auto_control/doc/TASK_DEV.md

任务开发指南

面向 platform-tools（uiautomator2 + Flask 单页应用，本地 SQLite 设备池 + adb 直连（IP:5555 / USB 远程 server）多设备并发执行框架）的新开发者。描述架构、核心概念，并给出从 0 到 1 新增一个 app 任务所需的全部模板与规范。看完本文即可上手开发新任务。

平台曾在代码层依赖 OpenSTF，现已完全摘除（STF 占用/释放、remoteConnect 桥接均已退役），迁移背景见 doc/STF_REMOVAL.md。项目代码目录为 auto_control，文档/README 仍沿用旧名 platform-tools——项目是否正式更名待人工核实。

---

1. 架构总览

1.1 分层设计

平台按"配置 / 核心 / 任务 / 前端 / 数据 / 日志 / 工具"分层，职责清晰、互不交叉：

层	路径	职责
配置层	config.py	项目根配置：adb 路径 / web 端口 / 数据与备份目录 / USB 远程 adb（220）/ Tailscale / 设备发现 / .env 注入。STF/SSH 键已废弃，仅历史保留。不放任务参数（任务参数属于 tasks/）
核心层	core/	框架运行时：logger 日志、device_pool 设备池（本地清单 + adb 在线）、device_discovery 设备自动发现、models 数据模型、adb_helper adb 操作（全局锁）、device_worker Worker 基类 + STFDevice、task_manager 调度器、u2_helper / uiauto_helper / ocr / clipboard_helper、apk_manager 应用管理、system_backup 数据备份、tailscale_client、actions 全局 Action 基类
任务层	tasks/	每个 app 一个子包，自包含 task.py + actions/，互不依赖
前端层	templates/admin/ + static/admin/	单页应用（纯 HTML+CSS+JS，无框架）：监控 / 任务 / 日志 / 用户 / 工具 / AI 控制台 / 系统 7 个 Tab；工具页等按子分栏分组；JS 拆分为 static/admin/ 下的 base/list/monitor/editor/tasks/tools/apps/admin/agent/system
数据层	data/	SQLite 持久化：users.db（用户 / 设备分组 / 任务计划 / 自定义动作 / APK 文件 / 设备池 device / 待连接池 pending_device / app_meta / AI 会话与经验库 等表）
日志层	logs/	四类日志：core.log / task.log / web.log / action.log，10MB 滚动保留 5 份
文档层	doc/	项目文档
工具层	bin/adb/	adb 可执行文件
脚本层	scripts/	实用脚本

1.2 目录树

platform-tools/
├── config.py                  # 根配置（部署值从 .env 读，不放任务参数）
├── web_server.py              # Flask 入口（app 装配 + init_db + 启动调度/看门狗/设备发现）
├── web/                       # Web 蓝图包（路由按功能域拆分，web_server.py 只做装配）
│   ├── auth.py                #   登录/登出/CSRF/权限/页面路由（/、/wall）
│   ├── monitor.py             #   状态/运行控制/扫描前台/远程看屏
│   ├── tasks_api.py           #   任务计划/分组/自定义动作/元素抓取/步骤测试
│   ├── admin_api.py           #   用户管理/日志
│   ├── tools_api.py           #   adb 终端/剪贴板注入/应用版本
│   ├── devices_api.py         #   设备池管理/自动发现
│   ├── apks_api.py            #   应用管理
│   ├── tailscale_api.py       #   Tailscale 管理
│   ├── system_api.py          #   系统备份/导入恢复
│   └── agent_api.py           #   AI 控制台（会话/SSE/经验库）
├── core/                      # 核心程序层
│   ├── __init__.py
│   ├── logger.py              # 日志器（分文件、10MB 滚动）
│   ├── device_pool.py         # 设备池：SQLite devices 清单 + adb 在线（list_ready 供调度）
│   ├── device_discovery.py    # 设备自动发现（扫描 5555 → 待连接池，用户确认入池）
│   ├── models.py              # SQLAlchemy 模型（User/DeviceGroup/TaskJob/Device/PendingDevice/CustomAction/ApkFile）
│   ├── adb_helper.py          # adb 操作（全局锁串行化；支持 220 远程 server -H/-P）
│   ├── device_worker.py       # BaseWorker 基类 + STFDevice(acquire/release) + 心跳看门狗
│   ├── task_manager.py        # TaskManager 调度器 + 前台 App 扫描器
│   ├── u2_helper.py           # uiautomator2 通用操作（ensure_app_running/wait_for_app_home/random_sleep）
│   ├── uiauto_helper.py       # uiautodev 本地服务客户端（步骤编辑器"抓取元素"）
│   ├── ocr.py                 # 屏幕 OCR（RapidOCR；if_el 的 ocr 选择器用）
│   ├── clipboard_helper.py    # 剪贴板注入（ClipInject APK 通道）
│   ├── apk_manager.py         # APK 上传/解析/批量安装
│   ├── system_backup.py       # 数据备份导出/导入（重启生效）
│   ├── tailscale_client.py    # Tailscale API v2 客户端
│   ├── ssh_client.py          # SSH 客户端（仅历史手动运维 220 用）
│   └── actions/
│       ├── __init__.py        # create_action_registry / register_action / should_trigger
│       └── base.py            # BaseAction 全局基类
├── tasks/                     # 任务定义层
│   ├── __init__.py            # 聚合导出 + import 各任务包触发注册（from .douyin/.generic import task）
│   ├── base.py                # BaseTask 基类 + _TASK_TYPES + register_task/list_task_types/get_task_class
│   ├── douyin/                # 抖音养号（task_type=douyin_nurture，示例）
│   │   ├── __init__.py
│   │   ├── task.py            #   DEFAULT_PARAMS + Worker + Task + @register_task
│   │   └── actions/
│   │       ├── __init__.py    #   先 from .base import ACTIONS，再 from . import like
│   │       ├── base.py        #   ACTIONS = create_action_registry()
│   │       └── like.py        #   @register_action(ACTIONS) LikeAction
│   └── generic/               # 通用步骤任务（task_type=generic_steps，步骤编辑器编排）
│       ├── __init__.py
│       └── task.py            #   STEP_TYPES + Worker + Task（按 steps 顺序执行）+ test_step
├── templates/admin/
│   ├── monitor.html           # 单页应用（7 Tab + 页内子分栏，纯前端渲染）
│   ├── login.html             # 登录页
│   └── wall.html              # 监控大屏（只读轮播展示）
├── static/admin/              # 前端 JS（base/list/monitor/editor/tasks/tools/apps/admin/agent/system.js）+ custom.css
├── data/                      # 持久化数据
│   ├── users.db               # SQLite（用户/分组/任务/自定义动作/APK/设备池/待连接池/app_meta/AI 会话）
│   └── apks/                  # 上传的 APK 文件
├── logs/                      # 日志（10MB 滚动保留 5 份）
├── doc/                       # 文档
├── bin/adb/                   # adb 工具
├── mcp_server/                # MCP 服务端（AI 控制台 19 个 de_* 设备工具）
├── mcp_agent/                 # MCP Agent 链路（DeepSeek 多模态，AI 控制台后端）
└── scripts/                   # 实用脚本

1.3 数据流

┌──────────────┐  创建 Job    ┌─────────────┐  分发   ┌──────────────┐
│ 单页应用前端  │ ───────────► │ TaskManager │ ──────► │  Worker(设备) │
│ (monitor.html│              │互斥:_running│         └──────┬───────┘
│  fetch + DOM)│              └──────┬──────┘                │ STFDevice.acquire：
└──────────────┘                     │                       │  · IP:5555 → adb connect
       │ JSON API                    │ 心跳/状态             │  · USB → 220 远程 adb server
       ▼                             │                       ▼
┌──────────────┐             ┌──────────────┐        ┌────────────────┐
│  web_server  │             │  看门狗监控   │        │  设备(adb+u2)   │
│  (Flask API) │             └──────────────┘        │  u2 操作执行任务 │
└──────┬───────┘                                     └────────────────┘
       │
       │ device_pool.list_ready() = SQLite 清单 ∩ (本机 adb + 220 远程)在线
       ▼
┌──────────────┐
│ data/users.db│  SQLite 持久化（设备池/分组/任务/自定义动作/...）
└──────────────┘

调度与扫描的设备数据源是 core/device_pool（清单取自 SQLite device 表、在线状态取自 adb devices），不再查询 STF；同一 serial 同时只允许一个 worker，互斥由 TaskManager._running 保证（见 §2.2、§4.1）。

1.4 关键设计决策

• Flask + Flask-Login：已移除 Flask-Admin（自定义场景下过于受限），改用纯 Flask + 单页应用
• 单页应用：web_server.py 只提供 JSON API + 登录页，monitor.html 纯前端渲染（fetch + DOM 操作），无服务端模板依赖
• SQLite 持久化：替代旧 JSON 文件，支持用户/分组/任务的关系存储
• 多线程模式：Flask 启用 threaded=True 处理并发请求
• 设备状态缓存：get_status 带 5 秒缓存，worker 运行状态实时组装

---

2. 核心概念

2.1 TaskType — 任务类型

一个 TaskType 描述"做什么"（例如抖音养号、通用步骤），由 Task 子类 + Worker 子类 + DEFAULT_PARAMS 组成。每个 TaskType 注册到全局 _TASK_TYPES 字典，key 为任务类型字符串，value 为 Task 类。

注册位置：_TASK_TYPES 及 register_task / list_task_types / get_task_class 定义在 tasks/base.py；tasks/__init__.py 只做两件事——from .base import ... 聚合导出，以及 from .douyin/.generic import task 触发各任务包注册。当前已注册的 task_type 仅 douyin_nurture 与 generic_steps，前端"新建任务"下拉来自 GET /api/task_types。新增 task_type 需在 tasks/ 下建子包、用 @register_task 装饰，并在 tasks/__init__.py 加 from .xxx import task，然后重启 web 生效。

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

2.2 TaskJob — 任务计划

TaskJob 是"什么时候、在哪些设备上、用什么参数执行某个 TaskType"的持久化计划，存于 SQLite（data/users.db 的 task_job 表）。包含字段：

• task_type — 任务类型（对应 _TASK_TYPES 的 key；当前仅 douyin_nurture / generic_steps）
• target — 目标设备：{"mode": "all"|"group"|"serial", "group_name": "", "serial": ""}，默认 {"mode":"all"}
• params — 任务参数（与 DEFAULT_PARAMS 深合并）。含两个隐藏开关：
  • skip_offline（默认 true）— serial/group 模式先跳过本机 adb 不可达（离线）的设备
  • preempt（默认 false）— all 模式是否抢占正在运行其他任务的设备
• schedule — 调度策略：{"mode": "once"|"cron"|"cron_stop", "cron": "...", "stop_cron": "..."}；cron/cron_stop 可含 window 运行窗口
• retry — 重试策略：{"max_attempts": 1, "delay": 60}
• enabled — 是否启用

resolve_serials 语义（core/task_manager.py，按 target 展开实际要跑的设备，数据源为 core.device_pool）：

• mode="serial" — 只跑目标单台设备
• mode="group" — 取分组 serial 列表，并过滤到设备池内（不在池内的手动旧 IP 不参与调度）
• mode="all" — 不是字面"全部设备"：不开 preempt = device_pool.list_ready()（设备池清单 ∩ 在线，即当前可调度的在线空闲设备）；开 preempt = 取设备池全部在线设备（含正在跑其他任务的，执行时逐个抢占）
• 单台设备同时只允许一个 worker（TaskManager._running 互斥）；preempt 抢占结束后，调度器会自动重新拉起被抢占的原任务（重试循环 finally 归还设备）

/api/jobs 校验语义：新建/更新任务（POST/PUT /api/jobs）后端只校验 name 非空 + task_type 已注册，其余字段（target/params/schedule/retry）JSON 原样盲存、不做参数合法性校验——编辑器是唯一参数正确性关卡，保存 generic_steps 前务必用编辑器 validate/"测试此步骤"自校验。

2.3 DeviceGroup — 设备分组

设备分组存于 SQLite（device_group 表），便于按批次/项目/客户分组下发任务。一个 Job 可指定 target.mode="group"，调度器展开为组内设备序列号，并过滤到设备池内（见 §2.2 resolve_serials）。

2.4 Worker — 单设备执行线程

每个被调度的设备对应一个 Worker 实例，跑在独立线程中，继承 BaseWorker（core/device_worker.py）。Worker 负责一台设备的完整生命周期：STFDevice.acquire（serial 含冒号 → adb connect 直连 IP:5555；USB 无冒号 → 先本机 adb、不在则走 220 远程 adb server）→ 连接 u2 → setup → run_task → teardown → release（空操作，绝不 disconnect/kill-server）。同一 serial 同时只允许一个 worker，互斥由 TaskManager._running 保证（不再有 STF occupy/release）。

2.5 Action — 操作

Action 是任务循环里执行的"原子操作"（点赞 / 关注 / 滑动等）。每个 app 有独立的 Action 注册表（通过 create_action_registry() 创建），互不污染。全局基类 core/actions/base.py::BaseAction 提供通用能力。

2.6 进度上报（通用，适配任意 app）

Worker 通过 self.set_progress(**fields) 上报进度，前端统一解析展示。不再硬编码"已看视频数"等业务字段。

通用字段：

字段	类型	说明
done	int	已完成数量
total	int	总数量（0=不限数量，只显示已完成数）
unit	str	计数单位（"视频"/"轮次"/"条"）
action_counts	dict	操作计数 {"like": 3, "comment": 1}
elapsed	int	已运行时长（秒，可选，前端显示为 "Xm Ys"）

前端展示：进度条（百分比，total>0 时）+ "done/total unit" + 运行时长 + 操作计数徽章

示例：

# 抖音任务（有数量限制）
self.set_progress(done=5, total=80, unit="视频",
                  action_counts={"like": 3}, elapsed=120)

# 抖音任务（仅时长限制，无数量）
self.set_progress(done=5, total=0, unit="视频",
                  action_counts={"like": 3}, elapsed=120)

# 快手任务
self.set_progress(done=3, total=20, unit="轮次",
                  action_counts={"like": 2}, elapsed=60)

2.7 运行时长终止（通用，适配任意 app）

BaseWorker 提供运行时长终止能力，与"数量终止"配合使用。两者哪个先到就停。

成员	说明
self.max_duration	最大运行时长（秒），0=不限时
self._start_timer()	子类在 run_task 开头调用，启动计时
self.is_time_up()	是否已达 max_duration（max_duration=0 永远返回 False）
self.elapsed()	已运行时长（秒）

循环条件模板：

while not self.stopped():
    if watch_count > 0 and watched >= watch_count:
        break  # 数量终止
    if self.is_time_up():
        break  # 时长终止
    # ... 业务逻辑

三种终止模式：

• 仅数量：watch_count=80, max_duration=0 → 看完 80 个视频停
• 仅时长：watch_count=0, max_duration=1800 → 跑满 30 分钟停
• 双条件：watch_count=80, max_duration=1800 → 哪个先到就停
• 都为 0：永不停止，需手动停止

2.8 任务调度模式（schedule）

任务计划 TaskJob.schedule 支持三种模式：

mode	字段	行为
once	无	手动执行（前端点"立即执行"或调 /api/jobs/<id>/run）
cron	cron	定时启动：到 cron 时间点自动启动 worker
cron_stop	cron + stop_cron	定时启停：启动 cron 到点启动，停止 cron 到点停止本任务的 worker

cron_stop 模式只停止本 job 启动的 worker，不影响其他正在运行的任务。cron / cron_stop 还可带 window 运行窗口（每天重复，支持跨午夜如 21:00-09:00）：cron 触发点落在窗口外时 本次不启动，调度器会找窗口内下一个触发点；未配置/非法窗口 = 不限制。手动执行不受 window 限制。 典型用法：

{
  "schedule": {
    "mode": "cron_stop",
    "cron": "0 9 * * *",
    "stop_cron": "0 18 * * *",
    "window": {"start": "09:00", "end": "18:00"}
  }
}

含义：每天 9:00-18:00 为运行窗口；9 点自动启动任务，18 点自动停止。

2.9 心跳看门狗

每个 Worker 在 set_action / set_progress / heartbeat 时更新心跳时间戳。看门狗线程（_Watchdog）定期扫描，若超过 _HEARTBEAT_TIMEOUT=120s 未更新则判定卡死，标记 error。

长耗时操作必须周期性调用 self.heartbeat()，否则会被误杀。

2.10 generic_steps 通用步骤任务

task_type="generic_steps"（tasks/generic/task.py）是把任意 App 操作编排成"步骤链"的通用任务： 前端步骤编辑器拖拽节点 → 保存为 params.steps（JSON 数组）→ worker 按顺序执行。顶层 steps 只顺序执行一次——需要重复跑的操作必须显式放进 loop 节点（见下表）。任务级参数：

{
  "max_duration": 0,
  "steps": [ { "id": "step_1", "type": "open_app", "label": "打开抖音", "params": {...} } ]
}

• max_duration — 最大运行时长（秒），0=不限时
• steps — 步骤数组。每步 {id, type, label, params}；id 前端生成保证唯一
• 进度上报：done = 累计已执行的非容器步骤数（loop/group/if_el 不计入，避免监控噪音）、total=0、unit="操作"（前端显示"已执行 N 次操作"）
• 公共参数：每步都可有 probability（0-100，默认 100，<100 时按百分比概率决定本次是否执行该步）
• 嵌套深度上限 5：loop/group 的 children、if_el 的 then/else 递归嵌套超过 5 层会被跳过并告警
• 步骤执行会做选择器健康跟踪：某 selector 连续未命中达阈值记 last_warning，提示 App 改版导致选择器失效

全量 18 种节点（STEP_TYPES）：

type	作用	必填 params	子步骤字段
open_app	启动 App	package；wait_home/home_feature 可选	-
stop_app	强制结束 App（冷启动）	package	-
screen_on	亮屏（息屏时唤醒并滑动解锁）	-	-
screen_off	息屏	-	-
keep_screen	保持亮屏/恢复自动息屏（svc power stayon）	mode=on/off	-
key_event	按键（返回/Home/回车/菜单等）	key	-
swipe	滑动	direction(up/down/left/right)；duration_min/duration_max	-
swipe_until	滑动直到元素出现（可找到后点击）	selector_type+selector_value；direction/max_swipes/click_when_found	-
click	点击元素	selector_type+selector_value；wait_timeout	-
click_xy	点击坐标（屏幕百分比，中心=50/50）	x/y	-
long_click	长按元素	selector_type+selector_value；duration	-
wait_el	等待元素出现（条件等待）	selector_type+selector_value；timeout	-
input_text	在当前焦点输入框输入	mode=random/fixed；texts(随机候选) 或 fixed_text；clear_first	-
clipboard	剪贴板注入（ClipInject 通道）	text；paste=是否立即粘贴	-
wait	等待时长	min/max	-
loop	循环块	loop_mode=rounds/time/forever；max_iterations 或 loop_duration	children
group	动作组（按序执行一次，可折叠复用）	-	children
if_el	条件判断：命中→then，超时→else	selector_type+selector_value；timeout	then / else

选择器 selector_type 允许：xpath / description / text / resourceId / descriptionContains / className；仅 if_el 额外支持 ocr（截屏 OCR 按文字匹配，UI 树里没有的文字也能找到，可选 ocr_click 命中后自动点击）。 带选择器的步骤（click/long_click/swipe_until/wait_el/if_el）都必须填 selector_value。

xpath 序号必须整体加括号（2026-09-10 修复）：同一属性多个实例时，"第 k 个匹配"要写 (//*[@resource-id="x"])[k]；写成 //*[@resource-id="x"][k] 是"在其父节点中排第 k"， 多实例时 [2..n] 全部匹配不到（表现为运行时"未找到元素"，但界面上明明有这个元素）。 元素抓取器现已生成带括号形式；执行器 _norm_legacy_xpath 会自动纠正旧任务里的前者（只改前缀， 结构路径 .../FrameLayout[2] 的兄弟序号不动）。等价工具定位优先用 text/resourceId，比序号 xpath 稳。

静默跳过语义：worker 对未知 type / 缺必填（如 package、selector_value 为空）只打 warning 跳过，不会报错失败——任务会"看起来成功但啥也没干"。因此写任务必须自行校验：用编辑器内置 validate + "测试此步骤"逐个验证选择器（见 §2.11）。 权威 schema：tasks/generic/task.py 的 STEP_TYPES（后端执行器）与 static/admin/editor.js 的 STEP_LIB（前端操作库）必须保持同步——改节点结构两边要一起改。 AI 辅助生成：用一句话需求 → AI 生成 generic_steps 任务（步骤 JSON → 编辑器预填 → 人工确认）的规划见 doc/AI_TASK_GEN.md。

2.11 自定义动作与单步测试

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

• POST /api/custom_actions 只校验 name 非空 + 至少 1 个步骤，否则 400；steps 整段以 JSON 存库
• 保存前前端先剥掉步骤 id（_stripIds），避免同一动作多次拖入后 id 冲突；拖入画布时前端把该动作展开成一个 group 节点（{type:"group", children: 动作步骤}），没有 action_ref 这类"引用型"节点——动作是复制展开而非引用
• 更新/删除：PUT/DELETE /api/custom_actions/<id>（更新同样要求 name + ≥1 步）

单步测试（编辑器"测试此步骤"）：POST /api/steps/test 传 {serial, step}，在指定设备上 adb + u2 只读连接试执行单步并验证选择器，返回 result = "命中" / "未找到" / "已执行" （后端 tasks/generic/task.py::test_step + 前端 editor.js _testStep）。与运行中的任务互不干扰。

---

3. 新增一个 app 任务（完整步骤）

以"快手养号"为例。完整步骤 6 步，全部代码可直接复制。

签名提醒：以下模板的 Worker.__init__ / Task.create_worker 已去掉 stf_client / stf 参数——现行签名是 BaseWorker.__init__(self, serial, params=None, daemon=True)、 Task.create_worker(self, serial, params)（对照 tasks/douyin/task.py），新增任务照此抄， 不要再带 stf 形参。

步骤 1：在 tasks/ 下建 kuaishou/ 子包

tasks/kuaishou/
├── __init__.py
├── task.py
└── actions/
    ├── __init__.py
    ├── base.py
    └── like.py

步骤 2：写 actions/base.py（本任务的注册表）

# tasks/kuaishou/actions/base.py
"""快手 Action 注册表。"""
from core.actions import (
    BaseAction, register_action, create_action_registry,
    list_actions, get_action, should_trigger,
)

# 快手专属操作注册表（独立 dict，不污染其他 app）
ACTIONS = create_action_registry()


def list_action_types():
    """返回所有已注册快手操作的元信息（供前端展示）。"""
    return list_actions(ACTIONS)


def get_action_class(action_type):
    """按 action_type 取快手操作类。"""
    return get_action(ACTIONS, action_type)

步骤 3：写 actions/like.py

# tasks/kuaishou/actions/like.py
"""快手点赞 Action。"""
from core.actions import BaseAction, register_action, should_trigger
from core.logger import get_logger
from . import ACTIONS  # 必须从 __init__ 导入注册表

_log = get_logger("action.kuaishou.like")


@register_action(ACTIONS)
class LikeAction(BaseAction):
    action_type = "like"
    name = "点赞"
    description = "看完视频后随机点赞"
    default_params = {
        "rate": 0.8,                 # 触发概率 0~1
        "method": "double_tap",      # double_tap | heart_icon
    }

    def execute(self, d, params, worker):
        rate = float(params.get("rate", 0.8))
        if not should_trigger(rate):
            return False
        method = params.get("method", "double_tap")
        try:
            if method == "double_tap":
                info = d.info
                w, h = info["displayWidth"], info["displayHeight"]
                d.double_click(int(w * 0.5), int(h * 0.5))
            else:
                el = d(description="点赞")
                if not el.exists:
                    _log.info("未找到点赞按钮")
                    return False
                el.click()
            _log.info("点赞成功")
            return True
        except Exception as e:
            _log.warning(f"点赞异常: {e}")
            return False

步骤 4：写 actions/__init__.py（注意循环导入顺序）

# tasks/kuaishou/actions/__init__.py
"""快手操作注册包。import 触发各操作注册。

⚠️ 循环导入坑：必须先从 base 导入 ACTIONS，再导入各操作模块！
"""
from .base import (
    BaseAction, register_action, create_action_registry,
    list_actions, get_action, should_trigger,
    ACTIONS, list_action_types, get_action_class,
)

# 再导入各操作模块，触发 @register_action(ACTIONS) 注册
from . import like  # noqa: F401
# 新增 action 在此 import，例如：from . import follow

__all__ = [
    "BaseAction", "register_action", "create_action_registry",
    "list_actions", "get_action", "should_trigger",
    "ACTIONS", "list_action_types", "get_action_class",
]

步骤 5：写 task.py

# tasks/kuaishou/task.py
"""快手养号任务定义。

本文件自包含所有快手养号参数，不依赖 core 的业务配置。
快手专属操作（点赞/关注等）在 actions/ 子包里，xpath 只适用于快手。
"""
import time
import random

from tasks.base import BaseTask, register_task
from .actions import list_action_types, get_action_class
from core.device_worker import BaseWorker, _update_status
from core.u2_helper import ensure_app_running, wait_for_app_home
from core.logger import get_logger

_log = get_logger("task.kuaishou")

KUAISHOU_PKG = "com.smile.gifmaker"

DEFAULT_PARAMS = {
    "watch_count": 50,        # 观看视频数量
    "watch_min": 5.0,         # 单个视频最短观看秒数
    "watch_max": 30.0,        # 单个视频最长观看秒数
    "swipe_min": 0.25,        # 上滑手势最短时长(秒)
    "swipe_max": 0.50,        # 上滑手势最长时长(秒)
    "gap_min": 1.0,           # 视频间隔最短秒数
    "gap_max": 3.0,           # 视频间隔最长秒数
    "actions": {
        "like": {
            "enabled": True,
            "params": {"rate": 0.3, "method": "double_tap"},
        },
    },
}


class KuaishouWorker(BaseWorker):
    """快手养号 worker。"""

    def __init__(self, serial, params=None):
        super().__init__(serial, params)
        p = {**DEFAULT_PARAMS, **(self.params or {})}
        self.watch_count = int(p["watch_count"])
        self.watch_min = float(p["watch_min"])
        self.watch_max = float(p["watch_max"])
        self.swipe_min = float(p["swipe_min"])
        self.swipe_max = float(p["swipe_max"])
        self.gap_min = float(p["gap_min"])
        self.gap_max = float(p["gap_max"])
        self.actions_cfg = p.get("actions", {})
        self._actions = []
        for atype, cfg in self.actions_cfg.items():
            if not cfg.get("enabled"):
                continue
            cls = get_action_class(atype)
            if cls:
                self._actions.append(cls())
        if self._actions:
            summary = ", ".join(
                f"{a.action_type}(rate={self.actions_cfg.get(a.action_type, {}).get('params', {}).get('rate', '?')})"
                for a in self._actions
            )
            _log.info(f"[{serial}] 启用操作: {summary}")
        else:
            _log.warning(f"[{serial}] 未启用任何操作")

    def run_task(self, d):
        """快手养号主逻辑。d 是 u2.Device，已由基类连好。"""
        def is_home(d):
            return (d(descriptionContains="首页").exists
                    or d(descriptionContains="拍摄").exists)

        d.app_start(KUAISHOU_PKG, wait=True)
        if not wait_for_app_home(d, KUAISHOU_PKG, is_home, timeout=40):
            self.set_action("首页加载超时，继续尝试")

        watched = 0
        action_counts = {a.action_type: 0 for a in self._actions}
        # 初始化通用进度上报（前端会解析 done/total/unit + action_counts）
        self.set_progress(done=0, total=self.watch_count, unit="视频",
                          action_counts=action_counts)

        while not self.stopped() and watched < self.watch_count:
            if not ensure_app_running(d, KUAISHOU_PKG):
                _update_status(self.serial, status="error",
                               last_error="快手连续重启失败，放弃该设备")
                return

            watch = random.uniform(self.watch_min, self.watch_max)
            self.set_action(f"观看视频 {watched+1}/{self.watch_count}，{watch:.0f}s")
            time.sleep(watch)

            # 执行启用的操作
            for action in self._actions:
                if self.stopped():
                    break
                cfg = self.actions_cfg.get(action.action_type, {})
                params = {**action.default_params, **cfg.get("params", {})}
                try:
                    ok = action.execute(d, params, self)
                    _log.info(f"[{self.serial}] 视频{watched+1}: {action.action_type} execute={ok}")
                    if ok:
                        action_counts[action.action_type] += 1
                        time.sleep(random.uniform(0.5, 1.5))
                except Exception as e:
                    _log.error(f"[{self.serial}] 视频{watched+1}: 操作 {action.action_type} 异常: {e}")

            if self.stopped():
                break
            d.swipe(500, 1000, 500, 300, random.uniform(self.swipe_min, self.swipe_max))
            time.sleep(random.uniform(self.gap_min, self.gap_max))
            watched += 1
            # 上报通用进度
            self.set_progress(done=watched, total=self.watch_count, unit="视频",
                              action_counts=dict(action_counts))

        summary = f"完成 {watched} 个视频" + "".join(
            f"，{k} {v}次" for k, v in action_counts.items() if v
        )
        _update_status(self.serial, current_action=summary)


@register_task
class KuaishouTask(BaseTask):
    """快手养号任务。"""
    task_type = "kuaishou_nurture"
    name = "快手养号"
    description = "自动观看快手视频，按配置执行点赞等操作"
    default_params = dict(DEFAULT_PARAMS)

    @classmethod
    def list_action_types(cls):
        return list_action_types()

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

    def create_worker(self, serial, params):
        merged = {**DEFAULT_PARAMS, **(params or {})}
        # actions 字段参数级深合并（保留前端没传的操作默认值）
        default_actions = DEFAULT_PARAMS["actions"]
        merged_actions = params.get("actions", {}) if params else {}
        for atype, dflt in default_actions.items():
            if atype not in merged_actions:
                merged_actions[atype] = dflt
            else:
                cfg = merged_actions[atype]
                merged_cfg = {}
                for k in ("enabled", "params"):
                    merged_cfg[k] = cfg.get(k, dflt.get(k))
                merged_params = dict(dflt.get("params", {}))
                merged_params.update(cfg.get("params", {}))
                merged_cfg["params"] = merged_params
                merged_actions[atype] = merged_cfg
        merged["actions"] = merged_actions
        return KuaishouWorker(serial, params=merged)

步骤 6：注册任务包

tasks/kuaishou/__init__.py：

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

tasks/__init__.py 加一行：

# tasks/__init__.py
from .base import BaseTask, register_task, list_task_types, get_task_class
from .douyin import task  # noqa: F401
from .generic import task  # noqa: F401
from .kuaishou import task  # noqa: F401  ← 新增这一行

完成。重启 web 后，前端任务类型下拉自动出现 kuaishou_nurture。

---

4. Worker 开发指南

BaseWorker 位于 core/device_worker.py，封装了设备生命周期、心跳、异常分类、与调度器的状态通信。

4.1 生命周期

STFDevice.acquire(serial)     # 连上设备（互斥由 TaskManager._running 保证）：
   │                          #   · IP:5555 → adb connect 直连（绝不 disconnect）
   │                          #   · USB 无冒号 → 本机 adb，不在则 220 远程 adb server
   ▼
u2.connect                    # 连接 uiautomator2（带 30s 超时保护）
   │
   ▼
setup(d)                     # 子类可选钩子（启动 app、授权、关闭弹窗）
   │
   ▼
run_task(d)  ◄── 必须实现     # 任务主循环
   │
   ▼
teardown(d)                  # 子类可选钩子（退出 app、清理）
   │
   ▼
STFDevice.release()          # 空操作（不 disconnect、不 kill-server）

任意阶段抛出 DeviceOfflineError → 立即终止，不重试。其他异常 → 按 Job 的 retry 策略重试。

4.2 必须实现 / 可选钩子

方法	是否必须	说明
run_task(self, d)	必须	任务主循环，d 为 uiautomator2.Device
setup(self, d)	可选	设备/应用初始化
teardown(self, d)	可选	收尾，即使出错也会执行
on_error(self, d, err)	可选	异常通知钩子

4.3 工具方法

方法	说明
self.stopped()	循环里必须检查，返回 True 表示收到停止信号
self.set_action(s)	设置当前动作（前端大屏"当前动作"列可见），同时刷新心跳
self.set_progress(**fields)	上报进度（见 §2.6），同时刷新心跳
self.heartbeat()	手动刷新心跳（长操作中间调）
self.params	已合并 DEFAULT_PARAMS 与 Job 参数后的最终参数
self.serial	当前设备序列号
self.d	u2.Device（run_task 的 d 参数）

4.4 进度上报规范（重要）

所有 app 任务必须用 set_progress 上报进度，前端会统一解析展示。

# ✅ 正确：用通用字段
self.set_progress(done=5, total=80, unit="视频",
                  action_counts={"like": 3, "comment": 1})

# ❌ 错误：硬编码业务字段（前端无法识别）
self.set_progress(videos_watched=5)  # 前端不认这个字段

前端展示效果：

• 进度条：████████░░░░ （按 done/total 算百分比）
• 计数文本：5/80 视频
• 操作徽章：点赞 3 评论 1

4.5 异常分类

异常	处理
DeviceOfflineError	设备掉线，不重试，立即释放
其他 Exception	按 Job 的 retry 次数重试，退避后重新申请设备

4.6 run_task 模板（可直接复制）

def run_task(self, d):
    """任务主循环模板。"""
    # 1. 启动 app
    d.app_start("com.xxx", wait=True)
    if not wait_for_app_home(d, "com.xxx", lambda d: d(text="首页").exists, timeout=40):
        self.set_action("首页加载超时，继续尝试")

    # 2. 初始化进度上报
    watched = 0
    action_counts = {a.action_type: 0 for a in self._actions}
    self.set_progress(done=0, total=self.watch_count, unit="视频",
                      action_counts=action_counts)

    # 3. 主循环
    while not self.stopped() and watched < self.watch_count:
        # 3.1 确保 app 在前台
        if not ensure_app_running(d, "com.xxx"):
            _update_status(self.serial, status="error",
                           last_error="app 连续重启失败")
            return

        # 3.2 观看
        watch = random.uniform(self.watch_min, self.watch_max)
        self.set_action(f"观看 {watched+1}/{self.watch_count}，{watch:.0f}s")
        time.sleep(watch)

        # 3.3 执行操作
        for action in self._actions:
            if self.stopped():
                break
            cfg = self.actions_cfg.get(action.action_type, {})
            params = {**action.default_params, **cfg.get("params", {})}
            try:
                if action.execute(d, params, self):
                    action_counts[action.action_type] += 1
            except Exception as e:
                _log.error(f"[{self.serial}] 操作异常: {e}")

        # 3.4 滑动
        if self.stopped():
            break
        d.swipe(500, 1000, 500, 300, 0.3)
        time.sleep(random.uniform(1.0, 3.0))

        # 3.5 上报进度
        watched += 1
        self.set_progress(done=watched, total=self.watch_count, unit="视频",
                          action_counts=dict(action_counts))

    # 4. 收尾
    summary = f"完成 {watched} 个视频"
    _update_status(self.serial, current_action=summary)

铁律：循环里必须高频调用 self.stopped()，否则停止按钮无响应、看门狗误杀。

---

5. Action 开发指南

5.1 全局基类与独立注册表

• 全局基类：core/actions/base.py::BaseAction，提供 should_trigger 等通用能力
• 每个 app 通过 create_action_registry() 创建独立注册表，避免不同 app 的 like / comment 同名冲突
• 注册装饰器：@register_action(ACTIONS)，ACTIONS 为本 app 的注册表

5.2 BaseAction 关键 API

成员	说明
action_type	类属性，注册 key，必须与 params["actions"] 的 key 一致
name	类属性，中文名（前端展示）
description	类属性，描述
default_params	类属性，自包含默认参数
execute(self, d, params, worker)	必须实现，返回 True=成功 / False=跳过
should_trigger(rate)	按 rate 概率返回是否触发（rate=0.8 → 80% 概率 True）

5.3 完整 Action 模板

下面以"关注"操作为例，展示一个完整 Action 的写法（多策略定位 + 概率触发 + 异常兜底）：

# tasks/xxx/actions/follow.py
import time
import random

from core.actions import BaseAction, register_action, should_trigger
from core.logger import get_logger
from . import ACTIONS  # 从 __init__ 导入本 app 注册表

_log = get_logger("action.xxx.follow")


@register_action(ACTIONS)
class FollowAction(BaseAction):
    action_type = "follow"
    name = "关注"
    description = "看完视频后随机关注作者"
    default_params = {
        "rate": 0.1,
    }

    def execute(self, d, params, worker):
        rate = float(params.get("rate", 0.1))
        if not should_trigger(rate):
            return False

        # 定位关注按钮（多策略组合，失败回退）
        for desc in ("关注", "未关注", "follow"):
            el = d(description=desc)
            if el.exists:
                el.click()
                break
        else:
            _log.info("未找到关注按钮")
            return False

        time.sleep(1.0)
        _log.info("关注成功")
        return True

返回值约定：True=成功执行；False=主动跳过（概率未中、元素不存在等）；抛异常=执行失败，由 Worker 捕获并记录。

---

6. 参数设计规范

6.1 自包含

DEFAULT_PARAMS 放在 task.py 顶部，所有该任务需要的参数都要列出，包括每个 action 的子参数。不允许"隐式默认值"散落在 action 内部。

6.2 参数合并（参数级深合并）

create_worker 时执行三层合并：

def create_worker(self, serial, params):
    merged = {**DEFAULT_PARAMS, **(params or {})}
    # actions 字段参数级深合并
    default_actions = DEFAULT_PARAMS["actions"]
    merged_actions = params.get("actions", {}) if params else {}
    for atype, dflt in default_actions.items():
        if atype not in merged_actions:
            merged_actions[atype] = dflt
        else:
            cfg = merged_actions[atype]
            merged_cfg = {}
            for k in ("enabled", "params"):
                merged_cfg[k] = cfg.get(k, dflt.get(k))
            # params 再深合并一层
            merged_params = dict(dflt.get("params", {}))
            merged_params.update(cfg.get("params", {}))
            merged_cfg["params"] = merged_params
            merged_actions[atype] = merged_cfg
    merged["actions"] = merged_actions
    return MyWorker(serial, params=merged)

即：

• 顶层字段：Job 参数覆盖默认参数
• actions 字段：参数级深合并，前端可只覆盖某个 action 的某个子字段（如只改 like.rate）

6.3 前端任务参数 JSON 示例

Job 下发时只传需要覆盖的字段，调度器做深合并。例如只想把点赞概率从 0.3 调到 0.5，Job params 只需：

{
  "actions": {
    "like": {"params": {"rate": 0.5}}
  }
}

其余字段自动取 DEFAULT_PARAMS。不要在 Job 里传完整 params——升级默认值时会丢失新字段。

---

7. 日志规范

7.1 获取 logger

from core.logger import get_logger

log = get_logger("task.kuaishou")          # 任务日志 → logs/task.log
log = get_logger("action.kuaishou.like")   # action 日志 → logs/action.log
log = get_logger("core.task_manager")      # 核心日志 → logs/core.log
log = get_logger("web")                    # web 日志 → logs/web.log

logger 名前缀决定写入哪个文件：

前缀	文件
core.*	logs/core.log
task.*	logs/task.log
action.*	logs/action.log
web.*	logs/web.log

7.2 级别

• DEBUG — 详细元素查找、参数 dump（生产关闭）
• INFO — 正常流程节点（启动、轮次、action 结果）
• WARNING — 可恢复异常（元素找不到、action 失败）
• ERROR — 不可恢复错误（设备掉线、调度失败）

7.3 规则

• 禁止 print，统一用 get_logger
• 日志里带 [{self.serial}] 设备前缀，多设备并发时才能区分
• 单文件 10MB 滚动，保留 5 份历史，无需手动清理
• 不要在循环里高频打 INFO（如每个 exists() 都打），用 DEBUG

---

8. 设备调试（STF 已摘除）

平台已在代码层完全摘除 OpenSTF（occupy/release、remoteConnect 桥接、网页看屏均已退役）， 调度与设备操作直接基于 adb 真实现状。迁移过程、决策与回滚方式见 doc/STF_REMOVAL.md， 本文不再展开 STF 排障。以下结论在无 STF 时代仍然成立：

8.1 不重试原则

DeviceOfflineError 一律不重试——设备掉线后短时间内不会自愈，重试只会占用调度队列并阻塞调度器。 该错误由 STFDevice.acquire（adb connect 失败 / 设备不在本机与 220 远程 adb server）或 u2 连接失败 触发，设备直接进入冷却。

8.2 u2.connect 30s 超时

u2.connect() 在 atx-agent 无响应时会永久 hang。基类用 ThreadPoolExecutor + future.result(timeout=30) 包裹（USB 设备经 220 远程 adb server 建连接同样带 30s 保护），超时抛异常并标记 status=error。 子类无需处理，但不要绕过超时保护在 run_task 里直接调 u2.connect()。

8.3 前台 App 扫描（不打扰设备）

Web 仍提供"扫描前台 App"按钮（POST /api/scan_foreground），按设备状态分类处理，不打扰设备：

设备状态	处理方式	是否打扰
worker 运行中（IP:5555）	复用已有 ADB 连接（remote_adb_url）查询	否
worker 运行中（USB）	经 220 远程 adb server 查询	否
空闲设备	不主动 adb connect，直接返回"空闲"	否

无"他人占用"概念（单实例部署，设备互斥由 TaskManager._running 保证）。空闲设备不主动 connect， 是因为 IP:5555 的 adb transport 为共享连接，反复 connect/disconnect 会扰动现有连接。

---

9. 定位元素技巧

9.1 抓界面

用 weditor（pip install weditor → python -m weditor）实时查看 UI 树，复制定位表达式。

9.2 定位优先级

description  >  descriptionContains  >  resourceId  >  text/textContains  >  xpath

• description 最稳，开发者较少改动 contentDescription
• descriptionContains 模糊匹配，适配不同版本文案（如"点赞"/"未点赞"）
• resourceId 注意带包名前缀（com.xxx:id/...），跨版本可能变，建议多候选
• xpath 用相对定位，禁止依赖 FrameLayout[2] / LinearLayout[3] 这类绝对序号

9.3 多策略组合 + 回退

def find_like_button(d):
    """多策略定位点赞按钮，失败回退。"""
    # 1. description 精确
    for desc in ("点赞", "未点赞", "like"):
        el = d(description=desc)
        if el.exists:
            return el
    # 2. descriptionContains 模糊
    for kw in ("赞", "like"):
        el = d(descriptionContains=kw)
        if el.exists:
            return el
    # 3. resourceId 列表（多候选）
    for rid in ("com.xxx:id/aky", "com.xxx:id/d-like-view-icon"):
        el = d(resourceId=rid)
        if el.exists:
            return el
    return None

9.4 xpath 写法

# ✅ 相对定位，稳
d.xpath('//android.widget.TextView[@text="关注"]').click()

# ❌ 绝对序号，UI 一变就崩
d.xpath('//FrameLayout[2]/LinearLayout[1]/TextView[3]').click()

---

10. 常见问题

10.1 循环导入

actions/__init__.py 必须先导入 base 再导入各 action 模块：

# tasks/xxx/actions/__init__.py
from .base import ACTIONS, ...   # 1. 先建注册表
from . import like               # 2. 再导入各 action，触发 @register_action

tasks/__init__.py 同理：先 from .base import BaseTask，再 from . import douyin、from . import generic。

10.2 中文输入

uiautomator2 默认 IME 不支持中文。需切到 fastinput：

try:
    d.set_fastinput_ime(True)    # 切入
    d.send_keys("中文内容")
finally:
    try:
        d.set_fastinput_ime(False)  # 用完切回
    except Exception:
        pass

设备未装 FastInput 输入法时 set_fastinput_ime 会静默失败，建议加 try/except + 日志。

10.3 多设备并发

adb_helper 内置全局锁串行化所有 adb 调用（adb connect / adb devices 等）。原因：

• 多线程并发调 adb 会触发 adb server 竞争，导致连接抖动
• 禁止在任务代码里调 adb kill-server——会踢掉所有设备的连接
• 设备申请/释放走 device_pool（清单/在线） + STFDevice.acquire（IP:5555 直连 / USB 走 220 远程 server），与 adb_helper 全局锁配合避免冲突

# ✅ 正确：用 adb_helper 封装
from core.adb_helper import adb_connect
adb_connect(serial)

# ❌ 错误：自己起 subprocess 调 adb，绕过全局锁
import subprocess
subprocess.run(["adb", "connect", serial])

# ❌ 严禁
subprocess.run(["adb", "kill-server"])

10.4 看门狗误杀

若任务有长耗时操作（如长视频播放等待 5 分钟），看门狗可能误判卡死。解决：

• 在长操作内部周期性调用 self.heartbeat()（如每 30 秒一次），而不是只在整个操作前后调
• 不要调高看门狗阈值——真卡死的设备需要尽快释放

# 长等待的正确写法
end = time.time() + 300
while time.time() < end:
    if self.stopped():
        return
    self.heartbeat()      # 长循环内部也要心跳
    time.sleep(5)

10.5 u2.connect 卡死

u2.connect() 在 atx-agent 无响应时会永久 hang。基类已用 ThreadPoolExecutor + future.result(timeout=30) 包裹，超时返回 None 并抛异常。子类无需处理，但要避免在 run_task 里直接调 u2.connect()。

10.6 任务参数前端覆盖

Job 下发时只传需要覆盖的字段，调度器做深合并（见 §6.2）。例如只想把点赞概率从 0.8 调到 0.5，Job params 只需：

{
  "actions": {
    "like": {"params": {"rate": 0.5}}
  }
}

其余字段自动取 DEFAULT_PARAMS。不要在 Job 里传完整 params——升级默认值时会丢失新字段。

10.7 进度上报必须用通用字段

前端只认 progress = {done, total, unit, action_counts} 结构。不要用 videos_watched、round_idx 等业务字段名——前端不会识别。

# ✅ 正确
self.set_progress(done=5, total=80, unit="视频",
                  action_counts={"like": 3})

# ❌ 错误（前端不认）
self.set_progress(videos_watched=5, round_idx=3)

---

附录：新增任务 Checklist

新建一个 app 任务时，按此清单逐项确认：

• tasks/<app>/__init__.py 有 from . import task
• tasks/<app>/task.py 有 DEFAULT_PARAMS（自包含）+ Worker(BaseWorker) + Task(BaseTask) + @register_task
• Worker.run_task 已实现，循环顶部和 action 之间都检查 self.stopped()
• Worker.run_task 用 self.set_progress(done=, total=, unit=, action_counts=) 上报进度
• 长循环内周期性调用 self.heartbeat()
• tasks/<app>/actions/__init__.py 先 from .base import ACTIONS 再导入各 action
• 每个 Action 有 action_type / name / default_params / execute，返回 True/False
• Task.create_worker 做 actions 参数级深合并（照抄抖音模板）
• tasks/__init__.py 已 from .<app> import task 注册
• 日志用 get_logger("task.<app>") / get_logger("action.<app>.<name>")，无 print
• 定位元素优先 description / descriptionContains，resourceId 多候选，xpath 用相对定位
• 中文输入用 set_fastinput_ime，加 try/except
• adb 操作走 adb_helper，未自起 subprocess，未 kill-server

面向 generic_steps / 步骤编辑器：

• 编排 generic_steps 用编辑器 validate + "测试此步骤"逐条自校验（未知 type / 缺必填只会 warning 跳过，不会报错失败）
• 新增/修改步骤节点时，tasks/generic/task.py 的 STEP_TYPES 与 static/admin/editor.js 的 STEP_LIB 同步更新
• 新增 task_type 后，tasks/__init__.py 的 import、GET /api/task_types 返回、前端"新建任务"下拉一致（改完需重启 web）

完成上述清单后，重启 web，前端单页应用即可看到新任务类型并可下发。