Python京东价格监控系统：爬虫、反爬与降价提醒实战

简介：这是一份基于Python的京东价格监控系统完整源码，面向价格监控、自动提醒等实际场景，适合具备基础爬虫知识、希望搭建完整项目的开发者和学习者。资源共23个文件，以10个py脚本为主干，分别承担任务调度、页面爬取、邮件发送、数据库操作与代理配置等职责；另含7张效果示意图、2份Markdown说明文档、2个TXT文件以及conf和license文件，压缩包仅506KB，目录清晰，便于按需查阅或二次开发。功能上，支持用户设置商品ID与预期价格，价格低于预期即自动邮件提醒；也支持品类订阅，当同类商品降价幅度超过7折时统一发送通知。爬取端提供Requests静态请求、Selenium动态渲染及Js接口三种方式，数据存储兼容Sqlite与Mysql，并集成免费或自定义代理池以降低IP封禁风险。已有64人学习/下载，配套README与效果图可帮助快速跑通监控通知流程，适合课程设计、毕业设计或电商数据采集入门实践。

1. 价格监控系统的核心不是“抓价格”，而是判断“什么时候不该提醒”

从网盘或开源仓库里拿到一份“基于Python的京东价格监控系统.zip”，解压装好依赖后，第一次跑通常卡在两个地方：要么请求被京东反爬拦截，要么收到几十条 0.01 到 0.5 元的“降价提醒”。这个标题背后真正要解决的问题不是写爬虫抓价格，而是几条链路叠加后的可靠性问题：价格从哪个接口拿、拿回来怎么判断真降假降、脚本扔到服务器上能不能持续跑几个月。这个源码对应的常见角色有两种，一是蹲自营商品好价的个人买家，二是做竞品价格分析或比价数据的从业者。后者更关心价格历史采样是否连续、入库是否完整、告警是否可追溯。下面这套方案按一线工程习惯来搭：详情页 HTML 为主，移动端 JSON 兜底，老接口做最后补充，SQLite 落库加阈值去抖，再挂通知。

2. 京东价格从哪里来：页面 DOM、移动端 JSON 与遗留价格接口

2.1 先认清详情页里的两个价格

打开任意一个京东自营商品页，例如https://item.jd.com/100012345.html，页面上至少存在两个价格：一个是“京东价”，通常被渲染成<span class="price" id="jd-price">2999.00</span>这样的结构；另一个是划线价，也就是被划掉的参考价，官方语义叫市场价。做监控时只关心京东价，但两个值建议同时记录，因为市场价本身就是活动强度的参照。

老手拿到一个 zip 源码的第一件事，就是搜索源码里是否还有以下几种正则特征：

re.search(r'id="jd-price"[^>]*>([\d.]+)<', html) re.search(r'class="p-price"[^>]*>\s*([\d.]+)', html) re.search(r'"price":"([\d.]+)"', html)

第二行匹配的是旧版页面结构，第一行近几年更常见。一个值得注意的细节是：价格标签不一定总是直接渲染在 HTML 里。当请求频率偏高或 Cookie 缺失时，京东可能延迟加载价格部分，此时用正则提取得到的是空字符串。所以抓取函数不能只写一种解析方式，至少要有三级降级链。

2.2 三种价格获取方式的优先级与失效判断

常见做法是按下述顺序尝试：

为什么第三级只能当兜底？这个p.3.cn接口在 2020 年前后是公开且稳定的，很多老源码直接拿它做主力。后来京东给这个接口加了签名校验和请求频率限制，未带签名的请求经常返回{"error":"..."}或者直接空数组。如果你下载的源码里主函数只调用了这一个接口，那么大概率第一次运行就会失败。拿到源码后先确认主请求函数里有没有对 403 和空数组做异常处理，没有的话要自己补上。

2.3 价格忽高忽低的真实原因

监控跑起来后，另一个常见现象是同一件商品每五分钟一个价，且变化毫无规律。这里有三层原因。第一，京东自营价格按收货地区分配，同一个 SKU 在不同省份仓库下展示价可以不同；第二，详情页价格存在短暂缓存，当你用同一个出口 IP 高频刷新时，看到的多半是上一次的结果；第三，部分商品价格字段在未加载完成时返回-1.00或0.00，这类脏数据如果不做过滤，会被直接写进数据库，导致告警误报。

所以代码里要写一条硬性约束：只有价格大于 0 的响应才允许入库。抓取函数返回值也建议带上数据来源标识，这样后续排查“这次价格为什么异常”时，可以直接看库里source字段判断是哪一级兜底拿到的数据。

3. 用 Python 和 SQLite 实现一个最小可运行的价格监控脚本

3.1 抓取函数的三级降级链

下面这段代码是这一整套系统的核心，直接保存为monitor.py可以独立运行。先引入依赖并定义请求头和会话：

import re import json import sqlite3 import time import random import datetime import requests UA_POOL = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Safari/605.1.15", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36", ] def build_session(cookie_str: str = ""): session = requests.Session() session.headers.update({ "User-Agent": random.choice(UA_POOL), "Referer": "https://www.jd.com/", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", }) if cookie_str: session.headers["Cookie"] = cookie_str return session

build_session里的 Cookie 参数是可选的。如果从浏览器开发者工具复制出一整段 Cookie，价格解析的成功率会明显上升，尤其是对大额自营商品。没有 Cookie 也能跑，只是更容易触发风控。

接下来是真正的抓取函数：

def fetch_price(sku: str, session: requests.Session): # 第一级：PC 详情页 url = f"https://item.jd.com/{sku}.html" resp = session.get(url, timeout=10) m = re.search(r'id="jd-price"[^>]*>\s*([\d.]+)\s*<', resp.text) if m and float(m.group(1)) > 0: return float(m.group(1)), "desktop" # 第二级：移动端页面 JSON mobile_url = f"https://item.m.jd.com/product/{sku}.html" mresp = session.get(mobile_url, timeout=10) m2 = re.search(r'"price"\s*:\s*"([\d.]+)"', mresp.text) if m2 and float(m2.group(1)) > 0: return float(m2.group(1)), "mobile" # 第三级：遗留价格接口 p3_url = f"https://p.3.cn/prices/mgets?skuIds=J_{sku}" p3 = session.get(p3_url, timeout=8) if p3.ok and p3.text.strip().startswith("["): arr = p3.json() if arr and float(arr[0].get("p", 0)) > 0: return float(arr[0]["p"]), "p3" return None, "none"

这段代码有三个关键点。第一，每个解析分支都加了> 0判断，直接过滤掉-1.00和0.00这类脏价。第二，三级降级的顺序是“页面优先、接口兜底”，因为页面结构再丑返回的数据也相对可信，p.3.cn既无签名又受频率限制，放在最后能减少触发风控的次数。第三，返回值的第二个元素标记了数据来源，这个信息写入数据库后，后续做数据质量分析会非常有用。

3.2 SQLite 表结构与去抖

价格数据量不大，单机场景 SQLite 远比 MySQL 合适。建表语句如下：

CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT NOT NULL, title TEXT, price REAL NOT NULL, market_price REAL, source TEXT, crawled_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_sku_time ON price_history(sku, crawled_at DESC);

这里把price和market_price分两列存，不要合并成一个字符串。后期画趋势图、算降价幅度时，单独一列可以直接进matplotlib或 pandas，不需要再解析字符串。crawled_at使用 SQLite 默认的当前时间，写入时不需要 Python 端传当前时间。

入库函数和“上一次价格”查询函数：

def insert_price(conn, sku, title, price, market_price, source): conn.execute( "INSERT INTO price_history (sku, title, price, market_price, source) VALUES (?, ?, ?, ?, ?)", (sku, title, price, market_price, source) ) conn.commit() def get_last_valid_price(conn, sku): row = conn.execute( "SELECT price FROM price_history WHERE sku=? AND price > 0 ORDER BY id DESC LIMIT 1", (sku,) ).fetchone() return row[0] if row else None

去抖的核心逻辑在查询这条 SQL 里：只取上一次有效价格，与本次价格做差值。如果差值的绝对值小于预设阈值，不写入告警队列，但仍然写入数据库。历史数据完整性比告警本身更重要。

3.3 降价到目标价才推送

很多源码项目把“每次价格变化都通知”做成默认行为，这是最典型的错误设计。京东价格在一天内可能出现多次 0.1 元级波动，每次变化都推送只会让人关闭通知。按从业习惯，告警条件通常是两个：价格跌破心理价位，或者单次降幅超过阈值。

def check_and_notify(conn, sku, title, price, target_price, threshold): last_price = get_last_valid_price(conn, sku) if last_price is None: return if price <= target_price: send_notification(title, sku, price, reason="跌破目标价") elif last_price - price >= threshold: send_notification(title, sku, price, reason="短期降价")

target_price是用户心理价，threshold是去抖阈值。两个条件互斥，避免同一个商品短时间内被推送两次。这里的send_notification函数留在下一节实现，先用print占位也可以跑通。

4. 部署到服务器：配置文件、定时任务与双通道通知

4.1 用 JSON 配置代替硬编码

源码项目里最烦的是把 SKU 和告警价全部写在代码中间。哪怕脚本只跑一次，也建议把监控项放到外部配置文件中，常见命名是config.json：

[ { "sku": "100012345", "title": "某品牌手机", "target_price": 3299, "threshold": 20 }, { "sku": "100678910", "title": "某型号显示器", "target_price": 1499, "threshold": 10 } ]

threshold的单位是元，表示“只有价格下降超过该数值才提醒”。如果把 20 改成 0.1，基本等于每次变动都提醒，不要这么做。加载配置并串行执行的主循环：

import json def main(): with open("config.json", "r", encoding="utf-8") as f: items = json.load(f) session = build_session() conn = sqlite3.connect("jd_price.db") for item in items: price, source = fetch_price(item["sku"], session) if price is None: print(f"[{datetime.datetime.now()}] {item['sku']} 抓取失败") time.sleep(random.randint(3, 6)) continue insert_price( conn, item["sku"], item["title"], price, None, source ) check_and_notify( conn, item["sku"], item["title"], price, item["target_price"], item["threshold"] ) time.sleep(random.randint(2, 5)) conn.close() if __name__ == "__main__": main()

主循环刻意写成单线程串行，原因有两点。第一，京东对同一出口 IP 的并发非常敏感，开ThreadPoolExecutor同时请求 20 个 SKU，大概率触发验证码；第二，监控场景对时效性的要求是分钟级，串行执行 50 个 SKU 也就几分钟，完全够用。

4.2 定时任务：Linux 用 Crontab，Windows 用计划任务

脚本本身没有守护进程的必要，crontab 是最好的载体。采集频率建议 10 到 30 分钟一次，不要低于 5 分钟。京东对价格有缓存，5 分钟以内的频繁请求拿到的几乎是同一份数据，只会增加被风控的概率。

*/10 * * * * cd /opt/jd_monitor && /usr/bin/python3 monitor.py >> logs/monitor.log 2>&1

注意三个细节。第一，cd /opt/jd_monitor必须写，否则config.json和数据库文件的相对路径全部失效。第二，/usr/bin/python3是绝对路径，crontab 环境下的 PATH 不会自动找到 Python。第三，logs/目录要先建好，否则 cron 日志也会静默失败。

Windows 服务器可以用计划任务，命令如下：

schtasks /create /tn "jd_price_monitor" /tr "C:\Python312\python.exe D:\jd_monitor\monitor.py" /sc minute /mo 10 /st 08:00

4.3 通知通道：邮件与 Server酱双保险

通知至少做两个通道，因为任何单一通道都有失败的可能。先写邮件，使用 QQ 邮箱 SMTP，授权码不是登录密码，需要到邮箱设置里单独开启：

import smtplib from email.mime.text import MIMEText from email.header import Header SMTP_HOST = "smtp.qq.com" SMTP_PORT = 465 FROM_ADDR = "xxx@qq.com" AUTH_CODE = "替换为授权码" TO_ADDR = "收件人邮箱" def send_mail(subject, content): msg = MIMEText(content, "plain", "utf-8") msg["Subject"] = Header(subject, "utf-8") msg["From"] = FROM_ADDR msg["To"] = TO_ADDR with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=10) as smtp: smtp.login(FROM_ADDR, AUTH_CODE) smtp.send_message(msg)

Server酱是目前免费推送里配置成本最低的方案，绑定微信后直接调用 HTTP 接口：

def push_serverchan(title, content): key = "SCT你的key" url = f"https://sctapi.ftqq.com/{key}.send" resp = requests.post(url, data={"title": title, "desp": content}, timeout=10) if resp.status_code != 200: print("serverchan push failed:", resp.text)

send_notification里同时调用两个通道：

def send_notification(title, sku, price, reason): content = f"{title} 当前价格 {price}，触发条件：{reason}" print(content) try: send_mail("京东降价提醒", content) except Exception as e: print("mail failed:", e) push_serverchan("京东降价提醒", content)

5. 验证码、105 页面与价格趋势图：把源码跑成长效工具

5.1 识别触发了反爬而不是一直暴力重试

当抓取频率过高时，京东给出的响应通常不是简单的 HTTP 状态码，而是返回一个带验证码的页面，页面 URL 里会出现verify关键字，或者 HTML 中明显包含“拖动滑块”“安全验证”一类文案。此时源码要做的是识别并冷却，而不是继续重试。在fetch_price的 PC 详情页分支里加一段判断：

if "verify" in resp.url.lower() or "滑块" in resp.text: return None, "blocked"

拿到blocked后，主循环最好对该 SKU 做 30 分钟冷却，冷却期间只记录日志，不再发请求。这是合规且有效的处理方式，比任何绕过方案都省心。

5.2 页面结构改版后的应对方向

京东改版频率不高，但每次改版都会让正则失效。真正可靠的做法是优先从页面内嵌的 JSON 里取数，移动端页面里经常存在window.pageConfig或类似结构，价格和库存都在这个全局变量里。解析方式为：

m = re.search(r'window\.pageConfig\s*=\s*(\{.*?\});', mresp.text) if m: data = json.loads(m.group(1)) price = data.get("item", {}).get("price")

这类结构解析的代码在 2024 年后的项目里更常见。如果以后正则失效，优先检查移动端 JSON 里字段是否改了名，而不是反过来去抓接口。

5.3 历史价格趋势图

去抖和告警都完成后，这套系统已经可以长时间运行。此时最有价值的增值功能是价格趋势可视化。基于 SQLite 里的历史数据，可以用 pandas 加 matplotlib 画 30 天曲线：

import sqlite3 import pandas as pd import matplotlib.pyplot as plt plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False conn = sqlite3.connect("jd_price.db") df = pd.read_sql_query( "SELECT crawled_at, price FROM price_history WHERE sku='100012345'", conn ) df["crawled_at"] = pd.to_datetime(df["crawled_at"]) df.plot(x="crawled_at", y="price", marker="o", figsize=(10, 4)) plt.ylabel("价格（元）") plt.xlabel("采样时间") plt.title("近一个月价格走势") plt.savefig("price_trend.png", dpi=150)

把这段画图逻辑单独存成trend.py，再用一条 cron 每天凌晨跑一次，就能生成每日趋势图。如果想找历史最低价，把 SQL 里的crawled_at过滤条件改成近 90 天，再对price列做min()聚合即可。整条链路到这里已经闭环：采集、去抖、告警、落库、可视化，每一层都能独立排查问题。

本文还有配套的精品资源，点击获取