简介一款针对京东电商平台设计的商品价格实时监控工具面向普通消费者、电商运营者以及正学习爬虫与数据采集的开发者帮助解决价格波动难追踪、促销时机易错过、比价耗时间等痛点。资源基于Python实现共8个文件压缩包体积仅7KB文件构成包括4个Python脚本分别承担页面抓取、数据库连接、价格数据写入和邮件发送功能、1个用于初始化MySQL数据表的SQL脚本、1个XML配置文件、1个说明文档以及License授权文件整体轻量易部署。已有283人学习/下载具备一定参考热度。通过阅读说明文档和源码可以快速搭建一套简易的价格监控流程定时采集目标商品价格将结果存入数据库并在价格达到关注条件时通过邮件自动提醒辅助用户把握最佳购买或调价时机资源还涉及京东页面结构解析、请求频率控制等反爬应对思路适合有一定Python基础、希望上手真实电商数据监控项目的开发者作为参考。1. 京东商品价格实时监控它替你盯住的不只是价格数字京东买东西最让人懊恼的不是价格贵而是你不知道它什么时候突然降价。同一款商品前一天还是4599第二天就变3999等你想入手时又涨回去了。这个「京东电商商品价格实时监控」方案要解决的问题就是让一套脚本定期抓取你关注的商品价格把每次变动记进数据库当价格跌破你设定的心理价位时第一时间通知你。网上有不少现成的开源版本可以直接下载试跑但把原理和坑摸清楚之后再动手比拿黑匣子回来折腾高效得多。适合谁用经常在京东买东西、想等降价又不想整天盯页面的人以及做电商选品、需要跟踪竞品价格变化的从业者。2. 价格从哪拿四条获取路径的可行性和反爬成本拆解在写第一行抓取代码之前最应该想清楚的问题是价格数据到底从哪里来。京东没有对外公开的免费价格查询网站所以“获取路径”的选择直接决定了后面你要应付的反爬强度、字段完整度和接口可靠性。这一章把实际用过的四条路径逐个过一遍。2.1 PC详情页HTML解析入口最浅代码最脆最常见的做法是直接抓https://item.jd.com/{sku_id}.html这个PC详情页然后在HTML里找价格字段。早期版本的页面会在window.pageConfig这样的全局变量里输出部分商品信息价格也有可能以预渲染的形式存在于HTML源码中。但最近两年京东PC详情页的价格大多变成了异步加载HTML骨架先返回价格由前端再发起一次XHR请求才能拿到。也就是说你直接解析HTML可能什么都拿不到或者页面改版后字段结构一夜之间全部变了。另一个问题是PC详情页的HTML非常大平均在2到4MB之间。如果监控的商品多、频率高光是下载页面就消耗大量带宽和请求配额也更容易触发风控。所以我不会优先走这条路径它只适合“偶尔手动看一眼”的场景。动手前先在浏览器里打开详情页源码搜索p.price或price几个字段确认价格是否预渲染确认后再写解析器能省很多重新调试的时间。2.2 移动端H5接口字段干净但签名参数绕不过去移动端https://item.m.jd.com/product/{sku_id}.html对应的接口字段比PC端干净返回的JSON里直接有单价、销量、促销信息等结构化字段。但问题是移动端的API普遍依赖签名参数functionId、appid、body等这些参数有些是固定值有些是前端实时生成的签名。对于一次性的小规模抓取把固定参数照抄下来、带上一个手机端的UA确实能跑通。但当京东更新前端代码、更换签名算法后你的请求就会静默失败——返回的JSON里不再有价格取而代之的是错误码。维护成本在这里你需要定期回来确认接口还能用。如果你只是做短期监控比如盯一个月以内的活动价格H5接口完全够用但我不建议把它作为长期监控的基础路径。2.3 价格直查接口p.3.cn个人监控优先尝试的数据源路径三是一条公开的价格查询接口形式上很直接https://p.3.cn/prices/mgets?skuIdsJ_{sku_id}用GET请求带上一个或多个skuId返回的就是JSON数组。单个商品也可以查多个商品用英文逗号拼在一起也可以。返回格式大概是这样的[{id:J_100012043978,p:3999.00,op:4999.00,m:4599.00,t:0}]其中p是当前售价op是划线价m是市场标价。这条路径的优势在于返回体极小一次请求只消耗几KB流量对服务器和风控的压力都小字段就三个解析逻辑永远不会复杂接口存在时间很长京东多次改版都没有动它很多比价工具都在用。它也有明显的边界拿不到促销规则详情比如满减、赠品、plus专享价要额外算以及p返回到手价的准确性依赖请求头里带的Cookie。总的来说个人监控场景里我会优先从这条接口入手。单个sku的监控代码量极低后面第3章就是围绕它实现的。2.4 开放平台API企业向选择个人申请门槛不低京东开放平台提供正式的API能力查询商品价格需要申请相应的应用权限并且走授权流程。审批周期通常以工作日计部分接口还要求企业资质和一定的调用量配额。如果是个人开发者申请成本偏高只跑几个自己关注的商品确实有点杀鸡用牛刀。开放平台API也有不可替代的价值稳定的调用配额、完整的商品快照、结构化的价格与促销字段以及官方支持的调用方式和限频策略。适合有团队维护、对数据准确性要求高、要对接进业务系统的场景。个人监控用不到这一层。四条路径的取舍用下面的表格可以看得很直接获取路径字段完整度维护成本风控风险适用场景PC详情页HTML完整但有异步高高人工偶尔查看移动端H5接口结构化完整高签名中短期小批量p.3.cn价格接口仅价格字段低低个人长期监控开放平台API完整官方低低企业集成2.5 确定监控对象从商品链接里提取SKU ID不管用哪条路径你都得先把京东的SKU ID拿到手。京东商品详情页的URL格式是https://item.jd.com/100012043978.html最后那一串数字就是SKU ID。在Excel或者文本文件里维护一份商品清单每一行放一个SKU ID再配一个备注名字方便识别比如“RTX 4060显卡”“米面油套装”。后面脚本读这个清单循环抓取即可。注意不要直接在代码里写死SKU列表否则换监控对象时要改代码很麻烦。3. 搭一个能长期跑的价格监控服务调度、存储与告警闭环核心服务用Python写依赖只用到三个库requests负责HTTP请求APScheduler负责定时调度sqlite3是Python标准库直接做数据存储。这三个库在Windows和Linux上都能跑下面所有代码在Python 3.8以上版本可直接运行。建议使用虚拟环境把依赖隔离起来避免污染系统Python环境。3.1 抓取价格从p.3.cn读取商品实时售价先写一个最小可运行的抓取函数单独放在fetcher.py里。它的职责只有一个给定SKU ID返回当前价格记录。# fetcher.py import requests import time from datetime import datetime HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36, Referer: https://item.jd.com/, Accept: application/json, text/plain, */*, } def fetch_price(sku_id: str) - dict | None: 从京东价格接口拉取单个商品的价格快照 url https://p.3.cn/prices/mgets params {skuIds: fJ_{sku_id}} try: resp requests.get(url, paramsparams, headersHEADERS, timeout10) data resp.json() except Exception as exc: print(f[{datetime.now()}] 请求失败: {exc}) return None if not data or not isinstance(data, list): print(f[{datetime.now()}] SKU {sku_id} 返回异常数据: {resp.text[:200]}) return None item data[0] try: price float(item.get(p, 0)) original_price float(item.get(op, price)) except (ValueError, KeyError): print(f[{datetime.now()}] SKU {sku_id} 价格字段解析失败) return None # 返回的价格是0或负数说明商品已下架或接口异常 if price 0: return None return { sku_id: sku_id, price: price, original_price: original_price, checked_at: int(time.time()), }几个实现上的细节值得说明。请求头里的Referer一定要设置成京东的商品域名有些反爬策略会校验来源页不带Referer的请求有较大概率被拒绝或返回空数据。timeout10是必须写的否则某个请求卡住时整个监控进程都会被拖停。返回判断里把“请求失败”和“商品失效”分开处理fetch_price返回None的原因不同后面的告警逻辑才能区分“网络抖动”和“监控对象没了”。3.2 定时调度用APScheduler控制抓取频率价格监控本质上是一个定时任务。APScheduler的BlockingScheduler适合独立脚本运行的模式如果是Web服务里的监控任务需要改用BackgroundScheduler。下面这段是调度部分的核心写法。# scheduler.py from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger from fetcher import fetch_price def monitor_job(): sku_list [100012043978, 100008348542] # 从外部文件读取 for sku in sku_list: record fetch_price(sku) if record: # 先入库再判断是否告警 insert_price(record) check_and_alert(record, threshold5) else: # 标记一次失败连续失败3次触发失效告警 mark_failure(sku) scheduler BlockingScheduler() scheduler.add_job( monitor_job, triggerIntervalTrigger(minutes15), idjd_price_job, max_instances1, coalesceTrue, misfire_grace_time300, ) scheduler.start()频率参数需要按场景调不是越快越好。我一般用下面的对照来设定IntervalTrigger参数实际频率适用场景注意点minutes160秒一次秒杀倒计时、限量抢购风控风险极高容易封IPminutes15每15分钟一次常规促销波动监控推荐默认值平衡时效和风险hours1每小时一次价格稳定的日用品对服务器压力最小max_instances1的意思是上一次任务还没跑完时本次触发的周期直接跳过避免任务堆积。coalesceTrue表示如果因为进程休眠等原因积压了多个任务周期只补执行最后一次。misfire_grace_time300给了任务5分钟的容错窗口错过调度时间在这个窗口内仍然会补跑超过则放弃。这三个参数一起保证了服务长时间运行后不会出现“任务雪崩”。3.3 存储设计SQLite一张表装下所有历史价格历史价格数据用SQLite存足够了不需要上MySQL。单表加一个索引就能覆盖所有查询场景后续做曲线分析、比价排序都很方便。# storage.py import sqlite3 import time def get_connection(db_pathprice_monitor.db): conn sqlite3.connect(db_path) conn.execute(PRAGMA journal_modeWAL;) # 开启WAL模式减少读写锁冲突 return conn def init_db(conn): conn.execute( CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku_id TEXT NOT NULL, price REAL NOT NULL, original_price REAL NOT NULL DEFAULT 0, checked_at INTEGER NOT NULL ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_sku_checked ON price_history(sku_id, checked_at) ) conn.commit() def insert_price(conn, record: dict): conn.execute( INSERT INTO price_history (sku_id, price, original_price, checked_at) VALUES (?, ?, ?, ?), (record[sku_id], record[price], record[original_price], record[checked_at]), ) conn.commit()表结构刻意保持极简sku_id、price、original_price、checked_at四个字段足够解决90%的问题。checked_at用Unix时间戳存储排序和范围过滤都比字符串时间高效。索引建在(sku_id, checked_at)组合列上这是查询频率最高的条件组合。WAL模式是血泪经验换来的默认的journal模式在高频写入时容易出现数据库锁等待表现为“database is locked”错误开了WAL之后写并发立刻缓解。3.4 告警接入邮件、钉钉机器人和Server酱参数对照价格抓到并且入库之后下一步是触发通知。告警渠道的选择不复杂邮件最通用钉钉机器人适合公司内部接收Server酱适合个人手机推送。下面给出发送函数和各自需要准备的材料。# notifier.py import requests def send_email(smtp_host, smtp_port, username, auth_code, to_addr, subject, content): 使用SMTP发送降价通知邮件 import smtplib from email.mime.text import MIMEText msg MIMEText(content, plain, utf-8) msg[Subject] subject msg[From] username msg[To] to_addr with smtplib.SMTP_SSL(smtp_host, smtp_port, timeout10) as server: server.login(username, auth_code) server.sendmail(username, [to_addr], msg.as_string()) def send_dingtalk(webhook: str, message: str): 发送到钉钉群机器人webhook从钉钉群设置里获取 payload {msgtype: text, text: {content: message}} requests.post(webhook, jsonpayload, timeout10) def send_serverchan(send_key: str, title: str, content: str): Server酱扫码登录后获得SendKey免费版每天有发送条数限制 url fhttps://sctapi.ftqq.com/{send_key}.send requests.post(url, data{title: title, desp: content}, timeout10)三个渠道各有取舍渠道需要准备的材料主要限制SMTP邮件邮箱授权码、SMTP服务器地址容易被邮箱判定为垃圾邮件钉钉机器人群里的Webhook地址必须建群适合团队内部Server酱微信扫码换取的SendKey免费版条数有限适合个人低频通知接入时把发送函数包一层异常捕获因为通知渠道随时可能挂掉比如邮箱授权码过期、钉钉Webhook被撤销。通知发送失败不应该影响主流程继续抓取顶多记一条日志。我在生产环境里会把通知次数也做个上限控制同一商品一天内最多推送三条防止促销抖动期间反复轰炸。4. 价格监控避坑指南五个真实踩坑记录这一章记录的坑全部来自实际跑监控服务时遇到的问题。每一条都按现象、原因、解决的顺序写清楚方便你对照排查。4.1 请求被风控拦截返回的是验证码而不是价格现象脚本跑了两天突然某天开始resp.json()抛异常打印出原始响应发现是一段HTML文本里面包含“请通过验证”和滑块验证的提示。价格数据全部拿不到但浏览器打开商品页面一切正常。原因请求频率太高而且所有请求的User-Agent完全一致加上没有Cookie风控系统很容易识别出这是脚本流量。p.3.cn虽然接口公开也有基本的反爬开关。解决把抓取频率从每分钟一次降为每15分钟一次请求头发齐User-Agent、Referer、Accept每次请求之间加0.5到2秒的随机等待。如果仍然被拦停止抓取等待1到2小时自然解封。更好的办法是把自己账号登录后的Cookie放进请求头风控概率会显著降低。注意Cookie里通常有过期时间发现返回异常后优先检查Cookie是不是失效了。4.2 秒杀价引发半夜误报需要状态机过滤瞬时价格现象设置降价5%即告警凌晨3点收到一条降价通知当时没在意早上起来一看价格已经回到原位。查历史记录发现那次低价只持续了20分钟。原因凌晨时段的闪购价、限量秒杀价是瞬时价格持续几分钟到半小时不等。如果告警逻辑只看“当前价低于上次记录价”这一个条件这种瞬时波动也会触发通知价值不大还打扰休息。解决引入二次确认机制。第一次发现降价时不直接发送通知而是把该SKU标记为“候选”状态等下一次抓取时如果价格仍然低于阈值再真正发送通知。这样秒杀价如果只出现一个抓取周期15分钟内结束第二次抓取价会恢复正常自然不会触发告警。相当于给价格信号加了一个状态机过滤掉单点异常。4.3 同一商品不同账号看到的价格不一样现象脚本监控某商品显示价格3200元同一个商品在手机上用Plus会员账号看到的价格是3100元换个普通账号又变成3250元。原因京东对不同用户展示不同价格。Plus会员有专享价部分用户还有隐藏券匿名接口拿到的只是基准价并非所有人群的到手价。解决把自己账号的Cookie传给请求头让脚本以固定身份获取价格保证监控口径一致。但同时要注意账号登录态过期后接口会静默返回基准价看起来数据正常实际上已经换口径了。我的做法是在返回结果里附带一个member字段标识当前是Plus价还是普通价每次变化时输出到日志一对比就能发现口径切换。4.4 SQLite数据库膨胀和损坏历史数据差点全没现象跑了几个月price_monitor.db文件接近500MB查询变慢。一次意外断电后重启发现数据库打开报错历史数据读不出来。原因每次插入都立刻commit加上WAL日志文件不断增长且从未checkpoint数据库文件碎片化严重。进程在写入中途被强杀也可能留下未恢复的WAL状态。解决开启WAL模式后定期执行PRAGMA wal_checkpoint(TRUNCATE);把WAL文件合并回主库并截断每次数据库连接用完及时关闭不要长时间持有增加过期数据清理逻辑超过90天的明细数据可以删除或归档明细价格只保留最近三个月更早的数据按天聚合保存。我把这些整理成一个maintenance_job每周日凌晨执行一次数据库体积被控制在几十MB以内。4.5 商品下架后监控静默失效现象某个SKU长期缺货监控脚本还在跑但日志里连续输出“返回异常数据”没有收到任何告警。用户以为监控坏了实际是商品已经下架。原因接口对失效商品返回空数组或价格字段为负数脚本没有区分“请求失败”和“商品失效”这两种不同的异常。网络抖动失败会自行恢复商品失效则不会需要人工介入重新选品。解决给每个SKU维护一个连续失败计数器连续失败3次触发“监控失效”告警文案里带上SKU ID和商品备注提醒人工确认。人工处理完将计数器清零。同时把“请求失败”“商品失效”“接口改动”三种情况分别打不同日志级别排查问题的时候能一眼定位是网络层、业务层还是接口层出了问题。这个区分在长期运行的服务里太重要了否则你只能靠猜来排障。5. 把历史价格变成决策依据识别伪促销与最佳入手窗口有了完整的历史价格曲线监控的用途就不再只是“降价喊我”而是可以反过来辅助购买决策。这里分享两个我用得最顺手的分析思路。5.1 用30天最高价戳穿先涨后降京东部分商品在大促前会先上调原价再通过“立减”制造折扣。仅看当天的促销力度根本识别不了但历史价格曲线能直接暴露这个操作。写一个简单的聚合查询取近30天最高价和当前价做对比def analyze_discount(conn, sku_id, days30): start int(time.time()) - days * 86400 rows conn.execute( SELECT price FROM price_history WHERE sku_id? AND checked_at?, (sku_id, start), ).fetchall() if not rows: return prices [r[0] for r in rows] high max(prices) current prices[-1] drop (high - current) / high * 100 print(f近{len(prices)}次采样, 最高价{high}, 当前价{current}, 折扣{drop:.1f}%)判断逻辑很简单折扣率超过15%且当前价确实低于近30天正常区间才算真正值得买。如果某商品大促当天的“折扣率”还不如一周前的普通价那种促销直接忽略。5.2 把价格走势和价保规则结合起来京东多数自营商品支持价格保护价保期限一般为7天或30天。我在监控告警里加了一条规则如果商品在购买后7天内降价超过阈值脚本会自动生成一条提醒提示去申请价保。这条规则配合历史曲线能覆盖“买完就降价”的后悔药场景。查询代码也很简单取购买时间点之后的最低价与购买价比较即可。5.3 我现在的监控习惯我现在常年在监控清单里放着两类商品一类是单价高、价格波动频繁的数码产品另一类是固定周期补货的日用品。数码产品设15分钟抓一次、降幅3%告警日用品每小时抓一次、降幅10%才通知。大促前两周会额外记录一次每日快照配合5.1的伪促销判断来看基本不会被大促氛围带节奏。这套方案跑了两年多最大的收获不是省了多少钱而是下单前有了可查的历史依据不再凭感觉猜“现在是不是最低点”。希望帮到你。本文还有配套的精品资源点击获取 京东商品价格实时监控全攻略：从接口爬取到自动告警
简介一款针对京东电商平台设计的商品价格实时监控工具面向普通消费者、电商运营者以及正学习爬虫与数据采集的开发者帮助解决价格波动难追踪、促销时机易错过、比价耗时间等痛点。资源基于Python实现共8个文件压缩包体积仅7KB文件构成包括4个Python脚本分别承担页面抓取、数据库连接、价格数据写入和邮件发送功能、1个用于初始化MySQL数据表的SQL脚本、1个XML配置文件、1个说明文档以及License授权文件整体轻量易部署。已有283人学习/下载具备一定参考热度。通过阅读说明文档和源码可以快速搭建一套简易的价格监控流程定时采集目标商品价格将结果存入数据库并在价格达到关注条件时通过邮件自动提醒辅助用户把握最佳购买或调价时机资源还涉及京东页面结构解析、请求频率控制等反爬应对思路适合有一定Python基础、希望上手真实电商数据监控项目的开发者作为参考。1. 京东商品价格实时监控它替你盯住的不只是价格数字京东买东西最让人懊恼的不是价格贵而是你不知道它什么时候突然降价。同一款商品前一天还是4599第二天就变3999等你想入手时又涨回去了。这个「京东电商商品价格实时监控」方案要解决的问题就是让一套脚本定期抓取你关注的商品价格把每次变动记进数据库当价格跌破你设定的心理价位时第一时间通知你。网上有不少现成的开源版本可以直接下载试跑但把原理和坑摸清楚之后再动手比拿黑匣子回来折腾高效得多。适合谁用经常在京东买东西、想等降价又不想整天盯页面的人以及做电商选品、需要跟踪竞品价格变化的从业者。2. 价格从哪拿四条获取路径的可行性和反爬成本拆解在写第一行抓取代码之前最应该想清楚的问题是价格数据到底从哪里来。京东没有对外公开的免费价格查询网站所以“获取路径”的选择直接决定了后面你要应付的反爬强度、字段完整度和接口可靠性。这一章把实际用过的四条路径逐个过一遍。2.1 PC详情页HTML解析入口最浅代码最脆最常见的做法是直接抓https://item.jd.com/{sku_id}.html这个PC详情页然后在HTML里找价格字段。早期版本的页面会在window.pageConfig这样的全局变量里输出部分商品信息价格也有可能以预渲染的形式存在于HTML源码中。但最近两年京东PC详情页的价格大多变成了异步加载HTML骨架先返回价格由前端再发起一次XHR请求才能拿到。也就是说你直接解析HTML可能什么都拿不到或者页面改版后字段结构一夜之间全部变了。另一个问题是PC详情页的HTML非常大平均在2到4MB之间。如果监控的商品多、频率高光是下载页面就消耗大量带宽和请求配额也更容易触发风控。所以我不会优先走这条路径它只适合“偶尔手动看一眼”的场景。动手前先在浏览器里打开详情页源码搜索p.price或price几个字段确认价格是否预渲染确认后再写解析器能省很多重新调试的时间。2.2 移动端H5接口字段干净但签名参数绕不过去移动端https://item.m.jd.com/product/{sku_id}.html对应的接口字段比PC端干净返回的JSON里直接有单价、销量、促销信息等结构化字段。但问题是移动端的API普遍依赖签名参数functionId、appid、body等这些参数有些是固定值有些是前端实时生成的签名。对于一次性的小规模抓取把固定参数照抄下来、带上一个手机端的UA确实能跑通。但当京东更新前端代码、更换签名算法后你的请求就会静默失败——返回的JSON里不再有价格取而代之的是错误码。维护成本在这里你需要定期回来确认接口还能用。如果你只是做短期监控比如盯一个月以内的活动价格H5接口完全够用但我不建议把它作为长期监控的基础路径。2.3 价格直查接口p.3.cn个人监控优先尝试的数据源路径三是一条公开的价格查询接口形式上很直接https://p.3.cn/prices/mgets?skuIdsJ_{sku_id}用GET请求带上一个或多个skuId返回的就是JSON数组。单个商品也可以查多个商品用英文逗号拼在一起也可以。返回格式大概是这样的[{id:J_100012043978,p:3999.00,op:4999.00,m:4599.00,t:0}]其中p是当前售价op是划线价m是市场标价。这条路径的优势在于返回体极小一次请求只消耗几KB流量对服务器和风控的压力都小字段就三个解析逻辑永远不会复杂接口存在时间很长京东多次改版都没有动它很多比价工具都在用。它也有明显的边界拿不到促销规则详情比如满减、赠品、plus专享价要额外算以及p返回到手价的准确性依赖请求头里带的Cookie。总的来说个人监控场景里我会优先从这条接口入手。单个sku的监控代码量极低后面第3章就是围绕它实现的。2.4 开放平台API企业向选择个人申请门槛不低京东开放平台提供正式的API能力查询商品价格需要申请相应的应用权限并且走授权流程。审批周期通常以工作日计部分接口还要求企业资质和一定的调用量配额。如果是个人开发者申请成本偏高只跑几个自己关注的商品确实有点杀鸡用牛刀。开放平台API也有不可替代的价值稳定的调用配额、完整的商品快照、结构化的价格与促销字段以及官方支持的调用方式和限频策略。适合有团队维护、对数据准确性要求高、要对接进业务系统的场景。个人监控用不到这一层。四条路径的取舍用下面的表格可以看得很直接获取路径字段完整度维护成本风控风险适用场景PC详情页HTML完整但有异步高高人工偶尔查看移动端H5接口结构化完整高签名中短期小批量p.3.cn价格接口仅价格字段低低个人长期监控开放平台API完整官方低低企业集成2.5 确定监控对象从商品链接里提取SKU ID不管用哪条路径你都得先把京东的SKU ID拿到手。京东商品详情页的URL格式是https://item.jd.com/100012043978.html最后那一串数字就是SKU ID。在Excel或者文本文件里维护一份商品清单每一行放一个SKU ID再配一个备注名字方便识别比如“RTX 4060显卡”“米面油套装”。后面脚本读这个清单循环抓取即可。注意不要直接在代码里写死SKU列表否则换监控对象时要改代码很麻烦。3. 搭一个能长期跑的价格监控服务调度、存储与告警闭环核心服务用Python写依赖只用到三个库requests负责HTTP请求APScheduler负责定时调度sqlite3是Python标准库直接做数据存储。这三个库在Windows和Linux上都能跑下面所有代码在Python 3.8以上版本可直接运行。建议使用虚拟环境把依赖隔离起来避免污染系统Python环境。3.1 抓取价格从p.3.cn读取商品实时售价先写一个最小可运行的抓取函数单独放在fetcher.py里。它的职责只有一个给定SKU ID返回当前价格记录。# fetcher.py import requests import time from datetime import datetime HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36, Referer: https://item.jd.com/, Accept: application/json, text/plain, */*, } def fetch_price(sku_id: str) - dict | None: 从京东价格接口拉取单个商品的价格快照 url https://p.3.cn/prices/mgets params {skuIds: fJ_{sku_id}} try: resp requests.get(url, paramsparams, headersHEADERS, timeout10) data resp.json() except Exception as exc: print(f[{datetime.now()}] 请求失败: {exc}) return None if not data or not isinstance(data, list): print(f[{datetime.now()}] SKU {sku_id} 返回异常数据: {resp.text[:200]}) return None item data[0] try: price float(item.get(p, 0)) original_price float(item.get(op, price)) except (ValueError, KeyError): print(f[{datetime.now()}] SKU {sku_id} 价格字段解析失败) return None # 返回的价格是0或负数说明商品已下架或接口异常 if price 0: return None return { sku_id: sku_id, price: price, original_price: original_price, checked_at: int(time.time()), }几个实现上的细节值得说明。请求头里的Referer一定要设置成京东的商品域名有些反爬策略会校验来源页不带Referer的请求有较大概率被拒绝或返回空数据。timeout10是必须写的否则某个请求卡住时整个监控进程都会被拖停。返回判断里把“请求失败”和“商品失效”分开处理fetch_price返回None的原因不同后面的告警逻辑才能区分“网络抖动”和“监控对象没了”。3.2 定时调度用APScheduler控制抓取频率价格监控本质上是一个定时任务。APScheduler的BlockingScheduler适合独立脚本运行的模式如果是Web服务里的监控任务需要改用BackgroundScheduler。下面这段是调度部分的核心写法。# scheduler.py from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger from fetcher import fetch_price def monitor_job(): sku_list [100012043978, 100008348542] # 从外部文件读取 for sku in sku_list: record fetch_price(sku) if record: # 先入库再判断是否告警 insert_price(record) check_and_alert(record, threshold5) else: # 标记一次失败连续失败3次触发失效告警 mark_failure(sku) scheduler BlockingScheduler() scheduler.add_job( monitor_job, triggerIntervalTrigger(minutes15), idjd_price_job, max_instances1, coalesceTrue, misfire_grace_time300, ) scheduler.start()频率参数需要按场景调不是越快越好。我一般用下面的对照来设定IntervalTrigger参数实际频率适用场景注意点minutes160秒一次秒杀倒计时、限量抢购风控风险极高容易封IPminutes15每15分钟一次常规促销波动监控推荐默认值平衡时效和风险hours1每小时一次价格稳定的日用品对服务器压力最小max_instances1的意思是上一次任务还没跑完时本次触发的周期直接跳过避免任务堆积。coalesceTrue表示如果因为进程休眠等原因积压了多个任务周期只补执行最后一次。misfire_grace_time300给了任务5分钟的容错窗口错过调度时间在这个窗口内仍然会补跑超过则放弃。这三个参数一起保证了服务长时间运行后不会出现“任务雪崩”。3.3 存储设计SQLite一张表装下所有历史价格历史价格数据用SQLite存足够了不需要上MySQL。单表加一个索引就能覆盖所有查询场景后续做曲线分析、比价排序都很方便。# storage.py import sqlite3 import time def get_connection(db_pathprice_monitor.db): conn sqlite3.connect(db_path) conn.execute(PRAGMA journal_modeWAL;) # 开启WAL模式减少读写锁冲突 return conn def init_db(conn): conn.execute( CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku_id TEXT NOT NULL, price REAL NOT NULL, original_price REAL NOT NULL DEFAULT 0, checked_at INTEGER NOT NULL ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_sku_checked ON price_history(sku_id, checked_at) ) conn.commit() def insert_price(conn, record: dict): conn.execute( INSERT INTO price_history (sku_id, price, original_price, checked_at) VALUES (?, ?, ?, ?), (record[sku_id], record[price], record[original_price], record[checked_at]), ) conn.commit()表结构刻意保持极简sku_id、price、original_price、checked_at四个字段足够解决90%的问题。checked_at用Unix时间戳存储排序和范围过滤都比字符串时间高效。索引建在(sku_id, checked_at)组合列上这是查询频率最高的条件组合。WAL模式是血泪经验换来的默认的journal模式在高频写入时容易出现数据库锁等待表现为“database is locked”错误开了WAL之后写并发立刻缓解。3.4 告警接入邮件、钉钉机器人和Server酱参数对照价格抓到并且入库之后下一步是触发通知。告警渠道的选择不复杂邮件最通用钉钉机器人适合公司内部接收Server酱适合个人手机推送。下面给出发送函数和各自需要准备的材料。# notifier.py import requests def send_email(smtp_host, smtp_port, username, auth_code, to_addr, subject, content): 使用SMTP发送降价通知邮件 import smtplib from email.mime.text import MIMEText msg MIMEText(content, plain, utf-8) msg[Subject] subject msg[From] username msg[To] to_addr with smtplib.SMTP_SSL(smtp_host, smtp_port, timeout10) as server: server.login(username, auth_code) server.sendmail(username, [to_addr], msg.as_string()) def send_dingtalk(webhook: str, message: str): 发送到钉钉群机器人webhook从钉钉群设置里获取 payload {msgtype: text, text: {content: message}} requests.post(webhook, jsonpayload, timeout10) def send_serverchan(send_key: str, title: str, content: str): Server酱扫码登录后获得SendKey免费版每天有发送条数限制 url fhttps://sctapi.ftqq.com/{send_key}.send requests.post(url, data{title: title, desp: content}, timeout10)三个渠道各有取舍渠道需要准备的材料主要限制SMTP邮件邮箱授权码、SMTP服务器地址容易被邮箱判定为垃圾邮件钉钉机器人群里的Webhook地址必须建群适合团队内部Server酱微信扫码换取的SendKey免费版条数有限适合个人低频通知接入时把发送函数包一层异常捕获因为通知渠道随时可能挂掉比如邮箱授权码过期、钉钉Webhook被撤销。通知发送失败不应该影响主流程继续抓取顶多记一条日志。我在生产环境里会把通知次数也做个上限控制同一商品一天内最多推送三条防止促销抖动期间反复轰炸。4. 价格监控避坑指南五个真实踩坑记录这一章记录的坑全部来自实际跑监控服务时遇到的问题。每一条都按现象、原因、解决的顺序写清楚方便你对照排查。4.1 请求被风控拦截返回的是验证码而不是价格现象脚本跑了两天突然某天开始resp.json()抛异常打印出原始响应发现是一段HTML文本里面包含“请通过验证”和滑块验证的提示。价格数据全部拿不到但浏览器打开商品页面一切正常。原因请求频率太高而且所有请求的User-Agent完全一致加上没有Cookie风控系统很容易识别出这是脚本流量。p.3.cn虽然接口公开也有基本的反爬开关。解决把抓取频率从每分钟一次降为每15分钟一次请求头发齐User-Agent、Referer、Accept每次请求之间加0.5到2秒的随机等待。如果仍然被拦停止抓取等待1到2小时自然解封。更好的办法是把自己账号登录后的Cookie放进请求头风控概率会显著降低。注意Cookie里通常有过期时间发现返回异常后优先检查Cookie是不是失效了。4.2 秒杀价引发半夜误报需要状态机过滤瞬时价格现象设置降价5%即告警凌晨3点收到一条降价通知当时没在意早上起来一看价格已经回到原位。查历史记录发现那次低价只持续了20分钟。原因凌晨时段的闪购价、限量秒杀价是瞬时价格持续几分钟到半小时不等。如果告警逻辑只看“当前价低于上次记录价”这一个条件这种瞬时波动也会触发通知价值不大还打扰休息。解决引入二次确认机制。第一次发现降价时不直接发送通知而是把该SKU标记为“候选”状态等下一次抓取时如果价格仍然低于阈值再真正发送通知。这样秒杀价如果只出现一个抓取周期15分钟内结束第二次抓取价会恢复正常自然不会触发告警。相当于给价格信号加了一个状态机过滤掉单点异常。4.3 同一商品不同账号看到的价格不一样现象脚本监控某商品显示价格3200元同一个商品在手机上用Plus会员账号看到的价格是3100元换个普通账号又变成3250元。原因京东对不同用户展示不同价格。Plus会员有专享价部分用户还有隐藏券匿名接口拿到的只是基准价并非所有人群的到手价。解决把自己账号的Cookie传给请求头让脚本以固定身份获取价格保证监控口径一致。但同时要注意账号登录态过期后接口会静默返回基准价看起来数据正常实际上已经换口径了。我的做法是在返回结果里附带一个member字段标识当前是Plus价还是普通价每次变化时输出到日志一对比就能发现口径切换。4.4 SQLite数据库膨胀和损坏历史数据差点全没现象跑了几个月price_monitor.db文件接近500MB查询变慢。一次意外断电后重启发现数据库打开报错历史数据读不出来。原因每次插入都立刻commit加上WAL日志文件不断增长且从未checkpoint数据库文件碎片化严重。进程在写入中途被强杀也可能留下未恢复的WAL状态。解决开启WAL模式后定期执行PRAGMA wal_checkpoint(TRUNCATE);把WAL文件合并回主库并截断每次数据库连接用完及时关闭不要长时间持有增加过期数据清理逻辑超过90天的明细数据可以删除或归档明细价格只保留最近三个月更早的数据按天聚合保存。我把这些整理成一个maintenance_job每周日凌晨执行一次数据库体积被控制在几十MB以内。4.5 商品下架后监控静默失效现象某个SKU长期缺货监控脚本还在跑但日志里连续输出“返回异常数据”没有收到任何告警。用户以为监控坏了实际是商品已经下架。原因接口对失效商品返回空数组或价格字段为负数脚本没有区分“请求失败”和“商品失效”这两种不同的异常。网络抖动失败会自行恢复商品失效则不会需要人工介入重新选品。解决给每个SKU维护一个连续失败计数器连续失败3次触发“监控失效”告警文案里带上SKU ID和商品备注提醒人工确认。人工处理完将计数器清零。同时把“请求失败”“商品失效”“接口改动”三种情况分别打不同日志级别排查问题的时候能一眼定位是网络层、业务层还是接口层出了问题。这个区分在长期运行的服务里太重要了否则你只能靠猜来排障。5. 把历史价格变成决策依据识别伪促销与最佳入手窗口有了完整的历史价格曲线监控的用途就不再只是“降价喊我”而是可以反过来辅助购买决策。这里分享两个我用得最顺手的分析思路。5.1 用30天最高价戳穿先涨后降京东部分商品在大促前会先上调原价再通过“立减”制造折扣。仅看当天的促销力度根本识别不了但历史价格曲线能直接暴露这个操作。写一个简单的聚合查询取近30天最高价和当前价做对比def analyze_discount(conn, sku_id, days30): start int(time.time()) - days * 86400 rows conn.execute( SELECT price FROM price_history WHERE sku_id? AND checked_at?, (sku_id, start), ).fetchall() if not rows: return prices [r[0] for r in rows] high max(prices) current prices[-1] drop (high - current) / high * 100 print(f近{len(prices)}次采样, 最高价{high}, 当前价{current}, 折扣{drop:.1f}%)判断逻辑很简单折扣率超过15%且当前价确实低于近30天正常区间才算真正值得买。如果某商品大促当天的“折扣率”还不如一周前的普通价那种促销直接忽略。5.2 把价格走势和价保规则结合起来京东多数自营商品支持价格保护价保期限一般为7天或30天。我在监控告警里加了一条规则如果商品在购买后7天内降价超过阈值脚本会自动生成一条提醒提示去申请价保。这条规则配合历史曲线能覆盖“买完就降价”的后悔药场景。查询代码也很简单取购买时间点之后的最低价与购买价比较即可。5.3 我现在的监控习惯我现在常年在监控清单里放着两类商品一类是单价高、价格波动频繁的数码产品另一类是固定周期补货的日用品。数码产品设15分钟抓一次、降幅3%告警日用品每小时抓一次、降幅10%才通知。大促前两周会额外记录一次每日快照配合5.1的伪促销判断来看基本不会被大促氛围带节奏。这套方案跑了两年多最大的收获不是省了多少钱而是下单前有了可查的历史依据不再凭感觉猜“现在是不是最低点”。希望帮到你。本文还有配套的精品资源点击获取