feat(任务): 任务级「公共巡检」——独立于步骤画布的守护条件(含 webhook 通知)

需求:任务编辑器里能单独配"这个任务每隔 N 秒检查一次"——熄屏就点亮、某个元素
出现就通知、掉出 App 就停本设备;通知标题正文要能自己写。

配置与执行分离(这是本次的关键设计):
- **配置是任务级的**(`params.watchers`),在任务编辑器单独一块,不进步骤画布;
- **执行是穿插的**:worker 每执行完一步、以及长等待的每个分片,看一眼哪个巡检
  到点了。不起线程 → 不需要并发模型,也不会和主流程抢屏幕(两边同时点屏幕会
  互相打断)。代价是精度受步长影响(某步卡 30s,巡检最多晚 30s),已在文档写明。

- core/patrol.py(新):检查项/动作注册表(CHECKS/ACTIONS)+ evaluate/act。
  检查:屏幕熄灭/亮着、元素存在/不存在、前台是/不是某 App;
  动作:只通知、点亮、息屏、停止本设备。屏幕走 `dumpsys power`(0.3s,
  不用 d.info——那玩意在部分设备要 14s),前台走 d.app_current()(0.7s)。
- tasks/generic/task.py:`_maybe_patrol` / `_run_patrol`(冷却、命中记一条
  步骤明细、发通知);**文案在动作之前渲染**——点亮后 {screen} 就成了"亮屏",
  用户要看的是"发现熄屏,已点亮"。
- 任务编辑器新增「公共巡检」块(static/admin/tasks.js)+ 样式;保存进 params.watchers。
- 通知:新增事件 `task.patrol.hit`(巡检命中)与 `task.notify.custom`(步骤发通知);
  给了 title 就用它当标题(不再拼前缀),level 字段可点名级别。

顺带(巡检需要的原语,也可单独用):
- if_el 条件判断支持 `selector_type=screen`(亮/熄)与 `foreground`(前台包名);
  非元素条件不参与「选择器健康」统计(否则会攒出假的"选择器失效"告警)。
- 新增两个步骤:`notify`(发自定义通知)、`stop_self`(停本设备,记"被停止"
  而不是失败,不触发重试)。步骤类型 18 → 20,相关文档计数一并更新。

修 bug:`wait` 步骤在巡检耗时超过剩余时间后 `sleep(负数)` 抛
"sleep length must be non-negative"(真机联调抓到,已 clamp 到 0)。

自测:假设备单测 12 组(命中/冷却/间隔/元素/前台/停止/静默/异常不炸);
真机联调:熄屏→点亮(False→True)+ 通知文案正确、掉出抖音按间隔命中 5 次、
长等待里穿插生效且步骤回到 ok、清理后用户通知配置原样恢复。
文档:TASK_DEV §4.5(含两个可抄的例子)与步骤表/条件类型、NOTIFY §3、README。
This commit is contained in:
2026-09-16 13:02:40 +08:00
parent 243df3071b
commit c29516cff5
15 changed files with 644 additions and 41 deletions
+29
View File
@@ -5,6 +5,7 @@
"""
import time
import random
import re
from core.logger import get_logger
@@ -65,6 +66,34 @@ def random_sleep(min_s, max_s):
time.sleep(random.uniform(min_s, max_s))
def screen_is_on(d):
"""屏幕是否亮着。返回 True/False,取不到返回 None(三态,别当 False 用)。
为什么不用 `d.info`:它在部分设备上一次要**十几秒**(见 doc/backlog/TODO.md);
`dumpsys power` 经 atx-agent 跑只要 ~0.3s,且 `mWakefulness` 字段全机型都有
(MIUI 的 `dumpsys display` 反而没有 mScreenState)。
"""
try:
out = d.shell("dumpsys power")
out = getattr(out, "output", None) or str(out)
m = re.search(r"mWakefulness=(\w+)", out)
if m:
return m.group(1).lower() != "asleep"
except Exception as e:
_log.warning(f"读取屏幕状态失败: {e}")
return None
def current_package(d):
"""当前前台 App 包名;取不到返回空串。"""
try:
cur = d.app_current() or {}
return cur.get("package") or ""
except Exception as e:
_log.warning(f"读取前台 App 失败: {e}")
return ""
def safe_click(el, timeout=1):
"""安全点击:元素存在才点,不抛异常。返回是否点击成功。"""
try: