From e0b8939ec5fd0025a933a95f84ffd4e68a5f1fc7 Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Thu, 8 Oct 2026 10:32:15 +0200 Subject: [PATCH] =?UTF-8?q?feat(4.0-P4):=20=E9=87=8D=E5=91=BD=E5=90=8D?= =?UTF-8?q?=E7=BB=8F=20rename=5Fstrategy=20=E8=A7=A3=E6=9E=90=EF=BC=8Cmedi?= =?UTF-8?q?a=5Ffiles=20/=20conflict=5Fpolicy=20=E6=94=B9=E4=B8=BA=20core?= =?UTF-8?q?=20Provider=EF=BC=8C=E5=8F=91=E5=B8=83=E6=95=B4=E7=90=86?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 重命名方式按 rename_method 从注册表解析；未登记时记录一次日志并按 none 处理， 不再给未改名的种子打 ab:renamed 标签。字幕改用 RenameInput.kind，不再拼 subtitle_*。 - 插件策略经 runner 同步调用（异常与无效返回值计入熔断）；策略抛出 RenameSkipped 或失败时文件保留原名、种子不打标签，每个种子同一原因只发一条 rename_skipped 通知。 - check_files 经 media_files 分类；版本冲突经 conflict_policy 判定，core 实现与 3.x 的原因文案一致。 - 每轮结束发布 file.renamed；种子整理完成（含 none）发布 torrent.organized。 - pn / advance 暂以 core 登记，下一步移入内置插件 rename。 --- backend/src/module/api/plugins.py | 1 + backend/src/module/downloader/path.py | 24 +- .../src/module/manager/rename_strategies.py | 66 +-- backend/src/module/manager/renamer.py | 379 ++++++++++++++---- backend/src/module/manager/revision_policy.py | 42 ++ backend/src/module/manager/revision_saga.py | 29 +- backend/src/module/notification/__init__.py | 2 + backend/src/module/notification/events.py | 2 + backend/src/module/plugin/host.py | 23 ++ backend/src/module/plugin/runner.py | 46 ++- backend/src/test/test_organize.py | 255 ++++++++++++ backend/src/test/test_plugin_extensions.py | 1 + backend/src/test/test_renamer.py | 12 +- 13 files changed, 732 insertions(+), 150 deletions(-) create mode 100644 backend/src/test/test_organize.py diff --git a/backend/src/module/api/plugins.py b/backend/src/module/api/plugins.py index 1045a9f2..21d7aa0b 100644 --- a/backend/src/module/api/plugins.py +++ b/backend/src/module/api/plugins.py @@ -119,5 +119,6 @@ async def list_plugin_providers(): points.NOTIFIER, points.SEARCH_SITE, points.METADATA_PROVIDER, + points.RENAME_STRATEGY, ) } diff --git a/backend/src/module/downloader/path.py b/backend/src/module/downloader/path.py index cf14cc37..a3c4b6f9 100644 --- a/backend/src/module/downloader/path.py +++ b/backend/src/module/downloader/path.py @@ -3,9 +3,12 @@ import re from os import PathLike from pathlib import PureWindowsPath +from ab_sdk import points +from ab_sdk.rename import CORE_ID, MediaFiles, MediaKind from module.conf import PLATFORM, settings from module.models import Bangumi, BangumiUpdate from module.models.movie import Movie, MovieUpdate +from module.plugin import host as plugin_host logger = logging.getLogger(__name__) @@ -38,15 +41,30 @@ def sanitize_path_fragment(name: str) -> str: return cleaned.rstrip(". ") +class SuffixMediaFiles: + """宿主自带的 ``media_files`` Provider：按扩展名区分正片与字幕。""" + + def classify(self, path: str) -> MediaKind: + suffix = Path(path).suffix.lower() + if suffix in _MEDIA_SUFFIXES: + return "media" + if suffix in _SUBTITLE_SUFFIXES: + return "subtitle" + return "ignore" + + def check_files(files: list[dict]): + media_files: MediaFiles = ( + plugin_host.get_registry().providers(points.MEDIA_FILES)[CORE_ID].factory() + ) media_list = [] subtitle_list = [] for f in files: file_name = f["name"] - suffix = Path(file_name).suffix.lower() - if suffix in _MEDIA_SUFFIXES: + kind = media_files.classify(file_name) + if kind == "media": media_list.append(file_name) - elif suffix in _SUBTITLE_SUFFIXES: + elif kind == "subtitle": subtitle_list.append(file_name) return media_list, subtitle_list diff --git a/backend/src/module/manager/rename_strategies.py b/backend/src/module/manager/rename_strategies.py index 4b8508f7..595ca074 100644 --- a/backend/src/module/manager/rename_strategies.py +++ b/backend/src/module/manager/rename_strategies.py @@ -8,57 +8,35 @@ 与 gen_save_path 的文件夹命名保持一致 (Title (Year))。 """ -from collections.abc import Callable +from ab_sdk.rename import RenameInput, pad -from module.models import EpisodeFile, SubtitleFile - -Strategy = Callable[[EpisodeFile | SubtitleFile, str, str, str], str] +NO_RENAME = "none" -def _subtitle(file_info: EpisodeFile | SubtitleFile) -> SubtitleFile: - assert isinstance( - file_info, SubtitleFile - ), "subtitle methods require a SubtitleFile" - return file_info +class NoRename: + """``none``：保留原路径。""" + + def target_name(self, f: RenameInput) -> str: + return f.media_path -def pn( - file_info: EpisodeFile | SubtitleFile, bangumi_name: str, season: str, episode: str -) -> str: - if file_info.episode_type == "movie": - return f"{file_info.title}{file_info.suffix}" - return f"{file_info.title} S{season}E{episode}{file_info.suffix}" +def standard_name(base: str, f: RenameInput) -> str: + """``{base} SxxEyy[.语言].ext``；电影为 ``{base}[.语言].ext``。""" + language = f".{f.language}" if f.kind == "subtitle" else "" + if f.episode_type == "movie": + return f"{base}{language}{f.suffix}" + return f"{base} S{pad(f.season)}E{pad(f.episode)}{language}{f.suffix}" -def advance( - file_info: EpisodeFile | SubtitleFile, bangumi_name: str, season: str, episode: str -) -> str: - if file_info.episode_type == "movie": - return f"{bangumi_name}{file_info.suffix}" - return f"{bangumi_name} S{season}E{episode}{file_info.suffix}" +class PnRename: + """``pn``：以文件名解析出的标题命名。""" + + def target_name(self, f: RenameInput) -> str: + return standard_name(f.title, f) -def subtitle_pn( - file_info: EpisodeFile | SubtitleFile, bangumi_name: str, season: str, episode: str -) -> str: - sub = _subtitle(file_info) - if sub.episode_type == "movie": - return f"{sub.title}.{sub.language}{sub.suffix}" - return f"{sub.title} S{season}E{episode}.{sub.language}{sub.suffix}" +class AdvanceRename: + """``advance``：以番剧文件夹名命名。""" - -def subtitle_advance( - file_info: EpisodeFile | SubtitleFile, bangumi_name: str, season: str, episode: str -) -> str: - sub = _subtitle(file_info) - if sub.episode_type == "movie": - return f"{bangumi_name}.{sub.language}{sub.suffix}" - return f"{bangumi_name} S{season}E{episode}.{sub.language}{sub.suffix}" - - -STRATEGIES: dict[str, Strategy] = { - "pn": pn, - "advance": advance, - "subtitle_pn": subtitle_pn, - "subtitle_advance": subtitle_advance, -} + def target_name(self, f: RenameInput) -> str: + return standard_name(f.bangumi_name, f) diff --git a/backend/src/module/manager/renamer.py b/backend/src/module/manager/renamer.py index 1a2b4e06..ea4fd6ed 100644 --- a/backend/src/module/manager/renamer.py +++ b/backend/src/module/manager/renamer.py @@ -2,7 +2,19 @@ import asyncio import logging from datetime import datetime, timedelta, timezone from pathlib import PurePath +from typing import Any +from ab_sdk import points +from ab_sdk.events import FileRenamed, OrganizedFile, TorrentOrganized +from ab_sdk.rename import ( + CORE_ID, + ConflictPolicy, + ConflictRequest, + FileKind, + RenameInput, + RenameSkipped, + RevisionTask, +) from module.conf import settings from module.database import Database from module.database.bangumi import ( @@ -13,11 +25,17 @@ from module.database.bangumi import ( from module.downloader import DownloadClient, RenameOutcome, RenameResult from module.downloader.path import check_files, is_ep, path_to_bangumi from module.models import EpisodeFile, Notification, SubtitleFile -from module.notification import RenameConflictEvent +from module.notification import RenameSkippedEvent, SystemEvent from module.parser import TitleParser +from module.plugin import host as plugin_host +from module.plugin.registry import ProviderEntry -from .rename_strategies import STRATEGIES -from .revision_policy import is_strict_upgrade, parse_revision_identity +from .rename_strategies import NO_RENAME +from .revision_policy import ( + is_strict_upgrade, + parse_revision_identity, + revision_snapshot, +) from .revision_saga import ( _RENAMED_TAG, MediaRenameReport, @@ -27,20 +45,89 @@ from .revision_saga import ( logger = logging.getLogger(__name__) +# 下载器 id：多下载器（P2.5）之前只有一个 +_DOWNLOADER_ID = "default" +# 已记录过「未登记」日志的重命名方式（每个进程只记一次，重新登记后清除） +_missing_methods: set[str] = set() +# 已发布过的 TorrentOrganized（种子 hash → 最终文件）。未打 ab:renamed 标签的 +# 种子（重命名方式为 none）每轮都会重新处理，文件不变时不重复发布。 +# ponytail: 进程内字典，重启后重发一次（事件约定为至少一次）；条目随种子数增长 +_organized_published: dict[str, tuple[OrganizedFile, ...]] = {} +# 已发送过「文件未重命名」通知的 (种子 hash, 原因) +_skip_notified: set[tuple[str, str]] = set() + + +def _rename_strategy(method: str) -> ProviderEntry: + """按 rename_method 取重命名策略；未登记（如 rename 插件未启用）时记录一次 + 日志并按 none 处理。""" + strategies = plugin_host.get_registry().providers(points.RENAME_STRATEGY) + entry = strategies.get(method) + if entry is not None: + _missing_methods.discard(method) + return entry + if method not in _missing_methods: + _missing_methods.add(method) + logger.warning( + "[Renamer] 重命名方式 %s 未登记（提供它的插件未启用？），文件保留原名", + method, + ) + return strategies[NO_RENAME] + + +def _valid_name(result: Any) -> bool: + return isinstance(result, RenameSkipped) or ( + isinstance(result, str) and result != "" + ) + + +def _target_name(entry: ProviderEntry, f: RenameInput) -> str: + """调用重命名策略。策略给不出合法名字时抛出 RenameSkipped。 + + 插件策略经 runner 调用：异常与无效返回值计入熔断；策略主动抛出的 + RenameSkipped（输入或用户配置有问题）不计入。 + """ + if entry.plugin_id == plugin_host.CORE: + return entry.factory().target_name(f) + + def call() -> str | RenameSkipped: + try: + return entry.factory().target_name(f) + except RenameSkipped as e: + return e + + runner = plugin_host.get_runner() + if runner is not None: + ok, result = runner.call_provider_sync( + entry.plugin_id, points.RENAME_STRATEGY, call, check=_valid_name + ) + else: + try: + result = call() + ok = _valid_name(result) + except Exception as e: + logger.warning("[Plugin:%s] 重命名策略失败：%s", entry.plugin_id, e) + ok = False + if isinstance(result, RenameSkipped): + raise result + if not ok or result is None: + raise RenameSkipped(f"重命名方式 {entry.id} 执行失败，详见日志") + return result + + +def _downloader_path(save_path: str, name: str) -> str: + """下载器视角的绝对路径，统一以 "/" 分隔。""" + relative = name.replace("\\", "/") + return f"{normalize_save_path(save_path)}/{relative}" + class Renamer(RevisionSaga): def __init__(self, client: DownloadClient): self.client = client self._parser = TitleParser() - self.events: list[RenameConflictEvent] = [] - - @staticmethod - def _format_episode(episode: int | float) -> str: - # 总集篇等半集（12.5）保留小数，否则会覆盖同季的整数集 (#667)； - # 整数值沿用两位补零 - if isinstance(episode, float) and episode.is_integer(): - episode = int(episode) - return f"0{episode}" if episode < 10 else str(episode) + self.events: list[SystemEvent] = [] + # 本轮各种子的实际重命名 (原路径, 新路径, 类别) 与被跳过的原因 + self._moves: dict[str, list[tuple[str, str, FileKind]]] = {} + self._skipped: dict[str, str] = {} @staticmethod def gen_path( @@ -50,24 +137,31 @@ class Renamer(RevisionSaga): episode_offset: int = 0, season_offset: int = 0, # Kept for API compatibility, but no longer used ) -> str: + """按重命名方式生成种子内的目标路径。 + + 策略给不出合法名字时抛出 RenameSkipped，调用方保留原文件名。 + """ # Season comes from the folder name which already includes the offset # (folder is now "Season {season + season_offset}") # So we use file_info.season directly without applying offset again - season_num = file_info.season - season = f"0{season_num}" if season_num < 10 else season_num - episode = Renamer._format_episode( - Renamer._adjust_episode(file_info.episode, episode_offset) - ) # 注意：group_tag 只影响 qB RSS 规则名（downloader/path.py 的 rule_name）， # 从不写进重命名后的文件名——已有做种媒体库的文件名必须保持稳定， # 否则升级后会触发整库批量重命名，破坏 Plex/Jellyfin 索引与硬链接 - if method == "none" or method == "subtitle_none": - return file_info.media_path - strategy = STRATEGIES.get(method) - if strategy is None: - logger.error(f"Unknown rename method: {method}") - return file_info.media_path - return strategy(file_info, bangumi_name, str(season), episode) + f = RenameInput( + kind="subtitle" if isinstance(file_info, SubtitleFile) else "media", + media_path=file_info.media_path, + title=file_info.title, + bangumi_name=bangumi_name, + season=file_info.season, + episode=Renamer._adjust_episode(file_info.episode, episode_offset), + suffix=file_info.suffix, + episode_type=file_info.episode_type, + language=( + file_info.language if isinstance(file_info, SubtitleFile) else "" + ), + group=file_info.group, + ) + return _target_name(_rename_strategy(method), f) async def rename_file( self, @@ -94,7 +188,7 @@ class Renamer(RevisionSaga): season_offset=season_offset, episode_type=episode_type, ) - if report.result.succeeded and method != "none": + if report.result.succeeded and _rename_strategy(method).id != NO_RENAME: await self._mark_renamed(_hash, existing_tags) return report.notification @@ -143,16 +237,19 @@ class Renamer(RevisionSaga): season_offset: int = 0, episode_type: str = "episode", ) -> MediaRenameReport: - prepared = self._prepare_media_rename( - torrent_name=torrent_name, - media_path=media_path, - bangumi_name=bangumi_name, - method=method, - season=season, - episode_offset=episode_offset, - season_offset=season_offset, - episode_type=episode_type, - ) + try: + prepared = self._prepare_media_rename( + torrent_name=torrent_name, + media_path=media_path, + bangumi_name=bangumi_name, + method=method, + season=season, + episode_offset=episode_offset, + season_offset=season_offset, + episode_type=episode_type, + ) + except RenameSkipped as e: + return self._skip(_hash, media_path, e) if prepared is None: logger.warning("%s parse failed", media_path) if settings.bangumi_manage.remove_bad_torrent: @@ -183,6 +280,14 @@ class Renamer(RevisionSaga): stem = stem[len(prefix) :] return f"{base} - {stem}{suffix}" + def _skip(self, _hash: str, path: str, error: RenameSkipped) -> MediaRenameReport: + """策略给不出名字：保留原文件名，记录原因（本轮结束时通知、不打标签）。""" + logger.warning("[Renamer] %s 保留原名：%s", path, error) + self._skipped.setdefault(_hash, str(error)) + return MediaRenameReport( + result=RenameResult(RenameOutcome.RETRYABLE_FAILURE, detail=str(error)) + ) + async def rename_collection( self, media_list: list[str], @@ -218,13 +323,18 @@ class Renamer(RevisionSaga): episode_type=episode_type, ) if ep: - new_path = self.gen_path( - ep, - bangumi_name, - method=method, - episode_offset=episode_offset, - season_offset=season_offset, - ) + try: + new_path = self.gen_path( + ep, + bangumi_name, + method=method, + episode_offset=episode_offset, + season_offset=season_offset, + ) + except RenameSkipped as e: + self._skip(_hash, media_path, e) + all_renamed = False + continue if ( movie_primary is not None and media_path != movie_primary @@ -261,13 +371,15 @@ class Renamer(RevisionSaga): old_path=media_path, new_path=new_path, ) + if result.outcome is RenameOutcome.RENAMED: + self._record_move(_hash, media_path, new_path, "media") if not result.succeeded: all_renamed = False logger.warning(f"{media_path} rename failed") else: # 解析失败的媒体文件不会被重命名——不能算处理完成 all_renamed = False - if all_renamed and mark_complete and method != "none": + if all_renamed and mark_complete and _rename_strategy(method).id != NO_RENAME: await self._mark_renamed(_hash, existing_tags) return all_renamed @@ -284,7 +396,6 @@ class Renamer(RevisionSaga): episode_type: str = "episode", **kwargs, ): - method = "subtitle_" + method for subtitle_path in subtitle_list: sub = self._parser.torrent_parser( torrent_path=subtitle_path, @@ -294,13 +405,17 @@ class Renamer(RevisionSaga): episode_type=episode_type, ) if sub: - new_path = self.gen_path( - sub, - bangumi_name, - method=method, - episode_offset=episode_offset, - season_offset=season_offset, - ) + try: + new_path = self.gen_path( + sub, + bangumi_name, + method=method, + episode_offset=episode_offset, + season_offset=season_offset, + ) + except RenameSkipped as e: + self._skip(_hash, subtitle_path, e) + continue if subtitle_path != new_path: # Skip verification for subtitles to reduce latency renamed = await self.client.rename_torrent_file( @@ -309,6 +424,8 @@ class Renamer(RevisionSaga): new_path=new_path, verify=False, ) + if renamed.outcome is RenameOutcome.RENAMED: + self._record_move(_hash, subtitle_path, new_path, "subtitle") if not renamed: logger.warning(f"{subtitle_path} rename failed") @@ -330,16 +447,19 @@ class Renamer(RevisionSaga): season_offset: int, episode_type: str, ) -> MediaRenameReport: - prepared = self._prepare_media_rename( - torrent_name=info["name"], - media_path=media_path, - bangumi_name=bangumi_name, - method=method, - season=season, - episode_offset=episode_offset, - season_offset=season_offset, - episode_type=episode_type, - ) + try: + prepared = self._prepare_media_rename( + torrent_name=info["name"], + media_path=media_path, + bangumi_name=bangumi_name, + method=method, + season=season, + episode_offset=episode_offset, + season_offset=season_offset, + episode_type=episode_type, + ) + except RenameSkipped as e: + return self._skip(info["hash"], media_path, e) if prepared is None: logger.warning("%s parse failed", media_path) if settings.bangumi_manage.remove_bad_torrent: @@ -439,21 +559,42 @@ class Renamer(RevisionSaga): ) if owners: owner = owners[0] if len(owners) == 1 else None - reason = "canonical path has more than one downloader owner" - can_replace = bool( - owner is not None + decision = self._conflict_policy().decide( + ConflictRequest( + target_path=prepared.target_path, + incoming=RevisionTask( + info["hash"], + info.get("name", ""), + len(files), + revision_snapshot(incoming_identity), + ), + owners=tuple( + RevisionTask( + o.info["hash"], + o.info.get("name", ""), + len(o.files), + revision_snapshot(o.identity), + ) + for o in owners + ), + configured=settings.bangumi_manage.revision_conflict_policy, + strict_upgrade=bool( + owner is not None + and incoming_identity is not None + and owner.identity is not None + and is_strict_upgrade(owner.identity, incoming_identity) + ), + ) + ) + # 替换 saga 只支持「唯一占用者、双方单文件、身份完整」的情形 + if ( + decision.action == "replace" + and owner is not None and len(files) == 1 and len(owner.files) == 1 and incoming_identity is not None and owner.identity is not None - and is_strict_upgrade(owner.identity, incoming_identity) - ) - if ( - settings.bangumi_manage.revision_conflict_policy == "replace" - and can_replace ): - assert owner is not None - assert incoming_identity is not None notification = await self._start_replacement( info=info, prepared=prepared, @@ -484,16 +625,7 @@ class Renamer(RevisionSaga): notification=notification, ) - if len(owners) == 1: - assert owner is not None - if len(files) != 1 or len(owner.files) != 1: - reason = "automatic replacement requires two single-file torrents" - elif incoming_identity is None or owner.identity is None: - reason = "revision identity is incomplete" - elif not is_strict_upgrade(owner.identity, incoming_identity): - reason = "existing and incoming releases are not a strict revision upgrade" - else: - reason = "revision conflict policy is hold" + reason = decision.reason or "automatic replacement is not possible" await self._persist_conflict( info=info, prepared=prepared, @@ -514,6 +646,75 @@ class Renamer(RevisionSaga): episode_offset=episode_offset, ) + def _finish_torrent( + self, + info: dict, + media_list: list[str], + subtitle_list: list[str], + method: str, + organized: bool, + ) -> None: + """发布该种子本轮的 file.renamed；整理完成时发布 torrent.organized； + 有文件因策略给不出名字而保留原名时改为发一条通知（同一原因只发一次）。""" + torrent_hash = info["hash"] + save_path = info["save_path"] + bangumi_id = self._parse_bangumi_id_from_tags(info.get("tags")) + final: dict[str, str] = {} + for old, new, kind in self._moves.pop(torrent_hash, []): + final[old] = new + plugin_host.publish( + FileRenamed( + bangumi_id=bangumi_id, + old_path=_downloader_path(save_path, old), + new_path=_downloader_path(save_path, new), + file_kind=kind, + downloader_id=_DOWNLOADER_ID, + ) + ) + reason = self._skipped.pop(torrent_hash, None) + if reason is not None: + if (torrent_hash, reason) not in _skip_notified: + _skip_notified.add((torrent_hash, reason)) + self.events.append( + RenameSkippedEvent( + task_id=torrent_hash, + torrent_name=info["name"], + strategy=method, + reason=reason, + ) + ) + return + if not organized: + return + kinds: tuple[tuple[list[str], FileKind], ...] = ( + (media_list, "media"), + (subtitle_list, "subtitle"), + ) + files = tuple( + OrganizedFile(_downloader_path(save_path, final.get(name, name)), kind) + for names, kind in kinds + for name in names + ) + if _organized_published.get(torrent_hash) == files: + return + _organized_published[torrent_hash] = files + plugin_host.publish( + TorrentOrganized( + torrent_hash=torrent_hash, + bangumi_id=bangumi_id, + files=files, + downloader_id=_DOWNLOADER_ID, + ) + ) + + @staticmethod + def _conflict_policy() -> ConflictPolicy: + return ( + plugin_host.get_registry() + .providers(points.CONFLICT_POLICY)[CORE_ID] + .factory() + ) + @staticmethod def _normalize_path(path: str) -> str: """Normalize path by removing trailing slashes and standardizing separators.""" @@ -726,7 +927,7 @@ class Renamer(RevisionSaga): async def rename(self) -> list[Notification]: logger.debug("Start rename process.") - rename_method = settings.bangumi_manage.rename_method + rename_method = _rename_strategy(settings.bangumi_manage.rename_method).id pending_infos = await self.client.get_torrent_info() # Owner counting and Saga recovery must see tasks outside the normal # Bangumi/completed filter (collections, paused tasks, changed category). @@ -821,8 +1022,15 @@ class Renamer(RevisionSaga): await self.rename_subtitles( subtitle_list=subtitle_list, **kwargs ) - if rename_method != "none": + if rename_method != NO_RENAME and torrent_hash not in self._skipped: await self._mark_renamed(torrent_hash, info.get("tags")) + self._finish_torrent( + info, + media_list, + subtitle_list, + rename_method, + report.result.succeeded, + ) elif len(media_list) > 1: logger.info("Start rename collection") file_sizes = {f["name"]: f.get("size") or 0 for f in files} @@ -836,9 +1044,12 @@ class Renamer(RevisionSaga): if collection_complete and subtitle_list: await self.rename_subtitles(subtitle_list=subtitle_list, **kwargs) if collection_complete: - if rename_method != "none": + if rename_method != NO_RENAME and torrent_hash not in self._skipped: await self._mark_renamed(torrent_hash, info.get("tags")) await self.client.set_category(torrent_hash, "BangumiCollection") + self._finish_torrent( + info, media_list, subtitle_list, rename_method, collection_complete + ) else: logger.warning(f"{torrent_name} has no media file") async with Database() as db: diff --git a/backend/src/module/manager/revision_policy.py b/backend/src/module/manager/revision_policy.py index c30c1086..3c26be5b 100644 --- a/backend/src/module/manager/revision_policy.py +++ b/backend/src/module/manager/revision_policy.py @@ -7,6 +7,7 @@ import unicodedata from dataclasses import dataclass from pathlib import PurePosixPath +from ab_sdk.rename import ConflictDecision, ConflictRequest, Revision from module.parser.analyser.selector import parse_configured_release_title from module.parser.analyser.tokenizer import MediaType from module.parser.release_policy import preference_identity, preference_revision @@ -91,6 +92,47 @@ def same_release_identity(old: RevisionIdentity, new: RevisionIdentity) -> bool: ) +def revision_snapshot(identity: RevisionIdentity | None) -> Revision | None: + """交给 conflict_policy 的只读快照。""" + if identity is None: + return None + return Revision( + bangumi_id=identity.bangumi_id, + media_type=identity.media_type.value, + season=identity.season, + episode=identity.episode, + group=identity.group, + resolution=identity.resolution, + revision=identity.revision, + ) + + +class CoreConflictPolicy: + """宿主自带的 ``conflict_policy``：只在唯一占用者与新种子都是单文件、 + 且新种子是严格的版本升级时，按用户设置决定是否替换。""" + + def decide(self, request: ConflictRequest) -> ConflictDecision: + if len(request.owners) != 1: + return ConflictDecision( + "hold", "canonical path has more than one downloader owner" + ) + owner = request.owners[0] + if request.incoming.file_count != 1 or owner.file_count != 1: + return ConflictDecision( + "hold", "automatic replacement requires two single-file torrents" + ) + if request.incoming.revision is None or owner.revision is None: + return ConflictDecision("hold", "revision identity is incomplete") + if not request.strict_upgrade: + return ConflictDecision( + "hold", + "existing and incoming releases are not a strict revision upgrade", + ) + if request.configured == "replace": + return ConflictDecision("replace") + return ConflictDecision("hold", "revision conflict policy is hold") + + def replacement_staged_path( target_path: str, *, old_task_id: str, old_revision: int ) -> str: diff --git a/backend/src/module/manager/revision_saga.py b/backend/src/module/manager/revision_saga.py index 4d107240..7634da60 100644 --- a/backend/src/module/manager/revision_saga.py +++ b/backend/src/module/manager/revision_saga.py @@ -15,13 +15,14 @@ from datetime import datetime, timedelta, timezone from sqlalchemy.exc import IntegrityError +from ab_sdk.rename import FileKind from module.conf import settings from module.database import Database from module.database.bangumi import normalize_save_path from module.downloader import DownloadClient, RenameOutcome, RenameResult from module.downloader.path import path_to_bangumi from module.models import EpisodeFile, Notification, RenameOperation, SubtitleFile -from module.notification import RenameConflictEvent +from module.notification import RenameConflictEvent, SystemEvent from .revision_policy import ( RevisionIdentity, @@ -62,10 +63,16 @@ class RevisionOwner: class RevisionSaga: - """由 Renamer 继承；子类提供下载器门面 ``client`` 与冲突事件列表 ``events``。""" + """由 Renamer 继承；子类提供下载器门面 ``client``、冲突事件列表 ``events`` + 与本轮重命名记录 ``_moves``。""" client: DownloadClient - events: list[RenameConflictEvent] + events: list[SystemEvent] + _moves: dict[str, list[tuple[str, str, FileKind]]] + + def _record_move(self, _hash: str, old: str, new: str, kind: FileKind) -> None: + """记录一次已生效的重命名，供本轮结束时发布 file.renamed / torrent.organized。""" + self._moves.setdefault(_hash, []).append((old, new, kind)) @staticmethod def _adjust_episode(original: int | float, episode_offset: int) -> int | float: @@ -116,6 +123,9 @@ class RevisionSaga: ) notification = None if result.outcome is RenameOutcome.RENAMED: + self._record_move( + _hash, prepared.source_path, prepared.target_path, "media" + ) notification = Notification( official_title=bangumi_name, season=prepared.episode.season, @@ -549,6 +559,13 @@ class RevisionSaga: old_path=operation.source_path, new_path=operation.target_path, ) + if result.outcome is RenameOutcome.RENAMED: + self._record_move( + operation.new_task_id, + operation.source_path, + operation.target_path, + "media", + ) if result.succeeded: operation = await self._set_operation_state( operation, "new_promoted" @@ -565,6 +582,12 @@ class RevisionSaga: == operation.target_path for item in new_files ): + self._record_move( + operation.new_task_id, + operation.source_path, + operation.target_path, + "media", + ) operation = await self._set_operation_state( operation, "new_promoted" ) diff --git a/backend/src/module/notification/__init__.py b/backend/src/module/notification/__init__.py index 78b8fc53..7ef5e37c 100644 --- a/backend/src/module/notification/__init__.py +++ b/backend/src/module/notification/__init__.py @@ -6,6 +6,7 @@ from .events import ( LLMPluginInstallFailedEvent, OffsetReviewEvent, RenameConflictEvent, + RenameSkippedEvent, RssFailureEvent, SystemEvent, UpdateAppliedEvent, @@ -23,6 +24,7 @@ __all__ = [ "NotificationProvider", "OffsetReviewEvent", "RenameConflictEvent", + "RenameSkippedEvent", "PROVIDER_REGISTRY", "RssFailureEvent", "SystemEvent", diff --git a/backend/src/module/notification/events.py b/backend/src/module/notification/events.py index 6830c78a..ae34d76a 100644 --- a/backend/src/module/notification/events.py +++ b/backend/src/module/notification/events.py @@ -26,6 +26,7 @@ from ab_sdk.events import ( LLMPluginInstallFailedEvent, OffsetReviewEvent, RenameConflictEvent, + RenameSkippedEvent, RssFailureEvent, SystemEvent, UpdateAppliedEvent, @@ -39,6 +40,7 @@ __all__ = [ "LLMPluginInstallFailedEvent", "OffsetReviewEvent", "RenameConflictEvent", + "RenameSkippedEvent", "RssFailureEvent", "SystemEvent", "UpdateAppliedEvent", diff --git a/backend/src/module/plugin/host.py b/backend/src/module/plugin/host.py index 8f3d4159..cd09ef6e 100644 --- a/backend/src/module/plugin/host.py +++ b/backend/src/module/plugin/host.py @@ -14,6 +14,7 @@ from typing import Any from ab_sdk import Event, points from ab_sdk.downloader import DownloaderConnection +from ab_sdk.rename import CORE_ID from .bus import EventBus from .registry import ExtensionPoint, ExtensionRegistry, ProviderEntry @@ -51,6 +52,12 @@ POINTS = ( ExtensionPoint(points.MCP_TOOL, "provider", "MCP 工具", scoped=True), ExtensionPoint(points.MCP_RESOURCE, "provider", "MCP 资源", scoped=True), ExtensionPoint(points.MESSAGE_TEMPLATE, "transform", "系统事件通知文案"), + # --- P4 organize --- + ExtensionPoint( + points.RENAME_STRATEGY, "provider", "重命名方式（bangumi_manage.rename_method）" + ), + ExtensionPoint(points.MEDIA_FILES, "provider", "种子内文件分类（正片 / 字幕）"), + ExtensionPoint(points.CONFLICT_POLICY, "provider", "目标路径被占用时保留或替换"), ) _registry: ExtensionRegistry | None = None @@ -174,6 +181,22 @@ def _register_core(registry: ExtensionRegistry) -> None: ProviderEntry(CORE, provider_id, metadata(provider_id)), ) + # organize：重命名方式 none、按扩展名分类、版本冲突策略 + from module.downloader.path import SuffixMediaFiles + from module.manager.rename_strategies import ( + NO_RENAME, + AdvanceRename, + NoRename, + PnRename, + ) + from module.manager.revision_policy import CoreConflictPolicy + + _core(registry, points.RENAME_STRATEGY, NO_RENAME, NoRename()) + _core(registry, points.RENAME_STRATEGY, "pn", PnRename()) + _core(registry, points.RENAME_STRATEGY, "advance", AdvanceRename()) + _core(registry, points.MEDIA_FILES, CORE_ID, SuffixMediaFiles()) + _core(registry, points.CONFLICT_POLICY, CORE_ID, CoreConflictPolicy()) + # --- P5 events/api --------------------------------------------------------- diff --git a/backend/src/module/plugin/runner.py b/backend/src/module/plugin/runner.py index ed050252..79dc11fd 100644 --- a/backend/src/module/plugin/runner.py +++ b/backend/src/module/plugin/runner.py @@ -139,6 +139,25 @@ class HookRunner: entry = HookEntry(plugin_id, func, 0, timeout) return await self._call(entry, point, args, {}, check) + def call_provider_sync( + self, + plugin_id: str, + point: str, + func: Callable[..., Any], + *args: Any, + check: Check | None = None, + ) -> tuple[bool, Any]: + """同步 Provider 方法（如重命名策略）的调用：没有超时，失败与返回值 + 校验的规则同 :meth:`call_provider`。""" + try: + result = func(*args) + _check(result, check) + except Exception as e: + self._fail(plugin_id, f"{point}: {type(e).__name__}: {e}") + return False, None + self._breaker.record_success(plugin_id) + return True, result + async def _call( self, entry: HookEntry, @@ -152,21 +171,30 @@ class HookRunner: result = entry.func(*args, **kwargs) if inspect.isawaitable(result): result = await asyncio.wait_for(result, timeout) - # 返回值校验在记成功之前：无效结果必须计入熔断，而不是先清零再计一次 - if result is not None and check is not None and not check(result): - raise TypeError(f"返回了无效结果 {type(result).__name__}，已忽略") + _check(result, check) except Exception as e: - reason = ( - f"{point} 超时（{timeout}s）" - if isinstance(e, TimeoutError) - else f"{point}: {type(e).__name__}: {e}" + self._fail( + entry.plugin_id, + ( + f"{point} 超时（{timeout}s）" + if isinstance(e, TimeoutError) + else f"{point}: {type(e).__name__}: {e}" + ), ) - logger.warning("[Plugin:%s] 钩子失败：%s", entry.plugin_id, reason) - self._breaker.record_failure(entry.plugin_id, reason) return False, None self._breaker.record_success(entry.plugin_id) return True, result + def _fail(self, plugin_id: str, reason: str) -> None: + logger.warning("[Plugin:%s] 钩子失败：%s", plugin_id, reason) + self._breaker.record_failure(plugin_id, reason) + + +def _check(result: Any, check: Check | None) -> None: + # 返回值校验在记成功之前：无效结果必须计入熔断，而不是先清零再计一次 + if result is not None and check is not None and not check(result): + raise TypeError(f"返回了无效结果 {type(result).__name__}，已忽略") + def _as_verdict(result: Any) -> Verdict | None: if isinstance(result, Verdict): diff --git a/backend/src/test/test_organize.py b/backend/src/test/test_organize.py new file mode 100644 index 00000000..7f2d1cca --- /dev/null +++ b/backend/src/test/test_organize.py @@ -0,0 +1,255 @@ +"""P4 organize 扩展点：rename_strategy 的解析、回退与跳过通知，conflict_policy， +以及 file.renamed / torrent.organized 事件。""" + +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from ab_sdk import points +from ab_sdk.events import ( + FileRenamed, + OrganizedFile, + RenameSkippedEvent, + TorrentOrganized, +) +from ab_sdk.rename import ( + ConflictRequest, + RenameInput, + RenameSkipped, + Revision, + RevisionTask, +) +from ab_sdk.testing import RecordingBus +from module.downloader import DownloadClient +from module.manager import renamer as renamer_module +from module.manager.renamer import Renamer +from module.manager.revision_policy import CoreConflictPolicy +from module.models import EpisodeFile, SubtitleFile +from module.plugin import host +from module.plugin.registry import ProviderEntry +from module.plugin.runner import CircuitBreaker, HookRunner + +SAVE_PATH = "/downloads/Bangumi/Anime (2024)/Season 1" +NAME = "[Sub] Anime - 01.mkv" +INFO = {"hash": "h1", "name": NAME, "save_path": SAVE_PATH, "tags": "ab:7"} +SUB = "[Sub] Anime - 01.zh.ass" + + +def parse(torrent_path: str, **kwargs) -> EpisodeFile | SubtitleFile: + if torrent_path == SUB: + return SubtitleFile( + media_path=SUB, + title="Anime", + season=1, + episode=1, + language="zh", + suffix=".ass", + ) + return EpisodeFile( + media_path=torrent_path, title="Anime", season=1, episode=1, suffix=".mkv" + ) + + +@pytest.fixture +def plugins(monkeypatch): + """全新的进程级注册表、runner 与记录事件的总线。""" + monkeypatch.setattr(host, "_registry", None) + registry = host.get_registry() + tripped: list[str] = [] + breaker = CircuitBreaker(2, on_trip=lambda plugin_id, _: tripped.append(plugin_id)) + monkeypatch.setattr(host, "_runner", HookRunner(registry, breaker)) + bus = RecordingBus() + monkeypatch.setattr(host, "_bus", bus) + monkeypatch.setattr(renamer_module, "_organized_published", {}) + monkeypatch.setattr(renamer_module, "_skip_notified", set()) + return SimpleNamespace(registry=registry, tripped=tripped, bus=bus) + + +def add_strategy(registry, provider_id: str, target_name, plugin_id: str = "ext"): + impl = SimpleNamespace(target_name=target_name) + registry.add_provider( + points.RENAME_STRATEGY, ProviderEntry(plugin_id, provider_id, lambda: impl) + ) + + +@pytest.fixture +def renamer(mock_qb_client): + with patch("module.downloader.download_client.settings") as mock_settings: + mock_settings.downloader.type = "qbittorrent" + mock_settings.downloader.host = "localhost:8080" + mock_settings.downloader.path = "/downloads/Bangumi" + with patch( + "module.downloader.download_client.DownloadClient._DownloadClient__getClient", + return_value=mock_qb_client, + ): + client = DownloadClient() + client.client = mock_qb_client + mock_qb_client.torrents_info.return_value = [INFO] + mock_qb_client.torrents_files.return_value = [ + {"name": NAME}, + {"name": SUB}, + ] + return Renamer(client) + + +async def run(renamer: Renamer, method: str) -> None: + with ( + patch.object(renamer._parser, "torrent_parser", side_effect=parse), + patch("module.manager.renamer.settings") as mock_settings, + patch("module.downloader.path.settings") as mock_path_settings, + ): + mock_settings.bangumi_manage.rename_method = method + mock_settings.bangumi_manage.remove_bad_torrent = False + mock_path_settings.downloader.path = "/downloads/Bangumi" + await renamer.rename() + + +def published(bus: RecordingBus, cls: type) -> list: + return [e for e in bus.published if isinstance(e, cls)] + + +async def test_rename_unregistered_method_keeps_names_without_tag(plugins, renamer): + await run(renamer, "jellyfin-style") + + renamer.client.client.torrents_rename_file.assert_not_called() + renamer.client.client.add_tag.assert_not_called() + assert published(plugins.bus, TorrentOrganized) + + +async def test_rename_plugin_strategy_renames_and_publishes_events(plugins, renamer): + seen: list[RenameInput] = [] + + def target_name(f: RenameInput) -> str: + seen.append(f) + language = f".{f.language}" if f.kind == "subtitle" else "" + return f"{f.title} E{f.episode}{language}{f.suffix}" + + add_strategy(plugins.registry, "custom", target_name) + await run(renamer, "custom") + + assert [(f.kind, f.bangumi_name) for f in seen] == [ + ("media", "Anime (2024)"), + ("subtitle", "Anime (2024)"), + ] + renamer.client.client.add_tag.assert_awaited_once_with("h1", "ab:renamed") + assert published(plugins.bus, FileRenamed) == [ + FileRenamed( + bangumi_id=7, + old_path=f"{SAVE_PATH}/{NAME}", + new_path=f"{SAVE_PATH}/Anime E1.mkv", + file_kind="media", + ), + FileRenamed( + bangumi_id=7, + old_path=f"{SAVE_PATH}/{SUB}", + new_path=f"{SAVE_PATH}/Anime E1.zh.ass", + file_kind="subtitle", + ), + ] + assert published(plugins.bus, TorrentOrganized) == [ + TorrentOrganized( + torrent_hash="h1", + bangumi_id=7, + files=( + OrganizedFile(f"{SAVE_PATH}/Anime E1.mkv", "media"), + OrganizedFile(f"{SAVE_PATH}/Anime E1.zh.ass", "subtitle"), + ), + ) + ] + + +async def test_rename_none_publishes_organized_once_per_process(plugins, renamer): + await run(renamer, "none") + await run(renamer, "none") + + renamer.client.client.torrents_rename_file.assert_not_called() + assert published(plugins.bus, FileRenamed) == [] + organized = published(plugins.bus, TorrentOrganized) + assert len(organized) == 1 + assert organized[0].files[0] == OrganizedFile(f"{SAVE_PATH}/{NAME}", "media") + + +def _skip(f: RenameInput) -> str: + raise RenameSkipped("模板渲染失败：'titel' is undefined") + + +def _crash(f: RenameInput) -> str: + raise ValueError("boom") + + +@pytest.mark.parametrize( + ("target_name", "counted"), + [(_skip, False), (_crash, True), (lambda f: "", True), (lambda f: 3, True)], +) +async def test_rename_strategy_failure_keeps_name_and_notifies_once( + plugins, renamer, target_name, counted +): + add_strategy(plugins.registry, "custom", target_name) + await run(renamer, "custom") + await run(renamer, "custom") + + renamer.client.client.torrents_rename_file.assert_not_called() + renamer.client.client.add_tag.assert_not_called() + assert published(plugins.bus, TorrentOrganized) == [] + events = [e for e in renamer.events if isinstance(e, RenameSkippedEvent)] + assert len(events) == 1 + assert events[0].task_id == "h1" and events[0].strategy == "custom" + assert plugins.tripped == (["ext"] if counted else []) + + +def _task(file_count: int = 1, revision: int | None = 1) -> RevisionTask: + identity = ( + None + if revision is None + else Revision(7, "episode", 1, 1, "ani", "1080p", revision) + ) + return RevisionTask("h", "name", file_count, identity) + + +@pytest.mark.parametrize( + ("owners", "incoming", "configured", "strict", "action", "reason"), + [ + ( + (_task(), _task()), + _task(), + "replace", + True, + "hold", + "canonical path has more than one downloader owner", + ), + ( + (_task(file_count=2),), + _task(), + "replace", + True, + "hold", + "automatic replacement requires two single-file torrents", + ), + ( + (_task(revision=None),), + _task(), + "replace", + True, + "hold", + "revision identity is incomplete", + ), + ( + (_task(),), + _task(), + "replace", + False, + "hold", + "existing and incoming releases are not a strict revision upgrade", + ), + ((_task(),), _task(), "hold", True, "hold", "revision conflict policy is hold"), + ((_task(),), _task(), "replace", True, "replace", ""), + ], +) +def test_core_conflict_policy_decide_matches_3x_reasons( + owners, incoming, configured, strict, action, reason +): + decision = CoreConflictPolicy().decide( + ConflictRequest("t.mkv", incoming, owners, configured, strict) + ) + assert (decision.action, decision.reason) == (action, reason) diff --git a/backend/src/test/test_plugin_extensions.py b/backend/src/test/test_plugin_extensions.py index 8e122620..fd553a01 100644 --- a/backend/src/test/test_plugin_extensions.py +++ b/backend/src/test/test_plugin_extensions.py @@ -525,6 +525,7 @@ class TestPluginsApi: "notifier": [], "search_site": [], "metadata_provider": [], + "rename_strategy": [], } def test_config_get_masks_plugin_secrets(self, authed_client, plugin_ctx): diff --git a/backend/src/test/test_renamer.py b/backend/src/test/test_renamer.py index 725e8ddd..49a4e6e8 100644 --- a/backend/src/test/test_renamer.py +++ b/backend/src/test/test_renamer.py @@ -79,7 +79,7 @@ class TestGenPath: language="zh", suffix=".ass", ) - result = Renamer.gen_path(sub, "Bangumi", method="subtitle_pn") + result = Renamer.gen_path(sub, "Bangumi", method="pn") assert result == "My Anime S01E03.zh.ass" def test_subtitle_advance_method(self): @@ -92,7 +92,7 @@ class TestGenPath: language="zh-tw", suffix=".srt", ) - result = Renamer.gen_path(sub, "Bangumi Name", method="subtitle_advance") + result = Renamer.gen_path(sub, "Bangumi Name", method="advance") assert result == "Bangumi Name S02E07.zh-tw.srt" def test_zero_padding_single_digit(self): @@ -215,7 +215,7 @@ class TestGenPathMovie: suffix=".ass", episode_type="movie", ) - result = Renamer.gen_path(sub, "天气之子 (2019)", method="subtitle_advance") + result = Renamer.gen_path(sub, "天气之子 (2019)", method="advance") assert result == "天气之子 (2019).zh.ass" def test_gen_path_movie_group_tag_enabled_keeps_clean_name(self): @@ -296,7 +296,7 @@ class TestGenPathGroupTagStability: suffix=".ass", ) with patch.object(settings.bangumi_manage, "group_tag", True): - result = Renamer.gen_path(sub, "Bangumi Name", method="subtitle_pn") + result = Renamer.gen_path(sub, "Bangumi Name", method="pn") assert result == "My Anime S01E05.zh.ass" def test_gen_path_group_tag_disabled_no_prefix(self): @@ -1686,9 +1686,7 @@ class TestGenPathWithOffsets: language="zh", suffix=".ass", ) - result = Renamer.gen_path( - sub, "Bangumi", method="subtitle_pn", episode_offset=-12 - ) + result = Renamer.gen_path(sub, "Bangumi", method="pn", episode_offset=-12) assert "E13" in result # 25 - 12 = 13 def test_offset_none_method_unchanged(self):