auto_control/core/task_manager.py

"""通用任务管理框架：任务类型注册 + 设备分组 + 任务计划 + 定时调度 + 重试 + 持久化。

设计目标：可扩展，未来加非抖音任务只需注册新的 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")


# ================== 设备分组 ==================
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 <serial> → 查询 → 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（含重试循环）。"""
        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 serial in serials:
            # 每台设备一个重试循环线程，互不影响
            t = threading.Thread(target=self._run_with_retry,
                                 args=(task, serial, job, max_attempts, delay), daemon=True)
            t.start()

    def _run_with_retry(self, task, serial, job, max_attempts, delay):
        """单设备任务执行 + 重试。

        异常分类：
          DeviceOfflineError — 设备掉线，立即放弃不重试（换设备也没用）
          其他异常           — 按 max_attempts 重试
        用户停止（stop_device）会加入 _stop_requested，阻止任何后续重试。
        """
        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()

        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", ""),
                "running_job": r.get("job_id", ""),
                "task_job": w.get("task_job", ""),
                "attempt": r.get("attempt", 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)