"""通用步骤任务定义。 前端步骤编辑器编排步骤链，保存为 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 - 停止本设备的任务（其它设备不受影响） 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 _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, "then": [], "else": []}}, {"type": "notify", "label": "发通知", "icon": "🔔", "params": {"title": "", "message": "", "level": "info"}}, {"type": "stop_self", "label": "停止本设备", "icon": "⛔", "params": {"reason": ""}}, ] DEFAULT_PARAMS = { "max_duration": 0, # 最大运行时长（秒），0=不限时 # 注意：**没有默认 steps**。步骤由前端步骤编辑器编排产出，新建任务时从零开始拖； # 缺 steps 的任务执行时会明确报错（见 run_task），不会静默空跑。 } class GenericStepsWorker(BaseWorker): """通用步骤 worker：按 params.steps 顺序执行步骤链。 复用 BaseWorker 的设备生命周期（STF 占用/释放、u2 连接、异常、状态上报、stop）。 只实现 run_task，按步骤类型分发到 _exec_ 方法。 """ 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}") _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 时自动点击该文字中心。 分支里可放任意步骤（含循环/嵌套条件判断），实现"有弹窗就关掉、没弹窗就继续"这类逻辑。 返回 True=命中 / False=未命中 / None=缺选择器或 OCR 不可用。 """ sel_type = params.get("selector_type", "xpath") sel_val = params.get("selector_value", "") timeout = float(params.get("timeout", 3)) if not sel_val: _log.warning(f"[{self.serial}] if_el 缺少选择器，跳过") return None found = False # 非元素类条件（屏幕状态/前台 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) 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) 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 if dynamic: _log.info(f"[{self.serial}] 条件判断 {detail} → {'命中' if found else '未命中'}") else: self._track_selector_health(sel_val, found) branch = params.get("then" if found else "else") or [] self.set_action(f"条件{'命中' if found else '未命中'} → 执行{'找到' if found 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 _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_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