在 Python 自动化脚本的开发里最常见的一类需求就是“到点之后自动查一个资源状态发现有余量就立刻提醒自己”。很多人会把它进一步扩展成自动抢票脚本甚至希望做到“看到有票就自动下单”。这里要先把边界说清楚自动检测余票、自动发通知属于学习 Python 自动化的正当场景但自动绕过登录、验证码、频率限制和订单接口去代下单既违反票务平台规则也不适合作为技术教程传播。这篇文章不会提供“成功率 100%”的抢票脚本而是用一套可复现的本地模拟接口实现一个合规的 Python 余票监控提醒脚本。它跑通之后也能迁移到库存提醒、价格监控、考试名额提醒等场景。1. 先拆需求自动检测和自动下单是两件完全不同的事1.1 为什么“抢票脚本”不适合作为教程主题“抢票”这个功能拆开看其实包含三个阶段查余票、发现有余票、替用户提交订单并完成支付。前两个阶段是信息获取和通知技术上确实值得学习第三个阶段涉及用户登录态、验证码、加密参数、接口频控、订单并发甚至支付环节强行自动化会带来几个现实问题。第一平台规则通常不允许脚本批量提交订单。自动抢票工具一旦被发现轻则请求被限流或封禁账号重则影响正常用户购票体验。第二这类脚本的难点根本不是 Python 语法而是如何在平台风控面前“伪装成正常用户”这个方向已经明确越界。第三任何声称“成功率 100%”的方案都不成立因为余票数量、服务器响应时间、验证码策略和账号状态都在动态变化没有任何本地脚本能保证必中。所以比较合理的学习路线是把自动化能力限制在“查票并提醒”这一步再由用户本人决定是否继续购票。这样既能练到 HTTP 请求、定时任务、异常重试、日志记录等实用技能又不会触碰平台规则。1.2 合规目标余票提醒而不是代下单本文要实现的脚本本质是一个轮询式监控器。它的工作流程是每隔一段时间请求一次票务余票接口。解析接口返回的 JSON判断活动是否还有余票。如果状态从“无票”切换为“有余票”输出一条提醒日志。如果状态没有变化不做任何输出避免重复刷屏。请求失败时自动重试有限次数并记录错误日志。这套流程不涉及登录态不绕过验证码不重复提交订单也不影响接口正常服务。用到的接口是本地模拟出来的读者可以先理解全流程再替换成自己具备合法调用权限的业务接口。1.3 本文成品效果和使用场景完成这篇文章的练习后你会得到三个文件一个 Flask 模拟票务服务、一个 HTTP 查询客户端、一个定时监控入口。运行python server_mock.py启动模拟服务后再运行python monitor.py --event-id 1001 --interval 10脚本就会每 10 秒查询一次示例活动 1001 的余票。当模拟活动从有剩余余票变为无票或者从无票变为有余票时终端会输出对应状态变化。在真实项目中这套流程可以用于内部系统资源池剩余量监控。学习用公开 API 的告警提醒。自己开发的业务系统定时巡检。考试报名名额、库存状态的简单提醒。它不适合也不应该用于对外部商业平台的高频撞票、自动下单。2. Python 环境准备先解决命令不识别和依赖混乱2.1 安装 Python 并确认环境变量在开始之前先确认本机是否已经安装 Python。终端里分别执行以下命令python --version pip --version如果输出的是类似Python 3.11.8和pip 23.2.1的内容说明环境正常。如果出现下面这行提示python : 无法将“python”项识别为 cmdlet、函数、脚本文件或可运行程序的名称。请检查名称的拼写如果存在路径请确保路径正确然后再试一次。原因通常是安装 Python 时没有勾选“Add Python to PATH”或者安装之后终端没有重启。Windows 下安装时可以到 Python 官网下载安装包第一屏注意勾选最下方的Add python.exe to PATH再点 Install Now。安装完成后重新打开一个终端窗口再执行验证命令。2.2 创建虚拟环境隔离项目依赖Python 项目最怕全局环境里的包互相冲突。建议每个项目都使用虚拟环境。在项目目录下执行mkdir ticket-monitor cd ticket-monitor python -m venv .venvWindows 下激活虚拟环境.\.venv\Scripts\activateLinux 或 macOS 下激活虚拟环境source .venv/bin/activate激活后终端命令提示符前面会出现(.venv)表示当前使用的解释器已经切换到虚拟环境内部。如果使用 VSCode还需要按CtrlShiftP输入Python: Select Interpreter选择项目里的.venv路径否则编辑器仍可能使用全局解释器。2.3 安装 Flask、requests 和 schedule本文代码依赖三个库Flask用来搭建本地模拟票务接口。requests用来在监控脚本里发送 HTTP 请求。schedule用来实现轻量级定时循环。在项目目录下创建requirements.txtflask3.0,4 requests2.31,3 schedule1.2,2然后安装pip install -r requirements.txt安装完成后可以用pip list检查三个包是否已经出现在列表里。如果下载速度慢可以临时使用国内镜像源pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple2.4 环境检查清单检查项命令预期结果Python 命令python --version输出 Python 3.x 版本pip 命令pip --version输出 pip 版本和 Python 路径虚拟环境Windows 下.\.venv\Scripts\activate命令提示符前出现(.venv)依赖安装pip list能看到 flask、requests、schedule项目解释器VSCode 中查看右下角解释器显示.venv路径实际开发中版本号会持续更新。如果遇到依赖兼容问题可以把版本号放宽让 pip 自动选择可用版本不要在一开始就锁死过于生僻的版本组合。3. 搭一个本地模拟票务 API把“查余票”变成 HTTP 调用3.1 为什么要先使用本地模拟接口真实的商业票务接口通常不会公开给普通用户直接调用。很多平台接口需要登录态、签名参数甚至验证码直接抓包去逆向这些内容已经超出“学习 Python 自动化”的安全边界。本地模拟接口的价值在于先把“监控脚本如何发请求、如何解析响应、如何处理异常”这套链路跑通之后再替换成自己有权调用的接口时只需要改 URL 和字段解析逻辑即可。不要把精力浪费在破解接口签名上那是另一个完全不同的领域而且很容易触碰平台风控底线。3.2 用 Flask 写一个最小的余票查询接口在项目目录下创建server_mock.pyfrom flask import Flask, jsonify, request import threading import time app Flask(__name__) ticket_pool { 1001: {name: 示例音乐会-周末场, total: 500, sold: 497}, 1002: {name: 示例话剧-晚场, total: 300, sold: 300}, } app.get(/api/ticket/check) def ticket_check(): event_id request.args.get(eventId, 1001) event ticket_pool.get(event_id) if not event: return jsonify({ code: 404, message: event not found, data: None }), 404 left event[total] - event[sold] return jsonify({ code: 0, message: ok, data: { eventId: event_id, name: event[name], ticketLeft: max(left, 0), serverTime: int(time.time()), }, }) def simulate_sell(): # 每 5 秒卖掉一张票观察余票变化用 while True: time.sleep(5) if ticket_pool[1001][sold] ticket_pool[1001][total]: ticket_pool[1001][sold] 1 if __name__ __main__: threading.Thread(targetsimulate_sell, daemonTrue).start() app.run(host127.0.0.1, port5000, debugFalse)这段代码包含一个内存字典ticket_pool其中total是总票数sold是已售数量。simulate_sell函数每 5 秒把活动 1001 的已售数量加 1用于模拟余票被逐渐买走的过程。接口设计了一个常见的响应结构{ code: 0, message: ok, data: { eventId: 1001, name: 示例音乐会-周末场, ticketLeft: 3, serverTime: 1730000000 } }实际项目中的接口结构各不相同但大多数都会包含业务状态码和具体数据。监控脚本只关心code 0并且data.ticketLeft 0两个条件。3.3 编写 HTTP 查询客户端超时、重试和异常处理在项目目录下创建ticket_client.pyimport logging import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) logger logging.getLogger(ticket_client) def create_session(): session requests.Session() retry Retry( total2, connect2, read2, status2, backoff_factor0.5, status_forcelist(500, 502, 503, 504), ) session.mount(http://, HTTPAdapter(max_retriesretry)) session.mount(https://, HTTPAdapter(max_retriesretry)) return session def query_ticket(event_id1001, base_urlhttp://127.0.0.1:5000, sessionNone): session session or create_session() url base_url.rstrip(/) /api/ticket/check try: resp session.get( url, params{eventId: event_id}, timeout(3, 5), ) resp.raise_for_status() try: payload resp.json() except ValueError: logger.error(接口返回内容不是 JSON: %s, resp.text[:200]) return None if payload.get(code) ! 0: logger.warning(接口返回业务错误: %s, payload) return None return payload.get(data) except requests.exceptions.Timeout: logger.error(请求超时: %s, url) except requests.exceptions.ConnectionError: logger.error(连接失败: %s, url) except requests.exceptions.HTTPError as exc: logger.error(HTTP 状态错误: %s, exc) return None if __name__ __main__: print(query_ticket(1001))这里有一个关键设计同一个requests.Session会被复用因为 HTTP 连接池可以复用底层 TCP 连接避免每次请求都重新建立连接。Retry表示最多重试 2 次backoff_factor0.5表示重试间隔会按 0.5、1、2 秒递增。timeout(3, 5)表示连接超时 3 秒读取超时 5 秒。没有设置超时的请求一旦遇到网络异常可能会长时间阻塞这是监控脚本的大忌。3.4 运行模拟服务并验证查询结果先启动模拟服务python server_mock.py看到 Flask 的输出后在另一个终端执行python ticket_client.py正常输出是{eventId: 1001, name: 示例音乐会-周末场, ticketLeft: 3, serverTime: 1730000000}如果先停止模拟服务再执行python ticket_client.py会看到ERROR ticket_client 连接失败: http://127.0.0.1:5000/api/ticket/check说明异常处理已经生效。这个最小闭环跑通之后剩下的事情就是加上定时调用和状态切换。4. 加入定时轮询和状态切换让监控脚本真正“自动运行”4.1 使用 schedule 而不是裸写 while True 做定时很多入门代码会写成while True: query_ticket(1001) time.sleep(10)这种写法的问题在于如果query_ticket内部出现阻塞整个循环都会受到拖累。schedule库的优势是可以用声明式的方式安排任务同时在每个循环周期里统一执行所有到期任务。import schedule import time schedule.every(10).seconds.do(job) while True: schedule.run_pending() time.sleep(1)这段代码本身仍然有一个while True但time.sleep(1)只负责让出 CPU真正的执行周期由schedule.every(...).seconds.do(...)来控制代码可读性更好也方便以后增加多个定时任务。4.2 用状态字典消除重复提醒如果每次发现“有余票”都打印一次脚本运行一小时会刷出成百上千行日志。实际情况是只有状态发生跳变时才需要提醒。状态切换的核心逻辑是记录每个活动上一次是否还有余票当前结果和上一次不同才输出提醒。last_state {} def check_and_notify(event_id, session): data query_ticket(event_id, sessionsession) if data is None: return left int(data.get(ticketLeft, 0)) event_name data.get(name, event_id) previous last_state.get(event_id, False) if left 0 and not previous: send_notification(f{event_name} 有余票剩余 {left} 张) elif left 0 and previous: send_notification(f{event_name} 已无票) last_state[event_id] left 0在模拟场景下初始状态为无票时第一次查到有余票会提醒后续每次查询仍有余票也不会继续提醒因为previous已经是True。4.3 通知模块先保留统一入口为了不让代码耦合到具体通知平台可以单独写一个send_notification函数def send_notification(text): # 本地先输出日志生产可替换为企业微信机器人、钉钉机器人、Server酱等 logging.info([通知] %s, text)真正接入通知平台时只需要修改这个函数内部的实现。例如企业微信机器人通常需要一个 webhook 地址用requests.post把消息发过去即可。不要把 webhook 地址硬编码到代码里应该通过环境变量或配置文件传入。4.4 完整的监控脚本入口创建monitor.pyimport argparse import logging import time import schedule from ticket_client import create_session, query_ticket logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) logger logging.getLogger(monitor) last_state {} def send_notification(text): logger.info([提醒] %s, text) def check_and_notify(event_id, base_url, session): data query_ticket( event_idevent_id, base_urlbase_url, sessionsession, ) if data is None: logger.error(活动 %s 查询失败本次跳过, event_id) return left int(data.get(ticketLeft, 0)) event_name data.get(name, event_id) previous last_state.get(event_id, False) if left 0 and not previous: send_notification(f{event_name} 有余票剩余 {left} 张) elif left 0 and previous: logger.info(%s 从有票变为无票, event_name) last_state[event_id] left 0 def parse_args(): parser argparse.ArgumentParser(description余票监控提醒脚本) parser.add_argument(--base-url, defaulthttp://127.0.0.1:5000) parser.add_argument(--event-id, default1001) parser.add_argument(--interval, typeint, default10) return parser.parse_args() def main(): args parse_args() logger.info( 监控启动: base_url%s event_id%s interval%ss, args.base_url, args.event_id, args.interval, ) session create_session() schedule.every(args.interval).seconds.do( check_and_notify, event_idargs.event_id, base_urlargs.base_url, sessionsession, ) try: while True: schedule.run_pending() time.sleep(1) except KeyboardInterrupt: logger.info(监控已手动停止) if __name__ __main__: main()启动监控python monitor.py --event-id 1001 --interval 10模拟服务端每 5 秒会卖出一张票活动 1001 原本只有 3 张余票所以在脚本运行一段时间后终端会看到状态从“有票”变成“无票”的变化。注意不要在同一终端同时运行服务和监控脚本需要开两个终端窗口。5. 从能跑到可运维配置、日志与常驻进程5.1 用配置文件管理环境差异在上面的脚本里地址和活动编号需要通过命令行参数传入。真实项目里更常见的是写一个config.ini文件把容易变化的内容集中管理。创建config.ini[ticket] base_url http://127.0.0.1:5000 event_id 1001 interval_seconds 60 [log] level INFO file logs/monitor.logmonitor.py中加载配置import configparser config configparser.ConfigParser() config.read(config.ini, encodingutf-8) base_url config.get(ticket, base_url) event_id config.get(ticket, event_id) interval config.getint(ticket, interval_seconds)引入配置文件之后开发环境、测试环境、生产环境之间只需要维护不同的配置内容不需要修改代码。登录凭据、webhook 地址、密钥这些敏感信息不要放进配置文件并提交到代码仓库应该使用环境变量或密钥管理服务。5.2 用日志文件记录关键链路print只适合临时调试。长期运行的监控脚本需要把日志写入文件并且按大小自动轮转否则日志文件会越来越大。import logging from logging.handlers import RotatingFileHandler logger logging.getLogger(monitor) handler RotatingFileHandler( logs/monitor.log, maxBytes5 * 1024 * 1024, backupCount3, encodingutf-8, ) formatter logging.Formatter(%(asctime)s %(levelname)s %(name)s %(message)s) handler.setFormatter(formatter) logger.addHandler(handler) logger.setLevel(logging.INFO)上面的配置表示单个日志文件最大 5 MB保留最近 3 个备份文件。每次启动脚本前建议先确保logs目录存在否则日志文件会创建失败。5.3 常驻运行Windows 任务计划与 Linux systemd学习阶段可以直接让脚本在终端里跑但生产环境必须考虑进程守护和开机自启。Windows 下可以使用任务计划程序创建基本任务。触发器选择“计算机启动时”或“每天”。操作选择“启动程序”。程序填写虚拟环境中的python.exe参数填写monitor.py的完整路径起始目录填写项目目录。Linux 下推荐使用 systemd。创建/etc/systemd/system/ticket-monitor.service[Unit] DescriptionTicket Monitor Service Afternetwork.target [Service] WorkingDirectory/opt/ticket-monitor ExecStart/opt/ticket-monitor/.venv/bin/python /opt/ticket-monitor/monitor.py Restartalways RestartSec10 [Install] WantedBymulti-user.target启动服务sudo systemctl daemon-reload sudo systemctl start ticket-monitor sudo systemctl enable ticket-monitor当进程意外退出时Restartalways会让 systemd 自动拉起来。这比手工在终端里盯着日志要可靠得多。5.4 学习环境与生产环境的主要差异维度学习环境生产环境运行方式前台终端运行systemd 或任务计划守护轮询间隔5 到 10 秒便于观察至少 60 秒以上避免限流日志终端输出滚动日志文件 监控告警配置命令行参数配置文件 环境变量异常恢复手动重启自动重启 重试退避通知日志或控制台企业微信、钉钉、邮件等历史记录内存状态SQLite 或数据库持久化生产环境里的监控脚本还需要考虑接口是否允许轮询、单次任务执行时长、重复任务堆积、时间同步和告警风暴等问题。不要因为本地跑通了就直接部署到公网服务器上。6. 高频问题排查从命令行报错到脚本行为不符合预期6.1 PowerShell 提示无法识别 python、pip 或相关命令现象python : 无法将“python”项识别为 cmdlet、函数、脚本文件或可运行程序的名称排查顺序先看安装是否成功执行where.exe python。如果输出为空说明 Python 没有安装或没有加入 PATH。Windows 下也可以尝试py --version因为部分安装方式只提供了py启动器。使用虚拟环境时不要直接输入python可以执行.venv\Scripts\python.exe --version。PowerShell 激活脚本报错时需要执行Set-ExecutionPolicy -Scope CurrentUser RemoteSigned修改当前用户的脚本执行策略。pip 命令出现同样问题时优先使用python -m pip --version因为这种写法可以避免调用到错误路径下的 pip。6.2 接口 404、请求超时或返回内容不是 JSON现象ERROR ticket_client 请求超时 ERROR ticket_client 连接失败 ERROR ticket_client 接口返回内容不是 JSON排查顺序先确认模拟服务是否启动访问http://127.0.0.1:5000/api/ticket/check?eventId1001。如果浏览器能看到 JSON说明服务和接口路径正常。在控制器台上使用curl http://127.0.0.1:5000/api/ticket/check?eventId1001确认返回结构。查看服务端终端日志有没有报错比如端口被占用、Flask 启动失败。检查客户端base_url是否写错比如多加了末尾斜杠或者写成了https。如果返回 404确认接口路径是/api/ticket/check而不是/api/ticket/。返回内容不是 JSON 时日志里应该打印响应内容的前 200 个字符方便快速判断是不是被网关拦截或返回了 HTML。监控脚本里不要直接resp.json()而不处理异常否则一旦接口返回异常 HTML整个任务会抛出未捕获异常。6.3 定时任务不执行、重复提醒或进程退出常见原因有三个第一任务函数内部抛出了未捕获异常导致schedule.run_pending()所在的循环退出。解决办法是在任务函数入口增加try/except Exception保证单次失败不影响后续轮询。第二状态只保存在内存里脚本一旦重启之前的“已提醒”状态就丢了。如果重启后余票仍然大于 0脚本会再次提醒一次。这在提醒场景下可以接受如果不想重复提醒可以把状态写入 SQLite 或本地 JSON 文件。第三轮询间隔设置得过短或者任务本身执行时间超过间隔时间导致任务堆积。使用schedule时如果任务执行时间很长应考虑用线程池隔离任务或者使用 APScheduler 等更完整的调度库。6.4 常见报错速查表问题现象可能原因检查方式处理方案python 命令不存在未安装或未配置 PATHwhere.exe python重新安装并勾选 Add
