简介：这份资源是一套面向电商价格监控场景的京东商品价格实时监控工具源码，适合具备一定Python基础的开发者、爬虫学习者及关注电商数据分析的读者参考。压缩包共8个文件，约7KB，以4个Python脚本为核心，分别承担商品页面抓取、邮件通知发送、数据库连接等职责，另附MySQL建表脚本、XML配置、开源许可证与说明文档，结构精简、模块划分清晰，便于快速理解监控流程。已有283人学习下载，说明其在同类小工具中具备一定参考价值。读者可从中了解定时抓取商品价格、入库存储、价格变动推送的完整实现思路，并基于现有脚本调整监控目标与通知方式，作为爬虫入门练习或价格监控原型搭建的起点。
1. 价格监控这件事，先从一次“手慢无”说起
去年帮朋友盯一款显卡的价格，我定了个闹钟每天手动刷三次页面，结果有天下午价格闪崩了二十分钟，等我晚上打开页面，早就恢复原价了。那次之后我才认真去找能落地的电商价格监控方案，翻到这个项目——一个围绕商品价格实时监控搭起来的完整工程，核心就是定时抓取、价格比对、变动告警这条链路。它适合两类人：一类是想给自己盯几件商品、不想再靠手动刷新的普通用户；另一类是想拿一套能跑通的采集加调度骨架，改成自己业务监控系统的开发者。项目本身不复杂，但把“定时任务怎么排、页面数据怎么抽、价格怎么存、变动怎么通知”这几件事串完整了，拿来当二次开发的底座比从零搭省事得多。
2. 项目定位与技术选型：这套监控骨架到底由什么撑起来
2.1 它解决的是“价格变了我知道”，不是“爬全站商品”
先把预期摆正。这个项目的边界很清晰：你给它一批商品链接或商品 ID，它按设定的频率去拉价格，和上一次记录做比对，发现变动就落库并触发通知。它不做全站商品发现，不做比价聚合，也不做历史价格曲线的高级分析——这些是二次开发可以往上叠的部分。
从工程角度看，一条完整的价格监控链路至少包含四段：调度层负责“什么时候抓”，采集层负责“怎么把价格抠出来”，存储层负责“历史价格怎么留”，通知层负责“变了怎么告诉我”。这个项目就是按这四段组织的，目录结构大致如下（不同分支命名可能略有差异，但职责划分一致）：
price-monitor/ ├── config/ # 商品清单、抓取频率、通知渠道配置 │ ├── products.yaml # 要监控的商品列表 │ └── settings.yaml # 全局参数 ├── scheduler/ # 调度层：定时触发采集任务 │ └── job_runner.py ├── crawler/ # 采集层：请求页面 + 解析价格 │ ├── fetcher.py │ └── parser.py ├── storage/ # 存储层：价格历史与变动记录 │ ├── models.py │ └── repository.py ├── notifier/ # 通知层：邮件/ webhook 告警 │ └── alert.py ├── main.py # 入口 └── requirements.txt这种分层的好处是每一层都能单独替换。比如你觉得某个解析规则不稳，只动parser.py；想把邮件换成企业微信机器人，只动notifier/alert.py，调度和存储完全不用碰。我一般接手这类项目第一件事就是看分层清不清楚，分层乱的采集项目改起来是灾难。
2.2 技术栈：Python + 定时调度 + 轻量存储
技术选型上，这个项目走的是“够用就好”的路线，没有上重型框架：
| 层次 | 选型 | 说明 |
|---|---|---|
| 语言 | Python 3.8+ | 采集和解析生态成熟，改起来门槛低 |
| 调度 | APScheduler / 系统 cron | 进程内定时或系统级定时二选一 |
| 请求 | requests | 同步请求，配合重试和超时控制 |
| 解析 | 正则 + BeautifulSoup | 静态价格走正则，结构复杂的走 DOM 选择器 |
| 存储 | SQLite（可换 MySQL） | 单机够用，换库只改连接配置 |
| 通知 | SMTP 邮件 / Webhook | 邮件开箱即用，Webhook 适合接群机器人 |
为什么用 SQLite 而不是直接上 MySQL？因为价格监控这种场景，单机、数据量不大、写入频率低（几分钟到几小时一次），SQLite 完全扛得住，还省了部署数据库的麻烦。等你要监控几千个商品、多个实例并发写的时候，再换 MySQL 或 PostgreSQL，repository.py里换掉连接串和少量方言即可。这种“先跑起来再扩容”的思路，比一上来堆中间件务实得多。
2.3 配置驱动：商品清单和频率都放在 YAML 里
项目把“监控什么、多久监控一次”抽成了配置，而不是写死在代码里。这是我觉得最值得保留的设计——加一个商品不用改代码，改配置重启即可：
# config/products.yaml products: - id: "100012043978" name: "示例显卡" url: "https://example.com/item/100012043978" parser: "jd" # 指定用哪套解析规则 interval_minutes: 30 # 该商品的抓取间隔 threshold: 50 # 价格变动超过 50 元才告警，过滤抖动parser字段是关键，它让不同站点的解析规则可以并存；interval_minutes支持按商品单独设频率，热门商品盯紧点，冷门的放宽；threshold是防噪的，价格上下浮动几块钱就告警会把人烦死，设个阈值只报有意义的变化。这三个字段基本覆盖了实际使用中最常调的东西。
3. 核心链路拆解：一次价格抓取从触发到告警的完整过程
3.1 调度层：定时任务怎么排才不互相踩
调度层负责按配置把采集任务派发出去。项目里用的是 APScheduler 的BackgroundScheduler，按每个商品的interval_minutes注册独立任务：
# scheduler/job_runner.py from apscheduler.schedulers.background import BackgroundScheduler from crawler.fetcher import fetch_price from storage.repository import save_price, get_last_price from notifier.alert import send_alert scheduler = BackgroundScheduler( job_defaults={ "coalesce": True, # 错过的多次触发合并成一次，避免堆积 "max_instances": 1, # 同一商品不并发跑，防止重复写 "misfire_grace_time": 60 # 错过 60 秒内仍执行，超时丢弃 } ) def monitor_job(product): price = fetch_price(product) # 采集当前价格 if price is None: return # 抓取失败直接跳过，不写脏数据 last = get_last_price(product["id"]) # 取上一次记录 save_price(product["id"], price) # 先落库 if last is not None and abs(price - last) >= product["threshold"]: send_alert(product, last, price) # 超过阈值才告警 def start(products): for p in products: scheduler.add_job( monitor_job, "interval", minutes=p["interval_minutes"], args=[p], id=f"monitor_{p['id']}" ) scheduler.start()这段代码有三个参数值得说清楚。coalesce=True解决的是“进程卡了一下，积压了好几次触发”的问题，合并成一次执行，不然会瞬间打出一串重复请求。max_instances=1保证同一商品不会有两个任务同时跑，否则两次写入可能把“上一次价格”覆盖乱。misfire_grace_time是容错窗口，机器短暂卡顿导致任务晚了几十秒还能补上，晚太多就放弃这次，避免雪崩。这三个参数是定时采集项目里最容易忽略、又最容易翻车的地方，我见过不少项目就是因为没设max_instances，网络一慢任务堆叠，最后把目标站点请求到被限流。
3.2 采集层：请求要稳，解析要能兜底
采集层分两步：fetcher.py负责把页面拿回来，parser.py负责把价格抠出来。请求部分的核心是超时、重试和请求头：
# crawler/fetcher.py import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() retry = Retry( total=3, # 最多重试 3 次 backoff_factor=1, # 退避：1s、2s、4s status_forcelist=[429, 500, 502, 503, 504] ) session.mount("https://", HTTPAdapter(max_retries=retry)) session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Accept-Language": "zh-CN,zh;q=0.9" }) def fetch_html(url, timeout=10): try: resp = session.get(url, timeout=timeout) resp.raise_for_status() return resp.text except requests.RequestException as e: print(f"[fetch] {url} 失败: {e}") return NoneRetry的backoff_factor=1意味着重试间隔按 1、2、4 秒递增，给目标站点喘息时间，也降低被判定为异常流量的概率。status_forcelist里放 429 是因为限流响应重试往往能过，放 5xx 是因为服务端偶发错误重试有效。超时设 10 秒是经验值，太短容易误判慢页面，太长会拖住整个调度。请求头里带Accept-Language不是玄学，部分站点会根据语言返回不同页面结构，固定住能减少解析规则失效的概率。
解析部分要能兜底，因为页面结构说变就变：
# crawler/parser.py import re from bs4 import BeautifulSoup def parse_price(html, parser_type="jd"): if html is None: return None # 优先走结构化数据，最稳 m = re.search(r'"price"\s*:\s*"?([\d.]+)"?', html) if m: return float(m.group(1)) # 兜底走 DOM 选择器 soup = BeautifulSoup(html, "html.parser") node = soup.select_one(".price .value, [class*=price]") if node: num = re.search(r"[\d.]+", node.get_text()) if num: return float(num.group(0)) return None解析策略是“先正则抠 JSON 里的价格字段，再退到 DOM 选择器”。很多电商页面会把价格塞在页面内嵌的 JSON 里，正则匹配"price":往往比 DOM 选择器稳定，因为前端改样式不影响 JSON 结构。DOM 选择器作为兜底，用[class*=price]这种模糊匹配提高命中率。返回None表示这次没解析出来，上层会跳过不写库——宁可漏一次，也不要把错误价格写进历史，否则后面告警全是误报。
3.3 存储层：历史价格怎么留，变动怎么查
存储层两张表就够：一张存每次抓到的价格，一张存变动事件。表结构大致这样：
-- storage/schema.sql CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id TEXT NOT NULL, price REAL NOT NULL, captured_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_product_time ON price_history(product_id, captured_at DESC); CREATE TABLE IF NOT EXISTS price_alert ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id TEXT NOT NULL, old_price REAL, new_price REAL, alerted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );idx_product_time这个复合索引是必须的，因为查“某商品最近一次价格”是最高频的操作，没有索引的话数据一多就慢。captured_at用数据库默认时间戳，避免应用层时区处理出错。写入逻辑上，save_price先写历史再返回，get_last_price按captured_at DESC LIMIT 1取最近一条——注意这里取的是“写入前的最近一条”，所以调用顺序必须是先查后写，顺序反了就会拿自己刚写的价格做比对，永远不触发告警。这个顺序坑我在别的项目里踩过，排查了半天才发现是比对对象搞错了。
3.4 通知层：告警要带上下文，别只发一个数字
通知层最容易被写得很敷衍，只发一句“价格变了”。实际用起来，告警必须带足够上下文，否则收到消息还得自己去查：
# notifier/alert.py import smtplib from email.mime.text import MIMEText def send_alert(product, old_price, new_price): direction = "下降" if new_price < old_price else "上涨" diff = abs(new_price - old_price) body = ( f"商品：{product['name']}\n" f"价格{direction}：{old_price} -> {new_price}（变动 {diff:.2f}）\n" f"链接：{product['url']}\n" f"时间：{__import__('datetime').datetime.now():%Y-%m-%d %H:%M:%S}" ) msg = MIMEText(body, "plain", "utf-8") msg["Subject"] = f"[价格{direction}] {product['name']}" msg["From"] = "monitor@example.com" msg["To"] = "you@example.com" with smtplib.SMTP_SSL("smtp.example.com", 465) as s: s.login("monitor@example.com", "your-auth-code") s.send_message(msg)邮件正文里带商品名、变动方向、新旧价格、差额、链接和时间，收到就能判断要不要行动。主题里带方向，方便邮箱规则过滤。SMTP 用 465 端口走 SSL，登录密码用授权码而不是账号密码，这是邮件通知的常规做法。如果不想用邮件，把send_alert换成 POST 一个 webhook 地址即可，函数签名不变，上层完全无感。
4. 跑起来：从零到收到第一条告警的操作步骤
4.1 环境准备与依赖安装
先把代码拉下来，建虚拟环境装依赖。Python 版本建议 3.8 以上，低版本可能缺zoneinfo之类的标准库：
python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txtrequirements.txt里核心就四个包：requests、beautifulsoup4、APScheduler、PyYAML。如果装的时候卡在编译，多半是某个包版本太老，升级 pip 再试。装完先跑一下python -c "import apscheduler, bs4, yaml"确认没报错，比直接启动再排查省事。
4.2 配置商品清单和通知参数
打开config/products.yaml，把示例商品换成你要盯的。第一次跑建议只放一个商品，间隔设短一点（比如 5 分钟），方便快速验证链路通不通。threshold先设小一点（比如 1），这样价格有任何变动都能触发，确认告警能收到之后再调回合理值。
config/settings.yaml里配数据库路径和通知渠道：
database: path: "./data/price.db" notify: channel: "email" # email 或 webhook smtp_host: "smtp.example.com" smtp_port: 465 username: "monitor@example.com" password: "your-auth-code" to: "you@example.com"数据库路径指向的目录要提前建好，SQLite 不会自动创建父目录，路径不存在会直接报错。通知密码用授权码，别填邮箱登录密码，这是新手最常卡住的一步。
4.3 启动与验证
启动入口是main.py，它会读配置、初始化数据库、注册任务然后常驻：
python main.py启动后看日志，正常会打印每个商品注册的任务 ID 和下次执行时间。验证链路是否通，最直接的办法是手动改一次数据库里的价格，让下一次抓取产生“变动”：
sqlite3 data/price.db "UPDATE price_history SET price = price + 100 WHERE id = (SELECT MAX(id) FROM price_history);"下一次任务跑起来，比对到的旧价格就被人为抬高了，如果新抓到的价格和它差超过阈值，就会触发告警。收到邮件说明整条链路通了。这个手动造变动的方法比干等价格真变要快得多，调试阶段我基本都这么干。
5. 避坑与排查：价格监控最容易翻车的几个地方
5.1 抓到的价格是“划线价”不是“到手价”
现象：告警说价格降了，打开页面发现实际到手价没变。原因：页面里同时存在原价、促销价、券后价多个数字，正则或选择器命中了第一个匹配到的，往往是划线原价。解决：解析时优先取带促销语义的字段，比如 JSON 里的finalPrice、salePrice，DOM 选择器限定在促销价容器内；解析完打印出来人工核对几个商品，确认取的是到手价再放开跑。
5.2 任务越跑越慢，最后卡死
现象：跑了一两天后，日志里任务执行间隔越来越长，甚至不再触发。原因：请求没设超时或超时太长，某个页面卡住导致任务实例不释放，加上没设max_instances，新触发不断堆积。解决：请求必须带timeout，调度器加max_instances=1和coalesce=True，再给整个任务包一层异常捕获，单次失败不能影响后续调度。
5.3 价格没变却反复告警
现象：同一商品短时间内连续收到多条告警，价格其实只差几毛钱。原因：threshold设得太小，或者页面价格带随机小数抖动，再或者比对时把“自己刚写入的价格”当成了旧价格。解决：把threshold调到有意义的值（比如 10 或 50），确认get_last_price在save_price之前调用，必要时对价格做取整或按分位过滤。
5.4 换了个商品解析就返回 None
现象：新加的商品一直抓不到价格，日志里全是解析失败。原因：不同站点甚至同站点不同品类页面结构不同，parser字段没配对，或者页面是前端渲染的，requests拿到的 HTML 里根本没有价格。解决：先curl一下目标 URL 看返回的 HTML 里有没有价格字符串，没有就是动态渲染，需要换方案（接口抓取或渲染引擎）；有的话针对该页面单独写解析规则，在配置里用不同的parser值区分。
5.5 数据库文件越来越大
现象：跑了几周后price.db涨到几百 MB，查询变慢。原因：每次抓取都写一条历史，高频商品一天几百条，长期累积。解决：加定期清理策略，比如只保留最近 90 天的明细，或者对长期无变动的商品降频；历史数据可以按天聚合后归档，明细表只留近期。清理任务同样挂到调度器上，每天凌晨跑一次。
6. 二次开发：把它改成你自己的监控系统
这套骨架最值钱的地方在于分层清晰，往上叠功能不用动地基。我一般会从三个方向改。
第一个方向是换通知渠道。邮件适合个人用，团队场景更适合群机器人。把notifier/alert.py里的send_alert改成 POST 一个 webhook，body 里带 markdown 格式的消息，函数签名保持不变，调度层一行不用改。如果要支持多渠道，加一个notifier/dispatcher.py，按配置把告警分发给邮件、webhook、短信，每个渠道实现同一个接口即可。
第二个方向是加价格分析。现在只有“变动告警”，可以基于price_history表算历史最低价、近 30 天均价、当前价处于历史什么分位。加一个analysis/目录，查询用 SQL 聚合就能出结果：
-- 某商品近 30 天最低价与均价 SELECT MIN(price) AS min_price, AVG(price) AS avg_price, COUNT(*) AS samples FROM price_history WHERE product_id = ? AND captured_at >= datetime('now', '-30 days');告警逻辑里加一条“当前价低于近 30 天均价 10% 才推送”，噪音会少很多，推送的也更有价值。这个改动只涉及查询和告警条件，不动采集。
第三个方向是提升采集稳定性。如果目标页面是动态渲染的，requests拿不到价格，就得换抓取方式。常见做法是找页面背后的价格接口直接请求 JSON，比渲染整页轻得多；实在没有接口再上无头浏览器，但要注意资源占用，别一个商品开一个浏览器实例，用连接池复用。改的时候把fetcher.py抽象成一个接口，静态请求和渲染请求各实现一份，配置里选，这样两种方式能共存。
改完怎么验证？我的习惯是先在测试库上跑，把threshold调到 1、间隔调到 1 分钟，人为改几次价格，确认告警内容、频率、去重都符合预期，再切到正式配置。有一次我图省事直接改生产配置，结果阈值忘了调回来，半夜被几十条告警轰醒——从那以后我每次改告警相关参数，都强制先在测试库上走一遍完整链路再上线。希望这套拆解能帮到你，少走几个我踩过的坑。
本文还有配套的精品资源，点击获取
