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 classprice idjd-price2999.00/span这样的结构另一个是划线价也就是被划掉的参考价官方语义叫市场价。做监控时只关心京东价但两个值建议同时记录因为市场价本身就是活动强度的参照。老手拿到一个 zip 源码的第一件事就是搜索源码里是否还有以下几种正则特征re.search(ridjd-price[^]*([\d.]), html) re.search(rclassp-price[^]*\s*([\d.]), html) re.search(rprice:([\d.]), html)第二行匹配的是旧版页面结构第一行近几年更常见。一个值得注意的细节是价格标签不一定总是直接渲染在 HTML 里。当请求频率偏高或 Cookie 缺失时京东可能延迟加载价格部分此时用正则提取得到的是空字符串。所以抓取函数不能只写一种解析方式至少要有三级降级链。2.2 三种价格获取方式的优先级与失效判断常见做法是按下述顺序尝试优先级数据来源请求地址特点与失效场景1PC 详情页 HTMLhttps://item.jd.com/{sku}.html结构稳定但可能延迟加载价格2移动端详情页 JSONhttps://item.m.jd.com/product/{sku}.htmlPC 页解析失败时兜底返回字段为price:2999.003遗留价格接口https://p.3.cn/prices/mgets?skuIdsJ_{sku}曾经返回[{id:J_100012345,p:2999.00,m:4699.00}]现在经常 403 或返回空为什么第三级只能当兜底这个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/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, }) if cookie_str: session.headers[Cookie] cookie_str return sessionbuild_session里的 Cookie 参数是可选的。如果从浏览器开发者工具复制出一整段 Cookie价格解析的成功率会明显上升尤其是对大额自营商品。没有 Cookie 也能跑只是更容易触发风控。接下来是真正的抓取函数def fetch_price(sku: str, session: requests.Session): # 第一级PC 详情页 url fhttps://item.jd.com/{sku}.html resp session.get(url, timeout10) m re.search(ridjd-price[^]*\s*([\d.])\s*, resp.text) if m and float(m.group(1)) 0: return float(m.group(1)), desktop # 第二级移动端页面 JSON mobile_url fhttps://item.m.jd.com/product/{sku}.html mresp session.get(mobile_url, timeout10) m2 re.search(rprice\s*:\s*([\d.]), mresp.text) if m2 and float(m2.group(1)) 0: return float(m2.group(1)), mobile # 第三级遗留价格接口 p3_url fhttps://p.3.cn/prices/mgets?skuIdsJ_{sku} p3 session.get(p3_url, timeout8) 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, encodingutf-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 用 CrontabWindows 用计划任务脚本本身没有守护进程的必要crontab 是最好的载体。采集频率建议 10 到 30 分钟一次不要低于 5 分钟。京东对价格有缓存5 分钟以内的频繁请求拿到的几乎是同一份数据只会增加被风控的概率。*/10 * * * * cd /opt/jd_monitor /usr/bin/python3 monitor.py logs/monitor.log 21注意三个细节。第一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:004.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 xxxqq.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, timeout10) as smtp: smtp.login(FROM_ADDR, AUTH_CODE) smtp.send_message(msg)Server酱是目前免费推送里配置成本最低的方案绑定微信后直接调用 HTTP 接口def push_serverchan(title, content): key SCT你的key url fhttps://sctapi.ftqq.com/{key}.send resp requests.post(url, data{title: title, desp: content}, timeout10) 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(rwindow\.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 sku100012345, conn ) df[crawled_at] pd.to_datetime(df[crawled_at]) df.plot(xcrawled_at, yprice, markero, figsize(10, 4)) plt.ylabel(价格元) plt.xlabel(采样时间) plt.title(近一个月价格走势) plt.savefig(price_trend.png, dpi150)把这段画图逻辑单独存成trend.py再用一条 cron 每天凌晨跑一次就能生成每日趋势图。如果想找历史最低价把 SQL 里的crawled_at过滤条件改成近 90 天再对price列做min()聚合即可。整条链路到这里已经闭环采集、去抖、告警、落库、可视化每一层都能独立排查问题。本文还有配套的精品资源点击获取