butubb
5c88cc6384
需求：`35377983067` 或 `35377983068` 都算我的号，要能一次列好几个。 - `cmp_value` 现在是**多值**：换行或 `|` 分隔，一行一个。 多个之间是 **OR**（任一命中即命中）；否定式（不等于/不包含）语义是 "一个都不许出现"。**逗号刻意不当分隔符**——要比对的文本本身就常带逗号 （`抖音号：123,456`），拆了就永远匹配不上。 - 比对核心抽成 `_cmp_hit(got, op, want) -> (是否命中, 命中的那个值)`， `_text_matches` 保留布尔版（多处调用只看结果，不改签名）。 日志因此能写清人话： `采集到 '抖音号：35377983068' 包含 2 个候选值 → 符合，命中『35377983068』` 否定式不成立时则指出"因为出现了『xxx』"——排查时一眼看出是哪个值惹的。 - 行为统一（**刻意的变更**）：候选值为空/全空白 = 误配 → 一律不命中。 以前"等于 + 两边都空"会返回 True（`"" == ""` 的巧合），等于选了"等于" 却没填值就永远命中、条件判断形同虚设；现在与编辑器告警、AI 草稿校验 的口径一致了（"选了运算符没填值 → 永远不命中"）。 - 前端：`cmp_value` 由单行输入换成**两行文本域**，标签与提示写明"可多行、 任一命中即可、逗号不是分隔符"。 - 文档：TASK_DEV.md §4.2 例子改成多号，并写明分隔符与 OR/否定语义。 自测：多值语义 15 项（换行/`|` 拆分、逗号不拆、空白清理、四种运算符的多值 OR 与否定语义、命中的是哪个值、候选全空不命中）+ 原有 38 项回归全过； `_exec_if_el` 走多值路径实测命中 then 分支、日志含候选数与命中值。
1104 lines
55 KiB
Python
1104 lines
55 KiB
Python
"""通用步骤任务定义。
|
||
|
||
前端步骤编辑器编排步骤链，保存为 params.steps（JSON 数组）。
|
||
worker 按 steps 顺序执行，支持 loop 步骤循环、停止信号、进度上报。
|
||
|
||
步骤 schema:
|
||
{
|
||
"id": "step_1", # 唯一 id（前端生成）
|
||
"type": "open_app", # 步骤类型
|
||
"label": "打开抖音", # 用户可读名称
|
||
"params": { ... } # 步骤参数（按 type 不同）
|
||
}
|
||
|
||
支持的步骤类型：
|
||
open_app - 启动 app
|
||
stop_app - 强制结束 app（am force-stop，下次打开冷启动）
|
||
screen_on - 亮屏（息屏时唤醒并滑动解锁）
|
||
screen_off - 息屏
|
||
keep_screen - 保持亮屏/恢复自动息屏（svc power stayon，设备充电时常亮）
|
||
key_event - 按键（返回/Home/回车/菜单等，d.press）
|
||
swipe - 滑动（方向/时长）
|
||
swipe_until - 滑动直到元素出现（最多 N 次，可选中找到后点击）
|
||
click - 点击元素（选择器）
|
||
click_xy - 点击坐标（屏幕百分比，无选择器时兜底）
|
||
long_click - 长按元素（选择器 + 时长）
|
||
wait_el - 等待元素出现（条件等待，替代固定时长 wait）
|
||
input_text - 输入文字（随机候选/指定文字，可先清空）
|
||
wait - 等待时长
|
||
notify - 发自定义通知（标题/正文自己写，推送 webhook）
|
||
stop_self - 停止本设备的任务（其它设备不受影响）
|
||
gesture - 回放录制的手势轨迹（纯录制：路径与时长照录制，见 core/gesture.py）
|
||
loop - 循环块（含 children 步骤列表 + max_iterations）
|
||
group - 动作组（含 children 步骤列表，按序执行一次，可折叠复用）
|
||
"""
|
||
import random
|
||
import re
|
||
import time
|
||
|
||
from tasks.base import BaseTask, register_task
|
||
from core.device_worker import BaseWorker, _update_status
|
||
from core.u2_helper import (ensure_app_running, wait_for_app_home, random_sleep,
|
||
screen_is_on, current_package)
|
||
from core.logger import get_logger
|
||
from core import notifier, step_log, humanize, patrol, gesture
|
||
|
||
_log = get_logger("task.generic")
|
||
|
||
# 历史抓取器生成的"伪序号"选择器：//*[@resource-id="x"][4]
|
||
# XPath 里这是"父节点内排第 4"，不是"第 4 个匹配"——同属性多实例时 [2..n] 全部失配。
|
||
_LEGACY_IDX_XPATH = re.compile(r'^(//\*\[@[^\]]+\])\[(\d+)\](.*)$')
|
||
|
||
|
||
# ---- 「保持亮屏」用的系统设置 ----
|
||
# 系统息屏超时（毫秒）：保持亮屏时顶到最大，恢复时写回原值
|
||
_SCREEN_TIMEOUT_MAX = "2147483647"
|
||
_SCREEN_TIMEOUT_DEFAULT = "600000" # 兜底恢复值：10 分钟（原值记不住时用）
|
||
_SCREEN_TIMEOUT_BACKUP = {} # serial -> 原始 screen_off_timeout
|
||
|
||
|
||
def _norm_legacy_xpath(sel_val):
|
||
"""把历史 `//*[@attr=…][k]` 纠正为 `(//*[@attr=…])[k]`（只改整体前缀，保留后续子路径）。
|
||
|
||
窄范围：只匹配"属性XPath + 数字谓词"开头的形态；结构路径里
|
||
`.../FrameLayout[2]` 的兄弟序号是有意为之，不受影响。
|
||
"""
|
||
if not isinstance(sel_val, str):
|
||
return sel_val
|
||
m = _LEGACY_IDX_XPATH.match(sel_val.strip())
|
||
if not m:
|
||
return sel_val
|
||
return f"({m.group(1)})[{m.group(2)}]{m.group(3)}"
|
||
|
||
|
||
# ================== 步骤类型定义（前端操作库 + 后端执行共用）==================
|
||
STEP_TYPES = [
|
||
{"type": "open_app", "label": "打开App", "icon": "▶",
|
||
"params": {"package": "", "wait_home": False, "home_feature": ""}},
|
||
{"type": "stop_app", "label": "结束App", "icon": "⏹",
|
||
"params": {"package": ""}},
|
||
{"type": "screen_on", "label": "亮屏", "icon": "💡",
|
||
"params": {}},
|
||
{"type": "screen_off", "label": "息屏", "icon": "🌙",
|
||
"params": {}},
|
||
{"type": "keep_screen", "label": "保持亮屏", "icon": "🔆",
|
||
"params": {"mode": "on"}},
|
||
{"type": "key_event", "label": "按键", "icon": "⌨️",
|
||
"params": {"key": "back"}},
|
||
{"type": "swipe", "label": "滑动", "icon": "↕",
|
||
"params": {"direction": "up", "duration_min": 0.25, "duration_max": 0.50,
|
||
"distance_ratio": 0.6, "jitter": 0.15, "humanize": True}},
|
||
{"type": "swipe_until", "label": "滑动直到元素", "icon": "🔍",
|
||
"params": {"selector_type": "xpath", "selector_value": "", "direction": "up",
|
||
"max_swipes": 8, "click_when_found": True,
|
||
"duration_min": 0.25, "duration_max": 0.50,
|
||
"distance_ratio": 0.6, "jitter": 0.15, "humanize": True}},
|
||
{"type": "click", "label": "点击元素", "icon": "✦",
|
||
"params": {"selector_type": "description", "selector_value": "", "wait_timeout": 2}},
|
||
{"type": "click_xy", "label": "点击坐标", "icon": "🎯",
|
||
"params": {"x": 50, "y": 50}},
|
||
{"type": "long_click", "label": "长按元素", "icon": "👆",
|
||
"params": {"selector_type": "xpath", "selector_value": "", "duration": 1.0, "wait_timeout": 2}},
|
||
{"type": "wait_el", "label": "等待元素", "icon": "⏳",
|
||
"params": {"selector_type": "xpath", "selector_value": "", "timeout": 10}},
|
||
{"type": "input_text", "label": "输入文字", "icon": "⌨",
|
||
"params": {"mode": "random", "texts": "你好\n有趣\n支持", "fixed_text": "", "clear_first": True}},
|
||
{"type": "clipboard", "label": "剪贴板注入", "icon": "📋",
|
||
"params": {"text": "", "paste": True}},
|
||
{"type": "wait", "label": "等待", "icon": "⏱",
|
||
"params": {"min": 1.0, "max": 3.0}},
|
||
{"type": "loop", "label": "循环块", "icon": "↻",
|
||
"params": {"loop_mode": "rounds", "max_iterations": 10, "loop_duration": 600, "children": []}},
|
||
{"type": "group", "label": "动作组", "icon": "📦",
|
||
"params": {"children": []}},
|
||
{"type": "if_el", "label": "条件判断", "icon": "❓",
|
||
"params": {"selector_type": "xpath", "selector_value": "", "timeout": 3,
|
||
"cmp_op": "", "cmp_value": "", "then": [], "else": []}},
|
||
{"type": "notify", "label": "发通知", "icon": "🔔",
|
||
"params": {"title": "", "message": "", "level": "info"}},
|
||
{"type": "stop_self", "label": "停止本设备", "icon": "⛔",
|
||
"params": {"reason": ""}},
|
||
{"type": "gesture", "label": "录制手势", "icon": "⏺",
|
||
"params": {"points": [], "speed": 1.0}},
|
||
]
|
||
|
||
|
||
DEFAULT_PARAMS = {
|
||
"max_duration": 0, # 最大运行时长（秒），0=不限时
|
||
# 注意：**没有默认 steps**。步骤由前端步骤编辑器编排产出，新建任务时从零开始拖；
|
||
# 缺 steps 的任务执行时会明确报错（见 run_task），不会静默空跑。
|
||
}
|
||
|
||
# 「条件判断」文本比对时读元素文本的超时（秒）：只读一次属性，正常几十毫秒
|
||
_TEXT_READ_TIMEOUT = 2
|
||
|
||
# 「条件判断」的文本比对运算符：中文为主（界面上就是这个），英文/符号也认，
|
||
# 方便 AI 建任务或手写 JSON 时不用猜用哪个词。
|
||
_CMP_ALIASES = {
|
||
"等于": ("等于", "==", "=", "eq", "equals", "equal"),
|
||
"不等于": ("不等于", "!=", "<>", "ne", "not_equal", "notequal"),
|
||
"包含": ("包含", "含", "contains", "in", "has"),
|
||
"不包含": ("不包含", "不含", "not_contains", "notcontains", "excludes"),
|
||
}
|
||
# 界面上给的运算符（顺序即下拉框顺序）；上面那张表另收了英文/符号别名。
|
||
# AI 草稿校验（core/task_draft.py）用它拦"运算符拼错 -> 永远不命中"。
|
||
CMP_OPS = tuple(_CMP_ALIASES)
|
||
|
||
|
||
def _canon_op(op):
|
||
"""运算符规范化成中文那个（认别名）；认不出返回 None。"""
|
||
key = (op or "").strip().lower()
|
||
for canon, aliases in _CMP_ALIASES.items():
|
||
if key in [a.lower() for a in aliases]:
|
||
return canon
|
||
return None
|
||
|
||
|
||
def split_cmp_values(raw):
|
||
"""把"比对的值"拆成多个候选（**任一命中即可**，即 OR 语义）。
|
||
|
||
分隔符只有两个：**换行** 和 `|`。
|
||
**刻意不拆英文/中文逗号**——要比对的文本本身就常常带逗号
|
||
（如 `抖音号：123,456`），拆了就永远匹配不上。界面上是一行一个。
|
||
"""
|
||
if raw is None:
|
||
return []
|
||
s = str(raw).replace("\r\n", "\n").replace("\r", "\n").replace("|", "\n")
|
||
return [v.strip() for v in s.split("\n") if v.strip()]
|
||
|
||
|
||
def _cmp_hit(got, op, want):
|
||
"""核心比对：返回 (是否命中, 命中的那个值/惹事的值)。
|
||
|
||
多个候选值之间是 **OR**（"这个号或那个号都算我"）：
|
||
· 等于 → 文本等于**任一**候选即命中
|
||
· 包含 → 文本含**任一**候选即命中
|
||
· 不等于 → 文本**不在**任何候选里才命中
|
||
· 不包含 → **一个**候选都不出现才命中
|
||
认不出的运算符、或候选全为空 → 一律不命中（"看不懂就算不命中"，比猜一个更好排查）。
|
||
"""
|
||
got = (got or "").strip()
|
||
wants = split_cmp_values(want)
|
||
canon = _canon_op(op)
|
||
if canon is None or not wants:
|
||
return False, ""
|
||
if canon == "等于":
|
||
hit = got in wants
|
||
return hit, (got if hit else "")
|
||
if canon == "包含":
|
||
for v in wants:
|
||
if v in got:
|
||
return True, v
|
||
return False, ""
|
||
if canon == "不等于":
|
||
return (got not in wants), ""
|
||
for v in wants: # 不包含
|
||
if v in got:
|
||
return False, v # 命中了"不该出现"的那个 -> 条件不成立
|
||
return True, ""
|
||
|
||
|
||
def _text_matches(got, op, want):
|
||
"""元素文本比对（布尔版，多处调用只看结果）。"""
|
||
return _cmp_hit(got, op, want)[0]
|
||
|
||
|
||
class GenericStepsWorker(BaseWorker):
|
||
"""通用步骤 worker：按 params.steps 顺序执行步骤链。
|
||
|
||
复用 BaseWorker 的设备生命周期（STF 占用/释放、u2 连接、异常、状态上报、stop）。
|
||
只实现 run_task，按步骤类型分发到 _exec_<type> 方法。
|
||
"""
|
||
|
||
def __init__(self, serial, params=None, ctx=None):
|
||
super().__init__(serial, params, ctx=ctx)
|
||
p = {**DEFAULT_PARAMS, **(self.params or {})}
|
||
self.max_duration = int(p.get("max_duration", 0))
|
||
self.steps = p.get("steps", [])
|
||
self._action_counts = {} # 步骤执行计数 {step_label: count}
|
||
self._miss_counts = {} # 选择器健康：selector -> 连续未命中次数
|
||
# 步骤路径：容器步骤（loop/group/if_el）进出时压栈，得到 "2.1.3" 这种位置，
|
||
# 写进「步骤明细」用来还原嵌套结构。worker 每台设备一个线程，实例级状态安全。
|
||
self._path = []
|
||
self._step_rows = 0 # 本次运行已记录的明细条数（封顶见 _record_step）
|
||
self._step_capped = False
|
||
# 公共巡检（任务级配置，独立于步骤画布；见 core/patrol.py）：
|
||
# _wid 内部标识，用来记"下次到点 / 上次命中"两个节奏
|
||
self._watchers = []
|
||
for i, w in enumerate(p.get("watchers") or []):
|
||
if not isinstance(w, dict) or not w.get("enabled", True):
|
||
continue
|
||
w = dict(w)
|
||
w["_wid"] = f"w{i}"
|
||
self._watchers.append(w)
|
||
self._watch_next = {} # wid -> 下次到点（缺省 0 = 任务一开始就先查一次）
|
||
self._watch_last = {} # wid -> 上次命中（冷却用）
|
||
|
||
# 某 click 选择器连续未找到元素的次数达到该值，判定可能失效（App 改版）
|
||
_MAX_CONSECUTIVE_MISS = 10
|
||
|
||
def _track_selector_health(self, sel_val, found):
|
||
"""记录某选择器的连续未命中次数，达到阈值上报 last_warning 并重置计数。
|
||
|
||
用于发现"选择器失效但任务仍显示成功"的静默空转问题（如 App 改版后元素找不到了）。
|
||
"""
|
||
if found:
|
||
self._miss_counts.pop(sel_val, None)
|
||
return
|
||
n = self._miss_counts.get(sel_val, 0) + 1
|
||
self._miss_counts[sel_val] = n
|
||
if n >= self._MAX_CONSECUTIVE_MISS:
|
||
_log.error(f"[{self.serial}] 选择器连续 {n} 次未找到元素，可能已失效（App 改版？）：{sel_val}")
|
||
_update_status(self.serial, last_warning=f"选择器连续{n}次未命中")
|
||
self._miss_counts.pop(sel_val, None) # 已上报，重置避免刷屏
|
||
# 通知：这是"任务显示成功但什么都没做"的隐蔽故障，值得人看一眼
|
||
# （重置计数后要再攒满 10 次才会再报，天然不刷屏）
|
||
notifier.notify("task.selector.invalid", serial=self.serial,
|
||
selector=str(sel_val)[:200], miss_count=n)
|
||
|
||
def run_task(self, d):
|
||
"""按 steps 顺序执行。loop 步骤递归执行其 children。"""
|
||
if not self.steps:
|
||
# 任务没配步骤（DEFAULT_PARAMS 不含默认步骤，步骤只能由编辑器产出）：
|
||
# 明确报错而不是"跑完 0 步"当成功——设备表/任务页会显示该错误
|
||
raise ValueError("通用步骤任务没有可执行步骤：请在「任务」页编辑该任务并添加步骤")
|
||
self._start_timer()
|
||
self.set_action("开始执行步骤链")
|
||
self.set_progress(done=0, total=0, unit="操作",
|
||
action_counts=dict(self._action_counts))
|
||
|
||
try:
|
||
self._exec_steps(d, self.steps)
|
||
finally:
|
||
elapsed = self.elapsed()
|
||
summary = f"完成步骤链，运行 {elapsed//60}m{elapsed%60}s"
|
||
_update_status(self.serial, current_action=summary)
|
||
|
||
def _exec_steps(self, d, steps, depth=0):
|
||
"""执行步骤列表。depth 防止无限嵌套。"""
|
||
if depth > 5:
|
||
_log.warning(f"[{self.serial}] 步骤嵌套深度超限(>5)，跳过")
|
||
return
|
||
for i, step in enumerate(steps):
|
||
if self.stopped() or self.is_time_up():
|
||
return
|
||
self._maybe_patrol(d) # 穿插：每步之前看一眼有没有巡检到点了
|
||
# 压入本步在兄弟里的序号（1 起）：容器步骤执行 children 时会在其后
|
||
# 继续追加，得到 "2.1" 这样的嵌套路径
|
||
self._path.append(str(i + 1))
|
||
try:
|
||
self._exec_one(d, step, depth)
|
||
finally:
|
||
self._path.pop()
|
||
|
||
def _exec_one(self, d, step, depth=0):
|
||
"""执行单个步骤。"""
|
||
stype = step.get("type", "")
|
||
label = step.get("label", stype)
|
||
params = step.get("params", {})
|
||
# 兼容历史选择器：//*[@attr=…][k] → (//*[@attr=…])[k]（XPath 位置谓词语义）
|
||
if params.get("selector_type") == "xpath" and params.get("selector_value"):
|
||
fixed = _norm_legacy_xpath(params["selector_value"])
|
||
if fixed != params["selector_value"]:
|
||
params = dict(params, selector_value=fixed)
|
||
_log.info(f"[{self.serial}] 选择器已纠正为: {fixed}")
|
||
# 概率触发：probability=100 必执行，<100 按百分比概率决定本次是否执行
|
||
prob = float(params.get("probability", 100))
|
||
if prob < 100 and random.random() * 100 > prob:
|
||
_log.info(f"[{self.serial}] 步骤 '{label}' 概率 {prob}% 未触发，跳过")
|
||
self._record_step(step, "skip", f"概率 {prob}% 未触发", 0)
|
||
return
|
||
self.set_action(f"执行: {label}")
|
||
if stype == "gesture": # 点列可能有几十个点，别整段甩进日志
|
||
_log.info(f"[{self.serial}] 步骤: {label}({stype}) "
|
||
f"{len(params.get('points') or [])} 个轨迹点")
|
||
else:
|
||
_log.info(f"[{self.serial}] 步骤: {label}({stype}) params={params}")
|
||
|
||
ret = None
|
||
# 结果归类：异常 > 未命中（handler 返回 False）> 正常
|
||
# （True/None 都算正常——None 是"没有判断语义"的步骤，如点击坐标）
|
||
result, detail = "ok", ""
|
||
t0 = time.time()
|
||
handler = getattr(self, f"_exec_{stype}", None)
|
||
if handler:
|
||
try:
|
||
ret = handler(d, params, depth)
|
||
if ret is False:
|
||
result, detail = "miss", "未命中/超时"
|
||
except Exception as e:
|
||
_log.error(f"[{self.serial}] 步骤 {label}({stype}) 异常: {e}")
|
||
result, detail = "error", str(e)
|
||
else:
|
||
_log.warning(f"[{self.serial}] 未知步骤类型: {stype}")
|
||
result, detail = "unknown", f"未知步骤类型: {stype}"
|
||
self._record_step(step, result, detail, int((time.time() - t0) * 1000))
|
||
# 容器步骤（loop/group/if_el）不计入操作计数，避免监控显示噪音
|
||
if stype not in ("loop", "group", "if_el"):
|
||
self._action_counts[label] = self._action_counts.get(label, 0) + 1
|
||
elapsed = self.elapsed()
|
||
# done=累计执行的操作次数（无固定总数），前端显示"已执行 N 次操作"
|
||
self.set_progress(done=sum(self._action_counts.values()),
|
||
total=0, unit="操作",
|
||
action_counts=dict(self._action_counts), elapsed=elapsed)
|
||
return ret # 命中结果（True/False/None），"测试此步骤"用
|
||
|
||
def _record_step(self, step, result, detail, duration_ms):
|
||
"""把这一步写进「步骤明细」（异步落库，不阻塞任务线程）。
|
||
|
||
单次运行条数封顶：`forever` 循环任务会执行成千上万步，不封顶这张表
|
||
会被瞬间写爆。到顶后只补一条 cap 说明行，此后本次运行不再记录。
|
||
"""
|
||
try:
|
||
# 只记"任务运行"里的步骤：编辑器里的「测试此步骤」也会走 _exec_one，
|
||
# 但没有 run_id（不归任何一次运行），记进来只是噪音
|
||
if not self.ctx.get("run_id") or self._step_capped:
|
||
return
|
||
params = step.get("params") or {}
|
||
label = step.get("label") or step.get("type", "")
|
||
if self._step_rows >= step_log.MAX_ROWS_PER_RUN:
|
||
self._step_capped = True
|
||
step_log.record(
|
||
run_id=self.ctx.get("run_id", ""), serial=self.serial,
|
||
device_name=self.ctx.get("device_name", ""),
|
||
job_id=self.ctx.get("job_id", ""),
|
||
job_name=self.ctx.get("job_name", ""),
|
||
step_path=".".join(self._path), step_label=label,
|
||
step_type=step.get("type", ""), result="cap",
|
||
detail=f"本次运行明细已达上限 {step_log.MAX_ROWS_PER_RUN} 条，"
|
||
f"后续步骤不再记录（保留期与上限见 core/step_log.py）")
|
||
return
|
||
self._step_rows += 1
|
||
step_log.record(
|
||
run_id=self.ctx.get("run_id", ""), serial=self.serial,
|
||
device_name=self.ctx.get("device_name", ""),
|
||
job_id=self.ctx.get("job_id", ""),
|
||
job_name=self.ctx.get("job_name", ""),
|
||
step_path=".".join(self._path), step_label=label,
|
||
step_type=step.get("type", ""),
|
||
selector=str(params.get("selector_value", "") or ""),
|
||
result=result, detail=detail, duration_ms=duration_ms)
|
||
except Exception:
|
||
pass # 记录失败绝不能影响任务执行
|
||
|
||
# ================== 公共巡检（任务级，穿插执行） ==================
|
||
def _maybe_patrol(self, d):
|
||
"""看一眼有没有巡检到点了；到点就跑一次检查（配置见 core/patrol.py）。
|
||
|
||
穿插式：不起线程、不抢屏幕，所以巡检精度受主流程步长影响——某一步卡
|
||
30 秒，巡检最多晚 30 秒（这是刻意取舍，换来的是绝不会和主流程打架）。
|
||
"""
|
||
if not self._watchers or self.stopped():
|
||
return
|
||
now = time.time()
|
||
for w in self._watchers:
|
||
wid = w["_wid"]
|
||
if now < self._watch_next.get(wid, 0):
|
||
continue
|
||
# 先排下一次再执行：检查本身可能慢（元素检查要 dump UI 树），
|
||
# 否则慢检查会导致"每分钟"变成"每检查一次"
|
||
self._watch_next[wid] = now + max(5, int(w.get("interval") or 60))
|
||
try:
|
||
self._run_patrol(d, w, now)
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 巡检[{w.get('name') or wid}]异常: {e}")
|
||
|
||
def _run_patrol(self, d, w, now):
|
||
"""执行一次巡检：命中 → 过冷却 → 做动作 + 发通知 + 记一条明细。"""
|
||
wid, name = w["_wid"], (w.get("name") or patrol.check_label(w.get("check")))
|
||
hit, detail = patrol.evaluate(d, w)
|
||
if not hit:
|
||
return
|
||
# 命中后冷却：条件持续成立（比如一直熄屏）时不要每分钟都动作/发通知
|
||
cooldown = max(0, int(w.get("cooldown") or 0))
|
||
if now - self._watch_last.get(wid, 0) < cooldown:
|
||
_log.info(f"[{self.serial}] 巡检[{name}]命中（{detail}）但在冷却期内，不重复动作")
|
||
return
|
||
self._watch_last[wid] = now
|
||
# 文案在**动作之前**渲染：动作会改变设备状态（点亮后 {screen} 就成了"亮屏"），
|
||
# 而用户要看的是"发现时"的样子——"发现熄屏，已点亮"而不是"发现亮屏，已点亮"
|
||
title = self._render_vars(d, w.get("title") or "") or f"巡检命中：{name}"
|
||
message = self._render_vars(d, w.get("message") or "") or detail
|
||
done = patrol.act(d, w, self)
|
||
_log.info(f"[{self.serial}] 巡检[{name}]命中：{detail}"
|
||
f"{(' → ' + done) if done else '（只通知）'}")
|
||
self.set_action(f"巡检[{name}]：{detail}{('，' + done) if done else ''}")
|
||
summary = f"{detail}{(' → ' + done) if done else ''}"
|
||
try: # 明细：只记命中，不记"没事发生"
|
||
step_log.record(run_id=self.ctx.get("run_id", ""), serial=self.serial,
|
||
device_name=self.ctx.get("device_name", ""),
|
||
job_id=self.ctx.get("job_id", ""),
|
||
job_name=self.ctx.get("job_name", ""),
|
||
step_label=f"巡检[{name}]", step_type="patrol",
|
||
result="hit", detail=summary)
|
||
except Exception:
|
||
pass
|
||
if w.get("notify", True):
|
||
notifier.notify("task.patrol.hit",
|
||
patrol_name=name,
|
||
check_label=patrol.check_label(w.get("check")),
|
||
action_label=patrol.action_label(w.get("action")),
|
||
detail=detail,
|
||
job_id=self.ctx.get("job_id", ""),
|
||
job_name=self.ctx.get("job_name", ""),
|
||
serial=self.serial,
|
||
device_name=self.ctx.get("device_name", ""),
|
||
title=title, message=message,
|
||
level=w.get("level", "warning"))
|
||
|
||
# ================== 步骤执行器 ==================
|
||
def _exec_screen_on(self, d, params, depth=0):
|
||
"""亮屏：息屏时唤醒并滑动解锁。任务执行前用（息屏时 u2 无法操作）。"""
|
||
try:
|
||
d.unlock()
|
||
_log.info(f"[{self.serial}] 亮屏")
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 亮屏异常: {e}")
|
||
|
||
def _exec_screen_off(self, d, params, depth=0):
|
||
"""息屏：任务结束后用，避免长时间亮屏（烧屏/发热）。"""
|
||
try:
|
||
d.screen_off()
|
||
_log.info(f"[{self.serial}] 息屏")
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 息屏异常: {e}")
|
||
|
||
def _exec_keep_screen(self, d, params, depth=0):
|
||
"""保持亮屏 / 恢复自动息屏。
|
||
|
||
⚠️ 只用 `svc power stayon true` **不管用**：它管的是「**充电时**屏幕常亮」
|
||
（= stay_on_while_plugged_in）。设备走 WiFi 跑任务、没插充电器时它完全不起作用
|
||
——以前"保持亮屏"看着像没生效就是这个原因（2026-09-14 用户反馈）。
|
||
|
||
真正管用的是把系统**息屏超时**顶到最大（`screen_off_timeout`）：不充电也不会
|
||
自动黑屏。进入时记下原值，`mode=off` 写回；进程重启丢了记录就写回常规的 10 分钟。
|
||
"""
|
||
mode = params.get("mode", "on")
|
||
try:
|
||
if mode == "off":
|
||
d.shell("svc power stayon false")
|
||
orig = _SCREEN_TIMEOUT_BACKUP.pop(self.serial, _SCREEN_TIMEOUT_DEFAULT)
|
||
d.shell(f"settings put system screen_off_timeout {orig}")
|
||
_log.info(f"[{self.serial}] 恢复自动息屏（超时 {orig}ms）")
|
||
else:
|
||
cur = ""
|
||
try:
|
||
cur = (d.shell("settings get system screen_off_timeout") or "").strip()
|
||
except Exception:
|
||
pass
|
||
if self.serial not in _SCREEN_TIMEOUT_BACKUP:
|
||
# 没设过时 `settings get` 会返回 "null"（走系统默认）——别把 null 写回去
|
||
_SCREEN_TIMEOUT_BACKUP[self.serial] = \
|
||
cur if cur.isdigit() else _SCREEN_TIMEOUT_DEFAULT
|
||
d.shell("svc power stayon true") # 充电场景（顺带）
|
||
d.shell(f"settings put system screen_off_timeout {_SCREEN_TIMEOUT_MAX}")
|
||
d.shell("input keyevent 224") # 立刻唤醒（KEYCODE_WAKEUP，单向）
|
||
_log.info(f"[{self.serial}] 保持亮屏（息屏超时→最大，原值 "
|
||
f"{_SCREEN_TIMEOUT_BACKUP.get(self.serial)}ms）")
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] keep_screen({mode}) 异常: {e}")
|
||
|
||
def _exec_stop_app(self, d, params, depth=0):
|
||
"""强制结束 App（am force-stop）。清后台不留进程，下次打开为冷启动。"""
|
||
package = params.get("package", "")
|
||
if not package:
|
||
_log.warning(f"[{self.serial}] stop_app 缺少 package 参数")
|
||
return
|
||
try:
|
||
d.app_stop(package)
|
||
_log.info(f"[{self.serial}] 已强制结束 {package}")
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 结束 {package} 异常: {e}")
|
||
|
||
def _exec_open_app(self, d, params, depth=0):
|
||
package = params.get("package", "")
|
||
if not package:
|
||
_log.warning("open_app 缺少 package 参数")
|
||
return
|
||
wait_home = params.get("wait_home", False)
|
||
home_feature = params.get("home_feature", "")
|
||
d.app_start(package, wait=True)
|
||
if wait_home:
|
||
home_check = (lambda d: d(descriptionContains=home_feature).exists) if home_feature else (lambda d: True)
|
||
if not wait_for_app_home(d, package, home_check, timeout=40):
|
||
self.set_action(f"{package} 首页加载超时，继续")
|
||
|
||
def _exec_key_event(self, d, params, depth=0):
|
||
"""按键：返回/Home/回车等（d.press）。退出评论、返回上一页必备。"""
|
||
key = params.get("key", "back")
|
||
try:
|
||
d.press(key)
|
||
_log.info(f"[{self.serial}] 按键: {key}")
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 按键 {key} 异常: {e}")
|
||
|
||
def _exec_swipe_until(self, d, params, depth=0):
|
||
"""滑动直到元素出现（最多 max_swipes 次），可选找到后点击。
|
||
|
||
养号核心动作：信息流刷到目标按钮（点赞/评论）再操作。
|
||
返回 True=找到, False=未找到, None=配置缺选择器。
|
||
"""
|
||
sel_type = params.get("selector_type", "xpath")
|
||
sel_val = params.get("selector_value", "")
|
||
direction = params.get("direction", "up")
|
||
max_swipes = max(1, int(params.get("max_swipes", 8)))
|
||
click_when_found = params.get("click_when_found", True)
|
||
if not sel_val:
|
||
_log.warning("swipe_until 缺少选择器")
|
||
return None
|
||
# 每次滑动都重新生成（弧线方向/幅度/时长都不一样），不用预算好的固定坐标
|
||
swipe_kw = dict(
|
||
serial=self.serial,
|
||
duration_min=float(params.get("duration_min", 0.25)),
|
||
duration_max=float(params.get("duration_max", 0.50)),
|
||
distance_ratio=float(params.get("distance_ratio", 0.6) or 0.6),
|
||
jitter=params.get("jitter", 0.15),
|
||
humanize=params.get("humanize", True) is not False)
|
||
for i in range(1, max_swipes + 1):
|
||
if self.stopped() or self.is_time_up():
|
||
return None
|
||
found = False
|
||
try:
|
||
if sel_type == "xpath":
|
||
el = d.xpath(sel_val)
|
||
found = el.wait(timeout=1)
|
||
else:
|
||
el = d(**{sel_type: sel_val})
|
||
found = el.exists(timeout=1)
|
||
except Exception:
|
||
found = False
|
||
if found:
|
||
self.set_action(f"滑动找到元素(第{i}次)")
|
||
if click_when_found:
|
||
try:
|
||
el.click()
|
||
_log.info(f"[{self.serial}] 滑动{i}次后找到并点击: {sel_val}")
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 找到但点击失败: {e}")
|
||
else:
|
||
_log.info(f"[{self.serial}] 滑动{i}次后找到: {sel_val}")
|
||
self._track_selector_health(sel_val, True)
|
||
return True
|
||
humanize.swipe(d, direction, **swipe_kw)
|
||
time.sleep(random.uniform(0.35, 0.75)) # 每次滑动后停顿也抖动一下
|
||
self._track_selector_health(sel_val, False)
|
||
_log.warning(f"[{self.serial}] 滑动{max_swipes}次未找到元素: {sel_val}")
|
||
return False
|
||
|
||
def _exec_click_xy(self, d, params, depth=0):
|
||
"""点击坐标（屏幕百分比 0-100，50/50=屏幕中心）。无选择器时兜底。"""
|
||
x_pct = float(params.get("x", 50))
|
||
y_pct = float(params.get("y", 50))
|
||
info = d.info
|
||
w, h = info["displayWidth"], info["displayHeight"]
|
||
x = int(w * min(max(x_pct, 0), 100) / 100)
|
||
y = int(h * min(max(y_pct, 0), 100) / 100)
|
||
d.click(x, y)
|
||
_log.info(f"[{self.serial}] 点击坐标 ({x},{y})")
|
||
|
||
def _exec_long_click(self, d, params, depth=0):
|
||
"""长按元素（选择器 + 时长秒）。复制链接/呼出菜单用。"""
|
||
sel_type = params.get("selector_type", "xpath")
|
||
sel_val = params.get("selector_value", "")
|
||
duration = float(params.get("duration", 1.0))
|
||
timeout = float(params.get("wait_timeout", 2))
|
||
if not sel_val:
|
||
_log.warning("long_click 缺少选择器")
|
||
return None
|
||
found = False
|
||
try:
|
||
if sel_type == "xpath":
|
||
el = d.xpath(sel_val)
|
||
if el.wait(timeout=timeout):
|
||
el.long_click(duration=duration)
|
||
found = True
|
||
else:
|
||
el = d(**{sel_type: sel_val})
|
||
if el.exists(timeout=timeout):
|
||
el.long_click(duration=duration)
|
||
found = True
|
||
self._track_selector_health(sel_val, found)
|
||
_log.info(f"[{self.serial}] 长按 {'命中' if found else '未找到'}: {sel_val}")
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 长按异常: {e}")
|
||
return found
|
||
|
||
def _exec_wait_el(self, d, params, depth=0):
|
||
"""等待元素出现（条件等待）。返回 True=出现, False=超时, None=缺选择器。"""
|
||
sel_type = params.get("selector_type", "xpath")
|
||
sel_val = params.get("selector_value", "")
|
||
timeout = float(params.get("timeout", 10))
|
||
if not sel_val:
|
||
_log.warning("wait_el 缺少选择器")
|
||
return None
|
||
try:
|
||
if sel_type == "xpath":
|
||
found = d.xpath(sel_val).wait(timeout=timeout)
|
||
else:
|
||
found = d(**{sel_type: sel_val}).exists(timeout=timeout)
|
||
self._track_selector_health(sel_val, found)
|
||
_log.info(f"[{self.serial}] 等待元素 {'出现' if found else '超时未出现'}: {sel_val}")
|
||
return found
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 等待元素异常: {e}")
|
||
return False
|
||
|
||
def _exec_if_el(self, d, params, depth=0):
|
||
"""条件判断：找到元素 → 执行 then 分支；超时未找到 → 执行 else 分支。
|
||
|
||
selector_type=ocr 时用屏幕 OCR 匹配文字（图片/画布/WebView 里的文字也能找到），
|
||
命中且 ocr_click=True 时自动点击该文字中心。
|
||
分支里可放任意步骤（含循环/嵌套条件判断），实现"有弹窗就关掉、没弹窗就继续"这类逻辑。
|
||
|
||
**文本比对（cmp_op / cmp_value，可选）**：光"元素在不在"不够用——常见需求是
|
||
"这台机器上登录的抖音号是不是我配的那个"。填了 cmp_op 就变成：
|
||
先按选择器**取到元素的文本**，再拿它和 cmp_value 比（等于/不等于/包含/不包含），
|
||
比对结果才是命中与否。例：`resourceId=com.ss.android.ugc.aweme:id/506` 的文本是
|
||
`抖音号：35377983067`，cmp_op=包含、cmp_value=35377983067 就是"这台上登录的是它"。
|
||
|
||
`cmp_value` **可以填多个值**（换行或 `|` 分隔，见 `split_cmp_values`），
|
||
多个之间是 **OR**——"这几个号都算我的"。要跑多个号的场景就靠它。
|
||
返回 True=命中 / False=未命中 / None=缺选择器或 OCR 不可用。
|
||
"""
|
||
sel_type = params.get("selector_type", "xpath")
|
||
sel_val = params.get("selector_value", "")
|
||
timeout = float(params.get("timeout", 3))
|
||
cmp_op = str(params.get("cmp_op") or "").strip()
|
||
cmp_val = str(params.get("cmp_value") or "")
|
||
if not sel_val:
|
||
_log.warning(f"[{self.serial}] if_el 缺少选择器，跳过")
|
||
return None
|
||
found = False
|
||
got = None # 比对用的"采集到的值"：元素文本 / OCR 命中文本 / 前台包名
|
||
# 非元素类条件（屏幕状态/前台 App）：不参与"选择器健康"统计——它们不是选择器，
|
||
# 连续未命中不该被记成"选择器失效"
|
||
dynamic = sel_type in ("screen", "foreground")
|
||
detail = ""
|
||
if sel_type == "screen":
|
||
# 屏幕亮/熄：selector_value 填 on / off（也认 熄屏/灭屏 这类中文）
|
||
want = str(sel_val).strip().lower()
|
||
want_off = want in ("off", "0", "false", "no", "熄屏", "灭屏", "黑屏")
|
||
state = screen_is_on(d)
|
||
found = (state is False) if want_off else (state is True)
|
||
detail = (f"屏幕{'熄屏' if state is False else '亮屏' if state else '状态未知'}"
|
||
f"（判断{'熄屏' if want_off else '亮屏'}）")
|
||
elif sel_type == "foreground":
|
||
cur = current_package(d)
|
||
got = cur or ""
|
||
found = cur == str(sel_val).strip()
|
||
detail = f"前台='{cur or '未知'}'（期望 {sel_val}）"
|
||
elif sel_type == "ocr":
|
||
# OCR 模式：截图 → 识别文字 → 关键词匹配（UI 树里没有的文字也能找到）
|
||
try:
|
||
from core.ocr import available as _ocr_available, find_on_screen
|
||
ok, hint = _ocr_available()
|
||
if not ok:
|
||
_log.error(f"[{self.serial}] OCR 条件判断不可用: {hint}")
|
||
return None
|
||
img = d.screenshot()
|
||
if img is None:
|
||
_log.warning(f"[{self.serial}] OCR 截图失败")
|
||
else:
|
||
found, center, matched = find_on_screen(img, sel_val)
|
||
got = matched or "" if found else ""
|
||
if found:
|
||
if params.get("ocr_click") and center:
|
||
d.click(*center)
|
||
_log.info(f"[{self.serial}] OCR 命中 '{matched}'，已点击 ({center[0]},{center[1]})")
|
||
else:
|
||
_log.info(f"[{self.serial}] OCR 命中 '{matched}'")
|
||
else:
|
||
_log.info(f"[{self.serial}] OCR 未命中: '{sel_val}'")
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] OCR 条件判断异常: {e}")
|
||
found = False
|
||
else:
|
||
try:
|
||
if sel_type == "xpath":
|
||
found = d.xpath(sel_val).wait(timeout=timeout)
|
||
else:
|
||
found = d(**{sel_type: sel_val}).exists(timeout=timeout)
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 条件判断异常: {e}")
|
||
found = False
|
||
|
||
# 元素"在不在"的原始结果：选择器健康统计要用它、**不能**用比对结果——
|
||
# "找到了但文本不对"是条件按预期走了 else 分支，不是选择器失效。
|
||
existed = found
|
||
|
||
# ---- 文本比对（可选）----
|
||
if cmp_op and sel_type != "screen":
|
||
if not found:
|
||
# 元素都没找到就没什么可比的：算未命中（走 else 分支）。
|
||
# 这条规则很重要——"没读到"永远不能被当成"和我设的不一样"，
|
||
# 否则界面还没加载出来时会误判成"账号被别人换了"。
|
||
nofind = f"未找到目标，无法比对（{cmp_op} {cmp_val}）"
|
||
detail = f"{detail}；{nofind}" if detail else nofind
|
||
else:
|
||
if got is None:
|
||
got = self._read_element_text(d, sel_type, sel_val)
|
||
hit, hit_val = _cmp_hit(got, cmp_op, cmp_val)
|
||
n_val = len(split_cmp_values(cmp_val))
|
||
if hit:
|
||
extra = f"，命中『{hit_val}』" if hit_val else ""
|
||
else:
|
||
extra = f"，因为出现了『{hit_val}』" if hit_val else ""
|
||
detail = (f"采集到 '{got}' {cmp_op} {n_val} 个候选值 → "
|
||
f"{'符合' if hit else '不符合'}{extra}")
|
||
found = hit
|
||
elif cmp_op and sel_type == "screen":
|
||
_log.warning(f"[{self.serial}] 屏幕状态没有文本可比，已忽略文本比对设置")
|
||
|
||
if dynamic:
|
||
_log.info(f"[{self.serial}] 条件判断 {detail} → {'命中' if found else '未命中'}")
|
||
else:
|
||
self._track_selector_health(sel_val, existed)
|
||
branch = params.get("then" if found else "else") or []
|
||
if cmp_op:
|
||
cmp_txt = "比对命中" if found else "比对未命中"
|
||
else:
|
||
cmp_txt = "命中" if found else "未命中"
|
||
self.set_action(f"条件{cmp_txt} → 执行{'then' if found else 'else'}分支({len(branch)}步)")
|
||
_log.info(f"[{self.serial}] 条件判断 {'命中' if found else '未命中'}: {detail or sel_val}，执行{'then' if found else 'else'}分支 {len(branch)} 步")
|
||
self._exec_steps(d, branch, depth + 1)
|
||
return found
|
||
|
||
def _read_element_text(self, d, sel_type, sel_val):
|
||
"""取选中元素的文本（文本比对用）。读不到返回 ""。
|
||
|
||
只读操作，不改任何状态；失败静默返回空串，由调用方按"空文本"去比对
|
||
（日志里会打出实际读到的值，排查时能看出是空还是不对）。
|
||
|
||
注意 u2 两种元素的 exists 语义不同，别"统一"掉：
|
||
· `d.xpath(...)` → XPathSelector.exists 是**纯 bool 属性**
|
||
· `d(text=...)` → UiObject.exists 是**可调用的 Exists 对象**
|
||
（`bool()` 即时判断、`exists(timeout=t)` 带等待）——两种写法都对
|
||
"""
|
||
try:
|
||
if sel_type == "xpath":
|
||
el = d.xpath(sel_val)
|
||
if not el.exists:
|
||
return ""
|
||
return (el.get_text() or "").strip()
|
||
el = d(**{sel_type: sel_val})
|
||
if not el.exists:
|
||
return ""
|
||
# 限时读：元素在"检查"与"读"之间消失时，别挂满默认等待（默认可能是 10s）
|
||
return (el.get_text(timeout=_TEXT_READ_TIMEOUT) or "").strip()
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] 读取元素文本失败（按空文本比对）: {e}")
|
||
return ""
|
||
|
||
def _exec_swipe(self, d, params, depth=0):
|
||
"""滑动一次（默认拟人：弧线 + 抖动 + 本设备手速，见 core/humanize.py）。
|
||
|
||
老行为（正中直线、每次都一样）可以用 humanize=false 还原；
|
||
jitter=0 则保留轨迹但去掉随机抖动（对照排障用）。
|
||
"""
|
||
direction = params.get("direction", "up")
|
||
info = humanize.swipe(
|
||
d, direction, serial=self.serial,
|
||
duration_min=float(params.get("duration_min", 0.25)),
|
||
duration_max=float(params.get("duration_max", 0.50)),
|
||
distance_ratio=float(params.get("distance_ratio", 0.6) or 0.6),
|
||
jitter=params.get("jitter", 0.15),
|
||
humanize=params.get("humanize", True) is not False)
|
||
_log.info(f"[{self.serial}] 滑动 {direction} 时长{info['duration']}s "
|
||
f"({info['from']}→{info['to']}"
|
||
f"{'，弧线' if info.get('human') else '，直线'})")
|
||
|
||
def _exec_notify(self, d, params, depth=0):
|
||
"""发一条自定义通知（标题/正文由任务自己写，支持占位符）。
|
||
|
||
事件是 `task.notify.custom`：**谁能收到取决于 webhook 的事件订阅**——
|
||
这个步骤只负责"把消息交出去"，不关心发给谁（与平台其它通知同一套口径）。
|
||
"""
|
||
title = self._render_vars(d, params.get("title") or "")
|
||
msg = self._render_vars(d, params.get("message") or "")
|
||
if not (title or msg):
|
||
_log.warning(f"[{self.serial}] notify 步骤既没标题也没内容，跳过")
|
||
return None
|
||
notifier.notify("task.notify.custom",
|
||
job_id=self.ctx.get("job_id", ""),
|
||
job_name=self.ctx.get("job_name", ""),
|
||
serial=self.serial,
|
||
device_name=self.ctx.get("device_name", ""),
|
||
title=title or "任务通知", message=msg,
|
||
level=params.get("level", "info"))
|
||
_log.info(f"[{self.serial}] 已发自定义通知: {title} {msg}")
|
||
return True
|
||
|
||
def _render_vars(self, d, text):
|
||
"""替换通知文本里的占位符：{device} {serial} {job} {time} {app} {screen}。
|
||
|
||
`{app}` / `{screen}` 要查设备（0.3~0.7s），**只在文本里真用到时才查**，
|
||
否则每条通知都白付一次设备往返。
|
||
"""
|
||
if not text or "{" not in text:
|
||
return text
|
||
vals = {"device": self.ctx.get("device_name") or self.serial,
|
||
"serial": self.serial,
|
||
"job": self.ctx.get("job_name", ""),
|
||
"time": time.strftime("%Y-%m-%d %H:%M:%S")}
|
||
if "{app}" in text:
|
||
vals["app"] = current_package(d) or "未知"
|
||
if "{screen}" in text:
|
||
st = screen_is_on(d)
|
||
vals["screen"] = "未知" if st is None else ("亮屏" if st else "熄屏")
|
||
for k, v in vals.items():
|
||
text = text.replace("{" + k + "}", str(v))
|
||
return text
|
||
|
||
def _exec_stop_self(self, d, params, depth=0):
|
||
"""停止本设备的任务（其它设备的任务不受影响）。
|
||
|
||
实现只是置 worker 的停止位：`_exec_steps` 每步前检查 `stopped()`，所以
|
||
后续步骤不会再跑；TaskManager 侧把它记成"被停止"而不是失败
|
||
（见 `core/task_manager.py` 的 `_run_with_retry` 自停分支）。
|
||
"""
|
||
reason = (params.get("reason") or "").strip()
|
||
_log.info(f"[{self.serial}] 任务内主动停止本设备（{reason or '未填原因'}）")
|
||
self.set_action(f"停止本设备任务{('：' + reason) if reason else ''}")
|
||
self.stop()
|
||
return True
|
||
|
||
def _exec_gesture(self, d, params, depth=0):
|
||
"""回放**录制**的手势（路径与时长照录制，不做任何再加工）。
|
||
|
||
与「滑动」步骤是两套东西：滑动每次重新生成（弧线/抖动/本设备手速），
|
||
这里只把录下来的点列原样交给设备插值（原因见 core/gesture.py 模块头）。
|
||
"""
|
||
pts = params.get("points") or []
|
||
if len(pts) < 2:
|
||
_log.warning(f"[{self.serial}] gesture 步骤没有有效轨迹（points 为空），跳过")
|
||
return None
|
||
info = gesture.replay(d, pts, speed=params.get("speed", 1.0))
|
||
_log.info(f"[{self.serial}] 录制回放：{info['points']} 个点 / {info['duration']}s")
|
||
return True
|
||
|
||
def _exec_click(self, d, params, depth=0):
|
||
sel_type = params.get("selector_type", "xpath")
|
||
sel_val = params.get("selector_value", "")
|
||
timeout = float(params.get("wait_timeout", 2))
|
||
if not sel_val:
|
||
_log.warning(f"[{self.serial}] click 跳过：selector_value 为空，请在步骤编辑器中填写或抓取元素")
|
||
return
|
||
# 兼容旧配置：selector_type 不是 xpath 但值是 XPath（以 // 开头）
|
||
if sel_type != "xpath" and sel_val.strip().startswith("//"):
|
||
sel_type = "xpath"
|
||
found = False
|
||
try:
|
||
if sel_type == "xpath":
|
||
# u2 的 d.xpath() 返回 XPathSelector，exists 是属性需用 wait()
|
||
el = d.xpath(sel_val)
|
||
if el.wait(timeout=timeout):
|
||
el.click()
|
||
_log.info(f"[{self.serial}] click 命中: xpath={sel_val}")
|
||
found = True
|
||
else:
|
||
_log.warning(f"[{self.serial}] click 未找到元素: xpath={sel_val} (等待{timeout}s)")
|
||
else:
|
||
# UiSelector 风格：d(description=xxx) 等，exists() 支持 timeout 参数
|
||
el = d(**{sel_type: sel_val})
|
||
if el.exists(timeout=timeout):
|
||
el.click()
|
||
_log.info(f"[{self.serial}] click 命中: {sel_type}={sel_val}")
|
||
found = True
|
||
else:
|
||
_log.warning(f"[{self.serial}] click 未找到元素: {sel_type}={sel_val} (等待{timeout}s)")
|
||
self._track_selector_health(sel_val, found)
|
||
except Exception as e:
|
||
_log.warning(f"[{self.serial}] click 异常: {e}")
|
||
return found # 命中结果，"测试此步骤"用
|
||
|
||
def _exec_clipboard(self, d, params, depth=0):
|
||
"""剪贴板注入（ClipInject 通道，Android 10+ 可用），可选立即粘贴。
|
||
|
||
批量发链接场景：先聚焦评论区输入框 → 注入+粘贴链接1 → 换链接 → 注入+粘贴链接2……
|
||
粘贴模式：ClipInject 写入剪贴板后调 atx-agent pasteClipboard 粘贴到当前
|
||
焦点输入框；动作失败时兜底 u2 set_text 直接设置聚焦输入框文本（不依赖
|
||
剪贴板，与"输入文字"兜底同机制）。
|
||
"""
|
||
text = str(params.get("text", "") or "")
|
||
if not text:
|
||
_log.warning(f"[{self.serial}] clipboard 内容为空，跳过")
|
||
return False
|
||
from core.clipboard_helper import inject_clipboard
|
||
ok, msg = inject_clipboard(self.serial, text, d=d)
|
||
if not ok:
|
||
_log.warning(f"[{self.serial}] 剪贴板注入失败: {msg}")
|
||
return False
|
||
_log.info(f"[{self.serial}] 剪贴板注入: {text[:60]}（{msg}）")
|
||
if not params.get("paste"):
|
||
return True
|
||
try:
|
||
d.jsonrpc.pasteClipboard()
|
||
_log.info(f"[{self.serial}] 已粘贴: {text[:60]}")
|
||
return True
|
||
except Exception as e:
|
||
try:
|
||
d(focused=True).set_text(text)
|
||
_log.info(f"[{self.serial}] 粘贴兜底 set_text: {text[:60]}")
|
||
return True
|
||
except Exception as e2:
|
||
_log.warning(f"[{self.serial}] 粘贴失败: {e} / {e2}")
|
||
return True # 剪贴板已写入，粘贴失败不致命
|
||
|
||
def _exec_wait(self, d, params, depth=0):
|
||
min_s = float(params.get("min", 1.0))
|
||
max_s = float(params.get("max", 3.0))
|
||
# 分片 sleep：每 0.5s 检查停止/超时信号，抢占任务能及时接管（纯 sleep 无法中断）
|
||
total = random.uniform(min_s, max_s)
|
||
# 可选：按本设备节奏缩放（勾了"设备节奏差异"才生效，默认关——
|
||
# 时长被放大/缩小是业务语义，不能悄悄改）
|
||
total = humanize.pace(self.serial, total, enabled=params.get("vary_pace", False))
|
||
end = time.time() + total
|
||
while time.time() < end:
|
||
if self.stopped() or self.is_time_up():
|
||
return
|
||
self._maybe_patrol(d) # 长等待里也看一眼，别让巡检被 wait 拖住
|
||
# max(0, …)：巡检本身要花时间（屏幕 0.3s、前台 0.7s、元素检查更久），
|
||
# 扣掉这段时间后可能已经过点了，负数 sleep 会直接抛异常
|
||
time.sleep(max(0.0, min(0.5, end - time.time())))
|
||
|
||
def _exec_loop(self, d, params, depth=0):
|
||
children = params.get("children", [])
|
||
if not children:
|
||
return
|
||
mode = params.get("loop_mode", "rounds")
|
||
if mode == "forever":
|
||
# 一直循环：无限跑直到任务被外部停止（定时停止 cron / 运行时长上限 / 手动停止）。
|
||
# 配合"定时启动+停止"调度（如 9:00 启动 / 18:00 停止）实现全天候运行。
|
||
i = 0
|
||
while not self.stopped() and not self.is_time_up():
|
||
i += 1
|
||
self.set_action(f"循环第 {i} 轮(持续)")
|
||
if i % 10 == 0 or i <= 2:
|
||
_log.info(f"[{self.serial}] loop 第 {i} 轮(持续运行)")
|
||
self._exec_steps(d, children, depth + 1)
|
||
elif mode == "time":
|
||
# 按时间循环：跑满 loop_duration 秒，同时受全局 stopped/max_duration 约束
|
||
duration = float(params.get("loop_duration", 0) or 0)
|
||
if duration <= 0:
|
||
_log.warning(f"[{self.serial}] 按时间循环未设置时长，跳过")
|
||
return
|
||
start = time.time()
|
||
i = 0
|
||
while not self.stopped() and not self.is_time_up() and (time.time() - start) < duration:
|
||
i += 1
|
||
self.set_action(f"循环第 {i} 轮(按时间)")
|
||
_log.info(f"[{self.serial}] loop 第 {i} 轮(按时间, 已{(time.time()-start):.0f}s/{duration:.0f}s)")
|
||
self._exec_steps(d, children, depth + 1)
|
||
else:
|
||
# 按轮次循环
|
||
max_iter = int(params.get("max_iterations", 10))
|
||
for i in range(max_iter):
|
||
if self.stopped() or self.is_time_up():
|
||
return
|
||
self.set_action(f"循环第 {i+1}/{max_iter} 轮")
|
||
_log.info(f"[{self.serial}] loop 第 {i+1}/{max_iter} 轮")
|
||
self._exec_steps(d, children, depth + 1)
|
||
|
||
def _exec_group(self, d, params, depth=0):
|
||
"""动作组：按序执行 children 一次（类似 loop 但不循环）。
|
||
|
||
用于把一系列步骤打包成一个可复用的动作。
|
||
"""
|
||
children = params.get("children", [])
|
||
if not children:
|
||
return
|
||
self._exec_steps(d, children, depth + 1)
|
||
|
||
def _exec_input_text(self, d, params, depth=0):
|
||
"""在当前焦点输入框输入文字。
|
||
|
||
需先用"点击元素"步骤定位到输入框，本步骤只负责输入。
|
||
mode=random: 从 texts（换行分隔）随机选一条
|
||
mode=fixed: 输入 fixed_text
|
||
"""
|
||
mode = params.get("mode", "random")
|
||
if mode == "fixed":
|
||
text = params.get("fixed_text", "")
|
||
else:
|
||
raw = params.get("texts", "")
|
||
candidates = [t.strip() for t in raw.splitlines() if t.strip()]
|
||
if not candidates:
|
||
_log.warning("input_text 无候选文字")
|
||
return
|
||
text = random.choice(candidates)
|
||
if not text:
|
||
return
|
||
clear_first = params.get("clear_first", True)
|
||
try:
|
||
# 优先用 u2 的 send_keys（对当前焦点元素输入）；clear=True 先清空再输入
|
||
d.send_keys(text, clear=clear_first)
|
||
_log.info(f"[{self.serial}] input_text: {text}" + ("（已先清空）" if clear_first else ""))
|
||
except Exception as e:
|
||
# 兜底：找 EditText 设置文本
|
||
try:
|
||
edit = d(className="android.widget.EditText")
|
||
if edit.exists:
|
||
if clear_first:
|
||
edit.clear_text()
|
||
edit.set_text(text)
|
||
_log.info(f"[{self.serial}] input_text(set_text): {text}")
|
||
else:
|
||
_log.warning(f"[{self.serial}] input_text 未找到输入框: {e}")
|
||
except Exception as e2:
|
||
_log.warning(f"[{self.serial}] input_text 失败: {e2}")
|
||
|
||
|
||
@register_task
|
||
class GenericStepsTask(BaseTask):
|
||
"""通用步骤任务：前端编辑器编排步骤链，worker 按顺序执行。"""
|
||
task_type = "generic_steps"
|
||
name = "通用步骤"
|
||
description = "通过步骤编辑器编排任务流程，支持循环/点击/滑动/点赞/评论等"
|
||
default_params = dict(DEFAULT_PARAMS)
|
||
|
||
@classmethod
|
||
def list_action_types(cls):
|
||
"""返回支持的步骤类型列表（前端操作库用）。"""
|
||
return STEP_TYPES
|
||
|
||
@classmethod
|
||
def get_action_class(cls, action_type):
|
||
return None
|
||
|
||
def create_worker(self, serial, params, ctx=None):
|
||
merged = {**DEFAULT_PARAMS, **(params or {})}
|
||
return GenericStepsWorker(serial, params=merged, ctx=ctx)
|
||
|
||
|
||
def test_step(serial, step):
|
||
"""在设备上单步试执行（编辑器"测试此步骤"按钮用）。
|
||
|
||
直接 adb connect + u2 连接后执行单个步骤，验证选择器是否命中。
|
||
返回 (ok, msg, result)：result 为执行器返回的命中结果（True/False/None）。
|
||
与运行中的任务互不干扰（只读连接，不占用/不释放 STF）。
|
||
"""
|
||
import uiautomator2 as u2
|
||
from concurrent.futures import ThreadPoolExecutor # 与 device_worker 一致（本机 stdlib threading 无此属性）
|
||
from core.adb_helper import adb_connect
|
||
from core.device_worker import _U2_CONNECT_TIMEOUT
|
||
|
||
try:
|
||
if not adb_connect(serial):
|
||
return False, f"adb connect {serial} 失败", None
|
||
except Exception as e:
|
||
return False, f"adb connect 异常: {e}", None
|
||
try:
|
||
with ThreadPoolExecutor(max_workers=1) as pool:
|
||
fut = pool.submit(u2.connect, serial)
|
||
d = fut.result(timeout=_U2_CONNECT_TIMEOUT)
|
||
except Exception as e:
|
||
return False, f"u2 连接失败: {e}", None
|
||
if d is None:
|
||
return False, "u2 连接超时", None
|
||
try:
|
||
w = GenericStepsWorker(serial, {})
|
||
result = w._exec_one(d, step)
|
||
return True, f"步骤已执行（{step.get('label') or step.get('type')}）", result
|
||
except Exception as e:
|
||
return False, f"执行异常: {e}", None
|