nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-11-04 14:45:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(core): best-effort poster cache, abort boot on migration failure, non-blocking start

Browse Source 
- cache_image/save_image skip None bytes; cross_version caches posters per-item best-effort (one bad URL no longer aborts boot)
- run_migrations re-raises after rollback so boot aborts on a half-migrated schema; run migrations before legacy data_migration
- AppContext.start_tasks() is non-blocking; startup task is cancelled+awaited on shutdown
- Settings.save/init create config/ dir; robust version.info parse; AB_SOCKS writes proxy type "socks5"
- checker delegates downloader health to the client (aria2 can pass); add webauthn/notification settings fields

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014w1Z6Nxy6XTRgkFXqPr9Zh
This commit is contained in:
Estrella Pan
2026-07-02 17:05:34 +02:00
parent 9f9c7042dc
commit 4bab9d00c8
17 changed files with 166 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
backend/src/main.py
View File

@@ -1,4 +1,3 @@
import asyncio
import logging
import os
from contextlib import asynccontextmanager
@@ -33,15 +32,6 @@ uvicorn_logging_config = {
}
def _supervise_start_tasks(task: asyncio.Task) -> None:
"""Log-and-surface any exception from the background start task."""
if task.cancelled():
return
exc = task.exception()
if exc is not None:
logger.error("[Core] Background start task failed", exc_info=exc)
@asynccontextmanager
async def lifespan(app: FastAPI):
ctx: AppContext = app.state.ctx
@@ -50,8 +40,10 @@ async def lifespan(app: FastAPI):
# First run just created the DB — do not auto-start loops (matches the old
# Program.startup early return); the user starts them after setup.
if not ctx.first_run_boot:
start_task = asyncio.create_task(ctx.start_tasks())
start_task.add_done_callback(_supervise_start_tasks)
# start_tasks() itself schedules the downloader-wait + loop-start as a
# supervised background task and returns immediately (CONTRACT #5),
# so this does not block server startup.
await ctx.start_tasks()
yield
# Shutdown
await ctx.stop()

28
backend/src/module/checker/checker.py
View File

@@ -1,8 +1,6 @@
import logging
from pathlib import Path
import httpx
from module.conf import VERSION, settings
from module.models import Config
from module.update import version_check
@@ -65,27 +63,13 @@ class Checker:
logger.info("[Checker] Using MockDownloader - skipping connection check")
return True
# Delegate to the concrete client's own auth/capabilities instead of
# grepping the downloader's homepage HTML (which only ever matches
# qBittorrent, so aria2 and future backends could never pass).
try:
prefix = "https" if settings.downloader.ssl else "http"
url = (
f"{prefix}://{settings.downloader.host}"
if "://" not in settings.downloader.host
else f"{settings.downloader.host}"
)
async with httpx.AsyncClient(timeout=2.0) as client:
response = await client.get(url)
if "qbittorrent" in response.text.lower() or "vuetorrent" in response.text.lower():
async with DownloadClient() as dl_client:
if dl_client.authed:
return True
else:
return False
else:
return False
except httpx.TimeoutException:
logger.error("[Checker] Downloader connect timeout.")
return False
except httpx.ConnectError:
async with DownloadClient() as dl_client:
return dl_client.authed
except ConnectionError:
logger.error("[Checker] Downloader connect failed.")
return False
except Exception as e:

1
backend/src/module/conf/config.py
View File

@@ -85,6 +85,7 @@ class Settings(Config):
"""Write configuration to ``CONFIG_PATH``. Uses current state when no dict supplied."""
if not config_dict:
config_dict = self.model_dump()
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
json.dump(config_dict, f, indent=4, ensure_ascii=False)

6
backend/src/module/conf/const.py
View File

@@ -40,7 +40,7 @@ DEFAULT_SETTINGS = {
"username": "",
"password": "",
},
"notification": {"enable": False, "providers": []},
"notification": {"enable": False, "providers": [], "base_url": ""},
"experimental_openai": {
"enable": False,
"api_key": "",
@@ -63,6 +63,8 @@ DEFAULT_SETTINGS = {
"fc00::/7",
],
"mcp_tokens": [],
"webauthn_rp_id": "",
"webauthn_origin": "",
},
}
@@ -106,7 +108,7 @@ ENV_TO_ATTR = {
],
"AB_SOCKS": [
("enable", lambda e: True),
("type", lambda e: "socks"),
("type", lambda e: "socks5"),
("host", lambda e: e.split(",")[0]),
("port", lambda e: int(e.split(",")[1])),
("username", lambda e: e.split(",")[2]),

1
backend/src/module/conf/search_provider.py
View File

@@ -15,6 +15,7 @@ def load_provider():
if PROVIDER_PATH.exists():
return json_config.load(PROVIDER_PATH)
else:
PROVIDER_PATH.parent.mkdir(parents=True, exist_ok=True)
json_config.save(PROVIDER_PATH, DEFAULT_PROVIDER)
return DEFAULT_PROVIDER

54
backend/src/module/core/context.py
View File

@@ -11,6 +11,7 @@ import time
from module.checker import Checker
from module.conf import LEGACY_DATA_PATH, VERSION, settings
from module.database import Database
from module.downloader.download_client import shutdown as downloader_shutdown
from module.models import ResponseModel
from module.network.request_url import reset_shared_client
@@ -74,6 +75,10 @@ class AppContext:
self._downloader_last_check: float = 0.0
self._tasks_started = False
self._startup_done = False
# 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.
self._start_task: asyncio.Task | None = None
# True when this boot created the database (genuine first run); the
# lifespan then skips auto-starting background tasks, matching the old
# Program.startup() early-return.
@@ -177,6 +182,11 @@ class AppContext:
logger.info(
"[Core] Legacy data detected, starting data migration, please wait patiently."
)
# data_migration() writes into the bangumi/rssitem tables directly,
# so the schema must exist and be up to date first.
async with Database() as db:
await db.create_table()
await db.run_migrations()
await data_migration()
else:
is_same, last_minor = Checker.check_version()
@@ -214,22 +224,54 @@ class AppContext:
logger.info("Waiting for downloader to start...")
await asyncio.sleep(_DOWNLOADER_RETRY_INTERVAL)
async def start_tasks(self) -> ResponseModel:
"""Wait for the downloader, then start all enabled background loops."""
settings.load()
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
logger.info("Program running.")
@staticmethod
def _supervise_start_task(task: asyncio.Task) -> None:
"""Log-and-surface any exception from the background start task."""
if task.cancelled():
return
exc = task.exception()
if exc is not None:
logger.error("[Core] Background start task failed", exc_info=exc)
async def start_tasks(self) -> ResponseModel:
"""Kick off the downloader-wait + loop-start in the background and return immediately.
The downloader wait-retry loop can take up to ~300s (CONTRACT #5); running
it inline used to block both the lifespan and the /start,/restart API
handlers for that long. It now runs as a supervised background task
tracked on ``self._start_task`` so ``stop()`` can cancel it cleanly.
"""
settings.load()
if 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(
status=True,
status_code=200,
msg_en="Program started.",
msg_zh="程序启动成功。",
msg_en="Program starting.",
msg_zh="程序启动中。",
)
async def stop(self) -> ResponseModel:
"""Stop background loops and close the cached downloader session."""
"""Cancel any pending start task, stop background loops, and close the cached downloader session."""
if self._start_task is not None:
self._start_task.cancel()
try:
await self._start_task
except asyncio.CancelledError:
pass
except Exception as e:
logger.warning(
f"[Core] Background start task failed during shutdown: {e}"
)
self._start_task = None
if self.is_running:
await self.scheduler.stop_all()
self._tasks_started = False

19
backend/src/module/core/loops.py
View File

@@ -29,11 +29,20 @@ async def rss_tick(analyser: RSSAnalyser) -> None:
try:
await analyser.rss_to_data(rss, engine)
except Exception:
# RSS 可能在遍历期间被 API 删除，跳过即可
logger.debug(
"[RSSThread] Skipping RSS id=%s, likely deleted",
rss.id if hasattr(rss, "id") else "?",
)
# 仅当该 RSS 确实已在遍历期间被 API 删除时才视为预期情况；
# 其余异常可能是真实 bug，需在 WARNING 级别带完整堆栈。
if await db.rss.search_id(rss.id) is None:
logger.debug(
"[RSSThread] Skipping RSS id=%s, deleted during iteration",
rss.id,
)
else:
logger.warning(
"[RSSThread] Error analysing RSS id=%s (%s)",
rss.id,
rss.url,
exc_info=True,
)
# Run RSS Engine
await engine.refresh_rss(client)
if settings.bangumi_manage.eps_complete:

9
backend/src/module/database/migrations.py
View File

@@ -184,8 +184,9 @@ def set_schema_version(conn: Connection, version: int) -> None:
def run_migrations_conn(conn: Connection) -> None:
"""在单个连接上执行所有待应用迁移。
每个迁移包在 SAVEPOINT 中执行：失败即回滚该迁移并停止，之前成功的
迁移保持生效（与旧实现的可观测行为一致）。
每个迁移包在 SAVEPOINT 中执行：失败即回滚该迁移并重新抛出异常，使外层
事务整体回滚、启动失败并高声中止 —— 不允许应用在半迁移的 schema 上继续
提供服务。
"""
ensure_schema_version_table(conn)
current = get_schema_version(conn)
@@ -213,7 +214,9 @@ def run_migrations_conn(conn: Connection) -> None:
except Exception as e:
savepoint.rollback()
logger.error(f"[Database] Migration v{migration.version} failed: {e}")
break
# Abort loudly instead of silently continuing to serve on a
# half-migrated schema; the outer transaction rolls back.
raise
set_schema_version(conn, migration.version)
logger.info(f"[Database] Schema version is now {get_schema_version(conn)}.")
fill_null_with_defaults_conn(conn)

21
backend/src/module/models/config.py
View File

@@ -180,6 +180,13 @@ class Notification(BaseModel):
providers: list[NotificationProvider] = Field(
default_factory=list, description="List of notification providers"
)
base_url: str = Field(
"",
description=(
"Public base URL used to build absolute poster URLs for "
"notification providers. Empty = omit the poster field entirely."
),
)
# Legacy fields for backward compatibility (deprecated)
type: Optional[str] = Field(None, description="[Deprecated] Use providers instead")
@@ -260,6 +267,20 @@ class Security(BaseModel):
default_factory=list,
description="API bearer tokens for MCP access.",
)
webauthn_rp_id: str = Field(
"",
description=(
"WebAuthn relying-party ID. Empty = derive from the request "
"headers instead."
),
)
webauthn_origin: str = Field(
"",
description=(
"Expected WebAuthn origin. Empty = derive from the request "
"headers instead."
),
)
class Config(BaseModel):

20
backend/src/module/update/cross_version.py
View File

@@ -57,9 +57,19 @@ async def cache_image():
async with RequestContent() as req:
for bangumi in bangumis:
if bangumi.poster_link:
# Hash local path
img = await req.get_content(bangumi.poster_link)
suffix = bangumi.poster_link.split(".")[-1]
img_path = save_image(img, suffix)
bangumi.poster_link = img_path
# Best-effort: a single dead/unreachable poster link must
# not abort the whole migration for every other bangumi.
try:
# Hash local path
img = await req.get_content(bangumi.poster_link)
suffix = bangumi.poster_link.split(".")[-1]
img_path = save_image(img, suffix)
if img_path:
bangumi.poster_link = img_path
except Exception as e:
logger.warning(
"[Migration] Failed to cache poster for %s: %s",
bangumi.official_title,
e,
)
await db.bangumi.update_all(bangumis)

18
backend/src/module/update/version_check.py
View File

@@ -1,7 +1,11 @@
import logging
import semver
from module.conf import VERSION, VERSION_PATH
logger = logging.getLogger(__name__)
def version_check() -> tuple[bool, int | None]:
"""Check if version has changed.
@@ -22,8 +26,18 @@ def version_check() -> tuple[bool, int | None]:
with open(VERSION_PATH, "r+") as f:
# Read last version
versions = f.readlines()
last_version = versions[-1].strip()
last_ver = semver.VersionInfo.parse(last_version)
try:
last_version = versions[-1].strip()
last_ver = semver.VersionInfo.parse(last_version)
except (IndexError, ValueError) as e:
logger.warning(
f"[Version] {VERSION_PATH} is empty or malformed ({e}); "
"rewriting with the current version."
)
f.seek(0)
f.truncate()
f.write(VERSION + "\n")
return True, None
now_ver = semver.VersionInfo.parse(VERSION)
if now_ver.minor == last_ver.minor:
return True, None

2
backend/src/module/utils/__init__.py
View File

@@ -1 +1 @@
from .cache_image import save_image, load_image
from .cache_image import load_image, save_image

4
backend/src/module/utils/cache_image.py
View File

@@ -2,6 +2,10 @@ import hashlib
def save_image(img, suffix):
if img is None:
# Fetching the poster failed upstream; skip caching instead of
# crashing on hashlib.md5(None).
return None
img_hash = hashlib.md5(img).hexdigest()[0:8]
image_path = f"data/posters/{img_hash}.{suffix}"
with open(image_path, "wb") as f:

7
backend/src/test/test_api_config.py
View File

@@ -1,18 +1,17 @@
"""Tests for Config API endpoints and config sanitization."""
import pytest
from unittest.mock import patch, AsyncMock, MagicMock
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from module.api import v1
from module.api.config import _sanitize_dict, _restore_masked
from module.api.config import _restore_masked, _sanitize_dict
from module.api.deps import get_context
from module.models.config import Config
from module.security.api import get_current_user
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

11
backend/src/test/test_config.py
View File

@@ -2,22 +2,25 @@
import json
import os
import pytest
from pathlib import Path
from unittest.mock import patch
import pytest
from module.conf.config import Settings
from module.conf.const import BCOLORS, DEFAULT_SETTINGS
from module.models.config import (
Config,
Downloader,
Notification as NotificationConfig,
NotificationProvider,
Program,
Proxy,
RSSParser,
Security,
)
from module.conf.config import Settings
from module.conf.const import BCOLORS, DEFAULT_SETTINGS
from module.models.config import (
Notification as NotificationConfig,
)
# ---------------------------------------------------------------------------
# Config model defaults

4
backend/src/test/test_context.py
View File

@@ -119,6 +119,10 @@ class TestStartTasks:
),
):
resp = await ctx.start_tasks()
# start_tasks() only kicks off the downloader-wait + scheduler-start
# as a supervised background task and returns immediately; await it
# here so the assertions below observe its effects.
await ctx._start_task
ctx.scheduler.start_all.assert_called_once()
assert ctx._tasks_started is True

13
backend/src/test/test_migrations_module.py
View File

@@ -1,6 +1,8 @@
"""Tests for the table-driven migration system (module.database.migrations)."""
import pytest
from sqlalchemy import inspect, text
from sqlalchemy.exc import OperationalError
from sqlmodel import create_engine
from module.database.migrations import (
@@ -101,13 +103,20 @@ class TestRunMigrations:
assert "weekday_locked" in bangumi_cols # later migrations still ran
def test_failure_stops_before_bumping_version(self):
"""A failing migration must not record its version as applied."""
"""A failing migration must not record its version as applied.
run_migrations aborts loudly on failure (so startup aborts too, see
AppContext.startup) rather than swallowing it -- the caller is
expected to observe the exception.
"""
engine = _make_v0_engine()
with engine.begin() as conn:
# Sabotage migration 1: its target table is gone, the guard says
# "not applied", and the ALTER then fails.
conn.execute(text("DROP TABLE bangumi"))
run_migrations(engine)
with pytest.raises(OperationalError):
run_migrations(engine)
with engine.connect() as conn:
version = get_schema_version(conn)