最近在音游玩家群里经常看到一句调侃明年才能打舞萌了。有人是因为常去的电玩城机台数量太少有人是周末排队时间动辄一两个小时也有人好不容易到店却发现机台全部处于维护状态。玩笑归玩笑这类问题背后的核心诉求其实很明确如何在机台空闲的第一时间就收到提醒避免反复到店扑空或长时间守在旁边排队。与其被动等待不如做一个面向机台状态监控与自动提醒的小系统。本文将从需求分析、系统设计、代码实现到运行验证完整演示一个 Python 监控工具的实现过程。你可以将它理解为“舞萌 DX 机台空闲提醒助手”稍加改造后也能用于其他街机、共享设备或展厅设备的在线状态监控。整个项目基于 Python 与 Flask 搭建不依赖复杂框架适合刚开始接触自动化监控与消息推送的开发者学习。1. 一个真实痛点为什么需要机台状态监控1.1 从“到店碰运气”到“自动化监控”很多玩家都有过类似的经历打开地图软件确认店铺还在营业骑车过去后却发现所有机台都在对战中工作日午休想去玩一局结果排在自己前面的还有五六个人机台偶尔进入维护状态工作人员也不确定恢复时间节假日期间机台资源变得更加紧张想预约比抢演唱会门票还难。“今年玩不到得明年再来”本质上反映的是信息不对称玩家无法实时获取机台当前处于什么状态。如果能有一个程序每隔一段时间自动获取机台状态并且在空闲或完成维护时立刻发送通知玩家就可以合理安排出行时间减少无意义的等待。1.2 需求拆解围绕“提前知道机台空闲”这一目标系统需要解决以下几个问题业务需求技术实现方式获取机台状态调用机台状态接口或解析公开页面数据周期性检查写一个常驻轮询循环间隔时间可配置检测状态变化保存上次状态与当前状态做 diff 比较发送提醒控制台日志、邮件、钉钉/企业微信机器人 Webhook运行与排错日志文件、状态文件、配置化参数1.3 这套方案适合谁如果你是 Python 初学者可以通过这个项目学会requests 发送 HTTP 请求、解析 JSON 响应状态快照的存储与对比思路多通道消息通知的封装定时轮询任务的基本写法简单的工程化组织方式如配置分离、日志记录、异常处理。如果你是有一定经验的开发人员则可以把它当作一个轻量级监控脚手架把机台状态替换成服务健康检查、设备在线率、商品库存等任何周期性变化的业务数据基本思路完全一致。2. 环境准备与项目结构2.1 版本说明本文示例使用以下环境操作系统Windows 10 / macOS / Linux 均可Python3.8 及以上第三方库requests、flask由于不同时期的三方库版本可能存在差异示例中不写死精确版本号而是使用宽松的范围约束flask3.0,4.0 requests2.31,3.0如果你的项目本身已经依赖某些框架请注意版本间的兼容性。本文重点演示配置和编码思路具体版本需要根据你的实际环境调整。2.2 创建虚拟环境建议先为项目创建一个独立虚拟环境避免污染系统级 Python 环境。mkdir maimai-monitor cd maimai-monitor python -m venv venv激活虚拟环境Windows 命令提示符venv\Scripts\activatemacOS / Linuxsource venv/bin/activate然后安装依赖pip install -r requirements.txt2.3 项目目录结构整个项目分为模拟状态源和监控程序两部分。建议按下面的结构组织文件maimai-monitor/ ├── mock_server.py # 模拟机台状态接口服务 ├── monitor.py # 监控主程序 ├── config.json # 配置文件 ├── requirements.txt # 依赖列表 ├── last_state.json # 程序运行时自动生成 └── monitor.log # 程序运行时自动生成日志last_state.json和monitor.log是程序运行后产生的文件。前者用于保存最近一次机台状态快照后者用于记录运行日志。3. 核心设计与数据模型3.1 机台状态定义为了让监控程序能够准确识别变化需要先明确机台状态的含义。本文以音游街机常见的三种状态为例状态值含义提醒场景idle空闲可以上机从其他状态变为空闲时提醒playing游戏中有人正在游玩记录变化不主动提醒maintenance维护中暂时无法使用从维护变为空闲时重点提醒实际业务中可能还有“已预约”“故障”等更多状态。为了便于扩展建议将状态值统一用英文字符串表示在展示层映射为中文。3.2 状态机与变化检测监控程序的核心是“状态变化检测”。简化后的判断逻辑如下第一次启动时没有历史状态此时将当前状态保存为快照不发送提醒避免启动即刷屏。后续每次轮询将本次获取到的状态与上次保存的状态做对比。如果同一机台的状态发生改变生成一条变化记录。所有变化记录统一通过通知渠道发送。按状态变化可以形成如下流程获取当前状态 - 读取历史快照 - 对比差异 - 生成变化记录 - 发送通知 - 保存新快照3.3 配置文件设计配置文件采用 JSON 格式把接口地址、轮询间隔、通知渠道和邮箱参数集中管理。这样做的好处是修改监控频率或通知方式时完全不需要改动代码。{ api_base_url: http://127.0.0.1:5000, poll_interval: 60, notify_channels: [console, webhook], email: { smtp_host: smtp.example.com, smtp_port: 465, username: your_accountexample.com, password: your_password_or_auth_code, receivers: [receiverexample.com] }, webhook: { url: 替换为你的机器人 Webhook 地址, keyword: 舞萌机台提醒 } }notify_channels支持多个渠道程序会按照列表顺序依次尝试推送。邮件配置中建议使用邮箱服务商提供的授权码而不是账号登录密码避免账号安全风险。3.4 通知渠道抽象不同的通知渠道发送方式不同但调用方只关心“把变化记录变成消息推送出去”。因此代码中为每个渠道单独实现一个函数统一在检测到状态变化后调用notify_console输出到控制台与日志文件notify_email通过 SMTP 发送邮件notify_webhook通过钉钉/企业微信机器人 Webhook 推送。这样后续新增短信、Server酱或即时通讯 App 推送时只需要增加一个函数并在check_and_notify中加一行调用即可。4. 完整实战搭建监控与提醒系统4.1 用 Flask 建一个模拟机台状态源在对接真实机台数据之前先用一个模拟接口把整个链路跑通。模拟服务的逻辑很简单每次有人请求状态接口时有 30% 的概率随机改变某台机器的状态。# 文件mock_server.py # 作用模拟机台状态服务方便本地验证监控程序 import random import time from flask import Flask, jsonify app Flask(__name__) CABINETS [ { id: cab_001, name: A01, status: idle, last_update: time.strftime(%Y-%m-%d %H:%M:%S), }, { id: cab_002, name: A02, status: playing, last_update: time.strftime(%Y-%m-%d %H:%M:%S), }, { id: cab_003, name: A03, status: maintenance, last_update: time.strftime(%Y-%m-%d %H:%M:%S), }, ] app.route(/api/cabinet/status, methods[GET]) def cabinet_status(): # 每调用一次有 30% 概率随机改变机台状态模拟真实环境 for cabinet in CABINETS: if random.random() 0.3: cabinet[status] random.choice([idle, playing, maintenance]) cabinet[last_update] time.strftime(%Y-%m-%d %H:%M:%S) return jsonify({ code: 0, message: success, data: CABINETS, timestamp: int(time.time()), }) if __name__ __main__: # 本地调试服务不需要在公网暴露 app.run(host127.0.0.1, port5000, debugFalse)这段代码的关键点有两个。第一状态数据保存在一个全局列表中方便在多次请求之间共享和修改第二随机变化逻辑保证了监控程序运行一段时间后能够观察到状态变化从而验证通知是否生效。4.2 编写监控主程序监控主程序承担所有核心逻辑包括读取配置、请求接口、对比状态、保存快照和触发通知。下面分段来看。首先是初始化部分# 文件monitor.py # 作用定时轮询机台状态检测变更后通过多个渠道发送提醒 import json import logging import smtplib import time from datetime import datetime from email.header import Header from email.mime.text import MIMEText from pathlib import Path import requests BASE_DIR Path(__file__).resolve().parent CONFIG_FILE BASE_DIR / config.json STATE_FILE BASE_DIR / last_state.json LOG_FILE BASE_DIR / monitor.log logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, handlers[ logging.StreamHandler(), logging.FileHandler(LOG_FILE, encodingutf-8), ], ) logger logging.getLogger(__name__)这里把日志同时输出到控制台和monitor.log文件方便长时间运行时追溯问题。然后读取配置、请求接口、状态映射、对比差异def load_config(): with open(CONFIG_FILE, r, encodingutf-8) as f: return json.load(f) def fetch_cabinet_status(config): api_url config[api_base_url].rstrip(/) /api/cabinet/status resp requests.get(api_url, timeout10) resp.raise_for_status() body resp.json() if body.get(code) ! 0: raise RuntimeError(f接口返回异常: {body}) return body[data] def format_status(status_text): mapping { idle: 空闲, playing: 游戏中, maintenance: 维护中, unknown: 未知, } return mapping.get(status_text, status_text) def diff_status(old_state, new_state): changes [] old_map {item[id]: item for item in old_state} if old_state else {} new_map {item[id]: item for item in new_state} for cabinet_id, current in new_map.items(): previous old_map.get(cabinet_id) if previous is None: changes.append( f新增机台 {current[name]}当前状态{format_status(current[status])} ) elif previous[status] ! current[status]: changes.append( f机台 {current[name]} 状态变化 f{format_status(previous[status])} - {format_status(current[status])} ) return changesdiff_status是变化检测的核心。它将旧状态列表转换成以id为键的字典新状态依次对比这样即使机台顺序发生变化也不会影响判断结果。4.3 实现多通道通知下面实现控制台、邮件、Webhook 三种通知方式。def notify_console(changes): for line in changes: logger.info([提醒] %s, line) def notify_email(config, changes): email_cfg config.get(email, {}) if not email_cfg: return smtp_host email_cfg[smtp_host] smtp_port email_cfg.get(smtp_port, 465) username email_cfg[username] password email_cfg[password] receivers email_cfg[receivers] subject f舞萌机台状态提醒 {datetime.now():%Y-%m-%d %H:%M:%S} content \n.join(changes) msg MIMEText(content, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] username msg[To] , .join(receivers) try: if smtp_port 465: server smtplib.SMTP_SSL(smtp_host, smtp_port, timeout15) else: server smtplib.SMTP(smtp_host, smtp_port, timeout15) server.starttls() server.login(username, password) server.sendmail(username, receivers, msg.as_string()) logger.info(邮件提醒发送成功) except Exception as e: logger.error(邮件发送失败: %s, e) finally: try: server.quit() except Exception: pass def notify_webhook(config, changes): webhook_cfg config.get(webhook, {}) if not webhook_cfg: return webhook_url webhook_cfg[url] keyword webhook_cfg.get(keyword, 机台提醒) content \n.join(changes) payload { msgtype: text, text: { content: f{keyword}\n{content}, }, } resp requests.post(webhook_url, jsonpayload, timeout10) logger.info(Webhook 推送结果%s, resp.text)邮件通知兼容了常规 SMTP587 端口和 SSL 加密通道465 端口。如果你的邮箱服务商要求starttls程序会在 587 端口模式下自动启用。4.4 状态持久化与主循环状态快照的保存与加载直接决定变化检测是否准确。为了避免程序异常退出造成快照文件损坏这里采用“临时文件 原子替换”的方式写入。def save_state(state): temp_file STATE_FILE.with_suffix(.tmp) with open(temp_file, w, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse, indent2) temp_file.replace(STATE_FILE) def load_state(): if not STATE_FILE.exists(): return None try: with open(STATE_FILE, r, encodingutf-8) as f: return json.load(f) except json.JSONDecodeError: logger.error(历史状态文件损坏将重新初始化) return None def check_and_notify(config): current fetch_cabinet_status(config) old load_state() changes diff_status(old, current) if changes: channels config.get(notify_channels, [console]) if console in channels: notify_console(changes) if email in channels: notify_email(config, changes) if webhook in channels: notify_webhook(config, changes) else: logger.info(机台状态未发生变化跳过通知) save_state(current) def main(): config load_config() logger.info(监控程序启动轮询间隔%s 秒, config.get(poll_interval, 60)) while True: try: check_and_notify(config) except requests.RequestException as e: logger.error(请求机台状态接口失败: %s, e) except Exception as e: logger.exception(监控流程出现未预期异常: %s, e) time.sleep(config.get(poll_interval, 60)) if __name__ __main__: main()主循环使用简单的while True sleep实现轮询。每次请求失败都会记录日志但不会导致程序退出。需要注意的是time.sleep只是在单机脚本场景下足够用如果后续要支持多任务并发调度建议换成APScheduler或Celery。4.5 添加依赖清单flask3.0,4.0 requests2.31,3.0将上述内容保存为requirements.txt然后执行安装命令pip install -r requirements.txt5. 运行验证与效果演示5.1 启动模拟状态源先打开一个终端窗口启动 Flask 模拟服务python mock_server.py正常情况下会看到 Flask 的启动日志监听地址为* Running on http://127.0.0.1:5000可以在浏览器中访问http://127.0.0.1:5000/api/cabinet/status确认接口返回 JSON 数据。5.2 启动监控程序另开一个终端窗口确认虚拟环境已激活然后运行python monitor.py程序启动后会读取config.json然后按照poll_interval设定的间隔轮询接口。由于模拟接口有 30% 的概率改变机台状态运行几分钟后大概率能在控制台看到类似输出2025-01-15 10:23:01 [INFO] 监控程序启动轮询间隔60 秒 2025-01-15 10:24:01 [INFO] 机台状态未发生变化跳过通知 2025-01-15 10:25:01 [INFO] [提醒] 机台 A01 状态变化空闲 - 游戏中 2025-01-15 10:25:01 [INFO] [提醒] 机台 A03 状态变化维护中 - 空闲 2025-01-15 10:25:01 [INFO] Webhook 推送结果{errcode:0,errmsg:ok}看到状态变化日志说明状态检测和通知链路已经正常运作。5.3 验证历史状态文件程序运行后会生成last_state.json内容大致如下[ { id: cab_001, name: A01, status: playing, last_update: 2025-01-15 10:25:01 }, { id: cab_002, name: A02, status: playing, last_update: 2025-01-15 10:23:01 }, { id: cab_003, name: A03, status: idle, last_update: 2025-01-15 10:25:01 } ]下次轮询时程序会以这份文件作为对比基准。6. 常见问题与排查思路6.1 高频问题速查表问题现象常见原因解决思路监控程序启动后马上退出config.json格式错误或字段缺失用python -m json.tool config.json检查 JSON 格式请求接口超时mock_server 未启动或端口被占用先访问http://127.0.0.1:5000/api/cabinet/status确认接口可用一直输出“未发生变化”模拟源随机概率较低或轮询间隔太短调大 mock_server 中的概率或手动修改状态数据邮件提醒发送失败SMTP 端口、授权码或发件地址配置错误使用邮箱授权码登录确认服务商 SMTP 地址正确Webhook 不生效推送数据结构不符合平台要求先用curl手动模拟一次推送核对文档格式重启监控后立刻收到大量提醒历史快照被删除或损坏首次运行不出提醒属于正常设计检查last_state.json是否存在6.2 如何排查接口返回异常当监控日志出现如下错误时请求机台状态接口失败: HTTPSConnectionPool(...) Max retries exceeded优先按顺序做三件事确认模拟服务是否在运行端口是否被占用确认config.json中的api_base_url是否拼写正确用浏览器或curl手动请求一次接口观察返回结构是否与代码预期一致。如果真实接口返回的字段名和本文示例不同只需要修改fetch_cabinet_status中的解析逻辑不需要改动其他模块。6.3 避免通知风暴模拟环境里状态变化频率不高但真实环境中可能出现一台机台状态频繁抖动的情况比如从“空闲”变为“游戏中”后又马上变回“空闲”。这会导致短时间内产生大量通知。一种简单有效的做法是增加“冷静时间”change_key f{cabinet_id}:{current[status]} last_notify_time notify_history.get(change_key, 0) if time.time() - last_notify_time config.get(notify_cooldown, 600): changes.append(...) notify_history[change_key] time.time()上面这段是思路示意实际接入时可以在diff_status之前维护一个notify_history字典同一个机台的相同状态在冷静时间内不重复提醒。7. 生产环境落地建议与工程实践7.1 对接真实机台状态接口模拟接口只是演示真实项目中需要找到合法、可公开访问的状态数据来源。对接时建议单独抽象一个datasource.py让监控主程序不直接感知请求细节# 文件datasource.py示例思路 # 作用适配不同数据源统一返回机台状态列表 def get_cabinet_status_from_web(): # 解析店铺页面中的机台状态返回统一结构 pass def get_cabinet_status_from_api(): # 请求官方或授权接口返回统一结构 pass对数据源做一层封装后后续切换数据来源只需要替换datasource内部实现不会影响变化检测和通知逻辑。7.2 轮询频率与请求压力轮询间隔并不是越短越好。如果频繁请求第三方接口可能会给对方服务造成压力也可能触发对方的风控策略。建议根据状态实时性要求合理设置人流量较大的商业区60 到 120 秒一次即可普通店铺5 分钟一次足够个人学习项目30 秒到 5 分钟之间自行调整。同时可以在每次请求中加入requests的timeout参数避免第三方接口长时间无响应时监控程序被一直阻塞。7.3 服务化与守护运行当前脚本使用CtrlC方式停止。如果希望长期在服务器上运行可以使用 systemd 管理。下面是一个简单的 systemd 服务配置示例[Unit] Descriptionmaimai cabinet monitor Afternetwork.target [Service] WorkingDirectory/opt/maimai-monitor ExecStart/opt/maimai-monitor/venv/bin/python /opt/maimai-monitor/monitor.py Restartalways RestartSec10 [Install] WantedBymulti-user.target将文件保存为/etc/systemd/system/maimai-monitor.service然后运行sudo systemctl daemon-reload sudo systemctl enable maimai-monitor sudo systemctl start maimai-monitor这样即使程序异常退出systemd 也会自动重新拉起。7.4 安全与合规提醒不管监控对象是机台、设备还是网页资源都需要注意合规边界只读取公开可见或你已经获得授权访问的数据不要尝试绕过登录、验证码或加密参数不要对目标接口发起高频压力请求生产环境密钥不要硬编码在代码中建议使用环境变量或密钥管理服务。本文示例中的config.json只是本地调试配置如果放到生产环境请务必把email.password、webhook.url等敏感字段替换为环境变量读取。7.5 日志与可观测性监控程序本身也需要被监控。建议至少记录以下内容日志级别记录内容INFO启动信息、轮询周期、正常状态变化WARNING单次请求失败、通知发送失败ERROR配置解析失败、接口结构异常CRITICAL连续多次轮询失败、状态文件无法写入如果希望主动收到程序异常告警可以在except中增加一个“监控自愈提醒”通道把异常信息发送到运维群。不过要注意避免异常反复触发导致刷屏一般可以做一个简单的失败次数计数达到阈值再发送告警。8. 总结与下一步优化方向这套机台状态监控与提醒系统看起来简单但已经覆盖了自动化监控项目的几个核心要素数据获取、状态快照、变化检测、多通道通知、异常处理和配置化运行。即使以后不再为“打舞萌”排队发愁这套代码架构也可以直接复用到其他场景比如库存监控、活动页面开售提醒、设备在线率巡检等。接下来可以继续尝试几个优化方向增加一个 Web 页面用表格展示当前机台状态和最近变化历史引入 SQLite 或 MySQL 保存长期状态数据方便统计机台空闲时段使用 APScheduler 替换while True sleep支持更复杂的定时策略将通知渠道扩展为短信、Server酱、企业微信应用消息等覆盖更多使用场景如果要同时监控多个店铺可以为每个店铺增加一个独立配置并统一汇聚到管理页。在“明年才能打舞萌了”的调侃背后不如把这次等待变成一个动手实践的机会。先把模拟服务跑起来再对照本文把监控程序部署到自己的电脑或服务器上之后无论是替换真实接口还是增加提醒渠道都是一件非常顺滑的事情。如果部署过程中遇到问题随时可以把报错信息贴在评论区大家一起排查。 Python+Flask实战：打造舞萌DX机台空闲监控与自动提醒系统
最近在音游玩家群里经常看到一句调侃明年才能打舞萌了。有人是因为常去的电玩城机台数量太少有人是周末排队时间动辄一两个小时也有人好不容易到店却发现机台全部处于维护状态。玩笑归玩笑这类问题背后的核心诉求其实很明确如何在机台空闲的第一时间就收到提醒避免反复到店扑空或长时间守在旁边排队。与其被动等待不如做一个面向机台状态监控与自动提醒的小系统。本文将从需求分析、系统设计、代码实现到运行验证完整演示一个 Python 监控工具的实现过程。你可以将它理解为“舞萌 DX 机台空闲提醒助手”稍加改造后也能用于其他街机、共享设备或展厅设备的在线状态监控。整个项目基于 Python 与 Flask 搭建不依赖复杂框架适合刚开始接触自动化监控与消息推送的开发者学习。1. 一个真实痛点为什么需要机台状态监控1.1 从“到店碰运气”到“自动化监控”很多玩家都有过类似的经历打开地图软件确认店铺还在营业骑车过去后却发现所有机台都在对战中工作日午休想去玩一局结果排在自己前面的还有五六个人机台偶尔进入维护状态工作人员也不确定恢复时间节假日期间机台资源变得更加紧张想预约比抢演唱会门票还难。“今年玩不到得明年再来”本质上反映的是信息不对称玩家无法实时获取机台当前处于什么状态。如果能有一个程序每隔一段时间自动获取机台状态并且在空闲或完成维护时立刻发送通知玩家就可以合理安排出行时间减少无意义的等待。1.2 需求拆解围绕“提前知道机台空闲”这一目标系统需要解决以下几个问题业务需求技术实现方式获取机台状态调用机台状态接口或解析公开页面数据周期性检查写一个常驻轮询循环间隔时间可配置检测状态变化保存上次状态与当前状态做 diff 比较发送提醒控制台日志、邮件、钉钉/企业微信机器人 Webhook运行与排错日志文件、状态文件、配置化参数1.3 这套方案适合谁如果你是 Python 初学者可以通过这个项目学会requests 发送 HTTP 请求、解析 JSON 响应状态快照的存储与对比思路多通道消息通知的封装定时轮询任务的基本写法简单的工程化组织方式如配置分离、日志记录、异常处理。如果你是有一定经验的开发人员则可以把它当作一个轻量级监控脚手架把机台状态替换成服务健康检查、设备在线率、商品库存等任何周期性变化的业务数据基本思路完全一致。2. 环境准备与项目结构2.1 版本说明本文示例使用以下环境操作系统Windows 10 / macOS / Linux 均可Python3.8 及以上第三方库requests、flask由于不同时期的三方库版本可能存在差异示例中不写死精确版本号而是使用宽松的范围约束flask3.0,4.0 requests2.31,3.0如果你的项目本身已经依赖某些框架请注意版本间的兼容性。本文重点演示配置和编码思路具体版本需要根据你的实际环境调整。2.2 创建虚拟环境建议先为项目创建一个独立虚拟环境避免污染系统级 Python 环境。mkdir maimai-monitor cd maimai-monitor python -m venv venv激活虚拟环境Windows 命令提示符venv\Scripts\activatemacOS / Linuxsource venv/bin/activate然后安装依赖pip install -r requirements.txt2.3 项目目录结构整个项目分为模拟状态源和监控程序两部分。建议按下面的结构组织文件maimai-monitor/ ├── mock_server.py # 模拟机台状态接口服务 ├── monitor.py # 监控主程序 ├── config.json # 配置文件 ├── requirements.txt # 依赖列表 ├── last_state.json # 程序运行时自动生成 └── monitor.log # 程序运行时自动生成日志last_state.json和monitor.log是程序运行后产生的文件。前者用于保存最近一次机台状态快照后者用于记录运行日志。3. 核心设计与数据模型3.1 机台状态定义为了让监控程序能够准确识别变化需要先明确机台状态的含义。本文以音游街机常见的三种状态为例状态值含义提醒场景idle空闲可以上机从其他状态变为空闲时提醒playing游戏中有人正在游玩记录变化不主动提醒maintenance维护中暂时无法使用从维护变为空闲时重点提醒实际业务中可能还有“已预约”“故障”等更多状态。为了便于扩展建议将状态值统一用英文字符串表示在展示层映射为中文。3.2 状态机与变化检测监控程序的核心是“状态变化检测”。简化后的判断逻辑如下第一次启动时没有历史状态此时将当前状态保存为快照不发送提醒避免启动即刷屏。后续每次轮询将本次获取到的状态与上次保存的状态做对比。如果同一机台的状态发生改变生成一条变化记录。所有变化记录统一通过通知渠道发送。按状态变化可以形成如下流程获取当前状态 - 读取历史快照 - 对比差异 - 生成变化记录 - 发送通知 - 保存新快照3.3 配置文件设计配置文件采用 JSON 格式把接口地址、轮询间隔、通知渠道和邮箱参数集中管理。这样做的好处是修改监控频率或通知方式时完全不需要改动代码。{ api_base_url: http://127.0.0.1:5000, poll_interval: 60, notify_channels: [console, webhook], email: { smtp_host: smtp.example.com, smtp_port: 465, username: your_accountexample.com, password: your_password_or_auth_code, receivers: [receiverexample.com] }, webhook: { url: 替换为你的机器人 Webhook 地址, keyword: 舞萌机台提醒 } }notify_channels支持多个渠道程序会按照列表顺序依次尝试推送。邮件配置中建议使用邮箱服务商提供的授权码而不是账号登录密码避免账号安全风险。3.4 通知渠道抽象不同的通知渠道发送方式不同但调用方只关心“把变化记录变成消息推送出去”。因此代码中为每个渠道单独实现一个函数统一在检测到状态变化后调用notify_console输出到控制台与日志文件notify_email通过 SMTP 发送邮件notify_webhook通过钉钉/企业微信机器人 Webhook 推送。这样后续新增短信、Server酱或即时通讯 App 推送时只需要增加一个函数并在check_and_notify中加一行调用即可。4. 完整实战搭建监控与提醒系统4.1 用 Flask 建一个模拟机台状态源在对接真实机台数据之前先用一个模拟接口把整个链路跑通。模拟服务的逻辑很简单每次有人请求状态接口时有 30% 的概率随机改变某台机器的状态。# 文件mock_server.py # 作用模拟机台状态服务方便本地验证监控程序 import random import time from flask import Flask, jsonify app Flask(__name__) CABINETS [ { id: cab_001, name: A01, status: idle, last_update: time.strftime(%Y-%m-%d %H:%M:%S), }, { id: cab_002, name: A02, status: playing, last_update: time.strftime(%Y-%m-%d %H:%M:%S), }, { id: cab_003, name: A03, status: maintenance, last_update: time.strftime(%Y-%m-%d %H:%M:%S), }, ] app.route(/api/cabinet/status, methods[GET]) def cabinet_status(): # 每调用一次有 30% 概率随机改变机台状态模拟真实环境 for cabinet in CABINETS: if random.random() 0.3: cabinet[status] random.choice([idle, playing, maintenance]) cabinet[last_update] time.strftime(%Y-%m-%d %H:%M:%S) return jsonify({ code: 0, message: success, data: CABINETS, timestamp: int(time.time()), }) if __name__ __main__: # 本地调试服务不需要在公网暴露 app.run(host127.0.0.1, port5000, debugFalse)这段代码的关键点有两个。第一状态数据保存在一个全局列表中方便在多次请求之间共享和修改第二随机变化逻辑保证了监控程序运行一段时间后能够观察到状态变化从而验证通知是否生效。4.2 编写监控主程序监控主程序承担所有核心逻辑包括读取配置、请求接口、对比状态、保存快照和触发通知。下面分段来看。首先是初始化部分# 文件monitor.py # 作用定时轮询机台状态检测变更后通过多个渠道发送提醒 import json import logging import smtplib import time from datetime import datetime from email.header import Header from email.mime.text import MIMEText from pathlib import Path import requests BASE_DIR Path(__file__).resolve().parent CONFIG_FILE BASE_DIR / config.json STATE_FILE BASE_DIR / last_state.json LOG_FILE BASE_DIR / monitor.log logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, handlers[ logging.StreamHandler(), logging.FileHandler(LOG_FILE, encodingutf-8), ], ) logger logging.getLogger(__name__)这里把日志同时输出到控制台和monitor.log文件方便长时间运行时追溯问题。然后读取配置、请求接口、状态映射、对比差异def load_config(): with open(CONFIG_FILE, r, encodingutf-8) as f: return json.load(f) def fetch_cabinet_status(config): api_url config[api_base_url].rstrip(/) /api/cabinet/status resp requests.get(api_url, timeout10) resp.raise_for_status() body resp.json() if body.get(code) ! 0: raise RuntimeError(f接口返回异常: {body}) return body[data] def format_status(status_text): mapping { idle: 空闲, playing: 游戏中, maintenance: 维护中, unknown: 未知, } return mapping.get(status_text, status_text) def diff_status(old_state, new_state): changes [] old_map {item[id]: item for item in old_state} if old_state else {} new_map {item[id]: item for item in new_state} for cabinet_id, current in new_map.items(): previous old_map.get(cabinet_id) if previous is None: changes.append( f新增机台 {current[name]}当前状态{format_status(current[status])} ) elif previous[status] ! current[status]: changes.append( f机台 {current[name]} 状态变化 f{format_status(previous[status])} - {format_status(current[status])} ) return changesdiff_status是变化检测的核心。它将旧状态列表转换成以id为键的字典新状态依次对比这样即使机台顺序发生变化也不会影响判断结果。4.3 实现多通道通知下面实现控制台、邮件、Webhook 三种通知方式。def notify_console(changes): for line in changes: logger.info([提醒] %s, line) def notify_email(config, changes): email_cfg config.get(email, {}) if not email_cfg: return smtp_host email_cfg[smtp_host] smtp_port email_cfg.get(smtp_port, 465) username email_cfg[username] password email_cfg[password] receivers email_cfg[receivers] subject f舞萌机台状态提醒 {datetime.now():%Y-%m-%d %H:%M:%S} content \n.join(changes) msg MIMEText(content, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] username msg[To] , .join(receivers) try: if smtp_port 465: server smtplib.SMTP_SSL(smtp_host, smtp_port, timeout15) else: server smtplib.SMTP(smtp_host, smtp_port, timeout15) server.starttls() server.login(username, password) server.sendmail(username, receivers, msg.as_string()) logger.info(邮件提醒发送成功) except Exception as e: logger.error(邮件发送失败: %s, e) finally: try: server.quit() except Exception: pass def notify_webhook(config, changes): webhook_cfg config.get(webhook, {}) if not webhook_cfg: return webhook_url webhook_cfg[url] keyword webhook_cfg.get(keyword, 机台提醒) content \n.join(changes) payload { msgtype: text, text: { content: f{keyword}\n{content}, }, } resp requests.post(webhook_url, jsonpayload, timeout10) logger.info(Webhook 推送结果%s, resp.text)邮件通知兼容了常规 SMTP587 端口和 SSL 加密通道465 端口。如果你的邮箱服务商要求starttls程序会在 587 端口模式下自动启用。4.4 状态持久化与主循环状态快照的保存与加载直接决定变化检测是否准确。为了避免程序异常退出造成快照文件损坏这里采用“临时文件 原子替换”的方式写入。def save_state(state): temp_file STATE_FILE.with_suffix(.tmp) with open(temp_file, w, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse, indent2) temp_file.replace(STATE_FILE) def load_state(): if not STATE_FILE.exists(): return None try: with open(STATE_FILE, r, encodingutf-8) as f: return json.load(f) except json.JSONDecodeError: logger.error(历史状态文件损坏将重新初始化) return None def check_and_notify(config): current fetch_cabinet_status(config) old load_state() changes diff_status(old, current) if changes: channels config.get(notify_channels, [console]) if console in channels: notify_console(changes) if email in channels: notify_email(config, changes) if webhook in channels: notify_webhook(config, changes) else: logger.info(机台状态未发生变化跳过通知) save_state(current) def main(): config load_config() logger.info(监控程序启动轮询间隔%s 秒, config.get(poll_interval, 60)) while True: try: check_and_notify(config) except requests.RequestException as e: logger.error(请求机台状态接口失败: %s, e) except Exception as e: logger.exception(监控流程出现未预期异常: %s, e) time.sleep(config.get(poll_interval, 60)) if __name__ __main__: main()主循环使用简单的while True sleep实现轮询。每次请求失败都会记录日志但不会导致程序退出。需要注意的是time.sleep只是在单机脚本场景下足够用如果后续要支持多任务并发调度建议换成APScheduler或Celery。4.5 添加依赖清单flask3.0,4.0 requests2.31,3.0将上述内容保存为requirements.txt然后执行安装命令pip install -r requirements.txt5. 运行验证与效果演示5.1 启动模拟状态源先打开一个终端窗口启动 Flask 模拟服务python mock_server.py正常情况下会看到 Flask 的启动日志监听地址为* Running on http://127.0.0.1:5000可以在浏览器中访问http://127.0.0.1:5000/api/cabinet/status确认接口返回 JSON 数据。5.2 启动监控程序另开一个终端窗口确认虚拟环境已激活然后运行python monitor.py程序启动后会读取config.json然后按照poll_interval设定的间隔轮询接口。由于模拟接口有 30% 的概率改变机台状态运行几分钟后大概率能在控制台看到类似输出2025-01-15 10:23:01 [INFO] 监控程序启动轮询间隔60 秒 2025-01-15 10:24:01 [INFO] 机台状态未发生变化跳过通知 2025-01-15 10:25:01 [INFO] [提醒] 机台 A01 状态变化空闲 - 游戏中 2025-01-15 10:25:01 [INFO] [提醒] 机台 A03 状态变化维护中 - 空闲 2025-01-15 10:25:01 [INFO] Webhook 推送结果{errcode:0,errmsg:ok}看到状态变化日志说明状态检测和通知链路已经正常运作。5.3 验证历史状态文件程序运行后会生成last_state.json内容大致如下[ { id: cab_001, name: A01, status: playing, last_update: 2025-01-15 10:25:01 }, { id: cab_002, name: A02, status: playing, last_update: 2025-01-15 10:23:01 }, { id: cab_003, name: A03, status: idle, last_update: 2025-01-15 10:25:01 } ]下次轮询时程序会以这份文件作为对比基准。6. 常见问题与排查思路6.1 高频问题速查表问题现象常见原因解决思路监控程序启动后马上退出config.json格式错误或字段缺失用python -m json.tool config.json检查 JSON 格式请求接口超时mock_server 未启动或端口被占用先访问http://127.0.0.1:5000/api/cabinet/status确认接口可用一直输出“未发生变化”模拟源随机概率较低或轮询间隔太短调大 mock_server 中的概率或手动修改状态数据邮件提醒发送失败SMTP 端口、授权码或发件地址配置错误使用邮箱授权码登录确认服务商 SMTP 地址正确Webhook 不生效推送数据结构不符合平台要求先用curl手动模拟一次推送核对文档格式重启监控后立刻收到大量提醒历史快照被删除或损坏首次运行不出提醒属于正常设计检查last_state.json是否存在6.2 如何排查接口返回异常当监控日志出现如下错误时请求机台状态接口失败: HTTPSConnectionPool(...) Max retries exceeded优先按顺序做三件事确认模拟服务是否在运行端口是否被占用确认config.json中的api_base_url是否拼写正确用浏览器或curl手动请求一次接口观察返回结构是否与代码预期一致。如果真实接口返回的字段名和本文示例不同只需要修改fetch_cabinet_status中的解析逻辑不需要改动其他模块。6.3 避免通知风暴模拟环境里状态变化频率不高但真实环境中可能出现一台机台状态频繁抖动的情况比如从“空闲”变为“游戏中”后又马上变回“空闲”。这会导致短时间内产生大量通知。一种简单有效的做法是增加“冷静时间”change_key f{cabinet_id}:{current[status]} last_notify_time notify_history.get(change_key, 0) if time.time() - last_notify_time config.get(notify_cooldown, 600): changes.append(...) notify_history[change_key] time.time()上面这段是思路示意实际接入时可以在diff_status之前维护一个notify_history字典同一个机台的相同状态在冷静时间内不重复提醒。7. 生产环境落地建议与工程实践7.1 对接真实机台状态接口模拟接口只是演示真实项目中需要找到合法、可公开访问的状态数据来源。对接时建议单独抽象一个datasource.py让监控主程序不直接感知请求细节# 文件datasource.py示例思路 # 作用适配不同数据源统一返回机台状态列表 def get_cabinet_status_from_web(): # 解析店铺页面中的机台状态返回统一结构 pass def get_cabinet_status_from_api(): # 请求官方或授权接口返回统一结构 pass对数据源做一层封装后后续切换数据来源只需要替换datasource内部实现不会影响变化检测和通知逻辑。7.2 轮询频率与请求压力轮询间隔并不是越短越好。如果频繁请求第三方接口可能会给对方服务造成压力也可能触发对方的风控策略。建议根据状态实时性要求合理设置人流量较大的商业区60 到 120 秒一次即可普通店铺5 分钟一次足够个人学习项目30 秒到 5 分钟之间自行调整。同时可以在每次请求中加入requests的timeout参数避免第三方接口长时间无响应时监控程序被一直阻塞。7.3 服务化与守护运行当前脚本使用CtrlC方式停止。如果希望长期在服务器上运行可以使用 systemd 管理。下面是一个简单的 systemd 服务配置示例[Unit] Descriptionmaimai cabinet monitor Afternetwork.target [Service] WorkingDirectory/opt/maimai-monitor ExecStart/opt/maimai-monitor/venv/bin/python /opt/maimai-monitor/monitor.py Restartalways RestartSec10 [Install] WantedBymulti-user.target将文件保存为/etc/systemd/system/maimai-monitor.service然后运行sudo systemctl daemon-reload sudo systemctl enable maimai-monitor sudo systemctl start maimai-monitor这样即使程序异常退出systemd 也会自动重新拉起。7.4 安全与合规提醒不管监控对象是机台、设备还是网页资源都需要注意合规边界只读取公开可见或你已经获得授权访问的数据不要尝试绕过登录、验证码或加密参数不要对目标接口发起高频压力请求生产环境密钥不要硬编码在代码中建议使用环境变量或密钥管理服务。本文示例中的config.json只是本地调试配置如果放到生产环境请务必把email.password、webhook.url等敏感字段替换为环境变量读取。7.5 日志与可观测性监控程序本身也需要被监控。建议至少记录以下内容日志级别记录内容INFO启动信息、轮询周期、正常状态变化WARNING单次请求失败、通知发送失败ERROR配置解析失败、接口结构异常CRITICAL连续多次轮询失败、状态文件无法写入如果希望主动收到程序异常告警可以在except中增加一个“监控自愈提醒”通道把异常信息发送到运维群。不过要注意避免异常反复触发导致刷屏一般可以做一个简单的失败次数计数达到阈值再发送告警。8. 总结与下一步优化方向这套机台状态监控与提醒系统看起来简单但已经覆盖了自动化监控项目的几个核心要素数据获取、状态快照、变化检测、多通道通知、异常处理和配置化运行。即使以后不再为“打舞萌”排队发愁这套代码架构也可以直接复用到其他场景比如库存监控、活动页面开售提醒、设备在线率巡检等。接下来可以继续尝试几个优化方向增加一个 Web 页面用表格展示当前机台状态和最近变化历史引入 SQLite 或 MySQL 保存长期状态数据方便统计机台空闲时段使用 APScheduler 替换while True sleep支持更复杂的定时策略将通知渠道扩展为短信、Server酱、企业微信应用消息等覆盖更多使用场景如果要同时监控多个店铺可以为每个店铺增加一个独立配置并统一汇聚到管理页。在“明年才能打舞萌了”的调侃背后不如把这次等待变成一个动手实践的机会。先把模拟服务跑起来再对照本文把监控程序部署到自己的电脑或服务器上之后无论是替换真实接口还是增加提醒渠道都是一件非常顺滑的事情。如果部署过程中遇到问题随时可以把报错信息贴在评论区大家一起排查。