auto_control/doc/TASK_DEV.md

任务与步骤开发（TASK_DEV）

适用读者：写任务、编排步骤、加步骤类型的开发者。 相关文档：ARCHITECTURE.md §5（调度链路）、API.md §5（任务接口）、AI_CONSOLE.md（AI 也会产步骤）。 现状：平台只有 generic_steps 一种任务类型，绝大多数 App 操作直接用步骤编辑器编排即可。

---

目录

• 1. 核心概念
• 2. 通用步骤任务 generic_steps
• 3. 18 种步骤全表
• 4. 容器步骤与公共参数
• 5. 选择器与元素定位
• 6. 自定义动作
• 7. 步骤编辑器（前端）
• 8. 新增任务类型
• 9. Action 框架（App 专属操作）
• 10. 测试与排查
• 11. 开发检查清单

---

1. 核心概念

1.1 TaskType — 任务类型

一个 TaskType = Task 子类 + Worker 子类 + DEFAULT_PARAMS，用 @register_task 注册进全局 _TASK_TYPES（key 为 task_type 字符串）。

# tasks/base.py
_TASK_TYPES = {}

def register_task(task_cls):
    """任务类型注册装饰器（无需传 name，用 task_cls.task_type）"""
    _TASK_TYPES[task_cls.task_type] = task_cls
    return task_cls

def get_task_class(task_type):
    return _TASK_TYPES.get(task_type)     # 未注册返回 None

注册位置：tasks/__init__.py 只做两件事——from .base import ... 聚合导出，以及 import 各任务包子包触发注册：

# tasks/__init__.py
from .base import BaseTask, register_task, list_task_types, get_task_class
from .generic import task  # 触发 @register_task（当前唯一任务类型）

前端「新建任务」的任务类型下拉来自 GET /api/task_types。新增类型要重启 Web 才生效。

未知任务类型的行为：调度时打 error 返回；POST /api/jobs/<id>/run 直接返回明确错误；启动加载时会把库里残留的已删类型任务列进日志告警（只告警，不自动删用户数据）。

1.2 TaskJob — 任务计划

字段	说明
task_type	任务类型（不传默认 generic_steps）
target	{"mode":"all"|"group"|"serial", "group_name":…, "serial":…}，默认 {"mode":"all"}
params	任务参数，与 DEFAULT_PARAMS 深合并；含两个隐藏开关：skip_offline（默认 true，serial/group 模式先跳过离线设备）、preempt（默认 false，是否抢占正在跑别的任务的设备）
schedule	{"mode":"once"|"cron"|"cron_stop", "cron":…, "stop_cron":…, "window":{…}}
retry	{"max_attempts":1, "delay":60}
enabled	是否参与调度

后端不做 params 校验：只校验 name 非空 + task_type 已注册，其余原样盲存——步骤编辑器是参数正确性的唯一关卡，保存前务必用编辑器的 validate / "测试此步骤" 自校验。

1.3 Worker — 单设备执行

每台设备一个 BaseWorker 线程（core/device_worker.py）。基类已封装：设备获取（IP:5555 直连 / USB 走远程 adb server，try/finally 保证释放）、u2 连接（30s 超时）、状态上报、异常分类（设备离线不重试）、self.stopped() 停止信号、心跳看门狗（120s）。子类只实现 run_task(d)。

可用的基类能力：

方法	用途
self.set_action(text)	上报"当前动作"（同时刷新心跳）
self.set_progress(done=, total=, unit=, action_counts=, elapsed=)	上报进度（前端统一渲染进度条 + 徽章）
self.heartbeat()	长循环内手动刷新心跳
self.stopped() / self.is_time_up()	检查停止信号 / 任务时长上限
self.serial / self.params	当前设备 / 合并后的参数
self._start_timer() / self.elapsed()	计时

---

2. 通用步骤任务 generic_steps

task_type="generic_steps"（tasks/generic/task.py）：把任意 App 操作编排成"步骤链"。前端步骤编辑器拖拽节点 → 存为 params.steps（JSON 数组）→ worker 按顺序执行。

{
  "max_duration": 0,
  "steps": [
    {"id": "step_1", "type": "open_app", "label": "打开抖音",
     "params": {"package": "com.ss.android.ugc.aweme", "wait_home": true, "home_feature": "拍摄"}}
  ]
}

• 任务级参数：DEFAULT_PARAMS = {"max_duration": 0}（0 = 不限时）
• 没有默认步骤：steps 只能由编辑器产出；为空时 worker 立即报错「通用步骤任务没有可执行步骤：请在「任务」页编辑该任务并添加步骤」（设备「最近错误」可见），不会静默空跑
• 顶层 steps 只顺序执行一次——需要重复的动作必须显式放进 loop 步骤
• 每步 schema：{id, type, label, params}；id 由前端生成保证唯一（导出为自定义动作时会剥掉 id）

进度上报：done = 累计执行的非容器步骤数（loop/group/if_el 不计入，避免噪音）、total=0、unit="操作"，前端显示"已执行 N 次操作"。

选择器健康跟踪：某选择器连续未命中达阈值（10 次）会记 last_warning（"选择器连续 N 次未命中"），提示 App 改版导致选择器失效。

---

3. 18 种步骤全表

参数与默认值以 tasks/generic/task.py 为准；前端 STEP_LIB（static/admin/editor.js）负责在编辑器里呈现这些字段。

#	type	名称	必填 params	可选 params（默认）	子步骤	语义要点
1	open_app	打开 App	package	wait_home(False)、home_feature("")	—	app_start(package, wait=True)；wait_home=true 时以 descriptionContains=home_feature 作为"到首页"判定，超时只提示不失败。缺 package 只告警跳过
2	stop_app	结束 App	package	—	—	app_stop（am force-stop，下次打开是冷启动）
3	screen_on	亮屏	—	—	—	d.unlock()（息屏时唤醒并滑动解锁）
4	screen_off	息屏	—	—	—	d.screen_off()
5	keep_screen	保持亮屏	mode("on")	—	—	mode="off" → 恢复自动息屏；否则 svc power stayon true（充电时常亮，适合长任务）
6	key_event	按键	key("back")	—	—	d.press(key)；值直传，无白名单校验
7	swipe	滑动	direction("up")	duration_min(0.25)、duration_max(0.50)	—	时长取随机值；up = 中轴从 0.8h → 0.2h（down 反向，left/right 同理）；非法 direction 静默不滑
8	swipe_until	滑动直到元素	selector_type、selector_value	direction("up")、max_swipes(8)、click_when_found(True)	—	每轮滑动后 sleep 0.5s；命中即（可选）点击并返回 True；只有 direction="down" 才是向下滑，其余按上滑
9	click	点击元素	selector_type、selector_value	wait_timeout(2)	—	xpath 用 wait(timeout)，其它选择器用 exists(timeout)；wait_timeout=0 为立即判断。空选择器只告警
10	click_xy	点击坐标	—	x(50)、y(50)	—	屏幕百分比（0-100，会夹取），50/50 = 屏幕中心。无选择器时的兜底，最脆的方式
11	long_click	长按元素	selector_type、selector_value	duration(1.0)、wait_timeout(2)	—	先等元素出现再 long_click
12	wait_el	等待元素	selector_type、selector_value	timeout(10)	—	等元素出现（条件等待，优于固定 wait）
13	input_text	输入文字	—	mode("random")、texts("你好\n有趣\n支持")、fixed_text("")、clear_first(True)	—	mode="fixed" 用 fixed_text，否则从 texts 按行随机选一条；只负责输入，不负责定位输入框（要先 click 输入框）
14	clipboard	剪贴板注入	text	paste(True)	—	走 ClipInject 通道（绕开 Android 10+ 后台写剪贴板限制）；paste=true 时再触发一次粘贴
15	wait	等待	—	min(1.0)、max(3.0)	—	随机时长；分片 sleep（每 ≤0.5s 检查停止/超时），可被抢占打断
16	loop	循环块	—	loop_mode("rounds")、max_iterations(10)、loop_duration(600)	children	见 §4.1
17	group	动作组	—	—	children	子步骤按序执行一次（不循环）；自定义动作拖入画布就是展开成 group
18	if_el	条件判断	selector_type、selector_value、timeout(3)	ocr_click(False)	then / else	见 §4.2

---

4. 容器步骤与公共参数

4.1 loop 三种模式

loop_mode	行为
rounds（默认）	跑 max_iterations 轮
forever	一直循环，直到任务被外部停止或达到 max_duration（配合调度"定时启动+停止"或任务时长上限使用）
time	跑满 loop_duration 秒（<=0 时跳过）

未知 loop_mode 会按 rounds 处理（不报错）。

4.2 if_el 条件判断

• 先做元素判断：命中 → 执行 then 分支；未命中 → 执行 else 分支（分支都可继续嵌套）
• selector_type="ocr" 时改用 OCR：截屏 → find_on_screen(img, selector_value) 子串匹配；命中且 ocr_click=true 会点击命中位置。
  • OCR 模式下 timeout 不生效（单次截图即判定）
  • OCR 不可用（未装 rapidocr_onnxruntime）时该步返回 None（不判定）
  • 命中长文本时点击位置按关键词在文本中的比例估算（避免点到整块中心偏离）

4.3 公共参数

参数	说明
probability	0-100，默认 100。小于 100 时按概率决定本次是否执行该步（如 30 ≈ 隔几次才触发一次）；未触发会打日志"概率 X% 未触发，跳过"

4.4 嵌套与容错

• 嵌套深度上限 5 层：loop.children / group.children / if_el.then/else 递归超过 5 层会整段跳过并告警
• 每步执行前检查 stopped() / is_time_up()
• 静默跳过：未知 type、缺必填参数（如 click 没选择器、open_app 没包名）只打 WARNING，任务照常"成功"结束。排查"任务成功了但什么都没做"时先看 logs/task.log
• 单个步骤内部异常被捕获并记 error，不中断后续步骤

---

5. 选择器与元素定位

5.1 selector_type 支持的值

值	匹配语义	谁支持
xpath	d.xpath(value)（存在性判断必须用 wait()）	所有带选择器的步骤
description	content-desc 精确匹配	同上
descriptionContains	content-desc 模糊匹配（适配文案版本差异）	同上
text	文本精确匹配	同上
resourceId	resource-id 精确匹配（要带包名前缀 com.xxx:id/…）	同上
className	类名匹配（如 android.widget.EditText）	同上
ocr	截屏 OCR 文字子串匹配	仅 if_el

选择器值是直接透传给 uiautomator2 的 kwarg，没有服务端校验——写错了只会"找不到元素"。

5.2 定位优先级（重要）

① 目标有可见文字 → 用 text / descriptionContains（最稳）
② 有稳定 resource-id → 用 resourceId 或 xpath
③ 文案会变但有结构 → 用 descriptionContains 或结构 XPath
④ 纯图形、树里找不到 → 才用 click_xy（坐标，最脆）

坐标是最脆的：分辨率/布局一变就失配，且抓取时的坐标在其他设备上未必有效。抓取元素请在任务实际运行的那台设备的同一界面上做。

5.3 XPath 序号语义（踩过的坑）

写法	含义
//*[@resource-id="x"][2]	「在其父节点中排第 2 的匹配」——不是"第 2 个匹配"
(//*[@resource-id="x"])[2]	「第 2 个匹配」← 抓取器生成的形式

历史实现写了前者，导致同 id 多实例（如底部导航 4 个 tab）时 [2..n] 全部失配。现在：

• 执行器 tasks/generic/task.py 的 _norm_legacy_xpath 会自动纠正旧任务里的 //*[@x][k] 写法（仅前缀，类名/位置步进不受影响）
• 抓取器生成 (…)[k] 前先做语义消歧（见 §5.4），序号型只是最后的退路

⚠️ 序号型选择器仍然脆弱：它依赖"抓取那一刻该属性有 ≥k 个实例"。界面不同（如 App 还在闪屏页）就会失配——优先选带文字的元素，抓取器会自动产出语义选择器。

5.4 抓取器给的建议选择器

GET /api/uiauto/elements、GET /api/uiauto/snapshot 返回的元素都带 suggested，生成优先级：

1. 有 resource-id → //*[@resource-id="v"]；无 id 但有 text/content-desc 同理
2. 该属性值全树有多个时，先用第二个属性把目标单独圈出来（语义选择器，标 semantic:true + via）： //*[@resource-id="v" and @text="我"]。抖音底部导航同 id 的几个 tab 靠这个解决—— 灰度版 tab 数量会变（4 个 ↔ 3 个），序号必然错位，文字不会
3. 两个属性组合仍分不开（列表里同 id 同文字）→ 才退回序号 (…)[k]（标 indexed:true + occ/total，前端打黄标 ⚠ 序号）
4. 都没有属性、但有"最近的有属性祖先" → {祖先}/*[@class="android.widget.ImageView"][n]
5. 连祖先都没有 → //hierarchy/*[i]/*[j]（按子节点位置逐层，标 broad:true，脆弱、前端会提示）
6. 类名也是空 → 标 invalid:true（提示无法生成可靠选择器）

两个必须记住的 XPath 坑（都实测踩过）：

• dump 的 XML 标签名一律是 <node>，class 在 @class 属性上。所以结构步进只能写 *[@class="…"] 或位置 *[i]；写成 //FrameLayout[1] 这种"类名当标签名"的路径永远零命中 （2026-09-13 修：248 个元素里曾有 26 条死选择器）
• 同级节点的 index 属性会重复（状态栏/内容区/导航栏三个兄弟的 index 全是 0）， 兜底结构路径要按位置 *[i+1] 定位，不能按 @index

5.5 抓取与验证（编辑器里的两个按钮）

• 「▶ 点一下」：按元素 bounds 中心在设备上真点一次（POST /api/screen/tap，snap=1 自动吸附到可点元素中心），返回吸附结果并刷新截图——确认位置是否可达
• 「✓ 测选择器」：用将填入的选择器真跑一次 click（POST /api/steps/test）→ 返回 命中 / 未找到 / 已执行——确认回填的选择器在真实界面能命中
• 「测试此步骤」：在编辑器里对任意步骤做真机试执行（不占用设备池、不影响运行中的任务）

5.6 u2 辅助函数（core/u2_helper.py）

函数	用途
ensure_app_running(d, package, max_restart=3, home_check=None)	确保 App 在前台，必要时重启
wait_for_app_home(d, package, home_check, timeout=40)	等 App 首页就绪。注意：主页 Activity 名可能是 SplashActivity，不能靠 Activity 名判断，要用界面特征元素
random_sleep(min_s, max_s)	随机间隔（模拟人类）
safe_click(el, timeout=1) / safe_set_text(el, text, timeout=1)	存在才点/才输，不抛异常
find_and_click(d, timeout=1, **selectors)	找元素并点击
random_swipe_up(d, ...)	按比例上滑，时长可随机

---

6. 自定义动作

自定义动作（custom_action 表）把一串常用步骤打包成可复用动作，供任何 generic_steps 任务拖入：

• 编辑器里 ☑ 多选步骤 → 「打包选中步骤」 即可生成
• 拖入画布时展开为 group 节点（执行语义等同顺序执行一次）
• 「任务 → 自定义动作」子分栏可查看/编辑/删除

想做"引用型"节点（改一处、处处生效）目前不支持，见 backlog/TODO.md。

---

7. 步骤编辑器（前端）

static/admin/editor.js 的 _stepEditor 单例，能力：

能力	说明
操作库拖拽	左侧分类库（interact / screen / flow），拖到画布；支持跨层级嵌套（循环套循环）、拖拽排序
参数表单	按步骤类型渲染对应字段（选择器行、时长、概率、循环模式等）
条件分支	if_el 的 then/else 两个分支区；OCR 模式下多出"命中后点击"复选框
多选与打包	☑ 多选 → 打包成自定义动作
元素抓取	「抓取元素」打开抓取弹窗（选设备 → 截图 + 元素树 → 点选回填），见 §5.5
测试	「测试此步骤」真机试执行
草稿自动保存	编辑中的步骤会存草稿，避免误关丢失

后端 STEP_TYPES 与前端 STEP_LIB 是两份定义，新增/修改步骤类型必须同时改，否则会出现"编辑器里有、后端不认识"（静默跳过）或反之。

---

8. 新增任务类型

平台当前只有 generic_steps。先确认是否真的需要：绝大多数 App 操作直接用步骤编辑器编排即可，新建任务类型的成本与维护负担都不低。

8.1 目录结构

tasks/<app>/
├── __init__.py        # from . import task  触发注册
├── task.py            # DEFAULT_PARAMS + Worker + Task + @register_task
└── actions/           # 该任务专属操作（可选）
    ├── __init__.py    # 先 from .base import ACTIONS，再 from . import xxx
    ├── base.py        # ACTIONS = create_action_registry()
    └── like.py        # @register_action(ACTIONS) LikeAction

参考实现：tasks/generic/task.py（步骤引擎）。

8.2 骨架

# tasks/<app>/task.py
from tasks.base import BaseTask, register_task
from core.device_worker import BaseWorker, _update_status
from core.logger import get_logger

log = get_logger("task.<app>")          # 写入 logs/task.log

DEFAULT_PARAMS = {
    "count": 20,                        # 任务自身参数放这里，不放 config.py
    "max_duration": 0,                  # 0 = 不限时（时长上限由基类实现）
}

class MyWorker(BaseWorker):
    """单设备执行体：只实现 run_task(d)，d 是已连好的 u2.Device。"""
    def __init__(self, serial, params=None):
        super().__init__(serial, params)
        p = {**DEFAULT_PARAMS, **(params or {})}
        self.count = int(p.get("count", 20))

    def run_task(self, d):
        self._start_timer()
        self.set_progress(done=0, total=self.count, unit="次")
        for i in range(self.count):
            if self.stopped() or self.is_time_up():     # 必须检查，否则无法停止/超时
                break
            self.set_action(f"第 {i+1}/{self.count} 次")  # 上报当前动作（同时刷新心跳）
            # ... 业务操作（用 d 做点击/滑动/输入）...
            self.set_progress(done=i + 1, total=self.count, unit="次",
                              action_counts={}, elapsed=self.elapsed())

@register_task
class MyTask(BaseTask):
    task_type = "my_task"
    name = "我的任务"
    description = "…"
    default_params = dict(DEFAULT_PARAMS)

    @classmethod
    def list_action_types(cls):
        return []                       # 有专属操作时返回其清单

    @classmethod
    def get_action_class(cls, action_type):
        return None

    def create_worker(self, serial, params):
        merged = {**DEFAULT_PARAMS, **(params or {})}
        return MyWorker(serial, params=merged)

# tasks/<app>/__init__.py
from . import task  # noqa: F401  触发 @register_task 注册

# tasks/__init__.py —— 加一行
from .base import BaseTask, register_task, list_task_types, get_task_class
from .generic import task
from .myapp import task      # ← 新增

重启 Web 后，「新建任务」的任务类型下拉会出现新类型。

8.3 签名约定

• BaseWorker.__init__(self, serial, params=None, daemon=True)
• Task.create_worker(self, serial, params) —— 不要带 stf_client / stf 形参（STF 已摘除，历史签名已清理）

---

9. Action 框架（App 专属操作）

core/actions/ 提供跨任务的通用操作抽象：BaseAction 接口、概率触发、注册机制；每个 App 有自己的注册表（create_action_registry()），互不污染。

# tasks/<app>/actions/like.py（示例）
from core.actions import BaseAction, register_action, should_trigger
from . import ACTIONS          # 本 App 的注册表

@register_action(ACTIONS)
class LikeAction(BaseAction):
    action_type = "like"
    name = "点赞"
    default_params = {"rate": 0.3}

    def execute(self, ctx):
        if not should_trigger(self.params):     # 按概率决定是否本次触发
            return False
        ...                                     # ctx 提供 device/worker/进度等
        return True

循环导入坑：actions/__init__.py 必须先 from .base import ACTIONS，再 from . import like。

当前 generic_steps 不使用 Action 框架（它直接分发到 _exec_<type>）；Action 框架留给"专属任务类型"使用。

---

10. 测试与排查

场景	做法
试执行单个步骤	编辑器「测试此步骤」（POST /api/steps/test）——真跑一次，返回 命中 / 未找到 / 已执行
试执行整个任务	建一条只含该步骤的临时任务，目标选一台设备，enabled=false +「执行任务」
看任务日志	logs/task.log（业务）、logs/core.log（调度/adb）
看设备错误	监控页「最近错误」列；失败原因会带真实错误（不再只有"重试N次失败"）
步骤没生效	检查是否有 WARNING：未知 type / 缺必填参数只告警不报错
选择器失效	看设备 last_warning（"选择器连续 N 次未命中"）→ 重新抓取

没有步骤的任务：执行时会明确报错，不会静默空跑。

---

11. 开发检查清单

新增/修改任务与步骤时，逐条打勾：

• 后端 STEP_TYPES 与前端 STEP_LIB 一致（type 名、字段、默认值、分类）
• 新步骤在 _exec_<type> 里实现了，且缺参数时有明确日志
• 长循环里调用了 self.stopped() / self.is_time_up() / self.heartbeat()
• 用文字/id 定位，没有滥用坐标
• 抓元素是在目标任务运行的那台设备上、停在同一界面时做的
• 新增任务类型已注册进 tasks/__init__.py
• 文档同步：TASK_DEV.md（步骤表）、API.md（若接口有变）、必要时 AI_CONSOLE.md（AI 沉淀动作的白名单）
• 临时测试数据已清理（任务/分组/自定义动作）