京东商品价格实时监控：Python接口解析与定时调度方案

简介京东电商商品价格实时监控项目面向有Python基础、希望掌握电商数据抓取与价格追踪的开发者解决在京东等平台及时掌握商品降价动态、自动接收通知的问题。压缩包虽仅7KB但结构完整共8个文件以4个Python脚本为核心涵盖商品爬取、邮件通知、数据库连接等模块另含SQL建表脚本、XML项目配置、LICENSE及README说明便于快速理解工程脉络。目前已有283人学习浏览适合作为轻量级参考实现。通过阅读源码可掌握爬虫请求解析、价格数据持久化、定时触发邮件告警的完整思路并可根据自身需求扩展监控商品列表、添加更多通知渠道或接入数据分析对电商数据应用入门具有实用价值。1. 京东商品价格实时监控在解决什么问题想蹲一个京东商品降到心理价位再下单靠人工每隔几分钟刷新一次页面基本坚持不过半天。做电商选品和同行价格跟踪的人更痛苦竞品改价不会提前通知等人肉发现往往慢了一整天。京东商品价格实时监控就是用一段脚本在后台把这件事自动做完按设定周期抓取商品价格、写入历史库、计算跌幅价格触达阈值时发出提醒。它解决的问题非常具体适合两类人一是想在大促前精准抄底的普通买家二是需要持续跟踪竞品定价的运营或独立开发者。这篇文章从选型、实现到排错按一个能长期跑的方案来讲。2. 京东商品价格从哪拿价格接口与页面解析的选型理由做价格监控的第一步不是写代码而是搞清楚目标商品的价格到底能从哪个入口拿到。京东的商品详情页是服务端渲染加动态组件的混合页面价格在首次访问时可能由异步脚本填充直接拿requests请求详情页有时候能读到价格有时候读不到。实际开发中我一般会同时准备两条取价路径优先调用京东公开的批量价格查询接口失败时回退到商品详情页的 HTML 解析。两条路径都不复杂但选型逻辑需要先说清楚。价格接口的最大优势是返回结构固定、体积小、延迟低适合高频轮询页面解析的优势是不依赖某个接口是否还在维护只要页面版式没大改就能用。监控任务要长期跑接口字段调整和页面改版都是必然事件所以两条路径互为兜底比赌其中一条永远稳定更可靠。这一章先把这条主路径的完整实现给出来顺带讲清楚每个参数为什么这样设。2.1 京东商品页的价格数据藏在哪价格接口的返回结构与字段含义对京东商品比较熟的开发者都知道商品详情页的 URL 是https://item.jd.com/{sku_id}.htmlsku_id 一般是 12 位数字。价格数据通常有三个藏身位置渲染后的 HTML 里的.p-price .price元素、页面内嵌的 JSON 数据、以及独立的价格查询接口。其中独立的价格查询接口https://p.3.cn/prices/mgets?skuIdsJ_{sku_id}是最常用的它返回一个 JSON 数组每个元素至少包含三个字段id带J_前缀的 sku 标识、p当前售价、op原价或划线价。这个接口有几个细节要注意skuIds参数必须带J_前缀否则返回结果可能为空p和op都是字符串而不是数字直接拿来算数之前要做类型转换op字段可能为空字符串尤其当商品无原价展示时。另外接口对请求头有最基本的校验裸请求偶尔会被拦带上完整的 User-Agent 和 Referer 能明显降低失败概率。请求时把参数交给requests的params字段去拼不要自己拼 URL因为requests会正确处理编码也能让你后面加参数时不用改 URL 模板。2.2 用 requests 拿到京东商品价格的最小实现下面是取价主路径的完整代码。这个函数只做一件事传入 sku_id返回当前价和原价。import requests def fetch_jd_price(sku_id: str, timeout: int 5) - dict: 通过京东价格接口获取商品当前售价与原价 # 接口要求 sku 带 J_ 前缀参数名固定为 skuIds url https://p.3.cn/prices/mgets params {skuIds: fJ_{sku_id}} headers { User-Agent: ( Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36 ), Referer: fhttps://item.jd.com/{sku_id}.html, Accept: application/json, } resp requests.get(url, paramsparams, headersheaders, timeouttimeout) resp.raise_for_status() payload resp.json() if not payload: raise ValueError(fsku {sku_id} 返回空数据) item payload[0] return { sku_id: sku_id, price: float(item[p]), original_price: float(item.get(op) or 0), raw: item, }代码逻辑分四步先拼接带J_前缀的 sku 标识然后用带 UA 和 Referer 的请求头去请求接口拿到 JSON 后判断数组是否为空最后把p和op转成浮点数返回。注意item.get(op) or 0这个写法它专门处理op字段为空字符串的情况避免float()直接抛 ValueError 把整个监控任务打断。timeout5是给网络请求设的上限防止 DNS 解析慢或连接被挂起时线程卡死。有人习惯不写 timeout但在定时任务里这是必须的不然一个卡死请求会把整轮监控堵住。resp.raise_for_status()会主动抛出 HTTP 错误让上层可以统一捕获并重试而不是拿到一个 403 页面后继续走解析逻辑解析出莫名其妙的数值。2.3 页面解析兜底接口失效时的备用取价路径接口偶尔会改参数结构或直接下线所以备一条页面解析路径是有必要的。一个实用的兜底函数如下from bs4 import BeautifulSoup import requests def parse_jd_page_price(sku_id: str, headers: dict) - float | None: 接口取不到值时从商品详情页 HTML 里提取价格 url fhttps://item.jd.com/{sku_id}.html resp requests.get(url, headersheaders, timeout8) if resp.status_code ! 200: return None soup BeautifulSoup(resp.text, html.parser) # 商品价格通常渲染在 .p-price 下的 .price 元素里 el soup.select_one(.p-price .price) if el is None: return None text el.get_text().strip().replace(¥, ).replace(,, ) try: return float(text) except ValueError: return None这段兜底逻辑并不复杂请求详情页用 BeautifulSoup 定位价格节点清理掉货币符号和千分位逗号后转浮点数。页面解析的稳定性取决于京东是否改版所以每次大促前后都要抽查一次页面结构。若页面里出现“无货”“暂时缺货”这类文案.p-price节点可能不存在函数会返回None这不算异常而是商品状态变化的信号后面监控层要单独处理不能直接当作价格缺失报错。两条取价路径的选型对比可以这样归纳对比项价格接口 mgets页面解析返回格式JSON结构固定HTML需解析单次延迟几十毫秒到几百毫秒几百毫秒到几秒稳定性字段偶尔调整页面改版时需同步更新依赖仅 requestsrequests BeautifulSoup适用场景默认主路径接口失效或需要页面展示价时3. 把单次抓取升级成实时监控SQLite 入库、定时调度与降价判断能抓到价格只是第一步实时监控的本质是“周期抓取 历史对比”。价格数据如果只停留在打印日志那下次抓取就无从对比。一个最小可用的监控闭环由三部分组成存储层把每次抓到的价格记下来调度层按固定周期触发抓取判断层拿最新价与历史价比较、决定要不要提醒。这一章按这个顺序把闭环完整搭出来。存储我选 SQLite理由是监控任务是单机单进程的数据量一天撑死几十条用 MySQL 是杀鸡用牛刀用 CSV 又扛不住并发读写和索引查询。SQLite 不需要额外服务一个文件搞定备份和迁移都很简单。调度用 APScheduler它是 Python 生态里最常用的定时任务库支持阻塞式和后台式两种调度器几分钟就跑起来的成本非常低。判断逻辑则需要注意一个容易被忽略的点不是价格有变化就算降价要设定合理的阈值同时避免把同一波价格波动重复报出来。3.1 价格历史表设计SQLite 建表语句与写入函数价格历史表的设计要考虑两个场景按商品查走势、按时间查快照。所以我建表时把 sku_id 和 check_time 作为联合索引这是最常用的查询路径。价格字段建议以“分”为单位存整数而不是存带小数位的元因为浮点数在比较时可能因为精度问题产生误判整数分则完全避免这个坑。CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku_id TEXT NOT NULL, title TEXT DEFAULT , price INTEGER NOT NULL, original_price INTEGER DEFAULT 0, check_time TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_sku_time ON price_history(sku_id, check_time);price存的是“分”例如页面价 2999.00 元存的是 299900。check_time用 ISO 格式文本格式如2025-01-15 09:30:00排序可以直接按字典序比用 Unix 时间戳直观也不损失查询性能。title字段是冗余存储目的是在推送提醒时直接带出商品名省得每次都去请求一遍页面。写入函数封装成独立模块调用方只需要传入元数据import sqlite3 from datetime import datetime DB_PATH jd_prices.db def init_db(): with sqlite3.connect(DB_PATH) as conn: conn.execute( CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku_id TEXT NOT NULL, title TEXT DEFAULT , price INTEGER NOT NULL, original_price INTEGER DEFAULT 0, check_time TEXT NOT NULL ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_sku_time ON price_history(sku_id, check_time) ) def save_price(sku_id: str, title: str, price_yuan: float, original_yuan: float): with sqlite3.connect(DB_PATH) as conn: conn.execute( INSERT INTO price_history (sku_id, title, price, original_price, check_time) VALUES (?, ?, ?, ?, ?) , ( sku_id, title, int(price_yuan * 100), int(original_yuan * 100), datetime.now().strftime(%Y-%m-%d %H:%M:%S), ), )init_db()建表并建索引save_price()把元转成整数分入库。注意with sqlite3.connect(...)语句块里的execute会自动提交事务不用手动调commit()这能避免忘记提交导致的数据丢失。如果后续数据量大可以考虑用executemany批量插入但对个人监控场景单条插入完全够用。3.2 定时调度用 APScheduler 按固定周期轮询所有商品实时监控不需要精确到秒级价格变动的频率远没有股市行情那么高一般 30 分钟采样一次就能满足绝大多数场景。间隔太短容易触发拦截太长又可能错过短促的秒杀价30 分钟是综合权衡之后的经验值。如果监控对象是热门的自营家电数码可以把间隔缩到 15 分钟如果是冷门商品1 小时一次也够。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger INTERVAL_MINUTES 30 def run_once(): 执行一轮全量抓取与入库 for sku in load_config()[skus]: try: info fetch_jd_price(sku[id]) # 接口返回的 p 是元save_price 内部转换成分 save_price(sku[id], sku.get(name, ), info[price], info[original_price]) except Exception as exc: logger.error([run_once] sku%s error%s, sku[id], exc) scheduler BlockingScheduler() scheduler.add_job( run_once, IntervalTrigger(minutesINTERVAL_MINUTES), idjd_price_monitor, max_instances1, ) scheduler.start()max_instances1很关键它保证上一轮还没跑完时不会启动新一轮防止抓取慢的任务堆积成雪崩。BlockingScheduler会阻塞主线程适合纯脚本方式跑如果后续还要在同进程里提供 Web 服务就换成BackgroundScheduler两者 API 基本一致。调度器跑起来后日志里能看到每轮任务开始和结束我会把run_once的执行耗时也记下来超过预期就要检查是不是某个商品请求卡住了。3.3 降价判断用相对跌幅和价格下限过滤噪音价格监控最容易犯的错是把每一次价格波动都当成事件推出去。实际运营场景里商品在一天内小幅度浮动 1% 到 2% 是常态用户对这些毛刺提醒会很快麻木。所以判断逻辑要同时满足两个条件才触发提醒相对历史价格跌幅超过阈值并且当前价低于自己设置的心理价位可选。MIN_DROP_RATE 0.05 # 相对上一次价格跌幅超过 5% 才触发 TARGET_PRICE 1999 # 低于该价格才提醒可配置 def query_latest_price(sku_id: str): 查询最近一条历史记录排除当前这一轮刚写入的数据 with sqlite3.connect(DB_PATH) as conn: row conn.execute( SELECT price, check_time FROM price_history WHERE sku_id ? ORDER BY check_time DESC LIMIT 1 OFFSET 1 , (sku_id,), ).fetchone() return {price: row[0], check_time: row[1]} if row else None def check_and_notify(sku: dict, current_price_yuan: float): 判断当前价是否比上一次明显下降是则触发提醒 latest query_latest_price(sku[id]) if latest is None: logger.info(%s 首次采样等待下一次对比, sku[id]) return last_price latest[price] # 单位分 current_price int(current_price_yuan * 100) drop_rate (last_price - current_price) / last_price if drop_rate MIN_DROP_RATE: # 配置了 target_price 时必须低于目标价才提醒 if sku.get(target_price) and current_price_yuan sku[target_price]: logger.info(%s 跌幅达标但未到目标价不提醒, sku[id]) return logger.info( 降价触发 %s: %.2f - %.2f, 跌幅 %.1f%%, sku[id], last_price / 100, current_price / 100, drop_rate * 100, ) push_notice(sku, current_price / 100)query_latest_price里用OFFSET 1取的是“上一条”而不是“当前这一条”原因是监控任务先入库再判断如果直接取最新就会把自己刚写入的价格当成基准跌幅永远是 0提醒永远不会触发。这个细节容易踩坑后面避坑章节还会再强调。MIN_DROP_RATE用相对比例而不是绝对金额是因为 50 元的降价对 399 元的商品和 5999 元的商品含义完全不同比例更能反映真实折扣力度。4. 监控脚本要能长期跑多商品配置、异常重试与日志排查脚本跑五分钟没问题跑五天就未必了。价格监控是要常驻后台的所以从第一天起就要按“能长期运行”的标准来设计商品列表不能写死在代码里抓取失败必须有重试和记录日志要能支撑事后排查。这一章解决的是监控任务自身的稳定性否则数据断档了提醒再及时也没有意义。这个阶段常见的现象是脚本连续跑了一周某天突然发现凌晨某次采样没数据但日志里什么都没有。原因通常是网络抖动导致的请求超时异常被上层吞掉或者没有被正确记录。所以异常处理不能只print得有重试退避、失败落库、日志轮转这三板斧。4.1 多商品管理用 YAML 配置替代写死的 SKU 列表写死列表的代码一时爽后面加商品要改代码、重启进程做选品跟踪的人一天可能要调整好几轮。把商品配置抽到 YAML 文件里运营同事也可以直接改不用碰代码。# monitor.yaml interval_minutes: 30 skus: - id: 100012043654 name: 示例商品A target_price: 1999.00 - id: 100012043655 name: 示例商品B target_price: 899.00加载配置的代码很简单import yaml def load_config(path: str monitor.yaml): with open(path, encodingutf-8) as f: return yaml.safe_load(f)YAML 比 JSON 更适合做这种配置文件因为支持注释、中文友好、不会因为多一个逗号直接报错。target_price是可选的不填就退化为“跌幅超过阈值即提醒”。interval_minutes放在配置里是为了让不同环境本地调试、服务器常驻用不同的采样频率调试时改成 1 分钟正式跑改成 30 分钟不用改代码。整个监控的装配逻辑也就几行from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger def main(): cfg load_config() interval cfg.get(interval_minutes, 30) scheduler BlockingScheduler() scheduler.add_job( run_once, IntervalTrigger(minutesinterval), idjd_price_monitor, max_instances1, ) scheduler.start() if __name__ __main__: main()4.2 异常重试与日志监控进程自己不能成为黑匣子监控脚本跑在无人值守的服务器上出错时如果没有日志和重试基本等于瞎跑。我给所有外部请求都套了一个统一的重试装饰器失败后先记日志再按指数退避的策略等待后重试。import logging import time from functools import wraps logger logging.getLogger(__name__) def retry_on_error(retries: int 3, base_delay: float 3.0): 对目标函数做最多 retries 次重试间隔逐步放大 def deco(fn): wraps(fn) def wrapper(*args, **kwargs): for attempt in range(1, retries 1): try: return fn(*args, **kwargs) except Exception as exc: logger.warning(attempt %s/%s failed: %s, attempt, retries, exc) if attempt retries: time.sleep(base_delay * attempt) raise RuntimeError(f{fn.__name__} failed after {retries} attempts) return wrapper return decotime.sleep(base_delay * attempt)就是指数退避的简化版第一次失败等 3 秒第二次等 6 秒第三次等 9 秒。这样既给了网络抖动恢复的时间又不会在请求持续失败时高频轰炸目标站点。重试要包在“可能因网络失败”的请求函数上比如fetch_jd_price而不要包在业务逻辑层否则数据库异常也会被错误地重试三次。日志这件事标准库的logging加一个按天轮转的 handler 就够了from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler(jd_monitor.log, whenmidnight, backupCount7) formatter logging.Formatter(%(asctime)s [%(levelname)s] %(message)s) handler.setFormatter(formatter) logger.addHandler(handler) logger.setLevel(logging.INFO)backupCount7表示保留最近 7 天的日志超过自动清理避免日志文件把磁盘撑满。我用logger而不是print的目的是日志自带时间和级别排查问题时能准确知道某个错误发生在哪一轮采样的哪个阶段。再进一步失败记录也可以单独落一张表方便统计失败率CREATE TABLE IF NOT EXISTS monitor_errors ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku_id TEXT, error_msg TEXT, created_at TEXT );5. 京东价格监控避坑清单4 个我踩过的坑与排查方法监控脚本的功能可以很快写完真正花时间的是排错。这一章列出我在实际跑监控过程中遇到频率最高的四类问题每一条按“现象、原因、解决”的顺序说清照着排查能省很多时间。5.1 抓得太勤触发滑块验证频率控制与请求间隔现象监控跑了十几分钟后突然某轮所有商品都抓不到价格返回的 HTML 里出现了滑块验证页面接口请求也开始返回空数组。原因请求频率太高且请求头特征太单一。连续快速请求同一个商品页面服务端很容易从 IP 和 UA 维度识别出自动化脚本进而弹出验证页拦截。解决把默认采样间隔拉长到 30 分钟以上并在每次请求前加入随机抖动比如在 25 到 35 分钟之间随机取值。同时保证每个商品都有独立的 Referer 和完整 UA不要让所有请求都顶着同一个裸 UA。出现验证页时应立刻停止该商品的本轮抓取记日志后跳过而不是连续重试越重试越容易被封得久。我之前就把间隔调成 5 分钟跑了一晚上第二天醒来发现凌晨 2 点之后所有采样全部被拦这一整晚的数据都白费了。5.2 接口正常返回却解析不出价格op 字段为空与店铺商品缺价现象接口返回[{id:J_100012043654,p:4999.00,op:}]代码里float(item[op])直接抛 ValueError整个监控任务中断。原因op字段是原价或划线价部分商品尤其第三方店铺商品没有展示原价接口就会返回空字符串。代码里直接float()自然就炸了。解决统一用item.get(op) or 0处理缺省值空字符串会被or 0兜住变成 0。运行前把所有目标商品各抓一次人工核对返回结构把那些返回字段不完整的商品单独标记出来不要指望每个商品都长一个样。5.3 接口价格和商品页显示价格对不上活动价、Plus 价与优惠券的差异现象接口返回 3699 元但商品详情页显示 2999 元或者反过来接口是 2999 元页面挂着“Plus 会员价 2699 元”。两种数据都对但放在一起对比就会产生误报。原因接口给的是标准售价页面展示的可能是会场活动价、Plus 会员价甚至叠加了京券、东券后的预估到手价。优惠券通常不体现在本金里属于下单链路中的减免。解决监控之前先明确自己要盯的是哪一种价格。盯“用户实际看到的页面价”就统一用页面解析路径盯“基准售价”就用接口路径。千万不要在一个监控任务里混用两条取价路径否则曲线图会非常锯齿提醒也频繁误报。如果一定要跟踪到手价那就得把领券、Plus 判断等逻辑也做进去复杂度会明显上升我建议第一批版本不做。5.4 定时任务凌晨静默挂掉网络抖动与异常被吞现象早上起来看数据发现凌晨 3 点那一轮采样没有任何记录但日志文件里也没有报错仿佛这一轮任务从来没跑过。原因定时任务确实启动了但网络请求超时后异常没有被捕获或者任务因为上一个周期还没结束被max_instances1跳过了。还有一种可能是进程本身在凌晨被系统杀掉比如内存不足或 SSH 断开导致前台进程退出。解决所有网络请求必须设置timeout异常统一走logger.error记录下来核心请求函数套上重试装饰器。进程守护方面我习惯用nohup加上系统服务托管保证进程退出后能被重新拉起。数据层面每天定时检查price_history中是否有某个商品超过 2 个采样周期没有新记录有就触发一次“断更告警”这比事后翻日志省事得多。6. 让监控真的能用历史价格曲线验证与降价提醒自检功能写完了不等于能直接用。我踩过最大的坑是监控跑了一个月才发现数据是假的某个商品连续一星期抓到同一个价格页面其实已经涨价了但解析代码拿不到节点每次返回 0而上层逻辑把 0 当成“价格没变”提醒自然全是哑的。所以我的固定习惯是每周做两件小事画价格曲线、模拟一次降价。6.1 画价格曲线看数据是不是真实的价格走势import sqlite3 import pandas as pd import matplotlib.pyplot as plt def plot_price_history(sku_id: str, db_path: str jd_prices.db): conn sqlite3.connect(db_path) df pd.read_sql_query( SELECT check_time, price FROM price_history WHERE sku_id ? ORDER BY check_time , conn, params(sku_id,), ) conn.close() df[price_yuan] df[price] / 100 df[check_time] pd.to_datetime(df[check_time]) df.plot( xcheck_time, yprice_yuan, titlefSKU {sku_id} 价格走势, figsize(8, 4), ) plt.savefig(f{sku_id}_price.png, dpi150) plt.close()正常的价格曲线应该呈现平台或台阶状长时间不变某一天突然跳变。如果曲线出现无规律的尖刺说明某个采样点解析到了异常价格需要回翻日志定位是哪一轮取坏了。曲线也直观反映了采样是否连续连续一个星期都长一个样很可能是解析在空转要主动验证。6.2 模拟一次降价验证提醒链路没断等商品真降价来测试提醒链路太被动我一般直接在库里把最新一条记录的价格调高让下一轮对比天然触发跌幅。用 SQL 把最新一条记录的价格上调 10 元UPDATE price_history SET price (SELECT price 1000 FROM price_history WHERE sku_id 100012043654 ORDER BY id DESC LIMIT 1) WHERE id (SELECT max(id) FROM price_history WHERE sku_id 100012043654);这条 SQL 先把该 sku 最新一条记录的价格取出来加 1000 分即 10 元再写回去。下一轮定时任务跑完后当前价相比这条被改高的记录跌幅就会超过阈值推送分支就会被触发。如果推送没来优先检查push_notice是不是被上层异常拦截了或者query_latest_price的偏移写错了。模拟成功之后最好把数据库恢复原状或者直接在测试环境里做别污染真实曲线。做到这两步监控数据才不会成为新的黑匣子。我的教训是早期只看日志没看曲线结果一个商品连续一星期抓到同一价格、页面实际已经涨价数据是假的提醒自然也不会响。现在我把曲线检查和模拟降价当成固定动作宁可多花十分钟也不让监控变成哑巴。这个方向的投入大概一个周末跑一个月后你会发现省下的盯价时间远比写代码的时间多。希望帮到你。本文还有配套的精品资源点击获取