简介一套基于Python的京东价格监控系统源码面向Python开发者和电商购物人群解决手动比价效率低、无法及时发现降价的问题。系统利用Requests与Selenium自动抓取商品价格支持自定义商品监控与品类订阅当价格低于预期或降价超七折时自动发送邮件提醒同时内置代理池与多种爬取方式可有效降低封禁风险并支持Sqlite/MySQL数据库存储。压缩包内共23个文件包含10个Python脚本、7张效果截图、2份Markdown说明、2个文本文件、1个配置文件和1份许可证整体仅506KB。源码模块划分清晰涵盖数据库初始化、JS接口爬取、Selenium渲染爬取、邮件发送、代理配置等核心功能配套文档和截图便于快速部署与二次开发。已有65人学习下载适合作为爬虫、自动化监控及邮件通知项目的实战参考。1. 京东价格监控系统值得自己用 Python 重写一遍吗盯着一件商品等降价每天手动刷新页面两三回既费时间又容易错过秒杀时段。用 Python 写一套价格监控系统定时抓取商品价格、写入本地 SQLite、行情触发时把提醒推到微信或邮箱整个核心链路不到两百行代码就能打通。对刚开始接触爬虫的人来说这个 zip 源码包是理解 requests、SQLite 和定时任务三者怎么协作的现成教材对已经写过爬虫的工程师来说它提供了一个能快速拆解改造的骨架把抓取、存储、通知、调度四条链路各自独立成模块替换其中任何一环都不影响其他部分。它解决的不是“能不能抢到货”而是把盯价格的重复劳动交给程序让异常波动及时到达你面前。2. 先定技术底座requests、SQLite 与调度框架的选型理由2.1 为什么不用 Scrapy监控场景只需要轻量请求这个系统的请求量级很明确盯 10 个商品每半小时抓一次一天也就 480 次请求而且集中在几个固定 URL 上。这个量级用 Scrapy 属于小题大做——Scrapy 的异步并发、去重队列、下载中间件是为“短时间内批量抓取成千上万页面”设计的配置负担明显高于收益。requests 的同步阻塞在这里反而是优点请求按顺序执行哪一步出错一目了然依赖只有一个库。新手上手这套源码前先把 Python 装好。安装时勾选 Add Python to PATH比事后配环境变量省事得多然后创建虚拟环境装依赖py -m venv venv venv\Scripts\activate pip install requests apscheduler用 vscode 写 Python 的话打开命令面板选择解释器指向项目目录下的 venv这样 requests、apscheduler 这些包都装在项目内部不会污染全局环境以后换机器也能一键重建。提示这套源码只需要 requests、apscheduler 两个第三方库邮件和 SQLite 用的都是 Python 标准库依赖面很小这也是它适合长年挂在服务器上的原因之一。2.2 存储选 SQLite 而非 CSV/MySQL 的两个理由价格历史是典型的“写多读少”数据。CSV 看起来简单但进程被杀时会丢尾部数据两个写入方同时操作时文件锁混乱排查起来玄学感十足。MySQL 对这个数据量是杀鸡用牛刀还得额外维护一个数据库服务。SQLite 单文件、事务完整、查询语法和 MySQL 几乎一致后续真要迁移改连接串就行。建表时把商品 SKU 和检查时间组成联合主键天然去重CREATE TABLE IF NOT EXISTS price_history ( sku TEXT NOT NULL, title TEXT, price REAL NOT NULL, check_time TEXT NOT NULL, PRIMARY KEY (sku, check_time) );联合主键的意义在于同一轮监控里重复抓取同一个商品第二条 INSERT 会直接触发冲突相当于免费获得了去重能力。check_time 用 ISO 格式字符串存储按天分组和区间查询都比时间戳直观SQLite 内置的日期函数也能直接处理。2.3 调度方案对比time.sleep、schedule 与 APScheduler最省事的调度是 while True 加 time.sleep(1800)半小时醒来抓一次。缺点是进程一重启调度状态就归零抓取抛异常时外层没有兜底循环可能直接死掉。schedule 库写法很简洁但任务抛异常会带崩调度线程对无人值守的监控场景不够稳。APScheduler 的 BackgroundScheduler 把调度器和任务执行隔离任务内部再套一层保护壳from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger import logging def safe_job(): try: run_price_check() except Exception: logging.exception(price check job failed) scheduler BackgroundScheduler(timezoneAsia/Shanghai) scheduler.add_job( safe_job, CronTrigger(minute*/30, misfire_grace_time300), idprice_check, coalesceTrue, max_instances1, ) scheduler.start()CronTrigger 的 minute*/30 表示每 30 分钟整点执行misfire_grace_time300 允许任务因为上一轮超时最多迟到 300 秒coalesceTrue 保证错过的任务补跑时只执行一次max_instances1 防止上一轮没跑完、下一轮又启动造成的并发错乱。这四个参数是调度稳定性的大半新手最容易漏掉的是 max_instances。2.4 源码包目录结构解读与 zip 分发要处理的三个细节拿到 zip 后先看目录正常的源码包会按职责拆文件而不是把一个上千行的脚本塞在一起。常见组织方式是这样price_monitor/ ├── config.py # 商品URL、SMTP、webhook 占位符 ├── collector.py # 抓取与页面解析 ├── notifier.py # 邮件与webhook推送 ├── scheduler.py # APScheduler入口 ├── requirements.txt # 依赖锁定 ├── README.md # 运行说明 └── data/ # 运行时生成 price_log.dbconfig.py 单独拆出来是为了分发安全SMTP 授权码、推送 key 都以占位符存在部署时由使用者自己填。requirements.txt 锁定版本避免几个月后依赖升级导致源码跑不通。zip 包在 Windows 和 Linux 之间传递时中文文件名的编码问题经常让人翻车。Windows 默认 GBK 存文件名Linux 下直接解压会看到乱码要这样处理unzip -O gbk price_monitor.zip -d price_monitor如果系统 unzip 不支持 -O 参数就用 Python 的 zipfile 模块解压后重命名。还有一类 zip 提示要密码、但压缩软件里能直接看到文件列表这是 zip 伪加密改一下中央目录的加密标志位即可绕过真正的密码保护不会出现“列表可见”的情况。提示源码包落地前先删掉 __pycache__、venv、.env 这类运行产物别人的本地路径对你没有意义。3. 把价格稳定抓下来请求伪装、价格提取与入库的三段式流水线3.1 HTML 解析与接口直取两条数据来源怎么选抓价格有两条路。第一条直接解析商品页 HTML在 DOM 里找价格节点第二条找页面背后返回 JSON 的接口直接取结构化数据。实际页面 HTML 里的价格经常被拆散存放同一个页面同时存在原价、会员价、满减价位置不同格式也不同接口数据则整齐得多字段名明确解析几乎不费劲。我的默认策略是优先请求页面本身再从页面源码里找内嵌的 JSON 数据。商品页里通常有一段 script 标签保存着商品信息把这段 JSON 提取出来后价格就在其中的某个字段里比在 DOM 树里到处找节点稳定得多。页面拿不到时再退回接口方案两条通道都失败才判定抓取失败。3.2 请求头伪装与 Cookie 处理UA、Referer 与登录态要让服务器把请求当作普通浏览器访问请求头至少带三样User-Agent、Referer、Accept-Language。UA 用最新浏览器的完整字符串Referer 指向商品列表页语言设成 zh-CN。用 requests.Session 保持整个会话后续请求自动沿用 Cookieimport requests session requests.Session() session.headers.update({ 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: https://search.jd.com/, Accept-Language: zh-CN,zh;q0.9, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, }) resp session.get(url, timeout10) resp.encoding utf-8 html resp.texttimeout10 必须写否则网络抖动时 requests 会一直挂着等响应调度任务越积越多最终拖垮整条监控链路。UA 写死一份够用不要追着浏览器版本更新真遇到页面返回异常验证页再考虑更换。部分商品需要登录才能看到真实价格做法是浏览器里手动登录一次把关键 Cookie 复制到配置里session 会自动携带。3.3 价格提取正则与候选字段双保险以及字体反爬的兜底价格在页面里常见格式是“888.00”或“8,888.00”千分位逗号必须优先去掉。写一个提取函数处理两种格式顺便做区间校验import re def extract_price(text: str) - float | None: 从一段文本中提取第一个合理价格失败返回 None text text.replace(,, ) m re.search(r(\d\.\d{2}|\d), text) if not m: return None price float(m.group(1)) if price 1 or price 100_000_000: return None return price正则匹配“整数或小数点后两位的整数”价格合理性检查放在最后一道防线。个别页面会用自定义字体做反爬把数字渲染成特殊字符源码里的“9”在页面上显示成另一个字形。应对方式需要下载字体文件、解析映射表对个人监控来说成本偏高我会直接跳过这类商品并记录日志不值得为单一监控对象花一整个下午。3.4 入库与变化判断SQLite 写入前置检查抓到的价格不能直接往库里塞先与最近一条历史记录比较波动超过 50% 时保留入库但打上异常标记。这样既能应对临时降价也不会因为解析错误污染后续的趋势判断import sqlite3 from datetime import datetime, timezone, timedelta DB_PATH data/price_log.db def save_price(sku: str, title: str, price: float): now datetime.now(timezone(timedelta(hours8))).strftime(%Y-%m-%d %H:%M:%S) with sqlite3.connect(DB_PATH) as conn: conn.execute( INSERT INTO price_history (sku, title, price, check_time) VALUES (?, ?, ?, ?), (sku, title, price, now), ) conn.commit()插入前先 SELECT 最近一条记录做差值计算差值超阈值时在日志里额外输出一行。时区固定为东八区并转成字符串避免部署到跨时区服务器后时间对不上。对个人监控而言这个数据量完全不需要索引一个月几千行全表扫描毫无压力。4. 降价通知不靠盯屏SMTP 邮件与 webhook 推送的落地配置4.1 邮件通知SMTP 配置与提醒内容模板邮件是最不容易被忽略的通知通道适合“睡前统一看一遍”的场景。用 QQ 邮箱或 163 邮箱开启 SMTP 服务后拿到授权码填入配置即可。SSL 连接方式端口固定为 465import smtplib from email.mime.text import MIMEText from email.header import Header def send_mail(subject: str, body: str, to_addr: str): smtp_host smtp.qq.com smtp_port 465 user your_accountqq.com auth_code your_auth_code # 登录密码不行必须用授权码 msg MIMEText(body, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] user msg[To] to_addr with smtplib.SMTP_SSL(smtp_host, smtp_port, timeout15) as server: server.login(user, auth_code) server.send_message(msg)邮件正文至少包含商品名、当前价格、目标阈值、商品链接四个要素标题直接带当前价格比如“某无线耳机已降至 1599关注价 1699”。邮件模板可以按 SKU 分类不同类别用不同的关注价。SMTP 偶尔会因网络超时报错调用处一定要包 try/except邮件失败不能拖垮主流程。4.2 Server 酱与企业微信群机器人把提醒推到手机邮件适合统一查看降价提醒更适合直接推到手机。Server 酱、企业微信群机器人都属于这类 webhook 服务一个 URLPOST 两个参数手机立刻收到。实现逻辑很薄import requests def push_webhook(title: str, desp: str): webhook https://example.webhook/send?keyYOUR_KEY resp requests.post( webhook, json{title: title, desp: desp}, timeout10, ) result resp.json() if result.get(code) ! 0: raise RuntimeError(fpush failed: {result})webhook 地址是敏感配置放在 config.py 里以占位符形式存在。desp 参数支持简单排版可以把商品名、价格、历史最低价、商品链接拼成多行推送可读性比邮件更好。这类推送服务的稳定性完全取决于服务商所以推送失败一定要落日志让邮件通知兜底两条通道搭配使用才可靠。4.3 两条节流策略时间窗口与二次降价百分比比漏发更常见的问题是刷屏。商品价格在目标价附近反复横跳时每 30 分钟推一次一天能收几十条。节流方案不能只限制时间窗口还要判断“价格是否真的更低了”。我的做法是双重判断距离上次通知超过 2 小时同时价格比上次通知时又低了 5% 以上才再次推送。def should_notify(sku: str, price: float, last_notify_price: float | None) - bool: 时间窗口 二次降价百分比双重限制 if last_notify_price is None: return True if price last_notify_price * 0.95: return False return True第一次触发当然立刻推后续推送条件更苛刻只有价格比上次通知更低才推送5% 的阈值能覆盖大部分促销节奏。时间窗口控制提醒密度百分比控制提醒价值两者缺一不可。节流状态存进 SQLite 的 notify_log 表重启进程也不会丢比内存变量可靠得多。整套通知链路在跑真实任务前先手动调用一次确认邮件能收到、webhook 能送达再挂到定时调度上。5. 常见问题与避坑反爬、解析异常与链路静默的排查5.1 现象固定请求头下价格页面偶尔返回 400 或验证页原因通常有两个方向请求频率过高或者 Cookie 过期。监控多个商品时如果所有任务在同一分钟启动短时间对同一页面发起多次请求很容易触发临时风控。Cookie 则会在登录态失效后被服务端主动拒绝。解决办法是打散请求节奏并在失败时停止重试import time import random time.sleep(random.uniform(5, 15)) resp session.get(url, timeout10)sleep 放在 request 之前均匀取 5 到 15 秒的随机数避免固定间隔被识别。同一轮里连续三次拿不到价格就跳过该商品把问题留到日志里不要死循环重试。Cookie 过期时标记“需要更新 Cookie”并暂停该 SKU人工介入处理比反复试探稳妥。5.2 现象解析到的价格比页面显示少一个数量级这类 bug 最阴险程序不报错库里存着错误数据。原因几乎都是截取到了划线价、预售价或者促销标签旁边的旧价格。页面里同时存在多个价格展示正则默认取第一个数字恰好命中不含优惠的原价。解决解析函数里增加候选字段的优先级排序优先找含 price 语义的 JSON 字段其次匹配带货币符号的节点拿到结果后与上次记录对比波动超 50% 就标记异常入库前再做一次区间校验。价格合理性检查要放在统一入口而不是散落在各个解析分支里否则新增商品时容易漏掉。5.3 现象定时任务看着在跑数据库里没有新数据这个问题三分之二出在路径上三分之一出在异常被吞。最典型的场景在项目根目录手动启动一切正常改用系统计划任务或服务方式运行后工作目录变成用户主目录data/price_log.db 的相对路径指向了错误位置SQLite 报错退出。任务内异常如果只 print 到控制台后台运行时输出直接丢弃看起来就像什么都没发生。解决所有路径基于项目根目录做绝对路径拼接日志写到文件而不是控制台import logging logging.basicConfig( filenamelogs/price_monitor.log, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s, )每轮抓取结束输出一行摘要包含成功、失败、异常计数。这样即使看不到控制台日志文件也能完整还原执行过程。如果 24 小时完全没有新日志说明调度入口没触发先查计划任务配置再查脚本是否真的被拉起。5.4 现象zip 源码包解压后无法运行提示缺模块或编码错误两个常见原因。第一Windows 下用记事本改过源码把 UTF-8 文件保存成了 GBKPython 3 默认按 UTF-8 读源码遇到非法字节直接 SyntaxError。第二requirements.txt 与当前 Python 版本不兼容比如老项目里的依赖不支持新解释器。还有一类是 zip 内打包了缓存目录解压后路径混乱间接导致找不到模块。解决源码统一 UTF-8 编码保存确认 Python 版本后按依赖清单安装python -m pip install -r requirements.txt如果在 Windows 命令提示符里执行 pip 提示“不是内部或外部命令”说明安装 Python 时没勾选 Add Python to PATH重新运行安装程序修复即可。遇到 zip 伪加密提示时先确认文件列表是否可见不要被提示误导7-Zip 通常能正常列出内容。5.5 现象抓取和入库都正常但收不到降价通知链路断开的位置通常在推送侧。先查 notify_log 表确认上次通知时间是否被误写为当前时间节流逻辑把后面的通知全部拦截了。再看 webhook 服务返回推送接口升级后返回结构可能变化旧代码里读 code 字段判断成功就会失效。解决写一个独立的测试脚本直接调用推送函数排除主流程干扰推送响应用完之后先 raw 打印一层接口变更后能第一时间发现字段差异。只要抓取链路和通知链路分别都能独立跑通合在一起出问题的概率就很小。6. 收尾技巧用 SQLite 历史价格回放验证监控链路6.1 一条 SQL 查出这周的抓取质量监控跑了一周最该问的问题是“它真的每半小时都上班了吗”。与其相信感觉不如直接看数据库按天分组统计SELECT substr(check_time, 1, 10) AS day, COUNT(*) AS samples, MIN(price) AS min_price, MAX(price) AS max_price FROM price_history WHERE sku 指定商品 GROUP BY day;samples 接近 48说明当天 30 分钟一次的任务正常执行远低于 48那天大概率发生过断档。min_price 和 max_price 差距过大则要对照日志确认是真实降价还是解析异常。用这个结果和浏览器里手动看到的历史价格交叉验证监控链路是否健康一目了然。6.2 用历史最低价反推通知阈值固定阈值容易失灵设太高永远不触发设太低促销来了也静默。更合理的做法是从历史数据里找参考线def lowest_price_last_days(sku: str, days: int 7): with sqlite3.connect(DB_PATH) as conn: rows conn.execute( SELECT price, check_time FROM price_history WHERE sku ? AND check_time datetime(now, ?) ORDER BY check_time, (sku, f-{days} days), ).fetchall() return min(rows, keylambda r: r[0]) if rows else None返回最近七天的最低价格和出现时间通知阈值取它的九折既参考了历史波动又留有触发余地。每次回放顺手看一眼趋势摘要哪天的价格异常、哪个商品整天没数据立刻能定位到链路故障点。我每周五会固定看一眼 price 库就像检查仪表盘一样——这份数据不只在降价时有用更是监控系统自身健康状况的客观凭证。希望这套回放思路能帮你把监控跑得更放心。本文还有配套的精品资源点击获取 Python 写一套价格监控系统：从 requests 到 SQLite 的实战教程
简介一套基于Python的京东价格监控系统源码面向Python开发者和电商购物人群解决手动比价效率低、无法及时发现降价的问题。系统利用Requests与Selenium自动抓取商品价格支持自定义商品监控与品类订阅当价格低于预期或降价超七折时自动发送邮件提醒同时内置代理池与多种爬取方式可有效降低封禁风险并支持Sqlite/MySQL数据库存储。压缩包内共23个文件包含10个Python脚本、7张效果截图、2份Markdown说明、2个文本文件、1个配置文件和1份许可证整体仅506KB。源码模块划分清晰涵盖数据库初始化、JS接口爬取、Selenium渲染爬取、邮件发送、代理配置等核心功能配套文档和截图便于快速部署与二次开发。已有65人学习下载适合作为爬虫、自动化监控及邮件通知项目的实战参考。1. 京东价格监控系统值得自己用 Python 重写一遍吗盯着一件商品等降价每天手动刷新页面两三回既费时间又容易错过秒杀时段。用 Python 写一套价格监控系统定时抓取商品价格、写入本地 SQLite、行情触发时把提醒推到微信或邮箱整个核心链路不到两百行代码就能打通。对刚开始接触爬虫的人来说这个 zip 源码包是理解 requests、SQLite 和定时任务三者怎么协作的现成教材对已经写过爬虫的工程师来说它提供了一个能快速拆解改造的骨架把抓取、存储、通知、调度四条链路各自独立成模块替换其中任何一环都不影响其他部分。它解决的不是“能不能抢到货”而是把盯价格的重复劳动交给程序让异常波动及时到达你面前。2. 先定技术底座requests、SQLite 与调度框架的选型理由2.1 为什么不用 Scrapy监控场景只需要轻量请求这个系统的请求量级很明确盯 10 个商品每半小时抓一次一天也就 480 次请求而且集中在几个固定 URL 上。这个量级用 Scrapy 属于小题大做——Scrapy 的异步并发、去重队列、下载中间件是为“短时间内批量抓取成千上万页面”设计的配置负担明显高于收益。requests 的同步阻塞在这里反而是优点请求按顺序执行哪一步出错一目了然依赖只有一个库。新手上手这套源码前先把 Python 装好。安装时勾选 Add Python to PATH比事后配环境变量省事得多然后创建虚拟环境装依赖py -m venv venv venv\Scripts\activate pip install requests apscheduler用 vscode 写 Python 的话打开命令面板选择解释器指向项目目录下的 venv这样 requests、apscheduler 这些包都装在项目内部不会污染全局环境以后换机器也能一键重建。提示这套源码只需要 requests、apscheduler 两个第三方库邮件和 SQLite 用的都是 Python 标准库依赖面很小这也是它适合长年挂在服务器上的原因之一。2.2 存储选 SQLite 而非 CSV/MySQL 的两个理由价格历史是典型的“写多读少”数据。CSV 看起来简单但进程被杀时会丢尾部数据两个写入方同时操作时文件锁混乱排查起来玄学感十足。MySQL 对这个数据量是杀鸡用牛刀还得额外维护一个数据库服务。SQLite 单文件、事务完整、查询语法和 MySQL 几乎一致后续真要迁移改连接串就行。建表时把商品 SKU 和检查时间组成联合主键天然去重CREATE TABLE IF NOT EXISTS price_history ( sku TEXT NOT NULL, title TEXT, price REAL NOT NULL, check_time TEXT NOT NULL, PRIMARY KEY (sku, check_time) );联合主键的意义在于同一轮监控里重复抓取同一个商品第二条 INSERT 会直接触发冲突相当于免费获得了去重能力。check_time 用 ISO 格式字符串存储按天分组和区间查询都比时间戳直观SQLite 内置的日期函数也能直接处理。2.3 调度方案对比time.sleep、schedule 与 APScheduler最省事的调度是 while True 加 time.sleep(1800)半小时醒来抓一次。缺点是进程一重启调度状态就归零抓取抛异常时外层没有兜底循环可能直接死掉。schedule 库写法很简洁但任务抛异常会带崩调度线程对无人值守的监控场景不够稳。APScheduler 的 BackgroundScheduler 把调度器和任务执行隔离任务内部再套一层保护壳from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger import logging def safe_job(): try: run_price_check() except Exception: logging.exception(price check job failed) scheduler BackgroundScheduler(timezoneAsia/Shanghai) scheduler.add_job( safe_job, CronTrigger(minute*/30, misfire_grace_time300), idprice_check, coalesceTrue, max_instances1, ) scheduler.start()CronTrigger 的 minute*/30 表示每 30 分钟整点执行misfire_grace_time300 允许任务因为上一轮超时最多迟到 300 秒coalesceTrue 保证错过的任务补跑时只执行一次max_instances1 防止上一轮没跑完、下一轮又启动造成的并发错乱。这四个参数是调度稳定性的大半新手最容易漏掉的是 max_instances。2.4 源码包目录结构解读与 zip 分发要处理的三个细节拿到 zip 后先看目录正常的源码包会按职责拆文件而不是把一个上千行的脚本塞在一起。常见组织方式是这样price_monitor/ ├── config.py # 商品URL、SMTP、webhook 占位符 ├── collector.py # 抓取与页面解析 ├── notifier.py # 邮件与webhook推送 ├── scheduler.py # APScheduler入口 ├── requirements.txt # 依赖锁定 ├── README.md # 运行说明 └── data/ # 运行时生成 price_log.dbconfig.py 单独拆出来是为了分发安全SMTP 授权码、推送 key 都以占位符存在部署时由使用者自己填。requirements.txt 锁定版本避免几个月后依赖升级导致源码跑不通。zip 包在 Windows 和 Linux 之间传递时中文文件名的编码问题经常让人翻车。Windows 默认 GBK 存文件名Linux 下直接解压会看到乱码要这样处理unzip -O gbk price_monitor.zip -d price_monitor如果系统 unzip 不支持 -O 参数就用 Python 的 zipfile 模块解压后重命名。还有一类 zip 提示要密码、但压缩软件里能直接看到文件列表这是 zip 伪加密改一下中央目录的加密标志位即可绕过真正的密码保护不会出现“列表可见”的情况。提示源码包落地前先删掉 __pycache__、venv、.env 这类运行产物别人的本地路径对你没有意义。3. 把价格稳定抓下来请求伪装、价格提取与入库的三段式流水线3.1 HTML 解析与接口直取两条数据来源怎么选抓价格有两条路。第一条直接解析商品页 HTML在 DOM 里找价格节点第二条找页面背后返回 JSON 的接口直接取结构化数据。实际页面 HTML 里的价格经常被拆散存放同一个页面同时存在原价、会员价、满减价位置不同格式也不同接口数据则整齐得多字段名明确解析几乎不费劲。我的默认策略是优先请求页面本身再从页面源码里找内嵌的 JSON 数据。商品页里通常有一段 script 标签保存着商品信息把这段 JSON 提取出来后价格就在其中的某个字段里比在 DOM 树里到处找节点稳定得多。页面拿不到时再退回接口方案两条通道都失败才判定抓取失败。3.2 请求头伪装与 Cookie 处理UA、Referer 与登录态要让服务器把请求当作普通浏览器访问请求头至少带三样User-Agent、Referer、Accept-Language。UA 用最新浏览器的完整字符串Referer 指向商品列表页语言设成 zh-CN。用 requests.Session 保持整个会话后续请求自动沿用 Cookieimport requests session requests.Session() session.headers.update({ 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: https://search.jd.com/, Accept-Language: zh-CN,zh;q0.9, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, }) resp session.get(url, timeout10) resp.encoding utf-8 html resp.texttimeout10 必须写否则网络抖动时 requests 会一直挂着等响应调度任务越积越多最终拖垮整条监控链路。UA 写死一份够用不要追着浏览器版本更新真遇到页面返回异常验证页再考虑更换。部分商品需要登录才能看到真实价格做法是浏览器里手动登录一次把关键 Cookie 复制到配置里session 会自动携带。3.3 价格提取正则与候选字段双保险以及字体反爬的兜底价格在页面里常见格式是“888.00”或“8,888.00”千分位逗号必须优先去掉。写一个提取函数处理两种格式顺便做区间校验import re def extract_price(text: str) - float | None: 从一段文本中提取第一个合理价格失败返回 None text text.replace(,, ) m re.search(r(\d\.\d{2}|\d), text) if not m: return None price float(m.group(1)) if price 1 or price 100_000_000: return None return price正则匹配“整数或小数点后两位的整数”价格合理性检查放在最后一道防线。个别页面会用自定义字体做反爬把数字渲染成特殊字符源码里的“9”在页面上显示成另一个字形。应对方式需要下载字体文件、解析映射表对个人监控来说成本偏高我会直接跳过这类商品并记录日志不值得为单一监控对象花一整个下午。3.4 入库与变化判断SQLite 写入前置检查抓到的价格不能直接往库里塞先与最近一条历史记录比较波动超过 50% 时保留入库但打上异常标记。这样既能应对临时降价也不会因为解析错误污染后续的趋势判断import sqlite3 from datetime import datetime, timezone, timedelta DB_PATH data/price_log.db def save_price(sku: str, title: str, price: float): now datetime.now(timezone(timedelta(hours8))).strftime(%Y-%m-%d %H:%M:%S) with sqlite3.connect(DB_PATH) as conn: conn.execute( INSERT INTO price_history (sku, title, price, check_time) VALUES (?, ?, ?, ?), (sku, title, price, now), ) conn.commit()插入前先 SELECT 最近一条记录做差值计算差值超阈值时在日志里额外输出一行。时区固定为东八区并转成字符串避免部署到跨时区服务器后时间对不上。对个人监控而言这个数据量完全不需要索引一个月几千行全表扫描毫无压力。4. 降价通知不靠盯屏SMTP 邮件与 webhook 推送的落地配置4.1 邮件通知SMTP 配置与提醒内容模板邮件是最不容易被忽略的通知通道适合“睡前统一看一遍”的场景。用 QQ 邮箱或 163 邮箱开启 SMTP 服务后拿到授权码填入配置即可。SSL 连接方式端口固定为 465import smtplib from email.mime.text import MIMEText from email.header import Header def send_mail(subject: str, body: str, to_addr: str): smtp_host smtp.qq.com smtp_port 465 user your_accountqq.com auth_code your_auth_code # 登录密码不行必须用授权码 msg MIMEText(body, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] user msg[To] to_addr with smtplib.SMTP_SSL(smtp_host, smtp_port, timeout15) as server: server.login(user, auth_code) server.send_message(msg)邮件正文至少包含商品名、当前价格、目标阈值、商品链接四个要素标题直接带当前价格比如“某无线耳机已降至 1599关注价 1699”。邮件模板可以按 SKU 分类不同类别用不同的关注价。SMTP 偶尔会因网络超时报错调用处一定要包 try/except邮件失败不能拖垮主流程。4.2 Server 酱与企业微信群机器人把提醒推到手机邮件适合统一查看降价提醒更适合直接推到手机。Server 酱、企业微信群机器人都属于这类 webhook 服务一个 URLPOST 两个参数手机立刻收到。实现逻辑很薄import requests def push_webhook(title: str, desp: str): webhook https://example.webhook/send?keyYOUR_KEY resp requests.post( webhook, json{title: title, desp: desp}, timeout10, ) result resp.json() if result.get(code) ! 0: raise RuntimeError(fpush failed: {result})webhook 地址是敏感配置放在 config.py 里以占位符形式存在。desp 参数支持简单排版可以把商品名、价格、历史最低价、商品链接拼成多行推送可读性比邮件更好。这类推送服务的稳定性完全取决于服务商所以推送失败一定要落日志让邮件通知兜底两条通道搭配使用才可靠。4.3 两条节流策略时间窗口与二次降价百分比比漏发更常见的问题是刷屏。商品价格在目标价附近反复横跳时每 30 分钟推一次一天能收几十条。节流方案不能只限制时间窗口还要判断“价格是否真的更低了”。我的做法是双重判断距离上次通知超过 2 小时同时价格比上次通知时又低了 5% 以上才再次推送。def should_notify(sku: str, price: float, last_notify_price: float | None) - bool: 时间窗口 二次降价百分比双重限制 if last_notify_price is None: return True if price last_notify_price * 0.95: return False return True第一次触发当然立刻推后续推送条件更苛刻只有价格比上次通知更低才推送5% 的阈值能覆盖大部分促销节奏。时间窗口控制提醒密度百分比控制提醒价值两者缺一不可。节流状态存进 SQLite 的 notify_log 表重启进程也不会丢比内存变量可靠得多。整套通知链路在跑真实任务前先手动调用一次确认邮件能收到、webhook 能送达再挂到定时调度上。5. 常见问题与避坑反爬、解析异常与链路静默的排查5.1 现象固定请求头下价格页面偶尔返回 400 或验证页原因通常有两个方向请求频率过高或者 Cookie 过期。监控多个商品时如果所有任务在同一分钟启动短时间对同一页面发起多次请求很容易触发临时风控。Cookie 则会在登录态失效后被服务端主动拒绝。解决办法是打散请求节奏并在失败时停止重试import time import random time.sleep(random.uniform(5, 15)) resp session.get(url, timeout10)sleep 放在 request 之前均匀取 5 到 15 秒的随机数避免固定间隔被识别。同一轮里连续三次拿不到价格就跳过该商品把问题留到日志里不要死循环重试。Cookie 过期时标记“需要更新 Cookie”并暂停该 SKU人工介入处理比反复试探稳妥。5.2 现象解析到的价格比页面显示少一个数量级这类 bug 最阴险程序不报错库里存着错误数据。原因几乎都是截取到了划线价、预售价或者促销标签旁边的旧价格。页面里同时存在多个价格展示正则默认取第一个数字恰好命中不含优惠的原价。解决解析函数里增加候选字段的优先级排序优先找含 price 语义的 JSON 字段其次匹配带货币符号的节点拿到结果后与上次记录对比波动超 50% 就标记异常入库前再做一次区间校验。价格合理性检查要放在统一入口而不是散落在各个解析分支里否则新增商品时容易漏掉。5.3 现象定时任务看着在跑数据库里没有新数据这个问题三分之二出在路径上三分之一出在异常被吞。最典型的场景在项目根目录手动启动一切正常改用系统计划任务或服务方式运行后工作目录变成用户主目录data/price_log.db 的相对路径指向了错误位置SQLite 报错退出。任务内异常如果只 print 到控制台后台运行时输出直接丢弃看起来就像什么都没发生。解决所有路径基于项目根目录做绝对路径拼接日志写到文件而不是控制台import logging logging.basicConfig( filenamelogs/price_monitor.log, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s, )每轮抓取结束输出一行摘要包含成功、失败、异常计数。这样即使看不到控制台日志文件也能完整还原执行过程。如果 24 小时完全没有新日志说明调度入口没触发先查计划任务配置再查脚本是否真的被拉起。5.4 现象zip 源码包解压后无法运行提示缺模块或编码错误两个常见原因。第一Windows 下用记事本改过源码把 UTF-8 文件保存成了 GBKPython 3 默认按 UTF-8 读源码遇到非法字节直接 SyntaxError。第二requirements.txt 与当前 Python 版本不兼容比如老项目里的依赖不支持新解释器。还有一类是 zip 内打包了缓存目录解压后路径混乱间接导致找不到模块。解决源码统一 UTF-8 编码保存确认 Python 版本后按依赖清单安装python -m pip install -r requirements.txt如果在 Windows 命令提示符里执行 pip 提示“不是内部或外部命令”说明安装 Python 时没勾选 Add Python to PATH重新运行安装程序修复即可。遇到 zip 伪加密提示时先确认文件列表是否可见不要被提示误导7-Zip 通常能正常列出内容。5.5 现象抓取和入库都正常但收不到降价通知链路断开的位置通常在推送侧。先查 notify_log 表确认上次通知时间是否被误写为当前时间节流逻辑把后面的通知全部拦截了。再看 webhook 服务返回推送接口升级后返回结构可能变化旧代码里读 code 字段判断成功就会失效。解决写一个独立的测试脚本直接调用推送函数排除主流程干扰推送响应用完之后先 raw 打印一层接口变更后能第一时间发现字段差异。只要抓取链路和通知链路分别都能独立跑通合在一起出问题的概率就很小。6. 收尾技巧用 SQLite 历史价格回放验证监控链路6.1 一条 SQL 查出这周的抓取质量监控跑了一周最该问的问题是“它真的每半小时都上班了吗”。与其相信感觉不如直接看数据库按天分组统计SELECT substr(check_time, 1, 10) AS day, COUNT(*) AS samples, MIN(price) AS min_price, MAX(price) AS max_price FROM price_history WHERE sku 指定商品 GROUP BY day;samples 接近 48说明当天 30 分钟一次的任务正常执行远低于 48那天大概率发生过断档。min_price 和 max_price 差距过大则要对照日志确认是真实降价还是解析异常。用这个结果和浏览器里手动看到的历史价格交叉验证监控链路是否健康一目了然。6.2 用历史最低价反推通知阈值固定阈值容易失灵设太高永远不触发设太低促销来了也静默。更合理的做法是从历史数据里找参考线def lowest_price_last_days(sku: str, days: int 7): with sqlite3.connect(DB_PATH) as conn: rows conn.execute( SELECT price, check_time FROM price_history WHERE sku ? AND check_time datetime(now, ?) ORDER BY check_time, (sku, f-{days} days), ).fetchall() return min(rows, keylambda r: r[0]) if rows else None返回最近七天的最低价格和出现时间通知阈值取它的九折既参考了历史波动又留有触发余地。每次回放顺手看一眼趋势摘要哪天的价格异常、哪个商品整天没数据立刻能定位到链路故障点。我每周五会固定看一眼 price 库就像检查仪表盘一样——这份数据不只在降价时有用更是监控系统自身健康状况的客观凭证。希望这套回放思路能帮你把监控跑得更放心。本文还有配套的精品资源点击获取