mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-09 09:26:55 +08:00
feat(download): recognize and organize existing tasks
This commit is contained in:
yanjunpu
@@ -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))
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
+53
-15
@@ -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
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user