简介这是一套面向Python爬虫与自动化运维学习者的京东价格监控系统源码适合希望掌握电商数据采集、定时监控与邮件告警的开发者练手或二次开发。项目通过Requests与Selenium双方案抓取商品价格支持自定义商品ID与预期价格、品类订阅降价提醒、代理池防封禁并可用Sqlite或Mysql存储商品与用户信息价格低于预期时自动发邮件通知。压缩包共23个文件约506KB包含10个py核心脚本、7张png运行截图、2个txt说明、2个md文档及conf、license等配置许可文件结构清晰覆盖爬虫、数据库连接、邮件发送、代理与日志等模块。已有66人学习下载。读者可从中获得完整的监控系统实现思路、Selenium与Js接口两种爬取方式对比、代理池与邮件提醒的落地代码以及数据库建表与配置示例便于快速搭建自己的价格监控工具。1. 从一份京东价格监控源码说起它到底能帮你盯住什么电商大促前夜运营同事最怕的不是没货而是竞品悄悄改了价自己还蒙在鼓里。一份基于 Python 的京东价格监控系统源码解决的正是这件事定时抓取指定商品的价格、促销信息和库存状态落库留存价格异动时触发提醒。它适合三类人——做竞品分析的小团队、想练手爬虫与定时任务的 Python 学习者、以及需要给自家商品做价格留痕的运营。很多人拿到源码第一反应是「跑起来看看」但真正决定这套系统能不能长期用的不是代码本身而是采集策略、存储结构和调度方式这三件事。下面我按自己落地过的思路把这份源码拆开讲透让你既能复现也知道边界在哪。2. 拆解京东价格监控系统的采集层请求、解析与反爬边界2.1 为什么价格监控的第一道坎是「拿到稳定数据」京东商品页的价格并不是写死在 HTML 里的静态文本。早期页面价格直接渲染在 DOM 中用 requests 加 BeautifulSoup 就能拿到现在多数价格通过异步接口返回页面初始 HTML 里往往只有占位符。这意味着如果你照着老教程写soup.find(span, class_price)大概率拿到空值这就是很多人第一次跑源码就翻车的地方。常见做法是两条路一是直接请求价格接口二是用浏览器自动化渲染后再取。前者轻量、快但接口参数可能带签名后者稳、接近真实用户但资源占用高。我一般先用接口方案跑不通再退回自动化。判断标准很简单用 curl 或 requests 请求一次看返回体里有没有你要的价格字段。有就走接口没有再考虑渲染。这里要提醒一句采集频率是玄学也是红线。你把间隔设成 1 秒短时间内请求量上去了轻则被限流返回验证页重则 IP 被临时封禁。我的血泪经验是单商品轮询间隔不低于 30 秒批量商品用队列错峰别让所有请求在同一秒打出去。2.2 用 requests 抓价格接口的最小可跑代码下面这段是采集层的骨架重点看请求头构造和异常兜底而不是照抄 URL。京东的接口地址会变参数也可能调整所以我把可变部分抽成配置方便你替换。import requests import time import random # 请求头是关键缺少 Referer 和 UA 很容易被识别为脚本 HEADERS { User-Agent: (Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36), Referer: https://item.jd.com/, Accept: application/json, text/plain, */*, } def fetch_price(sku_id, retry3): 按 SKU 抓取价格失败重试并退避 # 接口地址以实际抓包为准这里只做结构示意 url fhttps://p.3.cn/prices/mgets?skuIdsJ_{sku_id} for i in range(retry): try: resp requests.get(url, headersHEADERS, timeout8) if resp.status_code 200: data resp.json() # 返回是列表取第一个元素的价格字段 return float(data[0].get(p, 0)) except (requests.RequestException, ValueError, KeyError): # 网络抖动或返回体异常退避后重试 time.sleep(2 ** i random.random()) return None if __name__ __main__: price fetch_price(100012043978) print(当前价格:, price)逻辑说明fetch_price接收 SKU 编号拼出价格接口地址带请求头发起 GET。返回体是 JSON 数组价格在p字段里。重试采用指数退避第 i 次失败等2^i秒再加随机抖动避免多个任务同时重试形成脉冲。参数说明timeout8是单次请求上限太短会误判超时太长会拖慢整批任务retry3是重试次数再多意义不大因为连续失败通常意味着接口或 IP 出了问题Referer必须带否则部分接口直接拒绝。SKU 编号就是商品链接里那串数字比如item.jd.com/100012043978.html中的100012043978。2.3 解析层要处理的三种返回形态采集回来的数据不一定都是干净的价格数字。实际会遇到三种形态正常 JSON、带验证的 HTML 页面、以及空响应。源码里如果只处理第一种跑几天就会因为后两种崩溃。我的做法是在解析层加一层类型判断先看Content-Type是 JSON 就解析字段是 HTML 就记录告警并跳过空响应直接标记为失败。这一步的价值在于让系统「知道自己没拿到数据」而不是把 None 当成 0 存进数据库。很多价格曲线突然掉到 0就是解析层没做区分导致的。你可以加一个状态字段把每次采集标记为 success、blocked、empty 三种后续排查一眼就能看出是接口变了还是被限流了。3. 存储与调度把价格数据变成能查的曲线3.1 表结构设计为什么不能只存一个价格字段只存价格你只能看到「现在多少钱」看不到「什么时候变的、变了多少、是不是促销」。我一般会设计三张表商品表存 SKU、名称、目标价价格记录表存每次采集的时间戳和价格告警表存触发记录。价格记录表是核心字段至少包括sku_id、price、captured_at、status。CREATE TABLE price_record ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku_id TEXT NOT NULL, price REAL, status TEXT DEFAULT success, captured_at DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 按商品和时间建索引查历史曲线时不用全表扫 CREATE INDEX idx_sku_time ON price_record (sku_id, captured_at);逻辑说明price允许为空因为采集失败时不该伪造一个 0status记录采集状态方便区分「真降价」和「没抓到」索引建在sku_id和captured_at组合上因为最常见的查询是「某商品最近 7 天的价格」。参数说明SQLite 适合单机和小规模几百个 SKU 完全够用如果 SKU 上千、采集频率高换成 MySQL 或 PostgreSQL 更稳。时间字段用DATETIME而不是字符串方便做区间查询和聚合。3.2 用 APScheduler 做定时任务别用 while True新手最容易写的就是while True: fetch(); time.sleep(60)。这种写法的问题在于任务执行时间不确定如果一次采集花了 20 秒实际间隔就变成 80 秒而且没有失败隔离一个商品报错可能拖垮整轮。常见做法是用 APScheduler把每个商品或每批商品注册成独立任务。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger scheduler BlockingScheduler() def job_fetch_all(): 批量采集入口内部做错峰 sku_list load_sku_list() # 从数据库读要监控的商品 for sku in sku_list: price fetch_price(sku) save_price(sku, price) time.sleep(random.uniform(1, 3)) # 商品之间错峰降低被封风险 # 每 5 分钟跑一轮具体间隔按业务定 scheduler.add_job(job_fetch_all, IntervalTrigger(minutes5), idfetch_all, max_instances1) scheduler.start()逻辑说明job_fetch_all是整批采集的入口内部对每个 SKU 之间加 1 到 3 秒随机间隔避免请求扎堆。max_instances1保证同一任务不会重叠执行防止上一轮没跑完下一轮又启动。参数说明minutes5是轮询周期价格监控不需要秒级5 到 30 分钟都合理random.uniform(1, 3)是错峰区间太小起不到错峰作用太大拖长整轮时间。如果你的 SKU 很多可以把批次拆开用多个 job 分担。3.3 价格异动判断阈值怎么设才不误报判断降价不能只看「比上次低」。如果上次采集失败存了空值这次拿到正常价格差值会很大直接触发误报。正确做法是只在两次都成功采集的前提下比较并且设置一个最小变动幅度比如低于 1% 的变化忽略。def check_price_drop(sku_id, new_price, threshold0.01): 对比最近一次成功记录判断是否触发降价告警 last get_last_success_price(sku_id) # 只取 statussuccess 的记录 if last is None or new_price is None: return False drop_ratio (last - new_price) / last if drop_ratio threshold: send_alert(sku_id, last, new_price) return True return False逻辑说明get_last_success_price只查成功记录跳过失败采集避免空值干扰。drop_ratio是降价比例只有超过阈值才告警。参数说明threshold0.01表示降价 1% 才提醒你可以按品类调整快消品可以设 0.5%大家电设 2% 更合理。告警渠道常见的是邮件、企业微信机器人或钉钉机器人源码里一般会留一个send_alert的空实现你按自己的渠道补上即可。4. 避坑与排查价格监控系统最常见的 5 个翻车现场4.1 现象价格全是 None 或 0原因接口地址失效或请求头缺少关键字段被拒绝。京东的价格接口会调整老源码里的 URL 可能已经不通。解决先用浏览器开发者工具抓一次真实请求对比 URL、请求头和参数。重点检查Referer和User-Agent缺一个都可能返回空。确认接口通了再改代码别在解析逻辑上瞎调。4.2 现象跑一段时间后开始返回验证页原因请求频率过高或同一 IP 短时间内请求量太大触发限流。解决拉长轮询间隔商品之间加随机延迟必要时分散到不同时间段。别用多线程猛冲那只会更快被封。我的习惯是单 IP 每分钟不超过 10 次请求超过就拆任务。4.3 现象数据库里价格曲线出现断崖式下跌原因采集失败时把 None 或 0 存进了价格字段前端画图时当成真实价格。解决价格字段允许为空失败时存 NULL 并标记 status。查询曲线时过滤掉非成功记录或者用前一个有效值填充。这个坑我在早期项目里踩过排查了半天才发现是空值被当成了 0。4.4 现象定时任务跑着跑着不执行了原因任务函数抛异常没被捕获调度器把该任务挂起或者max_instances设置不当导致任务堆积。解决在任务入口包一层 try/except记录异常日志保证单次失败不影响后续调度。同时确认max_instances1避免上一轮没结束下一轮又进来。4.5 现象告警频繁误报运营被骚扰原因阈值设得太低或者没有过滤失败采集导致正常波动也触发提醒。解决提高最小变动幅度只在连续两次成功采集之间比较。还可以加一个冷却时间同一商品 1 小时内只告警一次避免价格反复波动时刷屏。5. 让这套源码真正可用从能跑到能长期维护的进阶技巧把源码跑起来只是起点能不能长期用取决于你有没有给它加上「自检」和「留痕」。我一般会做三件事。第一加一个每日健康检查任务统计当天采集成功率低于 90% 就给自己发提醒这样接口一变你当天就知道而不是等运营来问。第二把采集状态和异常信息写进日志文件按天切分出问题时能回溯到具体是哪次请求、返回了什么。第三给价格记录表加一个定期归档策略超过半年的明细可以聚合或转存避免单表无限膨胀拖慢查询。验证方法也很直接手动改一次数据库里的历史价格看告警会不会触发把接口地址改错看健康检查会不会报警把轮询间隔调到 1 秒观察多久被限流。这三个测试跑一遍你对这套系统的边界就心里有数了。维护项建议频率检查内容采集成功率每天低于 90% 排查接口和请求头日志清理每周保留 30 天避免磁盘占满数据归档每季度半年前明细聚合或转存告警阈值复核每月按品类调整减少误报还有一个容易被忽略的点SKU 会下架。下架后接口返回空或报错如果不处理你的失败率会一直偏高。常见做法是连续 N 次采集失败后把该商品标记为「暂停监控」而不是无限重试。N 取 5 到 10 都行看你的轮询频率。最后说个我自己的习惯每次改完采集逻辑先拿 3 个商品跑一天对比手工查价的结果确认无误再全量放开。价格监控这事数据准比功能多重要得多。希望帮到你。本文还有配套的精品资源点击获取