From 98e016ce2eae2d50f16a2846cc406751ffd6e074 Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Fri, 3 Jul 2026 10:52:09 +0200 Subject: [PATCH] fix: apply architecture-review fixes P2-P10 across update, downloader, core, parser, webui - update: snapshot DB before promote; rollback restores it only when the schema actually advanced, never from a stale on-disk fallback; snapshots are standalone (no WAL sidecars); dependency sync moved to apply time with a staged venv, plus boot-time venv reconcile after rollback; entrypoint restart loop forwards signals so docker stop stays graceful - downloader: AddResult carried through the client protocol; qB "Fails." only classified duplicate when the magnet hash is confirmed present; aria2 partial-success batches report ADDED, magnet followedBy gids are migrated and metadata stubs skipped in torrents_info; renamed paths persisted per gid so the rename loop converges - core: AppContext lifecycle serialized under one lock with running state derived from the scheduler; setup wizard starts background loops on completion - parser: LLM path gains episode_type classification (LLM results only), timeout/cache/concurrency/breaker controls, proxy-aware SDK clients closed after a grace period on config reload - webui: useApi returns an {ok} discriminant; config save and RSS refresh surface failures instead of false success - migrations: per-statement guards so a crash mid-migration recovers Implemented via Codex (P2-P10 from tasks/todo.md), then reviewed with a multi-agent verify pass; 10 confirmed review findings fixed on top. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FAxVyRwrY7z5NotTtgnwVs --- backend/src/boot_overlay.py | 107 +++++--- backend/src/module/api/setup.py | 1 + backend/src/module/api/update.py | 15 +- backend/src/module/core/context.py | 43 ++- backend/src/module/database/aria2.py | 75 ++++++ backend/src/module/database/migrations.py | 123 ++++++++- backend/src/module/downloader/__init__.py | 3 +- backend/src/module/downloader/base.py | 13 +- .../downloader/client/aria2_downloader.py | 149 ++++++++--- .../downloader/client/mock_downloader.py | 8 +- .../module/downloader/client/qb_downloader.py | 57 +++- .../src/module/downloader/download_client.py | 25 +- backend/src/module/models/aria2.py | 1 + backend/src/module/models/config.py | 21 ++ backend/src/module/parser/analyser/llm.py | 57 +++- backend/src/module/parser/title_parser.py | 110 +++++++- backend/src/module/rss/engine.py | 13 +- backend/src/module/update/updater.py | 177 ++++++++++++- backend/src/test/conftest.py | 8 +- backend/src/test/test_aria2_downloader.py | 133 +++++++++- backend/src/test/test_context.py | 62 ++++- backend/src/test/test_download_client.py | 19 +- backend/src/test/test_llm_parser.py | 36 ++- backend/src/test/test_migrations_module.py | 159 +++++++++++ backend/src/test/test_mock_downloader.py | 5 +- backend/src/test/test_qb_downloader.py | 72 +++++ backend/src/test/test_rss_engine_new.py | 31 +++ backend/src/test/test_setup.py | 4 +- backend/src/test/test_title_parser.py | 95 ++++++- backend/src/test/test_update.py | 247 ++++++++++++++++++ entrypoint.sh | 39 ++- webui/src/hooks/__tests__/useApi.test.ts | 31 ++- webui/src/hooks/useApi.ts | 22 +- webui/src/pages/index/config.vue | 7 +- webui/src/pages/index/rss.vue | 9 +- webui/src/store/config.ts | 8 +- 36 files changed, 1792 insertions(+), 193 deletions(-) diff --git a/backend/src/boot_overlay.py b/backend/src/boot_overlay.py index b8d5710c..344d04db 100644 --- a/backend/src/boot_overlay.py +++ b/backend/src/boot_overlay.py @@ -6,20 +6,18 @@ 逻辑：从持久卷 ``/app/config/updates/`` 读取 ``applied.json``，当覆盖层版本 **高于** 镜像基线版本（``/app/IMAGE_VERSION``）时，把覆盖层的 module 树覆盖到 -``/app/module``、前端 dist 覆盖到 ``/app/dist``，并在 ``uv.lock`` 相对镜像基线 -变化时 ``uv sync`` 同步依赖到 ``/app/.venv``；否则（镜像更新或相等）忽略并清除 -过期覆盖层，让镜像版本生效。 +``/app/module``、前端 dist 覆盖到 ``/app/dist``，并在更新包已预先准备好 +``.venv`` 时替换 ``/app/.venv``；否则（镜像更新或相等）忽略并清除过期覆盖层， +让镜像版本生效。 关键约束：本脚本位于 ``/app/boot_overlay.py``，**不在** 覆盖层会替换的 ``/app/module`` 里，因此始终是镜像自带的稳定版本在做决策。任何一步失败都不得 阻断启动——记录日志后回退到镜像版本继续启动。 """ -import hashlib import logging import os import shutil -import subprocess import sys from pathlib import Path @@ -31,8 +29,6 @@ logger = logging.getLogger("boot_overlay") APP_ROOT = Path("/app") UPDATES_ROOT = APP_ROOT / "config" / "updates" IMAGE_VERSION_PATH = APP_ROOT / "IMAGE_VERSION" -BASELINE_LOCK = APP_ROOT / "uv.lock" -VENV_PATH = APP_ROOT / ".venv" def _parse_semver(value: str): @@ -45,17 +41,6 @@ def _parse_semver(value: str): return None -def _sha256(path: Path): - try: - h = hashlib.sha256() - with open(path, "rb") as f: - for chunk in iter(lambda: f.read(1 << 20), b""): - h.update(chunk) - return h.hexdigest() - except OSError: - return None - - def _read_text(path: Path) -> str: try: return path.read_text(encoding="utf-8").strip() @@ -92,25 +77,66 @@ def _replace_tree(src: Path, dst: Path) -> None: shutil.rmtree(old) -def _uv_sync(backend_dir: Path) -> None: - """按覆盖层的 uv.lock 把依赖同步到应用 venv（best-effort）。""" +def _sha256_file(path: Path) -> str | None: + import hashlib + + try: + h = hashlib.sha256() + with path.open("rb") as f: + for chunk in iter(lambda: f.read(1 << 20), b""): + h.update(chunk) + return h.hexdigest() + except OSError: + return None + + +# 记录 /app/.venv 当前对应的 uv.lock 哈希；缺失即视为镜像基线 venv。 +_VENV_LOCK_MARKER = ".ab-lock-sha256" + + +def _sync_venv_to_lock( + current: Path, app_root: Path, baseline_lock: Path | None +) -> None: + """把 /app/.venv 与当前生效覆盖层的 uv.lock 对齐。 + + 正常路径下依赖在 apply 时已预装进 staged .venv，这里不会做任何事；只有 + 覆盖层带 uv.lock 却没有 staged .venv（回滚到旧覆盖层、或旧版 updater + 落盘的覆盖层）且 lock 与 venv 当前对应的 lock 不一致时才补一次 + uv sync——否则回滚后的旧代码会跑在新依赖上直接启动崩溃。 + 失败只记日志：用现有 venv 继续启动好过卡死在启动前。 + """ + overlay_lock = current / "backend" / "uv.lock" + if not overlay_lock.exists(): + return + image_lock = baseline_lock if baseline_lock is not None else app_root / "uv.lock" + desired = _sha256_file(overlay_lock) + marker = app_root / ".venv" / _VENV_LOCK_MARKER + actual = _read_text(marker) or ( + _sha256_file(image_lock) if image_lock.exists() else None + ) + if desired is None or actual == desired: + return + uv = shutil.which("uv") if not uv: - logger.error("uv not found on PATH; cannot sync dependencies for the overlay") + logger.error("uv not found; cannot reconcile venv with overlay lockfile.") return + + import subprocess + env = dict(os.environ) - env["UV_PROJECT_ENVIRONMENT"] = str(VENV_PATH) + env["UV_PROJECT_ENVIRONMENT"] = str((app_root / ".venv").resolve()) + logger.info("No staged venv; syncing /app/.venv against overlay uv.lock.") try: - logger.info("Syncing overlay dependencies with uv (uv.lock changed)...") subprocess.run( [uv, "sync", "--frozen", "--no-dev"], - cwd=str(backend_dir), + cwd=str(current / "backend"), env=env, check=True, ) - logger.info("Dependency sync complete.") - except (subprocess.CalledProcessError, OSError) as exc: - logger.error("uv sync failed: %s (continuing with existing venv)", exc) + marker.write_text(desired, encoding="utf-8") + except Exception as exc: + logger.error("venv reconcile failed; continuing with existing venv: %s", exc) def _clear_overlay(updates_root: Path) -> None: @@ -128,7 +154,7 @@ def apply_overlay( app_root: Path = APP_ROOT, updates_root: Path = UPDATES_ROOT, image_version_path: Path = IMAGE_VERSION_PATH, - baseline_lock: Path = BASELINE_LOCK, + baseline_lock: Path | None = None, ) -> bool: """应用覆盖层。返回是否实际把覆盖层落地（供测试断言）。""" applied = _read_applied(updates_root) @@ -161,7 +187,7 @@ def apply_overlay( current = updates_root / "current" src_module = current / "backend" / "src" / "module" src_dist = current / "webui-dist" - overlay_lock = current / "backend" / "uv.lock" + src_venv = current / ".venv" if not src_module.exists(): logger.warning( @@ -184,14 +210,21 @@ def apply_overlay( except Exception as exc: logger.error("Failed to overlay webui dist: %s", exc) - # 依赖同步：覆盖层 uv.lock 相对镜像基线 lock 变化时才 uv sync。 - if overlay_lock.exists(): - overlay_lock_sha = _sha256(overlay_lock) - baseline_lock_sha = _sha256(baseline_lock) if baseline_lock.exists() else None - if overlay_lock_sha and overlay_lock_sha != baseline_lock_sha: - _uv_sync(overlay_lock.parent) - else: - logger.info("Dependencies unchanged from image baseline; skip uv sync.") + if src_venv.exists(): + try: + _replace_tree(src_venv, app_root / ".venv") + lock_sha = _sha256_file(current / "backend" / "uv.lock") + if lock_sha: + (app_root / ".venv" / _VENV_LOCK_MARKER).write_text( + lock_sha, encoding="utf-8" + ) + except Exception as exc: + logger.error("Failed to overlay staged venv: %s", exc) + else: + try: + _sync_venv_to_lock(current, app_root, baseline_lock) + except Exception as exc: + logger.error("Failed to reconcile venv with overlay lock: %s", exc) logger.info("Overlay applied successfully.") return True diff --git a/backend/src/module/api/setup.py b/backend/src/module/api/setup.py index f0cc8c7b..7049298d 100644 --- a/backend/src/module/api/setup.py +++ b/backend/src/module/api/setup.py @@ -387,6 +387,7 @@ async def complete_setup( # 4. Create sentinel file SENTINEL_PATH.parent.mkdir(parents=True, exist_ok=True) SENTINEL_PATH.touch() + await ctx.start_tasks() return ResponseModel( status=True, diff --git a/backend/src/module/api/update.py b/backend/src/module/api/update.py index b0a94ec2..dd7884b2 100644 --- a/backend/src/module/api/update.py +++ b/backend/src/module/api/update.py @@ -1,8 +1,8 @@ """在线自动更新 API：检查 / 应用 / 回滚。 三个端点均由 ``get_current_user`` 鉴权保护。``apply`` / ``rollback`` 成功后会 -安排一次延迟进程退出——Docker 的 ``restart: unless-stopped`` 会重跑 entrypoint， -由 entrypoint 的覆盖层逻辑把更新落到应用目录（并在依赖变化时 uv sync）。 +安排一次延迟进程退出——退出前写入 ``config/updates/.restart``，entrypoint +内部循环会重跑覆盖层逻辑并再次启动应用。 进程退出被抽成可打桩的 ``schedule_restart``，测试据此断言“已安排重启”而不会真的 退出进程。 @@ -12,6 +12,7 @@ import asyncio import logging import os import signal +from pathlib import Path from fastapi import APIRouter, Depends from fastapi.responses import JSONResponse @@ -26,11 +27,15 @@ router = APIRouter(prefix="/update", tags=["update"]) # 退出前的延迟：先让 HTTP 响应与最后一帧 SSE 进度刷出，再退出进程。 _RESTART_DELAY_SECONDS = 1.5 +RESTART_SENTINEL = Path("config") / "updates" / ".restart" +_RESTART_TASKS: set[asyncio.Task] = set() def _request_restart() -> None: - """退出进程，交给容器 restart 策略重启（重跑 entrypoint 应用覆盖层）。""" + """请求 entrypoint 内部循环重启进程并重跑覆盖层。""" logger.info("[Update] restarting process to apply update") + RESTART_SENTINEL.parent.mkdir(parents=True, exist_ok=True) + RESTART_SENTINEL.touch() os.kill(os.getpid(), signal.SIGINT) @@ -41,7 +46,9 @@ async def _delayed_restart() -> None: def schedule_restart() -> None: """安排一次延迟重启（抽成独立函数以便测试打桩，避免真的退出进程）。""" - asyncio.create_task(_delayed_restart()) + task = asyncio.create_task(_delayed_restart()) + _RESTART_TASKS.add(task) + task.add_done_callback(_RESTART_TASKS.discard) @router.get("/check", dependencies=[Depends(get_current_user)]) diff --git a/backend/src/module/core/context.py b/backend/src/module/core/context.py index 1aa17dc9..772b0594 100644 --- a/backend/src/module/core/context.py +++ b/backend/src/module/core/context.py @@ -77,8 +77,8 @@ class AppContext: # Downloader-status TTL cache (was ProgramStatus.check_downloader_status). self._downloader_status = False self._downloader_last_check: float = 0.0 - self._tasks_started = False self._startup_done = False + self._lifecycle_lock = asyncio.Lock() # The background task spawned by start_tasks() (downloader-wait + # scheduler start). Tracked so stop() can cancel it if shutdown # happens while it is still waiting on the downloader. @@ -134,7 +134,10 @@ class AppContext: @property def is_running(self) -> bool: - return self._tasks_started and not Checker.check_first_run() + if Checker.check_first_run(): + return False + start_pending = self._start_task is not None and not self._start_task.done() + return self.scheduler.running or start_pending @property def first_run(self) -> bool: @@ -231,8 +234,9 @@ class AppContext: async def _run_start_tasks(self) -> None: """Body of the background start task: wait for the downloader, then start loops.""" await self._wait_for_downloader() - self.scheduler.start_all() - self._tasks_started = True + async with self._lifecycle_lock: + if not Checker.check_first_run() and not self.scheduler.running: + self.scheduler.start_all() logger.info("Program running.") @staticmethod @@ -256,8 +260,14 @@ class AppContext: point) rather than calling ``settings.load()`` directly; the scheduler is not yet running at this point so its restart branch is a no-op. """ - await self.reload_settings() - if self._start_task is None or self._start_task.done(): + async with self._lifecycle_lock: + return await self._start_tasks_unlocked() + + async def _start_tasks_unlocked(self) -> ResponseModel: + await self._reload_settings_unlocked() + if not self.scheduler.running and ( + self._start_task is None or self._start_task.done() + ): self._start_task = asyncio.create_task(self._run_start_tasks()) self._start_task.add_done_callback(self._supervise_start_task) return ResponseModel( @@ -269,6 +279,11 @@ class AppContext: async def stop(self) -> ResponseModel: """Cancel any pending start task, stop background loops, and close the cached downloader session.""" + async with self._lifecycle_lock: + return await self._stop_unlocked() + + async def _stop_unlocked(self) -> ResponseModel: + was_running = self.is_running if self._start_task is not None: self._start_task.cancel() try: @@ -280,9 +295,9 @@ class AppContext: f"[Core] Background start task failed during shutdown: {e}" ) self._start_task = None - if self.is_running: + if self.scheduler.running: await self.scheduler.stop_all() - self._tasks_started = False + if was_running: resp = ResponseModel( status=True, status_code=200, @@ -302,15 +317,19 @@ class AppContext: return resp async def restart(self) -> ResponseModel: + async with self._lifecycle_lock: + return await self._restart_unlocked() + + async def _restart_unlocked(self) -> ResponseModel: stop_ok = True try: - await self.stop() + await self._stop_unlocked() except Exception as e: logger.warning(f"[Core] Error during stop in restart: {e}") stop_ok = False start_ok = True try: - await self.start_tasks() + await self._start_tasks_unlocked() except Exception as e: logger.error(f"[Core] Error during start in restart: {e}") start_ok = False @@ -346,6 +365,10 @@ class AppContext: client, notifier, endpoint-keyed caches, and scheduler all stay in sync with the new config. """ + async with self._lifecycle_lock: + await self._reload_settings_unlocked() + + async def _reload_settings_unlocked(self) -> None: # settings.load() does synchronous file I/O; keep it off the event loop. await asyncio.to_thread(settings.load) await reset_shared_client() diff --git a/backend/src/module/database/aria2.py b/backend/src/module/database/aria2.py index b2659a00..fc4f54d5 100644 --- a/backend/src/module/database/aria2.py +++ b/backend/src/module/database/aria2.py @@ -4,6 +4,7 @@ 持久化 gid -> bangumi_id / category / dedup_key 的映射。 """ +import json import logging from sqlalchemy.ext.asyncio import AsyncSession @@ -24,6 +25,7 @@ class Aria2GidDatabase: bangumi_id: int | None = None, category: str | None = None, dedup_key: str | None = None, + renamed_paths: str | None = None, ) -> None: """新增一条 gid 记录，或者用非空字段覆盖已有记录。""" existing = await self.session.get(Aria2Gid, gid) @@ -34,6 +36,7 @@ class Aria2GidDatabase: bangumi_id=bangumi_id, category=category, dedup_key=dedup_key, + renamed_paths=renamed_paths, ) ) else: @@ -43,6 +46,8 @@ class Aria2GidDatabase: existing.category = category if dedup_key is not None: existing.dedup_key = dedup_key + if renamed_paths is not None: + existing.renamed_paths = renamed_paths self.session.add(existing) await self.session.commit() @@ -67,8 +72,78 @@ class Aria2GidDatabase: async def set_category(self, gid: str, category: str) -> None: await self.upsert(gid, category=category) + async def replace_gid(self, old_gid: str, new_gid: str) -> None: + """Move local metadata from an aria2 metadata gid to its followedBy gid.""" + if old_gid == new_gid: + return + old = await self.session.get(Aria2Gid, old_gid) + if old is None: + return + existing = await self.session.get(Aria2Gid, new_gid) + if existing is None: + self.session.add( + Aria2Gid( + gid=new_gid, + bangumi_id=old.bangumi_id, + category=old.category, + dedup_key=old.dedup_key, + renamed_paths=old.renamed_paths, + created_at=old.created_at, + ) + ) + else: + if existing.bangumi_id is None: + existing.bangumi_id = old.bangumi_id + if existing.category is None: + existing.category = old.category + if existing.dedup_key is None: + existing.dedup_key = old.dedup_key + existing.renamed_paths = _merge_renamed_paths( + existing.renamed_paths, old.renamed_paths + ) + self.session.add(existing) + await self.session.delete(old) + await self.session.commit() + + async def get_renamed_paths(self, gid: str) -> dict[str, str]: + record = await self.session.get(Aria2Gid, gid) + if record is None or not record.renamed_paths: + return {} + try: + data = json.loads(record.renamed_paths) + except json.JSONDecodeError: + logger.warning("[Aria2] Ignoring invalid renamed_paths for gid %s", gid) + return {} + if not isinstance(data, dict): + return {} + return {str(k): str(v) for k, v in data.items()} + + async def set_renamed_path(self, gid: str, old_path: str, new_path: str) -> None: + mapping = await self.get_renamed_paths(gid) + for original_path, renamed_path in list(mapping.items()): + if renamed_path == old_path: + mapping[original_path] = new_path + mapping[old_path] = new_path + await self.upsert(gid, renamed_paths=json.dumps(mapping, ensure_ascii=False)) + async def delete(self, gid: str) -> None: existing = await self.session.get(Aria2Gid, gid) if existing is not None: await self.session.delete(existing) await self.session.commit() + + +def _merge_renamed_paths(left: str | None, right: str | None) -> str | None: + if not left: + return right + if not right: + return left + try: + merged = json.loads(left) + incoming = json.loads(right) + except json.JSONDecodeError: + return left + if not isinstance(merged, dict) or not isinstance(incoming, dict): + return left + merged.update(incoming) + return json.dumps(merged, ensure_ascii=False) diff --git a/backend/src/module/database/migrations.py b/backend/src/module/database/migrations.py index 80700913..1b84739a 100644 --- a/backend/src/module/database/migrations.py +++ b/backend/src/module/database/migrations.py @@ -29,6 +29,7 @@ TABLE_MODELS: list[type[SQLModel]] = [Bangumi, RSSItem, Torrent, User, Passkey] # already_applied 守卫：接收 inspector，返回该迁移是否已生效 AppliedCheck = Callable[[Any], bool] +GuardedStatement = tuple[str, AppliedCheck] def column_exists(table: str, column: str) -> AppliedCheck: @@ -56,12 +57,31 @@ def index_exists(table: str, index: str) -> AppliedCheck: return check +def all_checks(*checks: AppliedCheck) -> AppliedCheck: + def check(inspector) -> bool: + return all(item(inspector) for item in checks) + + return check + + @dataclass(frozen=True) class Migration: version: int description: str statements: tuple[str, ...] already_applied: AppliedCheck + guarded_statements: tuple[GuardedStatement, ...] = () + + def pending_statements(self, inspector) -> tuple[str, ...]: + if self.guarded_statements: + return tuple( + statement + for statement, statement_applied in self.guarded_statements + if not statement_applied(inspector) + ) + if self.already_applied(inspector): + return () + return self.statements # 迁移按版本顺序执行；版本号与 3.2.x 的历史保持一致，旧数据库照常升级。 @@ -80,7 +100,25 @@ MIGRATIONS: tuple[Migration, ...] = ( "ALTER TABLE rssitem ADD COLUMN last_checked_at TEXT", "ALTER TABLE rssitem ADD COLUMN last_error TEXT", ), - column_exists("rssitem", "connection_status"), + all_checks( + column_exists("rssitem", "connection_status"), + column_exists("rssitem", "last_checked_at"), + column_exists("rssitem", "last_error"), + ), + ( + ( + "ALTER TABLE rssitem ADD COLUMN connection_status TEXT", + column_exists("rssitem", "connection_status"), + ), + ( + "ALTER TABLE rssitem ADD COLUMN last_checked_at TEXT", + column_exists("rssitem", "last_checked_at"), + ), + ( + "ALTER TABLE rssitem ADD COLUMN last_error TEXT", + column_exists("rssitem", "last_error"), + ), + ), ), Migration( 3, @@ -121,7 +159,30 @@ MIGRATIONS: tuple[Migration, ...] = ( "ALTER TABLE bangumi ADD COLUMN needs_review INTEGER DEFAULT 0", "ALTER TABLE bangumi ADD COLUMN needs_review_reason TEXT DEFAULT NULL", ), - column_exists("bangumi", "episode_offset"), + all_checks( + column_exists("bangumi", "episode_offset"), + column_exists("bangumi", "season_offset"), + column_exists("bangumi", "needs_review"), + column_exists("bangumi", "needs_review_reason"), + ), + ( + ( + "ALTER TABLE bangumi RENAME COLUMN offset TO episode_offset", + column_exists("bangumi", "episode_offset"), + ), + ( + "ALTER TABLE bangumi ADD COLUMN season_offset INTEGER DEFAULT 0", + column_exists("bangumi", "season_offset"), + ), + ( + "ALTER TABLE bangumi ADD COLUMN needs_review INTEGER DEFAULT 0", + column_exists("bangumi", "needs_review"), + ), + ( + "ALTER TABLE bangumi ADD COLUMN needs_review_reason TEXT DEFAULT NULL", + column_exists("bangumi", "needs_review_reason"), + ), + ), ), Migration( 6, @@ -130,7 +191,20 @@ MIGRATIONS: tuple[Migration, ...] = ( "ALTER TABLE torrent ADD COLUMN qb_hash TEXT", "CREATE INDEX IF NOT EXISTS ix_torrent_qb_hash ON torrent(qb_hash)", ), - column_exists("torrent", "qb_hash"), + all_checks( + column_exists("torrent", "qb_hash"), + index_exists("torrent", "ix_torrent_qb_hash"), + ), + ( + ( + "ALTER TABLE torrent ADD COLUMN qb_hash TEXT", + column_exists("torrent", "qb_hash"), + ), + ( + "CREATE INDEX IF NOT EXISTS ix_torrent_qb_hash ON torrent(qb_hash)", + index_exists("torrent", "ix_torrent_qb_hash"), + ), + ), ), Migration( 7, @@ -141,7 +215,22 @@ MIGRATIONS: tuple[Migration, ...] = ( "ALTER TABLE bangumi ADD COLUMN suggested_episode_offset " "INTEGER DEFAULT NULL", ), - column_exists("bangumi", "suggested_season_offset"), + all_checks( + column_exists("bangumi", "suggested_season_offset"), + column_exists("bangumi", "suggested_episode_offset"), + ), + ( + ( + "ALTER TABLE bangumi ADD COLUMN suggested_season_offset " + "INTEGER DEFAULT NULL", + column_exists("bangumi", "suggested_season_offset"), + ), + ( + "ALTER TABLE bangumi ADD COLUMN suggested_episode_offset " + "INTEGER DEFAULT NULL", + column_exists("bangumi", "suggested_episode_offset"), + ), + ), ), Migration( 8, @@ -162,7 +251,20 @@ MIGRATIONS: tuple[Migration, ...] = ( "ALTER TABLE bangumi ADD COLUMN preferred_group TEXT DEFAULT NULL", "ALTER TABLE bangumi ADD COLUMN preferred_resolution TEXT DEFAULT NULL", ), - column_exists("bangumi", "preferred_group"), + all_checks( + column_exists("bangumi", "preferred_group"), + column_exists("bangumi", "preferred_resolution"), + ), + ( + ( + "ALTER TABLE bangumi ADD COLUMN preferred_group TEXT DEFAULT NULL", + column_exists("bangumi", "preferred_group"), + ), + ( + "ALTER TABLE bangumi ADD COLUMN preferred_resolution TEXT DEFAULT NULL", + column_exists("bangumi", "preferred_resolution"), + ), + ), ), Migration( 11, @@ -222,6 +324,12 @@ MIGRATIONS: tuple[Migration, ...] = ( ("CREATE INDEX IF NOT EXISTS ix_torrent_bangumi_id ON torrent(bangumi_id)",), index_exists("torrent", "ix_torrent_bangumi_id"), ), + Migration( + 15, + "add renamed_paths column to aria2_gid for persisted file renames", + ("ALTER TABLE aria2_gid ADD COLUMN renamed_paths TEXT DEFAULT NULL",), + column_exists("aria2_gid", "renamed_paths"), + ), ) # 由迁移列表派生，新增迁移时无需手动同步 @@ -272,7 +380,8 @@ def run_migrations_conn(conn: Connection) -> None: if migration.version <= current: continue # 每轮重新 inspect：前一个迁移的 DDL 需要对守卫可见 - if migration.already_applied(inspect(conn)): + pending = migration.pending_statements(inspect(conn)) + if not pending: logger.debug( f"[Database] Migration v{migration.version} skipped " f"(already applied): {migration.description}" @@ -280,7 +389,7 @@ def run_migrations_conn(conn: Connection) -> None: else: savepoint = conn.begin_nested() try: - for stmt in migration.statements: + for stmt in pending: conn.execute(text(stmt)) savepoint.commit() logger.info( diff --git a/backend/src/module/downloader/__init__.py b/backend/src/module/downloader/__init__.py index 6891579c..7d190d7c 100644 --- a/backend/src/module/downloader/__init__.py +++ b/backend/src/module/downloader/__init__.py @@ -1,9 +1,10 @@ from .base import ( + AddResult, CoreDownloaderClient, DownloaderCapabilities, DownloaderClient, ) -from .download_client import AddResult, DownloadClient, shutdown +from .download_client import DownloadClient, shutdown __all__ = [ "AddResult", diff --git a/backend/src/module/downloader/base.py b/backend/src/module/downloader/base.py index d24d2b8e..2ff76194 100644 --- a/backend/src/module/downloader/base.py +++ b/backend/src/module/downloader/base.py @@ -8,9 +8,18 @@ unsupported operations instead of blowing up on a missing method. """ from dataclasses import dataclass +from enum import Enum from typing import Any, ClassVar, Protocol, runtime_checkable +class AddResult(Enum): + """Torrent-add result shared by concrete clients and the facade.""" + + ADDED = "added" + DUPLICATE = "duplicate" + FAILED = "failed" + + @dataclass(frozen=True) class DownloaderCapabilities: """What a concrete download client can do. @@ -39,7 +48,7 @@ class CoreDownloaderClient(Protocol): async def add_torrents( self, torrent_urls, torrent_files, save_path, category, tags=None - ) -> bool: ... + ) -> AddResult: ... @runtime_checkable @@ -70,7 +79,7 @@ class DownloaderClient(Protocol): # Torrent lifecycle async def add_torrents( self, torrent_urls, torrent_files, save_path, category, tags=None - ) -> bool: ... + ) -> AddResult: ... async def torrents_info(self, status_filter, category, tag=None) -> list[dict]: ... diff --git a/backend/src/module/downloader/client/aria2_downloader.py b/backend/src/module/downloader/client/aria2_downloader.py index 8035ffb3..c071dd96 100644 --- a/backend/src/module/downloader/client/aria2_downloader.py +++ b/backend/src/module/downloader/client/aria2_downloader.py @@ -8,7 +8,7 @@ from typing import ClassVar import httpx from module.database import Database -from module.downloader.base import DownloaderCapabilities +from module.downloader.base import AddResult, DownloaderCapabilities logger = logging.getLogger(__name__) @@ -167,19 +167,42 @@ class Aria2Downloader: return os.path.basename(files[0]["path"]) return download.get("gid", "") + @staticmethod + def _extract_followed_by_gid(status: dict | None) -> str | None: + if not isinstance(status, dict): + return None + followed_by = (status or {}).get("followedBy") or [] + if isinstance(followed_by, list) and followed_by: + return str(followed_by[0]) + return None + + @staticmethod + def _translate_renamed_path(path: str, renamed_paths: dict[str, str]) -> str: + seen = set() + while path in renamed_paths and path not in seen: + seen.add(path) + path = renamed_paths[path] + return path + @staticmethod def _move_file(old_abs: str, new_abs: str) -> None: """移动单个文件；目标已存在且不是同一个文件时拒绝覆盖。 - os.replace 会静默覆盖已有目标——多文件种子若映射到同一目标名， - 直接覆盖会毁掉已重命名好的文件，所以这里显式拦下来。 + 多文件种子若映射到同一目标名，直接覆盖会毁掉已重命名好的文件， + 所以这里显式拦下来。 """ + if ( + os.path.exists(old_abs) + and os.path.exists(new_abs) + and os.path.samefile(old_abs, new_abs) + ): + return if os.path.exists(new_abs) and not ( os.path.exists(old_abs) and os.path.samefile(old_abs, new_abs) ): raise FileExistsError(f"destination already exists: {new_abs}") os.makedirs(os.path.dirname(new_abs), exist_ok=True) - os.replace(old_abs, new_abs) + shutil.move(old_abs, new_abs) @staticmethod def _move_files(files: list[dict], old_dir: str, new_dir: str) -> None: @@ -217,7 +240,7 @@ class Aria2Downloader: # Adding # ------------------------------------------------------------------ - async def _dedup_record_is_stale(self, db: Database, gid: str) -> bool: + async def _dedup_record_is_stale(self, db: Database, gid: str) -> bool | None: """dedup 命中后校验 gid 是否仍存在于 aria2；陈旧则删记录并返回 True。 本地 aria2_gid 表可能和 aria2 真实状态脱节（用户在 aria2 UI 里删了 @@ -225,8 +248,8 @@ class Aria2Downloader: 会让同一个种子永远被当作"已添加"而无法重新下载。 - RPC 报 not found、或 status 为 "removed" → 记录陈旧，删掉，返回 True； - - aria2 不可达（连接错误）→ 无法断言，保守保留记录，返回 False； - - 其余 RPC 错误 → 同样保守保留。 + - aria2 不可达（连接错误）→ 无法断言，保守保留记录，返回 None； + - 其余 RPC 错误 → 同样保守保留，返回 None。 """ try: status = await self._call("tellStatus", [gid, ["status"]]) @@ -240,7 +263,7 @@ class Aria2Downloader: await db.aria2.delete(gid) return True logger.debug("[Aria2] Cannot verify gid %s, keeping record: %s", gid, e) - return False + return None except Aria2ConnectionError as e: # aria2 不可达 != 记录陈旧，绝不能因此删本地记录。 logger.debug( @@ -248,7 +271,7 @@ class Aria2Downloader: gid, e, ) - return False + return None if (status or {}).get("status") == "removed": logger.info( "[Aria2] Stale dedup record: gid %s was removed in aria2, " @@ -259,39 +282,51 @@ class Aria2Downloader: return True return False - async def _add_uri(self, url: str, options: dict) -> str | None: + async def _add_uri(self, url: str, options: dict) -> tuple[AddResult, str | None]: try: - return await self._call("addUri", [[url], options]) + return AddResult.ADDED, await self._call("addUri", [[url], options]) except Aria2RpcError as e: if self._is_duplicate_error(e): logger.debug("[Aria2] addUri reports duplicate: %s", e.message) + return AddResult.DUPLICATE, None else: logger.error("[Aria2] addUri failed for %s: %s", url, e) - return None + return AddResult.FAILED, None except Aria2ConnectionError as e: logger.error("[Aria2] addUri connection error for %s: %s", url, e) - return None + return AddResult.FAILED, None - async def _add_torrent_file(self, data: bytes, options: dict) -> str | None: + async def _add_torrent_file( + self, data: bytes, options: dict + ) -> tuple[AddResult, str | None]: import base64 b64 = base64.b64encode(data).decode() try: - return await self._call("addTorrent", [b64, [], options]) + return AddResult.ADDED, await self._call("addTorrent", [b64, [], options]) except Aria2RpcError as e: if self._is_duplicate_error(e): logger.debug("[Aria2] addTorrent reports duplicate: %s", e.message) + return AddResult.DUPLICATE, None else: logger.error("[Aria2] addTorrent failed: %s", e) - return None + return AddResult.FAILED, None except Aria2ConnectionError as e: logger.error("[Aria2] addTorrent connection error: %s", e) - return None + return AddResult.FAILED, None + + async def _resolve_followed_by_gid(self, gid: str) -> str: + try: + status = await self._call("tellStatus", [gid, ["followedBy"]]) + except (Aria2RpcError, Aria2ConnectionError) as e: + logger.debug("[Aria2] Could not resolve followedBy for %s: %s", gid, e) + return gid + return self._extract_followed_by_gid(status) or gid async def add_torrents( self, torrent_urls, torrent_files, save_path, category, tags=None - ) -> bool: - """添加下载任务，返回 True 当且仅当真的新增了至少一个任务。 + ) -> AddResult: + """添加下载任务，返回新增/重复/失败的三态结果。 去重优先靠本地的 dedup_key（url 或种子内容 hash）判断"已经添加过"， aria2 RPC 报错里带 already/duplicate 字样时也当作重复处理（见 @@ -301,6 +336,8 @@ class Aria2Downloader: bangumi_id = self._parse_bangumi_id_from_tag(tags) options = {"dir": save_path} added_any = False + duplicate_any = False + failed_any = False async with Database() as db: if torrent_urls: urls = ( @@ -309,14 +346,23 @@ class Aria2Downloader: for url in urls: dedup_key = f"url:{url}" existing_gid = await db.aria2.find_by_dedup_key(dedup_key) - if existing_gid and not await self._dedup_record_is_stale( - db, existing_gid - ): - logger.debug("[Aria2] Skip already-added url: %s", url) + if existing_gid: + stale = await self._dedup_record_is_stale(db, existing_gid) + if stale is None: + failed_any = True + continue + if not stale: + logger.debug("[Aria2] Skip already-added url: %s", url) + duplicate_any = True + continue + result, gid = await self._add_uri(url, options) + if result is AddResult.DUPLICATE: + duplicate_any = True continue - gid = await self._add_uri(url, options) - if gid is None: + if result is AddResult.FAILED or gid is None: + failed_any = True continue + gid = await self._resolve_followed_by_gid(gid) await db.aria2.upsert(gid, bangumi_id, category, dedup_key) added_any = True if torrent_files: @@ -328,17 +374,34 @@ class Aria2Downloader: for f in files: dedup_key = f"file:{hashlib.sha1(f).hexdigest()}" existing_gid = await db.aria2.find_by_dedup_key(dedup_key) - if existing_gid and not await self._dedup_record_is_stale( - db, existing_gid - ): - logger.debug("[Aria2] Skip already-added torrent file") + if existing_gid: + stale = await self._dedup_record_is_stale(db, existing_gid) + if stale is None: + failed_any = True + continue + if not stale: + logger.debug("[Aria2] Skip already-added torrent file") + duplicate_any = True + continue + result, gid = await self._add_torrent_file(f, options) + if result is AddResult.DUPLICATE: + duplicate_any = True continue - gid = await self._add_torrent_file(f, options) - if gid is None: + if result is AddResult.FAILED or gid is None: + failed_any = True continue + gid = await self._resolve_followed_by_gid(gid) await db.aria2.upsert(gid, bangumi_id, category, dedup_key) added_any = True - return added_any + # 部分成功按 ADDED 报告（与 qB 的批量语义一致）：已经真正开始下载 + # 的任务必须让上层入库，否则会与 aria2 状态脱节；全部失败才算 FAILED。 + if added_any: + return AddResult.ADDED + if failed_any: + return AddResult.FAILED + if duplicate_any: + return AddResult.DUPLICATE + return AddResult.FAILED # ------------------------------------------------------------------ # Querying @@ -353,8 +416,16 @@ class Aria2Downloader: logger.error("[Aria2] Failed to query downloads: %s", e) return [] raw = [*active, *waiting, *stopped] - gids = [d["gid"] for d in raw if d.get("gid")] + followed_by: dict[str, str] = {} + for d in raw: + gid = d.get("gid") + follower = self._extract_followed_by_gid(d) + if gid and follower: + followed_by[gid] = follower + gids = [followed_by.get(d["gid"], d["gid"]) for d in raw if d.get("gid")] async with Database() as db: + for old_gid, new_gid in followed_by.items(): + await db.aria2.replace_gid(old_gid, new_gid) meta = await db.aria2.get_many(gids) allowed_statuses = ( @@ -364,9 +435,14 @@ class Aria2Downloader: for d in raw: gid = d.get("gid", "") aria2_status = d.get("status", "") + if gid in followed_by: + # 磁力元数据 stub：真实下载由 followedBy gid 表示。stub 的 + # status 是 "complete"，若不跳过会以完成态混进结果，让 + # 重命名循环拿元数据文件当正片处理，且 UI 出现重复条目。 + continue if allowed_statuses is not None and aria2_status not in allowed_statuses: continue - info = meta.get(gid) + info = meta.get(followed_by.get(gid, gid)) torrent_category = info.category if info else None if category and torrent_category != category: continue @@ -405,10 +481,13 @@ class Aria2Downloader: save_dir = (status or {}).get("dir", "") except (Aria2RpcError, Aria2ConnectionError) as e: logger.debug("[Aria2] Could not resolve dir for %s: %s", torrent_hash, e) + async with Database() as db: + renamed_paths = await db.aria2.get_renamed_paths(torrent_hash) result = [] for f in files or []: path = f.get("path", "") rel = os.path.relpath(path, save_dir) if save_dir and path else path + rel = self._translate_renamed_path(rel, renamed_paths) result.append({"name": rel, "size": int(f.get("length", 0) or 0)}) return result @@ -454,6 +533,8 @@ class Aria2Downloader: "[Aria2] Rename reported success but %s is missing", new_abs ) return False + async with Database() as db: + await db.aria2.set_renamed_path(torrent_hash, old_path, new_path) return True # ------------------------------------------------------------------ diff --git a/backend/src/module/downloader/client/mock_downloader.py b/backend/src/module/downloader/client/mock_downloader.py index 391315fd..dbed008b 100644 --- a/backend/src/module/downloader/client/mock_downloader.py +++ b/backend/src/module/downloader/client/mock_downloader.py @@ -8,7 +8,7 @@ qBittorrent instance. All operations return success and log their actions. import logging from typing import Any, ClassVar -from module.downloader.base import DownloaderCapabilities +from module.downloader.base import AddResult, DownloaderCapabilities logger = logging.getLogger(__name__) @@ -105,8 +105,8 @@ class MockDownloader: save_path: str, category: str, tags: str | None = None, - ) -> bool: - """Add a torrent. Returns True for success.""" + ) -> AddResult: + """Add a torrent. Returns ADDED for success.""" import hashlib import time @@ -127,7 +127,7 @@ class MockDownloader: logger.info( f"[MockDownloader] add_torrents -> hash={mock_hash[:16]}... save_path={save_path}" ) - return True + return AddResult.ADDED @staticmethod def _normalize_hashes(hashes: str | list | tuple) -> list[str]: diff --git a/backend/src/module/downloader/client/qb_downloader.py b/backend/src/module/downloader/client/qb_downloader.py index 34ca43a0..64a9625d 100644 --- a/backend/src/module/downloader/client/qb_downloader.py +++ b/backend/src/module/downloader/client/qb_downloader.py @@ -1,15 +1,18 @@ import asyncio import json import logging +import re from typing import ClassVar import httpx from module.ab_decorator import qb_connect_failed_wait -from module.downloader.base import DownloaderCapabilities +from module.downloader.base import AddResult, DownloaderCapabilities logger = logging.getLogger(__name__) +_MAGNET_BTIH_RE = re.compile(r"xt=urn:btih:([0-9A-Fa-f]{40}|[A-Za-z2-7]{32})") + class QbDownloader: capabilities: ClassVar[DownloaderCapabilities] = DownloaderCapabilities( @@ -206,6 +209,40 @@ class QbDownloader: resp = await self._get("torrents/files", params={"hash": torrent_hash}) return resp.json() + async def _urls_already_added(self, torrent_urls) -> bool: + """尽力确认"Fails."是否因为任务已存在：从磁力链提取 btih 并查询 qB。 + + 只有全部 URL 都是可提取 hash 的磁力链、且每个 hash 都已在 qB 中时才 + 返回 True；.torrent URL 无法事先得知 hash，返回 False（按失败处理）。 + """ + urls = ( + torrent_urls + if isinstance(torrent_urls, list) + else [torrent_urls] if torrent_urls else [] + ) + if not urls: + return False + hashes: list[str] = [] + for url in urls: + m = _MAGNET_BTIH_RE.search(url or "") + if not m: + return False + btih = m.group(1) + if len(btih) == 32: # base32 -> hex + import base64 + + try: + btih = base64.b32decode(btih.upper()).hex() + except ValueError: + return False + hashes.append(btih.lower()) + try: + resp = await self._get("torrents/info", params={"hashes": "|".join(hashes)}) + found = {t.get("hash", "").lower() for t in resp.json()} + return set(hashes) <= found + except (httpx.RequestError, ValueError): + return False + async def add_torrents( self, torrent_urls, torrent_files, save_path, category, tags=None ): @@ -247,7 +284,23 @@ class QbDownloader: data=data, files=files if files else None, ) - return resp.text == "Ok." + if resp.status_code == 200 and resp.text == "Ok.": + return AddResult.ADDED + if resp.status_code == 200 and resp.text == "Fails.": + # "Fails." 覆盖所有被拒绝的 add（重复、种子损坏、磁力链 + # 无法解析……），不能一律当重复——否则损坏种子会被记成 + # 已下载、永远不重试。只有能通过 hash 确认任务已存在时 + # 才归类为重复，其余按失败抛出让上层重试。 + if await self._urls_already_added(torrent_urls): + return AddResult.DUPLICATE + raise ConnectionError( + "[Downloader] qBittorrent rejected torrent add " + "(200 'Fails.') and no matching torrent found" + ) + raise ConnectionError( + "[Downloader] qBittorrent rejected torrent add: " + f"HTTP {resp.status_code} {resp.text!r}" + ) except (httpx.ReadError, httpx.ConnectError, httpx.RequestError) as e: if attempt < max_retries - 1: logger.warning( diff --git a/backend/src/module/downloader/download_client.py b/backend/src/module/downloader/download_client.py index 9fad357b..0c885c93 100644 --- a/backend/src/module/downloader/download_client.py +++ b/backend/src/module/downloader/download_client.py @@ -1,31 +1,16 @@ import asyncio import logging -from enum import Enum from module.conf import settings from module.models import Bangumi, Torrent from module.network import RequestContent -from .base import DownloaderClient +from .base import AddResult, DownloaderClient from .path import gen_save_path logger = logging.getLogger(__name__) -class AddResult(Enum): - """``add_torrent`` 的三态结果。 - - - ``ADDED``: 真正新增了下载任务； - - ``DUPLICATE``: 下载器里已有同一种子（qB 重复 / aria2 去重命中）， - 对调用方来说等同于成功，不应触发失败通知； - - ``FAILED``: 投递失败（抓取种子文件失败或客户端异常），可以重试。 - """ - - ADDED = "added" - DUPLICATE = "duplicate" - FAILED = "failed" - - # --------------------------------------------------------------------------- # Module-level concrete-client cache (session reuse, #1039 / #900) # @@ -336,20 +321,22 @@ class DownloadClient: # Create tag with bangumi_id for offset lookup during rename tags = f"ab:{bangumi.id}" if bangumi.id else None try: - if await self.client.add_torrents( + result = await self.client.add_torrents( torrent_urls=torrent_url, torrent_files=torrent_file, save_path=bangumi.save_path, category="Bangumi", tags=tags, - ): + ) + if result is AddResult.ADDED: logger.debug("[Downloader] Add torrent: %s", bangumi.official_title) return AddResult.ADDED - else: + if result is AddResult.DUPLICATE: logger.debug( "[Downloader] Torrent added before: %s", bangumi.official_title ) return AddResult.DUPLICATE + return AddResult.FAILED except Exception as e: logger.error( f"[Downloader] Failed to add torrent for {bangumi.official_title}: {e}" diff --git a/backend/src/module/models/aria2.py b/backend/src/module/models/aria2.py index b64ee05b..59a7dfe7 100644 --- a/backend/src/module/models/aria2.py +++ b/backend/src/module/models/aria2.py @@ -20,4 +20,5 @@ class Aria2Gid(SQLModel, table=True): bangumi_id: Optional[int] = Field(default=None, foreign_key="bangumi.id") category: Optional[str] = None dedup_key: Optional[str] = Field(default=None, index=True) + renamed_paths: Optional[str] = None created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) diff --git a/backend/src/module/models/config.py b/backend/src/module/models/config.py index d85b1790..8cd7c99b 100644 --- a/backend/src/module/models/config.py +++ b/backend/src/module/models/config.py @@ -259,6 +259,27 @@ class LLM(BaseModel): "primary: LLM first, regex as safety net" ), ) + timeout: float = Field(default=20.0, ge=1.0, description="LLM request timeout") + cache_ttl: int = Field( + default=900, + ge=0, + description="Seconds to cache LLM parse successes and failures; 0 disables", + ) + max_concurrency: int = Field( + default=2, + ge=1, + description="Maximum concurrent LLM parse requests", + ) + failure_threshold: int = Field( + default=3, + ge=1, + description="Consecutive LLM failures before temporarily skipping calls", + ) + failure_backoff: int = Field( + default=300, + ge=0, + description="Seconds to skip LLM calls after failure_threshold is reached", + ) # [Deprecated] 旧版 OpenAI 解析配置，仅保留用于读取旧配置文件（向后兼容）。 diff --git a/backend/src/module/parser/analyser/llm.py b/backend/src/module/parser/analyser/llm.py index 417aa402..e19eacef 100644 --- a/backend/src/module/parser/analyser/llm.py +++ b/backend/src/module/parser/analyser/llm.py @@ -15,13 +15,17 @@ import logging from typing import Any, Optional import anthropic +import httpx import openai as openai_sdk from google import genai from google.genai import errors as genai_errors from google.genai import types as genai_types +from httpx_socks import AsyncProxyTransport from openai import AsyncOpenAI from pydantic import BaseModel +from module.conf import settings + logger = logging.getLogger(__name__) @@ -99,6 +103,7 @@ class LLMParser: provider: str = "openai", model: str = "gpt-4o-mini", base_url: str = "", + timeout: float = 20.0, ) -> None: """初始化解析器并构建对应提供商的异步客户端。 @@ -107,6 +112,7 @@ class LLMParser: provider: "openai" | "anthropic" | "gemini"。 model: 模型名。 base_url: 自定义端点，仅 openai 提供商使用；空串表示官方 API。 + timeout: 单次 LLM 请求超时秒数。 Raises: ValueError: api_key 为空或 provider 不受支持。 @@ -115,17 +121,31 @@ class LLMParser: raise ValueError("API key is required.") self.provider = provider self.model = model + self._http_client: httpx.AsyncClient | None = None if provider == "openai": + self._http_client = _build_http_client(timeout) self._openai_client = AsyncOpenAI( - api_key=api_key, base_url=base_url or None + api_key=api_key, + base_url=base_url or None, + timeout=timeout, + http_client=self._http_client, ) elif provider == "anthropic": - self._anthropic_client = anthropic.AsyncAnthropic(api_key=api_key) + self._http_client = _build_http_client(timeout) + self._anthropic_client = anthropic.AsyncAnthropic( + api_key=api_key, + timeout=timeout, + http_client=self._http_client, + ) elif provider == "gemini": self._gemini_client = genai.Client(api_key=api_key) else: raise ValueError(f"Unsupported LLM provider: {provider}") + async def aclose(self) -> None: + if self._http_client is not None: + await self._http_client.aclose() + async def parse(self, raw: str, asdict: bool = True) -> dict | None: """解析种子标题，返回 Episode 形状的 dict；失败返回 None。 @@ -212,3 +232,36 @@ class LLMParser: except json.JSONDecodeError as e: logger.warning("Cannot decode Gemini output for '%s': %s", raw, e) return None + + +def _build_http_client(timeout: float) -> httpx.AsyncClient: + http_timeout = httpx.Timeout( + connect=min(timeout, 10.0), + read=timeout, + write=min(timeout, 10.0), + pool=min(timeout, 10.0), + ) + kwargs: dict[str, Any] = {"timeout": http_timeout, "follow_redirects": True} + if not settings.proxy.enable: + return httpx.AsyncClient(**kwargs) + if "http" in settings.proxy.type: + if settings.proxy.username: + proxy_url = ( + f"http://{settings.proxy.username}:{settings.proxy.password}" + f"@{settings.proxy.host}:{settings.proxy.port}" + ) + else: + proxy_url = f"http://{settings.proxy.host}:{settings.proxy.port}" + return httpx.AsyncClient(proxy=proxy_url, **kwargs) + if settings.proxy.type == "socks5": + if settings.proxy.username: + proxy_url = ( + f"socks5://{settings.proxy.username}:{settings.proxy.password}" + f"@{settings.proxy.host}:{settings.proxy.port}" + ) + else: + proxy_url = f"socks5://{settings.proxy.host}:{settings.proxy.port}" + return httpx.AsyncClient( + transport=AsyncProxyTransport.from_url(proxy_url, rdns=True), **kwargs + ) + return httpx.AsyncClient(**kwargs) diff --git a/backend/src/module/parser/title_parser.py b/backend/src/module/parser/title_parser.py index f9884761..02d409b9 100644 --- a/backend/src/module/parser/title_parser.py +++ b/backend/src/module/parser/title_parser.py @@ -1,4 +1,6 @@ +import asyncio import logging +import time from module.conf import settings from module.models import Bangumi @@ -11,6 +13,7 @@ from module.parser.analyser import ( tmdb_parser, torrent_parser, ) +from module.parser.analyser.raw_parser import _detect_non_episodic_type logger = logging.getLogger(__name__) @@ -19,6 +22,11 @@ logger = logging.getLogger(__name__) # mode 不影响客户端构建（调用时读取），不参与缓存键。 _llm_parser: LLMParser | None = None _llm_parser_kwargs: dict | None = None +_llm_cache: dict[tuple, tuple[float, Episode | None]] = {} +_llm_failure_count = 0 +_llm_breaker_until = 0.0 +_llm_semaphore: asyncio.Semaphore | None = None +_llm_semaphore_limit: int | None = None def _get_llm_parser(kwargs: dict) -> LLMParser: @@ -33,8 +41,32 @@ def reset_cache() -> None: """清空 LLM 解析器单例。配置重载后必须调用，否则会继续使用旧配置 （如 provider/base_url/api_key）构建的客户端。""" global _llm_parser, _llm_parser_kwargs + global _llm_cache, _llm_failure_count, _llm_breaker_until + global _llm_semaphore, _llm_semaphore_limit + parser = _llm_parser + old_kwargs = _llm_parser_kwargs _llm_parser = None _llm_parser_kwargs = None + _llm_cache = {} + _llm_failure_count = 0 + _llm_breaker_until = 0.0 + _llm_semaphore = None + _llm_semaphore_limit = None + if parser is not None and hasattr(parser, "aclose"): + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return + # 不能立刻关：进行中的 _llm_parse 还握着这个共享 client，立即 + # aclose 会让它们全部报错、把 None 写进缓存并误触熔断。in-flight + # 调用被 wait_for(timeout) 兜底，等一个超时周期后再关必然安全。 + grace = float((old_kwargs or {}).get("timeout") or 30) + 5.0 + + async def _close_after_grace() -> None: + await asyncio.sleep(grace) + await parser.aclose() + + loop.create_task(_close_after_grace()) def _llm_config() -> LLM: @@ -58,25 +90,73 @@ def _llm_config() -> LLM: async def _llm_parse(raw: str) -> Episode | None: """用 LLM 解析标题，返回 Episode；任何失败（API 错误、拒答、 输出不可用）都返回 None，由调用方决定是否回退。""" + global _llm_failure_count, _llm_breaker_until conf = _llm_config() + now = time.monotonic() + if _llm_breaker_until > now: + logger.warning("LLM parser breaker is open; skipping '%s'", raw) + return None + parser_kwargs = { + "provider": conf.provider, + "api_key": conf.api_key, + "model": conf.model, + "base_url": conf.base_url, + "timeout": conf.timeout, + } + cache_key = (tuple(sorted(parser_kwargs.items())), raw) + if conf.cache_ttl > 0: + cached = _llm_cache.get(cache_key) + if cached is not None: + expires_at, cached_episode = cached + if expires_at > now: + return cached_episode + _llm_cache.pop(cache_key, None) try: - llm = _get_llm_parser( - { - "provider": conf.provider, - "api_key": conf.api_key, - "model": conf.model, - "base_url": conf.base_url, - } - ) - episode_dict = await llm.parse(raw, asdict=True) + llm = _get_llm_parser(parser_kwargs) + semaphore = _get_llm_semaphore(conf.max_concurrency) + async with semaphore: + episode_dict = await asyncio.wait_for( + llm.parse(raw, asdict=True), timeout=conf.timeout + ) if not isinstance(episode_dict, dict): + _record_llm_failure(conf) + _cache_llm_result(cache_key, None, conf.cache_ttl) return None - return Episode(**episode_dict) + episode = Episode(**episode_dict) + _llm_failure_count = 0 + _llm_breaker_until = 0.0 + _cache_llm_result(cache_key, episode, conf.cache_ttl) + return episode except Exception as e: logger.warning(f"LLM cannot parse '{raw}': {type(e).__name__}: {e}") + _record_llm_failure(conf) + _cache_llm_result(cache_key, None, conf.cache_ttl) return None +def _get_llm_semaphore(limit: int) -> asyncio.Semaphore: + global _llm_semaphore, _llm_semaphore_limit + if _llm_semaphore is None or _llm_semaphore_limit != limit: + _llm_semaphore = asyncio.Semaphore(limit) + _llm_semaphore_limit = limit + return _llm_semaphore + + +def _cache_llm_result( + cache_key: tuple, episode: Episode | None, cache_ttl: int +) -> None: + if cache_ttl <= 0: + return + _llm_cache[cache_key] = (time.monotonic() + cache_ttl, episode) + + +def _record_llm_failure(conf: LLM) -> None: + global _llm_failure_count, _llm_breaker_until + _llm_failure_count += 1 + if _llm_failure_count >= conf.failure_threshold: + _llm_breaker_until = time.monotonic() + conf.failure_backoff + + class TitleParser: def __init__(self): pass @@ -130,16 +210,26 @@ class TitleParser: try: llm_conf = _llm_config() episode = None + from_llm = False # primary 模式：LLM 优先解析每个标题；失败时用正则兜底，保证不丢标题 if llm_conf.enable and llm_conf.mode == "primary": episode = await _llm_parse(raw) + from_llm = episode is not None if episode is None: episode = raw_parser(raw) # fallback 模式（默认）：正则优先，仅当正则失败时才请求 LLM 兜底 if episode is None and llm_conf.enable and llm_conf.mode == "fallback": episode = await _llm_parse(raw) + from_llm = episode is not None if episode is None: return None + # 正则路径已在剥离字幕组名后的标题上判定过 episode_type，这里只 + # 给缺少该能力的 LLM 结果补分类；对完整 raw 判定会把组名/标签里 + # 撞词（如 "[Movie-Fan]"）的普通周更集误判成剧场版/特别篇。 + if from_llm: + detected_episode_type = _detect_non_episodic_type(raw) + if detected_episode_type is not None: + episode.episode_type = detected_episode_type titles = { "zh": episode.title_zh, diff --git a/backend/src/module/rss/engine.py b/backend/src/module/rss/engine.py index 91e088b7..c41bc1c3 100644 --- a/backend/src/module/rss/engine.py +++ b/backend/src/module/rss/engine.py @@ -386,7 +386,18 @@ class RSSEngine: ) if torrents: async with DownloadClient() as client: - await client.add_torrent(torrents, bangumi) + result = await client.add_torrent(torrents, bangumi) + if result is AddResult.FAILED: + return ResponseModel( + status=False, + status_code=502, + msg_en=( + f"[Engine] Download {bangumi.official_title} failed." + ), + msg_zh=f"[Engine] 下载 {bangumi.official_title} 失败。", + ) + for torrent in torrents: + torrent.downloaded = True await self.db.torrent.add_all(torrents) return ResponseModel( status=True, diff --git a/backend/src/module/update/updater.py b/backend/src/module/update/updater.py index f8e84c91..063fb034 100644 --- a/backend/src/module/update/updater.py +++ b/backend/src/module/update/updater.py @@ -21,17 +21,20 @@ import json import logging import os import shutil +import sqlite3 +import subprocess import time import zipfile from dataclasses import asdict, dataclass from pathlib import Path -from typing import Optional +from typing import Callable, Optional import httpx import semver from pydantic import BaseModel from module.conf import IMAGE_VERSION, VERSION +from module.database.migrations import CURRENT_SCHEMA_VERSION from module.network.request_url import get_shared_client logger = logging.getLogger(__name__) @@ -166,11 +169,17 @@ class Updater: root: Optional[Path] = None, current_version: Optional[str] = None, image_version: Optional[str] = None, + data_db: Optional[Path] = None, + app_root: Optional[Path] = None, + dependency_syncer: Optional[Callable[[Path], None]] = None, client: Optional[httpx.AsyncClient] = None, owner: str = GITHUB_OWNER, repo: str = GITHUB_REPO, ) -> None: self.root = root if root is not None else DEFAULT_UPDATES_ROOT + self.data_db = data_db if data_db is not None else Path("data") / "data.db" + self.app_root = app_root if app_root is not None else Path(".") + self._dependency_syncer = dependency_syncer self.current_version = ( current_version if current_version is not None else VERSION ) @@ -343,7 +352,13 @@ class Updater: return ApplyResult(success=False, message=message) def _promote( - self, unpacked: Path, version: str, sha256: str, manifest: dict + self, + unpacked: Path, + version: str, + sha256: str, + manifest: dict, + db_snapshot: Optional[dict], + schema_version_before: Optional[int], ) -> None: """把解包好的 staging 提升为 current，旧 current 移到 backup。""" current = self.root / "current" @@ -356,16 +371,104 @@ class Updater: lockfile = current / "backend" / "uv.lock" lock_sha = _sha256_file(lockfile) if lockfile.exists() else None + # schema_version_after 不能在这里记录：新版本的迁移要到下次启动才会 + # 执行，此刻读库得到的仍是旧值。回滚时直接读活库比较。 applied = { "version": version, "applied_at": int(time.time()), "sha256": sha256, "lockfile_sha256": manifest.get("lockfile_sha256") or lock_sha, + "schema_version_before": schema_version_before, + "db_backup": db_snapshot, } (self.root / "applied.json").write_text( json.dumps(applied, indent=2), encoding="utf-8" ) + def _snapshot_database(self) -> tuple[Optional[dict], Optional[int]]: + """Back up the runtime SQLite DB before applying an overlay update.""" + schema_version = _read_db_schema_version(self.data_db) + if not self.data_db.exists(): + return None, schema_version + + backup_dir = self.root / "db-backup" + if backup_dir.exists(): + shutil.rmtree(backup_dir) + backup_dir.mkdir(parents=True, exist_ok=True) + dest = backup_dir / self.data_db.name + + # sqlite3 backup API 透过 WAL 读出一致的独立快照，不需要（也不能） + # 搭配活库的 -wal/-shm 边车文件——那属于另一个数据库实例，配对恢复 + # 会导致 WAL 帧错乱。 + with sqlite3.connect(self.data_db) as src, sqlite3.connect(dest) as dst: + src.backup(dst) + + return ( + { + "path": str(dest.relative_to(self.root)), + "created_at": int(time.time()), + }, + schema_version, + ) + + def _prepare_dependencies(self, unpacked: Path) -> None: + """Resolve overlay dependencies before promotion so boot stays simple.""" + if self._dependency_syncer is not None: + self._dependency_syncer(unpacked) + return + + backend_dir = unpacked / "backend" + overlay_lock = backend_dir / "uv.lock" + if not overlay_lock.exists(): + return + baseline_lock = self.app_root / "uv.lock" + overlay_sha = _sha256_file(overlay_lock) + baseline_sha = _sha256_file(baseline_lock) if baseline_lock.exists() else None + if baseline_sha == overlay_sha: + logger.info( + "[Update] Dependencies unchanged from image baseline; skip sync." + ) + return + + uv = shutil.which("uv") + if not uv: + raise RuntimeError("uv not found on PATH; cannot sync update dependencies") + + env = dict(os.environ) + env["UV_PROJECT_ENVIRONMENT"] = str((unpacked / ".venv").resolve()) + logger.info("[Update] Syncing overlay dependencies into staged venv.") + subprocess.run( + [uv, "sync", "--frozen", "--no-dev"], + cwd=str(backend_dir), + env=env, + check=True, + ) + + def _restore_database_snapshot(self, applied: dict) -> Optional[int]: + # 只认 applied.json 里显式记录的快照；不回退到磁盘上残留的 + # db-backup/ 目录，否则第二次回滚会误还原上一轮更新的陈旧快照。 + snapshot = applied.get("db_backup") or {} + rel_path = snapshot.get("path") if isinstance(snapshot, dict) else None + if not rel_path: + return None + backup_db = self.root / rel_path + if not backup_db.exists(): + return None + + self.data_db.parent.mkdir(parents=True, exist_ok=True) + tmp = self.data_db.with_name(self.data_db.name + ".rollback_tmp") + shutil.copy2(backup_db, tmp) + os.replace(tmp, self.data_db) + # 快照是 backup API 生成的独立文件；活库残留的 -wal/-shm 必须清掉。 + for suffix in ("-wal", "-shm"): + sidecar = self.data_db.with_name(self.data_db.name + suffix) + try: + sidecar.unlink() + except FileNotFoundError: + pass + shutil.rmtree(backup_db.parent, ignore_errors=True) + return _read_db_schema_version(self.data_db) + async def apply_update(self, channel: str = "stable") -> ApplyResult: if self._lock.locked(): return ApplyResult( @@ -422,9 +525,19 @@ class Updater: f"update (requires >= {min_iv}); please pull a newer image" ) + await asyncio.to_thread(self._prepare_dependencies, unpacked) + db_snapshot, schema_version_before = await asyncio.to_thread( + self._snapshot_database + ) self._set_progress(phase="promoting", message="promoting update") await asyncio.to_thread( - self._promote, unpacked, version, actual_hash, manifest + self._promote, + unpacked, + version, + actual_hash, + manifest, + db_snapshot, + schema_version_before, ) self._cache = None # 版本已变，作废检查缓存 @@ -454,6 +567,19 @@ class Updater: current = self.root / "current" backup = self.root / "backup" applied = self.root / "applied.json" + applied_data: dict = {} + try: + applied_data = json.loads(applied.read_text(encoding="utf-8")) + except (OSError, ValueError): + pass + + live_schema = _read_db_schema_version(self.data_db) + if live_schema is not None and live_schema > CURRENT_SCHEMA_VERSION: + return self._fail( + "database schema is newer than this rollback code " + f"understands ({live_schema} > {CURRENT_SCHEMA_VERSION}); " + "refusing rollback" + ) if backup.exists(): # current <-> backup 互换，使回滚本身可再次撤销。 @@ -470,12 +596,37 @@ class Updater: version = _read_manifest(current).get("version") except (OSError, ValueError): version = None + # 只有当 schema 自 apply 后确实前进过（旧代码读不了新库） + # 才还原快照；schema 没变时保留活库，避免丢掉更新之后 + # 写入的全部数据。 + snapshot_schema = applied_data.get("schema_version_before") + restored_schema = None + if ( + live_schema is not None + and snapshot_schema is not None + and live_schema != snapshot_schema + ): + logger.warning( + "[Update] Schema advanced since update (%s -> %s); " + "restoring pre-update database snapshot. Data written " + "since the update will be lost.", + snapshot_schema, + live_schema, + ) + restored_schema = self._restore_database_snapshot(applied_data) + else: + logger.info( + "[Update] Schema unchanged since update; " + "keeping live database." + ) applied.write_text( json.dumps( { "version": version, "applied_at": int(time.time()), "rolled_back": True, + "schema_version_before": live_schema, + "schema_version_after": restored_schema, }, indent=2, ), @@ -518,3 +669,23 @@ updater = Updater() def get_update_progress() -> dict: """供 SSE 端点读取当前更新进度。""" return updater.progress_dict() + + +def _read_db_schema_version(db_path: Path) -> Optional[int]: + if not db_path.exists(): + return None + try: + with sqlite3.connect(db_path) as conn: + row = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' " + "AND name='schema_version'" + ).fetchone() + if row is None: + return 0 + version = conn.execute( + "SELECT version FROM schema_version WHERE id = 1" + ).fetchone() + return int(version[0]) if version else 0 + except sqlite3.Error as exc: + logger.warning("[Update] failed to read DB schema version: %s", exc) + return None diff --git a/backend/src/test/conftest.py b/backend/src/test/conftest.py index 2b06b4b7..d958cdbb 100644 --- a/backend/src/test/conftest.py +++ b/backend/src/test/conftest.py @@ -93,9 +93,7 @@ async def db_engine(): @pytest_asyncio.fixture async def db_session(db_engine): """Provide a fresh async database session per test.""" - factory = async_sessionmaker( - db_engine, class_=AsyncSession, expire_on_commit=False - ) + factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) async with factory() as session: yield session @@ -127,6 +125,8 @@ def mock_settings(test_settings): @pytest.fixture def mock_qb_client(): """Mock QbDownloader that simulates qBittorrent API responses.""" + from module.downloader import AddResult + client = AsyncMock() client.auth.return_value = True client.logout.return_value = None @@ -134,7 +134,7 @@ def mock_qb_client(): client.torrents_info.return_value = [] client.torrents_files.return_value = [] client.torrents_rename_file.return_value = True - client.add_torrents.return_value = True + client.add_torrents.return_value = AddResult.ADDED client.torrents_delete.return_value = None client.torrents_pause.return_value = None client.torrents_resume.return_value = None diff --git a/backend/src/test/test_aria2_downloader.py b/backend/src/test/test_aria2_downloader.py index 73bab1e9..389a0a0e 100644 --- a/backend/src/test/test_aria2_downloader.py +++ b/backend/src/test/test_aria2_downloader.py @@ -15,6 +15,7 @@ import httpx import pytest from module.database import Database +from module.downloader import AddResult from module.downloader.client.aria2_downloader import ( Aria2ConnectionError, Aria2Downloader, @@ -202,13 +203,40 @@ class TestAddTorrents: category="Bangumi", tags="ab:42", ) - assert result is True + assert result is AddResult.ADDED async with Database() as db: record = await db.aria2.get("gid001") assert record is not None assert record.bangumi_id == 42 assert record.category == "Bangumi" + async def test_magnet_followed_by_gid_is_persisted(self): + aria2 = _aria2() + + async def fake_call(method, params=None, timeout=10.0): + if method == "addUri": + return "metadata-gid" + if method == "tellStatus": + return {"followedBy": ["payload-gid"]} + return None + + with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)): + result = await aria2.add_torrents( + torrent_urls="magnet:?xt=urn:btih:abc", + torrent_files=None, + save_path="/downloads", + category="Bangumi", + tags="ab:42", + ) + + assert result is AddResult.ADDED + async with Database() as db: + assert await db.aria2.get("metadata-gid") is None + record = await db.aria2.get("payload-gid") + assert record is not None + assert record.bangumi_id == 42 + assert record.dedup_key == "url:magnet:?xt=urn:btih:abc" + async def test_duplicate_url_second_call_returns_false_and_skips_rpc(self): aria2 = _aria2() add_uri_calls = [] @@ -237,7 +265,7 @@ class TestAddTorrents: category="Bangumi", tags="ab:1", ) - assert result is False + assert result is AddResult.DUPLICATE # addUri only called once -- the second, duplicate add never hit aria2. assert len(add_uri_calls) == 1 @@ -268,7 +296,7 @@ class TestAddTorrents: category="Bangumi", tags="ab:1", ) - assert result is True + assert result is AddResult.ADDED assert len(add_uri_calls) == 2 async def test_all_urls_already_added_returns_false(self): @@ -296,7 +324,7 @@ class TestAddTorrents: category="Bangumi", tags="ab:1", ) - assert result is False + assert result is AddResult.DUPLICATE async def test_rpc_duplicate_error_treated_as_already_added(self): aria2 = _aria2() @@ -311,7 +339,7 @@ class TestAddTorrents: save_path="/downloads", category="Bangumi", ) - assert result is False + assert result is AddResult.DUPLICATE async with Database() as db: # Nothing to persist -- addUri never returned a gid. assert ( @@ -329,7 +357,7 @@ class TestAddTorrents: save_path="/downloads", category="Bangumi", ) - assert result is False + assert result is AddResult.FAILED async def test_torrent_file_bytes_dedup_by_content_hash(self): aria2 = _aria2() @@ -359,8 +387,8 @@ class TestAddTorrents: category="Bangumi", tags="ab:7", ) - assert first is True - assert second is False + assert first is AddResult.ADDED + assert second is AddResult.DUPLICATE assert len(add_torrent_calls) == 1 async def test_add_torrents_stale_dedup_gid_not_found_readds(self): @@ -388,7 +416,7 @@ class TestAddTorrents: tags="ab:1", ) - assert result is True + assert result is AddResult.ADDED async with Database() as db: assert await db.aria2.get("gid-stale") is None assert await db.aria2.get("gid-fresh") is not None @@ -415,7 +443,7 @@ class TestAddTorrents: category="Bangumi", ) - assert result is True + assert result is AddResult.ADDED async with Database() as db: assert await db.aria2.get("gid-removed") is None assert await db.aria2.get("gid-fresh") is not None @@ -442,7 +470,7 @@ class TestAddTorrents: category="Bangumi", ) - assert result is False + assert result is AddResult.DUPLICATE async with Database() as db: assert await db.aria2.get("gid-alive") is not None @@ -468,7 +496,7 @@ class TestAddTorrents: category="Bangumi", ) - assert result is False + assert result is AddResult.FAILED async with Database() as db: assert await db.aria2.get("gid-unknown") is not None @@ -497,7 +525,7 @@ class TestAddTorrents: category="Bangumi", ) - assert result is True + assert result is AddResult.ADDED async with Database() as db: assert await db.aria2.get("gid-stale-file") is None assert await db.aria2.get("gid-fresh-file") is not None @@ -676,6 +704,34 @@ class TestTorrentsInfo: assert info["dlspeed"] == 12345 assert info["upspeed"] == 678 + async def test_torrents_info_migrates_followed_by_gid_metadata(self): + aria2 = _aria2() + stub = _download("metadata-gid", status="complete") + stub["followedBy"] = ["payload-gid"] + payload = _download("payload-gid", status="active") + + async def fake_call(method, params=None, timeout=10.0): + if method == "tellActive": + return [payload] + if method == "tellStopped": + return [stub] + return [] + + async with Database() as db: + await db.aria2.upsert("metadata-gid", bangumi_id=5, category="Bangumi") + + with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)): + result = await aria2.torrents_info(status_filter=None, category="Bangumi") + + # 元数据 stub 必须被跳过，只报告真实下载（follower gid），否则 stub + # 会以完成态混进重命名循环、UI 里出现重复条目 + assert [t["hash"] for t in result] == ["payload-gid"] + assert result[0]["tags"] == "ab:5" + assert result[0]["category"] == "Bangumi" + async with Database() as db: + assert await db.aria2.get("metadata-gid") is None + assert await db.aria2.get("payload-gid") is not None + # --------------------------------------------------------------------------- # torrents_files: relative-path mapping @@ -704,6 +760,24 @@ class TestTorrentsFiles: {"name": "ep01.ass", "size": 1}, ] + async def test_returns_persisted_renamed_paths(self): + aria2 = _aria2() + + async def fake_call(method, params=None, timeout=10.0): + if method == "getFiles": + return [{"path": "/downloads/Show/Season 1/ep01.mkv", "length": "100"}] + if method == "tellStatus": + return {"dir": "/downloads/Show/Season 1"} + return None + + async with Database() as db: + await db.aria2.set_renamed_path("gidA", "ep01.mkv", "Show - 01.mkv") + + with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)): + files = await aria2.torrents_files("gidA") + + assert files == [{"name": "Show - 01.mkv", "size": 100}] + async def test_query_failure_returns_empty_list(self): aria2 = _aria2() with patch.object( @@ -733,6 +807,39 @@ class TestTorrentsRenameFile: assert not (tmp_path / "old.mkv").exists() assert (tmp_path / "new.mkv").read_bytes() == b"data" + async def test_rename_updates_torrents_files_path(self, tmp_path): + aria2 = _aria2() + old_file = tmp_path / "old.mkv" + old_file.write_bytes(b"data") + + async def fake_call(method, params=None, timeout=10.0): + if method == "getFiles": + return [{"path": str(old_file), "length": "4"}] + return {"dir": str(tmp_path)} + + with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)): + assert await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv") + files = await aria2.torrents_files("gidA") + + assert files == [{"name": "new.mkv", "size": 4}] + + async def test_chained_rename_keeps_original_aria2_path_in_sync(self, tmp_path): + aria2 = _aria2() + old_file = tmp_path / "old.mkv" + old_file.write_bytes(b"data") + + async def fake_call(method, params=None, timeout=10.0): + if method == "getFiles": + return [{"path": str(old_file), "length": "4"}] + return {"dir": str(tmp_path)} + + with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)): + assert await aria2.torrents_rename_file("gidA", "old.mkv", "mid.mkv") + assert await aria2.torrents_rename_file("gidA", "mid.mkv", "new.mkv") + files = await aria2.torrents_files("gidA") + + assert files == [{"name": "new.mkv", "size": 4}] + async def test_creates_parent_dirs_for_nested_new_path(self, tmp_path): aria2 = _aria2() (tmp_path / "old.mkv").write_bytes(b"data") diff --git a/backend/src/test/test_context.py b/backend/src/test/test_context.py index d7ebd4bf..3881a544 100644 --- a/backend/src/test/test_context.py +++ b/backend/src/test/test_context.py @@ -1,5 +1,6 @@ """Tests for module.core.context.AppContext.""" +import asyncio from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -21,6 +22,15 @@ def ctx(): # explicitly since a bare MagicMock attribute is truthy by default and # would make reload_settings() think a restart is needed. ctx.scheduler.running = False + + def _mark_running(): + ctx.scheduler.running = True + + async def _mark_stopped(): + ctx.scheduler.running = False + + ctx.scheduler.start_all.side_effect = _mark_running + ctx.scheduler.stop_all.side_effect = _mark_stopped return ctx @@ -60,7 +70,7 @@ class TestStartup: mock_first_run.assert_called_once() assert ctx.first_run_boot is True assert ctx._startup_done is True - assert ctx._tasks_started is False + assert ctx.scheduler.running is False async def test_existing_db_runs_pending_migrations(self, ctx): """Existing DB, same version => run_migrations() is invoked.""" @@ -118,6 +128,7 @@ class TestStartTasks: with ( patch("module.core.context.settings"), patch("module.core.context.reset_shared_client", new=AsyncMock()), + patch("module.core.context.Checker.check_first_run", return_value=False), patch( "module.core.context.Checker.check_downloader", new=AsyncMock(return_value=True), @@ -128,24 +139,49 @@ class TestStartTasks: # as a supervised background task and returns immediately; await it # here so the assertions below observe its effects. await ctx._start_task + assert ctx.is_running is True ctx.scheduler.start_all.assert_called_once() - assert ctx._tasks_started is True assert isinstance(resp, ResponseModel) assert resp.status is True - async def test_start_tasks_routes_through_reload_settings(self, ctx): - """start_tasks() must go through reload_settings() (the single reload - entry point) instead of calling settings.load() directly.""" - ctx.reload_settings = AsyncMock() - with patch( - "module.core.context.Checker.check_downloader", - new=AsyncMock(return_value=True), + async def test_start_tasks_runs_reload_choreography(self, ctx): + """start_tasks() must use the same reload choreography as config updates.""" + with ( + patch("module.core.context.settings") as mock_settings, + patch("module.core.context.reset_shared_client", new=AsyncMock()), + patch("module.core.context.Checker.check_first_run", return_value=False), + patch( + "module.core.context.Checker.check_downloader", + new=AsyncMock(return_value=True), + ), ): await ctx.start_tasks() await ctx._start_task - ctx.reload_settings.assert_awaited_once() + mock_settings.load.assert_called_once() + + async def test_concurrent_start_tasks_create_one_background_start(self, ctx): + start_gate = asyncio.Event() + + async def check_downloader(): + await start_gate.wait() + return True + + with ( + patch("module.core.context.settings"), + patch("module.core.context.reset_shared_client", new=AsyncMock()), + patch("module.core.context.Checker.check_first_run", return_value=False), + patch( + "module.core.context.Checker.check_downloader", + new=AsyncMock(side_effect=check_downloader), + ), + ): + await asyncio.gather(ctx.start_tasks(), ctx.start_tasks()) + start_gate.set() + await ctx._start_task + + ctx.scheduler.start_all.assert_called_once() # --------------------------------------------------------------------------- @@ -155,7 +191,7 @@ class TestStartTasks: class TestStop: async def test_stop_running_stops_scheduler_and_downloader(self, ctx): - ctx._tasks_started = True + ctx.scheduler.running = True with ( patch("module.core.context.Checker.check_first_run", return_value=False), patch( @@ -166,11 +202,11 @@ class TestStop: ctx.scheduler.stop_all.assert_awaited_once() mock_shutdown.assert_awaited_once() - assert ctx._tasks_started is False + assert ctx.scheduler.running is False assert resp.status_code == 200 async def test_stop_not_running_still_closes_downloader(self, ctx): - ctx._tasks_started = False + ctx.scheduler.running = False with ( patch("module.core.context.Checker.check_first_run", return_value=False), patch( diff --git a/backend/src/test/test_download_client.py b/backend/src/test/test_download_client.py index 2e2d6484..719e27f2 100644 --- a/backend/src/test/test_download_client.py +++ b/backend/src/test/test_download_client.py @@ -142,7 +142,7 @@ class TestAddTorrent: ): """When client.add_torrents returns False (already added), returns DUPLICATE, not a failure.""" - mock_qb_client.add_torrents.return_value = False + mock_qb_client.add_torrents.return_value = AddResult.DUPLICATE torrent = make_torrent(url="magnet:?xt=urn:btih:abc") bangumi = make_bangumi() @@ -155,6 +155,23 @@ class TestAddTorrent: assert result is AddResult.DUPLICATE + async def test_add_torrent_client_reports_failed_returns_failed( + self, download_client, mock_qb_client + ): + """When client.add_torrents returns FAILED, facade returns FAILED.""" + mock_qb_client.add_torrents.return_value = AddResult.FAILED + torrent = make_torrent(url="magnet:?xt=urn:btih:abc") + bangumi = make_bangumi() + + with patch("module.downloader.download_client.RequestContent") as MockReq: + mock_req = AsyncMock() + MockReq.return_value.__aenter__ = AsyncMock(return_value=mock_req) + MockReq.return_value.__aexit__ = AsyncMock(return_value=False) + + result = await download_client.add_torrent(torrent, bangumi) + + assert result is AddResult.FAILED + async def test_add_torrent_fetch_failure_returns_failed( self, download_client, mock_qb_client ): diff --git a/backend/src/test/test_llm_parser.py b/backend/src/test/test_llm_parser.py index cf6ba43f..648af707 100644 --- a/backend/src/test/test_llm_parser.py +++ b/backend/src/test/test_llm_parser.py @@ -3,7 +3,7 @@ import json from types import SimpleNamespace from typing import Any -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, patch import anthropic import httpx @@ -11,6 +11,8 @@ import openai as openai_sdk import pytest from google.genai import errors as genai_errors +from module.models.config import Proxy +from module.parser.analyser import llm as llm_module from module.parser.analyser.llm import ( DEFAULT_PROMPT, EPISODE_JSON_SCHEMA, @@ -45,6 +47,38 @@ class TestLLMParserInit: with pytest.raises(ValueError): LLMParser(api_key="key", provider="deepmind") + def test_http_proxy_is_passed_to_injected_http_client(self): + proxy = Proxy( + enable=True, + type="http", + host="proxy.local", + port=7890, + username="user", + password="pass", + ) + with patch.object(llm_module.settings, "proxy", proxy): + with patch("module.parser.analyser.llm.httpx.AsyncClient") as mock_client: + llm_module._build_http_client(5.0) + + assert mock_client.call_args.kwargs["proxy"] == ( + "http://user:pass@proxy.local:7890" + ) + + def test_socks_proxy_uses_proxy_transport(self): + proxy = Proxy(enable=True, type="socks5", host="proxy.local", port=7891) + with patch.object(llm_module.settings, "proxy", proxy): + with ( + patch("module.parser.analyser.llm.httpx.AsyncClient") as mock_client, + patch( + "module.parser.analyser.llm.AsyncProxyTransport.from_url", + return_value="transport", + ) as mock_transport, + ): + llm_module._build_http_client(5.0) + + mock_transport.assert_called_once_with("socks5://proxy.local:7891", rdns=True) + assert mock_client.call_args.kwargs["transport"] == "transport" + class TestOpenAIProvider: def _make_parser(self, **kwargs) -> LLMParser: diff --git a/backend/src/test/test_migrations_module.py b/backend/src/test/test_migrations_module.py index eb73c83f..8bc97f80 100644 --- a/backend/src/test/test_migrations_module.py +++ b/backend/src/test/test_migrations_module.py @@ -8,8 +8,10 @@ from sqlmodel import create_engine from module.database.migrations import ( CURRENT_SCHEMA_VERSION, MIGRATIONS, + ensure_schema_version_table, get_schema_version, run_migrations, + set_schema_version, ) @@ -68,6 +70,71 @@ def _make_v0_engine(): return engine +def _make_versioned_engine( + version: int, + *, + bangumi_extra: str, + torrent_extra: str = "", +) -> object: + engine = create_engine("sqlite://") + bangumi_extra_sql = f", {bangumi_extra}" if bangumi_extra else "" + torrent_extra_sql = f", {torrent_extra}" if torrent_extra else "" + with engine.begin() as conn: + conn.execute( + text( + "CREATE TABLE bangumi (" + " id INTEGER PRIMARY KEY," + " official_title TEXT," + " title_raw TEXT," + " deleted BOOLEAN DEFAULT 0," + " air_weekday INTEGER," + " archived BOOLEAN DEFAULT 0" + f"{bangumi_extra_sql}" + ")" + ) + ) + conn.execute( + text( + "CREATE TABLE rssitem (" + " id INTEGER PRIMARY KEY," + " name TEXT," + " url TEXT," + " connection_status TEXT," + " last_checked_at TEXT," + " last_error TEXT" + ")" + ) + ) + conn.execute( + text( + "CREATE TABLE torrent (" + " id INTEGER PRIMARY KEY," + " name TEXT," + " bangumi_id INTEGER," + " rss_id INTEGER," + " url TEXT" + f"{torrent_extra_sql}" + ")" + ) + ) + conn.execute( + text( + "CREATE TABLE user (" + " id INTEGER PRIMARY KEY," + " username TEXT," + " password TEXT" + ")" + ) + ) + ensure_schema_version_table(conn) + set_schema_version(conn, version) + return engine + + +def _columns(engine, table: str) -> set[str]: + return {c["name"] for c in inspect(engine).get_columns(table)} + + class TestMigrationTable: def test_every_migration_has_an_already_applied_guard(self): """No migration may rely on the old hardcoded if-ladder.""" @@ -130,6 +197,15 @@ class TestRunMigrations: torrent_indexes = {ix["name"] for ix in inspector.get_indexes("torrent")} assert "ix_torrent_bangumi_id" in torrent_indexes + def test_adds_aria2_renamed_paths_column(self): + """v15 stores local aria2 file-renames so getFiles can be translated.""" + engine = _make_v0_engine() + run_migrations(engine) + + inspector = inspect(engine) + aria2_cols = {c["name"] for c in inspector.get_columns("aria2_gid")} + assert "renamed_paths" in aria2_cols + def test_is_idempotent(self): engine = _make_v0_engine() run_migrations(engine) @@ -152,6 +228,89 @@ class TestRunMigrations: bangumi_cols = {c["name"] for c in inspector.get_columns("bangumi")} assert "weekday_locked" in bangumi_cols # later migrations still ran + def test_partial_v2_adds_missing_rssitem_connection_columns(self): + engine = _make_v0_engine() + with engine.begin() as conn: + conn.execute(text("ALTER TABLE rssitem ADD COLUMN connection_status TEXT")) + + run_migrations(engine) + + rss_cols = _columns(engine, "rssitem") + assert {"connection_status", "last_checked_at", "last_error"} <= rss_cols + + def test_partial_v5_adds_missing_columns_without_rerunning_rename(self): + engine = _make_versioned_engine( + 4, + bangumi_extra="episode_offset INTEGER DEFAULT 0", + ) + + run_migrations(engine) + + bangumi_cols = _columns(engine, "bangumi") + assert "episode_offset" in bangumi_cols + assert "season_offset" in bangumi_cols + assert "needs_review" in bangumi_cols + assert "needs_review_reason" in bangumi_cols + + def test_partial_v6_adds_missing_qb_hash_index(self): + engine = _make_versioned_engine( + 5, + bangumi_extra=( + "episode_offset INTEGER DEFAULT 0," + " season_offset INTEGER DEFAULT 0," + " needs_review INTEGER DEFAULT 0," + " needs_review_reason TEXT DEFAULT NULL" + ), + torrent_extra="qb_hash TEXT", + ) + + run_migrations(engine) + + torrent_indexes = {ix["name"] for ix in inspect(engine).get_indexes("torrent")} + assert "ix_torrent_qb_hash" in torrent_indexes + + def test_partial_v7_adds_missing_suggested_episode_offset(self): + engine = _make_versioned_engine( + 6, + bangumi_extra=( + "episode_offset INTEGER DEFAULT 0," + " season_offset INTEGER DEFAULT 0," + " needs_review INTEGER DEFAULT 0," + " needs_review_reason TEXT DEFAULT NULL," + " suggested_season_offset INTEGER DEFAULT NULL" + ), + torrent_extra="qb_hash TEXT", + ) + + run_migrations(engine) + + bangumi_cols = _columns(engine, "bangumi") + assert "suggested_season_offset" in bangumi_cols + assert "suggested_episode_offset" in bangumi_cols + + def test_partial_v10_adds_missing_preferred_resolution(self): + engine = _make_versioned_engine( + 9, + bangumi_extra=( + "episode_offset INTEGER DEFAULT 0," + " season_offset INTEGER DEFAULT 0," + " needs_review INTEGER DEFAULT 0," + " needs_review_reason TEXT DEFAULT NULL," + " suggested_season_offset INTEGER DEFAULT NULL," + " suggested_episode_offset INTEGER DEFAULT NULL," + " title_aliases TEXT DEFAULT NULL," + " weekday_locked BOOLEAN DEFAULT 0," + " preferred_group TEXT DEFAULT NULL" + ), + torrent_extra="qb_hash TEXT", + ) + + run_migrations(engine) + + bangumi_cols = _columns(engine, "bangumi") + assert "preferred_group" in bangumi_cols + assert "preferred_resolution" in bangumi_cols + def test_failure_stops_before_bumping_version(self): """A failing migration must not record its version as applied. diff --git a/backend/src/test/test_mock_downloader.py b/backend/src/test/test_mock_downloader.py index aa21d22d..5e129af5 100644 --- a/backend/src/test/test_mock_downloader.py +++ b/backend/src/test/test_mock_downloader.py @@ -2,6 +2,7 @@ import pytest +from module.downloader import AddResult from module.downloader.client.mock_downloader import MockDownloader @@ -112,7 +113,7 @@ class TestMockDownloaderAddTorrents: save_path="/downloads/Bangumi", category="Bangumi", ) - assert result is True + assert result is AddResult.ADDED async def test_add_torrent_stores_in_state(self, mock_dl): await mock_dl.add_torrents( @@ -143,7 +144,7 @@ class TestMockDownloaderAddTorrents: save_path="/downloads", category="Bangumi", ) - assert result is True + assert result is AddResult.ADDED async def test_two_different_torrents_stored_separately(self, mock_dl): await mock_dl.add_torrents( diff --git a/backend/src/test/test_qb_downloader.py b/backend/src/test/test_qb_downloader.py index 8f241682..2749b413 100644 --- a/backend/src/test/test_qb_downloader.py +++ b/backend/src/test/test_qb_downloader.py @@ -11,6 +11,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest +from module.downloader import AddResult from module.downloader.client.qb_downloader import QbDownloader # --------------------------------------------------------------------------- @@ -612,6 +613,77 @@ class TestAuthQb52Compat: assert result is False +# --------------------------------------------------------------------------- +# add_torrents +# --------------------------------------------------------------------------- + + +class TestAddTorrents: + async def test_add_returns_added_on_ok_body(self): + qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False) + qb._client = AsyncMock() + mock_resp = MagicMock(status_code=200, text="Ok.") + qb._client.request = AsyncMock(return_value=mock_resp) + + result = await qb.add_torrents( + torrent_urls="magnet:?xt=urn:btih:abc", + torrent_files=None, + save_path="/downloads", + category="Bangumi", + ) + + assert result is AddResult.ADDED + + async def test_add_fails_body_with_existing_magnet_is_duplicate(self): + qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False) + qb._client = AsyncMock() + btih = "a" * 40 + add_resp = MagicMock(status_code=200, text="Fails.") + info_resp = MagicMock(status_code=200) + info_resp.json.return_value = [{"hash": btih}] + qb._client.request = AsyncMock(side_effect=[add_resp, info_resp]) + + result = await qb.add_torrents( + torrent_urls=f"magnet:?xt=urn:btih:{btih}", + torrent_files=None, + save_path="/downloads", + category="Bangumi", + ) + + # 磁力 hash 已存在于 qB：确认是重复 + assert result is AddResult.DUPLICATE + + async def test_add_fails_body_without_confirmed_duplicate_raises(self): + # "Fails." 也可能是种子损坏/无法解析——无法确认重复时必须按失败抛出， + # 让上层保留重试机会，而不是记成已下载 + qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False) + qb._client = AsyncMock() + mock_resp = MagicMock(status_code=200, text="Fails.") + qb._client.request = AsyncMock(return_value=mock_resp) + + with pytest.raises(ConnectionError): + await qb.add_torrents( + torrent_urls="https://mikanani.me/Download/broken.torrent", + torrent_files=None, + save_path="/downloads", + category="Bangumi", + ) + + async def test_add_raises_on_non_ok_status(self): + qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False) + qb._client = AsyncMock() + mock_resp = MagicMock(status_code=415, text="Unsupported Media Type") + qb._client.request = AsyncMock(return_value=mock_resp) + + with pytest.raises(ConnectionError): + await qb.add_torrents( + torrent_urls="magnet:?xt=urn:btih:abc", + torrent_files=None, + save_path="/downloads", + category="Bangumi", + ) + + # --------------------------------------------------------------------------- # torrents_delete (#1046) # --------------------------------------------------------------------------- diff --git a/backend/src/test/test_rss_engine_new.py b/backend/src/test/test_rss_engine_new.py index d6b4f4a4..7210458f 100644 --- a/backend/src/test/test_rss_engine_new.py +++ b/backend/src/test/test_rss_engine_new.py @@ -531,6 +531,37 @@ class TestRefreshRssRetry: assert all_torrents[0].bangumi_id is None +class TestDownloadBangumi: + async def test_failed_add_returns_failure_and_does_not_persist(self, rss_engine): + bangumi = make_bangumi( + official_title="Mushoku Tensei", + rss_link="https://example.com/rss", + filter="", + ) + torrent = Torrent( + name="[Sub] Mushoku Tensei - 12 [1080p].mkv", + url="https://example.com/ep12.torrent", + ) + with ( + patch("module.rss.engine.RequestContent") as MockReq, + patch("module.rss.engine.DownloadClient") as MockClient, + ): + req = AsyncMock() + req.get_torrents = AsyncMock(return_value=[torrent]) + MockReq.return_value.__aenter__ = AsyncMock(return_value=req) + MockReq.return_value.__aexit__ = AsyncMock(return_value=False) + client = AsyncMock() + client.add_torrent = AsyncMock(return_value=AddResult.FAILED) + MockClient.return_value.__aenter__ = AsyncMock(return_value=client) + MockClient.return_value.__aexit__ = AsyncMock(return_value=False) + + resp = await rss_engine.download_bangumi(bangumi) + + assert resp.status is False + assert resp.status_code == 502 + assert await rss_engine.db.torrent.search_all() == [] + + # --------------------------------------------------------------------------- # add_rss # --------------------------------------------------------------------------- diff --git a/backend/src/test/test_setup.py b/backend/src/test/test_setup.py index 57f18e0b..74a9853c 100644 --- a/backend/src/test/test_setup.py +++ b/backend/src/test/test_setup.py @@ -10,9 +10,10 @@ from module.api.setup import SENTINEL_PATH, router @pytest.fixture def mock_ctx(): - """Mock AppContext whose reload_settings is an awaitable no-op.""" + """Mock AppContext whose lifecycle methods are awaitable no-ops.""" ctx = MagicMock() ctx.reload_settings = AsyncMock() + ctx.start_tasks = AsyncMock() return ctx @@ -352,6 +353,7 @@ class TestSetupComplete: data = response.json() assert data["status"] is True mock_ctx.reload_settings.assert_awaited_once() + mock_ctx.start_tasks.assert_awaited_once() class TestSentinelPath: diff --git a/backend/src/test/test_title_parser.py b/backend/src/test/test_title_parser.py index 13c5b621..f8f5b1ff 100644 --- a/backend/src/test/test_title_parser.py +++ b/backend/src/test/test_title_parser.py @@ -14,7 +14,7 @@ RAW_TITLE = ( ) -def _make_llm_episode() -> Episode: +def _make_llm_episode(episode_type: str = "episode") -> Episode: return Episode( title_en="LLM Title", title_zh=None, @@ -26,6 +26,7 @@ def _make_llm_episode() -> Episode: group="LLMGroup", resolution="1080P", source="Web", + episode_type=episode_type, ) @@ -131,6 +132,20 @@ class TestRawParserLLMPrimaryMode: assert result is not None assert result.official_title == "LLM Title" + async def test_raw_non_episodic_type_overrides_llm_episode_type(self): + with patch.object( + settings, "llm", LLM(enable=True, api_key="k", mode="primary") + ): + with patch( + "module.parser.title_parser._llm_parse", + new_callable=AsyncMock, + return_value=_make_llm_episode(episode_type="episode"), + ): + result = await TitleParser.raw_parser("[Group] LLM Title Movie [1080P]") + + assert result is not None + assert result.episode_type == "movie" + async def test_llm_failure_falls_back_to_regex(self): """LLM 失败（返回 None）时用正则兜底，标题不丢失。""" with patch.object( @@ -188,6 +203,78 @@ class TestLLMParseErrorHandling: result = await title_parser_module._llm_parse("some title") assert result is None + async def test_llm_success_is_cached(self): + mock_parser = AsyncMock() + mock_parser.parse = AsyncMock( + return_value={ + "title_en": "Cached", + "title_zh": None, + "title_jp": None, + "season": 1, + "season_raw": "S1", + "episode": 1, + "sub": "", + "group": "", + "resolution": "", + "source": "", + } + ) + with patch.object(settings, "llm", LLM(enable=True, api_key="k", cache_ttl=60)): + with patch( + "module.parser.title_parser._get_llm_parser", return_value=mock_parser + ): + first = await title_parser_module._llm_parse("cached title") + second = await title_parser_module._llm_parse("cached title") + + assert first == second + assert mock_parser.parse.await_count == 1 + + async def test_llm_failures_are_cached(self): + mock_parser = AsyncMock() + mock_parser.parse = AsyncMock(side_effect=RuntimeError("boom")) + with patch.object( + settings, + "llm", + LLM( + enable=True, + api_key="k", + cache_ttl=60, + failure_threshold=10, + ), + ): + with patch( + "module.parser.title_parser._get_llm_parser", return_value=mock_parser + ): + first = await title_parser_module._llm_parse("bad title") + second = await title_parser_module._llm_parse("bad title") + + assert first is None + assert second is None + assert mock_parser.parse.await_count == 1 + + async def test_llm_failure_breaker_skips_provider_calls(self): + mock_parser = AsyncMock() + mock_parser.parse = AsyncMock(side_effect=RuntimeError("boom")) + with patch.object( + settings, + "llm", + LLM( + enable=True, + api_key="k", + cache_ttl=0, + failure_threshold=2, + failure_backoff=60, + ), + ): + with patch( + "module.parser.title_parser._get_llm_parser", return_value=mock_parser + ): + assert await title_parser_module._llm_parse("bad title 1") is None + assert await title_parser_module._llm_parse("bad title 2") is None + assert await title_parser_module._llm_parse("bad title 3") is None + + assert mock_parser.parse.await_count == 2 + class TestLLMConfigLegacyFallback: """llm 段缺失时，_llm_config 回退读取旧的 experimental_openai 段。""" @@ -225,8 +312,14 @@ class TestLLMParserResetCache: # clears it, not that it's a real LLMParser. title_parser_module._llm_parser = object() # type: ignore[assignment] title_parser_module._llm_parser_kwargs = {"api_key": "old"} + title_parser_module._llm_cache = {("old",): (1, None)} + title_parser_module._llm_failure_count = 2 + title_parser_module._llm_breaker_until = 999 title_parser_module.reset_cache() assert title_parser_module._llm_parser is None assert title_parser_module._llm_parser_kwargs is None + assert title_parser_module._llm_cache == {} + assert title_parser_module._llm_failure_count == 0 + assert title_parser_module._llm_breaker_until == 0.0 diff --git a/backend/src/test/test_update.py b/backend/src/test/test_update.py index bf96810b..45d63556 100644 --- a/backend/src/test/test_update.py +++ b/backend/src/test/test_update.py @@ -5,9 +5,11 @@ 覆盖层落地走 tmp_path，进程退出被打桩以确保测试不会真的退出。 """ +import asyncio import hashlib import io import json +import sqlite3 import zipfile from pathlib import Path from unittest.mock import patch @@ -92,6 +94,28 @@ def _make_client(handler) -> httpx.AsyncClient: return httpx.AsyncClient(transport=httpx.MockTransport(handler)) +def _seed_schema_db(path: Path, version: int) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(path) as conn: + conn.execute( + "CREATE TABLE IF NOT EXISTS schema_version (" + " id INTEGER PRIMARY KEY," + " version INTEGER NOT NULL" + ")" + ) + conn.execute( + "INSERT OR REPLACE INTO schema_version (id, version) VALUES (1, ?)", + (version,), + ) + + +def _read_schema_db(path: Path) -> int: + with sqlite3.connect(path) as conn: + return conn.execute( + "SELECT version FROM schema_version WHERE id = 1" + ).fetchone()[0] + + def _default_handler(bundle_bytes: bytes, sha_hex: str): """路由 GitHub API 与下载 URL 到对应的响应。""" @@ -216,6 +240,7 @@ def _updater_for_apply(tmp_path, data, sha, current="3.1.0", image="3.3.0-beta.1 root=tmp_path / "u", current_version=current, image_version=image, + dependency_syncer=lambda unpacked: None, client=_make_client(_default_handler(data, sha)), ) @@ -237,6 +262,59 @@ class TestApplyUpdate: assert applied["sha256"] == sha assert applied["lockfile_sha256"] + @pytest.mark.asyncio + async def test_success_snapshots_db_and_records_schema(self, tmp_path, bundle): + data, sha = bundle + db_path = tmp_path / "data" / "data.db" + _seed_schema_db(db_path, 7) + (tmp_path / "data" / "data.db-wal").write_bytes(b"wal") + (tmp_path / "data" / "data.db-shm").write_bytes(b"shm") + up = Updater( + root=tmp_path / "u", + current_version="3.1.0", + image_version="3.3.0-beta.1", + data_db=db_path, + dependency_syncer=lambda unpacked: None, + client=_make_client(_default_handler(data, sha)), + ) + + res = await up.apply_update("beta") + + assert res.success is True + applied = json.loads((tmp_path / "u" / "applied.json").read_text()) + assert applied["schema_version_before"] == 7 + # schema_version_after 在 apply 时不可知（迁移下次启动才跑），不得记录 + assert "schema_version_after" not in applied + backup_db = tmp_path / "u" / applied["db_backup"]["path"] + assert _read_schema_db(backup_db) == 7 + # backup API 快照是独立一致文件，不携带活库的 -wal/-shm 边车 + assert not (backup_db.parent / "data.db-wal").exists() + assert not (backup_db.parent / "data.db-shm").exists() + + @pytest.mark.asyncio + async def test_dependency_sync_failure_aborts_without_promote( + self, tmp_path, bundle + ): + data, sha = bundle + + def fail_sync(unpacked: Path) -> None: + raise RuntimeError("uv sync failed") + + up = Updater( + root=tmp_path / "u", + current_version="3.1.0", + image_version="3.3.0-beta.1", + dependency_syncer=fail_sync, + client=_make_client(_default_handler(data, sha)), + ) + + res = await up.apply_update("beta") + + assert res.success is False + assert "uv sync failed" in res.message + assert not (tmp_path / "u" / "current").exists() + assert not (tmp_path / "u" / "applied.json").exists() + @pytest.mark.asyncio async def test_sha_mismatch_aborts_without_promote(self, tmp_path, bundle): data, _ = bundle @@ -301,6 +379,144 @@ class TestRollback: applied = json.loads((root / "applied.json").read_text()) assert applied["version"] == "3.2.0" + @pytest.mark.asyncio + async def test_rollback_restores_db_snapshot(self, tmp_path): + root = tmp_path / "u" + current = root / "current" + backup = root / "backup" + current.mkdir(parents=True) + backup.mkdir(parents=True) + (current / "manifest.json").write_text(json.dumps({"version": "3.3.0"})) + (backup / "manifest.json").write_text(json.dumps({"version": "3.2.0"})) + db_path = tmp_path / "data" / "data.db" + _seed_schema_db(db_path, 9) + backup_db = root / "db-backup" / "data.db" + _seed_schema_db(backup_db, 4) + (root / "applied.json").write_text( + json.dumps( + { + "version": "3.3.0", + "schema_version_before": 4, + "db_backup": {"path": "db-backup/data.db"}, + } + ) + ) + + up = Updater( + root=root, + current_version="3.3.0", + image_version="3.1.0", + data_db=db_path, + ) + res = await up.rollback() + + assert res.success is True + # schema 自 apply 后前进过（4 -> 9），旧代码读不了新库，必须还原快照 + assert _read_schema_db(db_path) == 4 + applied = json.loads((root / "applied.json").read_text()) + assert applied["schema_version_before"] == 9 + assert applied["schema_version_after"] == 4 + # 快照目录用后即焚，避免下一轮回滚误还原陈旧快照 + assert not (root / "db-backup").exists() + + @pytest.mark.asyncio + async def test_rollback_keeps_live_db_when_schema_unchanged(self, tmp_path): + root = tmp_path / "u" + current = root / "current" + backup = root / "backup" + current.mkdir(parents=True) + backup.mkdir(parents=True) + (current / "manifest.json").write_text(json.dumps({"version": "3.3.0"})) + (backup / "manifest.json").write_text(json.dumps({"version": "3.2.0"})) + db_path = tmp_path / "data" / "data.db" + _seed_schema_db(db_path, 4) + backup_db = root / "db-backup" / "data.db" + _seed_schema_db(backup_db, 3) + (root / "applied.json").write_text( + json.dumps( + { + "version": "3.3.0", + "schema_version_before": 4, + "db_backup": {"path": "db-backup/data.db"}, + } + ) + ) + + up = Updater( + root=root, + current_version="3.3.0", + image_version="3.1.0", + data_db=db_path, + ) + res = await up.rollback() + + assert res.success is True + # schema 没变：保留活库，不得用快照抹掉更新后写入的数据 + assert _read_schema_db(db_path) == 4 + + @pytest.mark.asyncio + async def test_second_rollback_ignores_leftover_backup_dir(self, tmp_path): + root = tmp_path / "u" + current = root / "current" + backup = root / "backup" + current.mkdir(parents=True) + backup.mkdir(parents=True) + (current / "manifest.json").write_text(json.dumps({"version": "3.3.0"})) + (backup / "manifest.json").write_text(json.dumps({"version": "3.2.0"})) + db_path = tmp_path / "data" / "data.db" + _seed_schema_db(db_path, 9) + # 上一轮更新遗留的快照目录；本轮 applied.json 没有 db_backup 记录 + stale_backup = root / "db-backup" / "data.db" + _seed_schema_db(stale_backup, 3) + (root / "applied.json").write_text( + json.dumps( + { + "version": "3.3.0", + "rolled_back": True, + "schema_version_before": 5, + } + ) + ) + + up = Updater( + root=root, + current_version="3.3.0", + image_version="3.1.0", + data_db=db_path, + ) + res = await up.rollback() + + assert res.success is True + # 不得回退到磁盘残留的陈旧快照 + assert _read_schema_db(db_path) == 9 + + @pytest.mark.asyncio + async def test_rollback_refuses_unknown_future_schema(self, tmp_path): + root = tmp_path / "u" + current = root / "current" + backup = root / "backup" + current.mkdir(parents=True) + backup.mkdir(parents=True) + (current / "manifest.json").write_text(json.dumps({"version": "3.3.0"})) + (backup / "manifest.json").write_text(json.dumps({"version": "3.2.0"})) + db_path = tmp_path / "data" / "data.db" + _seed_schema_db(db_path, 999) + + up = Updater( + root=root, + current_version="3.3.0", + image_version="3.1.0", + data_db=db_path, + ) + with patch("module.update.updater.CURRENT_SCHEMA_VERSION", 14): + res = await up.rollback() + + assert res.success is False + assert "schema" in res.message + assert (backup / "manifest.json").exists() + cur = json.loads((current / "manifest.json").read_text()) + assert cur["version"] == "3.3.0" + @pytest.mark.asyncio async def test_rollback_without_backup_reverts_to_image(self, tmp_path): root = tmp_path / "u" @@ -339,6 +555,8 @@ class TestBootOverlay: ) (current / "webui-dist").mkdir(parents=True) (current / "webui-dist" / "new.html").write_text("overlay dist") + (current / ".venv").mkdir(parents=True) + (current / ".venv" / "pyvenv.cfg").write_text("overlay venv") (updates / "applied.json").write_text(json.dumps({"version": overlay_version})) image_version_path = app / "IMAGE_VERSION" @@ -361,6 +579,7 @@ class TestBootOverlay: assert (app / "module" / "new.txt").exists() assert not (app / "module" / "old.txt").exists() # tree replaced, not merged assert (app / "dist" / "new.html").exists() + assert (app / ".venv" / "pyvenv.cfg").read_text() == "overlay venv" def test_skips_and_clears_overlay_when_image_newer(self, tmp_path): import boot_overlay @@ -492,3 +711,31 @@ class TestUpdateApiContract: resp = authed_client.post("/api/v1/update/apply") assert resp.status_code == 400 sched.assert_not_called() + + def test_request_restart_writes_entrypoint_sentinel(self, tmp_path): + from module.api import update as update_api + + sentinel = tmp_path / "config" / "updates" / ".restart" + with ( + patch("module.api.update.RESTART_SENTINEL", sentinel), + patch("module.api.update.os.kill") as kill, + ): + update_api._request_restart() + + assert sentinel.exists() + kill.assert_called_once() + + @pytest.mark.asyncio + async def test_schedule_restart_retains_task_reference(self): + from module.api import update as update_api + + update_api._RESTART_TASKS.clear() + with ( + patch("module.api.update._RESTART_DELAY_SECONDS", 0), + patch("module.api.update._request_restart"), + ): + update_api.schedule_restart() + assert len(update_api._RESTART_TASKS) == 1 + await asyncio.gather(*list(update_api._RESTART_TASKS)) + await asyncio.sleep(0) + assert len(update_api._RESTART_TASKS) == 0 diff --git a/entrypoint.sh b/entrypoint.sh index e14d1287..6d1b400e 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -7,13 +7,40 @@ if [ -f /config/bangumi.json ]; then mv /config/bangumi.json /app/data/bangumi.json fi -# 应用持久卷中的在线更新覆盖层（在启动应用前、以 root 运行）。脚本内部已全程 -# 兜底 try/except，这里再加 `|| true`：覆盖层失败绝不能阻断容器启动。 -python /app/boot_overlay.py || true - groupmod -o -g "${PGID}" ab usermod -o -u "${PUID}" ab -chown ab:ab -R /app/data /app/config /home/ab +# 重启循环让 shell 站在 tini 和应用之间，因此必须自己转发信号并等待 +# 应用退出，否则 docker stop 时 shell 先死、tini 跟着退出，python 会被 +# SIGKILL 打断（SQLite WAL 来不及落盘）。 +child="" +forward_signal() { + if [ -n "${child}" ]; then + kill -TERM "${child}" 2>/dev/null + wait "${child}" + exit $? + fi + exit 0 +} +trap forward_signal TERM INT -exec su-exec "${PUID}:${PGID}" python main.py \ No newline at end of file +while true; do + # 应用持久卷中的在线更新覆盖层（在启动应用前、以 root 运行）。脚本内部已全程 + # 兜底 try/except，这里再加 `|| true`：覆盖层失败绝不能阻断容器启动。 + python /app/boot_overlay.py || true + + chown ab:ab -R /app/data /app/config /home/ab + + su-exec "${PUID}:${PGID}" python main.py & + child=$! + wait "${child}" + status=$? + child="" + + if [ -f /app/config/updates/.restart ]; then + rm -f /app/config/updates/.restart + continue + fi + + exit "${status}" +done diff --git a/webui/src/hooks/__tests__/useApi.test.ts b/webui/src/hooks/__tests__/useApi.test.ts index 5ab51636..cd6dafc4 100644 --- a/webui/src/hooks/__tests__/useApi.test.ts +++ b/webui/src/hooks/__tests__/useApi.test.ts @@ -14,7 +14,10 @@ interface Options { onFinally?: () => void; } - +type ApiExecuteResult = + | { ok: true; data: TData } + | { ok: false; error: unknown }; + type AnyAsyncFunction = (...args: any[]) => Promise; function createUseApi( @@ -24,15 +27,19 @@ function createUseApi( let data: Awaited> | undefined; let isLoading = false; - const execute = async (...params: Parameters): Promise => { + const execute = async ( + ...params: Parameters + ): Promise>>> => { options.onBeforeExecute?.(); isLoading = true; try { const res: Awaited> = await api(...params); data = res; options.onSuccess?.(res); + return { ok: true, data: res }; } catch (err) { options.onError?.(err); + return { ok: false, error: err }; } finally { isLoading = false; options.onFinally?.(); @@ -57,6 +64,16 @@ describe('useApi logic', () => { expect(mockApi).toHaveBeenCalledWith('param1', 'param2', 123); }); + it('should return ok result with data on success', async () => { + const responseData = { msg_en: 'Success', value: 42 }; + const mockApi = vi.fn().mockResolvedValue(responseData); + const { execute } = createUseApi(mockApi); + + const result = await execute(); + + expect(result).toEqual({ ok: true, data: responseData }); + }); + it('should set data to API response on success', async () => { const responseData = { msg_en: 'Success', value: 42 }; const mockApi = vi.fn().mockResolvedValue(responseData); @@ -101,6 +118,16 @@ describe('useApi logic', () => { expect(onError).toHaveBeenCalledWith(error); }); + it('should return ok false result when API throws', async () => { + const error = new Error('API Error'); + const mockApi = vi.fn().mockRejectedValue(error); + const { execute } = createUseApi(mockApi, { onError: vi.fn() }); + + const result = await execute(); + + expect(result).toEqual({ ok: false, error }); + }); + it('should call onFinally after success', async () => { const onFinally = vi.fn(); const mockApi = vi.fn().mockResolvedValue({}); diff --git a/webui/src/hooks/useApi.ts b/webui/src/hooks/useApi.ts index 6dff6463..1b9a7714 100644 --- a/webui/src/hooks/useApi.ts +++ b/webui/src/hooks/useApi.ts @@ -1,4 +1,8 @@ -type AnyAsyncFuntion = (...args: any[]) => Promise; +type AnyAsyncFunction = (...args: any[]) => Promise; + +export type ApiExecuteResult = + | { ok: true; data: TData } + | { ok: false; error: unknown }; interface Options { showMessage?: boolean; @@ -9,7 +13,7 @@ interface Options { } export function useApi< - TApi extends AnyAsyncFuntion = AnyAsyncFuntion, + TApi extends AnyAsyncFunction = AnyAsyncFunction, TData = Awaited> >( api: TApi, @@ -19,7 +23,7 @@ export function useApi< onSuccess, onError, onFinally, - }: Options = {} + }: Options = {} ) { const data = ref(); const isLoading = ref(false); @@ -27,21 +31,25 @@ export function useApi< const message = useMessage(); const { returnUserLangMsg } = useMyI18n(); - async function execute(...params: Parameters) { + async function execute( + ...params: Parameters + ): Promise> { onBeforeExecute?.(); try { isLoading.value = true; const res = await api(...params); - data.value = res; + data.value = res as TData; - onSuccess?.(res); + onSuccess?.(res as TData); - if (showMessage && 'msg_en' in res) { + if (showMessage && res && typeof res === 'object' && 'msg_en' in res) { message.success(returnUserLangMsg(res)); } + return { ok: true, data: res as TData }; } catch (err) { onError?.(err); + return { ok: false, error: err }; } finally { isLoading.value = false; onFinally?.(); diff --git a/webui/src/pages/index/config.vue b/webui/src/pages/index/config.vue index ebea7c68..b39cca83 100644 --- a/webui/src/pages/index/config.vue +++ b/webui/src/pages/index/config.vue @@ -10,11 +10,13 @@ const { isMobileOrTablet } = useBreakpointQuery(); const isSaving = ref(false); const isResetting = ref(false); +const saveFailed = ref(false); async function handleSave() { isSaving.value = true; try { - await setConfig(); + const result = await setConfig(); + saveFailed.value = !result.ok; } finally { isSaving.value = false; } @@ -24,6 +26,7 @@ async function handleReset() { isResetting.value = true; try { await getConfig(); + saveFailed.value = false; } finally { isResetting.value = false; } @@ -70,7 +73,7 @@ onActivated(() => { { // Tracks which single row is refreshing so only its button spins — the // store's isRefreshingAll only covers the "refresh all" button. const refreshingId = ref(null); +const refreshFailedId = ref(null); async function onRefreshOne(id: number) { refreshingId.value = id; try { - await refreshRSS(id); + const result = await refreshRSS(id); + refreshFailedId.value = result.ok ? null : id; } finally { refreshingId.value = null; } @@ -103,10 +105,11 @@ const rssColumns = computed>(() => [ width: 90, render(rss: RSS) { const isRowRefreshing = refreshingId.value === rss.id; + const isRowFailed = refreshFailedId.value === rss.id; return ( row.id;
{ const config = ref(initConfig); + const lastSaveError = ref(null); async function getConfig() { const res = await apiConfig.getConfig(); @@ -17,7 +18,11 @@ export const useConfigStore = defineStore('config', () => { }, }); - const setConfig = () => set(config.value); + const setConfig = async () => { + const result = await set(config.value); + lastSaveError.value = result.ok ? null : result.error; + return result; + }; function getSettingGroup(key: Tkey) { return computed({ @@ -32,6 +37,7 @@ export const useConfigStore = defineStore('config', () => { return { config, + lastSaveError, getConfig, setConfig, getSettingGroup,