butubb
e82fd64695
用户提的（他说得对）："这个标记已做不是应该是我上面的条件判断命中哪一个就用哪一个做吗？
这怎么还要自己选择元素啊"。
原来的设计要求检查与记账**各配一次身份元素** —— 一旦一边填了、另一边忘了，两边算出的
key 不一致 → **去重静默失效**（现象就是"去重没生效、还是重复做"，他刚踩过）。
而我在"有效期"上为了避免两处配置特意强制放任务级，却在"身份"上要求填两遍，自相矛盾。
改成：
- `tasks/generic/task.py`：「去重」条件解析出的身份存进 `self._last_ident`；
`_exec_mark_done` 的身份元素**留空时复用它**（检查与记账必然是同一个字符串）。
三种情况分清（这是关键，别退化成"用设备"误标）：
① 自己填了身份元素 → 用自己填的（显式优先）
② 留空 + 上面跑过去重检查 → **复用那个身份**
③ 留空 + 检查**跑了但没读到身份**（页面没到）→ **不记账**，下次重跑重试
—— 绝不能退回设备身份，那会在"什么都没做成"时把设备标成已做
④ 留空 + 任务里根本没有去重检查 → 退回设备身份（一号一机场景）
- `static/admin/editor.js`：`mark_done` 面板文案改为"留空 = 自动跟随上面检查的身份（推荐）"；
校验里记账侧的**空值不再参与身份比对**（留空是合法的、且会自动对齐），
只有"填了、且和检查侧不一样"才告警，并在文案里建议留空。
- `doc/TASK_DEV.md` §4.6：身份只需在检查侧配一次（带一张四种组合的表）；步骤表同步。
验证：新增 11 项（复用后账本身份 == 检查身份、换设备仍能命中=去重真生效、
显式填身份仍优先、无检查时退回设备、**读不到身份时不记账也不误标设备**）；
原 dedup 套件（单元 + 真机）+ 编辑器语法回归全过；
把用户**生产任务的真实 JSON** 喂给新校验 —— **0 条警告**（旧版会报一条"身份不一致"，
但现在留空=自动跟随是正确配置）。
1212 lines
62 KiB
Python
1212 lines
62 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, dedup
|
||
|
||
_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": "",
|
||
"ident_type": "text", "ident_value": "",
|
||
"then": [], "else": []}},
|
||
{"type": "mark_done", "label": "记为已做", "icon": "✅",
|
||
"params": {"selector_type": "text", "selector_value": ""}},
|
||
{"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=不限时
|
||
# 去重账本的有效期（任务级，**只在这一处配**）。检查与记账两处各填一份的话，
|
||
# 填不一致就会算出两个不同的 key、去重静默失效——所以强制任务级。
|
||
# day（默认，每天一次）/ all（只做一次，永不清理）/ hours（每 dedup_hours 小时）
|
||
"dedup_reset": "day",
|
||
"dedup_hours": 6,
|
||
# 注意：**没有默认 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
|
||
# 「去重」条件最近一次解析出的身份：`mark_done` 留空时复用它，
|
||
# 保证"检查"与"记账"用的是**同一个身份字符串**（否则 key 对不上、去重失效）
|
||
self._last_ident = ""
|
||
self._dedup_seen = 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**——"这几个号都算我的"。要跑多个号的场景就靠它。
|
||
|
||
`selector_type="dedup"` 时换成**去重判断**（跨设备幂等，见 `core/dedup.py`）：
|
||
"这个身份在这个任务里做过了吗"——命中=做过。身份元素填在 `ident_value`
|
||
（留空则用设备 serial，适合一号一机）；有效期取任务级 `dedup_reset`。
|
||
多台手机跑同一任务时，任何一台做过了，其余设备的这一步都会命中。
|
||
返回 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 "")
|
||
# 去重条件不用 selector_value（身份元素在 ident_value 里），所以别按"缺选择器"拦
|
||
if not sel_val and sel_type != "dedup":
|
||
_log.warning(f"[{self.serial}] if_el 缺少选择器，跳过")
|
||
return None
|
||
found = False
|
||
got = None # 比对用的"采集到的值"：元素文本 / OCR 命中文本 / 前台包名
|
||
# 非元素类条件（屏幕状态/前台 App/去重）：不参与"选择器健康"统计——它们不是选择器，
|
||
# 连续未命中不该被记成"选择器失效"
|
||
dynamic = sel_type in ("screen", "foreground", "dedup")
|
||
detail = ""
|
||
if sel_type == "dedup":
|
||
# 去重条件：**这个身份在这个任务里做过了吗**（跨设备共享一份账本，
|
||
# 见 core/dedup.py）。命中=做过 → 走 then 分支（通常放"跳过/停止/发通知"）。
|
||
kind, hours = self._dedup_policy()
|
||
self._dedup_seen = True # 供「记为已做」区分"没配检查"与"配了但没读到"
|
||
ident = self._dedup_identity(d, params, "ident_type", "ident_value")
|
||
if ident is None:
|
||
found = False # 身份判定不了 → 当"没做过"，照常往下做（宁可重复不误拦）
|
||
detail = "去重：身份读不到，按未做过处理"
|
||
else:
|
||
self._last_ident = ident # 供后面「记为已做」复用（见 _exec_mark_done）
|
||
found = dedup.check(self.ctx.get("job_id", ""), ident, kind, hours)
|
||
detail = f"去重：身份『{ident}』{'做过' if found else '没做过'}（有效期 {kind}）"
|
||
elif 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 not in ("screen", "dedup"):
|
||
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 str(cmp_val or "").strip() and sel_type in ("screen", "dedup"):
|
||
# 只在**真的配了比对的值**时才提醒：从元素类切到「去重/屏幕状态」时
|
||
# 界面上那块是隐藏的，旧值会留在 params 里，那种残留每轮都刷警告没意义
|
||
_log.warning(f"[{self.serial}] {sel_type} 没有文本可比，已忽略文本比对设置")
|
||
|
||
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 _dedup_policy(self):
|
||
"""任务级去重有效期（`dedup_reset` / `dedup_hours`）→ (kind, hours)。
|
||
|
||
**刻意放任务级**：检查与记账两处各填一份的话，填不一致就会算出两个不同的 key、
|
||
去重静默失效（最难查的那种 bug）。认不出的值一律回退 `day`。
|
||
"""
|
||
p = self.params or {}
|
||
kind = str(p.get("dedup_reset") or dedup.KIND_DAY).strip()
|
||
if kind not in dedup.RESET_KINDS:
|
||
kind = dedup.KIND_DAY
|
||
try:
|
||
hours = int(p.get("dedup_hours") or dedup.DEFAULT_HOURS)
|
||
except (TypeError, ValueError):
|
||
hours = dedup.DEFAULT_HOURS
|
||
return kind, max(1, hours)
|
||
|
||
def _dedup_identity(self, d, params, tkey, vkey):
|
||
"""算去重身份：填了身份元素就读它的文本；留空则用设备 serial（一号一机场景）。
|
||
|
||
返回 None = **无法判定**（读不到值）→ 调用方必须"不去重"。
|
||
宁可漏拦一次，也不能把"读不到"当成空身份——那会让所有设备共用一个 key，
|
||
第一台记账之后其余全被误跳过（同一个号一台都没做，却被判"做过"）。
|
||
"""
|
||
sel_val = (params.get(vkey) or "").strip()
|
||
if not sel_val:
|
||
return self.serial
|
||
txt = self._read_element_text(d, params.get(tkey) or "text", sel_val)
|
||
if not txt:
|
||
_log.warning(f"[{self.serial}] 去重身份元素读不到值（{sel_val}）→ 本次不去重")
|
||
return None
|
||
return txt
|
||
|
||
def _exec_mark_done(self, d, params, depth=0):
|
||
"""「记为已做」：把身份值记进去重账本（跨设备幂等）。
|
||
|
||
**放在动作成功之后**：动作失败就不记账，下次重跑会重试该设备——
|
||
拆成"检查在前、记账在后"两步，天然就是"成功才记账、失败不丢"。
|
||
身份读不到时**不记账**（记一个空身份会把别人挡住）。
|
||
|
||
**身份留空 = 复用前面「去重」条件刚解析出的那个身份**（推荐这样配）：
|
||
检查与记账必须用**同一个身份字符串**，key 才算得到一起；让用户在两处各配一遍
|
||
元素，一旦不一致（填了一个、另一个忘了）去重就静默失效——这是最容易踩的坑，
|
||
所以默认行为改成"自动跟随上面的检查"。
|
||
"""
|
||
kind, hours = self._dedup_policy()
|
||
if (params.get("selector_value") or "").strip():
|
||
ident = self._dedup_identity(d, params, "selector_type", "selector_value")
|
||
elif self._last_ident:
|
||
ident = self._last_ident
|
||
_log.info(f"[{self.serial}] 记为已做：复用上面去重检查的身份『{ident}』")
|
||
elif self._dedup_seen:
|
||
# 上面**跑了**去重检查、但身份读不到（元素没出来/页面没到）。
|
||
# 这时**绝不能**退回设备身份记一笔：那会在"什么都没做成"的情况下
|
||
# 把设备标成已做。不记 → 下次重跑会重试。
|
||
ident = None
|
||
_log.warning(f"[{self.serial}] 记为已做：上面去重检查没读到身份（页面没到？）"
|
||
f"→ 本次不记账，下次重跑会重试")
|
||
else:
|
||
# 任务里根本没有前置去重检查、又没填身份元素 → 退回设备身份（一号一机场景）
|
||
ident = self.serial
|
||
_log.info(f"[{self.serial}] 记为已做：没填身份元素也没有前置去重检查 → "
|
||
f"按设备身份记账（{ident}）")
|
||
if not ident:
|
||
self.set_action("记为已做：身份读不到 → 本次不记（下次重跑会重试）")
|
||
return True
|
||
first = dedup.mark(self.ctx.get("job_id", ""), ident, kind, hours,
|
||
serial=self.serial,
|
||
device_name=self.ctx.get("device_name", ""),
|
||
job_name=self.ctx.get("job_name", ""))
|
||
who = "首次" if first else "已存在（别处先记过）"
|
||
_log.info(f"[{self.serial}] 记为已做：身份『{ident}』 → {who}")
|
||
self.set_action(f"记为已做：『{ident}』{who}")
|
||
return True
|
||
|
||
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
|