"""歌曲表（keeps）批量下载驱动器：**持续补队 + 定时巡检 + 致命错误自动停止**。 设计要点（对应"服务器只按歌曲表下载/更新文件"）： 1. **持续补队（只补歌曲表里的）**：保持队列里始终有 `batch_target_pending` 首待下载 （默认 200），每 `batch_check_interval_sec` 补一次；补队来源是 `store.keep_next_batch()` —— 即 `keeps` 里 `state NOT IN ('done','downloading')` 的歌，不在歌曲表里的歌永远不会被下载。 2. **定时巡检**：每轮采样水位/活动任务/完成数；每 `batch_report_interval_sec` 往 事件日志写一条 `BATCH 进度`（% / 速率 / 预计剩余 / 本地与云盘水位），界面可查。 3. **致命错误即刻停止**（停止后不再补队，并暂停下载池、写入 `FATAL` 事件）： - 空间不足**连续** `batch_low_space_checks` 次（默认 5 次 ≈ 2.5 分钟）仍未缓解； - 连续失败次数 ≥ `batch_max_consecutive_failures`（默认 20）； - 长时间无进展：`batch_stale_minutes`（默认 20 分钟）内 0 完成**且队列非空** （队列为空说明只是"没有歌要下"，不算致命）。 4. **歌曲表里的全部下完 = 正常收工**：队列为空且没有未完成的歌曲表条目时，写一条 `INFO BATCH 已按歌曲表下载完毕` 并以 `fatal=False` 正常停止（不是致命错误）。 """ from __future__ import annotations import threading import time class BatchManager: def __init__(self, cfg, store, downloader, guard=None): self.cfg = cfg self.store = store self.downloader = downloader self.guard = guard self._lock = threading.Lock() self._stop = threading.Event() self._thread: threading.Thread | None = None self.state = { "enabled": False, "fatal": False, "reason": "", "started_at": 0, "stopped_at": 0, "last_check": 0, "last_report": 0, "checks": 0, "low_space_streak": 0, "consecutive_failures": 0, "last_done": 0, "last_progress_at": 0, "baseline_done": 0, "enqueued_total": 0, "retries": 0, "next_retry_at": 0, "retry_timer": None, } # ── 只读状态 ── def _keeps(self) -> dict: """歌曲表统计（旧假 store 没有该方法时返回空字典，绝不因此让巡检崩溃）。""" fn = getattr(self.store, "keep_stats", None) try: return fn() if callable(fn) else {} except Exception: return {} def _keep_remaining(self) -> int | None: """歌曲表里还没下完的数量；`None` = store 没有这个方法（此时不判"收工"）。""" fn = getattr(self.store, "keep_remaining", None) if not callable(fn): return None try: return int(fn()) except Exception: return None def _keep_candidates(self, limit: int, order: str, desc: bool) -> list[dict]: """补队候选：**只来自歌曲表**（keeps 里未完成的歌）。""" fn = getattr(self.store, "keep_next_batch", None) if not callable(fn): return [] try: return fn(int(limit), order, desc) except Exception: return [] def status(self) -> dict: with self._lock: st = dict(self.state) stats = self.store.stats() st.update({"done": stats["done"], "pending": stats["pending"], "failed": stats["failed"], "library": stats["songs"], "enqueued": stats["enqueued"]}) st["remaining"] = max(0, stats["songs"] - stats["done"]) # 歌曲表维度（界面显示 待下/已下/失败；下载只认歌曲表） ks = self._keeps() st.update({"keeps_want": int(ks.get("want", 0)), "keeps_done": int(ks.get("done", 0)), "keeps_failed": int(ks.get("failed", 0)), "keeps_total": int(ks.get("total", 0)), "keeps_bytes": int(ks.get("bytes", 0)), "keeps_remaining": self._keep_remaining() or 0}) return st # ── 启停 ── def start(self) -> dict: with self._lock: if self.state["enabled"] and not self.state["fatal"]: return {"ok": False, "error": "批量任务已在运行"} stats = self.store.stats() self.state.update({"enabled": True, "fatal": False, "reason": "", "started_at": int(time.time()), "stopped_at": 0, "checks": 0, "low_space_streak": 0, "consecutive_failures": 0, "last_done": stats["done"], "baseline_done": stats["done"], "last_progress_at": int(time.time()), "enqueued_total": 0}) self._stop.clear() if self.downloader is not None and self.downloader.paused_by_space: self.downloader.resume(by="space") if self.downloader is not None and not self.downloader._threads: self.downloader.start() if self._thread is None or not self._thread.is_alive(): self._thread = threading.Thread(target=self.loop, name="batch", daemon=True) self._thread.start() self.store.event("INFO", "BATCH 已启动：按排序持续补队下载**歌曲表（keeps）**里的歌" "（致命错误会自动停止）") return {"ok": True} def stop(self, reason: str = "手动停止", fatal: bool = False) -> dict: with self._lock: if not self.state["enabled"] and not fatal: return {"ok": False, "error": "批量任务未在运行"} self.state.update({"enabled": False, "fatal": fatal, "reason": reason, "stopped_at": int(time.time())}) if not fatal: self.state["next_retry_at"] = 0 # 人工停止 = 取消后续自动重试 self._stop.set() if self.downloader is not None and fatal: self.downloader.pause(by="batch") self.store.event("ERROR" if fatal else "INFO", f"BATCH {'致命错误停止' if fatal else '已停止'}：{reason}") if fatal and bool(self.cfg.get("batch_fatal_auto_retry")): self._schedule_retry(reason) return {"ok": True, "fatal": fatal, "reason": reason, "retry": self.state.get("next_retry_at", 0)} def _schedule_retry(self, reason: str) -> None: """致命停止后安排自动重试（上游限流/断网这类问题等一段时间会自愈）。""" with self._lock: retries = int(self.state["retries"]) + 1 max_retries = int(self.cfg.get("batch_fatal_max_retries") or 6) if retries > max_retries: self.store.event("ERROR", f"BATCH 自动重试已达上限（{max_retries} 次），彻底停止：{reason}") return wait = int(self.cfg.get("batch_fatal_retry_minutes") or 60) * 60 try: wait = int(wait * (float(self.cfg.get("batch_fatal_retry_backoff") or 2.0) ** (retries - 1))) except Exception: pass self.state.update({"retries": retries, "next_retry_at": int(time.time()) + wait}) def _retry(): time.sleep(max(5, wait)) if not self.state.get("next_retry_at"): return # 已被人工停止取消 if self.state.get("enabled"): return # 已经有人在跑 if bool(self.cfg.get("batch_retry_reset_failed")): n = self.store.retry_failed(int(self.cfg.get("max_attempts") or 3)) self.store.event("INFO", f"BATCH 重试前重置失败/卡住任务：{n} 首") self.store.event("INFO", f"BATCH 自动重试第 {retries} 次（上次致命原因：{reason}）") self.start() t = threading.Thread(target=_retry, name="batch-retry", daemon=True) t.start() with self._lock: self.state["retry_timer"] = id(t) self.store.event("WARN", f"BATCH 将在 {wait // 60} 分钟后自动重试（第 {retries} 次；原因：{reason}）") # ── 巡检 ── def evaluate(self, stats: dict, low_space: bool, now: int | None = None) -> str: """返回致命原因；空串表示健康。纯函数，便于测试。""" now = now or int(time.time()) max_fail = int(self.cfg.get("batch_max_consecutive_failures") or 20) low_need = int(self.cfg.get("batch_low_space_checks") or 5) stale_min = int(self.cfg.get("batch_stale_minutes") or 20) streak = self.state["low_space_streak"] + (1 if low_space else 0) if not low_space: streak = 0 if streak >= low_need: return f"存储空间不足持续 {streak} 次检查仍未缓解（本地/云盘低于下限）" if self.state["consecutive_failures"] >= max_fail: return f"连续失败 {self.state['consecutive_failures']} 首（≥{max_fail}）" if stats["pending"] > 0 and now - self.state["last_progress_at"] > stale_min * 60: return f"超过 {stale_min} 分钟没有任何新完成（队列仍有 {stats['pending']} 首待下载）" return "" def loop(self) -> None: while not self._stop.wait(self._check_interval()): try: if not self.tick(): return except Exception as exc: # pragma: no cover self.store.event("WARN", f"BATCH 巡检异常：{exc}") def _check_interval(self) -> int: return max(5, int(self.cfg.get("batch_check_interval_sec") or 30)) def tick(self) -> bool: """巡检一轮（补队 → 收工判定 → 定时报告）；返回 False 表示本轮已停止（循环应退出）。 独立成方法是为了可单测：不必真的等 `batch_check_interval_sec`（最小 5 秒）。 """ report = max(60, int(self.cfg.get("batch_report_interval_sec") or 600)) target = max(1, int(self.cfg.get("batch_target_pending") or 200)) stats = self.store.stats() low_space = bool(self.guard and not self.guard.state.get("ok", True)) now = int(time.time()) with self._lock: self.state["checks"] += 1 self.state["last_check"] = now if stats["done"] > self.state["last_done"]: self.state["last_done"] = stats["done"] self.state["last_progress_at"] = now self.state["consecutive_failures"] = 0 self.state["low_space_streak"] = (self.state["low_space_streak"] + 1) if low_space else 0 self.state["consecutive_failures"] = stats["failed"] reason = self.evaluate(stats, low_space, now) if reason: self.stop(reason, fatal=True) return False # 持续补队：**只补歌曲表（keeps）里的**（不在表里的一律不下） if stats["pending"] < target: need = target - stats["pending"] order = str(self.cfg.get("download_order") or "hot") desc = bool(self.cfg.get("download_order_desc", True)) cands = self._keep_candidates(need, order, desc) if cands: res = self.store.enqueue([int(c["tid"]) for c in cands]) with self._lock: self.state["enqueued_total"] += res["added"] # 歌曲表里的全部下完（队列空 + 歌曲表没有未完成条目）→ 正常收工，不是致命错误 if stats["pending"] == 0 and self._keep_remaining() == 0: self.store.event("INFO", "BATCH 已按歌曲表下载完毕") self.stop("已按歌曲表下载完毕", fatal=False) return False # 定时报告 if now - self.state["last_report"] >= report: with self._lock: self.state["last_report"] = now self.report(stats) return True def report(self, stats: dict) -> None: lib = max(1, stats["songs"]) done = stats["done"] elapsed = max(1, int(time.time()) - self.state["started_at"]) if self.state["started_at"] else 1 gained = max(0, done - self.state["baseline_done"]) rate = gained / elapsed * 3600.0 # 首/小时 remain = max(0, lib - done) eta_h = (remain / rate) if rate > 0 else -1 g = self.guard.state if self.guard else {} ks = self._keeps() self.store.event("INFO", f"BATCH 进度 {done}/{lib}（{done * 100.0 / lib:.2f}%）" f"，本轮 {gained} 首、{rate:.0f} 首/小时" + (f"、预计剩余 {eta_h / 24:.1f} 天" if eta_h >= 0 else "") + f"；队列 {stats['pending']} 待下载、失败 {stats['failed']}" + f"；歌曲表 待下 {ks.get('want', 0)}" f"（其中 {self._keep_remaining() or 0} 未完成）" f"、已下 {ks.get('done', 0)}、失败 {ks.get('failed', 0)}" + (f"；本地剩余 {g['local_free'] / 1e9:.1f}GB、云盘剩余 {g['media_free'] / 1e9:.1f}GB" if g else ""))