butubb
7ce4ae9016
一、账号台账（新表 device_account，schema v8） - core/ledger.py：CRUD、Excel 粘贴解析（Tab 分隔 / 表头乱序缺列 / 是·否布尔 / 逐行留痕）、 按范围取号（本机台账 / 全部 / 按设备分组）、设备端账号块、dry_run 预览 - 新「账号」顶级 Tab（权限 devices）：列表 + 搜索 + 设备筛选 + 增删改 + 粘贴导入（默认先预览再写，逐行显示 新增/覆盖/跳过/失败 + 留痕） - 设备池加「账号 N」列（0 个显示"未登记"），点开看该设备账号明细 - 任务「条件判断」新增 cmp_source / cmp_group：候选值可直接来自台账， 与手填值**合并（OR）**；取不到号时回落手填值，并把原因写进步骤明细与日志 （否则"永远走 else 分支"而任务照样显示成功，最难查） - 设备端契约：身份页多推 accounts_b64（base64 JSON；默认不含手机号； 超预算按整条丢且绝不算字节切），见 doc/DEVICE_AGENT.md §5.2.1 - 备份/文档红线：TABLE_LABELS 加「账号台账」；DATA_MODEL/API/ARCHITECTURE/DEPLOY/ TASK_DEV/DEVICE_AGENT/DEVELOPMENT/README 同步；顺手补上 DATA_MODEL 漏列的 done_mark ⚠ 台账里的抖音号是**纯号**，只当"比对用的候选值"：绝不能拿去填「去重」的身份元素 （身份是元素原文逐字算 key，格式不同会让去重静默失效）。代码与文档都写明了。 二、剪贴板注入通道（修：平台还在调早期的独立 APK） - 改为按顺序尝试：设备端 Agent（com.example.deviceagent/.ClipActivity）→ 旧版独立 ClipInject 兜底；两个都没有时报"需要设备端 Agent"（不再只说 ClipInject） - 读回验证改成**轮询到 3 秒**：透明 Activity 要等窗口拿到焦点才写， 原来只睡 0.5s 会读到上一次的内容 → 误报"写入可能被拒"（实测内容已写入却回失败）
579 lines
28 KiB
Python
579 lines
28 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_el 用「去重」条件时不需要 selector_value，
|
||
# 它的身份元素在 ident_value 里，按自己的规则校验）
|
||
if stype in NEED_SELECTOR and not (stype == "if_el"
|
||
and params.get("selector_type") == "dedup"):
|
||
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()
|
||
cmp_src = (params.get("cmp_source") or "").strip()
|
||
if cmp_src not in ("", "manual", "device", "all", "group"):
|
||
errors.append(f"{here}.params.cmp_source: 不支持的候选值来源 {cmp_src!r}"
|
||
f"（可用：空/manual/device/all/group）")
|
||
if cmp_src == "group" and not (params.get("cmp_group") or "").strip():
|
||
warnings.append(f"{here}: 候选值来源是「按设备分组」，但没填分组名 —— 取不到号")
|
||
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() \
|
||
and cmp_src not in ("device", "all", "group"):
|
||
# 候选值来自台账时允许手填留空（台账取号就够）；
|
||
# 其余情况仍要求填值，否则永远不命中
|
||
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 会被忽略")
|
||
# 去重条件（见 tasks/generic/task.py 的 selector_type="dedup"）：
|
||
# 不用 selector_value，身份元素填在 ident_value
|
||
if (params.get("selector_type") or "") == "dedup":
|
||
iv = (params.get("ident_value") or "").strip()
|
||
if not iv:
|
||
warnings.append(f"{here}: 去重条件没填身份元素，将按「设备」当身份"
|
||
"（一号一机时没问题；一台机器多个号时要填账号那个元素）")
|
||
elif (params.get("ident_type") or "text") == "xpath" \
|
||
and not (iv.startswith("//") or iv.startswith("(//")):
|
||
errors.append(f"{here}.params.ident_value: 身份元素的类型是 xpath，"
|
||
f"取值应以 // 或 (// 开头（当前 {iv[:40]!r}）")
|
||
|
||
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 and not (sel_type == "dedup" and stype == "if_el"):
|
||
errors.append(f"{here}.params.selector_type: 不支持 {sel_type!r}"
|
||
f"（可用：{'、'.join(stype_ok + ('dedup',))}）")
|
||
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
|