京东商品价格实时监控：从抓取到告警的完整实现-开发者社区

简介：这是一份面向电商价格监控场景的实战型工具包，适合对爬虫、数据采集与自动化通知感兴趣的开发者及运营人员参考。资源围绕京东商品价格实时跟踪展开，涵盖数据抓取、价格存储与邮件提醒等核心环节，可帮助读者理解价格监控从采集到推送的完整链路。压缩包共8个文件，约7KB，以4个Python脚本为主体，分别承担爬取、邮件发送、数据库连接等职责，另含1个SQL建表脚本、1个XML配置文件、1份LICENSE及1份README说明文档，结构精简、便于快速上手。目前已有283人学习下载。通过阅读源码，读者可掌握定时抓取商品价格、写入MySQL数据库、触发邮件通知的实现思路，并借鉴其模块划分与配置方式，为自建价格监控工具或二次开发提供可复用的参考模板。

1. 京东电商商品价格实时监控：从“手动刷新”到“自动盯盘”的落地思路

做电商运营或者自己搞副业的朋友，大概率都经历过这种场景：盯着一款商品等它降价，一天手动刷新几十次，结果刚去接杯水，回来发现秒杀价已经过了。更别提做竞品分析的时候，要同时盯十几个链接，靠人力根本盯不过来。京东电商商品价格实时监控这件事，本质上就是把这套“手动刷新”的动作自动化——定时抓取商品页面的价格字段，存下来，变化了再通知你。它解决的不是什么高深问题，就是“我不想一直盯着，但价格一变我就要知道”。适合谁呢？做店群比价的运营、倒腾数码产品的黄牛、写比价工具的后端开发，以及想给自己买东西省点钱的技术爱好者。这一篇不讲虚的，从抓取策略、接口选择、数据存储到告警触发，把一条能跑通的链路拆开讲清楚。

2. 先搞清楚抓什么：京东价格数据的三个来源与选型

在动手写代码之前，得先弄明白一件事：京东商品页面上显示的价格，并不是只有一个数字。你看到的“到手价”“秒杀价”“Plus会员价”“满减后价格”，背后对应的是不同的数据接口和计算逻辑。如果抓错了字段，监控出来的结果就是错的，后面做再多告警也是白搭。

2.1 页面直出价格与异步接口价格的区别

最常见的做法是直接请求商品详情页的 HTML，然后用正则或者解析库去提取价格。这种方式在几年前还能用，但现在京东的商品详情页大部分价格信息是异步加载的。你直接请求item.jd.com的页面，拿到的 HTML 里往往只有一个价格占位符，真实价格是页面加载后通过 JavaScript 再发一个请求去取的。

所以更稳妥的方式是找到那个异步接口。常见做法是打开浏览器的开发者工具，切到 Network 面板，筛选 XHR 请求，然后刷新商品页，观察哪个请求返回了价格字段。通常你会看到一个类似p.3.cn或者item-soa.jd.com的请求，返回的是 JSON 格式的数据，里面包含了p（价格）、m（市场价）等字段。

这里有个血泪经验：不同品类的商品，价格接口可能不一样。自营商品和第三方商家商品的接口路径就有差异。我一般会先拿几个不同品类的链接做测试，确认接口模式是否统一，再决定用一套逻辑还是分套逻辑。

2.2 自营与第三方店铺的价格字段差异

自营商品的价格相对稳定，接口返回的字段也比较规范。第三方店铺就麻烦一些，因为商家可以自己设置促销策略，价格字段里可能会多出“促销价”“拼购价”等变体。如果你监控的是第三方商品，建议把接口返回的完整 JSON 先打印出来看一遍，找到那个真正代表“用户最终支付价格”的字段。

下面是一个典型的请求示例，用来获取某个商品的实时价格：

import requests import json # 商品ID，从商品链接中提取，例如 https://item.jd.com/100012345678.html sku_id = "100012345678" # 京东价格接口的常见形式，实际使用时需要根据抓包结果替换 # 这里用的是一个示意地址，真实地址以你抓到的为准 price_api = f"https://p.3.cn/prices/mgets?skuIds=J_{sku_id}" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": f"https://item.jd.com/{sku_id}.html" } try: resp = requests.get(price_api, headers=headers, timeout=10) # 接口返回通常是列表，取第一个元素 data = resp.json() if isinstance(data, list) and len(data) > 0: price_info = data[0] # p 字段通常代表当前售价 current_price = price_info.get("p") # m 字段通常代表市场价 market_price = price_info.get("m") print(f"当前售价: {current_price}, 市场价: {market_price}") else: print("返回数据格式异常:", data) except Exception as e: print("请求失败:", e)

这段代码的逻辑很直接：构造价格接口的 URL，带上必要的请求头，然后解析 JSON 响应。参数说明上，skuIds是商品编号，前缀J_是京东价格接口的常见格式，但不同接口可能要求不同的前缀，这个必须根据你实际抓包的结果来调整。headers里的Referer很重要，有些接口会校验来源，不带的话可能返回空数据。

2.3 价格接口的请求频率与反爬边界

京东的价格接口不是无限让你调的。如果你每秒请求几十次，很快就会被限流，表现为返回 403 或者直接超时。我一般会把单商品的请求间隔控制在 30 秒到 1 分钟之间，如果是批量监控，会加一个队列，串行发送请求，每个请求之间 sleep 一个随机值，比如 1 到 3 秒。

另外，不要用同一个 IP 去高频请求。如果你监控的商品数量比较多，建议把请求分散到不同的时间段，或者使用多个出口 IP 轮换。但这里要注意，不要用那种来路不明的代理池，稳定性差不说，还可能带来安全风险。比较稳妥的做法是控制总量，只监控你真正关心的那几十个链接，而不是盲目铺量。

3. 把价格存下来：数据表设计与增量判断逻辑

抓到了价格只是第一步，接下来要解决的是“怎么存”和“怎么判断变化”。如果每次抓取都无脑入库，数据量会膨胀得很快，而且你也没法快速看出价格走势。合理的做法是设计一张价格历史表，每次抓取时先和上一条记录比对，只有价格变了才插入新记录，同时更新一个“最后监控时间”字段。

3.1 价格历史表的字段与索引设计

下面是一个用 SQLite 建表的示例，适合单机跑的小规模监控。如果你要监控上千个商品，建议换成 MySQL 或者 PostgreSQL。

-- 商品基础信息表 CREATE TABLE IF NOT EXISTS products ( sku_id TEXT PRIMARY KEY, -- 商品ID product_name TEXT, -- 商品名称，方便识别 created_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_checked_at DATETIME -- 最后一次检查时间 ); -- 价格历史表 CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku_id TEXT NOT NULL, -- 关联商品ID price REAL NOT NULL, -- 当前售价 market_price REAL, -- 市场价，可空 recorded_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (sku_id) REFERENCES products(sku_id) ); -- 为 sku_id 和 recorded_at 建联合索引，加速查询某个商品的价格走势 CREATE INDEX IF NOT EXISTS idx_sku_time ON price_history(sku_id, recorded_at);

字段设计上，price存的是实际售价，market_price存市场价，方便后续做折扣计算。recorded_at默认取当前时间，这样你查历史的时候可以直接按时间排序。索引idx_sku_time是为了加速“查某个商品最近 30 天价格”这类查询，没有索引的话，数据量一大查询就会变慢。

3.2 增量写入与价格变化触发条件

每次抓取到价格后，不要直接 INSERT，而是先查一下这个商品最近一条记录的价格。如果价格没变，只更新products表的last_checked_at；如果变了，才往price_history插一条新记录。

import sqlite3 from datetime import datetime def save_price(sku_id, current_price, market_price=None): conn = sqlite3.connect("price_monitor.db") cursor = conn.cursor() # 查询该商品最近一条价格记录 cursor.execute( "SELECT price FROM price_history WHERE sku_id = ? ORDER BY recorded_at DESC LIMIT 1", (sku_id,) ) last_record = cursor.fetchone() # 更新商品的最后检查时间 cursor.execute( "UPDATE products SET last_checked_at = ? WHERE sku_id = ?", (datetime.now(), sku_id) ) # 判断价格是否变化 if last_record is None or abs(last_record[0] - current_price) > 0.001: # 价格变了，或者这是第一条记录，插入新数据 cursor.execute( "INSERT INTO price_history (sku_id, price, market_price) VALUES (?, ?, ?)", (sku_id, current_price, market_price) ) print(f"价格变化: {sku_id} 从 {last_record[0] if last_record else '无'} 变为 {current_price}") else: print(f"价格未变: {sku_id} 仍为 {current_price}") conn.commit() conn.close()

这段代码的关键在于abs(last_record[0] - current_price) > 0.001这个判断。浮点数比较不能直接用!=，因为精度问题可能导致误判。用一个小阈值来判断，可以避免因为浮点误差而频繁插入重复数据。另外，last_checked_at的更新是每次都会执行的，这样你能知道监控任务有没有正常跑。

3.3 用 SQL 查最近价格走势的实用语句

存完数据之后，你可能会想看某个商品最近一周的价格变化。下面这条 SQL 可以直接用：

-- 查询某个商品最近 7 天的价格记录，按时间倒序 SELECT price, market_price, recorded_at FROM price_history WHERE sku_id = '100012345678' AND recorded_at >= datetime('now', '-7 days') ORDER BY recorded_at DESC;

如果你想知道某个商品的历史最低价，可以这样查：

-- 查询历史最低售价及出现时间 SELECT price, recorded_at FROM price_history WHERE sku_id = '100012345678' ORDER BY price ASC LIMIT 1;

这些查询在监控商品数量不多的时候完全够用。如果商品数量上千，建议把price_history按月分表，或者换用时序数据库，否则单表数据量大了之后查询会明显变慢。

4. 避坑与排查：价格监控常见的五个翻车点

这一章是我自己踩过的坑，以及帮别人排查问题时遇到最多的几种情况。如果你刚开始做，建议先看一遍，能省不少时间。

4.1 抓到的价格是“划线价”而不是“到手价”

现象：监控程序跑了一周，发现价格一直是同一个数字，但手动去京东看，明明有秒杀活动。

原因：你抓的接口返回的是市场价或者划线价，而不是用户实际支付的促销价。京东的价格体系里，p字段有时候是原价，促销价可能在另一个字段里，比如promoPrice或者需要结合促销信息计算。

解决：把接口返回的完整 JSON 打印出来，对比手动看到的价格，找到那个真正对应的字段。如果接口里没有，可能需要再调一个促销接口，或者从页面渲染后的 DOM 里提取。我一般会先用一个商品做对照实验，确认字段无误后再批量跑。

4.2 请求头缺失导致返回空数据

现象：代码在本地跑得好好的，放到服务器上就返回空列表或者 403。

原因：京东的接口会校验Referer和User-Agent，有些还会校验Cookie里的某些字段。本地浏览器可能带着缓存，服务器上裸请求就被拦了。

解决：把Referer设置成商品详情页的 URL，User-Agent用一个常见的浏览器标识。如果还不行，可以尝试在请求之间加上time.sleep，降低频率。不要试图用伪造的 Cookie 去绕过，稳定性很差，而且容易触发风控。

4.3 价格浮点数比较导致重复入库

现象：数据库里同一个商品同一分钟出现了好几条价格记录，但价格看起来是一样的。

原因：浮点数精度问题。比如99.9和99.90000001在程序里可能被判定为不相等，于是每次都插入新记录。

解决：比较价格时用阈值判断，比如abs(a - b) > 0.001才认为价格变了。或者在存库之前，先把价格四舍五入到两位小数，再进行比较。

4.4 监控任务被系统休眠或网络波动打断

现象：晚上睡觉前还在跑，早上起来发现监控停了几个小时。

原因：如果是跑在本地电脑上，系统休眠会导致进程挂起。如果是跑在服务器上，网络波动可能导致请求超时，而代码里没有做重试。

解决：给请求加上重试机制，比如用requests的Retry适配器，设置重试次数和退避策略。如果是本地跑，建议放到服务器或者树莓派这类常开设备上。另外，加一个心跳检测，每隔一段时间往日志里写一条记录，方便判断任务是否存活。

4.5 商品下架或链接失效导致程序报错退出

现象：监控列表里某个商品被下架了，程序直接抛异常终止，后面的商品也没监控到。

原因：代码里没有对异常情况做处理，一个请求失败就导致整个循环中断。

解决：把每个商品的抓取逻辑包在try...except里，单个失败不影响其他商品。同时记录失败日志，方便后续排查。如果某个商品连续多次失败，可以把它标记为“失效”，暂时跳过，避免浪费请求次数。

5. 让监控真正有用：告警触发与价格走势可视化

监控跑通了，数据也存下来了，但如果价格变了没人知道，那这套系统就只是个数据收集器。最后一章讲两个让监控真正产生价值的点：怎么把价格变化推送到你面前，以及怎么用一张图看懂价格走势。

5.1 用邮件或 Webhook 做价格变化通知

最简单的告警方式是邮件。Python 自带的smtplib就能发，不需要额外装库。下面是一个发送价格变化邮件的示例：

import smtplib from email.mime.text import MIMEText from email.header import Header def send_alert(sku_id, product_name, old_price, new_price): # 邮件服务器配置，以常见邮箱为例 smtp_server = "smtp.example.com" smtp_port = 587 sender = "your_email@example.com" password = "your_auth_code" # 注意不是登录密码，是授权码 receiver = "receiver@example.com" subject = f"价格变动提醒: {product_name}" content = f""" 商品ID: {sku_id} 商品名称: {product_name} 原价: {old_price} 现价: {new_price} 降价幅度: {old_price - new_price:.2f} """ msg = MIMEText(content, "plain", "utf-8") msg["Subject"] = Header(subject, "utf-8") msg["From"] = sender msg["To"] = receiver try: server = smtplib.SMTP(smtp_server, smtp_port) server.starttls() server.login(sender, password) server.sendmail(sender, [receiver], msg.as_string()) server.quit() print("告警邮件已发送") except Exception as e: print("邮件发送失败:", e)

参数说明：smtp_server和smtp_port根据你用的邮箱服务商来填，password通常是授权码而不是登录密码，这个在邮箱设置里可以生成。如果你不想用邮件，也可以换成 Webhook，比如把消息推送到某个群机器人，原理是一样的，只是把发送方式换成 HTTP POST。

5.2 用 matplotlib 画一张价格走势图

数据存下来之后，画图是最直观的。下面这段代码用matplotlib画某个商品最近 30 天的价格走势：

import sqlite3 import matplotlib.pyplot as plt import matplotlib.dates as mdates from datetime import datetime def plot_price_trend(sku_id): conn = sqlite3.connect("price_monitor.db") cursor = conn.cursor() cursor.execute( """SELECT price, recorded_at FROM price_history WHERE sku_id = ? AND recorded_at >= datetime('now', '-30 days') ORDER BY recorded_at ASC""", (sku_id,) ) rows = cursor.fetchall() conn.close() if not rows: print("没有足够的数据来画图") return prices = [row[0] for row in rows] # 将时间字符串转换为 datetime 对象 dates = [datetime.strptime(row[1], "%Y-%m-%d %H:%M:%S") for row in rows] plt.figure(figsize=(12, 5)) plt.plot(dates, prices, marker="o", linestyle="-", color="b", markersize=3) plt.title(f"商品 {sku_id} 最近 30 天价格走势") plt.xlabel("时间") plt.ylabel("价格 (元)") plt.gca().xaxis.set_major_formatter(mdates.DateFormatter("%m-%d")) plt.gca().xaxis.set_major_locator(mdates.DayLocator(interval=3)) plt.gcf().autofmt_xdate() plt.grid(True, alpha=0.3) plt.tight_layout() plt.savefig(f"price_trend_{sku_id}.png", dpi=150) plt.show() print(f"图表已保存为 price_trend_{sku_id}.png")

这段代码的逻辑是先从数据库里查出最近 30 天的价格记录，然后用matplotlib画折线图。mdates.DateFormatter用来控制 X 轴的时间显示格式，DayLocator设置每隔 3 天显示一个刻度，避免时间标签太密集。保存图片的时候用dpi=150，清晰度足够发到手机上看。