121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
"""任务类型基类与注册机制。
|
||
|
||
目录约定：
|
||
config.py — 项目根配置（STF/adb/web 等，核心程序用，不放任务参数）
|
||
core/ — 核心程序（STF、adb、BaseWorker、任务管理器）
|
||
tasks/ — 任务包，每个任务一个子文件夹，自包含任务+专属操作+参数
|
||
tasks/<name>/
|
||
__init__.py — import task 触发注册
|
||
task.py — BaseTask 子类 + Worker + DEFAULT_PARAMS
|
||
actions/ — 该任务专属操作（xpath 等只适用于本任务，不污染全局）
|
||
__init__.py
|
||
base.py — 本任务的操作基类 + 注册器
|
||
xxx.py — 具体操作
|
||
|
||
新增任务步骤（照着 tasks/douyin/ 抄即可）：
|
||
|
||
1. 在 tasks/ 下新建 my_task/ 子包（文件夹 + __init__.py）
|
||
2. 在 my_task/task.py 顶部写默认参数：
|
||
|
||
DEFAULT_PARAMS = {"loop": 10, "actions": {...}}
|
||
|
||
3. 写一个 BaseWorker 子类，实现 run_task(d)：
|
||
|
||
from core.device_worker import BaseWorker, _update_status
|
||
|
||
class MyWorker(BaseWorker):
|
||
def run_task(self, d): # d 是 u2.Device，基类已连好
|
||
d.app_start("com.xxx", wait=True)
|
||
while not self.stopped():
|
||
self.set_action("干活中")
|
||
self.set_progress(done=1)
|
||
|
||
4. 写一个 BaseTask 子类，用 @register_task 装饰，实现 create_worker：
|
||
|
||
@register_task
|
||
class MyTask(BaseTask):
|
||
task_type = "my_task"
|
||
name = "我的任务"
|
||
default_params = dict(DEFAULT_PARAMS)
|
||
|
||
# 有专属操作时重写这两个方法，从 actions 包取
|
||
@classmethod
|
||
def list_action_types(cls):
|
||
from .actions import list_action_types
|
||
return list_action_types()
|
||
|
||
@classmethod
|
||
def get_action_class(cls, action_type):
|
||
from .actions import get_action_class
|
||
return get_action_class(action_type)
|
||
|
||
def create_worker(self, serial, params):
|
||
return MyWorker(serial, params=params)
|
||
|
||
5. 在 my_task/__init__.py 加：from . import task （触发注册）
|
||
6. 在 tasks/__init__.py 加：from .my_task import task （触发注册）
|
||
|
||
做完前端自动出现新任务。有专属操作时前端"操作"区自动列出。
|
||
|
||
BaseWorker 已封装好的（子类不用管）：
|
||
- STF 占用/释放（try/finally 保证释放）
|
||
- adb 连接 / u2.connect
|
||
- 状态上报（status/current_action/last_error/model/remote_adb_url）
|
||
- 异常捕获（run_task 里抛异常 → status=error，设备仍会释放）
|
||
- stop 信号（self.stopped() 检查）
|
||
|
||
子类可用的工具：
|
||
self.stopped() — 是否被要求停止
|
||
self.set_action(s) — 上报当前动作
|
||
self.set_progress() — 上报进度字段
|
||
self.d — u2.Device
|
||
self.params — 任务参数 dict
|
||
core.u2_helper — 通用 u2 辅助函数（ensure_app_running/safe_click 等）
|
||
"""
|
||
|
||
|
||
class BaseTask:
|
||
"""任务类型基类。子类实现 create_worker。
|
||
|
||
有专属操作的任务重写 list_action_types / get_action_class，
|
||
从自己的 actions 子包取（操作 xpath 等只适用于本任务）。
|
||
"""
|
||
task_type = "base"
|
||
name = "基础任务"
|
||
description = ""
|
||
default_params = {}
|
||
|
||
def create_worker(self, serial, params):
|
||
"""返回一个 threading.Thread（已启动或待启动），执行实际任务。"""
|
||
raise NotImplementedError
|
||
|
||
@classmethod
|
||
def list_action_types(cls):
|
||
"""返回本任务支持的操作列表。默认无操作，子类按需重写。"""
|
||
return []
|
||
|
||
@classmethod
|
||
def get_action_class(cls, action_type):
|
||
"""按 action_type 取操作类。默认无操作，子类按需重写。"""
|
||
return None
|
||
|
||
|
||
_TASK_TYPES = {}
|
||
|
||
|
||
def register_task(task_cls):
|
||
"""注册新任务类型。用作装饰器。"""
|
||
_TASK_TYPES[task_cls.task_type] = task_cls
|
||
return task_cls
|
||
|
||
|
||
def list_task_types():
|
||
"""返回所有已注册任务类型的元信息（供前端展示）。"""
|
||
return [{"task_type": t.task_type, "name": t.name, "description": t.description,
|
||
"default_params": t.default_params} for t in _TASK_TYPES.values()]
|
||
|
||
|
||
def get_task_class(task_type):
|
||
"""按 task_type 取任务类。"""
|
||
return _TASK_TYPES.get(task_type)
|