From 9a74802d2b802e36ad7471bf02b2fa5bd0025fa9 Mon Sep 17 00:00:00 2001 From: yanjunpu <541726449@qq.com> Date: Mon, 7 Sep 2026 15:52:11 +0800 Subject: [PATCH] feat(download): recognize and organize existing tasks --- app/api/endpoints/download.py | 24 +- .../download/source_organization.py | 306 ++++++++++++++++++ app/schemas/download.py | 68 +++- tests/test_download_source_organization.py | 218 +++++++++++++ 4 files changed, 584 insertions(+), 32 deletions(-) create mode 100644 app/application/download/source_organization.py create mode 100644 tests/test_download_source_organization.py diff --git a/app/api/endpoints/download.py b/app/api/endpoints/download.py index 118ea648e..3660f5d55 100644 --- a/app/api/endpoints/download.py +++ b/app/api/endpoints/download.py @@ -4,7 +4,7 @@ import anyio from fastapi import Body, Depends from app.adapters.web.security.access import verify_token -from app.api.dependencies.auth import get_current_active_user +from app.api.dependencies.auth import get_current_active_user, get_current_active_manage_user from app.api.dependencies.site import get_site_sync_query_service from app.api.principal import ApiPrincipal from app.api.response import ( @@ -14,7 +14,7 @@ from app.api.response import ( ) from app.application.configuration import get_configured_system_config from app.application.directory import DirectoryHelper -from app.application.download.classification import DownloadSourceClassificationService +from app.application.download.source_organization import organize_existing_source from app.application.download.tasks import DownloadTaskMutationService from app.application.security.url import SecurityUtils from app.application.site.query import ( @@ -401,31 +401,21 @@ async def update_task( ) -@router.post( # type: ignore[misc] +@router.post( "/{hashString}/classify-source", - summary="按媒体类别重新定位资源目录", + summary="识别并归类已有下载任务", response_model=_SchemaResponse[_SchemaDownloadSourceClassificationData], ) async def classify_source( hashString: str, payload: _SchemaDownloadSourceClassificationRequest, - _: ApiPrincipal = Depends(get_current_active_user), + _: ApiPrincipal = Depends(get_current_active_manage_user), ) -> _SchemaResponse[Any]: - """预览或通过下载器执行已有任务的资源目录分类。""" + """复用媒体识别链生成资源目录和根目录名，确认后仅通过下载器执行。""" chain = DownloadChain() - service = DownloadSourceClassificationService( - list_torrents=chain.list_torrents, - get_history_by_hash=chain.download_history_repository.get_by_hash, - update_torrent=chain.update_torrent, - ) try: data = await anyio.to_thread.run_sync( - lambda: service.plan( - hash_value=hashString, - downloader=payload.downloader, - execute=payload.execute, - media_category=payload.media_category, - ) + lambda: organize_existing_source(hashString, payload, chain, MediaChain()) ) except ValueError as error: return _SchemaResponse(success=False, message=str(error)) diff --git a/app/application/download/source_organization.py b/app/application/download/source_organization.py new file mode 100644 index 000000000..d3ea7114f --- /dev/null +++ b/app/application/download/source_organization.py @@ -0,0 +1,306 @@ +"""下载器已有任务的媒体识别、资源归类与根目录重命名。""" + +import re +from pathlib import PurePosixPath +from typing import Any + +from app.application.configuration import get_configured_system_config +from app.application.directory import DirectoryHelper, validate_download_save_path +from app.domain.meta.metabase import MetaBase +from app.domain.meta.metamusic import MetaMusic +from app.domain.metainfo import MetaInfo +from app.schemas.types import MediaSource, MediaType, SystemConfigKey + +_INVALID_NAME = re.compile(r'[<>:"/\\|?*\x00-\x1f]') +_MUSIC_PRIMARY_TYPES = { + "album": "Album", + "ep": "EP", + "single": "Single", + "broadcast": "Broadcast", + "other": "Other", +} + + +def _safe_relative_name(value: Any, *, label: str) -> str: + """把识别结果规范为单层路径名，避免展示分隔符变成真实目录。""" + text = _INVALID_NAME.sub(" - ", str(value or "").strip()) + text = " ".join(text.split()).strip(" .") + if not text or text in (".", ".."): + raise ValueError(f"{label}为空或无法生成安全名称") + return text + + +def _normalize_music_category(media: Any) -> tuple[str, list[str]]: + """音乐目录只使用主类型，副类型只作识别信息展示。""" + primary = str(getattr(media, "album_type", None) or "").strip() + if not primary: + primary = str(getattr(media, "category", None) or "").split("/")[0].strip() + if not primary and str(getattr(media, "music_type", None) or "").casefold() == "recording": + primary = "Single" + primary = _MUSIC_PRIMARY_TYPES.get(primary.casefold(), primary) + secondary = [ + str(item).strip() + for item in (getattr(media, "secondary_types", None) or []) + if str(item).strip() and str(item).strip() != primary + ] + return _safe_relative_name(primary, label="音乐主类型"), secondary + + +def _resolve_media(request: Any, history: Any, torrent: Any, media_chain: Any) -> tuple[MetaBase, Any]: + """复用 MoviePilot 的媒体识别链，支持自动识别和显式原生媒体 ID。""" + raw_type = request.type_name or history.type + try: + media_type = MediaType(raw_type) if raw_type else None + except ValueError as error: + raise ValueError("下载历史缺少有效媒体类型，请手动选择") from error + title = str(torrent.title or history.torrent_name or history.title or "").strip() + if not title: + raise ValueError("任务缺少可用于识别的名称") + is_music = media_type == MediaType.MUSIC + metainfo = ( + MetaMusic.parse_query(title) + if is_music + else MetaInfo(title=title, subtitle=history.torrent_description) + ) + source = request.media_source or getattr(history, "media_source", None) + media_id = request.media_id or (getattr(history, "media_id", None) if source else None) + music_type = ( + request.music_type.value + if getattr(request.music_type, "value", None) + else request.music_type or getattr(history, "music_type", None) or ("album" if is_music else None) + ) + # 媒体链在未显式给出音乐源时会按单曲路由；资源根目录默认按专辑识别。 + if is_music and not source: + source = MediaSource.MusicBrainz + if source and media_id: + media = media_chain.recognize_media( + meta=metainfo, + mtype=media_type, + media_source=source, + media_id=media_id, + episode_group=request.episode_group or getattr(history, "episode_group", None), + music_type=music_type, + ) + else: + media = media_chain.recognize_by_meta( + metainfo, + mtype=media_type, + media_source=source, + episode_group=request.episode_group or getattr(history, "episode_group", None), + obtain_images=False, + music_type=music_type, + ) + if media is None: + raise ValueError("无法识别媒体信息，可搜索并指定媒体 ID，或改用手动指定目录") + return metainfo, media + + +def _download_root(current: PurePosixPath, media_type: str, category: str) -> tuple[Any, PurePosixPath]: + """按媒体类型与主类别选择资源目录，优先保持在当前配置根内。""" + candidates = [] + for directory in DirectoryHelper().get_download_dirs(): + if directory.storage != "local" or not directory.download_path: + continue + root = PurePosixPath(directory.download_path) + if not root.is_absolute() or ".." in root.parts: + continue + if directory.media_type and directory.media_type != media_type: + continue + if directory.media_category and directory.media_category != category: + continue + candidates.append((int(current.is_relative_to(root)), -directory.priority, len(root.parts), directory, root)) + if not candidates: + raise ValueError("没有找到匹配识别结果的本地资源目录") + _, _, _, directory, root = max(candidates, key=lambda item: item[:3]) + target = root + if not directory.media_type and directory.download_type_folder: + target /= media_type + if not directory.media_category and directory.download_category_folder: + target /= category + return directory, target + + +def _manual_target(value: str) -> PurePosixPath: + """校验手动目录是已配置资源目录本身或其子目录。""" + validated = validate_download_save_path(value) + target = PurePosixPath(validated) + if not target.is_absolute() or ".." in target.parts: + raise ValueError("手动目标路径无效") + return target + + +def _root_name(media: Any) -> str: + """生成跨电影、电视剧和音乐通用的规范任务根目录名。""" + if getattr(media, "type", None) == MediaType.MUSIC: + title = getattr(media, "album", None) or getattr(media, "title", None) + artist = getattr(media, "album_artist", None) or getattr(media, "artist", None) + base = " - ".join(str(part).strip() for part in (artist, title) if str(part or "").strip()) + else: + base = str(getattr(media, "title", None) or "").strip() + year = str(getattr(media, "year", None) or "").strip() + if year and f"({year})" not in base: + base = f"{base} ({year})" + return _safe_relative_name(base, label="规范目录名") + + +def _downloader_kind(name: str) -> str | None: + """返回已配置下载器类型，用于在预览阶段显式限定重命名能力。""" + downloaders = get_configured_system_config().get(SystemConfigKey.Downloaders) or [] + match = next((item for item in downloaders if item.get("name") == name), None) + return str(match.get("type") or "").casefold() if match else None + + +def _rename_qb_root(chain: Any, downloader: str, hash_value: str, old_name: str, new_name: str) -> bool: + """通过已运行的 qBittorrent 模块调用官方 renameFolder API。 + + 此变更由下载器维护任务与文件的对应关系，不直接操作文件系统。 + """ + module = chain.modulemanager.get_running_module("QbittorrentModule") + server = module.get_instance(downloader) if module else None + client = getattr(server, "qbc", None) + if client is None: + return False + try: + client.torrents_rename_folder( + torrent_hash=hash_value, + old_path=old_name, + new_path=new_name, + ) + return True + except Exception: + return False + + +def organize_existing_source(hash_value: str, request: Any, chain: Any, media_chain: Any) -> dict[str, Any]: + """生成可重放的预览计划，确认后仅通过下载器修改任务路径。""" + if not re.fullmatch(r"[0-9a-fA-F]{40}", hash_value): + raise ValueError("hash 格式无效") + history = chain.download_history_repository.get_by_hash(hash_value) + if history is None: + raise ValueError("未找到该任务的下载历史") + downloader = request.downloader or history.downloader + matches = [ + item for item in (chain.list_torrents( + hashs=[hash_value], downloader=downloader, include_all_tags=True + ) or []) + if str(item.hash or "").casefold() == hash_value.casefold() + ] + if len(matches) != 1: + raise ValueError("无法唯一定位下载任务，请检查任务是否仍在下载器中") + torrent = matches[0] + downloader = downloader or torrent.downloader + if not downloader: + raise ValueError("无法确定下载器实例") + current = PurePosixPath(str(torrent.save_path or "").strip()) + if not current.is_absolute() or ".." in current.parts: + raise ValueError("下载器返回的保存路径无效") + + media = None + category = None + secondary_categories: list[str] = [] + if request.mode == "recognize" or request.smart_rename: + _, media = _resolve_media(request, history, torrent, media_chain) + media_type = media.type.value + if media.type == MediaType.MUSIC: + category, secondary_categories = _normalize_music_category(media) + else: + category = _safe_relative_name( + request.media_category or getattr(media, "category", None) or history.media_category, + label="媒体类别", + ) + else: + media_type = request.type_name or history.type + + if request.mode == "manual": + target = _manual_target(str(request.target_path)) + else: + _, target = _download_root(current, media_type, category) + target = PurePosixPath(validate_download_save_path(target.as_posix())) + + content_text = str(torrent.content_path or torrent.path or "").strip() + content = PurePosixPath(content_text) if content_text else None + current_root_name = None + if content and content.is_absolute() and content.is_relative_to(current): + relative_content = content.relative_to(current) + if len(relative_content.parts) == 1 and not content.suffix: + current_root_name = relative_content.name + rename_supported = _downloader_kind(downloader) == "qbittorrent" and current_root_name is not None + proposed_root_name = _root_name(media) if request.smart_rename and media else current_root_name + rename_required = bool( + request.smart_rename + and rename_supported + and proposed_root_name + and proposed_root_name != current_root_name + ) + if request.smart_rename and not rename_supported: + raise ValueError("当前任务不是 qBittorrent 的单根目录任务，无法安全智能重命名") + + target_text = target.as_posix() + changed = current.as_posix() != target_text or rename_required + relocated = False + renamed = False + if request.execute and changed: + expected = ( + request.expected_current_path == current.as_posix() + and request.expected_target_path == target_text + and request.expected_content_path == (content.as_posix() if content else None) + and request.expected_root_name == proposed_root_name + ) + if not expected: + raise ValueError("任务路径或识别计划已变化，请重新预览后确认") + if rename_required: + renamed = _rename_qb_root( + chain, + downloader, + hash_value, + current_root_name, + proposed_root_name, + ) + if not renamed: + raise ValueError("下载器未接受根目录重命名") + if current.as_posix() != target_text: + result = chain.update_torrent( + hash_string=hash_value, + downloader=downloader, + save_path=target_text, + ) or {} + relocated = bool(result.get("save_path")) + if not relocated: + if renamed: + rolled_back = _rename_qb_root( + chain, + downloader, + hash_value, + proposed_root_name, + current_root_name, + ) + if not rolled_back: + raise ValueError("保存位置修改失败，且根目录名无法自动回滚，请检查下载器") + renamed = False + raise ValueError("下载器未接受保存位置修改，根目录重命名已回滚") + + media_source = getattr(media, "media_source", None) + return { + "hash": hash_value, + "downloader": downloader, + "mode": request.mode, + "recognized": media is not None, + "media_type": getattr(getattr(media, "type", None), "value", media_type), + "media_source": getattr(media_source, "value", media_source), + "media_id": str(getattr(media, "media_id", None) or "") or None, + "title": getattr(media, "album", None) or getattr(media, "title", None), + "year": str(getattr(media, "year", None) or "") or None, + "current_save_path": current.as_posix(), + "target_save_path": target_text, + "current_content_path": content.as_posix() if content else None, + "category": category, + "secondary_categories": secondary_categories, + "current_root_name": current_root_name, + "proposed_root_name": proposed_root_name, + "rename_supported": rename_supported, + "rename_required": rename_required, + "changed": changed, + "executed": bool(request.execute and changed), + "relocated": relocated, + "renamed": renamed, + } diff --git a/app/schemas/download.py b/app/schemas/download.py index f06398d6d..ad4aad927 100644 --- a/app/schemas/download.py +++ b/app/schemas/download.py @@ -1,6 +1,8 @@ from typing import Literal, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator + +from app.schemas.types import MediaSource, MusicTargetEntityType class DownloadTask(BaseModel): @@ -76,21 +78,57 @@ class DownloadTaskUpdateData(BaseModel): # type: ignore[misc] results: list[DownloadTaskMutationResult] = Field(default_factory=list, description="各修改动作结果") -class DownloadSourceClassificationRequest(BaseModel): # type: ignore[misc] - """已有下载任务的资源目录分类请求。""" +class DownloadSourceClassificationRequest(BaseModel): + """已有任务的识别、归类与种子根目录重命名请求。""" - downloader: Optional[str] = Field(default=None, description="下载器实例") - execute: bool = Field(default=False, description="是否执行下载器位置移动") - media_category: Optional[str] = Field(default=None, description="可选的手动媒体分类路径") + downloader: Optional[str] = None + execute: bool = False + mode: Literal["recognize", "manual"] = "recognize" + target_path: Optional[str] = None + type_name: Optional[Literal["电影", "电视剧", "音乐"]] = None + media_source: Optional[MediaSource] = None + media_id: Optional[str] = None + music_type: Optional[MusicTargetEntityType] = None + episode_group: Optional[str] = None + media_category: Optional[str] = None + smart_rename: bool = True + expected_current_path: Optional[str] = None + expected_target_path: Optional[str] = None + expected_content_path: Optional[str] = None + expected_root_name: Optional[str] = None + + @model_validator(mode="after") + def validate_mode_and_identity(self): + """手动模式必须给出目录，ID 不能脱离其所属数据源。""" + if self.mode == "manual" and not str(self.target_path or "").strip(): + raise ValueError("手动指定目录模式必须填写目标路径") + if self.media_source is None and str(self.media_id or "").strip(): + raise ValueError("填写媒体 ID 时必须选择数据源") + return self -class DownloadSourceClassificationData(BaseModel): # type: ignore[misc] - """资源目录分类预览或执行结果。""" +class DownloadSourceClassificationData(BaseModel): + """识别与资源目录变更计划，包含可审计的执行结果。""" - hash: str = Field(description="下载任务 Hash") - downloader: str = Field(description="实际使用的下载器实例") - current_save_path: str = Field(description="当前保存目录") - target_save_path: str = Field(description="按类别分类后的目标目录") - category: str = Field(description="命中的媒体分类路径") - changed: bool = Field(description="当前目录是否需要变更") - executed: bool = Field(description="是否已请求下载器移动") + hash: str + downloader: str + mode: Literal["recognize", "manual"] + recognized: bool + media_type: Optional[str] = None + media_source: Optional[str] = None + media_id: Optional[str] = None + title: Optional[str] = None + year: Optional[str] = None + current_save_path: str + target_save_path: str + current_content_path: Optional[str] = None + category: Optional[str] = None + secondary_categories: list[str] = Field(default_factory=list) + current_root_name: Optional[str] = None + proposed_root_name: Optional[str] = None + rename_supported: bool = False + rename_required: bool = False + changed: bool + executed: bool + relocated: bool = False + renamed: bool = False diff --git a/tests/test_download_source_organization.py b/tests/test_download_source_organization.py new file mode 100644 index 000000000..dff19d718 --- /dev/null +++ b/tests/test_download_source_organization.py @@ -0,0 +1,218 @@ +"""Isolated safety tests for the deployment-only source organization service.""" + +import importlib.util +import sys +import unittest +from enum import Enum +from pathlib import Path +from types import ModuleType, SimpleNamespace as NS +from unittest.mock import Mock, patch + + +class MediaType(str, Enum): + MUSIC = "音乐" + MOVIE = "电影" + TV = "电视剧" + + +class SystemConfigKey(str, Enum): + Downloaders = "Downloaders" + + +class MediaSource(str, Enum): + MusicBrainz = "musicbrainz" + + +directory_module = ModuleType("app.application.directory") +directory_module.DirectoryHelper = Mock() +directory_module.validate_download_save_path = lambda value: str(value) +configuration_module = ModuleType("app.application.configuration") +configuration_module.get_configured_system_config = lambda: { + SystemConfigKey.Downloaders: [{"name": "qb", "type": "qbittorrent"}] +} +metabase_module = ModuleType("app.domain.meta.metabase") +metabase_module.MetaBase = object +metamusic_module = ModuleType("app.domain.meta.metamusic") +metamusic_module.MetaMusic = NS(parse_query=lambda _value: NS()) +metainfo_module = ModuleType("app.domain.metainfo") +metainfo_module.MetaInfo = lambda **_kwargs: NS() +types_module = ModuleType("app.schemas.types") +types_module.MediaSource = MediaSource +types_module.MediaType = MediaType +types_module.SystemConfigKey = SystemConfigKey + +with patch.dict( + sys.modules, + { + "app.application.configuration": configuration_module, + "app.application.directory": directory_module, + "app.domain.meta.metabase": metabase_module, + "app.domain.meta.metamusic": metamusic_module, + "app.domain.metainfo": metainfo_module, + "app.schemas.types": types_module, + }, +): + spec = importlib.util.spec_from_file_location( + "source_organization", + Path(__file__).parent.parent / "app/application/download/source_organization.py", + ) + organization = importlib.util.module_from_spec(spec) + spec.loader.exec_module(organization) + + +class SourceOrganizationTests(unittest.TestCase): + def setUp(self): + self.hash_value = "a" * 40 + self.request = NS( + downloader=None, + execute=False, + mode="recognize", + target_path=None, + type_name="音乐", + media_source=None, + media_id=None, + music_type="album", + episode_group=None, + media_category=None, + smart_rename=True, + expected_current_path=None, + expected_target_path=None, + expected_content_path=None, + expected_root_name=None, + ) + self.history = NS( + downloader="qb", + type="音乐", + media_source=None, + media_id=None, + music_type=None, + media_category=None, + episode_group=None, + torrent_name="Karen Mok - Loving Gaze 2002", + torrent_description=None, + title="含情脉脉", + ) + self.torrent = NS( + hash=self.hash_value, + downloader="qb", + title="Karen Mok - Loving Gaze 2002", + save_path="/volume1/UT/Musics", + content_path="/volume1/UT/Musics/Karen Mok - Loving Gaze 2002 FLAC", + path=None, + ) + self.media = NS( + type=MediaType.MUSIC, + album_type="Album", + secondary_types=["Compilation"], + album="含情脉脉", + title="含情脉脉", + album_artist="莫文蔚", + artist="莫文蔚", + year="2002", + media_source="musicbrainz", + media_id="release-group-id", + ) + self.media_chain = Mock() + self.media_chain.recognize_by_meta.return_value = self.media + self.qbc = Mock() + module = NS(get_instance=lambda _name: NS(qbc=self.qbc)) + self.chain = Mock() + self.chain.download_history_repository.get_by_hash.return_value = self.history + self.chain.list_torrents.return_value = [self.torrent] + self.chain.modulemanager.get_running_module.return_value = module + self.chain.update_torrent.return_value = {"save_path": True} + directory_module.DirectoryHelper.return_value.get_download_dirs.return_value = [ + NS( + storage="local", + download_path="/volume1/UT/Musics", + media_type="音乐", + media_category="", + download_type_folder=False, + download_category_folder=True, + priority=2, + ) + ] + + def preview(self): + return organization.organize_existing_source( + self.hash_value, + self.request, + self.chain, + self.media_chain, + ) + + def test_music_secondary_type_never_becomes_a_path_segment(self): + result = self.preview() + self.assertEqual(result["category"], "Album") + self.assertEqual(result["secondary_categories"], ["Compilation"]) + self.assertEqual(result["target_save_path"], "/volume1/UT/Musics/Album") + self.assertNotIn("Compilation", result["target_save_path"]) + + def test_preview_is_read_only_and_includes_qb_root_rename(self): + result = self.preview() + self.assertEqual(result["proposed_root_name"], "莫文蔚 - 含情脉脉 (2002)") + self.assertTrue(result["rename_required"]) + self.chain.update_torrent.assert_not_called() + self.qbc.torrents_rename_folder.assert_not_called() + + def test_execution_requires_exact_preview_replay(self): + self.request.execute = True + with self.assertRaisesRegex(ValueError, "重新预览"): + self.preview() + self.chain.update_torrent.assert_not_called() + + def test_confirmed_execution_uses_qb_rename_then_set_location(self): + plan = self.preview() + self.request.execute = True + self.request.expected_current_path = plan["current_save_path"] + self.request.expected_target_path = plan["target_save_path"] + self.request.expected_content_path = plan["current_content_path"] + self.request.expected_root_name = plan["proposed_root_name"] + result = self.preview() + self.assertTrue(result["renamed"]) + self.assertTrue(result["relocated"]) + self.qbc.torrents_rename_folder.assert_called_once_with( + torrent_hash=self.hash_value, + old_path="Karen Mok - Loving Gaze 2002 FLAC", + new_path="莫文蔚 - 含情脉脉 (2002)", + ) + self.chain.update_torrent.assert_called_once_with( + hash_string=self.hash_value, + downloader="qb", + save_path="/volume1/UT/Musics/Album", + ) + + def test_manual_directory_without_rename_skips_recognition(self): + self.request.mode = "manual" + self.request.target_path = "/volume1/UT/Musics/EP" + self.request.smart_rename = False + result = self.preview() + self.assertFalse(result["recognized"]) + self.assertEqual(result["target_save_path"], "/volume1/UT/Musics/EP") + self.media_chain.recognize_by_meta.assert_not_called() + + def test_failed_relocation_rolls_back_root_rename(self): + plan = self.preview() + self.request.execute = True + self.request.expected_current_path = plan["current_save_path"] + self.request.expected_target_path = plan["target_save_path"] + self.request.expected_content_path = plan["current_content_path"] + self.request.expected_root_name = plan["proposed_root_name"] + self.chain.update_torrent.return_value = {"save_path": False} + with self.assertRaisesRegex(ValueError, "已回滚"): + self.preview() + self.assertEqual(self.qbc.torrents_rename_folder.call_count, 2) + self.qbc.torrents_rename_folder.assert_called_with( + torrent_hash=self.hash_value, + old_path="莫文蔚 - 含情脉脉 (2002)", + new_path="Karen Mok - Loving Gaze 2002 FLAC", + ) + + def test_smart_rename_rejects_multi_root_tasks(self): + self.torrent.content_path = self.torrent.save_path + with self.assertRaisesRegex(ValueError, "无法安全智能重命名"): + self.preview() + + +if __name__ == "__main__": + unittest.main()