feat(通知): 系统级 Webhook 通知子系统（事件目录 + 可插拔适配器 + 防刷屏）
平台此前出了问题只能靠人盯页面。现在各组件统一走 `notifier.notify(事件, **字段)`， 推到企业微信 / 自建服务；**所有可通知点都登记进事件目录，默认全关，用户按 webhook 勾选**。 ## 架构（`core/notifier.py` + `core/notify_events.py`） 业务线程调 `notify()` → 只做内存操作（读配置快照/匹配订阅/入队）→ 返回； 后台 1 个 dispatcher（聚合 + 每 hook 限流 + 折叠摘要）+ 3 个 sender（真实 HTTP、退避重试） 负责真正发出去。硬约束：**notify 零 DB、零 HTTP、零阻塞、异常不冒泡**——所以任务线程里 可以直接调（不用 app_context、不用 try/except），但**必须放在所有 `with self._lock` 之外**。 - **事件目录 36 条**（任务批次/单设备/设备/Worker/业务/安装/系统/AI），支持 `task.*` 通配订阅； 语义分工避免重复告警：`worker.*` 是单次尝试级，`task.device.success/failed` 是唯一权威结论。 - **适配器可插拔**：`wecom`（markdown，按 4096 **字节**截断、超限不截半个汉字）+ `json`（模板占位符，替换值按 JSON 转义，保存前干跑校验）；钉钉/飞书留了插槽（前端置灰）。 - **防打爆四层**：聚合窗口（默认 30s，同批次合并成一条并带样本）→ 令牌桶限流（默认 18/分， 对齐企微硬限 20）→ 被限流的**折叠成摘要不丢弃** → 有界队列背压。取舍：失败通知最多延迟 一个窗口，换来群不被刷屏。 ## 安全与存储 - 配置只落 `app_meta.notify_webhooks` 一个键（**不建表** → 不涉及备份覆盖红线）。 - URL 本身就是凭据（企微 `?key=`）→ 接口回显/发送记录/日志一律 `mask_url()/scrub()`； 编辑时留空即不修改；secret 永不回显。DATA_MODEL 的明文凭据告警补上了这一条。 - 发送记录：内存环形缓冲 200 条（重启清空）+ 独立 `logs/notify.log`。 ## 接入点（每个都放在锁外、不改 return 顺序） task_manager（批次开始/结束用新增的 `_BatchTracker` 统一在 finally 计数、单设备成功/失败/ 离线/重试/停止/抢占/归还/cron 停止）、device_worker 心跳看门狗、generic 任务选择器连续失效、 apk 安装开始/完成、设备上下线（**状态沿检测**，只报新变化）、备份导出/恢复、经验巡检、 用户登录、服务启停。 ## 前端 系统 Tab 新增「通知 / Webhook」子分栏：多条 webhook 列表（URL 打码）+ 编辑弹窗（格式/URL/ 密钥/事件勾选树带 ★建议/聚合/限流/自定义模板/预览）+ 发送测试 + 发送记录。 ## 自测 - 进程内逻辑 10 组断言全绿：聚合合并、限流+折叠、无配置/全局关静默丢弃、未知事件、 内部异常不外泄、JSON 转义（标题含引号换行仍合法）、URL/异常消息脱敏、配置校验。 - 端到端（假 webhook 接收端）17 项断言全绿：真实事件投递（user.login / task.batch.no_device）、 企微请求体形状、**HTTP 200 + errcode 93000 判为失败**、500 重试 3 次、记录里 URL 打码。 - 韧性：webhook 指向黑洞地址时登录耗时 100~114ms（基线 107~133ms，**异步隔离生效**）； 配置写成坏 JSON 服务照常启动、通知静默不发、日志有 error（服务端实测后已复原）。 - 页面：系统 → 通知 面板/弹窗/36 个事件复选框/预览全部正常，无 JS 报错。 - 自测数据已清理（webhook、自建任务、写坏又复原的配置键）。 文档：新增 doc/NOTIFY.md（事件表/配置/格式约束/防刷屏/加事件三步骤/排障）并登记进 doc/README； API.md §2.11；DATA_MODEL 的 app_meta 键表与明文凭据告警；ARCHITECTURE 线程表/分层/扩展点； 根 README 功能索引与日志表。
This commit is contained in:
+142
-12
@@ -21,6 +21,7 @@ import json
|
||||
import time
|
||||
import uuid
|
||||
import threading
|
||||
from collections import Counter
|
||||
from datetime import datetime
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
@@ -31,6 +32,7 @@ from config import DATA_DIR
|
||||
from core.logger import get_logger
|
||||
from core.models import db, DeviceGroup as GroupRow, TaskJob as JobRow
|
||||
from core import device_pool
|
||||
from core import notifier
|
||||
from .adb_helper import get_foreground_app, get_foreground_app_remote
|
||||
from .device_worker import (
|
||||
DeviceOfflineError,
|
||||
@@ -47,6 +49,42 @@ _log = get_logger("core.tm")
|
||||
_START_STAGGER_SEC = 0.2
|
||||
|
||||
|
||||
class _BatchTracker:
|
||||
"""一次任务批次的收尾统计：**所有设备都出结果**后发一条 `task.batch.finished`。
|
||||
|
||||
为什么不每台设备各发一条"批次结束"：100 台设备就是 100 条刷屏。这里只计数，
|
||||
归零那一刻发一条汇总（锁外发，见 done()）。
|
||||
|
||||
每台设备的结果由 `_run_with_retry` 的 finally 调一次 `done(serial, outcome)`——
|
||||
放在 finally 里是为了保证"任何一个 return 分支都会被计数一次"，不会漏也不会重。
|
||||
"""
|
||||
|
||||
def __init__(self, job, total, names=None):
|
||||
self.job = job
|
||||
self.total = total
|
||||
# serial -> 设备名（通知里显示名字而不是裸地址）；取不到就是空
|
||||
self.names = dict(names or {})
|
||||
self._lock = threading.Lock()
|
||||
self._left = total
|
||||
self._stats = Counter()
|
||||
self._t0 = time.time()
|
||||
|
||||
def done(self, serial, outcome="failed"):
|
||||
"""登记一台设备的结果（success/failed/stopped/skipped）。"""
|
||||
with self._lock:
|
||||
self._left -= 1
|
||||
self._stats[outcome] += 1
|
||||
left = self._left
|
||||
s = dict(self._stats)
|
||||
if left > 0:
|
||||
return
|
||||
notifier.notify("task.batch.finished", job_id=self.job.id,
|
||||
job_name=self.job.name, total=self.total,
|
||||
success=s.get("success", 0), failed=s.get("failed", 0),
|
||||
stopped=s.get("stopped", 0), skipped=s.get("skipped", 0),
|
||||
duration_s=int(time.time() - self._t0))
|
||||
|
||||
|
||||
def _in_run_window(schedule, now=None):
|
||||
"""是否在当前运行窗口内。
|
||||
|
||||
@@ -633,6 +671,8 @@ class TaskManager:
|
||||
w.stop()
|
||||
stopped.append(serial)
|
||||
_log.info(f"定时停止 {job.name}({job.id}): 停止 {len(stopped)} 台设备 {stopped}")
|
||||
notifier.notify("task.cron.stopped", job_id=job.id, job_name=job.name,
|
||||
stopped_count=len(stopped), serials=stopped[:10])
|
||||
|
||||
def next_run_of(self, job):
|
||||
"""任务下次真正执行的时间（考虑运行窗口）。停用/无 cron 返回 None。"""
|
||||
@@ -664,23 +704,42 @@ class TaskManager:
|
||||
serials = job.resolve_serials(self)
|
||||
if not serials:
|
||||
_log.warning(f"任务 {job.name} 无可用设备")
|
||||
notifier.notify("task.batch.no_device", job_id=job.id, job_name=job.name,
|
||||
target=job.target)
|
||||
return
|
||||
task_cls = get_task_class(job.task_type)
|
||||
if not task_cls:
|
||||
_log.error(f"未知任务类型: {job.task_type}")
|
||||
notifier.notify("task.batch.unknown_type", job_id=job.id, job_name=job.name,
|
||||
task_type=job.task_type)
|
||||
return
|
||||
task = task_cls()
|
||||
max_attempts = max(1, job.retry.get("max_attempts", 1))
|
||||
delay = job.retry.get("delay", 60)
|
||||
|
||||
notifier.notify("task.batch.started", job_id=job.id, job_name=job.name,
|
||||
task_type=job.task_type, device_count=len(serials),
|
||||
serials_preview=serials[:5])
|
||||
# 批次收尾统计：所有设备都出结果后发一条汇总（见 _BatchTracker）
|
||||
# 设备名一次查好带下去（worker 线程里没有 app context，查库要显式包 context）
|
||||
names = {}
|
||||
try:
|
||||
with self._db():
|
||||
names = {d["serial"]: (d.get("name") or "")
|
||||
for d in device_pool.list_devices()}
|
||||
except Exception as e:
|
||||
_log.warning(f"读取设备名失败（通知里将显示地址）: {e}")
|
||||
tracker = _BatchTracker(job, len(serials), names)
|
||||
|
||||
for idx, serial in enumerate(serials):
|
||||
# 每台设备一个重试循环线程，互不影响；错峰延迟在各自线程内等待
|
||||
t = threading.Thread(target=self._run_with_retry,
|
||||
args=(task, serial, job, max_attempts, delay,
|
||||
idx * _START_STAGGER_SEC), daemon=True)
|
||||
idx * _START_STAGGER_SEC, tracker), daemon=True)
|
||||
t.start()
|
||||
|
||||
def _run_with_retry(self, task, serial, job, max_attempts, delay, start_delay=0):
|
||||
def _run_with_retry(self, task, serial, job, max_attempts, delay, start_delay=0,
|
||||
tracker=None):
|
||||
"""单设备任务执行 + 重试。
|
||||
|
||||
异常分类：
|
||||
@@ -691,36 +750,58 @@ class TaskManager:
|
||||
# 错峰启动：在各自线程内等待，分摊批量启动的连接/占用压力
|
||||
if start_delay > 0:
|
||||
time.sleep(start_delay)
|
||||
# 设备名（通知里显示"设备：A08"而不是裸地址）
|
||||
dname = (tracker.names.get(serial) if tracker is not None else "") or serial
|
||||
# 被抢占的原任务 id（抢占结束后归还）——必须在循环外：
|
||||
# 重试时若重置为 None，finally 归还逻辑会丢失信息，被抢占任务永不恢复
|
||||
preempted_job = None
|
||||
# 本台设备的结果（供批次统计与通知用）。默认 "failed"：走到重试耗尽就是失败；
|
||||
# 各提前 return 的分支会覆盖它。**在 finally 里统一上报**，保证不漏不重。
|
||||
outcome = "failed"
|
||||
try:
|
||||
for attempt in range(1, max_attempts + 1):
|
||||
# 用户已请求停止 → 不再启动新 attempt
|
||||
stopped_by_user = False
|
||||
with self._lock:
|
||||
if serial in self._stop_requested:
|
||||
_log.info(f"{serial} 用户已请求停止，取消重试 (job={job.name})")
|
||||
return
|
||||
stopped_by_user = True
|
||||
if stopped_by_user:
|
||||
outcome = "stopped"
|
||||
notifier.notify("task.device.stopped", serial=serial, device_name=dname,
|
||||
job_id=job.id, job_name=job.name, attempt=attempt,
|
||||
phase="start")
|
||||
return
|
||||
# 同一 serial 同时只能一个 worker；不同任务可配置抢占
|
||||
preempt = False
|
||||
skip_reason = ""
|
||||
with self._lock:
|
||||
if serial in self._running:
|
||||
cur = self._running[serial]
|
||||
if cur.get("job_id") == job.id:
|
||||
# 同一任务重复触发：跳过（原行为）
|
||||
_log.warning(f"{serial} 已有任务在跑，跳过 (job={job.name})")
|
||||
return
|
||||
if not job.params.get("preempt"):
|
||||
skip_reason = "同一任务已在运行"
|
||||
elif not job.params.get("preempt"):
|
||||
# 不同任务且本任务未开启抢占：跳过
|
||||
_log.warning(f"{serial} 已有任务在跑，跳过 (job={job.name})")
|
||||
return
|
||||
preempt = True
|
||||
if not preempted_job:
|
||||
preempted_job = cur.get("job_id")
|
||||
skip_reason = "设备被其他任务占用"
|
||||
else:
|
||||
preempt = True
|
||||
if not preempted_job:
|
||||
preempted_job = cur.get("job_id")
|
||||
if skip_reason:
|
||||
outcome = "skipped"
|
||||
return
|
||||
if preempt:
|
||||
# 抢占：锁外停止该设备上的其他任务（stop_device 内部拿同一把锁，
|
||||
# 在锁内调用会死锁！），等其释放后接管
|
||||
_log.warning(f"{serial} 抢占：停止任务 {preempted_job} 后执行 {job.name}")
|
||||
notifier.notify("task.device.preempted", serial=serial, device_name=dname,
|
||||
job_id=job.id, job_name=job.name,
|
||||
preempted_job_id=preempted_job,
|
||||
preempted_job_name=(self.jobs.get(preempted_job).name
|
||||
if self.jobs.get(preempted_job) else ""))
|
||||
self.stop_device(serial)
|
||||
deadline = time.time() + 30
|
||||
while time.time() < deadline:
|
||||
@@ -730,6 +811,10 @@ class TaskManager:
|
||||
time.sleep(0.5)
|
||||
else:
|
||||
_log.warning(f"{serial} 抢占超时（旧任务 30s 未退出），跳过 (job={job.name})")
|
||||
outcome = "skipped"
|
||||
notifier.notify("task.device.preempt_timeout", serial=serial,
|
||||
device_name=dname, job_id=job.id, job_name=job.name,
|
||||
preempted_job_id=preempted_job)
|
||||
return
|
||||
with self._lock:
|
||||
self._running[serial] = {"job_id": job.id, "started_at": time.time(),
|
||||
@@ -739,6 +824,7 @@ class TaskManager:
|
||||
max_attempts=max_attempts)
|
||||
|
||||
worker = None
|
||||
t_attempt = time.time() # 单次 attempt 耗时（通知里带）
|
||||
try:
|
||||
worker = task.create_worker(serial, job.params)
|
||||
with self._lock:
|
||||
@@ -756,11 +842,22 @@ class TaskManager:
|
||||
_update_status(serial, task_job="")
|
||||
if st == "done":
|
||||
_log.info(f"{serial} 任务 {job.name} 成功完成")
|
||||
outcome = "success"
|
||||
notifier.notify("task.device.success", serial=serial,
|
||||
device_name=dname, job_id=job.id,
|
||||
job_name=job.name, attempt=attempt,
|
||||
duration_s=int(time.time() - t_attempt),
|
||||
is_time_up=bool(worker.is_time_up()))
|
||||
return
|
||||
# 用户请求停止（无论 attempt 第几次、status 是什么）→ 不重试
|
||||
with self._lock:
|
||||
if serial in self._stop_requested:
|
||||
_log.info(f"{serial} 用户已请求停止，不再重试 (job={job.name})")
|
||||
outcome = "stopped"
|
||||
notifier.notify("task.device.stopped", serial=serial,
|
||||
device_name=dname, job_id=job.id,
|
||||
job_name=job.name, attempt=attempt,
|
||||
phase="after")
|
||||
return
|
||||
_log.warning(f"{serial} 任务未成功(status={st})")
|
||||
except DeviceOfflineError as e:
|
||||
@@ -770,30 +867,46 @@ class TaskManager:
|
||||
self._running.pop(serial, None)
|
||||
_update_status(serial, status="failed", task_job="",
|
||||
last_error=f"设备离线: {e}")
|
||||
outcome = "failed"
|
||||
notifier.notify("task.device.offline", serial=serial, device_name=dname,
|
||||
job_id=job.id, job_name=job.name, error=str(e)[:200])
|
||||
return
|
||||
except Exception as e:
|
||||
_log.error(f"{serial} 执行异常: {e}", exc_info=True)
|
||||
with self._lock:
|
||||
self._running.pop(serial, None)
|
||||
_update_status(serial, task_job="")
|
||||
notifier.notify("task.device.error", serial=serial, device_name=dname,
|
||||
job_name=job.name, attempt=attempt,
|
||||
error=str(e)[:200])
|
||||
|
||||
if attempt < max_attempts:
|
||||
# 用户在 sleep 期间点停止也能中断
|
||||
with self._lock:
|
||||
if serial in self._stop_requested:
|
||||
_log.info(f"{serial} 用户已请求停止，取消重试 (job={job.name})")
|
||||
outcome = "stopped"
|
||||
notifier.notify("task.device.stopped", serial=serial,
|
||||
device_name=dname, job_name=job.name,
|
||||
attempt=attempt, phase="retry-wait")
|
||||
return
|
||||
# 检查是否是端口耗尽类临时错误，需要更长退避等端口释放
|
||||
with _WORKERS_LOCK:
|
||||
err = _WORKERS.get(serial, {}).get("last_error", "")
|
||||
if err.startswith("[transient]"):
|
||||
transient = err.startswith("[transient]")
|
||||
if transient:
|
||||
# Windows TCP 端口耗尽，TIME_WAIT 默认 2-4 分钟，等 120 秒
|
||||
extra_delay = max(delay, 120)
|
||||
_log.info(f"{serial} ADB 连接临时错误（端口耗尽），{extra_delay}s 后重试 ({attempt+1}/{max_attempts})")
|
||||
time.sleep(extra_delay)
|
||||
wait = extra_delay
|
||||
else:
|
||||
_log.info(f"{serial} {delay}s 后重试 ({attempt+1}/{max_attempts})")
|
||||
time.sleep(delay)
|
||||
wait = delay
|
||||
notifier.notify("task.device.retry", serial=serial, device_name=dname,
|
||||
job_name=job.name, attempt=attempt,
|
||||
next_attempt=attempt + 1, delay_s=int(wait),
|
||||
reason="transient" if transient else "normal")
|
||||
time.sleep(wait)
|
||||
|
||||
_log.error(f"{serial} 任务 {job.name} 重试耗尽，放弃")
|
||||
# 带上最后一次的真实失败原因——只写"重试N次失败"会让用户看不到为什么失败
|
||||
@@ -805,20 +918,37 @@ class TaskManager:
|
||||
if cause and cause != msg:
|
||||
msg += f"：{cause}"
|
||||
_update_status(serial, status="failed", last_error=msg[:200])
|
||||
outcome = "failed"
|
||||
notifier.notify("task.device.failed", serial=serial, device_name=dname,
|
||||
job_id=job.id, job_name=job.name, attempts=max_attempts,
|
||||
cause=cause, msg=msg)
|
||||
finally:
|
||||
# 清除停止标志：整个重试循环结束（成功/失败/停止）后允许下次任务
|
||||
with self._lock:
|
||||
self._stop_requested.discard(serial)
|
||||
# 批次统计：**每台设备只在这里上报一次**（所有 return 分支都会走到 finally）
|
||||
if tracker is not None:
|
||||
try:
|
||||
tracker.done(serial, outcome)
|
||||
except Exception as e:
|
||||
_log.warning(f"批次统计上报失败（不影响任务）: {e}")
|
||||
# 归还：本任务是抢占任务，结束后自动重新启动被抢占的原任务
|
||||
if preempted_job:
|
||||
j = self.jobs.get(preempted_job)
|
||||
if j and j.enabled:
|
||||
_log.info(f"{serial} 抢占任务 {job.name} 结束，归还设备给任务 {j.name}")
|
||||
notifier.notify("task.device.released", serial=serial, device_name=dname,
|
||||
job_name=job.name, preempted_job_id=preempted_job,
|
||||
returned=True, reason="归还并重启被抢占任务")
|
||||
threading.Thread(target=self._run_job, args=(j,), daemon=True).start()
|
||||
else:
|
||||
# 打日志便于排查：被抢占任务已删除/停用时不会归还，但要知道原因
|
||||
_log.info(f"{serial} 抢占任务 {job.name} 结束，"
|
||||
f"被抢占任务 {preempted_job} {'已停用，不归还' if j else '已不存在，不归还'}")
|
||||
notifier.notify("task.device.released", serial=serial, device_name=dname,
|
||||
job_name=job.name, preempted_job_id=preempted_job,
|
||||
returned=False,
|
||||
reason="被抢占任务已停用" if j else "被抢占任务已不存在")
|
||||
|
||||
# ---- 运行控制 ----
|
||||
def stop_device(self, serial):
|
||||
|
||||
Reference in New Issue
Block a user