butubb
c29516cff5
需求：任务编辑器里能单独配"这个任务每隔 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。
111 lines
4.6 KiB
Python
111 lines
4.6 KiB
Python
"""任务内**公共巡检**：任务运行期间按间隔穿插检查设备状态，命中就动作 + 通知。
|
||
|
||
和步骤编辑器的关系（重要，别混）：
|
||
|
||
| | 步骤（`params.steps`） | 公共巡检（`params.watchers`） |
|
||
|---|---|---|
|
||
| 配置位置 | 步骤画布，拖出来的 | 任务编辑器里单独一块 |
|
||
| 执行方式 | 顺序执行 | **穿插**：每执行完一步（以及长等待的每个分片）看一眼"哪个巡检到点了" |
|
||
| 适合什么 | 一段固定流程 | 全程都要盯的守护条件（熄屏就点亮、掉到桌面就通知） |
|
||
|
||
为什么用穿插而不是另起线程：不需要并发模型，不会和主流程抢屏幕（否则巡检点屏幕、
|
||
主流程也在点，动作会互相打断）。代价是巡检精度受步长影响——某一步卡 30 秒，
|
||
巡检最多晚 30 秒；这也是刻意的取舍（见 doc/TASK_DEV.md §4.5）。
|
||
|
||
配置结构（`params.watchers` 数组，每项）：
|
||
|
||
```json
|
||
{"name": "熄屏点亮", "enabled": true, "interval": 60,
|
||
"check": "screen_off", // 见 CHECKS
|
||
"selector_type": "xpath", "selector_value": "", // 元素类检查才用
|
||
"action": "screen_on", // 见 ACTIONS
|
||
"notify": true, "title": "{device} 熄屏了",
|
||
"message": "任务 {job} 在 {time} 发现 {screen}，已点亮",
|
||
"cooldown": 300} // 命中后多久内不再重复动作/通知（防刷屏）
|
||
```
|
||
"""
|
||
from core.logger import get_logger
|
||
from core.u2_helper import screen_is_on, current_package
|
||
|
||
_log = get_logger("task.generic")
|
||
|
||
# 检查项：need 说明该项还要填什么（selector=元素选择器 / package=包名）
|
||
CHECKS = {
|
||
"screen_off": {"label": "屏幕熄灭", "need": ""},
|
||
"screen_on": {"label": "屏幕亮着", "need": ""},
|
||
"element_exists": {"label": "元素存在", "need": "selector"},
|
||
"element_missing": {"label": "元素不存在", "need": "selector"},
|
||
"foreground_is": {"label": "前台是该App", "need": "package"},
|
||
"foreground_not": {"label": "前台不是该App", "need": "package"},
|
||
}
|
||
|
||
ACTIONS = {
|
||
"none": {"label": "只发通知，不动设备"},
|
||
"screen_on": {"label": "点亮屏幕"},
|
||
"screen_off": {"label": "熄灭屏幕"},
|
||
"stop_self": {"label": "停止本设备任务"},
|
||
}
|
||
|
||
# 通知正文里可用的占位符（前端提示、文档、渲染三处共用一份，避免漂移）
|
||
PLACEHOLDERS = ("device", "serial", "job", "time", "app", "screen")
|
||
|
||
|
||
def _element_exists(d, sel_type, sel_val):
|
||
"""元素是否存在（不等超时，立即判定）。一次 UI dump 的代价，慢设备上可能要几秒。"""
|
||
if sel_type == "xpath":
|
||
return bool(d.xpath(sel_val).exists())
|
||
return bool(d(**{sel_type: sel_val}).exists())
|
||
|
||
|
||
def evaluate(d, spec):
|
||
"""跑一次检查，返回 (是否命中, 一句人话说明)。异常由调用方兜。"""
|
||
check = str(spec.get("check") or "")
|
||
val = str(spec.get("selector_value") or "").strip()
|
||
|
||
if check in ("screen_off", "screen_on"):
|
||
state = screen_is_on(d)
|
||
if state is None:
|
||
return False, "屏幕状态读不到"
|
||
hit = (not state) if check == "screen_off" else state
|
||
return hit, f"屏幕{'亮着' if state else '熄灭'}"
|
||
|
||
if check in ("element_exists", "element_missing"):
|
||
if not val:
|
||
return False, "没填选择器"
|
||
found = _element_exists(d, spec.get("selector_type") or "xpath", val)
|
||
return (found if check == "element_exists" else not found), \
|
||
f"元素{'存在' if found else '不存在'}"
|
||
|
||
if check in ("foreground_is", "foreground_not"):
|
||
if not val:
|
||
return False, "没填包名"
|
||
cur = current_package(d)
|
||
return (cur == val) if check == "foreground_is" else (cur != val), \
|
||
f"前台={cur or '未知'}"
|
||
|
||
return False, f"未知检查项 {check}"
|
||
|
||
|
||
def act(d, spec, worker=None):
|
||
"""执行命中后的动作，返回一句人话（没动作返回空串）。"""
|
||
action = str(spec.get("action") or "none")
|
||
if action == "screen_on":
|
||
d.unlock() # 与 screen_on 步骤同语义：唤醒 + 解锁
|
||
return "已点亮屏幕"
|
||
if action == "screen_off":
|
||
d.screen_off()
|
||
return "已熄灭屏幕"
|
||
if action == "stop_self":
|
||
if worker is not None:
|
||
worker.stop() # 置停止位，主流程下一步就退出
|
||
return "已停止本设备任务"
|
||
return ""
|
||
|
||
|
||
def check_label(check):
|
||
return (CHECKS.get(check) or {}).get("label", check)
|
||
|
||
|
||
def action_label(action):
|
||
return (ACTIONS.get(action) or {}).get("label", action)
|