简介这是一套面向Python初学者与电商自动化爱好者的学习型工具源码聚焦京东平台商品库存监控与自动下单场景解决用户在抢购热门商品时需手动刷新、易错过补货时机的痛点。资源包含61个文件以9个核心Python脚本如JdBuyer.py、JdSession.py、timer.py、10个备份配置.zbak、36个地域ID文本覆盖全国34个省级行政区及港澳台、海外、钓鱼岛等特殊区域、以及config.json、README.md、logo.ico等支撑文件总大小仅605KB结构清晰、模块职责分明。已有103人学习下载适合希望掌握HTTP会话管理、定时任务调度、GUI开发基于tkinter及跨平台脚本封装的实践者。读者可直接运行图形界面模式快速上手或通过命令行脚本模式深入理解日志记录、异常捕获与微信/钉钉通知集成逻辑配套使用指南详述环境配置、Cookie获取与参数调优方法。 去年年底我想买一款显卡蹲了半个月硬是没抢到。后来实在受不了手动刷新页面就写了一套自动化监控脚本先实现了库存检测和消息推送后来又逐步补上了自动加购和提交订单的链路最终沉淀成一套可以部署的库存监控与自动下单系统。这里把源码设计和部署使用心得整理出来给有类似需求的读者一个可参考的落地方案。这个项目适合谁如果你经常需要紧盯某个商品的补货时间或者想在商品上架的第一时间获取通知再或者你本身就在做电商数据相关的技术研究这套系统都能派上用场。它不是一个灰产工具箱而是一个典型的数据采集、状态判断、任务调度和自动化操作的实战项目技术上涉及 requests、Redis、定时任务、Web 管理端读完你会对这类系统有完整的认知。1. 需求拆解与整体方案设计1.1 库存监控到底在监控什么很多人以为库存监控就是定时请求一下商品详情页判断“有货”还是“无货”。真做起来就会发现这里面的细节比想象中多。一个商品链接往往对应多个 SKU比如手机有黑色、白色、不同内存版本每个 SKU 的库存状态是独立的。只看商品主图页面的库存字段可能永远显示有货但具体到你想要的那个颜色其实一直缺货。所以监控的粒度必须下沉到 SKU 维度至少要拿到商品 ID、SKU ID、规格名、实时价格和库存数量。另外电商平台的库存返回通常不是单一字段。有些接口返回的是“可售状态”有些返回“库存数量”有些则要区分“现货”和“预售”还有的平台会有区域库存概念不同收货地址看到的库存不一样。这些因素都会影响系统判断。我在最初版本里只读取了一个字段结果同一商品在不同账号下看到的状态完全不同排查了很久才发现是区域参数导致的。设计监控规则时一定要把“哪些字段变化才算是真正的库存变化”想清楚否则会收到大量无效告警。1.2 自动下单系统的核心难点自动下单的上手门槛比监控高一个量级主要难点有三块。第一是链路长。一次完整的下单至少要经过登录态校验、加入购物车、访问结算页、获取订单确认信息、提交订单。每一步都有独立的接口接口之间还可能依赖上一步返回的 token、订单号、地址 ID 等参数。任何一个环节出错整个流程都要重新来。第二是风控拦截。平台对高频请求、异常行为、非人工操作有比较成熟的识别策略。短时间频繁加购、提交订单很容易触发验证码、滑块甚至账号限制。这套系统如果追求高并发下单就必须面对风控而风控策略往往是黑盒只能靠降低频率、模拟人工节奏、轮换账号和 IP 来缓解。第三是时序竞争。你监控到“有货”的同时可能已经有成千上万个用户也在下单商品从“有货”到“已售罄”可能只需要几秒钟。系统不仅要监控得快还要在下单环节尽量缩短时间损耗哪怕省掉一次无效的页面跳转成功率都能高一些。所以我最终把系统拆成两个独立的模块监控模块和下单模块。监控模块独立运行负责轮询库存、记录数据、发送通知下单模块默认关闭只有在人工确认开启时才执行自动加购和提交订单。这样的设计既保证了日常监控的稳定性也降低了误操作带来的账号风险。1.3 技术选型为什么这么搭技术栈上我选了 Python 作为主力开发语言理由很简单生态成熟、写起来快、调试方便requests/httpx 做 HTTP 请求足够用配合 BeautifulSoup 或正则就能解析页面内容。任务调度这块我没有引入太重的东西。最开始用的是 Python 的 threading.Timer 自己做定时循环简单但不好管理重启、暂停、修改监控频率都得改代码。后来换成 APScheduler支持 cron 表达式、后台运行、持久化任务已经能满足大部分监控场景。再配合 Celery 可以实现分布式任务队列但如果只是单机跑Celery 有点杀鸡用牛刀建议先从 APScheduler 入手。数据存储分两部分Redis 用来做实时状态缓存和接口幂等控制MySQL 用来存历史库存记录、通知日志和商品配置。Redis 的 key 设计成stock:{sku_id}value 直接存 JSON 字符串包含库存状态、价格、更新时间。这样监控模块每次轮询后只要对比 Redis 里的旧状态就知道有没有变化避免每次变化都写数据库。Web 管理端我选了 Flask原因就是轻ORM 用 SQLAlchemy配合一个简单的前端页面就能实现对商品监控任务、下单开关、通知渠道的管理。如果你对 Spring Boot 那套更熟悉也可以把核心服务用 Java 重写架构思路完全一致只是请求库和调度库需要换成对应的技术组件。这里放一个组件选型表方便参考模块技术选型用途HTTP 请求requests / httpx封装商品详情、加购、订单提交接口页面解析BeautifulSoup / 正则从 HTML 中提取 SKU、价格、库存信息任务调度APScheduler管理每个商品的轮询任务缓存Redis状态去重、短时间防重复下单数据库MySQL / SQLite持久化商品配置、历史库存、操作日志管理端Flask SQLAlchemy商品配置、监控日志、任务控制通知SMTP / Server酱 / 钉钉机器人向手机推送库存变化和下单结果这套组合的优点是上手门槛低、部署成本小单台 1 核 2G 的云服务器就能跑得很稳。缺点是 Python 的 GIL 限制了对多核 CPU 的利用不过监控任务本身就是 I/O 密集型的网络请求把并发放在协程或者线程池里解决就够了GIL 影响不大。2. 核心模块设计与源码解析2.1 监控模块轮询、状态判断与数据记录监控模块是整个系统的心脏。它的核心逻辑并不复杂定时请求商品数据解析出自己想要的信息和上一次记录做对比发生变化就处理没变化就继续安静地等待下个周期。轮询间隔怎么定是个值得琢磨的问题。间隔太短容易给平台造成压力也可能触发风控间隔太长又可能错过补货窗口。我的经验是普通商品 30 秒到 60 秒轮询一次就够如果商品非常抢手可以降低到 5 秒到 10 秒但这时候必须搭配代理 IP 和账号池否则很容易被限制登录态。下面是一段核心轮询代码import requests import json import redis import time session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://item.jd.com/, }) r redis.Redis(hostlocalhost, port6379, db0) def fetch_sku_info(sku_id): url fhttps://item.jd.com/{sku_id}.html resp session.get(url, timeout10) # 实际项目中会从 HTML 或内嵌 JSON 中解析出价格、库存等信息 # 这里用模拟数据展示结构 info { sku_id: sku_id, stock: in_stock, # in_stock / out_of_stock price: 2999.00, ts: int(time.time()), } return info def check_and_notify(sku_id): data fetch_sku_info(sku_id) cache_key fstock:{sku_id} old r.get(cache_key) if old is None or json.loads(old)[stock] ! data[stock]: r.set(cache_key, json.dumps(data)) print(f[{sku_id}] 库存变化 - {data[stock]}) # 触发通知、记录数据库等后续逻辑 if __name__ __main__: while True: check_and_notify(100012043978) time.sleep(30)这段代码展示的是一个最简原型生产环境里需要把请求重试、异常捕获、代理切换、多 SKU 遍历这些逻辑补全。我实际使用中踩过的一个坑是有些接口返回的不是 JSON而是 JavaScript 变量赋值语句比如window._itemInfo {...}直接 json.loads 会报错。这时候需要先用正则把大括号内的内容提取出来再解析或者干脆用 BeautifulSoup 解析 HTML 里隐藏的 data 属性。库存数据的落库我采用了按天分表或定期清理的策略。每 30 秒一条监控记录一个 SKU 一天就能产生 2880 条数据如果不加清理三个月后数据库就堆满了。我只保留变化记录也就是只有在库存状态、价格发生变化时才写一条历史记录这样数据量能减少一个数量级。2.2 自动下单模块加购、结算与提交订单自动下单模块的核心思路是模拟人工操作的完整链路。以京东的网页端为例一次购买包含以下几个关键步骤登录并保持会话、把商品加入购物车、访问购物车页面获取结算信息、提交订单并获取订单号。每一步都有对应的接口且这些接口需要携带登录后的 Cookie 和 Token。我建议把每一步拆成独立的函数方便单独调试和失败重试。比如加购失败时不需要重新走登录流程只需要重新执行加购那一步函数即可。def add_to_cart(session, sku_id, count1): url https://cart.jd.com/gate.action params { pid: sku_id, pcount: count, ptype: 1, } resp session.get(url, paramsparams, timeout10) return resp.status_code 200 def submit_order(session, address_id, pay_type4): url https://trade.jd.com/shopping/order/submitOrder.action data { address: address_id, payType: pay_type, # 订单确认页返回的 token 和订单号等参数 } resp session.post(url, datadata, timeout10) result resp.json() if result.get(success): return result[orderId] else: raise RuntimeError(result.get(message, submit order failed))这里有一个非常关键的细节提交订单接口通常需要携带一个从订单确认页面解析出来的 token而且这个 token 是一次性的用一次就失效。所以每次提交前必须重新访问订单确认页拿到最新的 token不能把这个 token 缓存后再用。否则会一直报“订单信息过期”或者“请重新下单”。另外一个实际经验是提交订单前最好做一次库存二次确认。因为从监控模块发现库存变化到执行下单中间可能隔着几十秒商品很可能已经被别人抢完了。提前调用一次库存查询接口确认当前 SKU 仍然有货再执行加购和提交可以在一定程度上减少无效请求降低被风控盯上的概率。自动下单模块的触发策略我也踩过不少坑。最初版本是监控发现“有货”后立刻自动下单结果经常误判。后来加了“连续 N 次有货才触发”的规则比如连续 3 次轮询都显示有货才认为是真实补货这才把误报率降下来。这个 N 的值要结合轮询间隔调整如果间隔是 30 秒连续 3 次就是观察了 90 秒如果间隔是 5 秒连续 3 次只观察了 15 秒误报可能性高一些。2.3 库存状态机与通知体系设计库存监控不能只处理“有货”和“无货”两种状态。在很多场景下商品还会处于“预售”“限购”“区域无货”“价格波动”等中间状态。我在系统中抽象了一个状态机每个 SKU 在任何时刻都处于几种有限状态之一状态含义触发动作OUT_OF_STOCK无货等待下一次轮询IN_STOCK有货可下单发送通知、按策略执行下单PRE_SALE预售中发送预售通知DECOMMISSIONED商品已下架停用监控任务并告警PRICE_CHANGED价格变化单独发送价格变动通知状态机的优势是逻辑清晰、便于扩展。比如你可以在 IN_STOCK 状态上配置“自动下单”标志位在 PRICE_CHANGED 状态上配置“低于目标价才提醒”的阈值。这样整个系统就像搭积木一样每个状态对应一组处理动作互相不干扰。通知模块我同时支持了三种渠道邮件、Server酱微信推送、钉钉机器人。日常使用中Server酱的微信推送体验最好延迟低、免装 App钉钉机器人适合团队内部做协作通知邮件则作为兜底方案防止第三方推送服务临时不可用。每种渠道的配置都放在配置文件中启用哪个就填哪个的 API Key。为了不把通知消息做成“轰炸机”我增加了消息聚合逻辑同一 SKU 在五分钟内的变化只推送一条消息避免高频率轮询时手机响个不停。3. 源码部署与使用指南3.1 环境准备与依赖安装部署这套系统不需要多高的服务器配置一台能跑 Python 的 Linux 机器就够了。我自己用的是 1 核 2G 内存的轻量服务器同时跑 Redis、MySQL、Flask 和监控脚本负载一直很稳定。操作系统建议 Ubuntu 20.04 或 CentOS 7 以上版本Python 版本建议 3.8 到 3.10。依赖安装直接用 pip 装pip install flask flask-sqlalchemy pymysql requests beautifulsoup4 redis apscheduler如果你不想在服务器上安装 MySQL可以先用 SQLite 顶替单机项目的数据量完全够用。SQLAlchemy 的配置只需要改一行连接串从mysqlpymysql://user:passlocalhost/dbname换成sqlite:///data.db其余代码不用动。这种方式适合快速验证系统逻辑确认稳定后再切换到 MySQL。Redis 是必须要装的。它是一个轻量级的内存数据库安装方式apt install redis-server systemctl enable redis-server systemctl start redis-serverRedis 在这个系统里主要做两件事一是保存每个 SKU 的最新库存快照用于状态对比二是实现分布式锁防止多个监控进程对同一个 SKU 重复下单。这两件事如果用文件或数据库来做不是不行但性能和可靠性都会差一些。3.2 配置文件详解我把所有可变参数都集中在一个config.yaml文件里这样修改监控频率、添加新商品、切换通知渠道都不用改代码。下面是一份典型配置monitor: interval: 30 # 全局轮询间隔单位秒 retry_times: 3 # 请求失败重试次数 timeout: 10 # 单次请求超时时间单位秒 user_agents: - Mozilla/5.0 (Windows NT 10.0; Win64; x64) - Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) notify: enable: true channels: - type: serverchan key: your_sendkey - type: email smtp_host: smtp.qq.com smtp_port: 465 username: yourqq.com password: your_auth_code receivers: - receiverexample.com - type: dingtalk webhook: https://oapi.dingtalk.com/robot/send?access_tokenxxx order: auto_order: false # 总开关默认关闭自动下单 trigger_count: 3 # 连续 N 次有货才触发 pay_type: 4 # 支付方式4 在线支付 address_id: 123456789配置项里面最危险的就是auto_order这个开关。日常跑监控建议保持false只做通知确认商品确实进入稳定补货期后再临时打开自动下单。我自己一般会先关着跑两天观察通知的准确率再决定开不开自动下单。商品列表我放在数据库里通过 Web 管理端添加也可以直接写 SQL 插入INSERT INTO product (sku_id, name, monitor_interval, notify_enabled) VALUES (100012043978, RTX 4070 显卡, 30, 1), (100016034872, 某品牌内存条 32G, 60, 1);每个商品可以单独设置监控间隔这个设计在实战中非常有用。热门款用短间隔冷门款用长间隔既保证了灵敏度又减少了整体请求量。3.3 启动流程与常用命令启动分三步。第一步是初始化数据库建好表结构第二步是启动 Redis如果还没启动的话第三步是运行监控主程序。# 初始化数据库 python init_db.py # 启动监控主进程前台运行方便看日志 python monitor.py # 启动 Web 管理端后台运行 nohup python web_admin.py --port 8080 web.log 21 如果你希望监控主进程也保持后台运行推荐用 systemd 写一个服务文件或者用nohup丢到后台。不建议用screen或者tmux来守护因为服务器一重启进程就丢了还得手动恢复。systemd 可以设置开机自启进程崩了还能自动拉起更适合长期运行的服务。监控进程起来后正常情况下会循环输出轮询日志。看到日志里出现库存变化 - in_stock这类信息就说明监控链路已经通了。# 查看监控日志 tail -f monitor.log # 重启监控服务systemd 方式 systemctl restart jd-monitor.service日志文件建议按天切割避免单文件过大。可以在代码里用logging.handlers.TimedRotatingFileHandler设置每天凌晨切换到新文件保留最近 7 天的日志。排查问题的时候历史日志是唯一能还原现场的资料非常重要。4. 常见问题与排查技巧实录4.1 请求频繁被限制怎么办这是监控系统遇到最多的问题。平台对单一 IP 的请求频率有明确限制尤其是登录接口和下单接口短时间大量请求很容易触发风险提示。我的解决方法有四个层级降低轮询频率。这是最直接有效的方法30 秒间隔不行就换 60 秒牺牲一点实时性换来账号稳定。随机化请求间隔。固定间隔容易被识别为机器行为可以在基础间隔上加一个随机偏移量比如30 random.randint(0, 15)秒。配置代理 IP 池。单个 IP 请求上限大约在每分钟 30 次左右超过就容易触发验证。准备几十个住宅代理 IP请求时分发到不同 IP 上能显著降低限制概率。轮换 Cookie。监控请求最好使用一个不常用的“游客 Cookie”或者低权重账号避免主账号因为高频监控被标记。另外要特别注意不要在请求频率降下来之前调试下单功能。我早期为了测试下单每分钟请求了二十多次结果账号被要求短信验证最后花了不少时间才解封。测试时尽量把auto_order关闭用 Mock 数据验证逻辑。4.2 库存误报怎么处理误报是这个系统绕不开的问题。最常见的场景是监控显示“有货”点进去其实没货或者一直显示“无货”但商品明明能买。出现这类问题先排查四个方面缓存干扰。平台页面有 CDN 缓存返回的数据不一定是最新的。解决方式是请求时带一个随机查询参数比如?t1234567890绕过 CDN 缓存。SKU 粒度问题。确认你解析出的库存信息确实属于目标 SKU而不是整个商品维度。有些页面结构里默认显示的是第一个有货 SKU 的信息导致误报。区域库存差异。平台会根据 IP 归属地或收货地址判断库存不同区域看到的库存不一样。确认请求头中的区域参数是否固定。状态字段混淆。有些字段表示“是否可售”有些表示“库存数量”可售为 true 但库存数量为 0 也是常见的组合。解析时要把这些字段含义搞清楚。排查误报有一个好习惯每次监控到库存变化时把当时抓取的原始响应内容存一份快照。这样出现疑义时可以复盘原始数据而不是只看到一个处理后的结果。4.3 自动下单失败率高怎么排查自动下单失败率高是另一个常见痛点。我从实践中总结了一套排查顺序第一步看加购返回。加购失败通常是 Cookie 失效或 SKU 已下架。如果加购接口返回需要登录说明会话过期需要重新走登录流程。第二步看订单确认页。订单确认页里包含的地址列表、支付方式、发票信息、商品清单每一次都可能因为某个字段缺失或格式不正确导致无法提交订单。这一步是失败率最高的环节建议在代码中把确认页的解析结果打日志人工核对关键字段。第三步看提交订单返回。回到之前说过的 token 一次性问题。如果提交订单报“订单已过期”或者“session timeout”多半就是 token 复用了重新访问确认页后再提交即可。最后一步是时序问题。如果前面都正常但订单仍然提交失败可能真的就是商品被别人抢完了。这时候系统应该发一条“下单失败库存被抢完”的通知而不是无脑重试。无脑重试只会增加账号风控风险。我实际统计过一组数据在不做任何优化的情况下从监控到库存发生变化到完成下单平均要花 8 秒到 12 秒。通过预热登录态、提前缓存地址信息、跳过不必要的页面跳转可以把耗时缩短到 3 秒以内成功率有明显提升。4.4 监控进程崩溃和数据积压长时间运行的进程最怕突然挂掉而挂了之后没有任何通知。这个问题我靠两层防护解决一是 systemd 守护进程崩溃后自动重启二是增加一个“心跳告警”机制监控进程每隔 10 分钟向通知渠道发送一条心跳消息如果超过 20 分钟没收到心跳就说明进程异常了。数据积压的问题主要出在数据库写入上。监控频率高的时候如果每次都同步写 MySQL磁盘 I/O 会成为瓶颈。我的方案是异步写入先用 Redis 的LPUSH把数据推到队列里后台单独起一个消费者进程批量写入数据库。这样即使数据库短暂卡顿也不会影响监控主流程的运行。这套系统我持续跑了两个多月帮我稳稳地拿下过几件商品也踩了不少坑。如果你决定部署第一次运行时建议先严格测试轮询、通知和日志记录确认各项功能都符合预期后再打开自动下单开关。任何自动化操作的前提都是合理使用、控制频率、保护账号安全千万别拿它做损害平台生态的事情。代码结构和技术思路是通用的换到任何类似的电商场景都能复用这也是我把它整理成完整项目的原因。本文还有配套的精品资源点击获取 基于Python的电商库存监控与自动下单系统设计与实战
简介这是一套面向Python初学者与电商自动化爱好者的学习型工具源码聚焦京东平台商品库存监控与自动下单场景解决用户在抢购热门商品时需手动刷新、易错过补货时机的痛点。资源包含61个文件以9个核心Python脚本如JdBuyer.py、JdSession.py、timer.py、10个备份配置.zbak、36个地域ID文本覆盖全国34个省级行政区及港澳台、海外、钓鱼岛等特殊区域、以及config.json、README.md、logo.ico等支撑文件总大小仅605KB结构清晰、模块职责分明。已有103人学习下载适合希望掌握HTTP会话管理、定时任务调度、GUI开发基于tkinter及跨平台脚本封装的实践者。读者可直接运行图形界面模式快速上手或通过命令行脚本模式深入理解日志记录、异常捕获与微信/钉钉通知集成逻辑配套使用指南详述环境配置、Cookie获取与参数调优方法。 去年年底我想买一款显卡蹲了半个月硬是没抢到。后来实在受不了手动刷新页面就写了一套自动化监控脚本先实现了库存检测和消息推送后来又逐步补上了自动加购和提交订单的链路最终沉淀成一套可以部署的库存监控与自动下单系统。这里把源码设计和部署使用心得整理出来给有类似需求的读者一个可参考的落地方案。这个项目适合谁如果你经常需要紧盯某个商品的补货时间或者想在商品上架的第一时间获取通知再或者你本身就在做电商数据相关的技术研究这套系统都能派上用场。它不是一个灰产工具箱而是一个典型的数据采集、状态判断、任务调度和自动化操作的实战项目技术上涉及 requests、Redis、定时任务、Web 管理端读完你会对这类系统有完整的认知。1. 需求拆解与整体方案设计1.1 库存监控到底在监控什么很多人以为库存监控就是定时请求一下商品详情页判断“有货”还是“无货”。真做起来就会发现这里面的细节比想象中多。一个商品链接往往对应多个 SKU比如手机有黑色、白色、不同内存版本每个 SKU 的库存状态是独立的。只看商品主图页面的库存字段可能永远显示有货但具体到你想要的那个颜色其实一直缺货。所以监控的粒度必须下沉到 SKU 维度至少要拿到商品 ID、SKU ID、规格名、实时价格和库存数量。另外电商平台的库存返回通常不是单一字段。有些接口返回的是“可售状态”有些返回“库存数量”有些则要区分“现货”和“预售”还有的平台会有区域库存概念不同收货地址看到的库存不一样。这些因素都会影响系统判断。我在最初版本里只读取了一个字段结果同一商品在不同账号下看到的状态完全不同排查了很久才发现是区域参数导致的。设计监控规则时一定要把“哪些字段变化才算是真正的库存变化”想清楚否则会收到大量无效告警。1.2 自动下单系统的核心难点自动下单的上手门槛比监控高一个量级主要难点有三块。第一是链路长。一次完整的下单至少要经过登录态校验、加入购物车、访问结算页、获取订单确认信息、提交订单。每一步都有独立的接口接口之间还可能依赖上一步返回的 token、订单号、地址 ID 等参数。任何一个环节出错整个流程都要重新来。第二是风控拦截。平台对高频请求、异常行为、非人工操作有比较成熟的识别策略。短时间频繁加购、提交订单很容易触发验证码、滑块甚至账号限制。这套系统如果追求高并发下单就必须面对风控而风控策略往往是黑盒只能靠降低频率、模拟人工节奏、轮换账号和 IP 来缓解。第三是时序竞争。你监控到“有货”的同时可能已经有成千上万个用户也在下单商品从“有货”到“已售罄”可能只需要几秒钟。系统不仅要监控得快还要在下单环节尽量缩短时间损耗哪怕省掉一次无效的页面跳转成功率都能高一些。所以我最终把系统拆成两个独立的模块监控模块和下单模块。监控模块独立运行负责轮询库存、记录数据、发送通知下单模块默认关闭只有在人工确认开启时才执行自动加购和提交订单。这样的设计既保证了日常监控的稳定性也降低了误操作带来的账号风险。1.3 技术选型为什么这么搭技术栈上我选了 Python 作为主力开发语言理由很简单生态成熟、写起来快、调试方便requests/httpx 做 HTTP 请求足够用配合 BeautifulSoup 或正则就能解析页面内容。任务调度这块我没有引入太重的东西。最开始用的是 Python 的 threading.Timer 自己做定时循环简单但不好管理重启、暂停、修改监控频率都得改代码。后来换成 APScheduler支持 cron 表达式、后台运行、持久化任务已经能满足大部分监控场景。再配合 Celery 可以实现分布式任务队列但如果只是单机跑Celery 有点杀鸡用牛刀建议先从 APScheduler 入手。数据存储分两部分Redis 用来做实时状态缓存和接口幂等控制MySQL 用来存历史库存记录、通知日志和商品配置。Redis 的 key 设计成stock:{sku_id}value 直接存 JSON 字符串包含库存状态、价格、更新时间。这样监控模块每次轮询后只要对比 Redis 里的旧状态就知道有没有变化避免每次变化都写数据库。Web 管理端我选了 Flask原因就是轻ORM 用 SQLAlchemy配合一个简单的前端页面就能实现对商品监控任务、下单开关、通知渠道的管理。如果你对 Spring Boot 那套更熟悉也可以把核心服务用 Java 重写架构思路完全一致只是请求库和调度库需要换成对应的技术组件。这里放一个组件选型表方便参考模块技术选型用途HTTP 请求requests / httpx封装商品详情、加购、订单提交接口页面解析BeautifulSoup / 正则从 HTML 中提取 SKU、价格、库存信息任务调度APScheduler管理每个商品的轮询任务缓存Redis状态去重、短时间防重复下单数据库MySQL / SQLite持久化商品配置、历史库存、操作日志管理端Flask SQLAlchemy商品配置、监控日志、任务控制通知SMTP / Server酱 / 钉钉机器人向手机推送库存变化和下单结果这套组合的优点是上手门槛低、部署成本小单台 1 核 2G 的云服务器就能跑得很稳。缺点是 Python 的 GIL 限制了对多核 CPU 的利用不过监控任务本身就是 I/O 密集型的网络请求把并发放在协程或者线程池里解决就够了GIL 影响不大。2. 核心模块设计与源码解析2.1 监控模块轮询、状态判断与数据记录监控模块是整个系统的心脏。它的核心逻辑并不复杂定时请求商品数据解析出自己想要的信息和上一次记录做对比发生变化就处理没变化就继续安静地等待下个周期。轮询间隔怎么定是个值得琢磨的问题。间隔太短容易给平台造成压力也可能触发风控间隔太长又可能错过补货窗口。我的经验是普通商品 30 秒到 60 秒轮询一次就够如果商品非常抢手可以降低到 5 秒到 10 秒但这时候必须搭配代理 IP 和账号池否则很容易被限制登录态。下面是一段核心轮询代码import requests import json import redis import time session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://item.jd.com/, }) r redis.Redis(hostlocalhost, port6379, db0) def fetch_sku_info(sku_id): url fhttps://item.jd.com/{sku_id}.html resp session.get(url, timeout10) # 实际项目中会从 HTML 或内嵌 JSON 中解析出价格、库存等信息 # 这里用模拟数据展示结构 info { sku_id: sku_id, stock: in_stock, # in_stock / out_of_stock price: 2999.00, ts: int(time.time()), } return info def check_and_notify(sku_id): data fetch_sku_info(sku_id) cache_key fstock:{sku_id} old r.get(cache_key) if old is None or json.loads(old)[stock] ! data[stock]: r.set(cache_key, json.dumps(data)) print(f[{sku_id}] 库存变化 - {data[stock]}) # 触发通知、记录数据库等后续逻辑 if __name__ __main__: while True: check_and_notify(100012043978) time.sleep(30)这段代码展示的是一个最简原型生产环境里需要把请求重试、异常捕获、代理切换、多 SKU 遍历这些逻辑补全。我实际使用中踩过的一个坑是有些接口返回的不是 JSON而是 JavaScript 变量赋值语句比如window._itemInfo {...}直接 json.loads 会报错。这时候需要先用正则把大括号内的内容提取出来再解析或者干脆用 BeautifulSoup 解析 HTML 里隐藏的 data 属性。库存数据的落库我采用了按天分表或定期清理的策略。每 30 秒一条监控记录一个 SKU 一天就能产生 2880 条数据如果不加清理三个月后数据库就堆满了。我只保留变化记录也就是只有在库存状态、价格发生变化时才写一条历史记录这样数据量能减少一个数量级。2.2 自动下单模块加购、结算与提交订单自动下单模块的核心思路是模拟人工操作的完整链路。以京东的网页端为例一次购买包含以下几个关键步骤登录并保持会话、把商品加入购物车、访问购物车页面获取结算信息、提交订单并获取订单号。每一步都有对应的接口且这些接口需要携带登录后的 Cookie 和 Token。我建议把每一步拆成独立的函数方便单独调试和失败重试。比如加购失败时不需要重新走登录流程只需要重新执行加购那一步函数即可。def add_to_cart(session, sku_id, count1): url https://cart.jd.com/gate.action params { pid: sku_id, pcount: count, ptype: 1, } resp session.get(url, paramsparams, timeout10) return resp.status_code 200 def submit_order(session, address_id, pay_type4): url https://trade.jd.com/shopping/order/submitOrder.action data { address: address_id, payType: pay_type, # 订单确认页返回的 token 和订单号等参数 } resp session.post(url, datadata, timeout10) result resp.json() if result.get(success): return result[orderId] else: raise RuntimeError(result.get(message, submit order failed))这里有一个非常关键的细节提交订单接口通常需要携带一个从订单确认页面解析出来的 token而且这个 token 是一次性的用一次就失效。所以每次提交前必须重新访问订单确认页拿到最新的 token不能把这个 token 缓存后再用。否则会一直报“订单信息过期”或者“请重新下单”。另外一个实际经验是提交订单前最好做一次库存二次确认。因为从监控模块发现库存变化到执行下单中间可能隔着几十秒商品很可能已经被别人抢完了。提前调用一次库存查询接口确认当前 SKU 仍然有货再执行加购和提交可以在一定程度上减少无效请求降低被风控盯上的概率。自动下单模块的触发策略我也踩过不少坑。最初版本是监控发现“有货”后立刻自动下单结果经常误判。后来加了“连续 N 次有货才触发”的规则比如连续 3 次轮询都显示有货才认为是真实补货这才把误报率降下来。这个 N 的值要结合轮询间隔调整如果间隔是 30 秒连续 3 次就是观察了 90 秒如果间隔是 5 秒连续 3 次只观察了 15 秒误报可能性高一些。2.3 库存状态机与通知体系设计库存监控不能只处理“有货”和“无货”两种状态。在很多场景下商品还会处于“预售”“限购”“区域无货”“价格波动”等中间状态。我在系统中抽象了一个状态机每个 SKU 在任何时刻都处于几种有限状态之一状态含义触发动作OUT_OF_STOCK无货等待下一次轮询IN_STOCK有货可下单发送通知、按策略执行下单PRE_SALE预售中发送预售通知DECOMMISSIONED商品已下架停用监控任务并告警PRICE_CHANGED价格变化单独发送价格变动通知状态机的优势是逻辑清晰、便于扩展。比如你可以在 IN_STOCK 状态上配置“自动下单”标志位在 PRICE_CHANGED 状态上配置“低于目标价才提醒”的阈值。这样整个系统就像搭积木一样每个状态对应一组处理动作互相不干扰。通知模块我同时支持了三种渠道邮件、Server酱微信推送、钉钉机器人。日常使用中Server酱的微信推送体验最好延迟低、免装 App钉钉机器人适合团队内部做协作通知邮件则作为兜底方案防止第三方推送服务临时不可用。每种渠道的配置都放在配置文件中启用哪个就填哪个的 API Key。为了不把通知消息做成“轰炸机”我增加了消息聚合逻辑同一 SKU 在五分钟内的变化只推送一条消息避免高频率轮询时手机响个不停。3. 源码部署与使用指南3.1 环境准备与依赖安装部署这套系统不需要多高的服务器配置一台能跑 Python 的 Linux 机器就够了。我自己用的是 1 核 2G 内存的轻量服务器同时跑 Redis、MySQL、Flask 和监控脚本负载一直很稳定。操作系统建议 Ubuntu 20.04 或 CentOS 7 以上版本Python 版本建议 3.8 到 3.10。依赖安装直接用 pip 装pip install flask flask-sqlalchemy pymysql requests beautifulsoup4 redis apscheduler如果你不想在服务器上安装 MySQL可以先用 SQLite 顶替单机项目的数据量完全够用。SQLAlchemy 的配置只需要改一行连接串从mysqlpymysql://user:passlocalhost/dbname换成sqlite:///data.db其余代码不用动。这种方式适合快速验证系统逻辑确认稳定后再切换到 MySQL。Redis 是必须要装的。它是一个轻量级的内存数据库安装方式apt install redis-server systemctl enable redis-server systemctl start redis-serverRedis 在这个系统里主要做两件事一是保存每个 SKU 的最新库存快照用于状态对比二是实现分布式锁防止多个监控进程对同一个 SKU 重复下单。这两件事如果用文件或数据库来做不是不行但性能和可靠性都会差一些。3.2 配置文件详解我把所有可变参数都集中在一个config.yaml文件里这样修改监控频率、添加新商品、切换通知渠道都不用改代码。下面是一份典型配置monitor: interval: 30 # 全局轮询间隔单位秒 retry_times: 3 # 请求失败重试次数 timeout: 10 # 单次请求超时时间单位秒 user_agents: - Mozilla/5.0 (Windows NT 10.0; Win64; x64) - Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) notify: enable: true channels: - type: serverchan key: your_sendkey - type: email smtp_host: smtp.qq.com smtp_port: 465 username: yourqq.com password: your_auth_code receivers: - receiverexample.com - type: dingtalk webhook: https://oapi.dingtalk.com/robot/send?access_tokenxxx order: auto_order: false # 总开关默认关闭自动下单 trigger_count: 3 # 连续 N 次有货才触发 pay_type: 4 # 支付方式4 在线支付 address_id: 123456789配置项里面最危险的就是auto_order这个开关。日常跑监控建议保持false只做通知确认商品确实进入稳定补货期后再临时打开自动下单。我自己一般会先关着跑两天观察通知的准确率再决定开不开自动下单。商品列表我放在数据库里通过 Web 管理端添加也可以直接写 SQL 插入INSERT INTO product (sku_id, name, monitor_interval, notify_enabled) VALUES (100012043978, RTX 4070 显卡, 30, 1), (100016034872, 某品牌内存条 32G, 60, 1);每个商品可以单独设置监控间隔这个设计在实战中非常有用。热门款用短间隔冷门款用长间隔既保证了灵敏度又减少了整体请求量。3.3 启动流程与常用命令启动分三步。第一步是初始化数据库建好表结构第二步是启动 Redis如果还没启动的话第三步是运行监控主程序。# 初始化数据库 python init_db.py # 启动监控主进程前台运行方便看日志 python monitor.py # 启动 Web 管理端后台运行 nohup python web_admin.py --port 8080 web.log 21 如果你希望监控主进程也保持后台运行推荐用 systemd 写一个服务文件或者用nohup丢到后台。不建议用screen或者tmux来守护因为服务器一重启进程就丢了还得手动恢复。systemd 可以设置开机自启进程崩了还能自动拉起更适合长期运行的服务。监控进程起来后正常情况下会循环输出轮询日志。看到日志里出现库存变化 - in_stock这类信息就说明监控链路已经通了。# 查看监控日志 tail -f monitor.log # 重启监控服务systemd 方式 systemctl restart jd-monitor.service日志文件建议按天切割避免单文件过大。可以在代码里用logging.handlers.TimedRotatingFileHandler设置每天凌晨切换到新文件保留最近 7 天的日志。排查问题的时候历史日志是唯一能还原现场的资料非常重要。4. 常见问题与排查技巧实录4.1 请求频繁被限制怎么办这是监控系统遇到最多的问题。平台对单一 IP 的请求频率有明确限制尤其是登录接口和下单接口短时间大量请求很容易触发风险提示。我的解决方法有四个层级降低轮询频率。这是最直接有效的方法30 秒间隔不行就换 60 秒牺牲一点实时性换来账号稳定。随机化请求间隔。固定间隔容易被识别为机器行为可以在基础间隔上加一个随机偏移量比如30 random.randint(0, 15)秒。配置代理 IP 池。单个 IP 请求上限大约在每分钟 30 次左右超过就容易触发验证。准备几十个住宅代理 IP请求时分发到不同 IP 上能显著降低限制概率。轮换 Cookie。监控请求最好使用一个不常用的“游客 Cookie”或者低权重账号避免主账号因为高频监控被标记。另外要特别注意不要在请求频率降下来之前调试下单功能。我早期为了测试下单每分钟请求了二十多次结果账号被要求短信验证最后花了不少时间才解封。测试时尽量把auto_order关闭用 Mock 数据验证逻辑。4.2 库存误报怎么处理误报是这个系统绕不开的问题。最常见的场景是监控显示“有货”点进去其实没货或者一直显示“无货”但商品明明能买。出现这类问题先排查四个方面缓存干扰。平台页面有 CDN 缓存返回的数据不一定是最新的。解决方式是请求时带一个随机查询参数比如?t1234567890绕过 CDN 缓存。SKU 粒度问题。确认你解析出的库存信息确实属于目标 SKU而不是整个商品维度。有些页面结构里默认显示的是第一个有货 SKU 的信息导致误报。区域库存差异。平台会根据 IP 归属地或收货地址判断库存不同区域看到的库存不一样。确认请求头中的区域参数是否固定。状态字段混淆。有些字段表示“是否可售”有些表示“库存数量”可售为 true 但库存数量为 0 也是常见的组合。解析时要把这些字段含义搞清楚。排查误报有一个好习惯每次监控到库存变化时把当时抓取的原始响应内容存一份快照。这样出现疑义时可以复盘原始数据而不是只看到一个处理后的结果。4.3 自动下单失败率高怎么排查自动下单失败率高是另一个常见痛点。我从实践中总结了一套排查顺序第一步看加购返回。加购失败通常是 Cookie 失效或 SKU 已下架。如果加购接口返回需要登录说明会话过期需要重新走登录流程。第二步看订单确认页。订单确认页里包含的地址列表、支付方式、发票信息、商品清单每一次都可能因为某个字段缺失或格式不正确导致无法提交订单。这一步是失败率最高的环节建议在代码中把确认页的解析结果打日志人工核对关键字段。第三步看提交订单返回。回到之前说过的 token 一次性问题。如果提交订单报“订单已过期”或者“session timeout”多半就是 token 复用了重新访问确认页后再提交即可。最后一步是时序问题。如果前面都正常但订单仍然提交失败可能真的就是商品被别人抢完了。这时候系统应该发一条“下单失败库存被抢完”的通知而不是无脑重试。无脑重试只会增加账号风控风险。我实际统计过一组数据在不做任何优化的情况下从监控到库存发生变化到完成下单平均要花 8 秒到 12 秒。通过预热登录态、提前缓存地址信息、跳过不必要的页面跳转可以把耗时缩短到 3 秒以内成功率有明显提升。4.4 监控进程崩溃和数据积压长时间运行的进程最怕突然挂掉而挂了之后没有任何通知。这个问题我靠两层防护解决一是 systemd 守护进程崩溃后自动重启二是增加一个“心跳告警”机制监控进程每隔 10 分钟向通知渠道发送一条心跳消息如果超过 20 分钟没收到心跳就说明进程异常了。数据积压的问题主要出在数据库写入上。监控频率高的时候如果每次都同步写 MySQL磁盘 I/O 会成为瓶颈。我的方案是异步写入先用 Redis 的LPUSH把数据推到队列里后台单独起一个消费者进程批量写入数据库。这样即使数据库短暂卡顿也不会影响监控主流程的运行。这套系统我持续跑了两个多月帮我稳稳地拿下过几件商品也踩了不少坑。如果你决定部署第一次运行时建议先严格测试轮询、通知和日志记录确认各项功能都符合预期后再打开自动下单开关。任何自动化操作的前提都是合理使用、控制频率、保护账号安全千万别拿它做损害平台生态的事情。代码结构和技术思路是通用的换到任何类似的电商场景都能复用这也是我把它整理成完整项目的原因。本文还有配套的精品资源点击获取