"""步骤默认值:新建步骤时预填的参数(后台「任务 → 动作配置」页可改)。 存在 `app_meta.step_defaults` 一个键里(**不建表** → 不碰备份覆盖红线), 与 `notify_webhooks` 同一套做法:**出厂默认 + 用户覆盖**,读的时候合并。 为什么是"出厂 + 覆盖"而不是只存用户值:以后调整某个步骤的出厂默认时, **没配过的字段**会跟着变好,**配过的字段**不动——这符合直觉。 字段白名单 + 范围校验都在这里(`SPEC`):写进库的东西必须能校验,否则手改 JSON 塞个奇怪值进去,之后新建步骤就把它带进任务里了。 ⚠️ 这里只管**编辑器预填**;`tasks/generic/task.py` 的 `STEP_TYPES` 仍是执行侧 的默认值来源(AI/MCP 造步骤也用它)。两处都改了要一起同步。 """ import json # 出厂默认(前端「动作配置」页的初始内容,也是没配过时的回退) FACTORY = { "swipe": {"direction": "up", "duration_min": 0.25, "duration_max": 0.5, "distance_ratio": 0.6, "jitter": 0.15, "humanize": True}, "click": {"wait_timeout": 2}, "long_click": {"duration": 1.0, "wait_timeout": 2}, "wait": {"min": 1.0, "max": 3.0}, "key_event": {"key": "back"}, "input_text": {"mode": "random", "texts": "你好\n有趣\n支持", "fixed_text": "", "clear_first": True}, } # 字段规格:("num", 最小, 最大) / ("bool",) / ("choice", 可选值) / ("text", 最大长度) SPEC = { "swipe": { "direction": ("choice", ("up", "down", "left", "right")), "duration_min": ("num", 0.05, 5.0), "duration_max": ("num", 0.05, 5.0), "distance_ratio": ("num", 0.1, 0.9), "jitter": ("num", 0.0, 0.4), "humanize": ("bool",), }, "click": {"wait_timeout": ("num", 0.0, 30.0)}, "long_click": {"duration": ("num", 0.2, 10.0), "wait_timeout": ("num", 0.0, 30.0)}, "wait": {"min": ("num", 0.1, 600.0), "max": ("num", 0.1, 600.0)}, "key_event": {"key": ("text", 30)}, "input_text": {"mode": ("choice", ("random", "fixed")), "texts": ("text", 2000), "fixed_text": ("text", 200), "clear_first": ("bool",)}, } # 与 notifier 的键同住 app_meta(`key` 是 MySQL 保留字,读写只走 db_config.meta_get/set) META_KEY = "step_defaults" def factory(): """出厂默认的深拷贝(别让调用方改到常量)。""" return json.loads(json.dumps(FACTORY, ensure_ascii=False)) def _clean_one(stype, raw): """校验并清洗单个步骤类型的默认值;非法字段**丢弃**(不让脏值进库)。""" spec = SPEC.get(stype) if not spec or not isinstance(raw, dict): return None out = {} for k, v in raw.items(): rule = spec.get(k) if not rule: continue kind = rule[0] try: if kind == "bool": out[k] = bool(v) elif kind == "num": n = float(v) lo, hi = rule[1], rule[2] out[k] = max(lo, min(hi, n)) elif kind == "choice": s = str(v) if s in rule[1]: out[k] = s elif kind == "text": out[k] = str(v)[:rule[1]] except (TypeError, ValueError): continue return out or None def clean(raw): """清洗整份配置(只保留白名单里的步骤类型与字段)。""" if not isinstance(raw, dict): return {} out = {} for stype in SPEC: got = _clean_one(stype, raw.get(stype)) if got: out[stype] = got return out def merged(user): """出厂默认 + 用户覆盖(按字段合并)。""" out = factory() for stype, vals in (clean(user) or {}).items(): out.setdefault(stype, {}).update(vals) return out def get(): """当前生效的默认值(读库,必须在 app context 内调)。""" from core.db_config import meta_get raw = meta_get(META_KEY) or "" if not raw: return factory() try: return merged(json.loads(raw)) except (ValueError, TypeError): return factory() def save(raw): """保存(只落用户显式给的字段)。返回 (ok, msg)。 与出厂默认**相同的字段不入库**:这样以后调出厂值,这些字段能跟着更新 ——库里只留"用户真的改过"的部分。 """ from core.db_config import meta_set cleaned = clean(raw) fac = FACTORY diff = {} for stype, vals in cleaned.items(): keep = {k: v for k, v in vals.items() if not (stype in fac and k in fac[stype] and fac[stype][k] == v)} if keep: diff[stype] = keep try: meta_set(META_KEY, json.dumps(diff, ensure_ascii=False)) except Exception as e: return False, f"保存失败: {e}" return True, "已保存" def as_payload(): """给前端的完整载荷:生效值 + 出厂值 + 字段规格(前端据此画表单与范围)。""" spec = {st: {k: (list(r[1]) if r[0] == "choice" else (list(r[1:]) if r[0] == "num" else None)) for k, r in fields.items()} for st, fields in SPEC.items()} kinds = {st: {k: r[0] for k, r in fields.items()} for st, fields in SPEC.items()} return {"defaults": get(), "factory": factory(), "spec": spec, "kinds": kinds}