"""歌曲表**自动维护**：把"补队下载 / 扫描登记 / 失败重试 / 设备身份时效 / 心跳"合并成**一个**后台任务。

## 为什么合并（2026 新语义）
改之前是**三个独立守护线程**（`BatchManager` / `media_scan_scheduler` / `FailedRetry`）再加上
`WhoamiReporter`，各自定时、各自写日志 —— 谁在跑、跑到哪一步、为什么没动静，界面上说不清
（线上出现过"下载 worker 不动作"，但三处都有各自的定时器，排查时互相干扰）。
现在只有**一条时间线**，每 `maintain_interval_sec`（默认 600 = 10 分钟）跑一轮：

    a. 身份时效检查 → b. 扫描媒体目录（登记新文件 + done 但文件缺失/过小 → 回队）→ c. 失败重试
    → d. 按歌曲表补队下载 → e. 等队列排空（drain）→ f. 记结果 + 算 `next_run_at`

## 部署机硬性要求（本文件里处处体现）
媒体目录在**云盘 FUSE** 上，`stat`/`listdir` 可能长时间阻塞，历史上多次因此出现
"进程活着但不监听端口"、"下载 worker 不动作"。因此：
  * 访问媒体目录一律走 `store.keep_adopt_media()`（内部是**带超时的守护线程**），
    超时/异常只记 `scan_error`，**不影响本轮后续步骤**；
  * 网络动作（whoami）只在**后台线程**里做，且只在"本轮确实要下载"时启停；
  * 本任务自身是**守护线程**，而且必须在 `main.py` 里**两个 HTTP 端口监听之后**才 `start()`；
  * `run_now()` 只负责"起线程"，**立刻返回**，绝不在 HTTP 请求线程里同步跑长任务。

## 状态机（`state`）
`stopped`（未运行）/ `idle`（运行中，在等下一轮）/ `checking_identity` / `scanning` /
`retrying` / `downloading` / `error`（本轮出现未预期异常）。

## 心跳（whoami）与身份时效
  * 心跳**不再独立定时**：由本任务在 `downloading` 之前 `start()`，drain 结束在 `finally` 里
    `stop()`（成功/失败/超时/异常都停）；本轮没有要下载的歌 → **一次都不启动**；
    `whoami_enabled=False` 时任何情况下都不启动；一次长下载期间仍按 `next_interval` 周期上报
    （`WhoamiReporter.loop()` 自带）；
  * 身份：`device-identity.json` 的 `created_at` 距今 > `device_identity_ttl_hours`（默认 24）
    → 自动 `regenerate()` 并写事件；换身份走 `identity.regenerate_identity()`（**一处**完成
    `apply_to_cfg` + 清 token），保证 `i.php` / `do.php` / whoami 三处立刻同源。
"""
from __future__ import annotations

import logging
import threading
import time

from . import identity as identity_mod

log = logging.getLogger("ktv.maintain")

# ── 状态机取值（界面按它显示中文，见 `STATE_TEXT`）──
STATE_STOPPED = "stopped"
STATE_IDLE = "idle"
STATE_IDENTITY = "checking_identity"
STATE_SCANNING = "scanning"
STATE_RETRYING = "retrying"
STATE_DOWNLOADING = "downloading"
STATE_ERROR = "error"

# 状态码 → 界面中文（"运行中 · 正在下载"这种口径；后端给一份，界面直接用，避免两处各写一套）
STATE_TEXT = {
    STATE_STOPPED: "未运行",
    STATE_IDLE: "运行中 · 空闲（等待下一轮）",
    STATE_IDENTITY: "运行中 · 检查设备身份",
    STATE_SCANNING: "运行中 · 扫描媒体目录",
    STATE_RETRYING: "运行中 · 重试失败任务",
    STATE_DOWNLOADING: "运行中 · 正在下载",
    STATE_ERROR: "出错（见错误信息）",
}

DEFAULT_INTERVAL = 600        # 默认每十分钟一轮
MIN_INTERVAL = 30            # 间隔下限（防止有人填 0/1 把轮询打疯）
DRAIN_POLL_SEC = 1.0         # drain 的轮询粒度（秒）；停止/超时都能及时退出
IDLE_POLL_SEC = 30           # 已停止/无排程时守护线程的睡眠粒度


class MaintainTask:
    """歌曲表自动维护的**唯一**后台任务（可单测：注入 clock / scan / retry / drain）。

    参数里的 `scan_fn` / `retry_fn` / `drain_fn` / `now_fn` / `sleep_fn` 只为**单测**注入
    （虚拟时钟、断言调用顺序、制造异常）；生产路径全部走默认实现。
    """

    def __init__(self, cfg, store, downloader=None, guard=None, whoami=None, identity=None,
                 scan_fn=None, retry_fn=None, drain_fn=None, now_fn=None):
        self.cfg = cfg
        self.store = store
        self.downloader = downloader
        self.guard = guard
        self.whoami = whoami
        self.identity = identity
        self._scan_fn = scan_fn if callable(scan_fn) else self._default_scan
        self._retry_fn = retry_fn if callable(retry_fn) else self._default_retry
        self._drain_fn = drain_fn if callable(drain_fn) else self._default_drain
        self._now = now_fn if callable(now_fn) else time.time
        self._lock = threading.RLock()
        self._stop = threading.Event()      # 守护线程生命周期（start 清、stop 置）
        self._abort = threading.Event()     # "中止当前这一轮"（stop 置；守护线程每轮开头清）
        self._cycle_lock = threading.Lock()  # 防重入：同一时刻只允许一轮
        self._thread: threading.Thread | None = None
        self._bg_threads: list[threading.Thread] = []   # run_now() 起的临时线程（供 join()）
        self.state: dict = {
            "enabled": False, "state": STATE_STOPPED, "fatal": False, "reason": "",
            "started_at": 0, "stopped_at": 0, "last_run_at": 0, "next_run_at": 0,
            "last_result": {}, "cycle_count": 0, "error": "", "running": False,
            "baseline_done": 0, "last_heartbeat": False,
        }

    # ────────────── 只读状态 ──────────────
    def _stats(self) -> dict:
        """队列统计；store 没有该方法/抛异常时返回空字典（绝不让状态查询崩）。"""
        fn = getattr(self.store, "stats", None)
        try:
            return dict(fn() or {}) if callable(fn) else {}
        except Exception:
            return {}

    def _keeps(self) -> dict:
        fn = getattr(self.store, "keep_stats", None)
        try:
            return dict(fn() or {}) if callable(fn) else {}
        except Exception:
            return {}

    def _keep_remaining(self) -> int:
        fn = getattr(self.store, "keep_remaining", None)
        if not callable(fn):
            return 0
        try:
            return int(fn() or 0)
        except Exception:
            return 0

    def _pending(self) -> int:
        return max(0, int(self._stats().get("pending") or 0))

    def _active(self) -> int:
        """下载池里**正在下载**的任务数（downloader 缺失/异常时按 0 算）。"""
        d = self.downloader
        if d is None:
            return 0
        fn = getattr(d, "status", None)
        if callable(fn):
            try:
                return max(0, int((fn() or {}).get("active_count") or 0))
            except Exception:
                pass
        try:
            return max(0, len(getattr(d, "_active", {}) or {}))
        except Exception:
            return 0

    def _heartbeat_running(self) -> bool:
        if self.whoami is None:
            return False
        fn = getattr(self.whoami, "status", None)
        if not callable(fn):
            return False
        try:
            return bool((fn() or {}).get("running"))
        except Exception:
            return False

    def _identity_dict(self) -> dict:
        """当前生效的设备身份（优先 `DeviceIdentity.identity`，否则配置里的运行时值）。"""
        obj = getattr(self.identity, "identity", None)
        if isinstance(obj, dict) and obj:
            return obj
        try:
            val = self.cfg.get("device_identity")
        except Exception:
            return {}
        return val if isinstance(val, dict) else {}

    def _ttl_hours(self) -> float:
        try:
            raw = self.cfg.get("device_identity_ttl_hours")
        except Exception:
            return 24.0
        if raw is None or raw == "":
            return 24.0
        try:
            return max(0.0, float(raw))
        except Exception:
            return 24.0

    def _interval(self) -> int:
        """本轮结束到下一轮的间隔（秒）；低于下限按 `MIN_INTERVAL` 处理（防忙轮询）。"""
        try:
            v = int(self.cfg.get("maintain_interval_sec") or DEFAULT_INTERVAL)
        except Exception:
            v = DEFAULT_INTERVAL
        return max(MIN_INTERVAL, v)

    def _drain_timeout(self) -> float:
        try:
            return max(0.0, float(self.cfg.get("maintain_drain_timeout_sec") or 3600))
        except Exception:
            return 3600.0

    def _heartbeat_enabled(self) -> bool:
        """是否允许在本轮启停心跳：维护开关 + whoami 总开关 + 装配了 reporter。"""
        if self.whoami is None:
            return False
        if not bool(self.cfg.get("maintain_heartbeat", True)):
            return False
        return bool(self.cfg.get("whoami_enabled", True))

    def status(self) -> dict:
        """完整状态（界面 4 秒轮询直接渲染；键一定齐全）。"""
        with self._lock:
            st = dict(self.state)
            last = dict(st.get("last_result") or {})
        stats = self._stats()
        ks = self._keeps()
        ident = self._identity_dict()
        ttl = self._ttl_hours()
        created = 0
        try:
            created = int((ident or {}).get("created_at") or 0)
        except Exception:
            created = 0
        age = max(0, int(self._now()) - created) if created > 0 else 0
        ttl_sec = int(ttl * 3600)
        expires_in = max(0, ttl_sec - age) if (created > 0 and ttl > 0) else 0
        state = str(st.get("state") or STATE_STOPPED)
        st.update({
            "ok": True,
            "enabled": bool(st.get("enabled")),
            "state": state,
            "state_text": STATE_TEXT.get(state, state),
            "interval_sec": self._interval(),
            "running": bool(self._thread is not None and self._thread.is_alive()),
            "last_run_at": int(st.get("last_run_at") or 0),
            "next_run_at": int(st.get("next_run_at") or 0),
            "last_result": last,
            "cycle_count": int(st.get("cycle_count") or 0),
            "checks": int(st.get("cycle_count") or 0),         # 兼容旧 /api/batch/status 的字段名
            "error": str(st.get("error") or ""),
            "heartbeat_running": self._heartbeat_running(),
            "heartbeat_enabled": self._heartbeat_enabled(),
            "identity_age_sec": age,
            "identity_expires_in_sec": expires_in,
            "identity_ttl_hours": ttl,
            "identity_expired": bool(created > 0 and ttl > 0 and age > ttl_sec),
            "identity_mac": str((ident or {}).get("mac") or ""),
            "done": int(stats.get("done") or 0),
            "pending": int(stats.get("pending") or 0),
            "failed": int(stats.get("failed") or 0),
            "library": int(stats.get("songs") or 0),
            "enqueued": int(stats.get("enqueued") or 0),
            "baseline_done": int(st.get("baseline_done") or 0),
            "keeps_total": int(ks.get("total") or 0),
            "keeps_want": int(ks.get("want") or 0),
            "keeps_done": int(ks.get("done") or 0),
            "keeps_failed": int(ks.get("failed") or 0),
            "keeps_bytes": int(ks.get("bytes") or 0),
            "keeps_remaining": self._keep_remaining(),
        })
        return st

    # ────────────── 启停 ──────────────
    def start(self) -> dict:
        """启动（**立即返回**）：第一轮马上跑，之后每 `interval` 秒一轮。"""
        now = int(self._now())
        with self._lock:
            if self.state.get("enabled") and self._thread is not None and self._thread.is_alive():
                return {"ok": False, "error": "歌曲表自动维护已在运行"}
            self.state.update({
                "enabled": True, "state": STATE_IDLE, "fatal": False, "reason": "",
                "started_at": now, "stopped_at": 0, "error": "",
                "next_run_at": now, "baseline_done": int(self._stats().get("done") or 0),
                "cycle_count": 0,
            })
        # 先让可能还活着的老循环退出，再清停止位 —— 顺序反了会出现"两个循环线程"。
        self._stop.set()
        old = self._thread
        if old is not None and old.is_alive():
            old.join(timeout=1.0)
        self._abort.clear()
        self._stop.clear()
        self._thread = threading.Thread(target=self.loop, name="maintain", daemon=True)
        self._thread.start()
        self._note("INFO", f"歌曲表自动维护已启动：每 {self._interval()} 秒一轮"
                          f"（扫描={bool(self.cfg.get('maintain_scan_media', True))}、"
                          f"重试={bool(self.cfg.get('maintain_retry_failed', True))}、"
                          f"心跳={self._heartbeat_enabled()}）")
        return {"ok": True, "enabled": True, "next_run_at": now, "state": STATE_IDLE}

    def stop(self, reason: str = "手动停止") -> dict:
        """停止：置停止位 + **立刻停心跳**；drain 中调用也会尽快退出（状态 `stopped`）。"""
        with self._lock:
            self.state.update({
                "enabled": False, "state": STATE_STOPPED, "next_run_at": 0,
                "stopped_at": int(self._now()), "reason": str(reason or "手动停止"),
                "running": False,
            })
        self._stop.set()
        self._abort.set()
        self._stop_heartbeat()
        self._note("INFO", f"歌曲表自动维护已停止：{reason}")
        return {"ok": True, "enabled": False, "state": STATE_STOPPED, "reason": str(reason or "手动停止")}

    def restart(self, reason: str = "重启") -> dict:
        """重启 = 停止后**立即开始新的一轮**（不等 interval）。"""
        self.stop(reason=f"重启：{reason}")
        res = self.start()
        with self._lock:
            self.state["next_run_at"] = int(self._now())      # 立即，不等 10 分钟
        res.update({"restart": True, "state": STATE_IDLE,
                    "next_run_at": int(self.state.get("next_run_at") or 0)})
        return res

    def run_now(self) -> dict:
        """**立即跑一轮，不影响排程**：起后台线程后立刻返回（绝不在请求线程里跑长任务）。"""
        t = threading.Thread(target=self._run_now_bg, name="maintain-run-now", daemon=True)
        with self._lock:
            self._bg_threads = [x for x in self._bg_threads if x.is_alive()]
            self._bg_threads.append(t)
        t.start()
        with self._lock:
            return {"ok": True, "started": True, "state": str(self.state.get("state") or STATE_STOPPED),
                    "next_run_at": int(self.state.get("next_run_at") or 0)}

    def _run_now_bg(self) -> None:
        try:
            self.run_cycle(update_schedule=False, reset_abort=True)
        except Exception as exc:                              # pragma: no cover - 兜底
            self._note("ERROR", f"维护（立即一次）异常：{exc}")
        finally:
            with self._lock:
                self._bg_threads = [x for x in self._bg_threads
                                    if x.is_alive() and x is not threading.current_thread()]

    def join(self, timeout: float = 3.0) -> None:
        """等**所有**维护线程（调度循环 + 正在跑的 `run_now`）结束。

        生产上不必须（daemon 线程随进程退出）；但"关数据库/落盘"之前调用它，
        可以避免"子线程还在写事件时父线程关连接"这类竞争（sqlite 同连接跨线程会段错误）。
        """
        end = time.time() + max(0.0, float(timeout))
        with self._lock:
            threads = [self._thread] + list(self._bg_threads)
        for th in threads:
            if th is None or not th.is_alive():
                continue
            remain = end - time.time()
            if remain <= 0:
                return
            th.join(timeout=remain)
        remain = end - time.time()
        if remain > 0 and self._cycle_lock.acquire(timeout=remain):
            self._cycle_lock.release()

    # ────────────── 调度循环 ──────────────
    def loop(self) -> None:
        while not self._stop.is_set():
            with self._lock:
                nxt = int(self.state.get("next_run_at") or 0)
            wait = nxt - int(self._now())
            if wait > 0:
                if self._stop.wait(min(wait, IDLE_POLL_SEC)):
                    return
                continue
            try:
                self.run_cycle()
            except Exception as exc:                          # pragma: no cover - run_cycle 自带兜底
                self._note("ERROR", f"维护循环异常：{exc}")
                if self._stop.wait(IDLE_POLL_SEC):
                    return

    # ────────────── 一轮 ──────────────
    def run_cycle(self, now=None, update_schedule: bool = True, reset_abort: bool = False) -> dict:
        """跑一轮（顺序固定，见模块顶部）；返回本轮结果字典（可直接放进 `last_result`）。

        * `now`：虚拟时钟入口（单测不真等 10 分钟）；不传就取 `now_fn()`；
        * `update_schedule=False`：`run_now()` 用 —— 跑完**不改** `next_run_at`（不影响排程）；
        * `reset_abort=True`：`run_now()` 在任务已停止时也要完整跑一轮，故先清中止位。
        """
        if reset_abort:
            self._abort.clear()
        started = int(now if now is not None else self._now())
        if not self._cycle_lock.acquire(blocking=False):
            return {"ok": False, "error": "上一轮维护还在进行中",
                    "state": str(self.state.get("state") or STATE_STOPPED)}
        result: dict = {
            "started_at": started, "ended_at": started, "elapsed_ms": 0,
            "identity_regenerated": False, "identity_age_sec": 0, "identity_expires_in_sec": 0,
            "identity_mac": "", "scanned": 0, "adopted": 0, "updated": 0, "skipped": 0,
            "requeued": 0,
            "retried": 0, "enqueued": 0, "done": 0, "failed": 0,
            "pending_left": 0, "active_left": 0, "heartbeat": False, "drain_timeout": False,
            "scan_error": "", "retry_error": "", "error": "",
        }
        heartbeat_on = False
        heartbeat_attempted = False
        done_before = 0
        try:
            with self._lock:
                self.state.update({"running": True, "error": ""})
            stats0 = self._stats()
            done_before = int(stats0.get("done") or 0)

            # a. 身份时效检查（超时/异常不影响后续步骤）
            self._set_state(STATE_IDENTITY)
            ident = self._check_identity(started)
            result.update({
                "identity_regenerated": bool(ident.get("regenerated")),
                "identity_age_sec": int(ident.get("age_sec") or 0),
                "identity_expires_in_sec": int(ident.get("expires_in_sec") or 0),
                "identity_mac": str(ident.get("mac") or ""),
                "identity_error": str(ident.get("error") or ""),
            })

            # b. 扫描媒体目录（登记新文件 + "done 但文件缺失/过小"回队；自带超时守护线程，失败只记日志）
            if bool(self.cfg.get("maintain_scan_media", True)):
                self._set_state(STATE_SCANNING)
                try:
                    res = self._scan_fn() or {}
                    result["scanned"] = int(res.get("scanned") or 0)
                    result["adopted"] = int(res.get("adopted_count",
                                                    len(res.get("adopted") or [])) or 0)
                    result["updated"] = int(res.get("updated") or 0)
                    result["skipped"] = int(res.get("skipped") or 0)
                    result["requeued"] = int(res.get("requeued") or 0)
                    if res.get("ok") is False:
                        result["scan_error"] = str(res.get("error") or "扫描失败")
                except Exception as exc:
                    result["scan_error"] = str(exc)
                    self._note("WARN", f"维护：扫描媒体目录失败（只记日志，不影响本轮后续）：{exc}")

            # c. 失败重试（failed/missing → 待下载）
            if bool(self.cfg.get("maintain_retry_failed", True)):
                self._set_state(STATE_RETRYING)
                try:
                    result["retried"] = int(self._retry_fn() or 0)
                except Exception as exc:
                    result["retry_error"] = str(exc)
                    self._note("WARN", f"维护：失败重试失败（只记日志，不影响本轮后续）：{exc}")

            # d. 按歌曲表补队（沿用 keep_next_batch + enqueue，绝不自己实现一套）
            result["enqueued"] = int(self._top_up() or 0)

            # e. **确实有歌要下**才起心跳 + 等排空；没有要下载的歌 → 一次都不启动心跳
            if self._pending() > 0 or self._active() > 0:
                self._ensure_pool()
                if self._heartbeat_enabled():
                    heartbeat_attempted = True
                    try:
                        self.whoami.start()               # 进入 downloading **之前**
                        heartbeat_on = True
                        result["heartbeat"] = True
                    except Exception as exc:
                        self._note("WARN", f"维护：whoami 心跳启动失败（不影响下载）：{exc}")
                self._set_state(STATE_DOWNLOADING)
                try:
                    drained = self._drain_fn(self._drain_timeout()) or {}
                    result["drain_timeout"] = bool(drained.get("drain_timeout"))
                    result["pending_left"] = int(drained.get("pending_left") or 0)
                    result["active_left"] = int(drained.get("active_left") or 0)
                except Exception as exc:
                    result["drain_error"] = str(exc)
                    self._note("WARN", f"维护：等待下载排空异常（已停心跳，进入下一轮）：{exc}")
        except Exception as exc:
            result["error"] = f"{type(exc).__name__}: {exc}"
            with self._lock:
                self.state["error"] = result["error"]
            self._note("ERROR", f"维护：本轮异常：{exc}")
        finally:
            try:
                # ⚠️ 无论成功、失败、超时还是异常，**心跳必须停**（只有起过的轮次才需要停）
                if heartbeat_on or heartbeat_attempted:
                    self._stop_heartbeat()
                stats1 = self._stats()
                result["done"] = max(0, int(stats1.get("done") or 0) - done_before)
                result["failed"] = int(stats1.get("failed") or 0)
                result["pending_left"] = self._pending()
                result["active_left"] = self._active()
                ended = int(self._now())
                result["ended_at"] = ended
                result["elapsed_ms"] = max(0, (ended - started) * 1000)
                result["ok"] = not result.get("error")
                with self._lock:
                    self.state["cycle_count"] = int(self.state.get("cycle_count") or 0) + 1
                    self.state["last_run_at"] = started
                    self.state["last_result"] = dict(result)
                    self.state["last_heartbeat"] = bool(result.get("heartbeat"))
                    self.state["running"] = False
                    self.state["state"] = (STATE_ERROR if result.get("error")
                                           else (STATE_IDLE if self.state.get("enabled")
                                                 else STATE_STOPPED))
                    if update_schedule:
                        # next_run_at = **本轮结束** + interval（间隔 0 视为"不再自动排下一轮"→ 0）
                        interval = self._interval()
                        self.state["next_run_at"] = (ended + interval) if self.state.get("enabled") else 0
                    result["state"] = self.state["state"]
                    result["next_run_at"] = int(self.state.get("next_run_at") or 0)
            finally:
                self._cycle_lock.release()
            self._note("INFO", self._summary(result))
        return result

    # ────────────── 各步骤实现 ──────────────
    def _set_state(self, state: str) -> None:
        """切换阶段状态；**已停止/已中止时不再改**。

        否则 `stop()` 之后仍在收尾的那一轮（例如刚 `run_now()` 过）会把状态从 `stopped`
        改回 `scanning`/`retrying`，界面上看起来"点了停止却还在跑"。
        """
        if self._abort.is_set():
            return
        with self._lock:
            self.state["state"] = state

    def _check_identity(self, now: int) -> dict:
        """身份时效：超期就自动重新生成（并写事件）；未装配身份管理时直接跳过。"""
        out = {"checked": False, "regenerated": False, "age_sec": 0, "expires_in_sec": 0,
               "mac": "", "error": ""}
        if self.identity is None or not callable(getattr(self.identity, "regenerate", None)):
            return out
        ident = self._identity_dict()
        if not isinstance(ident, dict) or not ident:
            return out
        out["checked"] = True
        try:
            created = int(ident.get("created_at") or 0)
        except Exception:
            created = 0
        ttl = self._ttl_hours()
        ttl_sec = int(ttl * 3600)
        age = max(0, int(now) - created) if created > 0 else 0
        out.update({"age_sec": age, "mac": str(ident.get("mac") or ""),
                    "expires_in_sec": max(0, ttl_sec - age) if (created > 0 and ttl > 0) else 0})
        if created <= 0 or ttl <= 0 or age <= ttl_sec:
            return out
        old_mac = str(ident.get("mac") or "")
        try:
            # 一处完成"生成 + 落盘 + apply_to_cfg + 清 token"（i.php/do.php/whoami 同时换新身份）
            new = identity_mod.regenerate_identity(self.cfg, self.identity, self.downloader)
        except Exception as exc:
            out["error"] = f"{type(exc).__name__}: {exc}"
            self._note("WARN", f"维护：设备身份已过期但自动更换失败（本轮继续）：{exc}")
            return out
        out.update({"regenerated": True, "age_sec": 0, "expires_in_sec": ttl_sec,
                    "mac": str(new.get("mac") or ""), "old_mac": old_mac})
        self._note("WARN", f"设备身份已过期（{age / 3600.0:.1f}h），自动更换："
                           f"新 mac={new.get('mac')} sn={new.get('sn')}")
        return out

    def _default_scan(self) -> dict:
        from . import admin                               # 延迟导入，避免模块级循环依赖
        return admin.run_media_scan(self.cfg, self.store)

    def _max_attempts(self) -> int:
        fn = getattr(self.cfg, "retry_failed_max_attempts", None)
        if callable(fn):
            try:
                return int(fn())
            except Exception:
                pass
        try:
            return max(1, int(self.cfg.get("max_attempts") or 3))
        except Exception:
            return 3

    def _default_retry(self) -> int:
        return int(self.store.retry_failed(self._max_attempts()) or 0)

    def _top_up(self) -> int:
        """按歌曲表补队：只补 `keeps` 里未完成的歌（沿用 store.keep_next_batch）。"""
        target = max(1, int(self.cfg.get("batch_target_pending") or 200))
        pending = self._pending()
        if pending >= target:
            return 0
        fn = getattr(self.store, "keep_next_batch", None)
        if not callable(fn):
            return 0
        order = str(self.cfg.get("download_order") or "hot")
        desc = bool(self.cfg.get("download_order_desc", True))
        try:
            cands = fn(target - pending, order, desc)
        except Exception as exc:
            self._note("WARN", f"维护：补队候选查询失败（不影响本轮）：{exc}")
            return 0
        if not cands:
            return 0
        res = self.store.enqueue([int(c["tid"]) for c in cands])
        return int((res or {}).get("added") or 0)

    def _ensure_pool(self) -> None:
        """确保下载池在跑（空间原因暂停过就先恢复）；失败只记日志。"""
        d = self.downloader
        if d is None:
            return
        try:
            if getattr(d, "paused_by_space", False):
                d.resume(by="space")
        except Exception as exc:
            self._note("WARN", f"维护：恢复下载池失败：{exc}")
        try:
            if not getattr(d, "_threads", None):
                d.start()
        except Exception as exc:
            self._note("WARN", f"维护：启动下载池失败：{exc}")

    def _default_drain(self, timeout_sec: float) -> dict:
        """等队列排空：`pending == 0 且 active == 0`，或超时；`stop()` 也会立刻退出。"""
        end = self._now() + max(0.0, float(timeout_sec))
        timed_out = False
        while not self._abort.is_set():
            if self._pending() <= 0 and self._active() <= 0:
                break
            remain = end - self._now()
            if remain <= 0:
                timed_out = True
                break
            if self._abort.wait(min(DRAIN_POLL_SEC, max(0.05, remain))):
                break
        return {"drain_timeout": timed_out, "pending_left": self._pending(),
                "active_left": self._active()}

    def _stop_heartbeat(self) -> None:
        if self.whoami is None:
            return
        try:
            self.whoami.stop()
        except Exception:                                     # pragma: no cover - 兜底
            pass

    def _summary(self, result: dict) -> str:
        """一轮结束写一条 INFO 事件（界面「上一轮结果」与日志页都看它）。"""
        parts = [f"扫描登记 {result.get('adopted', 0)}",
                 f"缺失回队 {result.get('requeued', 0)}",
                 f"重试 {result.get('retried', 0)}",
                 f"入队 {result.get('enqueued', 0)}",
                 f"完成 {result.get('done', 0)}",
                 f"失败 {result.get('failed', 0)}"]
        if result.get("scan_error"):
            parts.append(f"扫描失败（{result['scan_error'][:60]}）")
        if result.get("retry_error"):
            parts.append(f"重试失败（{result['retry_error'][:60]}）")
        if result.get("drain_timeout"):
            parts.append("本轮等待排空超时，下一轮继续")
        extra = f"，下一轮 {self.state.get('next_run_at')}" if self.state.get("next_run_at") else ""
        return (f"维护一轮完成（第 {self.state.get('cycle_count', 0)} 轮）："
                + "、".join(parts) + extra)

    def _note(self, level: str, text: str) -> None:
        """写日志 + 事件；**任何异常都吞掉**（日志/事件绝不能影响维护主流程）。"""
        try:
            log.info("%s", text)
        except Exception:                                     # pragma: no cover - 兜底
            pass
        if self.store is None:
            return
        try:
            self.store.event(level, text)
        except Exception:                                     # pragma: no cover - 兜底
            pass