"""通用任务管理框架：任务类型注册 + 设备分组 + 任务计划 + 定时调度 + 重试 + 持久化。 设计目标：可扩展，未来加非抖音任务只需注册新的 Task 类。 核心概念： TaskType — 任务类型（如"抖音养号"），可注册，含默认参数和 worker 工厂 DeviceGroup — 设备分组，持久化到 SQLite（core.models.DeviceGroup） TaskJob — 任务计划（类型+目标+参数+调度+重试），持久化到 SQLite（core.models.TaskJob） TaskManager — 统管调度器、分组、任务、运行实例、状态 定时：APScheduler BackgroundScheduler，cron 触发后为每个目标设备起 worker。 支持三种调度模式： once — 手动/立即执行，不注册 cron cron — 仅注册启动 cron（到点启动 worker） cron_stop — 注册启动 cron + 停止 cron（到点启动，到点停止） 重试：单设备 worker 失败后按 retry.max_attempts 重试，间隔 retry.delay。 并发：同一 serial 同时只允许一个 worker，避免冲突。 """ import os import json import time import uuid import threading from concurrent.futures import ThreadPoolExecutor, as_completed from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from config import DATA_DIR from core.logger import get_logger from core.models import db, DeviceGroup as GroupRow, TaskJob as JobRow from .stf_client import STFClient, DeviceOfflineError from .adb_helper import get_foreground_app, adb_connect_light, adb_disconnect from .device_worker import ( get_all_worker_status, _update_status, _remove_worker, _WORKERS, _WORKERS_LOCK, start_watchdog, stop_watchdog, ) from tasks import list_task_types, get_task_class _log = get_logger("core.tm") # 错峰启动间隔（秒）：批量触发时设备逐个开始占用/连接，避免 adb 全局锁串行堆积 # 和 STF occupy 并发风暴。100 台 × 0.2s = 20s 全部开始启动。 _START_STAGGER_SEC = 0.2 # ================== 设备分组 ================== class DeviceGroup: def __init__(self, name, serials=None, description=""): self.name = name self.serials = serials or [] self.description = description def to_dict(self): return {"name": self.name, "serials": self.serials, "description": self.description} @classmethod def from_dict(cls, d): return cls(d["name"], d.get("serials", []), d.get("description", "")) # ================== 任务计划 ================== class TaskJob: """一个任务计划：什么任务、跑哪些设备、什么参数、何时跑、失败重试策略。""" def __init__(self, job_id=None, name="", task_type="douyin_nurture", target=None, params=None, schedule=None, retry=None, enabled=True): self.id = job_id or uuid.uuid4().hex[:8] self.name = name self.task_type = task_type # target: {"mode": "all"|"group"|"serial", "group_name": "", "serial": ""} self.target = target or {"mode": "all"} self.params = params or {} self.schedule = schedule or {"mode": "once"} # {"mode":"once"|"cron", "cron":"0 9 * * *"} self.retry = retry or {"max_attempts": 1, "delay": 60} self.enabled = enabled def to_dict(self): return {"id": self.id, "name": self.name, "task_type": self.task_type, "target": self.target, "params": self.params, "schedule": self.schedule, "retry": self.retry, "enabled": self.enabled} @classmethod def from_dict(cls, d): return cls(d["id"], d["name"], d["task_type"], d.get("target"), d.get("params"), d.get("schedule"), d.get("retry"), d.get("enabled", True)) def resolve_serials(self, manager): """根据 target 解析出实际要跑的 serial 列表。""" mode = self.target.get("mode", "all") if mode == "serial": return [self.target["serial"]] if mode == "group": g = manager.groups.get(self.target.get("group_name")) return list(g.serials) if g else [] # all：返回所有空闲设备 try: return [d["serial"] for d in manager.stf.list_free_devices()] except Exception: return [] # ================== 前台 App 扫描器 ================== class _ForegroundScanner: """前台 App 扫描器：手动触发，获取所有在线设备的前台 App。 设计原则：**不打扰设备**，扫描过程不会让设备退出当前 App。 策略（按设备状态区分获取逻辑）： - worker 运行中设备：用已有 remote_adb_url 直接查询（已有 adb 连接，无额外开销） - 空闲设备：直接 adb connect → 查询 → adb disconnect（绕过 STF，不打扰设备） - 被别人占用的设备：标记 "他人占用" - STF occupy/release 会打扰设备（可能退回桌面），绝不使用 serial 格式为 IP:5555（设备本身的 adb 网络地址），可直接 adb connect。 adb connect/disconnect 只建立/断开调试连接，不影响设备 UI。 dumpsys window 只读取窗口状态，不执行任何操作。 """ def __init__(self, stf): self.stf = stf self._cache = {} # serial -> app_name self._cache_lock = threading.Lock() self._scanning = threading.Event() # 标记是否正在扫描 self._last_scan_time = 0 # 上次扫描完成时间戳 self._last_scan_count = 0 # 上次扫描设备数 def stop(self): """兼容 TaskManager.shutdown() 调用。""" pass def get(self, serial): """读取缓存的前台 App。""" with self._cache_lock: return self._cache.get(serial) @property def is_scanning(self): return self._scanning.is_set() @property def last_scan_time(self): return self._last_scan_time def scan_once(self): """手动触发一次扫描（后台线程执行，不阻塞调用者）。 如果已有扫描在进行中，直接返回 False。 """ if self._scanning.is_set(): return False t = threading.Thread(target=self._scan_all, name="fg-scan-once", daemon=True) t.start() return True def _scan_all(self): """扫描所有在线设备（后台线程执行）。 按设备归属分四类处理： 1. worker 运行中：用已有 remote_adb_url 直接查询（无额外开销） 2. 自己账户占用但无 worker：调用 STF remoteConnect 获取隧道查询 （不 occupy/release，不打扰设备 UI） 3. 完全空闲设备（using=False）：尝试轻量 adb connect serial （单次尝试，不 kill-server，不影响其他 worker） 4. 被他人占用：标记 "(他人占用)" """ self._scanning.set() _log.info("前台 App 扫描已启动") try: all_devices = self.stf.list_all_devices() except Exception as e: _log.error("前台 App 扫描: 获取设备列表失败: %s", e) self._scanning.clear() return # 自己账户已占用的设备列表（用于区分"自己占用"vs"他人占用"） try: my_serials = {d["serial"] for d in self.stf.list_my_devices()} except Exception: my_serials = set() worker_status = {w["serial"]: w for w in get_all_worker_status()} # 分类设备 have_conn = {} # serial -> remote_adb_url（worker 运行中，已有 adb 连接） my_owned = [] # 自己账户占用但无 running worker，需 remoteConnect 获取隧道 free_serials = [] # 完全空闲设备，尝试轻量 adb connect skip_serials = [] # 被他人占用 for dev in all_devices: if not dev.get("present"): continue serial = dev.get("serial", "") if not serial: continue w = worker_status.get(serial, {}) url = w.get("remote_adb_url") if url and w.get("status") in ("running", "connecting"): # 1. 设备正在执行任务，已有 adb 连接，直接查询 have_conn[serial] = url elif serial in my_serials: # 2. 自己账户占用但无 worker，可安全 remoteConnect（不打扰设备） my_owned.append(serial) elif dev.get("using"): # 4. 被他人占用 skip_serials.append(serial) else: # 3. 完全空闲，尝试轻量 adb connect（不 kill-server） free_serials.append(serial) results = {} _log.info("前台 App 扫描分类: 运行中=%d, 自己占用=%d, 空闲=%d, 他人占用=%d", len(have_conn), len(my_owned), len(free_serials), len(skip_serials)) # 1. worker 运行中设备：用已有 remote_adb_url 查询（并发 10） if have_conn: with ThreadPoolExecutor(max_workers=min(10, len(have_conn))) as pool: futures = {pool.submit(get_foreground_app, url): s for s, url in have_conn.items()} for fut in as_completed(futures, timeout=15): s = futures[fut] try: results[s] = fut.result() except Exception: results[s] = None # 2. 自己占用设备：remoteConnect 获取隧道 → 查询 → 断开（并发 5，不打扰设备） if my_owned: with ThreadPoolExecutor(max_workers=min(5, len(my_owned))) as pool: futures = {pool.submit(self._scan_my_owned, s): s for s in my_owned} for fut in as_completed(futures, timeout=30): s = futures[fut] try: results[s] = fut.result() except Exception: results[s] = None # 3. 空闲设备：轻量 adb connect serial → 查询 → disconnect（并发 10） if free_serials: with ThreadPoolExecutor(max_workers=min(10, len(free_serials))) as pool: futures = {pool.submit(self._scan_free, s): s for s in free_serials} for fut in as_completed(futures, timeout=20): s = futures[fut] try: results[s] = fut.result() except Exception: results[s] = None # 4. 被别人占用的设备 for s in skip_serials: results[s] = "(他人占用)" # 更新缓存 with self._cache_lock: for s, app in results.items(): self._cache[s] = app self._last_scan_time = time.time() self._last_scan_count = len(results) self._scanning.clear() _log.info("前台 App 扫描完成: %d 台设备", len(results)) def _scan_my_owned(self, serial): """自己账户占用的设备：通过 STF remoteConnect 获取隧道查询。 不调用 occupy/release，只建立/断开 ADB 隧道，不打扰设备 UI。 """ try: url = self.stf.remote_connect(serial) if not url: return None try: if not adb_connect_light(url): return None return get_foreground_app(url) finally: adb_disconnect(url) self.stf.remote_disconnect(serial) except Exception: return None def _scan_free(self, serial): """空闲设备：不扫描前台 App，直接返回"空闲"。 原因：STF provider 内部通过 IP:5555 维持 adb 连接监控设备。 外部 adb connect/disconnect IP:5555 会让 adb server 断开该地址的 transport，连带 STF provider 的连接一起断，STF 误判设备 offline 并触发重连——表现就是"一扫描前台 App 设备就离线、需要重连"。 因此空闲设备绝不主动 adb connect/disconnect，只对运行中设备 （走 STF 隧道端口，不动 5555）和自占设备（remoteConnect 隧道）扫描。 """ return "空闲" # ================== 任务管理器 ================== class TaskManager: def __init__(self, stf_client=None, app=None): self.stf = stf_client or STFClient() self.app = app # Flask app，用于 db context self.scheduler = BackgroundScheduler(timezone="Asia/Shanghai") self.scheduler.start() start_watchdog() # 启动心跳看门狗 self.groups = {} # name -> DeviceGroup(内存业务对象) self.jobs = {} # id -> TaskJob(内存业务对象) self._running = {} # serial -> {"worker", "job_id", "started_at", "attempt"} self._stop_requested = set() # serial 集合：用户请求停止，阻止后续重试 self._lock = threading.Lock() self._fg_scanner = _ForegroundScanner(self.stf) self._load() def _db(self): """获取 db context。task_manager 在后台线程跑，需自己推 app context。""" if self.app is None: raise RuntimeError("TaskManager 未关联 Flask app，无法访问数据库") return self.app.app_context() # ---- 持久化（SQLite，替代旧 JSON 文件）---- def _load(self): """从数据库加载分组和任务到内存。""" try: with self._db(): for row in GroupRow.query.all(): g = DeviceGroup(row.name, row.get_serials(), row.description or "") self.groups[g.name] = g dirty = False # 标记是否有任务参数需要写回 for row in JobRow.query.all(): j = TaskJob(job_id=row.id, name=row.name, task_type=row.task_type, target=row.get_target(), params=row.get_params(), schedule=row.get_schedule(), retry=row.get_retry(), enabled=row.enabled) # 数据规整：移除已废弃的 action 配置（如抖音的 comment 已删除） if j.task_type == "douyin_nurture": actions = j.params.get("actions", {}) if "comment" in actions: del actions["comment"] row.set_params(j.params) dirty = True _log.info(f"任务 {j.name}({j.id}): 已移除废弃的 comment 配置") self.jobs[j.id] = j if dirty: db.session.commit() _log.info("已将规整后的任务参数写回数据库") _log.info(f"从数据库加载 {len(self.groups)} 个分组, {len(self.jobs)} 个任务") except Exception as e: _log.error(f"从数据库加载失败: {e}") # 恢复启用的 cron 任务 for job in self.jobs.values(): if job.enabled and job.schedule.get("mode") in ("cron", "cron_stop"): self._add_cron(job) def _save_groups(self): """保存所有分组到数据库（upsert）。""" with self._db(): for g in self.groups.values(): row = GroupRow.query.filter_by(name=g.name).first() if not row: row = GroupRow(name=g.name) db.session.add(row) row.set_serials(g.serials) row.description = g.description db.session.commit() def _save_jobs(self): """保存所有任务到数据库（upsert）。""" with self._db(): for j in self.jobs.values(): row = JobRow.query.get(j.id) if not row: row = JobRow(id=j.id) db.session.add(row) row.name = j.name row.task_type = j.task_type row.set_target(j.target) row.set_params(j.params) row.set_schedule(j.schedule) row.set_retry(j.retry) row.enabled = j.enabled db.session.commit() # ---- 分组管理 ---- def add_group(self, name, serials, description=""): g = DeviceGroup(name, serials, description) self.groups[name] = g self._save_groups() return g def update_group(self, name, serials=None, description=None): g = self.groups.get(name) if not g: return None if serials is not None: g.serials = serials if description is not None: g.description = description self._save_groups() return g def delete_group(self, name): if name in self.groups: del self.groups[name] self._save_groups() return True return False # ---- 任务计划管理 ---- def add_job(self, name, task_type, target, params, schedule, retry, enabled=True): job = TaskJob(name=name, task_type=task_type, target=target, params=params, schedule=schedule, retry=retry, enabled=enabled) self.jobs[job.id] = job if enabled and schedule.get("mode") in ("cron", "cron_stop"): self._add_cron(job) self._save_jobs() return job def update_job(self, job_id, **fields): job = self.jobs.get(job_id) if not job: return None # 更新前先移除旧 cron was_cron = job.enabled and job.schedule.get("mode") in ("cron", "cron_stop") if was_cron: self._remove_cron(job_id) for k, v in fields.items(): if v is not None and hasattr(job, k): setattr(job, k, v) if job.enabled and job.schedule.get("mode") in ("cron", "cron_stop"): self._add_cron(job) self._save_jobs() return job def delete_job(self, job_id): self._remove_cron(job_id) if job_id in self.jobs: del self.jobs[job_id] self._save_jobs() return True return False def toggle_job(self, job_id, enabled): job = self.jobs.get(job_id) if not job: return None job.enabled = enabled if enabled and job.schedule.get("mode") in ("cron", "cron_stop"): self._add_cron(job) else: self._remove_cron(job_id) self._save_jobs() return job # ---- APScheduler 集成 ---- def _cron_id(self, job_id, kind="start"): """生成 cron job id。kind: start=启动 cron，stop=停止 cron。""" return f"job_{job_id}_{kind}" def _add_cron(self, job): """注册定时任务。支持三种模式： - once: 不注册 cron - cron: 只注册启动 cron（到点启动 worker） - cron_stop: 注册启动 cron + 停止 cron（到点启动，到点停止） """ mode = job.schedule.get("mode", "once") if mode not in ("cron", "cron_stop"): return start_cron = job.schedule.get("cron", "") if start_cron: try: trigger = CronTrigger.from_crontab(start_cron) except Exception as e: _log.error(f"启动 cron 解析失败 {start_cron}: {e}") else: self.scheduler.add_job(self._on_cron_trigger, trigger, args=[job.id], id=self._cron_id(job.id, "start"), replace_existing=True) _log.info(f"已注册启动定时 {job.name}({job.id}) cron={start_cron}") if mode == "cron_stop": stop_cron = job.schedule.get("stop_cron", "") if stop_cron: try: trigger = CronTrigger.from_crontab(stop_cron) except Exception as e: _log.error(f"停止 cron 解析失败 {stop_cron}: {e}") else: self.scheduler.add_job(self._on_cron_stop, trigger, args=[job.id], id=self._cron_id(job.id, "stop"), replace_existing=True) _log.info(f"已注册停止定时 {job.name}({job.id}) stop_cron={stop_cron}") def _remove_cron(self, job_id): """移除该任务的所有 cron job（启动 + 停止）。""" for kind in ("start", "stop"): try: self.scheduler.remove_job(self._cron_id(job_id, kind)) except Exception: pass def _on_cron_trigger(self, job_id): """cron 触发：启动该任务的所有目标设备 worker。""" job = self.jobs.get(job_id) if not job: return _log.info(f"定时触发: {job.name}({job.id})") self._run_job(job) def _on_cron_stop(self, job_id): """停止 cron 触发：停止该任务正在运行的所有 worker。 只停本 job 启动的 worker，不影响其他任务。 """ job = self.jobs.get(job_id) if not job: return with self._lock: items = [(s, info) for s, info in self._running.items() if info.get("job_id") == job_id] stopped = [] for serial, info in items: w = info.get("worker") if w and w.is_alive(): w.stop() stopped.append(serial) _log.info(f"定时停止 {job.name}({job.id}): 停止 {len(stopped)} 台设备 {stopped}") def run_job_now(self, job_id): """立即执行任务（手动触发）。""" job = self.jobs.get(job_id) if not job: return {"ok": False, "error": "任务不存在"} # 在独立线程跑，不阻塞调用方 t = threading.Thread(target=self._run_job, args=(job,), daemon=True) t.start() return {"ok": True, "msg": f"任务 {job.name} 已触发"} def _run_job(self, job): """执行任务：为每个目标设备起 worker（含重试循环）。 大量设备同时启动会触发 adb connect 全局锁串行 + STF occupy 并发风暴， 这里按 _START_STAGGER_SEC 间隔逐个启动，分摊连接压力（100 台 × 0.2s = 20s）。 """ serials = job.resolve_serials(self) if not serials: _log.warning(f"任务 {job.name} 无可用设备") return task_cls = get_task_class(job.task_type) if not task_cls: _log.error(f"未知任务类型: {job.task_type}") return task = task_cls() max_attempts = max(1, job.retry.get("max_attempts", 1)) delay = job.retry.get("delay", 60) for idx, serial in enumerate(serials): # 每台设备一个重试循环线程，互不影响；错峰延迟在各自线程内等待 t = threading.Thread(target=self._run_with_retry, args=(task, serial, job, max_attempts, delay, idx * _START_STAGGER_SEC), daemon=True) t.start() def _run_with_retry(self, task, serial, job, max_attempts, delay, start_delay=0): """单设备任务执行 + 重试。 异常分类： DeviceOfflineError — 设备掉线，立即放弃不重试（换设备也没用） 其他异常 — 按 max_attempts 重试 用户停止（stop_device）会加入 _stop_requested，阻止任何后续重试。 """ # 错峰启动：在各自线程内等待，分摊批量启动的连接/占用压力 if start_delay > 0: time.sleep(start_delay) try: for attempt in range(1, max_attempts + 1): # 用户已请求停止 → 不再启动新 attempt with self._lock: if serial in self._stop_requested: _log.info(f"{serial} 用户已请求停止，取消重试 (job={job.name})") return # 同一 serial 同时只能一个 worker with self._lock: if serial in self._running: _log.warning(f"{serial} 已有任务在跑，跳过 (job={job.name})") return self._running[serial] = {"job_id": job.id, "started_at": time.time(), "attempt": attempt, "task_type": job.task_type} _update_status(serial, task_job=job.name, attempt=attempt, max_attempts=max_attempts) worker = None try: worker = task.create_worker(self.stf, serial, job.params) with self._lock: self._running[serial]["worker"] = worker _log.info(f"{serial} 开始任务 {job.name} (第{attempt}/{max_attempts}次)") worker.start() worker.join() # 等待 worker 结束 # worker 正常结束（done 或被 stop） with self._lock: self._running.pop(serial, None) # 判断是否成功：看 status with _WORKERS_LOCK: st = _WORKERS.get(serial, {}).get("status") # 清除 task_job 标记，避免前端误判"运行中" _update_status(serial, task_job="") if st == "done": _log.info(f"{serial} 任务 {job.name} 成功完成") return # 用户请求停止（无论 attempt 第几次、status 是什么）→ 不重试 with self._lock: if serial in self._stop_requested: _log.info(f"{serial} 用户已请求停止，不再重试 (job={job.name})") return _log.warning(f"{serial} 任务未成功(status={st})") except DeviceOfflineError as e: # 设备掉线，立即放弃，不重试 _log.error(f"{serial} 设备离线，放弃任务 {job.name}: {e}") with self._lock: self._running.pop(serial, None) _update_status(serial, status="failed", task_job="", last_error=f"设备离线: {e}") return except Exception as e: _log.error(f"{serial} 执行异常: {e}", exc_info=True) with self._lock: self._running.pop(serial, None) _update_status(serial, task_job="") if attempt < max_attempts: # 用户在 sleep 期间点停止也能中断 with self._lock: if serial in self._stop_requested: _log.info(f"{serial} 用户已请求停止，取消重试 (job={job.name})") return # 检查是否是端口耗尽类临时错误，需要更长退避等端口释放 with _WORKERS_LOCK: err = _WORKERS.get(serial, {}).get("last_error", "") if err.startswith("[transient]"): # Windows TCP 端口耗尽，TIME_WAIT 默认 2-4 分钟，等 120 秒 extra_delay = max(delay, 120) _log.info(f"{serial} ADB 连接临时错误（端口耗尽），{extra_delay}s 后重试 ({attempt+1}/{max_attempts})") time.sleep(extra_delay) else: _log.info(f"{serial} {delay}s 后重试 ({attempt+1}/{max_attempts})") time.sleep(delay) _log.error(f"{serial} 任务 {job.name} 重试耗尽，放弃") _update_status(serial, status="failed", last_error=f"{job.name} 重试{max_attempts}次失败") finally: # 清除停止标志：整个重试循环结束（成功/失败/停止）后允许下次任务 with self._lock: self._stop_requested.discard(serial) # ---- 运行控制 ---- def stop_device(self, serial): """停止指定设备的 worker，并阻止后续重试。 无论 worker 当前在运行还是在重试 sleep 中，都会阻止下一次重试。 """ with self._lock: self._stop_requested.add(serial) info = self._running.get(serial) if info: w = info.get("worker") if w and w.is_alive(): w.stop() return True # worker 已结束但重试循环可能还在 sleep —— 仍然返回 True 表示已阻止重试 return True def stop_all(self): """停止所有运行中的 worker，并阻止后续重试。""" with self._lock: items = list(self._running.items()) for s, _ in items: self._stop_requested.add(s) stopped = [] for serial, info in items: w = info.get("worker") if w and w.is_alive(): w.stop() stopped.append(serial) return stopped def get_running(self): with self._lock: return {s: dict(v) for s, v in self._running.items()} # ---- 状态查询（带缓存，避免 STF 请求阻塞前端）---- _status_cache = None # (timestamp, data, error) _status_cache_lock = threading.Lock() _STATUS_CACHE_TTL = 5.0 # 缓存 5 秒，前端 5 秒刷新刚好命中 def get_status(self): """综合状态：STF 设备池 + 本地 worker + 运行中的任务。 带 5 秒缓存：STF 请求慢时避免每次 /api/status 都打 STF 阻塞 Flask。 worker 状态实时读（内存，无 IO），不受缓存影响。 """ with self._status_cache_lock: if self._status_cache: ts, cached, err = self._status_cache if time.time() - ts < self._STATUS_CACHE_TTL: if err: return None, err # 用缓存的设备列表 + 实时 worker 状态重新组装 return self._merge_status(cached), None # 缓存过期或不存在，重新拉 STF try: all_devices = self.stf.list_all_devices() except Exception as e: with self._status_cache_lock: self._status_cache = (time.time(), None, str(e)) return None, f"获取设备列表失败: {e}" with self._status_cache_lock: self._status_cache = (time.time(), all_devices, None) return self._merge_status(all_devices), None def _merge_status(self, all_devices): """用 STF 设备列表 + 实时 worker 状态 + 前台 App 组装返回结果。""" worker_status = {w["serial"]: w for w in get_all_worker_status()} running = self.get_running() # 清理陈旧状态：serial 已不在 STF 设备池、且没有在跑 worker 的条目， # 避免设备被删除后其失败记录仍残留在"异常汇总"里 present = {dev.get("serial", "") for dev in all_devices} for serial, w in list(worker_status.items()): if serial in present or serial in running: continue if w.get("status") in ("running", "connecting"): continue _remove_worker(serial) result = [] for dev in all_devices: serial = dev.get("serial", "") owner = dev.get("owner") w = worker_status.get(serial, {}) r = running.get(serial, {}) result.append({ "serial": serial, "model": dev.get("model") or w.get("model", "") or dev.get("product", ""), "device_name": dev.get("name") or "", "present": dev.get("present", False), "ready": dev.get("ready", False), "stf_occupied": dev.get("using", False), "owner": owner.get("name", "") if owner else "", "worker_status": w.get("status", "idle"), "foreground_app": self._fg_scanner.get(serial), # 通用进度字段（任意 app 通用，前端统一解析展示） # 结构：{"done": int, "total": int, "unit": str, "action_counts": dict} "progress": w.get("progress", {}), "current_action": w.get("current_action", ""), "last_error": w.get("last_error", ""), "last_warning": w.get("last_warning", ""), "running_job": r.get("job_id", ""), "task_job": w.get("task_job", ""), "attempt": r.get("attempt", 0), "end_time": w.get("end_time", 0), }) return result def list_all_serials(self): """返回 STF 上所有在线设备的 serial 列表（供分组表单勾选用）。 复用 get_status 缓存，避免开页面时阻塞。 """ devices, err = self.get_status() if err or not devices: return [] return [d["serial"] for d in devices if d.get("present")] def shutdown(self): self.stop_all() self._fg_scanner.stop() stop_watchdog() self.scheduler.shutdown(wait=False)