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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FAxVyRwrY7z5NotTtgnwVs
This commit is contained in:
Estrella Pan
2026-07-03 10:52:09 +02:00
parent b34fa62403
commit 98e016ce2e
36 changed files with 1792 additions and 193 deletions

View File

@@ -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

View File

@@ -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,

View File

@@ -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)])

View File

@@ -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()

View File

@@ -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)

View File

@@ -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(

View File

@@ -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",

View File

@@ -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]: ...

View File

@@ -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
# ------------------------------------------------------------------

View File

@@ -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]:

View File

@@ -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(

View File

@@ -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}"

View File

@@ -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))

View File

@@ -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 解析配置,仅保留用于读取旧配置文件(向后兼容)。

View File

@@ -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)

View File

@@ -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,

View File

@@ -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,

View File

@@ -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

View File

@@ -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

View File

@@ -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")

View File

@@ -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(

View File

@@ -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
):

View File

@@ -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:

View File

@@ -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.

View File

@@ -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(

View File

@@ -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)
# ---------------------------------------------------------------------------

View File

@@ -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
# ---------------------------------------------------------------------------

View File

@@ -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:

View File

@@ -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

View File

@@ -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

View File

@@ -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
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

View File

@@ -14,7 +14,10 @@ interface Options<T = unknown> {
onFinally?: () => void;
}
type ApiExecuteResult<TData> =
| { ok: true; data: TData }
| { ok: false; error: unknown };
type AnyAsyncFunction = (...args: any[]) => Promise<any>;
function createUseApi<TApi extends AnyAsyncFunction>(
@@ -24,15 +27,19 @@ function createUseApi<TApi extends AnyAsyncFunction>(
let data: Awaited<ReturnType<TApi>> | undefined;
let isLoading = false;
const execute = async (...params: Parameters<TApi>): Promise<void> => {
const execute = async (
...params: Parameters<TApi>
): Promise<ApiExecuteResult<Awaited<ReturnType<TApi>>>> => {
options.onBeforeExecute?.();
isLoading = true;
try {
const res: Awaited<ReturnType<TApi>> = 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({});

View File

@@ -1,4 +1,8 @@
type AnyAsyncFuntion<TData = any> = (...args: any[]) => Promise<TData>;
type AnyAsyncFunction<TData = any> = (...args: any[]) => Promise<TData>;
export type ApiExecuteResult<TData> =
| { ok: true; data: TData }
| { ok: false; error: unknown };
interface Options<T = any> {
showMessage?: boolean;
@@ -9,7 +13,7 @@ interface Options<T = any> {
}
export function useApi<
TApi extends AnyAsyncFuntion = AnyAsyncFuntion,
TApi extends AnyAsyncFunction = AnyAsyncFunction,
TData = Awaited<ReturnType<TApi>>
>(
api: TApi,
@@ -19,7 +23,7 @@ export function useApi<
onSuccess,
onError,
onFinally,
}: Options = {}
}: Options<TData> = {}
) {
const data = ref<TData>();
const isLoading = ref(false);
@@ -27,21 +31,25 @@ export function useApi<
const message = useMessage();
const { returnUserLangMsg } = useMyI18n();
async function execute(...params: Parameters<TApi>) {
async function execute(
...params: Parameters<TApi>
): Promise<ApiExecuteResult<TData>> {
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?.();

View File

@@ -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(() => {
<NButton
:size="isMobileOrTablet ? 'large' : 'medium'"
:class="[{ 'flex-1': isMobileOrTablet }]"
type="primary"
:type="saveFailed ? 'error' : 'primary'"
:loading="isSaving"
:disabled="isResetting || isSaving"
@click="handleSave"

View File

@@ -25,11 +25,13 @@ 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<number | null>(null);
const refreshFailedId = ref<number | null>(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<DataTableColumns<RSS>>(() => [
width: 90,
render(rss: RSS) {
const isRowRefreshing = refreshingId.value === rss.id;
const isRowFailed = refreshFailedId.value === rss.id;
return (
<NButton
size="small"
type="primary"
type={isRowFailed ? 'error' : 'primary'}
secondary
{...{ title: t('rss.refresh') }}
disabled={isRowRefreshing}
@@ -189,7 +192,7 @@ const rssRowKey = (row: RSS) => row.id;
<div class="rss-card-actions" @click.stop>
<NButton
size="small"
type="primary"
:type="refreshFailedId === item.id ? 'error' : 'primary'"
secondary
:title="$t('rss.refresh')"
:disabled="refreshingId === item.id"

View File

@@ -2,6 +2,7 @@ import { type Config, initConfig } from '#/config';
export const useConfigStore = defineStore('config', () => {
const config = ref<Config>(initConfig);
const lastSaveError = ref<unknown>(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<Tkey extends keyof Config>(key: Tkey) {
return computed<Config[Tkey]>({
@@ -32,6 +37,7 @@ export const useConfigStore = defineStore('config', () => {
return {
config,
lastSaveError,
getConfig,
setConfig,
getSettingGroup,