diff --git a/backend/src/ab_sdk/events.py b/backend/src/ab_sdk/events.py index 2faeb769..f16f6672 100644 --- a/backend/src/ab_sdk/events.py +++ b/backend/src/ab_sdk/events.py @@ -166,12 +166,13 @@ class DownloaderUnavailableEvent(SystemEvent): host: str reason: str # unreachable | credentials | banned + instance_id: str = "default" # 下载器实例 id（plugins.instances） def dedup_key(self) -> str | None: - return f"downloader:{self.host}" + return f"downloader:{self.instance_id}" def payload(self) -> dict: - return {"host": self.host, "reason": self.reason} + return {"host": self.host, "reason": self.reason, "instance": self.instance_id} def describe(self) -> tuple[str, str]: details = { @@ -180,7 +181,10 @@ class DownloaderUnavailableEvent(SystemEvent): "unreachable": "无法连接下载器，请检查地址、端口和网络。", } detail = details.get(self.reason, details["unreachable"]) - return ("下载器连接异常", f"下载器：{self.host}\n{detail}") + return ( + "下载器连接异常", + f"下载器：{self.instance_id}（{self.host}）\n{detail}", + ) @dataclass(frozen=True, slots=True) diff --git a/backend/src/module/api/bangumi.py b/backend/src/module/api/bangumi.py index c2138249..0312f33b 100644 --- a/backend/src/module/api/bangumi.py +++ b/backend/src/module/api/bangumi.py @@ -6,8 +6,8 @@ from pydantic import BaseModel from module.conf import settings from module.database import Database, get_db -from module.downloader import DownloadClient -from module.manager import Renamer, TorrentManager +from module.manager import TorrentManager +from module.manager.renamer import rename_all from module.models import APIResponse, Bangumi, BangumiUpdate, ResponseModel, Torrent from module.parser.analyser.offset_detector import detect_offset_mismatch from module.parser.analyser.tmdb_parser import tmdb_parser @@ -376,9 +376,7 @@ async def dismiss_review(bangumi_id: int): async def _trigger_rename() -> None: """Run a rename pass so applied offsets take effect immediately.""" - async with DownloadClient() as client: - renamer = Renamer(client) - await renamer.rename() + await rename_all() # Registered before /apply-offset/{bangumi_id} so "many" is never captured as an id. diff --git a/backend/src/module/api/downloader.py b/backend/src/module/api/downloader.py index 040b9936..52eabfe0 100644 --- a/backend/src/module/api/downloader.py +++ b/backend/src/module/api/downloader.py @@ -4,13 +4,15 @@ import re from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel +from module.conf import settings from module.database import Database from module.database.bangumi import ( build_save_path_index, match_bangumi_in_list, normalize_save_path, ) -from module.downloader import DownloadClient +from module.downloader import DownloadClient, downloader_ids, list_torrents +from module.models.config import DOWNLOADER_POINT from module.security.api import get_current_user logger = logging.getLogger(__name__) @@ -20,10 +22,11 @@ router = APIRouter(prefix="/downloader", tags=["downloader"]) class TorrentHashesRequest(BaseModel): hashes: list[str] + # 种子所在的下载器实例；为空时为默认实例 + downloader_id: str | None = None -class TorrentDeleteRequest(BaseModel): - hashes: list[str] +class TorrentDeleteRequest(TorrentHashesRequest): delete_files: bool = False @@ -32,12 +35,37 @@ class TorrentTagRequest(BaseModel): hash: str bangumi_id: int + downloader_id: str | None = None + + +def _client(downloader_id: str | None) -> DownloadClient: + if downloader_id is not None and downloader_id not in downloader_ids(): + raise HTTPException( + status_code=404, detail=f"Downloader instance {downloader_id} not found" + ) + return DownloadClient(downloader_id) + + +@router.get("/instances", dependencies=[Depends(get_current_user)]) +async def get_instances(): + """下载器实例列表（供规则 / 订阅选择下载器）。""" + return { + "default": settings.plugins.slots.downloader, + "instances": [ + {"id": i.id, "provider": i.provider} + for i in settings.plugins.instances + if i.point == DOWNLOADER_POINT + ], + } @router.get("/torrents", dependencies=[Depends(get_current_user)]) async def get_torrents(): - async with DownloadClient() as client: - return await client.get_torrent_info(category="Bangumi", status_filter=None) + """所有下载器实例的种子（每条带 downloader_id）；不可用的实例跳过。""" + torrents = await list_torrents() + if torrents is None: + raise HTTPException(status_code=503, detail="No downloader is reachable") + return torrents @router.get("/rename-conflicts", dependencies=[Depends(get_current_user)]) @@ -79,7 +107,7 @@ async def retry_rename_conflict(operation_id: int): @router.post("/torrents/pause", dependencies=[Depends(get_current_user)]) async def pause_torrents(req: TorrentHashesRequest): hashes = "|".join(req.hashes) - async with DownloadClient() as client: + async with _client(req.downloader_id) as client: await client.pause_torrent(hashes) return {"msg_en": "Torrents paused", "msg_zh": "种子已暂停"} @@ -87,7 +115,7 @@ async def pause_torrents(req: TorrentHashesRequest): @router.post("/torrents/resume", dependencies=[Depends(get_current_user)]) async def resume_torrents(req: TorrentHashesRequest): hashes = "|".join(req.hashes) - async with DownloadClient() as client: + async with _client(req.downloader_id) as client: await client.resume_torrent(hashes) return {"msg_en": "Torrents resumed", "msg_zh": "种子已恢复"} @@ -95,7 +123,7 @@ async def resume_torrents(req: TorrentHashesRequest): @router.post("/torrents/delete", dependencies=[Depends(get_current_user)]) async def delete_torrents(req: TorrentDeleteRequest): hashes = "|".join(req.hashes) - async with DownloadClient() as client: + async with _client(req.downloader_id) as client: ok = await client.delete_torrent(hashes, delete_files=req.delete_files) if not ok: return { @@ -124,7 +152,7 @@ async def tag_torrent(req: TorrentTagRequest): } tag = f"ab:{req.bangumi_id}" - async with DownloadClient() as client: + async with _client(req.downloader_id) as client: await client.add_tag(req.hash, tag) return { @@ -150,42 +178,45 @@ async def auto_tag_torrents(): bangumi_list = await db.bangumi.search_all() save_path_index = build_save_path_index(bangumi_list) - async with DownloadClient() as client: - # Get all Bangumi torrents - torrents = await client.get_torrent_info(category="Bangumi", status_filter=None) + for instance_id in downloader_ids(): + async with DownloadClient(instance_id) as client: + # Get all Bangumi torrents + torrents = await client.get_torrent_info( + category="Bangumi", status_filter=None + ) - for torrent in torrents: - torrent_hash = torrent["hash"] - torrent_name = torrent["name"] - save_path = torrent["save_path"] - tags = torrent.get("tags", "") + for torrent in torrents: + torrent_hash = torrent["hash"] + torrent_name = torrent["name"] + save_path = torrent["save_path"] + tags = torrent.get("tags", "") - # Skip if already has an ab: link tag。必须精确匹配数字 id： - # ab:renamed（处理完成标记）等同前缀标签不代表已关联番剧 - if re.search(r"ab:\d+", tags): - continue + # Skip if already has an ab: link tag。必须精确匹配数字 id： + # ab:renamed（处理完成标记）等同前缀标签不代表已关联番剧 + if re.search(r"ab:\d+", tags): + continue - # First try by torrent name, then fall back to save_path - bangumi = match_bangumi_in_list(torrent_name, bangumi_list) - if not bangumi: - bangumi = save_path_index.get(normalize_save_path(save_path)) + # First try by torrent name, then fall back to save_path + bangumi = match_bangumi_in_list(torrent_name, bangumi_list) + if not bangumi: + bangumi = save_path_index.get(normalize_save_path(save_path)) - if bangumi and not bangumi.deleted: - tag = f"ab:{bangumi.id}" - await client.add_tag(torrent_hash, tag) - tagged_count += 1 - logger.info( - f"Tagged '{torrent_name[:50]}...' with {tag} " - f"(matched: {bangumi.official_title})" - ) - else: - unmatched.append( - { - "hash": torrent_hash, - "name": torrent_name, - "save_path": save_path, - } - ) + if bangumi and not bangumi.deleted: + tag = f"ab:{bangumi.id}" + await client.add_tag(torrent_hash, tag) + tagged_count += 1 + logger.info( + f"Tagged '{torrent_name[:50]}...' with {tag} " + f"(matched: {bangumi.official_title})" + ) + else: + unmatched.append( + { + "hash": torrent_hash, + "name": torrent_name, + "save_path": save_path, + } + ) return { "status": True, diff --git a/backend/src/module/api/events.py b/backend/src/module/api/events.py index ca00bbbc..2d3070e4 100644 --- a/backend/src/module/api/events.py +++ b/backend/src/module/api/events.py @@ -22,7 +22,7 @@ from module.api.log import _read_log_tail from module.conf import LOG_PATH, VERSION from module.core import AppContext from module.database import Database -from module.downloader import DownloadClient +from module.downloader import list_torrents from module.notification.inbox import InboxChanged, inbox_revision from module.plugin.host import get_bus from module.security.api import get_current_user @@ -54,25 +54,14 @@ def _status_payload(ctx: AppContext) -> dict: } -async def _fetch_torrents() -> list[dict]: - async with DownloadClient() as client: - return await client.get_torrent_info(category="Bangumi", status_filter=None) - - async def _downloader_payload() -> list[dict] | None: - """获取种子列表；下载器未配置、不可达或超时时返回 None。 + """获取所有下载器实例的种子列表；全部实例不可达或超时时返回 None。 - asyncio.wait_for 超时会取消内部任务并等待其退出，慢查询不会泄漏； - None 会以 null 推送给前端，作为显式的"下载器不可用"信号。 + 超时按实例计算，一个实例不可用不影响其它实例的种子；asyncio.wait_for + 超时会取消内部任务并等待其退出，慢查询不会泄漏。None 会以 null 推送给 + 前端，作为显式的"下载器不可用"信号。 """ - try: - return await asyncio.wait_for(_fetch_torrents(), _DOWNLOADER_TIMEOUT_SECONDS) - except asyncio.TimeoutError: - logger.debug("SSE: downloader status fetch timed out") - return None - except Exception: - logger.debug("SSE: downloader status unavailable", exc_info=True) - return None + return await list_torrents(timeout=_DOWNLOADER_TIMEOUT_SECONDS) async def _notification_payload() -> dict: diff --git a/backend/src/module/api/rss.py b/backend/src/module/api/rss.py index 5152b466..eb4588f6 100644 --- a/backend/src/module/api/rss.py +++ b/backend/src/module/api/rss.py @@ -2,7 +2,7 @@ from fastapi import APIRouter, Depends from fastapi.responses import JSONResponse from module.database import Database, get_db -from module.downloader import DownloadClient +from module.downloader import DownloadClient, DownloaderPool, resolve_downloader_id from module.manager import SeasonCollector from module.models import APIResponse, Bangumi, Movie, RSSItem, RSSUpdate, Torrent from module.rss import RSSAnalyser, RSSEngine @@ -140,9 +140,9 @@ async def update_rss( dependencies=[Depends(get_current_user)], ) async def refresh_all(db: Database = Depends(get_db)): - async with DownloadClient() as client: + async with DownloaderPool() as downloaders: engine = RSSEngine(db) - await engine.refresh_rss(client) + await engine.refresh_rss(downloaders) return JSONResponse( status_code=200, content={ @@ -158,9 +158,9 @@ async def refresh_all(db: Database = Depends(get_db)): dependencies=[Depends(get_current_user)], ) async def refresh_rss(rss_id: int, db: Database = Depends(get_db)): - async with DownloadClient() as client: + async with DownloaderPool() as downloaders: engine = RSSEngine(db) - await engine.refresh_rss(client, rss_id) + await engine.refresh_rss(downloaders, rss_id) return JSONResponse( status_code=200, content={"msg_en": "Refresh RSS successfully.", "msg_zh": "刷新 RSS 成功。"}, @@ -200,7 +200,7 @@ async def analysis(rss: RSSItem): "/collect", response_model=APIResponse, dependencies=[Depends(get_current_user)] ) async def download_collection(data: Bangumi): - async with DownloadClient() as client: + async with DownloadClient(resolve_downloader_id(data.downloader_id)) as client: collector = SeasonCollector(client) resp = await collector.collect_season(data, data.rss_link) return u_response(resp) diff --git a/backend/src/module/core/context.py b/backend/src/module/core/context.py index 3f474c67..3da95aaa 100644 --- a/backend/src/module/core/context.py +++ b/backend/src/module/core/context.py @@ -271,7 +271,9 @@ class AppContext: try: await self.notifier.send_event( DownloaderUnavailableEvent( - host=self.settings.downloader.host, reason=reason + host=self.settings.downloader.host, + reason=reason, + instance_id=self.settings.plugins.slots.downloader, ) ) except Exception: diff --git a/backend/src/module/core/loops.py b/backend/src/module/core/loops.py index 261fce9d..da405e96 100644 --- a/backend/src/module/core/loops.py +++ b/backend/src/module/core/loops.py @@ -9,8 +9,9 @@ import logging from module.conf import settings from module.database import Database -from module.downloader import DownloadClient -from module.manager import Renamer, TorrentManager, eps_complete +from module.downloader import DownloaderPool +from module.manager import TorrentManager, eps_complete +from module.manager.renamer import rename_all from module.notification import NotificationManager, UpdateAvailableEvent from module.rss import RSSAnalyser, RSSEngine from module.update.updater import updater @@ -22,7 +23,7 @@ logger = logging.getLogger(__name__) async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None: """Analyse aggregate RSS feeds and refresh the RSS engine once.""" - async with DownloadClient() as client: + async with DownloaderPool() as downloaders: async with Database() as db: engine = RSSEngine(db) # Analyse RSS @@ -46,7 +47,7 @@ async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None exc_info=True, ) # Run RSS Engine - events = await engine.refresh_rss(client) + events = await engine.refresh_rss(downloaders) if events: # 站内通知中心要求事件总是送达 notifier（send_event 内部先落库， # 再按 settings.notification.enable 决定是否外发）。 @@ -61,11 +62,8 @@ async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None async def rename_tick(notifier: NotificationManager) -> None: - """Rename completed downloads and notify via the shared notifier.""" - async with DownloadClient() as client: - renamer = Renamer(client) - renamed_info = await renamer.rename() - rename_events = list(renamer.events) + """Rename completed downloads on every downloader instance and notify.""" + renamed_info, rename_events = await rename_all() if rename_events: await asyncio.gather(*[notifier.send_event(event) for event in rename_events]) if settings.notification.enable and renamed_info: diff --git a/backend/src/module/database/migrations.py b/backend/src/module/database/migrations.py index f7ea4687..cb5d0878 100644 --- a/backend/src/module/database/migrations.py +++ b/backend/src/module/database/migrations.py @@ -100,8 +100,14 @@ class Migration: return self.statements -# 存量行归属默认下载器实例（plugins.instances 中 id 为 default 的实例） -_DOWNLOADER_ID_TABLES = ("bangumi", "movie", "rssitem", "torrent") +# 规则与订阅的 downloader_id 为空表示「用默认实例」，存量行保持为空； +# 种子记录实际所在的实例，存量种子都在 3.3 的下载器，即实例 default +_DOWNLOADER_ID_COLUMNS = { + "bangumi": "downloader_id VARCHAR", + "movie": "downloader_id VARCHAR", + "rssitem": "downloader_id VARCHAR", + "torrent": "downloader_id VARCHAR DEFAULT 'default'", +} def _has_downloader_id(table: str) -> AppliedCheck: @@ -817,17 +823,13 @@ MIGRATIONS: tuple[Migration, ...] = ( 26, "add downloader_id to bangumi, movie, rssitem and torrent", tuple( - f"ALTER TABLE {table} ADD COLUMN downloader_id VARCHAR DEFAULT 'default'" - for table in _DOWNLOADER_ID_TABLES + f"ALTER TABLE {table} ADD COLUMN {column}" + for table, column in _DOWNLOADER_ID_COLUMNS.items() ), - all_checks(*(_has_downloader_id(table) for table in _DOWNLOADER_ID_TABLES)), + all_checks(*(_has_downloader_id(table) for table in _DOWNLOADER_ID_COLUMNS)), tuple( - ( - f"ALTER TABLE {table} ADD COLUMN downloader_id VARCHAR " - "DEFAULT 'default'", - _has_downloader_id(table), - ) - for table in _DOWNLOADER_ID_TABLES + (f"ALTER TABLE {table} ADD COLUMN {column}", _has_downloader_id(table)) + for table, column in _DOWNLOADER_ID_COLUMNS.items() ), ), ) diff --git a/backend/src/module/database/rename_operation.py b/backend/src/module/database/rename_operation.py index 625d9a8b..f07deeff 100644 --- a/backend/src/module/database/rename_operation.py +++ b/backend/src/module/database/rename_operation.py @@ -356,10 +356,14 @@ class RenameOperationDatabase: ) return list(result.scalars().all()) - async def list_active_replacements(self, limit: int = 100) -> list[RenameOperation]: + async def list_active_replacements( + self, downloader_type: str, limit: int = 100 + ) -> list[RenameOperation]: + """某个下载器实例（``downloader_type``）上未完成的版本替换事务。""" result = await self.session.execute( select(RenameOperation) .where( + RenameOperation.downloader_type == downloader_type, RenameOperation.kind == "replacement", col(RenameOperation.state).in_( ("planned", "old_staged", "new_promoted", "old_removed") diff --git a/backend/src/module/downloader/__init__.py b/backend/src/module/downloader/__init__.py index abf2f5bb..1fdb1903 100644 --- a/backend/src/module/downloader/__init__.py +++ b/backend/src/module/downloader/__init__.py @@ -6,15 +6,26 @@ from .base import ( RenameOutcome, RenameResult, ) -from .download_client import DownloadClient, shutdown +from .download_client import ( + DownloadClient, + DownloaderPool, + downloader_ids, + list_torrents, + resolve_downloader_id, + shutdown, +) __all__ = [ "AddResult", "CoreDownloaderClient", "DownloadClient", + "DownloaderPool", "DownloaderCapabilities", "DownloaderClient", "RenameOutcome", "RenameResult", + "downloader_ids", + "list_torrents", + "resolve_downloader_id", "shutdown", ] diff --git a/backend/src/module/downloader/download_client.py b/backend/src/module/downloader/download_client.py index 63a878bd..27e47f17 100644 --- a/backend/src/module/downloader/download_client.py +++ b/backend/src/module/downloader/download_client.py @@ -1,12 +1,14 @@ import asyncio import logging from collections import defaultdict +from contextlib import AsyncExitStack from urllib.parse import urlparse from ab_sdk import points from ab_sdk.ingest import AddRequest from module.conf import settings from module.models import Bangumi, Torrent +from module.models.config import DOWNLOADER_POINT, DownloaderInstance from module.network import RequestContent from module.plugin import host as plugin_host from module.plugin.views import bangumi_info, torrent_info @@ -116,24 +118,44 @@ async def _apply_adding_hooks( # `_bookkeeping_lock` serializes reads/writes of this shared state across the # awaits in ``__aenter__``/``__aexit__``. # --------------------------------------------------------------------------- -_client_cache: tuple[tuple, DownloaderClient] | None = None +# 实例 id → (连接设置 key, 具体客户端)。每个下载器实例各缓存一个客户端。 +_client_cache: dict[str, tuple[tuple, DownloaderClient]] = {} _stale_clients: list[DownloaderClient] = [] _active_holders: dict[int, int] = {} _pending_close: set[int] = set() _bookkeeping_lock = asyncio.Lock() -# 凭据被服务端明确拒绝后的闩锁：记录失败时的连接设置 key。命中时 enter 直接 -# 失败、不再发 login POST——每个 tick 重试一次登录，约 5 次即触发 qB 的 +# 凭据被服务端明确拒绝后的闩锁：实例 id → 失败时的连接设置 key。命中时 enter +# 直接失败、不再发 login POST——每个 tick 重试一次登录，约 5 次即触发 qB 的 # WebUI IP ban。设置变更（key 不同）自然解锁；同值重存经 # clear_credential_latch()（AppContext.reload_settings）解锁。 -_credential_failed_key: tuple | None = None +_credential_failed: dict[str, tuple] = {} # Warn at most once per (client type, operation) when a backend cannot perform # an operation, so aria2 users are not spammed every rename cycle. _warned_unsupported: set[tuple[str, str]] = set() -def _settings_key() -> tuple: - d = settings.downloader +def downloader_ids() -> list[str]: + """所有下载器实例 id（配置顺序）。""" + return [i.id for i in settings.plugins.instances if i.point == DOWNLOADER_POINT] + + +def resolve_downloader_id(*candidates: str | None) -> str: + """取第一个指向已配置实例的候选 id；都为空时用默认实例（slots.downloader）。 + + 指向已删除实例的候选记录告警后跳过，不让一条规则卡住整轮。 + """ + known = downloader_ids() + for candidate in candidates: + if not candidate: + continue + if candidate in known: + return candidate + logger.warning("Downloader instance %r no longer exists; skipped", candidate) + return settings.plugins.slots.downloader + + +def _settings_key(d: DownloaderInstance) -> tuple: # 带上 Provider 登记项：插件重载后登记项换新，旧客户端随之退役 entry = plugin_host.get_registry().providers(points.DOWNLOADER).get(d.type) return (d.type, d.host, d.username, d.password, d.ssl, entry) @@ -141,8 +163,8 @@ def _settings_key() -> tuple: def _reset_client_cache() -> None: """Drop the cached/stale concrete clients and refcount state (used by tests).""" - global _client_cache, _stale_clients, _active_holders, _pending_close - _client_cache = None + global _stale_clients, _active_holders, _pending_close + _client_cache.clear() _stale_clients = [] _active_holders = {} _pending_close = set() @@ -150,9 +172,8 @@ def _reset_client_cache() -> None: def clear_credential_latch() -> None: - """解除凭据失败闩锁（配置保存后调用，允许用户重试同值凭据）。""" - global _credential_failed_key - _credential_failed_key = None + """解除所有实例的凭据失败闩锁（配置保存后调用，允许用户重试同值凭据）。""" + _credential_failed.clear() async def _close_client(client) -> None: @@ -163,15 +184,13 @@ async def _close_client(client) -> None: async def shutdown() -> None: - """Log out and close the cached concrete client. + """Log out and close every cached concrete client. Invoked by the composition root (`AppContext`) on application shutdown. """ - global _client_cache, _stale_clients - clients = list(_stale_clients) - if _client_cache is not None: - clients.append(_client_cache[1]) - _client_cache = None + global _stale_clients + clients = list(_stale_clients) + [c for _, c in _client_cache.values()] + _client_cache.clear() _stale_clients = [] for client in clients: await _close_client(client) @@ -187,30 +206,37 @@ class DownloadClient: down by :func:`shutdown`. """ - def __init__(self): - global _client_cache - key = _settings_key() + def __init__(self, instance_id: str | None = None): + # None：默认实例（slots.downloader） + self.instance: DownloaderInstance = ( + settings.downloader + if instance_id is None + else settings.downloader_instance(instance_id) + ) + self.instance_id = self.instance.id + key = _settings_key(self.instance) self.client: DownloaderClient self._cache_key = key - if _client_cache is not None and _client_cache[0] == key: - self.client = _client_cache[1] + cached = _client_cache.get(self.instance_id) + if cached is not None and cached[0] == key: + self.client = cached[1] else: - if _client_cache is not None: + if cached is not None: # Settings changed: retire the previous client, close it later. # 用列表累积——连续两次改设置（期间没有 enter）不得把第一个 # 被撤下的客户端顶掉，否则它的连接池泄漏到进程结束。 - _stale_clients.append(_client_cache[1]) - self.client = self.__getClient() - _client_cache = (key, self.client) + _stale_clients.append(cached[1]) + self.client = self.__getClient(self.instance) + _client_cache[self.instance_id] = (key, self.client) self.authed = False @staticmethod - def __getClient() -> DownloaderClient: - """按 ``downloader.type`` 从扩展注册表取下载器工厂并实例化。 + def __getClient(instance: DownloaderInstance) -> DownloaderClient: + """按实例的 Provider id 从扩展注册表取下载器工厂并实例化。 内置的 qbittorrent / aria2 / mock 与插件提供的下载器走同一条路径。 """ - downloader_type = settings.downloader.type + downloader_type = instance.type factory = plugin_host.provider(points.DOWNLOADER, downloader_type) if factory is None: logger.error("Unsupported downloader type: %s", downloader_type) @@ -218,10 +244,10 @@ class DownloadClient: if downloader_type == "mock": logger.debug("Using MockDownloader for local development") conn = DownloaderConnection( - host=settings.downloader.host, - username=settings.downloader.username, - password=settings.downloader.password, - ssl=settings.downloader.ssl, + host=instance.host, + username=instance.username, + password=instance.password, + ssl=instance.ssl, ) # 只实现 CoreDownloaderClient 的后端（如 aria2 没有 qB 的 RSS 规则） # 由 _supports() 按 capabilities 跳过不支持的操作 @@ -251,11 +277,7 @@ class DownloadClient: return False async def __aenter__(self): - global _client_cache, _credential_failed_key - if ( - _credential_failed_key is not None - and _credential_failed_key == self._cache_key - ): + if _credential_failed.get(self.instance_id) == self._cache_key: # 凭据上次已被服务端明确拒绝且设置未变：不再发 login POST， # 避免逐 tick 累积到 qB 的 IP ban。checker/等待循环读的是 # 具体客户端上的失败原因，这里补齐。 @@ -297,15 +319,16 @@ class DownloadClient: raise if not self.authed: if self.last_auth_error == "credentials": - _credential_failed_key = self._cache_key + _credential_failed[self.instance_id] = self._cache_key # Release our slot and close the concrete client's connection # pool now or it leaks on every failed connect (#1043) -- # unless another already-entered block still holds it, in # which case defer to its __aexit__ instead of yanking the # pool out from under it. async with _bookkeeping_lock: - if _client_cache is not None and _client_cache[1] is self.client: - _client_cache = None + cached = _client_cache.get(self.instance_id) + if cached is not None and cached[1] is self.client: + del _client_cache[self.instance_id] _pending_close.add(id(self.client)) await self._release_holder() raise ConnectionError("Download client authentication failed") @@ -441,7 +464,10 @@ class DownloadClient: 调用方据此决定是否重试或发送失败通知。 """ if not bangumi.save_path: - bangumi.save_path = gen_save_path(bangumi) + bangumi.save_path = gen_save_path(bangumi, self.instance.path) + # 种子行记录实际投递的实例，之后的重命名 / 删除按它路由 + for t in torrent if isinstance(torrent, list) else [torrent]: + t.downloader_id = self.instance_id torrent_url: str | list[str] | None async with RequestContent() as req: if isinstance(torrent, list): @@ -513,3 +539,67 @@ class DownloadClient: return await self.client.add_tag(torrent_hash, tag) logger.debug("Added tag '%s' to torrent %s...", tag, torrent_hash[:8]) + + +class DownloaderPool: + """一次操作内按实例 id 惰性进入 :class:`DownloadClient`，退出时全部释放。 + + 进入失败的实例在本池内记住、直接抛 ``ConnectionError``，不再重复登录 + （一轮 RSS 可能有很多种子指向同一个不可用实例）。 + """ + + def __init__(self) -> None: + self._stack = AsyncExitStack() + self._clients: dict[str, DownloadClient] = {} + self._failed: set[str] = set() + + async def __aenter__(self) -> "DownloaderPool": + return self + + async def __aexit__(self, *_: object) -> None: + await self._stack.aclose() + + async def get(self, instance_id: str) -> DownloadClient: + if instance_id in self._failed: + raise ConnectionError(f"Downloader instance {instance_id} is unavailable") + if instance_id not in self._clients: + try: + self._clients[instance_id] = await self._stack.enter_async_context( + DownloadClient(instance_id) + ) + except Exception as e: + self._failed.add(instance_id) + raise ConnectionError( + f"Downloader instance {instance_id} is unavailable: {e}" + ) from e + return self._clients[instance_id] + + +async def list_torrents( + status_filter: str | None = None, timeout: float | None = None +) -> list[dict] | None: + """所有下载器实例中 Bangumi 分类的种子，每条带 ``downloader_id``。 + + 不可用（或超过 ``timeout`` 秒）的实例跳过；全部实例都不可用时返回 None。 + """ + + async def one(instance_id: str) -> list[dict]: + async with DownloadClient(instance_id) as client: + torrents = await client.get_torrent_info( + category="Bangumi", status_filter=status_filter + ) + return [{**t, "downloader_id": instance_id} for t in torrents] + + ids = downloader_ids() + results = await asyncio.gather( + *(asyncio.wait_for(one(i), timeout) for i in ids), return_exceptions=True + ) + listed: list[dict] = [] + reachable = False + for instance_id, result in zip(ids, results): + if isinstance(result, BaseException): + logger.debug("Downloader instance %s unavailable: %s", instance_id, result) + continue + reachable = True + listed.extend(result) + return listed if reachable else None diff --git a/backend/src/module/downloader/path.py b/backend/src/module/downloader/path.py index 13dfa703..b8db69fe 100644 --- a/backend/src/module/downloader/path.py +++ b/backend/src/module/downloader/path.py @@ -70,12 +70,15 @@ def check_files(files: list[dict]): return media_list, subtitle_list -def path_to_bangumi(save_path: PathLike[str] | str, torrent_name: str = ""): +def path_to_bangumi( + save_path: PathLike[str] | str, torrent_name: str = "", root: str | None = None +): + """``root`` 为种子所在下载器实例的下载目录，缺省为默认实例的。""" # Use PureWindowsPath regardless of the host AB runs on: it accepts # both "\" and "/" separators, so a qBittorrent-on-Windows save_path # reaching a Linux AB still splits into segments correctly (#1016). save_parts = PureWindowsPath(save_path).parts - download_parts = PureWindowsPath(settings.downloader.path).parts + download_parts = PureWindowsPath(root or settings.downloader.path).parts # Get bangumi name and season bangumi_name = "" season = 1 @@ -107,9 +110,13 @@ def _media_folder(data: Bangumi | BangumiUpdate | Movie | MovieUpdate) -> str: return "Unknown Bangumi" -def gen_save_path(data: Bangumi | BangumiUpdate | Movie | MovieUpdate) -> str: +def gen_save_path( + data: Bangumi | BangumiUpdate | Movie | MovieUpdate, root: str | None = None +) -> str: """Generate save path for a bangumi. + ``root`` 为目标下载器实例的下载目录，缺省为默认实例的。 + The save path uses the adjusted season number (season + season_offset) so files are saved directly to the correct season folder. @@ -121,7 +128,7 @@ def gen_save_path(data: Bangumi | BangumiUpdate | Movie | MovieUpdate) -> str: episode_type = getattr(data, "episode_type", "episode") if isinstance(data, (Movie, MovieUpdate)) or episode_type == "movie": # 电影/剧场版：Title (Year)/Title (Year).ext，不建 Season 子目录 - return str(Path(settings.downloader.path) / folder) + return str(Path(root or settings.downloader.path) / folder) # Apply season_offset to get the adjusted season number for the folder adjusted_season = data.season + getattr(data, "season_offset", 0) # 季号下限：普通剧集最小为 1——偏移到 Season 0 会被 Plex/Jellyfin 当作 @@ -132,7 +139,9 @@ def gen_save_path(data: Bangumi | BangumiUpdate | Movie | MovieUpdate) -> str: logger.warning( f"Season offset would result in invalid season for {data.official_title}, using original season" ) - save_path = Path(settings.downloader.path) / folder / f"Season {adjusted_season}" + save_path = ( + Path(root or settings.downloader.path) / folder / f"Season {adjusted_season}" + ) return str(save_path) diff --git a/backend/src/module/manager/collector.py b/backend/src/module/manager/collector.py index 79fa14c4..4c8ac591 100644 --- a/backend/src/module/manager/collector.py +++ b/backend/src/module/manager/collector.py @@ -2,7 +2,12 @@ import logging from module.database import Database from module.database.bangumi import release_fits_bangumi -from module.downloader import AddResult, DownloadClient +from module.downloader import ( + AddResult, + DownloadClient, + DownloaderPool, + resolve_downloader_id, +) from module.models import Bangumi, ResponseModel from module.network import RequestContent from module.rss import RSSEngine @@ -129,10 +134,12 @@ async def eps_complete(): datas = await db.bangumi.not_complete() if datas: logger.info("Start collecting full season...") - async with DownloadClient() as client: - collector = SeasonCollector(client) + async with DownloaderPool() as downloaders: for data in datas: if not data.eps_collect: - await collector.collect_season(data) + client = await downloaders.get( + resolve_downloader_id(data.downloader_id) + ) + await SeasonCollector(client).collect_season(data) data.eps_collect = True await db.bangumi.update_all(datas) diff --git a/backend/src/module/manager/renamer.py b/backend/src/module/manager/renamer.py index 1dff7295..278158bc 100644 --- a/backend/src/module/manager/renamer.py +++ b/backend/src/module/manager/renamer.py @@ -21,10 +21,19 @@ from module.database.bangumi import ( match_bangumi_in_list, normalize_save_path, ) -from module.downloader import DownloadClient, RenameOutcome, RenameResult +from module.downloader import ( + DownloadClient, + RenameOutcome, + RenameResult, + downloader_ids, +) from module.downloader.path import check_files, is_ep, path_to_bangumi from module.models import EpisodeFile, Notification, SubtitleFile -from module.notification import RenameSkippedEvent, SystemEvent +from module.notification import ( + DownloaderUnavailableEvent, + RenameSkippedEvent, + SystemEvent, +) from module.parser import TitleParser from module.plugin import host as plugin_host from module.plugin.registry import ProviderEntry @@ -44,8 +53,8 @@ from .revision_saga import ( logger = logging.getLogger(__name__) -# 下载器 id：多下载器（P2.5）之前只有一个 -_DOWNLOADER_ID = "default" +# 上一轮不可用的下载器实例：只在「可用 → 不可用」时通知一次 +_unavailable: set[str] = set() # 已记录过「未登记」日志的重命名方式（每个进程只记一次，重新登记后清除） _missing_methods: set[str] = set() # 已发布过的 TorrentOrganized（种子 hash → 最终文件）。未打 ab:renamed 标签的 @@ -589,7 +598,7 @@ class Renamer(RevisionSaga): old_path=_downloader_path(save_path, old), new_path=_downloader_path(save_path, new), file_kind=kind, - downloader_id=_DOWNLOADER_ID, + downloader_id=self.client.instance_id, ) ) reason = self._skipped.pop(torrent_hash, None) @@ -624,7 +633,7 @@ class Renamer(RevisionSaga): torrent_hash=torrent_hash, bangumi_id=bangumi_id, files=files, - downloader_id=_DOWNLOADER_ID, + downloader_id=self.client.instance_id, ) ) @@ -651,6 +660,10 @@ class Renamer(RevisionSaga): # Collect all hashes for batch query hashes = [info["hash"] for info in torrents_info] torrent_records = await db.torrent.search_by_qb_hashes(hashes) + # 同一 hash 在多个实例都有记录时，以本实例的记录为准（排在后面覆盖） + torrent_records.sort( + key=lambda r: r.downloader_id == self.client.instance_id + ) hash_to_bangumi_id = { r.qb_hash: r.bangumi_id for r in torrent_records if r.bangumi_id } @@ -758,7 +771,9 @@ class Renamer(RevisionSaga): info_by_hash.setdefault(info["hash"], info) all_infos = list(info_by_hash.values()) async with Database() as db: - active_replacements = await db.rename_operation.list_active_replacements() + active_replacements = await db.rename_operation.list_active_replacements( + self._downloader_type() + ) active_replacement_ids = { operation.new_task_id for operation in active_replacements } @@ -808,7 +823,9 @@ class Renamer(RevisionSaga): ) continue media_list, subtitle_list = check_files(files) - bangumi_name, season = path_to_bangumi(save_path, torrent_name) + bangumi_name, season = path_to_bangumi( + save_path, torrent_name, self.client.instance.path + ) episode_offset, episode_type = offset_map[torrent_hash] kwargs = { "torrent_name": torrent_name, @@ -875,3 +892,38 @@ class Renamer(RevisionSaga): ) logger.debug("Rename process finished.") return renamed_info + + +async def rename_all() -> tuple[list[Notification], list[SystemEvent]]: + """逐个下载器实例运行一轮重命名。 + + 进不去的实例（连不上、凭据被拒、Provider 未登记）跳过，不影响其它实例； + 它从可用变为不可用时产生一条 :class:`DownloaderUnavailableEvent`。 + """ + renamed: list[Notification] = [] + events: list[SystemEvent] = [] + for instance_id in downloader_ids(): + client: DownloadClient | None = None + try: + client = DownloadClient(instance_id) + await client.__aenter__() + except Exception as e: + logger.warning("Downloader %s unavailable, skipped: %s", instance_id, e) + if instance_id not in _unavailable: + _unavailable.add(instance_id) + events.append( + DownloaderUnavailableEvent( + host=settings.downloader_instance(instance_id).host, + reason=(client and client.last_auth_error) or "unreachable", + instance_id=instance_id, + ) + ) + continue + _unavailable.discard(instance_id) + try: + renamer = Renamer(client) + renamed += await renamer.rename() + events += renamer.events + finally: + await client.__aexit__(None, None, None) + return renamed, events diff --git a/backend/src/module/manager/revision_saga.py b/backend/src/module/manager/revision_saga.py index 7634da60..cceb1609 100644 --- a/backend/src/module/manager/revision_saga.py +++ b/backend/src/module/manager/revision_saga.py @@ -167,13 +167,14 @@ class RevisionSaga: return value <= datetime.now(timezone.utc) def _downloader_type(self) -> str: - configured = settings.downloader.type + instance = self.client.instance + configured = instance.type downloader_type = ( configured if isinstance(configured, str) else type(self.client.client).__name__.lower() ) - host = settings.downloader.host + host = instance.host if not isinstance(host, str) or not host: return downloader_type instance_hash = hashlib.sha256(host.strip().lower().encode()).hexdigest()[:12] @@ -191,7 +192,9 @@ class RevisionSaga: incoming_id = self._parse_bangumi_id_from_tags(incoming.get("tags")) _, incoming_season = path_to_bangumi( - incoming.get("save_path", ""), incoming.get("name", "") + incoming.get("save_path", ""), + incoming.get("name", ""), + self.client.instance.path, ) incoming_identity = parse_revision_identity( incoming.get("name", ""), @@ -229,7 +232,9 @@ class RevisionSaga: ): continue _, old_season = path_to_bangumi( - info.get("save_path", ""), info.get("name", "") + info.get("save_path", ""), + info.get("name", ""), + self.client.instance.path, ) owners.append( RevisionOwner( diff --git a/backend/src/module/manager/torrent.py b/backend/src/module/manager/torrent.py index 7ed9fedf..6c4cfeaf 100644 --- a/backend/src/module/manager/torrent.py +++ b/backend/src/module/manager/torrent.py @@ -3,7 +3,7 @@ import logging from module.conf import settings from module.database import Database from module.database.bangumi import normalize_save_path -from module.downloader import DownloadClient +from module.downloader import DownloadClient, downloader_ids, resolve_downloader_id from module.downloader.path import gen_save_path from module.downloader.rules import build_rss_rule from module.models import Bangumi, BangumiUpdate, ResponseModel @@ -19,8 +19,10 @@ class TorrentManager: self.db = db @staticmethod - async def __match_torrents_list(data: Bangumi | BangumiUpdate) -> list: - async with DownloadClient() as client: + async def __match_torrents_list( + data: Bangumi | BangumiUpdate, instance_id: str | None = None + ) -> list: + async with DownloadClient(instance_id) as client: torrents = await client.get_torrent_info(status_filter=None) target_save_path = normalize_save_path(data.save_path) return [ @@ -29,16 +31,33 @@ class TorrentManager: if normalize_save_path(torrent.get("save_path")) == target_save_path ] - async def delete_torrents(self, data: Bangumi, client: DownloadClient): - hash_list = await self.__match_torrents_list(data) - if hash_list: - if not await client.delete_torrent(hash_list): + async def _torrent_instances(self, data: Bangumi) -> list[str]: + """番剧的种子所在的下载器实例：种子行记录的实例，加上规则当前的实例。 + + 规则换过下载器时，旧种子仍在原实例上，删除要逐个实例处理。 + """ + rows = await self.db.torrent.search_by_bangumi_id(data.id) + ids = {r.downloader_id for r in rows} + ids.add(resolve_downloader_id(data.downloader_id)) + return sorted(ids & set(downloader_ids())) + + async def delete_torrents(self, data: Bangumi, instance_ids: list[str]): + found = False + for instance_id in instance_ids: + hash_list = await self.__match_torrents_list(data, instance_id) + if not hash_list: + continue + found = True + async with DownloadClient(instance_id) as client: + deleted = await client.delete_torrent(hash_list) + if not deleted: return ResponseModel( status_code=500, status=False, msg_en=f"Failed to delete torrents for {data.official_title}", msg_zh=f"删除 {data.official_title} 种子失败", ) + if found: logger.info(f"Delete rule and torrents for {data.official_title}") return ResponseModel( status_code=200, @@ -83,6 +102,8 @@ class TorrentManager: async def delete_rule(self, _id: int | str, file: bool = False): data = await self.db.bangumi.search_id(int(_id)) if isinstance(data, Bangumi): + # 种子行删除前先记下种子所在的实例 + instance_ids = await self._torrent_instances(data) if file else [] # Clean up torrent records so re-adding the same anime can re-download await self.db.torrent.delete_by_bangumi_id(int(_id)) await self.db.bangumi.delete_one(int(_id)) @@ -92,16 +113,15 @@ class TorrentManager: if file: # Only the file-cleanup path needs the downloader, so an # unreachable downloader shouldn't block a DB-only delete. - async with DownloadClient() as client: - torrent_message = await self.delete_torrents(data, client) - if torrent_message.status_code == 500: - return ResponseModel( - status_code=500, - status=False, - msg_en=f"Deleted rule for {data.official_title}, " - "but deleting its torrents failed.", - msg_zh=f"已删除 {data.official_title} 规则，但删除种子失败。", - ) + torrent_message = await self.delete_torrents(data, instance_ids) + if torrent_message.status_code == 500: + return ResponseModel( + status_code=500, + status=False, + msg_en=f"Deleted rule for {data.official_title}, " + "but deleting its torrents failed.", + msg_zh=f"已删除 {data.official_title} 规则，但删除种子失败。", + ) logger.info(f"Delete rule for {data.official_title}") return ResponseModel( status_code=200, @@ -125,8 +145,9 @@ class TorrentManager: if file: # Only the file-cleanup path needs the downloader, so an # unreachable downloader shouldn't block a DB-only disable. - async with DownloadClient() as client: - return await self.delete_torrents(data, client) + return await self.delete_torrents( + data, await self._torrent_instances(data) + ) logger.info(f"Disable rule for {data.official_title}") return ResponseModel( status_code=200, @@ -165,25 +186,29 @@ class TorrentManager: async def update_rule(self, bangumi_id, data: BangumiUpdate): old_data = await self.db.bangumi.search_id(bangumi_id) if old_data: - # Move torrent - match_list = await self.__match_torrents_list(old_data) - async with DownloadClient() as client: - new_path = gen_save_path(data) - old_path = old_data.save_path + old_instance = resolve_downloader_id(old_data.downloader_id) + new_instance = resolve_downloader_id(data.downloader_id) + new_path = gen_save_path( + data, settings.downloader_instance(new_instance).path + ) + old_path = old_data.save_path + # 换了下载器实例：旧种子留在原实例原位置，之后的新种子进新实例 + if new_instance == old_instance and new_path != old_path: + match_list = await self.__match_torrents_list(old_data, old_instance) + async with DownloadClient(old_instance) as client: + # Move existing torrents to new location if path changed + if match_list: + await client.move_torrent(match_list, new_path) + logger.info(f"Moved torrents from {old_path} to {new_path}") - # Move existing torrents to new location if path changed - if match_list and new_path != old_path: - await client.move_torrent(match_list, new_path) - logger.info(f"Moved torrents from {old_path} to {new_path}") - - # Update qBittorrent RSS rule if save_path changed - if new_path != old_path and old_data.rule_name: - # Recreate the rule with the new save_path - rule = build_rss_rule(data, new_path) - await client.set_rss_rule(old_data.rule_name, rule) - logger.info( - f"Updated RSS rule {old_data.rule_name} with new save_path" - ) + # Update qBittorrent RSS rule if save_path changed + if old_data.rule_name: + # Recreate the rule with the new save_path + rule = build_rss_rule(data, new_path) + await client.set_rss_rule(old_data.rule_name, rule) + logger.info( + f"Updated RSS rule {old_data.rule_name} with new save_path" + ) data.save_path = new_path await self.db.bangumi.update(data, bangumi_id) diff --git a/backend/src/module/mcp/tools.py b/backend/src/module/mcp/tools.py index 7ab891ff..551f8e20 100644 --- a/backend/src/module/mcp/tools.py +++ b/backend/src/module/mcp/tools.py @@ -6,7 +6,7 @@ from mcp import types from module.conf import VERSION from module.database import Database -from module.downloader import DownloadClient +from module.downloader import DownloaderPool, list_torrents from module.manager import SeasonCollector, TorrentManager from module.models import Bangumi, BangumiUpdate, RSSItem from module.rss import RSSAnalyser, RSSEngine @@ -301,10 +301,7 @@ async def _unsubscribe_anime(bangumi_id: int, delete: bool) -> dict: async def _list_downloads(status: str) -> list[dict]: status_filter = None if status == "all" else status - async with DownloadClient() as client: - torrents = await client.get_torrent_info( - status_filter=status_filter, category="Bangumi" - ) + torrents = await list_torrents(status_filter) or [] return [ { "name": t.get("name", ""), @@ -314,6 +311,7 @@ async def _list_downloads(status: str) -> list[dict]: "dlspeed": t.get("dlspeed", 0), "upspeed": t.get("upspeed", 0), "eta": t.get("eta", 0), + "downloader": t["downloader_id"], } for t in torrents ] @@ -348,10 +346,10 @@ def _get_program_status() -> dict: async def _refresh_feeds() -> dict: - async with DownloadClient() as client: + async with DownloaderPool() as downloaders: async with Database() as db: engine = RSSEngine(db) - await engine.refresh_rss(client) + await engine.refresh_rss(downloaders) return {"status": True, "message": "RSS feeds refreshed successfully"} diff --git a/backend/src/module/models/bangumi.py b/backend/src/module/models/bangumi.py index 33c48815..37dc72a7 100644 --- a/backend/src/module/models/bangumi.py +++ b/backend/src/module/models/bangumi.py @@ -61,8 +61,9 @@ class Bangumi(SQLModel, table=True): episode_type: str = Field( default="episode", alias="episode_type", title="剧集类型" ) # "episode" | "movie" | "special" - downloader_id: str = Field( - default="default", alias="downloader_id", title="下载器实例" + # 为空时用默认下载器实例（plugins.slots.downloader） + downloader_id: Optional[str] = Field( + default=None, alias="downloader_id", title="下载器实例" ) @@ -87,6 +88,9 @@ class BangumiUpdate(SQLModel): added: bool = Field(default=False, alias="added", title="是否已添加") rule_name: Optional[str] = Field(alias="rule_name", title="番剧规则名") save_path: Optional[str] = Field(alias="save_path", title="番剧保存路径") + downloader_id: Optional[str] = Field( + default=None, alias="downloader_id", title="下载器实例" + ) deleted: bool = Field(False, alias="deleted", title="是否已删除") archived: bool = Field(default=False, alias="archived", title="是否已归档") air_weekday: Optional[int] = Field( diff --git a/backend/src/module/models/config.py b/backend/src/module/models/config.py index cc658012..2ea89877 100644 --- a/backend/src/module/models/config.py +++ b/backend/src/module/models/config.py @@ -466,10 +466,15 @@ class Config(BaseModel): def downloader_instance(self, instance_id: str) -> DownloaderInstance: instance = next( - i - for i in self.plugins.instances - if i.id == instance_id and i.point == DOWNLOADER_POINT + ( + i + for i in self.plugins.instances + if i.id == instance_id and i.point == DOWNLOADER_POINT + ), + None, ) + if instance is None: + raise KeyError(f"unknown downloader instance {instance_id!r}") options = DownloaderOptions.model_validate(instance.options) return DownloaderInstance( id=instance.id, diff --git a/backend/src/module/models/movie.py b/backend/src/module/models/movie.py index 2eef8bef..5161e3ea 100644 --- a/backend/src/module/models/movie.py +++ b/backend/src/module/models/movie.py @@ -25,8 +25,9 @@ class Movie(SQLModel, table=True): save_path: str | None = Field(default=None, alias="save_path", title="保存路径") rule_name: str | None = Field(default=None, alias="rule_name", title="规则名") filter: str = Field(default="", alias="filter", title="过滤器") - downloader_id: str = Field( - default="default", alias="downloader_id", title="下载器实例" + # 为空时用默认下载器实例 + downloader_id: str | None = Field( + default=None, alias="downloader_id", title="下载器实例" ) @@ -42,3 +43,4 @@ class MovieUpdate(SQLModel): rss_link: str | None = None save_path: str | None = None filter: str | None = None + downloader_id: str | None = None diff --git a/backend/src/module/models/rss.py b/backend/src/module/models/rss.py index 739f25ef..ab4a4f24 100644 --- a/backend/src/module/models/rss.py +++ b/backend/src/module/models/rss.py @@ -13,7 +13,8 @@ class RSSItem(SQLModel, table=True): connection_status: Optional[str] = Field(None, alias="connection_status") last_checked_at: Optional[str] = Field(None, alias="last_checked_at") last_error: Optional[str] = Field(None, alias="last_error") - downloader_id: str = Field("default", alias="downloader_id") + # 由此订阅新建的规则继承；为空时用默认下载器实例 + downloader_id: Optional[str] = Field(None, alias="downloader_id") class RSSUpdate(SQLModel): @@ -22,3 +23,4 @@ class RSSUpdate(SQLModel): aggregate: Optional[bool] = Field(True, alias="aggregate") parser: Optional[str] = Field("mikan", alias="parser") enabled: Optional[bool] = Field(True, alias="enabled") + downloader_id: Optional[str] = Field(None, alias="downloader_id") diff --git a/backend/src/module/rss/analyser.py b/backend/src/module/rss/analyser.py index ed4c9ae2..516805e6 100644 --- a/backend/src/module/rss/analyser.py +++ b/backend/src/module/rss/analyser.py @@ -70,6 +70,7 @@ class RSSAnalyser: movie=result, rss=rss, torrent=torrent ) result.rss_link = rss.url + result.downloader_id = rss.downloader_id seen_identities.add(identity) new_movies.append(result) logger.info(f"New movie found: {result.official_title}") @@ -78,6 +79,7 @@ class RSSAnalyser: bangumi=result, rss=rss, torrent=torrent ) result.rss_link = rss.url + result.downloader_id = rss.downloader_id if not full_parse: return [result], new_movies seen_identities.add(identity) @@ -98,6 +100,7 @@ class RSSAnalyser: fetch_poster=fetch_poster, ) result.rss_link = rss.url + result.downloader_id = rss.downloader_id else: await self.official_title_parser( bangumi=result, @@ -106,6 +109,7 @@ class RSSAnalyser: fetch_poster=fetch_poster, ) result.rss_link = rss.url + result.downloader_id = rss.downloader_id return result return None diff --git a/backend/src/module/rss/engine.py b/backend/src/module/rss/engine.py index 4b10963a..b9967228 100644 --- a/backend/src/module/rss/engine.py +++ b/backend/src/module/rss/engine.py @@ -14,7 +14,12 @@ from module.database.bangumi import ( match_bangumi_in_list, release_fits_bangumi, ) -from module.downloader import AddResult, DownloadClient +from module.downloader import ( + AddResult, + DownloadClient, + DownloaderPool, + resolve_downloader_id, +) from module.models import Bangumi, ResponseModel, RSSItem, Torrent from module.network import RequestContent from module.notification.events import ( @@ -323,14 +328,17 @@ class RSSEngine: return skip_ids async def refresh_rss( - self, client: DownloadClient, rss_id: Optional[int] = None + self, downloaders: DownloaderPool, rss_id: Optional[int] = None ) -> list[SystemEvent]: - """Refresh feeds with one parser engine for the complete workflow.""" + """Refresh feeds with one parser engine for the complete workflow. + + 匹配到的种子投递到规则的下载器实例，规则未指定时用订阅的，再退回默认实例。 + """ with parser_engine_snapshot(): - return await self._refresh_rss(client, rss_id) + return await self._refresh_rss(downloaders, rss_id) async def _refresh_rss( - self, client: DownloadClient, rss_id: Optional[int] = None + self, downloaders: DownloaderPool, rss_id: Optional[int] = None ) -> list[SystemEvent]: # Get All RSS Items if not rss_id: @@ -438,6 +446,17 @@ class RSSEngine: matched_data.official_title, ) continue + try: + client = await downloaders.get( + resolve_downloader_id( + matched_data.downloader_id, rss_item.downloader_id + ) + ) + except ConnectionError as e: + # 实例不可用：不入库，下一轮重试；不可用由重命名轮次通知 + logger.warning("Skip %s: %s", torrent.name, e) + failed_ids.add(id(torrent)) + continue result = await client.add_torrent(torrent, matched_data) if result is AddResult.FAILED: # 投递失败：不入库（check_new 按 URL 去重，入库就 @@ -480,7 +499,9 @@ class RSSEngine: t for t in torrents if await self.plugin_accepts(t, bangumi) ] if torrents: - async with DownloadClient() as client: + async with DownloadClient( + resolve_downloader_id(bangumi.downloader_id) + ) as client: result = await client.add_torrent(torrents, bangumi) if result is AddResult.FAILED: return ResponseModel( diff --git a/backend/src/test/factories.py b/backend/src/test/factories.py index b8d2bc43..f597d785 100644 --- a/backend/src/test/factories.py +++ b/backend/src/test/factories.py @@ -2,6 +2,7 @@ from datetime import datetime, timezone +from module.downloader import DownloaderPool from module.models import Bangumi, RSSItem, Torrent from module.models.config import Config from module.models.passkey import Passkey @@ -85,3 +86,14 @@ def make_passkey(**overrides) -> Passkey: ) defaults.update(overrides) return Passkey(**defaults) + + +class SingleClientPool(DownloaderPool): + """所有实例都返回同一个（模拟）客户端的下载器池。""" + + def __init__(self, client) -> None: + super().__init__() + self.client = client + + async def get(self, instance_id: str): + return self.client diff --git a/backend/src/test/test_api_bangumi_extended.py b/backend/src/test/test_api_bangumi_extended.py index 9b1d302c..c5efaec2 100644 --- a/backend/src/test/test_api_bangumi_extended.py +++ b/backend/src/test/test_api_bangumi_extended.py @@ -297,79 +297,64 @@ class TestApplyOffset: """POST /bangumi/apply-offset/{id} applies offset and reruns the renamer.""" mock_db = MagicMock() mock_db.bangumi.apply_offset = AsyncMock(return_value=True) - mock_renamer = AsyncMock() + mock_rename_all = AsyncMock(return_value=([], [])) with patch("module.api.bangumi.Database", _async_db_mock(mock_db)): - with patch("module.api.bangumi.Renamer", return_value=mock_renamer): - with patch("module.api.bangumi.DownloadClient") as MockClient: - MockClient.return_value.__aenter__ = AsyncMock( - return_value=MagicMock() - ) - MockClient.return_value.__aexit__ = AsyncMock(return_value=False) - response = authed_client.post("/api/v1/bangumi/apply-offset/1") + with patch("module.api.bangumi.rename_all", mock_rename_all): + response = authed_client.post("/api/v1/bangumi/apply-offset/1") assert response.status_code == 200 data = response.json() assert data["status"] is True mock_db.bangumi.apply_offset.assert_awaited_once_with(1) - mock_renamer.rename.assert_awaited_once() + mock_rename_all.assert_awaited_once() def test_apply_offset_not_found_skips_rename(self, authed_client): """POST /bangumi/apply-offset/{id} with a non-existent bangumi does not rename.""" mock_db = MagicMock() mock_db.bangumi.apply_offset = AsyncMock(return_value=False) with patch("module.api.bangumi.Database", _async_db_mock(mock_db)): - with patch("module.api.bangumi.Renamer") as MockRenamer: + with patch("module.api.bangumi.rename_all") as mock_rename_all: response = authed_client.post("/api/v1/bangumi/apply-offset/999") assert response.status_code == 404 - MockRenamer.assert_not_called() + mock_rename_all.assert_not_called() def test_apply_offset_many_success(self, authed_client): """POST /bangumi/apply-offset/many applies offsets to every id and renames once.""" mock_db = MagicMock() mock_db.bangumi.apply_offset = AsyncMock(side_effect=[True, True]) - mock_renamer = AsyncMock() + mock_rename_all = AsyncMock(return_value=([], [])) with patch("module.api.bangumi.Database", _async_db_mock(mock_db)): - with patch("module.api.bangumi.Renamer", return_value=mock_renamer): - with patch("module.api.bangumi.DownloadClient") as MockClient: - MockClient.return_value.__aenter__ = AsyncMock( - return_value=MagicMock() - ) - MockClient.return_value.__aexit__ = AsyncMock(return_value=False) - response = authed_client.post( - "/api/v1/bangumi/apply-offset/many", json=[1, 2] - ) + with patch("module.api.bangumi.rename_all", mock_rename_all): + response = authed_client.post( + "/api/v1/bangumi/apply-offset/many", json=[1, 2] + ) assert response.status_code == 200 assert mock_db.bangumi.apply_offset.await_count == 2 - mock_renamer.rename.assert_awaited_once() + mock_rename_all.assert_awaited_once() def test_apply_offset_many_partial_failure(self, authed_client): """POST /bangumi/apply-offset/many reports a partial failure and still renames.""" mock_db = MagicMock() mock_db.bangumi.apply_offset = AsyncMock(side_effect=[True, False]) - mock_renamer = AsyncMock() + mock_rename_all = AsyncMock(return_value=([], [])) with patch("module.api.bangumi.Database", _async_db_mock(mock_db)): - with patch("module.api.bangumi.Renamer", return_value=mock_renamer): - with patch("module.api.bangumi.DownloadClient") as MockClient: - MockClient.return_value.__aenter__ = AsyncMock( - return_value=MagicMock() - ) - MockClient.return_value.__aexit__ = AsyncMock(return_value=False) - response = authed_client.post( - "/api/v1/bangumi/apply-offset/many", json=[1, 2] - ) + with patch("module.api.bangumi.rename_all", mock_rename_all): + response = authed_client.post( + "/api/v1/bangumi/apply-offset/many", json=[1, 2] + ) assert response.status_code == 500 - mock_renamer.rename.assert_awaited_once() + mock_rename_all.assert_awaited_once() def test_apply_offset_many_empty_list(self, authed_client): """POST /bangumi/apply-offset/many with an empty list is a 400, no DB/rename calls.""" - with patch("module.api.bangumi.Renamer") as MockRenamer: + with patch("module.api.bangumi.rename_all") as mock_rename_all: response = authed_client.post("/api/v1/bangumi/apply-offset/many", json=[]) assert response.status_code == 400 - MockRenamer.assert_not_called() + mock_rename_all.assert_not_called() def test_apply_offset_auth_required(self, unauthed_client): """POST /bangumi/apply-offset/{id} requires authentication.""" diff --git a/backend/src/test/test_api_downloader.py b/backend/src/test/test_api_downloader.py index 9fddbb66..7d444844 100644 --- a/backend/src/test/test_api_downloader.py +++ b/backend/src/test/test_api_downloader.py @@ -182,7 +182,7 @@ class TestOtherAuthRequired: class TestGetTorrents: def test_get_torrents_success(self, authed_client, mock_download_client): """GET /downloader/torrents returns list of torrents.""" - with patch("module.api.downloader.DownloadClient") as MockClient: + with patch("module.downloader.download_client.DownloadClient") as MockClient: MockClient.return_value.__aenter__ = AsyncMock( return_value=mock_download_client ) @@ -194,11 +194,12 @@ class TestGetTorrents: data = response.json() assert len(data) == 2 assert data[0]["hash"] == "abc123" + assert data[0]["downloader_id"] == "default" def test_get_torrents_empty(self, authed_client, mock_download_client): """GET /downloader/torrents returns empty list when no torrents.""" mock_download_client.get_torrent_info.return_value = [] - with patch("module.api.downloader.DownloadClient") as MockClient: + with patch("module.downloader.download_client.DownloadClient") as MockClient: MockClient.return_value.__aenter__ = AsyncMock( return_value=mock_download_client ) diff --git a/backend/src/test/test_api_events.py b/backend/src/test/test_api_events.py index 36acdbf7..f5ebe5c5 100644 --- a/backend/src/test/test_api_events.py +++ b/backend/src/test/test_api_events.py @@ -84,10 +84,12 @@ class TestDownloaderPayload: mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) - with patch("module.api.events.DownloadClient", return_value=mock_client): + with patch( + "module.downloader.download_client.DownloadClient", return_value=mock_client + ): result = await _downloader_payload() - assert result == torrents + assert result == [{**torrents[0], "downloader_id": "default"}] mock_client.get_torrent_info.assert_called_once_with( category="Bangumi", status_filter=None ) @@ -99,7 +101,9 @@ class TestDownloaderPayload: mock_client.__aenter__ = AsyncMock(side_effect=RuntimeError("no downloader")) mock_client.__aexit__ = AsyncMock(return_value=False) - with patch("module.api.events.DownloadClient", return_value=mock_client): + with patch( + "module.downloader.download_client.DownloadClient", return_value=mock_client + ): result = await _downloader_payload() assert result is None @@ -132,7 +136,10 @@ class TestDownloaderPayloadTimeout: mock_client = _make_hung_client(cancelled) with ( - patch("module.api.events.DownloadClient", return_value=mock_client), + patch( + "module.downloader.download_client.DownloadClient", + return_value=mock_client, + ), patch("module.api.events._DOWNLOADER_TIMEOUT_SECONDS", 0.05), ): start = time.monotonic() @@ -163,7 +170,10 @@ class TestEventGeneratorNonBlocking: mock_client = _make_hung_client(cancelled) with ( - patch("module.api.events.DownloadClient", return_value=mock_client), + patch( + "module.downloader.download_client.DownloadClient", + return_value=mock_client, + ), patch("module.api.events._DOWNLOADER_TIMEOUT_SECONDS", 0.05), patch("module.api.events.LOG_PATH", tmp_path / "missing.log"), ): diff --git a/backend/src/test/test_integration.py b/backend/src/test/test_integration.py index 7a54eb19..1af60381 100644 --- a/backend/src/test/test_integration.py +++ b/backend/src/test/test_integration.py @@ -6,7 +6,7 @@ from module.database import Database from module.downloader import AddResult from module.models import EpisodeFile, Torrent from module.rss.engine import RSSEngine -from test.factories import make_bangumi, make_rss_item +from test.factories import SingleClientPool, make_bangumi, make_rss_item # --------------------------------------------------------------------------- # RSS → Download Flow @@ -58,7 +58,7 @@ class TestRssToDownloadFlow: mock_client.add_torrent = AsyncMock(return_value=AddResult.ADDED) # 5. Execute refresh_rss - await engine.refresh_rss(mock_client) + await engine.refresh_rss(SingleClientPool(mock_client)) # 6. Verify: matched torrents were downloaded assert mock_client.add_torrent.call_count == 2 @@ -109,7 +109,7 @@ class TestRssToDownloadFlow: mock_get.return_value = torrents mock_client = AsyncMock() mock_client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await engine.refresh_rss(mock_client) + await engine.refresh_rss(SingleClientPool(mock_client)) # Only 1080p should be downloaded (720p is filtered) assert mock_client.add_torrent.call_count == 1 @@ -149,7 +149,7 @@ class TestRssToDownloadFlow: mock_get.return_value = torrents mock_client = AsyncMock() mock_client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await engine.refresh_rss(mock_client) + await engine.refresh_rss(SingleClientPool(mock_client)) # Only ep02 should be downloaded (ep01 already exists) assert mock_client.add_torrent.call_count == 1 diff --git a/backend/src/test/test_loops.py b/backend/src/test/test_loops.py index 0473e51e..54cd1604 100644 --- a/backend/src/test/test_loops.py +++ b/backend/src/test/test_loops.py @@ -99,7 +99,7 @@ class TestRssTick: with ( patch( - "module.core.loops.DownloadClient", + "module.core.loops.DownloaderPool", return_value=_async_cm(AsyncMock()), ), patch("module.core.loops.Database", return_value=_async_cm(mock_db)), @@ -125,7 +125,7 @@ class TestRssTick: with ( patch( - "module.core.loops.DownloadClient", + "module.core.loops.DownloaderPool", return_value=_async_cm(AsyncMock()), ), patch("module.core.loops.Database", return_value=_async_cm(mock_db)), @@ -144,6 +144,15 @@ class TestRssTick: # --------------------------------------------------------------------------- +def _rename_all(renamer): + """rename_all 的替身：返回模拟 Renamer 的结果与事件。""" + + async def run(): + return await renamer.rename(), list(renamer.events) + + return run + + class TestRenameTick: async def test_persists_and_dispatches_conflict_event_when_push_disabled(self): event = RenameConflictEvent( @@ -158,11 +167,7 @@ class TestRenameTick: mock_renamer.events = [event] with ( - patch( - "module.core.loops.DownloadClient", - return_value=_async_cm(AsyncMock()), - ), - patch("module.core.loops.Renamer", return_value=mock_renamer), + patch("module.core.loops.rename_all", _rename_all(mock_renamer)), patch("module.core.loops.settings") as mock_settings, ): mock_settings.notification.enable = False @@ -180,11 +185,7 @@ class TestRenameTick: mock_renamer.rename = AsyncMock(return_value=[notify]) with ( - patch( - "module.core.loops.DownloadClient", - return_value=_async_cm(AsyncMock()), - ), - patch("module.core.loops.Renamer", return_value=mock_renamer), + patch("module.core.loops.rename_all", _rename_all(mock_renamer)), patch("module.core.loops.settings") as mock_settings, ): mock_settings.notification.enable = True @@ -200,11 +201,7 @@ class TestRenameTick: mock_renamer.rename = AsyncMock(return_value=[notify]) with ( - patch( - "module.core.loops.DownloadClient", - return_value=_async_cm(AsyncMock()), - ), - patch("module.core.loops.Renamer", return_value=mock_renamer), + patch("module.core.loops.rename_all", _rename_all(mock_renamer)), patch("module.core.loops.settings") as mock_settings, ): mock_settings.notification.enable = False @@ -232,11 +229,7 @@ class TestRenameTick: mock_renamer.rename = AsyncMock(return_value=notifications) with ( - patch( - "module.core.loops.DownloadClient", - return_value=_async_cm(AsyncMock()), - ), - patch("module.core.loops.Renamer", return_value=mock_renamer), + patch("module.core.loops.rename_all", _rename_all(mock_renamer)), patch("module.core.loops.settings") as mock_settings, ): mock_settings.notification.enable = True diff --git a/backend/src/test/test_mcp_tools.py b/backend/src/test/test_mcp_tools.py index 5d683cfd..cc53a869 100644 --- a/backend/src/test/test_mcp_tools.py +++ b/backend/src/test/test_mcp_tools.py @@ -401,7 +401,9 @@ class TestDispatch: mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) - with patch("module.mcp.tools.DownloadClient", return_value=mock_client): + with patch( + "module.downloader.download_client.DownloadClient", return_value=mock_client + ): result = await _dispatch("list_downloads", {"status": "all"}) mock_client.get_torrent_info.assert_called_once_with( @@ -417,7 +419,9 @@ class TestDispatch: mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) - with patch("module.mcp.tools.DownloadClient", return_value=mock_client): + with patch( + "module.downloader.download_client.DownloadClient", return_value=mock_client + ): await _dispatch("list_downloads", {"status": "downloading"}) mock_client.get_torrent_info.assert_called_once_with( @@ -443,7 +447,9 @@ class TestDispatch: mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=False) - with patch("module.mcp.tools.DownloadClient", return_value=mock_client): + with patch( + "module.downloader.download_client.DownloadClient", return_value=mock_client + ): result = await _dispatch("list_downloads", {}) expected_keys = { @@ -454,6 +460,7 @@ class TestDispatch: "dlspeed", "upspeed", "eta", + "downloader", } assert set(result[0].keys()) == expected_keys @@ -516,7 +523,7 @@ class TestDispatch: mock_engine.refresh_rss = AsyncMock(return_value=None) with ( - patch("module.mcp.tools.DownloadClient", return_value=mock_client), + patch("module.mcp.tools.DownloaderPool", return_value=mock_client), patch("module.mcp.tools.RSSEngine", return_value=mock_engine), ): result = await _dispatch("refresh_feeds", {}) diff --git a/backend/src/test/test_migrations_module.py b/backend/src/test/test_migrations_module.py index acd4bc74..e2d3d0ee 100644 --- a/backend/src/test/test_migrations_module.py +++ b/backend/src/test/test_migrations_module.py @@ -705,8 +705,10 @@ class TestRunMigrations: assert prefix == "legacy_cccccccc" @pytest.mark.parametrize("preexisting", [None, "torrent"]) - def test_v26_downloader_id_existing_rows_default(self, preexisting, monkeypatch): - """v26：存量番剧 / 电影 / 订阅 / 种子都归属默认下载器实例； + def test_v26_downloader_id_existing_rows_follow_default( + self, preexisting, monkeypatch + ): + """v26：存量种子归属实例 default；番剧 / 电影 / 订阅为空（跟随默认实例）。 已有该列的表（如被 create_all 提前建出）跳过，其它表照常补列。""" engine = _make_v0_engine() with engine.begin() as conn: @@ -728,12 +730,13 @@ class TestRunMigrations: run_migrations(engine) + expected = {"bangumi": None, "movie": None, "rssitem": None} with engine.connect() as conn: for table in ("bangumi", "movie", "rssitem", "torrent"): values = conn.execute( text(f"SELECT downloader_id FROM {table}") ).scalars() - assert list(values) == ["default"], table + assert list(values) == [expected.get(table, "default")], table def test_is_idempotent(self): engine = _make_v0_engine() diff --git a/backend/src/test/test_multi_downloader.py b/backend/src/test/test_multi_downloader.py new file mode 100644 index 00000000..a5d8da59 --- /dev/null +++ b/backend/src/test/test_multi_downloader.py @@ -0,0 +1,330 @@ +"""多下载器实例（P2.5）：按实例缓存客户端、按规则 / 订阅路由新种子、 +重命名逐实例运行、不可用实例跳过并通知一次。两个实例都用 mock 下载器。""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from ab_sdk.events import DownloaderUnavailableEvent, TorrentOrganized +from module.conf import settings +from module.database import Database +from module.downloader import ( + DownloadClient, + DownloaderPool, + list_torrents, + resolve_downloader_id, +) +from module.downloader.client.mock_downloader import MockDownloader +from module.manager import renamer as renamer_module +from module.manager.renamer import Renamer, rename_all +from module.models import Torrent +from module.models.config import PluginInstance, Plugins, Slots +from module.rss.engine import RSSEngine +from test.factories import make_bangumi, make_rss_item + + +def _instance(instance_id: str, path: str) -> PluginInstance: + return PluginInstance( + id=instance_id, + point="downloader", + provider="mock", + options={"host": f"{instance_id}:8080", "path": path}, + ) + + +@pytest.fixture +def two_instances(monkeypatch): + """实例 a（默认）与 b，各自的下载目录不同。""" + plugins = Plugins( + instances=[_instance("a", "/a/Bangumi"), _instance("b", "/b/Bangumi")], + slots=Slots(downloader="a"), + ) + monkeypatch.setattr(settings, "plugins", plugins) + monkeypatch.setattr(renamer_module, "_unavailable", set()) + return plugins + + +def _mock(instance_id: str) -> MockDownloader: + client = DownloadClient(instance_id).client + assert isinstance(client, MockDownloader) + return client + + +def _completed(name: str, torrent_hash: str, save_path: str) -> dict: + return { + "hash": torrent_hash, + "name": name, + "save_path": save_path, + "category": "Bangumi", + "state": "completed", + "progress": 1.0, + "files": [{"name": f"{name}.mkv", "size": 100}], + "tags": "", + } + + +class TestClientCache: + def test_download_client_two_instances_separate_cached_clients(self, two_instances): + a, b = _mock("a"), _mock("b") + assert a is not b + assert _mock("a") is a + assert DownloadClient().instance_id == "a" + + async def test_download_client_credentials_rejected_latch_is_per_instance( + self, two_instances + ): + b = _mock("b") + b.auth = AsyncMock(return_value=False) # type: ignore[method-assign] + b.last_auth_error = "credentials" # type: ignore[attr-defined] + with pytest.raises(ConnectionError): + async with DownloadClient("b"): + pass + with pytest.raises(ConnectionError): + async with DownloadClient("b"): + pass + assert b.auth.await_count == 1 # 闩锁命中，不再登录 + async with DownloadClient("a") as client: + assert client.authed + + +@pytest.mark.parametrize( + "candidates, expected", + [ + ((None, None), "a"), + (("", None), "a"), + (("b", None), "b"), + ((None, "b"), "b"), + (("a", "b"), "a"), + (("gone", "b"), "b"), + (("gone",), "a"), + ], +) +def test_resolve_downloader_id_candidates_first_known_else_default( + two_instances, candidates, expected +): + assert resolve_downloader_id(*candidates) == expected + + +class TestRssRouting: + @pytest.mark.parametrize( + "bangumi_instance, rss_instance, expected", + [("b", None, "b"), (None, "b", "b"), (None, None, "a"), ("a", "b", "a")], + ) + async def test_refresh_rss_adds_to_rule_then_feed_then_default_instance( + self, two_instances, db_engine, bangumi_instance, rss_instance, expected + ): + engine = RSSEngine(Database(engine=db_engine)) + await engine.db.rss.add(make_rss_item(enabled=True, downloader_id=rss_instance)) + await engine.db.bangumi.add( + make_bangumi( + title_raw="Mushoku Tensei", + filter="", + save_path=None, + downloader_id=bangumi_instance, + ) + ) + torrent = Torrent( + name="[Sub] Mushoku Tensei - 12 [1080p].mkv", + url="magnet:?xt=urn:btih:12", + ) + with patch.object( + RSSEngine, "_get_torrents", AsyncMock(return_value=[torrent]) + ): + async with DownloaderPool() as pool: + await engine.refresh_rss(pool) + + rows = await engine.db.torrent.search_all() + assert [(r.downloaded, r.downloader_id) for r in rows] == [(True, expected)] + other = "a" if expected == "b" else "b" + assert len(_mock(expected)._torrents) == 1 + assert _mock(other)._torrents == {} + # 新规则的保存目录取目标实例的下载目录 + (added,) = _mock(expected)._torrents.values() + assert added["save_path"].startswith(f"/{expected}/Bangumi/") + + async def test_refresh_rss_unavailable_instance_torrent_not_persisted( + self, two_instances, db_engine + ): + engine = RSSEngine(Database(engine=db_engine)) + await engine.db.rss.add(make_rss_item(enabled=True)) + await engine.db.bangumi.add( + make_bangumi(title_raw="Mushoku Tensei", filter="", downloader_id="b") + ) + _mock("b").auth = AsyncMock(return_value=False) # type: ignore[method-assign] + torrent = Torrent( + name="[Sub] Mushoku Tensei - 12 [1080p].mkv", + url="magnet:?xt=urn:btih:12", + ) + with patch.object( + RSSEngine, "_get_torrents", AsyncMock(return_value=[torrent]) + ): + async with DownloaderPool() as pool: + events = await engine.refresh_rss(pool) + + assert events == [] + assert await engine.db.torrent.search_all() == [] # 下一轮重试 + + +class TestRenameAll: + async def test_rename_all_events_carry_instance_of_torrent(self, two_instances): + _mock("b")._torrents["hb"] = _completed( + "[Sub] Show - 01 [1080p]", "hb", "/b/Bangumi/Show/Season 1" + ) + published: list[object] = [] + with patch.object(renamer_module.plugin_host, "publish", published.append): + renamed, events = await rename_all() + + organized = [e for e in published if isinstance(e, TorrentOrganized)] + assert [(e.torrent_hash, e.downloader_id) for e in organized] == [("hb", "b")] + assert events == [] + assert [n.official_title for n in renamed] == ["Show"] + + async def test_rename_all_unavailable_instance_skipped_and_notified_once( + self, two_instances + ): + _mock("a")._torrents["ha"] = _completed( + "[Sub] Show - 01 [1080p]", "ha", "/a/Bangumi/Show/Season 1" + ) + _mock("b").auth = AsyncMock(return_value=False) # type: ignore[method-assign] + + first_renamed, first_events = await rename_all() + _, second_events = await rename_all() + + assert [n.official_title for n in first_renamed] == ["Show"] + assert first_events == [ + DownloaderUnavailableEvent( + host="b:8080", reason="unreachable", instance_id="b" + ) + ] + assert second_events == [] + + async def test_rename_all_recovered_instance_notified_again_on_next_outage( + self, two_instances + ): + counts = [] + for reachable in (False, True, False): + # 登录失败后缓存的客户端被丢弃，每轮给新取到的客户端打桩 + _mock("b").auth = AsyncMock(return_value=reachable) # type: ignore[method-assign] + counts.append(len((await rename_all())[1])) + assert counts == [1, 0, 1] + + async def test_rename_replacement_of_other_instance_not_recovered( + self, two_instances + ): + """实例 a 的版本替换事务不能被实例 b 的重命名当成「新种子已消失」去恢复。""" + from module.models.rename_operation import RenameOperation + + async with DownloadClient("a") as client_a: + a_type = Renamer(client_a)._downloader_type() + async with Database() as db: + db.add( + RenameOperation( + kind="replacement", + state="planned", + downloader_type=a_type, + new_task_id="only-on-a", + save_path="/a/Bangumi/Show/Season 1", + target_path="Show S01E01.mkv", + source_path="x.mkv", + ) + ) + await db.commit() + + with patch.object( + Renamer, "_recover_missing_replacement", AsyncMock() + ) as recover: + async with DownloadClient("b") as client_b: + await Renamer(client_b).rename() + recover.assert_not_awaited() + + +async def test_list_torrents_one_instance_down_lists_the_other(two_instances): + _mock("a")._torrents["ha"] = _completed("A", "ha", "/a/Bangumi/A/Season 1") + _mock("b").auth = AsyncMock(return_value=False) # type: ignore[method-assign] + listed = await list_torrents() + assert [(t["hash"], t["downloader_id"]) for t in listed or []] == [("ha", "a")] + + +class TestRuleOnAnotherInstance: + """规则换到另一个实例后，旧种子留在原实例；删除规则时两边的种子都删。""" + + async def _rule_with_torrent_on_a(self) -> int: + _mock("a")._torrents["old"] = _completed( + "Old", "old", "/a/Bangumi/Test Anime (2024)/Season 1" + ) + async with Database() as db: + await db.bangumi.add( + make_bangumi(save_path="/a/Bangumi/Test Anime (2024)/Season 1") + ) + await db.torrent.add( + Torrent(name="Old", url="u-old", bangumi_id=1, downloader_id="a") + ) + return 1 + + async def test_update_rule_new_instance_keeps_old_torrents_in_place( + self, two_instances + ): + from module.manager import TorrentManager + from module.models import BangumiUpdate + + bangumi_id = await self._rule_with_torrent_on_a() + async with Database() as db: + old = await db.bangumi.search_id(bangumi_id) + assert old is not None + update = BangumiUpdate(**old.model_dump(exclude={"id"})) + update.downloader_id = "b" + await TorrentManager(db).update_rule(bangumi_id, update) + new = await db.bangumi.search_id(bangumi_id) + + assert new is not None and new.downloader_id == "b" + assert new.save_path == "/b/Bangumi/Test Anime (2024)/Season 1" + moved = _mock("a")._torrents["old"]["save_path"] + assert moved == "/a/Bangumi/Test Anime (2024)/Season 1" + + async def test_delete_rule_deletes_torrents_on_every_instance(self, two_instances): + from module.manager import TorrentManager + + bangumi_id = await self._rule_with_torrent_on_a() + _mock("b")._torrents["new"] = _completed( + "New", "new", "/a/Bangumi/Test Anime (2024)/Season 1" + ) + async with Database() as db: + bangumi = await db.bangumi.search_id(bangumi_id) + assert bangumi is not None + bangumi.downloader_id = "b" + await db.bangumi.update(bangumi) + resp = await TorrentManager(db).delete_rule(bangumi_id, file=True) + + assert resp.status + assert _mock("a")._torrents == {} + assert _mock("b")._torrents == {} + + +class TestDownloaderApi: + def test_pause_routes_to_given_instance(self, two_instances, authed_client): + _mock("b")._torrents["hb"] = _completed("B", "hb", "/b/Bangumi/B") + resp = authed_client.post( + "/api/v1/downloader/torrents/pause", + json={"hashes": ["hb"], "downloader_id": "b"}, + ) + assert resp.status_code == 200 + assert _mock("b")._torrents["hb"]["state"] == "paused" + + def test_pause_unknown_instance_404(self, two_instances, authed_client): + resp = authed_client.post( + "/api/v1/downloader/torrents/pause", + json={"hashes": ["hb"], "downloader_id": "gone"}, + ) + assert resp.status_code == 404 + + def test_instances_lists_downloaders_and_default( + self, two_instances, authed_client + ): + resp = authed_client.get("/api/v1/downloader/instances") + assert resp.json() == { + "default": "a", + "instances": [ + {"id": "a", "provider": "mock"}, + {"id": "b", "provider": "mock"}, + ], + } diff --git a/backend/src/test/test_notification.py b/backend/src/test/test_notification.py index 2756bf91..53bd40ca 100644 --- a/backend/src/test/test_notification.py +++ b/backend/src/test/test_notification.py @@ -561,9 +561,13 @@ class TestEventInboxContract: assert event.kind == "downloader_unavailable" assert event.severity == "error" assert event.once is False - # reason 不进 dedup_key：unreachable→credentials 的翻转合并为一条 - assert event.dedup_key() == "downloader:http://qb:8080" - assert event.payload() == {"host": "http://qb:8080", "reason": "credentials"} + # 按实例去重，reason 不进 dedup_key：unreachable→credentials 的翻转合并为一条 + assert event.dedup_key() == "downloader:default" + assert event.payload() == { + "host": "http://qb:8080", + "reason": "credentials", + "instance": "default", + } title, body = event.describe() assert "下载器" in title assert "密码" in body diff --git a/backend/src/test/test_organize.py b/backend/src/test/test_organize.py index 12c1e6e9..c7f70a71 100644 --- a/backend/src/test/test_organize.py +++ b/backend/src/test/test_organize.py @@ -79,6 +79,7 @@ def add_strategy(registry, provider_id: str, target_name, plugin_id: str = "ext" @pytest.fixture def renamer(mock_qb_client): with patch("module.downloader.download_client.settings") as mock_settings: + mock_settings.downloader.id = "default" mock_settings.downloader.type = "qbittorrent" mock_settings.downloader.host = "localhost:8080" mock_settings.downloader.path = "/downloads/Bangumi" diff --git a/backend/src/test/test_plugin_ingest.py b/backend/src/test/test_plugin_ingest.py index 1a09d27d..f84fa84b 100644 --- a/backend/src/test/test_plugin_ingest.py +++ b/backend/src/test/test_plugin_ingest.py @@ -31,7 +31,7 @@ from module.plugin.registry import HookEntry, ProviderEntry from module.plugin.runner import CircuitBreaker, HookRunner from module.rss.analyser import RSSAnalyser from module.rss.engine import RSSEngine -from test.factories import make_bangumi, make_rss_item, make_torrent +from test.factories import SingleClientPool, make_bangumi, make_rss_item, make_torrent TORRENT_NAME = "[Sub] Mushoku Tensei - 12 [1080p].mkv" @@ -69,7 +69,7 @@ async def refresh_with(engine: RSSEngine, torrents: list[Torrent]): client.add_torrent = AsyncMock(return_value=AddResult.ADDED) with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as get: get.return_value = torrents - await engine.refresh_rss(client) + await engine.refresh_rss(SingleClientPool(client)) return client diff --git a/backend/src/test/test_rss_engine_new.py b/backend/src/test/test_rss_engine_new.py index 9c5df32c..1576a7f9 100644 --- a/backend/src/test/test_rss_engine_new.py +++ b/backend/src/test/test_rss_engine_new.py @@ -14,7 +14,7 @@ from module.notification.events import DownloadFailureEvent, RssFailureEvent from module.parser.analyser.selector import parse_configured_release_title from module.parser.analyser.tokenizer import ReleaseKind from module.rss.engine import RSSEngine -from test.factories import make_bangumi, make_rss_item, make_torrent +from test.factories import SingleClientPool, make_bangumi, make_rss_item, make_torrent @pytest_asyncio.fixture @@ -281,7 +281,7 @@ class TestRefreshRss: client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) # Verify download was attempted client.add_torrent.assert_called_once() @@ -305,7 +305,7 @@ class TestRefreshRss: ) as mock_get: mock_get.return_value = [unmatched] client = AsyncMock() - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) client.add_torrent.assert_not_called() all_torrents = await rss_engine.db.torrent.search_all() @@ -341,7 +341,7 @@ class TestRefreshRss: mock_get.return_value = [matched, unmatched] client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) all_torrents = await rss_engine.db.torrent.search_all() assert len(all_torrents) == 1 @@ -372,7 +372,7 @@ class TestRefreshRss: mock_get.return_value = [matched, unmatched] client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) all_torrents = await rss_engine.db.torrent.search_all() assert len(all_torrents) == 2 @@ -390,7 +390,7 @@ class TestRefreshRss: ) as mock_get: mock_get.return_value = [] client = AsyncMock() - await rss_engine.refresh_rss(client, rss_id=2) + await rss_engine.refresh_rss(SingleClientPool(client), rss_id=2) # Only called once (for rss_id=2) mock_get.assert_called_once() @@ -401,7 +401,7 @@ class TestRefreshRss: RSSEngine, "_get_torrents", new_callable=AsyncMock ) as mock_get: client = AsyncMock() - await rss_engine.refresh_rss(client, rss_id=999) + await rss_engine.refresh_rss(SingleClientPool(client), rss_id=999) mock_get.assert_not_called() @@ -426,7 +426,7 @@ class TestRefreshRssEvents: ) as mock_pull: mock_pull.return_value = ([], "connection refused") client = AsyncMock() - events = await rss_engine.refresh_rss(client) + events = await rss_engine.refresh_rss(SingleClientPool(client)) assert len(events) == 1 assert isinstance(events[0], RssFailureEvent) @@ -443,8 +443,8 @@ class TestRefreshRssEvents: ) as mock_pull: mock_pull.return_value = ([], "connection refused") client = AsyncMock() - first_events = await rss_engine.refresh_rss(client) - second_events = await rss_engine.refresh_rss(client) + first_events = await rss_engine.refresh_rss(SingleClientPool(client)) + second_events = await rss_engine.refresh_rss(SingleClientPool(client)) assert len(first_events) == 1 assert second_events == [] @@ -459,7 +459,7 @@ class TestRefreshRssEvents: ) as mock_pull: mock_pull.return_value = ([], None) client = AsyncMock() - events = await rss_engine.refresh_rss(client) + events = await rss_engine.refresh_rss(SingleClientPool(client)) assert events == [] @@ -481,7 +481,7 @@ class TestRefreshRssEvents: client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.FAILED) - events = await rss_engine.refresh_rss(client) + events = await rss_engine.refresh_rss(SingleClientPool(client)) assert len(events) == 1 assert isinstance(events[0], DownloadFailureEvent) @@ -512,7 +512,7 @@ class TestRefreshRssEvents: client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - events = await rss_engine.refresh_rss(client) + events = await rss_engine.refresh_rss(SingleClientPool(client)) assert events == [] @@ -537,7 +537,7 @@ class TestRefreshRssEvents: client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.DUPLICATE) - events = await rss_engine.refresh_rss(client) + events = await rss_engine.refresh_rss(SingleClientPool(client)) assert events == [] all_torrents = await rss_engine.db.torrent.search_all() @@ -573,14 +573,14 @@ class TestRefreshRssRetry: # Tick 1: transient failure mock_get.return_value = [feed_torrent()] client.add_torrent = AsyncMock(return_value=AddResult.FAILED) - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) client.add_torrent.assert_called_once() assert await rss_engine.db.torrent.search_all() == [] # Tick 2: same feed item is still there, add now succeeds mock_get.return_value = [feed_torrent()] client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) client.add_torrent.assert_called_once() all_torrents = await rss_engine.db.torrent.search_all() @@ -610,9 +610,9 @@ class TestRefreshRssRetry: client.add_torrent = AsyncMock(return_value=AddResult.ADDED) mock_get.return_value = [feed_torrent()] - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) mock_get.return_value = [feed_torrent()] - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) client.add_torrent.assert_not_called() all_torrents = await rss_engine.db.torrent.search_all() @@ -811,7 +811,7 @@ class TestRefreshRssConcurrency: rss_engine, "_pull_rss_with_status", side_effect=track_concurrency ): client = AsyncMock() - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) assert max_active <= 5 @@ -855,7 +855,7 @@ class TestRefreshRssPerHostThrottle: ), patch("module.rss.engine.RSS_PER_HOST_DELAY", 0), ): - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) assert max_active["nyaa.example"] == 1 assert max_active["mikan.example"] == 1 @@ -878,7 +878,7 @@ class TestRefreshRssPerHostThrottle: ), patch("module.rss.engine.RSS_PER_HOST_DELAY", 0), ): - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) for rss_id in (1, 2): item = await rss_engine.db.rss.search_id(rss_id) @@ -1334,7 +1334,7 @@ class TestRefreshRssPreferenceDedup: client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) parsed_after_refresh = parse_configured_release_title(torrent.name) assert parsed_after_refresh is not None @@ -1365,7 +1365,7 @@ class TestRefreshRssPreferenceDedup: client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) client.add_torrent.assert_called_once() called_torrent = client.add_torrent.call_args[0][0] @@ -1400,6 +1400,6 @@ class TestRefreshRssPreferenceDedup: client = AsyncMock() client.add_torrent = AsyncMock(return_value=AddResult.ADDED) - await rss_engine.refresh_rss(client) + await rss_engine.refresh_rss(SingleClientPool(client)) assert client.add_torrent.call_count == 2