121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
"""Action 基类与注册机制（全局通用，所有 app 任务共享接口）。
|
||
|
||
设计要点：
|
||
1. BaseAction 定义统一接口：execute(d, params, worker) -> bool
|
||
2. 每个 app 任务有自己的注册表（create_action_registry），
|
||
互不污染——抖音的 like 和快手的 like 各注册各的。
|
||
3. 概率触发、状态上报等通用逻辑放这里，子类不重复写。
|
||
4. ActionContext 封装执行上下文，方便扩展（传 d/worker/进度/计数器）。
|
||
|
||
新增 app 专属操作步骤（照着 tasks/douyin/actions/like.py 抄）：
|
||
|
||
1. 在 tasks/<app>/actions/ 下新建 my_action.py
|
||
2. 写一个 BaseAction 子类，用 @register_action 装饰，实现 execute：
|
||
|
||
from core.actions import BaseAction, register_action, should_trigger
|
||
from . import ACTIONS # 本 app 的注册表
|
||
|
||
@register_action(ACTIONS)
|
||
class MyAction(BaseAction):
|
||
action_type = "my_action"
|
||
name = "我的操作"
|
||
description = "做什么"
|
||
default_params = {"rate": 0.5}
|
||
|
||
def execute(self, d, params, worker):
|
||
if not should_trigger(params["rate"]):
|
||
return False
|
||
# ... 用 d(xpath=...) / d(text=...) 操作
|
||
return True
|
||
|
||
3. 在 tasks/<app>/actions/__init__.py 加：from . import my_action
|
||
|
||
做完前端自动出现该操作的可勾选项。
|
||
"""
|
||
import random
|
||
|
||
from core.logger import get_logger
|
||
|
||
_log = get_logger("action")
|
||
|
||
|
||
class ActionContext:
|
||
"""操作执行上下文，封装给 Action 用的所有依赖。
|
||
|
||
现在只是简单封装，后续可扩展（如传任务参数、设备信息、进度回调）。
|
||
Action 子类通过 worker 间接访问，或直接用 execute 的 d 参数。
|
||
"""
|
||
|
||
def __init__(self, worker=None, params=None):
|
||
self.worker = worker
|
||
self.params = params or {}
|
||
|
||
|
||
class BaseAction:
|
||
"""所有操作的基类。子类实现 execute。
|
||
|
||
约定：
|
||
execute 返回 True = 执行成功（会计入 action_counts）
|
||
execute 返回 False = 跳过/失败（不影响主流程，不抛异常）
|
||
execute 内部异常由调用方捕获，不影响其他操作和主循环
|
||
"""
|
||
action_type = "base"
|
||
name = "基础操作"
|
||
description = ""
|
||
default_params = {}
|
||
|
||
def execute(self, d, params, worker):
|
||
"""执行操作。
|
||
|
||
d: u2.Device — 已连好的设备
|
||
params: 本操作的参数 dict（已合并默认值）
|
||
worker: Worker 实例，可调 worker.set_action() 上报状态
|
||
返回 True=执行成功，False=跳过/失败。
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
def __repr__(self):
|
||
return f"<{self.__class__.__name__}({self.action_type})>"
|
||
|
||
|
||
def should_trigger(rate):
|
||
"""按概率决定是否触发（rate 是 0~1 的概率）。"""
|
||
return random.random() < rate
|
||
|
||
|
||
def register_action(registry):
|
||
"""装饰器：把 Action 类注册到指定注册表。
|
||
|
||
用法：
|
||
ACTIONS = create_action_registry()
|
||
@register_action(ACTIONS)
|
||
class MyAction(BaseAction): ...
|
||
|
||
每个 app 任务有自己的 ACTIONS 注册表，互不污染。
|
||
"""
|
||
def _decorator(action_cls):
|
||
registry[action_cls.action_type] = action_cls
|
||
_log.debug(f"注册操作: {action_cls.action_type} -> {action_cls.__name__}")
|
||
return action_cls
|
||
return _decorator
|
||
|
||
|
||
def create_action_registry():
|
||
"""创建一个独立的操作注册表（每个 app 任务一个）。
|
||
|
||
返回 dict: action_type -> ActionClass
|
||
"""
|
||
return {}
|
||
|
||
|
||
def list_actions(registry):
|
||
"""返回注册表中所有操作的元信息（供前端展示）。"""
|
||
return [{"action_type": a.action_type, "name": a.name,
|
||
"description": a.description,
|
||
"default_params": a.default_params} for a in registry.values()]
|
||
|
||
|
||
def get_action(registry, action_type):
|
||
"""按 action_type 取操作类。"""
|
||
return registry.get(action_type)
|