简介这份资源是一套面向电商价格监控场景的京东商品价格实时监控工具源码适合具备一定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 或 PostgreSQLrepository.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, minutesp[interval_minutes], args[p], idfmonitor_{p[id]} ) scheduler.start()这段代码有三个参数值得说清楚。coalesceTrue解决的是“进程卡了一下积压了好几次触发”的问题合并成一次执行不然会瞬间打出一串重复请求。max_instances1保证同一商品不会有两个任务同时跑否则两次写入可能把“上一次价格”覆盖乱。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( total3, # 最多重试 3 次 backoff_factor1, # 退避1s、2s、4s status_forcelist[429, 500, 502, 503, 504] ) session.mount(https://, HTTPAdapter(max_retriesretry)) session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Accept-Language: zh-CN,zh;q0.9 }) def fetch_html(url, timeout10): try: resp session.get(url, timeouttimeout) resp.raise_for_status() return resp.text except requests.RequestException as e: print(f[fetch] {url} 失败: {e}) return NoneRetry的backoff_factor1意味着重试间隔按 1、2、4 秒递增给目标站点喘息时间也降低被判定为异常流量的概率。status_forcelist里放 429 是因为限流响应重试往往能过放 5xx 是因为服务端偶发错误重试有效。超时设 10 秒是经验值太短容易误判慢页面太长会拖住整个调度。请求头里带Accept-Language不是玄学部分站点会根据语言返回不同页面结构固定住能减少解析规则失效的概率。解析部分要能兜底因为页面结构说变就变# crawler/parser.py import re from bs4 import BeautifulSoup def parse_price(html, parser_typejd): if html is None: return None # 优先走结构化数据最稳 m re.search(rprice\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] monitorexample.com msg[To] youexample.com with smtplib.SMTP_SSL(smtp.example.com, 465) as s: s.login(monitorexample.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: monitorexample.com password: your-auth-code to: youexample.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、salePriceDOM 选择器限定在促销价容器内解析完打印出来人工核对几个商品确认取的是到手价再放开跑。5.2 任务越跑越慢最后卡死现象跑了一两天后日志里任务执行间隔越来越长甚至不再触发。原因请求没设超时或超时太长某个页面卡住导致任务实例不释放加上没设max_instances新触发不断堆积。解决请求必须带timeout调度器加max_instances1和coalesceTrue再给整个任务包一层异常捕获单次失败不能影响后续调度。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 一个 webhookbody 里带 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 分钟人为改几次价格确认告警内容、频率、去重都符合预期再切到正式配置。有一次我图省事直接改生产配置结果阈值忘了调回来半夜被几十条告警轰醒——从那以后我每次改告警相关参数都强制先在测试库上走一遍完整链路再上线。希望这套拆解能帮到你少走几个我踩过的坑。本文还有配套的精品资源点击获取