"""通知事件目录（**唯一真相**）。 约定： - **key 点分层**：`模块.对象.动作`（如 `task.device.failed`），支持通配订阅 （`task.*`、`device.*`、`*`），匹配规则见 `match()`。 - **登记 ≠ 会发**：所有事件 `default` 一律 False —— 用户在「系统 → 通知」里 按 webhook 勾选才会推送。`recommend` 只用于界面高亮"建议开启"，不改默认值。 - **聚合**：`agg_window` 秒内的同 `(hook, event, agg_key 取值)` 事件合并成一条 （保留前几个样本）；`agg_window=0` = 不聚合，立即发（低频高危事件用它）。 - **fields**：该事件保证携带的上下文字段，供前端画字段表、也是模板占位符白名单。 新增事件：在 `EVENTS` 里加一条 + 在触发点调 `notifier.notify(key, **fields)`， 并同步 `doc/NOTIFY.md` 的事件表（见 doc/README.md 的文档同步红线）。 """ from collections import namedtuple # key, label, category, default, recommend, agg_window, agg_key, fields, desc EventDef = namedtuple( "EventDef", "key label category default recommend agg_window agg_key fields desc") def _e(key, label, category, fields, desc="", agg_window=30, agg_key=None, recommend=False): """构造一条事件定义（default 恒为 False：登记不等于推送）。""" return EventDef(key, label, category, False, recommend, agg_window, agg_key, tuple(fields), desc) # 聚合键：按"任务批次"合并（同一次任务的多台设备结果合成一条） _BY_JOB = ("job_id",) # 按"设备"合并（一台设备的重复抖动合成一条） _BY_SERIAL = ("serial",) EVENTS = [ # ---------------- 任务批次 ---------------- _e("task.batch.started", "任务批次开始", "任务批次", ["job_id", "job_name", "task_type", "device_count", "serials_preview"], "一次任务开始铺开到 N 台设备", agg_window=0, recommend=True), _e("task.batch.finished", "任务批次结束", "任务批次", ["job_id", "job_name", "total", "success", "failed", "stopped", "skipped", "failed_devices", "duration_s"], "整批跑完（成功/失败/停止台数 + 耗时；有失败时列出失败设备与型号）", agg_window=0, recommend=True), _e("task.batch.no_device", "任务无可用设备", "任务批次", ["job_id", "job_name", "target"], "触发时一台可用设备都没有（任务空跑）", agg_window=0, recommend=True), _e("task.batch.unknown_type", "任务类型不存在", "任务批次", ["job_id", "job_name", "task_type"], "任务类型已从代码里删掉，任务永远不会执行", agg_window=0, recommend=True), _e("task.cron.stopped", "定时停止任务", "任务批次", ["job_id", "job_name", "stopped_count", "serials"], "cron_stop 到点，停掉了正在跑的设备", agg_window=0), # ---------------- 任务 · 单设备（权威结论点） ---------------- _e("task.device.success", "设备任务成功", "任务·单设备", ["serial", "device_name", "model", "job_id", "job_name", "attempt", "duration_s"], "某台设备上的任务最终成功。**只在目标设备只有 1 台时发送**——多设备批次看" "「任务批次结束」就够了，13 台设备就是 13 条刷屏", agg_key=_BY_JOB), _e("task.device.failed", "设备任务失败", "任务·单设备", ["serial", "device_name", "model", "job_id", "job_name", "attempts", "cause", "msg"], "某台设备上的任务最终失败（带真实原因与型号）", agg_key=_BY_JOB, recommend=True), _e("task.device.offline", "设备离线放弃", "任务·单设备", ["serial", "device_name", "model", "job_id", "job_name", "error"], "设备离线，不重试直接失败", agg_key=_BY_JOB, recommend=True), _e("task.device.error", "单次尝试异常", "任务·单设备", ["serial", "device_name", "model", "job_name", "attempt", "error"], "某次尝试抛异常（后面还会重试，噪音较大）", agg_key=_BY_JOB), _e("task.device.retry", "设备任务重试", "任务·单设备", ["serial", "device_name", "job_name", "attempt", "next_attempt", "delay_s", "reason"], "即将重试", agg_key=_BY_JOB), _e("task.device.stopped", "设备任务被停止", "任务·单设备", ["serial", "device_name", "model", "job_name", "attempt", "phase"], "用户手动停止 / cron 停止", agg_key=_BY_JOB), _e("task.device.preempted", "设备被抢占", "任务·单设备", ["serial", "device_name", "job_name", "preempted_job_id", "preempted_job_name"], "本任务抢占了该设备上正在跑的其他任务", agg_key=_BY_JOB, recommend=True), _e("task.device.preempt_timeout", "抢占超时跳过", "任务·单设备", ["serial", "device_name", "job_name", "preempted_job_id"], "等被抢占任务退出超时，本设备放弃执行", agg_window=0, recommend=True), _e("task.device.released", "抢占结束归还", "任务·单设备", ["serial", "device_name", "job_name", "preempted_job_id", "returned", "reason"], "抢占结束时重新拉起被抢占的任务", agg_key=_BY_JOB), # ---------------- Worker 状态机（单次 attempt 级，默认关） ---------------- _e("worker.connected", "设备已连接", "Worker", ["serial", "model", "remote_adb_url"], "worker 连上设备（单次尝试级，噪音大）"), _e("worker.attempt.done", "单次执行完成", "Worker", ["serial", "task_type", "duration_s", "max_duration_hit"], "单次 attempt 正常跑完（不代表任务最终成功）"), _e("worker.attempt.error", "单次执行出错", "Worker", ["serial", "task_type", "error", "transient"], "单次 attempt 抛异常"), # ---------------- 业务（任务内容） ---------------- _e("task.patrol.hit", "任务巡检命中", "业务", ["title", "message", "patrol_name", "check_label", "detail", "action_label", "job_id", "job_name", "serial", "device_name", "model"], "任务里配置的「公共巡检」命中（熄屏/元素/前台…），已执行动作——" "标题正文可在巡检项里自己写，支持 {device}/{job}/{time}/{app}/{screen}", agg_window=0, recommend=True), _e("task.notify.custom", "任务自定义通知", "业务", ["title", "message", "job_id", "job_name", "serial", "device_name", "model"], "任务里的「发通知」步骤被触发——标题与正文由任务自己写，支持 " "{device}/{serial}/{job}/{time}/{app}/{screen} 占位符", agg_window=0, recommend=True), _e("task.selector.invalid", "选择器连续失效", "业务", ["serial", "device_name", "selector", "miss_count"], "某选择器连续 10 次未命中——任务可能显示成功但什么都没做", agg_window=0, recommend=True), # ---------------- 设备 ---------------- _e("device.online", "设备恢复在线", "设备", ["serials", "count"], "原本断联的设备又能连上了", agg_window=0, recommend=True), _e("device.offline", "设备断联", "设备", ["serials", "count", "devices"], "设备池里的设备连不上了", agg_window=0, recommend=True), _e("device.discovered", "发现新设备", "设备", ["serials", "count"], "扫描到尚未在池内的设备（待认领）", agg_window=0), _e("device.claimed", "设备自动认领", "设备", ["pairs", "count"], "指纹匹配成功，自动把旧记录迁到新地址", agg_window=0), _e("device.heartbeat_timeout", "心跳超时", "设备", ["serial", "device_name", "timeout_s", "task_job", "model"], "设备卡死（长时间没心跳），任务可能已中断", agg_window=0, recommend=True), # ---------------- 应用安装 ---------------- _e("apk.install.started", "应用安装开始", "安装", ["apk_id", "apk_name", "package_name", "total"], "开始往 N 台设备推装", agg_window=0), _e("apk.install.finished", "应用安装完成", "安装", ["apk_name", "success", "failed", "skipped", "total", "failed_items"], "批量安装结束（含失败台数与原因）", agg_window=0, recommend=True), # ---------------- 系统 ---------------- _e("system.backup.exported", "备份已导出", "系统", ["filename", "size", "tables", "include_apk", "user"], "有人导出了整库备份", agg_window=0, recommend=True), _e("system.backup.imported", "备份导入已挂起", "系统", ["token_prefix", "env_label", "force"], "上传了备份并确认导入（重启后生效）", agg_window=0), _e("system.backup.restored", "备份已恢复", "系统", ["applied_rows", "schema_version"], "启动时应用了待恢复的备份（所以这条是重启后才发）", agg_window=0, recommend=True), _e("system.backup.restore_failed", "备份恢复失败", "系统", ["error", "fail_dir"], "恢复校验没过，已搁置（数据未被改动）", agg_window=0, recommend=True), _e("service.started", "服务已启动", "系统", ["version", "env", "db_target", "device_count", "job_count"], "平台进程起来了", agg_window=0, recommend=True), _e("service.stopping", "服务正在停止", "系统", ["uptime_s"], "平台进程收到退出信号", agg_window=0), _e("user.login", "用户登录", "系统", ["username"], "有人登录了后台", agg_window=0), # ---------------- AI ---------------- _e("ai.audit.finished", "经验巡检完成", "AI", ["reviewed", "suggested", "summary"], "每日经验库巡检跑完（或手动触发）", agg_window=0, recommend=True), _e("ai.audit.failed", "经验巡检异常", "AI", ["error"], "巡检过程中报错", agg_window=0), _e("ai.audit.skipped", "经验巡检跳过", "AI", ["reason"], "没配模型 Key 等原因跳过巡检", agg_window=0), # ---------------- 其它 ---------------- _e("notify.test", "测试通知", "其它", ["hook_name", "operator"], "「发送测试」按钮专用", agg_window=0), ] BY_KEY = {e.key: e for e in EVENTS} # 界面上的分组顺序 CATEGORIES = ("任务批次", "任务·单设备", "业务", "设备", "安装", "系统", "AI", "Worker", "其它") def get(key): """按 key 取事件定义；未知 key 返回 None（未知事件会被 notify 丢弃）。""" return BY_KEY.get(key) def match(pattern, key): """事件订阅匹配。 - `*` → 全部 - `task.*` → task 下所有（含 task.device.failed） - `task.device.*` → 任意层级前缀匹配 - `task.device.failed` → 精确 """ if not pattern: return False if pattern == "*" or pattern == key: return True if pattern.endswith(".*"): return key.startswith(pattern[:-1]) # "task.*" → "task." return False def list_events(): """给接口/前端用的事件目录（按 CATEGORIES 排序）。""" order = {c: i for i, c in enumerate(CATEGORIES)} rows = [{"key": e.key, "label": e.label, "category": e.category, "default": e.default, "recommend": e.recommend, "agg_window": e.agg_window, "agg_key": list(e.agg_key or ()), "fields": list(e.fields), "desc": e.desc} for e in EVENTS] rows.sort(key=lambda r: (order.get(r["category"], 99), r["key"])) return rows