Files
auto_control/tasks/generic/task.py
T
butubb b98e6deac1 feat(发布计划): 视频发布计划(批量上传配对 → 时间线 → 推送到手机 → 发布任务 → 分享链接)
一、平台侧(账号 → 发布计划页)
- 新表 video_plan(schema v9→v10):账号×发布日期×编号 → 素材 + 标题 + 发布状态 + 分享链接;
  状态机 pending/ready/pushing/publishing/done/failed/unknown/skipped(**failed 与 unknown 必须分开**:
  推送阶段的失败可安全重试;碰过抖音之后的岔子只能算"结果未知",绝不自动重发)
- 素材上传:文件名 `手机号_日期_编号`(编号可省)解析配对;标题 txt `标题内容_手机号_日期_编号`;
  内容寻址落盘 data/videos/YYYY-MM/(sha1 分块算,同名不存两份),**不进整库备份**但进 manifest 反查
- 新蓝图 web/video_plan_api.py:上传/时间线/统计/单条增删改/推送到手机/标记结果/裁决/链接导出 CSV/
  任务列表与一键新建、**就地编辑**(GET/PUT /tasks/<id>)、**一键推送**(POST /push_all,按设备分组、设备内串行)
- 账号页拆子分栏(台账 / 发布计划)+ static/admin/release.js;清理 job(04:41 僵尸回收+过期行、04:47 素材文件)
- 上传体积:MAX_CONTENT_LENGTH(默认 2GiB)+ 413 JSON + nginx client_max_body_size(修现有 APK 上传隐患)

二、任务侧(平台推素材,抖音流程你自己写)
- 新步骤 push_release「推送发布视频」:原子占位 → adb push → **touch 改成"现在"** → 清旧目录同名副本 →
  触发扫描并**按路径**校验相册索引 → 标题写进剪贴板;默认目录 /sdcard/DCIM/Camera
- 新步骤 mark_release「标记发布结果」:回写 done/failed/unknown,成功时抓作品分享链接、删手机素材
- input_text 支持 text_source=release_title(自动取计划标题 + 回读校验);
  if_el 的候选值来源新增 release(**本机当前发布计划**的抖音号/昵称,发布前校验"登的是不是要发的号")
- build_release_steps 骨架 15 步:⓪ 亮屏 → ① 打开抖音(等首页) → ② 点「我」→ ③ 等抖音号出现 →
  ④ 条件判断(账号) → then ⑤ 推送 ⑥⑦⑧⑨⑩⑪⑫ 抖音点击/填标题 → ⑬ 标记 / else 发通知跳过

三、修(推送这一路的检测机制)
- **uiautomator2 3.x 的 d.shell() 返回 ShellResponse(tuple 子类)不是 str**:`'x' in resp` 恒 False、
  `.strip()` 不存在 → "推上去的文件大小不对"每次都判失败(文件其实推上去了)、相册校验永远报没进、
  删除确认永远判没删掉。新增 publish_flow._sh() 统一取 .output;大小改成解析 ls -l 的大小列
- **adb push 保留本地 mtime** → 推 3 天前上传的素材在按时间排序的相册里排不到最前,
  "点第一个 = 刚推的那个"不成立 → 推完 touch
- 相册校验**按路径**比(MediaStore 的 _data 会把目录小写、/storage/emulated/0 ≡ /sdcard),
  只比文件名会被老目录的同名残留骗过去
- 屏幕没亮就启动抖音会永远停在启动页(UI 树为空)→ 后面"点我/等抖音号"必然 miss,
  最后报成误导人的"账号不符" → 骨架第一步固定加「亮屏」,open_app 等「首页」出现

四、其它
- core/ledger.serial_of():设备名 → 当前地址(设备换 IP 后快照是错的)
- 通知事件 task.video.published / task.video.failed;备份清单加 video_plan 与素材统计
- 文档同步:DATA_MODEL §2.11 + schema v10、API(新接口与语义)、TASK_DEV §4.7 专章、
  ARCHITECTURE(账号页子分栏/release.js/两个 job)、DEPLOY(表数/nginx)、NOTIFY、DEVELOPMENT、README
2026-09-28 15:59:09 +08:00

1531 lines
80 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""通用步骤任务定义。
前端步骤编辑器编排步骤链,保存为 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, ledger
_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, "text_source": "manual", "release_topics": ""}},
{"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": "", "cmp_source": "", "cmp_group": "",
"ident_type": "text", "ident_value": "",
"then": [], "else": []}},
{"type": "mark_done", "label": "记为已做", "icon": "✅",
"params": {"selector_type": "text", "selector_value": ""}},
{"type": "push_release", "label": "推送发布视频", "icon": "📤",
"params": {"date_source": "today", "fixed_date": "",
"account": "device", "account_phone": "",
"album_dir": "/sdcard/DCIM/Camera", "to_clipboard": True,
"retry_failed": True, "max_same_run": 1}},
{"type": "mark_release", "label": "标记发布结果", "icon": "🏁",
"params": {"result": "published", "why": "", "capture_link": True,
"delete_phone": True, "album_dir": "/sdcard/DCIM/Camera"}},
{"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._last_release = None # 「推送发布视频」推的那条计划,供「标记发布结果」用
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 "")
# 候选值可以来自「账号台账」(cmp_source)——这里只把 cmp_val 这个**字符串**
# 换掉来路,下面 762-780 的比对、候选计数、日志文案全自动跟着对
# (_cmp_hit 保持纯函数,不掺 DB)
src_note = ""
if cmp_op:
cmp_val, src_note = self._resolve_cmp_value(params, cmp_val)
# 去重条件不用 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} 没有文本可比,已忽略文本比对设置")
# 候选值来源(台账/手填)也要进日志与步骤明细:取 0 个号会让这一步
# **永远走 else 分支**,而任务照样"跑完了"——只有这里能看出原因
if src_note:
detail = f"{detail};{src_note}" if detail else src_note
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 _resolve_cmp_value(self, params, manual_raw):
"""「条件判断」的候选值 = 手填值 + 台账取号(合并,OR 语义)。
返回 (候选值文本, 给日志/步骤明细的来源说明)。
· **合并**而不是二选一:台账取空时手填值仍然生效(台账没维护好不会把任务
直接打哑);"台账里的号 + 临时补一个"也是自然需求 —— split_cmp_values
与 `_cmp_hit` 本来就是多值 OR 语义。
· 取 0 个号时**必须留痕**:候选为空 ⇒ 这一步永远走 else 分支,而任务本身
"跑完了、显示成功",不写日志就只能靠人猜。
· ⚠ 这里产出的只是**比对用的文本**,绝不参与去重 identity:
身份是元素原文逐字算 key,台账里是纯号,格式不同会让去重静默失效
(见 core/ledger.py 顶部的警告)。
"""
source = str(params.get("cmp_source") or "").strip()
manual = split_cmp_values(manual_raw)
if source == "release":
# 「发布计划」:本机**下一条将要推送**的那条计划(推送前校验账号就靠它)。
# 候选值给两个:抖音号(纯号)+ 账号名称 —— 元素原文可能是
# `抖音号:35377983067`,也可能只显示昵称,两个都可命中(OR 语义)。
ids, note = self._release_candidates()
elif source in ("device", "all", "group"):
ids, note = ledger.douyin_ids(
source, serial=self.serial,
device_name=(self.ctx or {}).get("device_name", ""),
group=str(params.get("cmp_group") or "").strip())
else:
return manual_raw, "" # 手填(默认):行为与以前完全一致
merged = list(ids) + [v for v in manual if v not in ids]
if not ids:
_log.warning(
f"[{self.serial}] 条件判断取号为空(来源 {source}):{note};手填候选 "
f"{len(manual)} 个 → "
+ ("仅用手填值" if manual else "候选为空,这一步将永远不命中(走未找到分支)"))
where = "发布计划" if source == "release" else f"台账({source})"
src_txt = (f"候选来自{where} {len(ids)} 个号" if ids
else f"{where}取号为空:{note}")
if manual:
src_txt += f",另加手填 {len(manual)} 个"
return "\n".join(merged), src_txt
def _release_targets(self, params):
"""推送/标记步骤共用的"取本机账号手机号"逻辑。返回 (phones, who)。"""
from core import ledger
phone = (params.get("account_phone") or "").strip()
if (params.get("account") or "device") == "phone" and phone:
return [phone], f"指定手机号 {phone}"
_, rows, note = ledger.resolve_device(
serial=self.serial, device_name=(self.ctx or {}).get("device_name", ""))
return [r.get("phone") for r in rows if r.get("phone")], f"本机账号({note})"
def _release_candidates(self):
"""条件判断的「候选值来源 = 发布计划」:本机要发的那条计划的 **抖音号 + 账号名称**。
优先用本次运行里「推送发布视频」已经拿到的那条(`_last_release`);
还没有就**看**下一条(peek,不占位)—— 这正是"推送前先校验账号对不对"的用法:
命中(当前登录的就是要发的那个号)→ 走 then 去推送/发布;未命中 → else 跳过并通知。
给两个候选值是刻意的:元素原文可能是 `抖音号:35377983067`,也可能只显示昵称,
两个都算命中(候选值之间是 OR)。配「包含」运算符即可。
"""
from core import video_plan as vp
plan = getattr(self, "_last_release", None)
src = "本次已推送的那条"
if not plan:
phones, _who = self._release_targets({})
plan = vp.peek_for_device(phones, vp._today()) if phones else None
src = "本机下一条待发"
if not plan:
return [], "本机没有待发的发布计划"
cands = [c for c in (plan.get("douyin_id"), plan.get("nickname")) if c]
note = (f"{src}计划的抖音号 {plan.get('douyin_id')}"
f"({plan.get('nickname') or plan.get('phone')} "
f"{plan.get('release_date')} #{plan.get('seq')})")
return cands, note
def _exec_push_release(self, d, params, depth=0):
"""推送发布视频:把本机某天待发的计划**推到手机相册**(**不发布**)。
**平台只负责到这里**:推文件 + 触发相册刷新 + 把标题写进剪贴板;
抖音里怎么发(打开→点+→相册→选视频→下一步→粘贴标题→发布)由你在这条步骤
**后面自己画步骤**,最后放一个「标记发布结果」回写状态。
参数:`date_source`(today/fixed) `fixed_date` `account`(device/phone) `account_phone`
`album_dir`(默认 `/sdcard/DCIM/Camera`)`to_clipboard`(True) `retry_failed`(True)
`max_same_run`(1 —— 一次运行推几条;**多条时「标记」只认最后一条**,要发多条就重复"推送+标记"两步)
返回 None=本机没有待推的(跳过);True/False=至少推成一条/全失败。
"""
from tasks.generic import publish_flow as flow
from core import video_plan as vp
date = ((params.get("fixed_date") or "").strip()
if params.get("date_source") == "fixed" else vp._today()) or vp._today()
phones, who = self._release_targets(params)
if not phones:
_log.warning(f"[{self.serial}] 推送发布视频:本机没有台账账号 → 跳过")
self.set_action("本机没有台账账号 → 跳过")
return None
plans = vp.take_for_device(phones, date,
limit=int(params.get("max_same_run") or 1),
retry_failed=params.get("retry_failed", True) is not False)
if not plans:
_log.info(f"[{self.serial}] 推送发布视频:{who} 在 {date} 没有待推送的计划 → 跳过")
self.set_action(f"{date} 没有待推送计划")
return None
opts = {"album_dir": (params.get("album_dir") or "").strip() or flow.ALBUM_DIR,
"to_clipboard": params.get("to_clipboard", True) is not False}
okN = badN = 0
for plan in plans:
if self.stopped():
break
self._last_release = plan # 供后面的「标记发布结果」用
self.set_action(f"推送中:{plan.get('nickname') or plan['phone']} "
f"{plan['release_date']} #{plan['seq']}")
ok, msg = flow.push_one(self.serial, plan, opts,
log=lambda s: _log.info(f"[{self.serial}] {s}"))
if ok:
okN += 1
# 相册校验结果**单独说一句**:文件上去了 ≠ 相册里点得到它,
# 后面那步"点相册里第一个"靠的就是索引(不写这句,出了问题只能猜)
row = vp.get_plan(plan["id"]) or {}
if row.get("push_verify") == "no_index":
_log.warning(f"[{self.serial}] ⚠ 素材推上去了但**没进相册索引**"
f"(抖音选视频时可能看不到它)")
else:
badN += 1
self._notify_publish(plan, ok=False, msg=msg)
if okN and self._last_release:
lr = self._last_release
title = (lr.get("title") or "")[:20]
self.set_action(f"已推送到手机:{lr.get('nickname') or lr.get('phone')} "
f"{lr['release_date']} #{lr['seq']}"
+ (f"(标题『{title}』已进剪贴板)" if title else "")
+ ";发完记得放「标记发布结果」")
_log.info(f"[{self.serial}] 已推到手机,标题『{title}』已进剪贴板 —— "
f"后面用「粘贴」把它填进抖音,发完记得放一个「标记发布结果」")
else:
self.set_action(f"推送发布视频:成功 {okN} 条"
+ (f",失败 {badN} 条" if badN else ""))
return okN > 0 if (okN or badN) else None
def _exec_mark_release(self, d, params, depth=0):
"""标记发布结果:把"发出去了没有"回写到计划行(**你自己发布流程的最后一步**)。
参数:`result` = `published`(发出去了) / `failed`(没发出去,可重试) / `unknown`(不确定)
· `why` 失败原因(可选)· `capture_link`(True) 成功时顺带抓作品分享链接
· `delete_phone`(True) 成功/失败后删掉手机上的素材
放在你画布的**成功分支**里填 `published`、失败分支里填 `failed`;
拿不准有没有发出去就填 `unknown`(平台不会自动重发它)。
返回 True=标记成功 / False=发布失败 / None=找不到对应的计划(跳过)。
"""
from tasks.generic import publish_flow as flow
from core import video_plan as vp
import uiautomator2 as u2
plan = getattr(self, "_last_release", None)
if not plan:
phones, _who = self._release_targets(params)
plan = vp.last_pushed_for_device(phones) if phones else None
if not plan:
_log.warning(f"[{self.serial}] 标记发布结果:找不到对应的计划行 → 跳过"
f"(前面要先有一个「推送发布视频」步骤)")
self.set_action("没有可标记的计划")
return None
result = (params.get("result") or "published").strip().lower()
if result not in ("published", "failed", "unknown"):
_log.warning(f"[{self.serial}] 标记发布结果:result={result!r} 认不出 → 按 unknown 处理")
result = "unknown"
why = (params.get("why") or "").strip()
# 手机上的素材:先删再回写状态 —— 删失败不影响发布结论。
# 远程路径是**确定的**({相册目录}/{计划id}.mp4),所以这里能直接算出来,
# 不用把"上次推到哪"存进库(存了还得担心不一致)
if params.get("delete_phone", True) is not False:
try:
dd = d or u2.connect(self.serial)
album = (params.get("album_dir") or "").strip() or flow.ALBUM_DIR
flow.clean_phone_video(dd, f"{album}/{plan['id']}.mp4")
except Exception as e:
_log.warning(f"[{self.serial}] 删手机素材失败(不影响标记): {e}")
if result == "published":
url = ""
if params.get("capture_link", True) is not False:
try:
dd = d or u2.connect(self.serial)
url, lmsg = flow.capture_share_link(dd, self.serial)
_log.info(f"[{self.serial}] 抓分享链接:{url or ('失败:' + lmsg)}")
except Exception as e:
_log.warning(f"[{self.serial}] 抓分享链接异常: {e}")
vp.mark_done(plan["id"], share_url=url, msg=why)
self._notify_publish(dict(plan, share_url=url), ok=True, msg=why or "已标记发布成功")
self.set_action(f"已标记发布成功({'链接 ' + url if url else '没抓到链接'})")
return True
if result == "failed":
# **确认没发出去** → 直接落 `failed`(可重试),不能走 mark_failed('post')
# —— 那会落到"结果未知"、反而不敢重试。这里人是亲眼确认过的。
vp.set_status(plan["id"], vp.ST_FAILED, note=why or "人工标记:没发出去")
self._notify_publish(plan, ok=False, msg=why or "人工标记:没发出去")
self.set_action("已标记失败(下次任务会再推一次)")
return False
vp.mark_failed(plan["id"], "verify", why or "人工标记:结果不确定")
self._notify_publish(plan, ok=False, msg=why or "人工标记:结果不确定")
self.set_action("已标记「结果未知」(不会自动重发,需人工确认)")
return False
def _notify_publish(self, plan, ok, msg):
"""发布结果推通知(notify 零 DB:字段都在计划行里,不查库)。"""
try:
notifier.notify("task.video.published" if ok else "task.video.failed",
serial=self.serial,
device_name=(self.ctx or {}).get("device_name", ""),
model=(self.ctx or {}).get("model", ""),
phone=plan.get("phone", ""),
nickname=plan.get("nickname", ""),
release_date=plan.get("release_date", ""),
seq=plan.get("seq", 0),
title=(plan.get("title") or "")[:60],
share_url=plan.get("share_url", ""),
msg=msg)
except Exception as e:
_log.warning(f"发布通知发送失败(不影响发布):{e}")
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
**`text_source`(可选):文字从哪来**
· `manual`(默认)= 用上面的 mode(行为与以前完全一致)
· `release_title` = **本机当前那条发布计划的标题**("推送发布视频"推的那条;
本机没有就取最近一条已推送的)· 配 `release_topics` 还能自动加话题
选发布计划来源时**会回读校验**(读输入框的 text):不校验的后果是
"抖音不吃 send_keys → 发出一篇空文案,而每一步都显示成功";
校验不过就自动改走剪贴板通道粘贴(设备端 Agent)再验一次。
"""
src = str(params.get("text_source") or "manual").strip()
text = ""
if src in ("release_title", "release_title_topics"):
from core import video_plan as vp
plan = getattr(self, "_last_release", None)
if not plan:
phones, _who = self._release_targets(params)
plan = vp.last_pushed_for_device(phones) if phones else None
if plan:
text, _err = vp.build_caption(
plan, params.get("release_topics") or ""
if src == "release_title_topics" else "")
self._last_release = self._last_release or plan
_log.info(f"[{self.serial}] 输入文字取计划标题:"
f"{plan.get('nickname') or plan.get('phone')} "
f"{plan.get('release_date')} #{plan.get('seq')}")
if not text:
_log.warning(f"[{self.serial}] 输入文字:本机没有可用的发布计划标题 → 跳过"
f"(前面要先有「推送发布视频」步骤)")
return None
if not 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}")
# 取计划标题时**必须回读校验**:抖音的 EditText 有时不吃 send_keys,
# 不校验就可能发出一篇空文案而每一步都显示成功
if src in ("release_title", "release_title_topics"):
return self._verify_input(d, text, clear_first)
return None
def _verify_input(self, d, text, clear_first):
"""回读输入框,确认文字真的进去了;不行就改走剪贴板通道再验一次。"""
def _read():
try:
el = d(className="android.widget.EditText")
if el.exists:
return el.get_text(timeout=2) or ""
except Exception:
pass
return ""
for attempt in (1, 2):
got = _read()
if got and text[:12] in got:
return True
_log.warning(f"[{self.serial}] 文案回读不一致(第 {attempt} 次):"
f"期望『{text[:20]}…』读到『{got[:20]}』")
if attempt == 1:
# 第 2 次改走剪贴板通道(设备端 Agent 写剪贴板 + 粘贴),中文不会变形
try:
from core.clipboard_helper import inject_clipboard
okc, msgc = inject_clipboard(self.serial, text, d=d)
if okc:
d.jsonrpc.pasteClipboard()
time.sleep(1)
else:
_log.warning(f"[{self.serial}] 剪贴板兜底注入失败: {msgc}")
except Exception as e:
_log.warning(f"[{self.serial}] 剪贴板兜底异常: {e}")
time.sleep(1)
got = _read()
if got and text[:12] in got:
return True
_log.warning(f"[{self.serial}] ⚠ 文案没填进去(读回『{got[:30]}』)——"
f"接着发布就会发出空文案,这一步判失败")
return False
@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