butubb
68c16a5fde
用户要求："把你每一步操作都放出来让我自己写，你都封在一起我也不知道切号/认号的流程"
→ 把两个打包步骤的内部动作，拆成**画布上拖得出来的普通步骤**（能拆是因为补了下面两个零件）。
① 选择器值里的**变量**（tasks/generic/task.py `_vars_sub`）
`{本次要发的名字}`（循环块设的）/ `{本机账号名字}`（读名字读到的），中文别名与变量名都认。
用在 click / long_click / wait_el / swipe_until / expect_el / read_text / read_account_name /
switch_account 的选择器值，以及 if_el 的选择器与手填候选值。
· 没写花括号的步骤一个字不改（老任务零影响）
· 变量为空 → **换成空串 + 记警告**：宁可"选择器变空、找不到元素"明着报，
也不留 `{…}` 字面量去点到歪的地方
· 不做通用表达式（只认表里那几个名字）
② 「点击元素」的选择器类型下拉补上 **textContains**（文本包含）——
以前只能用打包步骤做到的"按名字点列表那一行"，现在自己就能拼：
`textContains` + 值 `{本次要发的名字}`
③ 新步骤 **read_text「读取文字」**：把元素文字读出来存进指定变量
（`var`=account_name 时顺手把昵称控件记给切号/推送核对复用；`clean`=按账号名清洗/只去空白；
`on_missing`=skip 默认只记一句 / fail 算失败）。之前**没有**"读取元素文字"这种步骤。
④ 「条件判断」的候选值来源加 **「变量」**（`cmp_source=var` + `cmp_var`）——
不然 read_text 读出来的东西没处用。空变量 = 候选为空 = 不命中（同"读不到名字"的规矩）。
⑤ fix(发布) 实测踩出来的两处（A06，截图/日志为证）：
· **剪辑页校验 12 → 45 秒**：点完「下一步」抖音在「正在导入」（一条 25 秒的视频要导 25 秒+），
导入期间工具栏不渲染，10~12 秒必然判"没到剪辑页"→ 这一条白发
· **切号前先回主界面**（`_goto_me`）：点不到「我」页签 = 不在主界面（剪辑页/发布页），
那种页面没有底部页签 → **强杀 + 冷启动**抖音再来。实测卡在剪辑页时，本轮剩下的号全部报废
· **切号后回读为空时点一次「我」再读**：换号会整体重启抖音、起来常落在首页（首页没有昵称控件），
读多少次都是空 —— 等完 wait_after、读两遍仍空才补这一下导航，不打扰它的重启过程
前端：编辑器调色板/字段面板认 read_text（含变量说明）、选择器类型加 textContains、
placeholder 提示 `{本次要发的名字}`、候选值来源加「变量」+ 变量名输入、
「发布计划」页就地编辑字段表同步（release.js）。
文档：TASK_DEV §3 步骤表 28→29 种 + §3.3「选择器里的变量」+ §3.4「自己拼切号+认号」；
API.md 骨架步数修正（18 → 4 步 / 含嵌套 23 步）。
测试：新增 scripts/tests/test_read_text.py（变量代入、空变量不许瞎点、读取文字存变量/清洗、
条件判断取变量、草稿校验），全套 8 个用例通过。
585 lines
28 KiB
Python
585 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", "expect_el",
|
||
"read_text")
|
||
# 需要包名的步骤类型
|
||
NEED_PACKAGE = ("open_app", "stop_app")
|
||
# 需要非空 children 的容器
|
||
NEED_CHILDREN = ("loop", "group", "foreach_account")
|
||
|
||
# 有副作用的动作关键词：命中的步骤在 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", "release", "account_target", "var"):
|
||
errors.append(f"{here}.params.cmp_source: 不支持的候选值来源 {cmp_src!r}"
|
||
f"（可用：空/manual/release/account_target/var）")
|
||
if cmp_src == "var" and not (params.get("cmp_var") or "").strip():
|
||
errors.append(f"{here}.params.cmp_var: 候选值来源选了「变量」就必须填变量名"
|
||
f"（默认 account_name —— 由「读取文字」写入）")
|
||
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 ("release", "account_target", "var"):
|
||
# 候选值来自发布计划 / 本次要发的名字 / 变量时，允许手填留空（候选不是手填的）；
|
||
# 其余情况仍要求填值，否则永远不命中
|
||
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":
|
||
# 文案来自「发布计划」（`text_source` 不是 manual）时 texts/fixed_text **都不参与输入** ——
|
||
# 别按"手填"的规则要求候选文案（发布任务骨架里那两步就是这么配的：
|
||
# mode 留默认、内容由计划给，靠 `release_title`/`release_body` 现取）
|
||
if str(params.get("text_source") or "manual").strip() == "manual":
|
||
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
|