"""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) 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