butubb
34a03db1b9
需求原话：条件判断需要能选择元素，比如我设置了一个抖音号，让它采集这个元素的
文本和我设的值做条件判断。
原来的条件判断只会判「元素在不在」（找到走 then、没找到走 else），没法判"内容对不对"。
现在加两个可选参数：
- `cmp_op`：等于 / 不等于 / 包含 / 不包含（`==`/`!=`/`contains` 这类别名也认）；
留空 = 老行为，**完全向后兼容**（现有任务不用改）。
- `cmp_value`：要比对的值。
填了就变成：先按选择器取到元素文本 → 和 cmp_value 比 → 比对结果才是命中与否。
用户那个例子就是：元素 `com.ss.android.ugc.aweme:id/506` 的文本是
`抖音号：35377983067`，选「包含」、值填 `35377983067` 即可；
要抓"登录的不是这个号"就用「不等于」。
- 读取只走 `get_text()`（只读，2 秒超时），不点击、不改状态。
- **两条防误判规则**（都写进文档了）：
· 元素**没找到**时一律算未命中（走 else）——"没读到"绝不能被当成"和我设的不一样"，
否则界面还没加载出来就会误判成"账号被换了"；
· 元素在但文本为空时按空文本参与比对（用于抓"栏位是空的"），日志会打出实际读到的值。
- 比对**不污染「选择器健康」统计**：统计用的仍是"元素在不在"——"找到了但值不对"
是条件按预期走了 else，不是选择器失效，不该攒出「连续未命中」告警。
- 支持比对的类型：所有元素类选择器 + `foreground`（比前台包名）+ `ocr`（比 OCR 命中的那段字）。
`screen`（亮/熄）没有文本可比，设了会忽略并告警。
- 前端：条件判断面板多一块「文本比对」（运算符下拉 + 比对的值），
只在元素/OCR/前台App 条件下出现；保存前校验"选了运算符却没填值"和"screen 上设了比对"。
- AI 建任务草稿校验（`core/task_draft.py`）同步拦"运算符拼错""有运算符没值"——
执行器对认不出的运算符是"一律不命中"，不拦就会变成悄悄走 else。
自测：比对函数 16 项（含别名、空值、None、拼错运算符）、`_exec_if_el` 全分支 20 项
（含"没找到+不等于"这个关键边界、健康统计用存在性、screen 忽略比对、u2 抛异常不炸）、
**真机验证**（cs1 上真读一个 TextView 的文本再比对：包含/等于/不等于 全对，
元素不存在时不误判）、草稿校验 5 项、GET 路由冒烟 55 个 0 个 500。
（测试里踩到一个坑记一笔：u2 的 `XPathSelector.exists` 是纯 bool 属性，而
`UiObject.exists` 是"可调用的 Exists 对象"（`bool()` 即时判断、`exists(timeout=)` 带等待），
两套语义不能想当然统一——单测的假设备一开始没模拟对，是测试写错了不是代码写错了。）
557 lines
26 KiB
Python
557 lines
26 KiB
Python
"""AI 建任务的 draft 契约：**校验 + 归一化**（唯一真相）。
|
||
|
||
## 为什么必须有这个模块
|
||
|
||
任务执行器对"写错的地方"是**静默跳过**的：未知步骤类型只打一条 warning 继续
|
||
（`tasks/generic/task.py` 的 `_exec_one`），空 `selector_value` 直接 return。
|
||
而 `POST /api/jobs` 只校验 `name` + `task_type`，`params` 完全盲存
|
||
（`web/tasks_api.py`）。两者叠加的后果是：AI（或任何人）交上一条写错的任务，
|
||
**创建成功、运行不报错、但什么都没做**。
|
||
|
||
所以这里把执行器的每个"静默跳过点"前移成显式 error，在**落库之前**拦住。
|
||
错误文案是给**模型**看的（走 `submit_task` 的返回值回灌，让它逐条改完重提），
|
||
所以每条都要说清"哪个节点的哪个字段、错在哪、应该是什么"。
|
||
|
||
## 用法
|
||
|
||
res = validate_draft(draft, groups=..., pool=..., default_serial="...")
|
||
# → {"ok": bool, "errors": [...], "warnings": [...], "draft": 归一化后的草稿}
|
||
|
||
`errors` 非空就是不能落库；`warnings` 只是提醒（前端醒目展示，不拦）。
|
||
|
||
契约与规则的权威描述见 `doc/AI_TASK_GEN.md`。
|
||
"""
|
||
import re
|
||
|
||
from core.logger import get_logger
|
||
|
||
_log = get_logger("core.task_draft")
|
||
|
||
# ================== 常量（规则的唯一来源） ==================
|
||
|
||
# 单条任务最多多少个步骤节点（含嵌套），防止模型输出失控/被输出长度截断
|
||
MAX_NODES = 60
|
||
# 嵌套深度上限（执行器 >5 层直接跳过该分支，见 tasks/generic/task.py）
|
||
MAX_DEPTH = 5
|
||
# 选择器值长度上限
|
||
MAX_SELECTOR_LEN = 200
|
||
# 单条 input_text 的候选文案条数
|
||
MAX_TEXTS = 20
|
||
|
||
# 与编辑器下拉（static/admin/editor.js 的 selector_type 选项）严格同集。
|
||
# 故意不含 textContains —— 下拉里没有它，用户一改就丢。
|
||
SELECTOR_TYPES = ("xpath", "description", "text", "resourceId",
|
||
"descriptionContains", "className")
|
||
IF_SELECTOR_TYPES = SELECTOR_TYPES + ("ocr",)
|
||
|
||
# 按键白名单（u2 的 d.press 支持更多，这里只放"不会把设备弄坏"的那些：
|
||
# 不含 power/camera —— 探索期误按 power 会让设备息屏、任务中途失联）
|
||
KEY_WHITELIST = ("back", "home", "enter", "menu", "recent", "delete",
|
||
"volume_up", "volume_down", "search")
|
||
|
||
DIRECTIONS = ("up", "down", "left", "right")
|
||
LOOP_MODES = ("rounds", "time", "forever")
|
||
SCHEDULE_MODES = ("once", "cron", "cron_stop")
|
||
TARGET_MODES = ("all", "group", "serial")
|
||
|
||
# 需要选择器的步骤类型（缺 selector_value 时执行器静默跳过）
|
||
NEED_SELECTOR = ("click", "long_click", "wait_el", "swipe_until", "if_el")
|
||
# 需要包名的步骤类型
|
||
NEED_PACKAGE = ("open_app", "stop_app")
|
||
# 需要非空 children 的容器
|
||
NEED_CHILDREN = ("loop", "group")
|
||
|
||
# 有副作用的动作关键词：命中的步骤在 AI 通道只给 warning（人工复核），不拦
|
||
DESTRUCTIVE_HINTS = ("评论", "发送", "发布", "投稿", "转发", "分享到",
|
||
"购买", "下单", "支付", "付款", "删除", "卸载",
|
||
"退出登录", "注销", "举报", "拉黑", "清空")
|
||
# 命中上面关键词的步骤，触发概率被压到这个值（可人工改回）
|
||
SAFE_PROBABILITY = 30
|
||
|
||
_PKG_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]*(\.[A-Za-z0-9_]+)+$")
|
||
_HHMM_RE = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$")
|
||
# 「序号型」选择器的两种形态：
|
||
# (//*[@resource-id="x"])[4] ← 抓取器在语义消歧失败时给的（当前形态）
|
||
# //*[@resource-id="x"][4] ← 历史形态（执行器会自动纠正成上面那种）
|
||
# 两者都依赖"同类元素有几个"，界面一变就指到别的元素上，都值得提醒换语义选择器。
|
||
_SEQ_SELECTOR_RE = re.compile(r"^(?:\(\s*//\*\[@[^\]]+\]\s*\)|//\*\[@[^\]]+\])\[\d+\]")
|
||
|
||
|
||
def _step_types():
|
||
"""步骤类型 → (label, 默认 params)，直接取自执行器的权威定义。"""
|
||
from tasks.generic.task import STEP_TYPES
|
||
return {s["type"]: (s.get("label", s["type"]), dict(s.get("params") or {}))
|
||
for s in STEP_TYPES}
|
||
|
||
|
||
def _is_num(v):
|
||
return isinstance(v, (int, float)) and not isinstance(v, bool)
|
||
|
||
|
||
def _num(params, key, default=None):
|
||
v = params.get(key, default)
|
||
return v if _is_num(v) else default
|
||
|
||
|
||
# ================== 步骤树校验 ==================
|
||
|
||
def validate_steps(steps, depth=1, path="steps", errors=None, warnings=None,
|
||
types=None, counter=None):
|
||
"""递归校验步骤树，返回 (errors, warnings, 节点总数)。
|
||
|
||
每条 error 都带上路径（如 `steps[0].params.children[2]`），模型据此能直接定位。
|
||
"""
|
||
errors = [] if errors is None else errors
|
||
warnings = [] if warnings is None else warnings
|
||
types = _step_types() if types is None else types
|
||
counter = [0] if counter is None else counter
|
||
|
||
if not isinstance(steps, list):
|
||
errors.append(f"{path}: 必须是数组")
|
||
return errors, warnings, counter[0]
|
||
|
||
for i, step in enumerate(steps):
|
||
here = f"{path}[{i}]"
|
||
if not isinstance(step, dict):
|
||
errors.append(f"{here}: 必须是对象")
|
||
continue
|
||
stype = step.get("type")
|
||
if not stype:
|
||
errors.append(f"{here}: 缺少 type 字段")
|
||
continue
|
||
if stype not in types:
|
||
errors.append(f"{here}: 未知步骤类型 {stype!r}（可用类型："
|
||
f"{'、'.join(types)}）")
|
||
continue
|
||
if stype == "click_xy":
|
||
errors.append(f"{here}: 不允许坐标点击（click_xy）—— 坐标在不同分辨率/"
|
||
"设备上必失配，请改用 click + 选择器定位元素")
|
||
continue
|
||
|
||
counter[0] += 1
|
||
if counter[0] > MAX_NODES:
|
||
errors.append(f"{here}: 步骤节点总数超过 {MAX_NODES} 个，请精简"
|
||
"（去掉冗余的等待/滑动，合并重复结构）")
|
||
return errors, warnings, counter[0]
|
||
|
||
params = step.get("params")
|
||
if params is None:
|
||
params = {}
|
||
if not isinstance(params, dict):
|
||
errors.append(f"{here}.params: 必须是对象")
|
||
continue
|
||
|
||
# 触发概率（执行器对所有类型通用）
|
||
prob = params.get("probability")
|
||
if prob is not None:
|
||
if not _is_num(prob) or not (0 <= prob <= 100):
|
||
errors.append(f"{here}.params.probability: 必须是 0~100 的数字"
|
||
f"（当前 {prob!r}）")
|
||
|
||
# 选择器类步骤
|
||
if stype in NEED_SELECTOR:
|
||
sel = params.get("selector_value")
|
||
if not isinstance(sel, str) or not sel.strip():
|
||
errors.append(f"{here}.params.selector_value: 不能为空 —— "
|
||
"该步骤没有定位目标，执行时会被静默跳过。"
|
||
"如果这一步还没探索出可靠的定位，请把它从 steps 里去掉，"
|
||
"需要人工确认的写进 notes")
|
||
else:
|
||
_check_selector(params, here, stype, errors, warnings)
|
||
|
||
# 包名类步骤
|
||
if stype in NEED_PACKAGE:
|
||
pkg = params.get("package")
|
||
if not isinstance(pkg, str) or not _PKG_RE.match(pkg.strip()):
|
||
errors.append(f"{here}.params.package: 必须是包名（如 "
|
||
"com.ss.android.ugc.aweme），当前 "
|
||
f"{pkg!r}")
|
||
|
||
# 容器
|
||
if stype in NEED_CHILDREN:
|
||
children = params.get("children")
|
||
if not isinstance(children, list) or not children:
|
||
errors.append(f"{here}.params.children: 不能为空 —— "
|
||
"空容器等于什么都没做")
|
||
continue
|
||
if depth >= MAX_DEPTH:
|
||
errors.append(f"{here}: 嵌套层级超过 {MAX_DEPTH} 层"
|
||
"（执行器会直接跳过这个分支）")
|
||
continue
|
||
validate_steps(children, depth + 1, f"{here}.params.children",
|
||
errors, warnings, types, counter)
|
||
|
||
if stype == "loop":
|
||
mode, fixed = _norm_loop_mode(params.get("loop_mode"))
|
||
params["loop_mode"] = mode
|
||
if fixed:
|
||
warnings.append(f"{here}.params.loop_mode: 已按 {fixed!r} 归一为 "
|
||
f"{mode!r}（执行器对未知取值按轮次处理）")
|
||
if mode == "rounds":
|
||
it = params.get("max_iterations")
|
||
if it is None:
|
||
# 执行器默认 10（tasks/generic/task.py 的 _exec_loop）——
|
||
# 能跑通的就不该拦，补默认值并提示即可
|
||
params["max_iterations"] = 10
|
||
warnings.append(f"{here}.params.max_iterations: 未填，按默认 10 轮")
|
||
elif not _is_num(it) or it < 1:
|
||
errors.append(f"{here}.params.max_iterations: 按轮次循环时"
|
||
f"必须 ≥1（当前 {it!r}）")
|
||
elif mode == "time":
|
||
dur = params.get("loop_duration")
|
||
# 按时间循环缺时长时执行器**直接跳过**整块（静默无操作）→ 必须拦
|
||
if not _is_num(dur) or dur < 10:
|
||
errors.append(f"{here}.params.loop_duration: 按时间循环时"
|
||
f"必须 ≥10 秒（当前 {dur!r}）")
|
||
|
||
if stype == "if_el":
|
||
then = params.get("then")
|
||
if not isinstance(then, list) or not then:
|
||
errors.append(f"{here}.params.then: 不能为空 —— "
|
||
"条件命中后必须做点什么")
|
||
if depth >= MAX_DEPTH:
|
||
errors.append(f"{here}: 嵌套层级超过 {MAX_DEPTH} 层")
|
||
continue
|
||
if isinstance(then, list):
|
||
validate_steps(then, depth + 1, f"{here}.params.then",
|
||
errors, warnings, types, counter)
|
||
else_ = params.get("else")
|
||
if isinstance(else_, list) and else_:
|
||
validate_steps(else_, depth + 1, f"{here}.params.else",
|
||
errors, warnings, types, counter)
|
||
# 文本比对（可选）：运算符必须认得，且配了运算符就得有值——
|
||
# 执行器对认不出的运算符是"一律不命中"，写错了会静默走 else，
|
||
# 所以在这里拦住，别让 AI 草稿悄悄变成"永远不命中"
|
||
cmp_op = (params.get("cmp_op") or "").strip()
|
||
if cmp_op:
|
||
from tasks.generic.task import CMP_OPS
|
||
if cmp_op not in CMP_OPS:
|
||
errors.append(f"{here}.params.cmp_op: 不支持的比对运算符 {cmp_op!r}"
|
||
f"（可用：{'、'.join(CMP_OPS)}）")
|
||
elif not str(params.get("cmp_value") or "").strip():
|
||
errors.append(f"{here}.params.cmp_value: 选了文本比对"
|
||
f"（{cmp_op}）就必须填比对的值，否则永远不命中")
|
||
elif (params.get("selector_type") or "xpath") == "screen":
|
||
warnings.append(f"{here}: 屏幕状态没有文本可比，"
|
||
"cmp_op/cmp_value 会被忽略")
|
||
|
||
if stype == "swipe":
|
||
_check_direction(params, here, errors)
|
||
dmin, dmax = _num(params, "duration_min"), _num(params, "duration_max")
|
||
if dmin is not None and dmax is not None and dmin > dmax:
|
||
errors.append(f"{here}.params: duration_min({dmin}) 不能大于 "
|
||
f"duration_max({dmax})")
|
||
|
||
if stype == "swipe_until":
|
||
_check_direction(params, here, errors)
|
||
n = params.get("max_swipes")
|
||
if not _is_num(n) or not (1 <= n <= 50):
|
||
errors.append(f"{here}.params.max_swipes: 必须是 1~50（当前 {n!r}）")
|
||
|
||
if stype == "wait":
|
||
lo, hi = _num(params, "min"), _num(params, "max")
|
||
if lo is None or hi is None or lo < 0 or hi > 600 or lo > hi:
|
||
errors.append(f"{here}.params: 需要 0 ≤ min ≤ max ≤ 600"
|
||
f"（当前 min={lo!r} max={hi!r}）")
|
||
|
||
if stype == "input_text":
|
||
mode = params.get("mode", "random")
|
||
if mode not in ("random", "fixed"):
|
||
errors.append(f"{here}.params.mode: 必须是 random 或 fixed"
|
||
f"（当前 {mode!r}）")
|
||
elif mode == "fixed":
|
||
if not str(params.get("fixed_text") or "").strip():
|
||
errors.append(f"{here}.params.fixed_text: mode=fixed 时不能为空")
|
||
else:
|
||
texts = params.get("texts")
|
||
if not isinstance(texts, str) or not texts.strip():
|
||
errors.append(f"{here}.params.texts: mode=random 时必须给候选"
|
||
"文案（换行分隔，如 \"你好\\n不错\"）")
|
||
elif len([t for t in texts.split("\n") if t.strip()]) > MAX_TEXTS:
|
||
errors.append(f"{here}.params.texts: 候选文案最多 {MAX_TEXTS} 条")
|
||
|
||
if stype == "clipboard":
|
||
if not str(params.get("text") or "").strip():
|
||
errors.append(f"{here}.params.text: 剪贴板内容不能为空")
|
||
|
||
if stype == "keep_screen":
|
||
if params.get("mode", "on") not in ("on", "off"):
|
||
errors.append(f"{here}.params.mode: 必须是 on 或 off")
|
||
|
||
if stype == "key_event":
|
||
key = params.get("key", "back")
|
||
if key not in KEY_WHITELIST:
|
||
errors.append(f"{here}.params.key: 不支持的按键 {key!r}"
|
||
f"（可用：{'、'.join(KEY_WHITELIST)}）")
|
||
|
||
_check_destructive(step, here, warnings)
|
||
|
||
return errors, warnings, counter[0]
|
||
|
||
|
||
def _check_selector(params, here, stype, errors, warnings):
|
||
"""选择器类型/取值合法性（跨 xpath 与 u2 kwarg 两套语义）。"""
|
||
stype_ok = IF_SELECTOR_TYPES if stype == "if_el" else SELECTOR_TYPES
|
||
sel_type = params.get("selector_type") or "xpath"
|
||
value = (params.get("selector_value") or "").strip()
|
||
if sel_type not in stype_ok:
|
||
errors.append(f"{here}.params.selector_type: 不支持 {sel_type!r}"
|
||
f"（可用：{'、'.join(stype_ok)}）")
|
||
return
|
||
if len(value) > MAX_SELECTOR_LEN:
|
||
errors.append(f"{here}.params.selector_value: 太长（>{MAX_SELECTOR_LEN} 字符）")
|
||
return
|
||
if sel_type == "xpath":
|
||
if not (value.startswith("//") or value.startswith("(//")):
|
||
errors.append(f"{here}.params.selector_value: 类型是 xpath，但取值不是 "
|
||
f"// 或 (// 开头（当前 {value[:40]!r}）")
|
||
elif _SEQ_SELECTOR_RE.match(value):
|
||
# 执行器有 _norm_legacy_xpath 兜底纠正，所以只提醒不拦：
|
||
# 序号型选择器「同类元素个数一变就失配」，能换成文字限定就用文字
|
||
warnings.append(f"{here}.params.selector_value: 序号型选择器"
|
||
f"（{value[:40]}）依赖同类元素个数，界面一变就会点空，"
|
||
"建议改用 @text/@content-desc 限定")
|
||
|
||
|
||
# 模型常见的近义写法 → 执行器认的枚举值。
|
||
# 归一化而不是报错：执行器对未知 loop_mode 是按"轮次"处理的（else 分支），
|
||
# 拦下来只会让模型反复重试（实测把一个 40 步的探索硬生生耗在改这一个字段上）。
|
||
_LOOP_MODE_ALIAS = {
|
||
"count": "rounds", "times": "rounds", "round": "rounds",
|
||
"iterations": "rounds", "iteration": "rounds", "n": "rounds",
|
||
"duration": "time", "seconds": "time", "secs": "time", "timeout": "time",
|
||
"infinite": "forever", "infinity": "forever", "until_stop": "forever",
|
||
"while": "forever", "loop": "forever",
|
||
}
|
||
|
||
|
||
def _norm_loop_mode(mode):
|
||
"""返回 (规范化后的 loop_mode, 原始值或 None)。"""
|
||
if mode in LOOP_MODES:
|
||
return mode, None
|
||
m = str(mode or "").strip().lower()
|
||
if m in _LOOP_MODE_ALIAS:
|
||
return _LOOP_MODE_ALIAS[m], mode
|
||
return "rounds", mode if mode is not None else None
|
||
|
||
|
||
def _check_direction(params, here, errors):
|
||
d = params.get("direction", "up")
|
||
if d not in DIRECTIONS:
|
||
errors.append(f"{here}.params.direction: 必须是 "
|
||
f"{'/'.join(DIRECTIONS)} 之一（当前 {d!r}）")
|
||
|
||
|
||
def _check_destructive(step, here, warnings):
|
||
"""有副作用的步骤给 warning（不拦）：让人知道哪里需要复核。"""
|
||
text = " ".join(str(step.get(f) or "") for f in ("label",))
|
||
text += " " + str((step.get("params") or {}).get("selector_value") or "")
|
||
text += " " + str((step.get("params") or {}).get("texts") or "")
|
||
hits = [w for w in DESTRUCTIVE_HINTS if w in text]
|
||
if not hits:
|
||
return
|
||
tags = "、".join(hits)
|
||
# 有副作用的动作默认调低触发概率（安全默认值，编辑器里可改回）：
|
||
# 探索期本来就不许真做这类动作（只核对元素存在），产物更不该一上来就每次都触发
|
||
params = step.setdefault("params", {})
|
||
prob = params.get("probability", 100)
|
||
if _is_num(prob) and prob > SAFE_PROBABILITY:
|
||
params["probability"] = SAFE_PROBABILITY
|
||
warnings.append(f"{here}: 含「{tags}」类有副作用的操作，触发概率已从 {prob}% "
|
||
f"降到 {SAFE_PROBABILITY}% —— 核对无误后可在步骤编辑器里改回")
|
||
else:
|
||
warnings.append(f"{here}: 含「{tags}」类有副作用的操作，"
|
||
"请人工复核后再启用")
|
||
|
||
|
||
# ================== draft 整体 ==================
|
||
|
||
def validate_draft(draft, groups=None, pool=None, default_serial="", overrides=None):
|
||
"""校验并归一化 AI 提交的任务草稿。
|
||
|
||
draft —— 模型提交的原始对象
|
||
groups —— 现有分组名集合（校验 target.mode=group 时用；None=不校验）
|
||
pool —— 设备池 serial 集合（用于给"设备不在池"的 warning；None=不校验）
|
||
default_serial —— 本次探索用的设备（target 缺省时兜底）
|
||
overrides —— 页面上的"任务设置"覆盖项（用户在页面上填的优先）
|
||
|
||
→ {"ok", "errors", "warnings", "draft"}
|
||
"""
|
||
errors, warnings = [], []
|
||
if not isinstance(draft, dict):
|
||
return {"ok": False, "errors": ["draft 必须是对象"], "warnings": [],
|
||
"draft": None}
|
||
|
||
# ---- summary ----
|
||
summary = str(draft.get("summary") or "").strip()
|
||
if not summary:
|
||
errors.append("summary: 不能为空 —— 用一句话说明这条任务做什么")
|
||
elif len(summary) > 200:
|
||
summary = summary[:200]
|
||
|
||
# ---- task ----
|
||
task = draft.get("task")
|
||
if not isinstance(task, dict):
|
||
errors.append("task: 缺少任务对象")
|
||
task = {}
|
||
task = _normalize_task(task, summary, default_serial, overrides or {},
|
||
errors, warnings)
|
||
|
||
types = _step_types()
|
||
steps = ((task.get("params") or {}).get("steps")) or []
|
||
if not steps:
|
||
errors.append("task.params.steps: 不能为空 —— 没有步骤的任务创建后"
|
||
"每次运行都会报错")
|
||
else:
|
||
validate_steps(steps, 1, "task.params.steps", errors, warnings, types)
|
||
|
||
_check_target(task, groups, pool, errors, warnings)
|
||
_check_schedule(task, errors, warnings)
|
||
|
||
return {"ok": not errors, "errors": errors, "warnings": warnings,
|
||
"draft": {"summary": summary, "task": task,
|
||
"notes": [str(n) for n in (draft.get("notes") or [])][:5],
|
||
"evidence": list(draft.get("evidence") or [])[:10]}}
|
||
|
||
|
||
def _normalize_task(task, summary, default_serial, overrides, errors, warnings):
|
||
"""补齐缺失字段（缺什么补什么，但**不掩盖**该报错的必填项）。"""
|
||
name = str(overrides.get("name") or task.get("name") or "").strip()
|
||
if not name:
|
||
name = (summary[:20] or "AI 任务").strip()
|
||
task["name"] = name[:40]
|
||
|
||
task_type = task.get("task_type") or "generic_steps"
|
||
if task_type != "generic_steps":
|
||
errors.append(f"task.task_type: 目前只支持 generic_steps（收到 {task_type!r}）")
|
||
task["task_type"] = "generic_steps"
|
||
|
||
target = task.get("target")
|
||
if not isinstance(target, dict) or not target.get("mode"):
|
||
target = {"mode": "serial", "serial": default_serial} if default_serial \
|
||
else {"mode": "all"}
|
||
warnings.append(f"task.target: 未指定，已按探索设备兜底为 {target}")
|
||
# 页面上显式选了目标就以页面为准
|
||
ov_mode = overrides.get("target_mode")
|
||
if ov_mode in TARGET_MODES:
|
||
target = {"mode": ov_mode}
|
||
if ov_mode == "group":
|
||
target["group_name"] = overrides.get("group_name") or ""
|
||
elif ov_mode == "serial":
|
||
target["serial"] = overrides.get("serial") or default_serial
|
||
task["target"] = target
|
||
|
||
schedule = task.get("schedule")
|
||
if not isinstance(schedule, dict) or not schedule.get("mode"):
|
||
schedule = {"mode": "once"}
|
||
if overrides.get("schedule"):
|
||
schedule = overrides["schedule"]
|
||
task["schedule"] = schedule
|
||
|
||
retry = task.get("retry")
|
||
if not isinstance(retry, dict):
|
||
retry = {}
|
||
attempts = retry.get("max_attempts", 1)
|
||
delay = retry.get("delay", 60)
|
||
if not _is_num(attempts) or not (1 <= attempts <= 10):
|
||
warnings.append(f"task.retry.max_attempts: 应为 1~10，已改为 1（原 {attempts!r}）")
|
||
attempts = 1
|
||
if not _is_num(delay) or not (10 <= delay <= 3600):
|
||
warnings.append(f"task.retry.delay: 应为 10~3600 秒，已改为 60（原 {delay!r}）")
|
||
delay = 60
|
||
task["retry"] = {"max_attempts": int(attempts), "delay": int(delay)}
|
||
task["enabled"] = bool(task.get("enabled", True))
|
||
|
||
params = task.get("params")
|
||
if not isinstance(params, dict):
|
||
params = {}
|
||
dur = overrides.get("max_duration", params.get("max_duration", 0))
|
||
if not _is_num(dur) or dur < 0 or dur > 604800:
|
||
errors.append(f"task.params.max_duration: 必须是 0~604800 的秒数"
|
||
f"（当前 {dur!r}）")
|
||
dur = 0
|
||
params["max_duration"] = int(dur)
|
||
# 步骤里的 id 由编辑器生成，落库前不需要（保留会让前端 id 重复）
|
||
if isinstance(params.get("steps"), list):
|
||
_strip_ids(params["steps"])
|
||
task["params"] = params
|
||
return task
|
||
|
||
|
||
def _strip_ids(steps):
|
||
for s in steps:
|
||
if isinstance(s, dict):
|
||
s.pop("id", None)
|
||
p = s.get("params")
|
||
if isinstance(p, dict):
|
||
for key in ("children", "then", "else"):
|
||
if isinstance(p.get(key), list):
|
||
_strip_ids(p[key])
|
||
|
||
|
||
def _check_target(task, groups, pool, errors, warnings):
|
||
target = task.get("target") or {}
|
||
mode = target.get("mode")
|
||
if mode not in TARGET_MODES:
|
||
errors.append(f"task.target.mode: 必须是 {'/'.join(TARGET_MODES)} 之一"
|
||
f"（当前 {mode!r}）")
|
||
return
|
||
if mode == "group":
|
||
gname = (target.get("group_name") or "").strip()
|
||
if not gname:
|
||
errors.append("task.target.group_name: mode=group 时必须给分组名")
|
||
elif groups is not None and gname not in groups:
|
||
errors.append(f"task.target.group_name: 分组 {gname!r} 不存在"
|
||
f"（现有：{'、'.join(groups) or '无'}）")
|
||
elif mode == "serial":
|
||
serial = (target.get("serial") or "").strip()
|
||
if not serial:
|
||
errors.append("task.target.serial: mode=serial 时必须给设备 serial")
|
||
elif pool is not None and serial not in pool:
|
||
warnings.append(f"task.target.serial: 设备 {serial} 不在设备池里，"
|
||
"任务运行时会被跳过")
|
||
|
||
|
||
def _check_schedule(task, errors, warnings):
|
||
sch = task.get("schedule") or {}
|
||
mode = sch.get("mode")
|
||
if mode not in SCHEDULE_MODES:
|
||
errors.append(f"task.schedule.mode: 必须是 {'/'.join(SCHEDULE_MODES)} 之一"
|
||
f"（当前 {mode!r}）")
|
||
return
|
||
if mode == "once":
|
||
return
|
||
cron = sch.get("cron")
|
||
if not isinstance(cron, str) or not _cron_ok(cron):
|
||
errors.append(f"task.schedule.cron: 不是合法的 5 段 crontab（分 时 日 月 周）"
|
||
f"，当前 {cron!r}")
|
||
if mode == "cron_stop":
|
||
stop = sch.get("stop_cron")
|
||
if not isinstance(stop, str) or not _cron_ok(stop):
|
||
errors.append("task.schedule.stop_cron: 定时停止必须给合法的 5 段 "
|
||
f"crontab，当前 {stop!r}")
|
||
win = sch.get("window")
|
||
if isinstance(win, dict):
|
||
start, end = win.get("start"), win.get("end")
|
||
if not (_HHMM_RE.match(str(start or "")) and _HHMM_RE.match(str(end or ""))):
|
||
warnings.append("task.schedule.window: 时间窗应形如 {\"start\":\"09:00\","
|
||
"\"end\":\"18:00\"}，格式不对会被忽略")
|
||
elif start == end:
|
||
warnings.append("task.schedule.window: 起止相同等于不限制")
|
||
dur = (task.get("params") or {}).get("max_duration", 0)
|
||
if mode == "cron_stop" and not dur:
|
||
warnings.append("task.params.max_duration 为 0：本任务靠定时停止结束，"
|
||
"建议同时给一个运行时长上限（秒）兜底")
|
||
|
||
|
||
def _cron_ok(expr):
|
||
"""5 段 crontab 且能被 APScheduler 解析（任务是**静默**不注册的，必须提前拦）。"""
|
||
if len(str(expr).split()) != 5:
|
||
return False
|
||
try:
|
||
from apscheduler.triggers.cron import CronTrigger
|
||
CronTrigger.from_crontab(expr)
|
||
return True
|
||
except Exception:
|
||
return False
|