mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-01 21:07:01 +08:00
feat(music): use original torrent evidence in album recognition
This commit is contained in:
jxxghp
@@ -1,5 +1,6 @@
|
||||
"""专辑目录扫描、曲目对齐与缓存编排 owner。"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
@@ -28,9 +29,47 @@ def _album_directory_cache_key(
|
||||
directory: Path,
|
||||
regions: tuple[str, ...],
|
||||
scripts: tuple[str, ...],
|
||||
contextual_meta: Optional[MetaMusic] = None,
|
||||
) -> str:
|
||||
"""返回包含发行偏好的目录键，避免不同手动选择错误共享结果。"""
|
||||
return "|".join((os.path.abspath(directory), ",".join(regions), ",".join(scripts)))
|
||||
"""将发行偏好及资源证据纳入缓存键，同目录更换种子线索时重新识别。"""
|
||||
evidence = {
|
||||
key: getattr(contextual_meta, key, None)
|
||||
for key in ("album", "artists", "album_artist", "year", "version")
|
||||
} if contextual_meta else None
|
||||
if evidence and evidence["album_artist"]:
|
||||
evidence["artists"] = [evidence["album_artist"]]
|
||||
if evidence and not any(evidence.values()):
|
||||
evidence = None
|
||||
return json.dumps([os.path.abspath(directory), regions, scripts, evidence], ensure_ascii=False, sort_keys=True)
|
||||
|
||||
|
||||
def _album_context_with_resource(
|
||||
directory: Path,
|
||||
metas: list[MetaMusic],
|
||||
contextual_meta: Optional[MetaMusic],
|
||||
) -> MetaMusic:
|
||||
"""以文件自身专辑线索为先，用绑定到本次文件的资源证据补足目录查询。"""
|
||||
album_meta = MetaMusic.from_album_context(directory.name, metas)
|
||||
if not contextual_meta:
|
||||
return album_meta
|
||||
if contextual_meta.album and not any(meta.album for meta in metas):
|
||||
album_meta.album = contextual_meta.album
|
||||
album_meta.title = contextual_meta.album
|
||||
if not album_meta.artists:
|
||||
album_meta.artists = (
|
||||
[contextual_meta.album_artist] if contextual_meta.album_artist
|
||||
else list(contextual_meta.artists)
|
||||
)
|
||||
if not album_meta.album_artist and contextual_meta.album_artist:
|
||||
album_meta.album_artist = contextual_meta.album_artist
|
||||
for key in ("year", "version"):
|
||||
local_values = {getattr(meta, key) for meta in metas if getattr(meta, key)}
|
||||
if not getattr(album_meta, key):
|
||||
if len(local_values) == 1:
|
||||
setattr(album_meta, key, next(iter(local_values)))
|
||||
elif not local_values:
|
||||
setattr(album_meta, key, getattr(contextual_meta, key))
|
||||
return album_meta
|
||||
|
||||
|
||||
class MediaAlbumOwner(_MediaOwnerBase):
|
||||
@@ -203,10 +242,11 @@ class MediaAlbumOwner(_MediaOwnerBase):
|
||||
files: list[Path],
|
||||
music_release_regions: Optional[tuple[str, ...]] = None,
|
||||
music_release_scripts: Optional[tuple[str, ...]] = None,
|
||||
contextual_meta: Optional[MetaMusic] = None,
|
||||
) -> dict[str, MusicInfo]:
|
||||
"""同步汇总本地专辑证据并委托 MusicBrainz 来源链匹配。"""
|
||||
metas = AudioMetadataHelper.read_many(files)
|
||||
album_meta = MetaMusic.from_album_context(directory.name, metas)
|
||||
album_meta = _album_context_with_resource(directory, metas, contextual_meta)
|
||||
regions, scripts = self._music_release_preferences(
|
||||
list(music_release_regions) if music_release_regions is not None else None,
|
||||
list(music_release_scripts) if music_release_scripts is not None else None,
|
||||
@@ -227,10 +267,11 @@ class MediaAlbumOwner(_MediaOwnerBase):
|
||||
files: list[Path],
|
||||
music_release_regions: Optional[tuple[str, ...]] = None,
|
||||
music_release_scripts: Optional[tuple[str, ...]] = None,
|
||||
contextual_meta: Optional[MetaMusic] = None,
|
||||
) -> dict[str, MusicInfo]:
|
||||
"""异步汇总本地专辑证据并委托 MusicBrainz 来源链匹配。"""
|
||||
metas = await run_in_threadpool(AudioMetadataHelper.read_many, files)
|
||||
album_meta = MetaMusic.from_album_context(directory.name, metas)
|
||||
album_meta = _album_context_with_resource(directory, metas, contextual_meta)
|
||||
regions, scripts = self._music_release_preferences(
|
||||
list(music_release_regions) if music_release_regions is not None else None,
|
||||
list(music_release_scripts) if music_release_scripts is not None else None,
|
||||
@@ -250,6 +291,7 @@ class MediaAlbumOwner(_MediaOwnerBase):
|
||||
path: Union[str, Path],
|
||||
music_release_regions: Optional[list[str]] = None,
|
||||
music_release_scripts: Optional[list[str]] = None,
|
||||
contextual_meta: Optional[MetaMusic] = None,
|
||||
) -> dict[str, MusicInfo]:
|
||||
"""按目录级线索批量识别整张专辑并返回文件到曲目的映射。"""
|
||||
directory = Path(path)
|
||||
@@ -263,12 +305,14 @@ class MediaAlbumOwner(_MediaOwnerBase):
|
||||
music_release_scripts,
|
||||
)
|
||||
custom_preference = music_release_regions is not None or music_release_scripts is not None
|
||||
key = _album_directory_cache_key(directory, regions, scripts)
|
||||
key = _album_directory_cache_key(directory, regions, scripts, contextual_meta)
|
||||
signature = self._album_directory_signature(directory, files)
|
||||
matched = self._album_dir_cache.resolve(
|
||||
key,
|
||||
signature,
|
||||
lambda: (
|
||||
self._match_music_album_directory(directory, files, regions, scripts, contextual_meta)
|
||||
if contextual_meta is not None else
|
||||
self._match_music_album_directory(directory, files, regions, scripts)
|
||||
if custom_preference
|
||||
else self._match_music_album_directory(directory, files)
|
||||
@@ -288,6 +332,7 @@ class MediaAlbumOwner(_MediaOwnerBase):
|
||||
path: Union[str, Path],
|
||||
music_release_regions: Optional[list[str]] = None,
|
||||
music_release_scripts: Optional[list[str]] = None,
|
||||
contextual_meta: Optional[MetaMusic] = None,
|
||||
) -> dict[str, MusicInfo]:
|
||||
"""异步按目录级线索批量识别整张专辑。"""
|
||||
directory = Path(path)
|
||||
@@ -301,12 +346,14 @@ class MediaAlbumOwner(_MediaOwnerBase):
|
||||
music_release_scripts,
|
||||
)
|
||||
custom_preference = music_release_regions is not None or music_release_scripts is not None
|
||||
key = _album_directory_cache_key(directory, regions, scripts)
|
||||
key = _album_directory_cache_key(directory, regions, scripts, contextual_meta)
|
||||
signature = self._album_directory_signature(directory, files)
|
||||
matched = await self._album_dir_cache.async_resolve(
|
||||
key,
|
||||
signature,
|
||||
lambda: (
|
||||
self._async_match_music_album_directory(directory, files, regions, scripts, contextual_meta)
|
||||
if contextual_meta is not None else
|
||||
self._async_match_music_album_directory(directory, files, regions, scripts)
|
||||
if custom_preference
|
||||
else self._async_match_music_album_directory(directory, files)
|
||||
|
||||
@@ -171,6 +171,7 @@ if TYPE_CHECKING:
|
||||
path: Union[str, Path],
|
||||
music_release_regions: Optional[list[str]] = None,
|
||||
music_release_scripts: Optional[list[str]] = None,
|
||||
contextual_meta: Optional[MetaMusic] = None,
|
||||
) -> dict[str, MusicInfo]:
|
||||
"""同步识别音乐专辑目录。"""
|
||||
...
|
||||
@@ -190,6 +191,7 @@ if TYPE_CHECKING:
|
||||
path: Union[str, Path],
|
||||
music_release_regions: Optional[list[str]] = None,
|
||||
music_release_scripts: Optional[list[str]] = None,
|
||||
contextual_meta: Optional[MetaMusic] = None,
|
||||
) -> dict[str, MusicInfo]:
|
||||
"""异步识别音乐专辑目录。"""
|
||||
...
|
||||
|
||||
@@ -19,7 +19,11 @@ from app.chain._contracts import TransferMixinHost
|
||||
from app.chain.media import MediaChain
|
||||
from app.chain.storage import StorageChain
|
||||
from app.chain.transfer.contract import _TransferOwnerBase
|
||||
from app.chain.transfer.music import prepare_music_batch_context, resolve_music_batch_file_context
|
||||
from app.chain.transfer.music import (
|
||||
prepare_music_batch_context,
|
||||
resolve_music_batch_file_context,
|
||||
restore_music_resource_meta,
|
||||
)
|
||||
from app.domain.context import MediaInfo, MusicAlbumInfo, MusicInfo
|
||||
from app.domain.media import normalize_music_type
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
@@ -186,14 +190,12 @@ class _MusicFileFilterBase(_TransferOwnerBase):
|
||||
if file_meta.media_id or getattr(file_item, "storage", "local") != "local":
|
||||
return file_meta, None
|
||||
try:
|
||||
if music_release_regions is None and music_release_scripts is None:
|
||||
matched = MediaChain().recognize_music_album_directory(file_path.parent)
|
||||
else:
|
||||
matched = MediaChain().recognize_music_album_directory(
|
||||
file_path.parent,
|
||||
music_release_regions=music_release_regions,
|
||||
music_release_scripts=music_release_scripts,
|
||||
)
|
||||
matched = MediaChain().recognize_music_album_directory(
|
||||
file_path.parent,
|
||||
music_release_regions=music_release_regions,
|
||||
music_release_scripts=music_release_scripts,
|
||||
contextual_meta=file_meta,
|
||||
)
|
||||
except Exception as err:
|
||||
logger.debug(f"音乐专辑目录匹配失败：{file_path} - {err}")
|
||||
return file_meta, None
|
||||
@@ -410,20 +412,28 @@ class FileFilterMixin(_TransferOwnerBase):
|
||||
|
||||
种子未提供的语义字段由历史中已选媒体补缺；实体类型和来源身份始终
|
||||
沿用已选媒体，不根据专辑名或文件曲名在单曲与专辑之间转换。
|
||||
多音轨批次误带单曲身份时只保留文件自身标签，避免把同一 recording
|
||||
身份传播到整张专辑；调用方随后可使用目录级证据重新匹配专辑。
|
||||
多音轨批次误带单曲身份时保留文件标签和独立的原始种子线索，避免
|
||||
把同一 recording 身份传播到整张专辑；随后可按目录重新匹配专辑。
|
||||
手动选中多条历史且未要求复用历史身份时，专辑身份也必须丢弃，确保
|
||||
目录级识别能够重新补齐发行版、分类和规范名称。
|
||||
"""
|
||||
note = getattr(download_history, "note", None)
|
||||
music_note = note.get("music") if isinstance(note, dict) else None
|
||||
if not isinstance(music_note, dict) or music_note.get("version") != 1:
|
||||
return None, None
|
||||
if not download_history or not MediaChain.is_audio_path(file_path):
|
||||
return None, None
|
||||
return restore_music_resource_meta(download_history, file_path), None
|
||||
try:
|
||||
saved_meta = MetaMusic.from_dict(music_note.get("meta") or {})
|
||||
saved_info = MusicInfo.from_dict(music_note.get("media") or {})
|
||||
except (TypeError, ValueError):
|
||||
return None, None
|
||||
resource_meta = (
|
||||
restore_music_resource_meta(download_history, file_path)
|
||||
if MediaChain.is_audio_path(file_path) else None
|
||||
)
|
||||
return resource_meta, None
|
||||
if not (saved_info.title or saved_meta.title or saved_meta.album):
|
||||
return restore_music_resource_meta(download_history, file_path), None
|
||||
|
||||
file_tags = MediaChain.read_path_meta(file_path)
|
||||
should_discard_identity = discard_saved_identity or (
|
||||
@@ -431,12 +441,11 @@ class FileFilterMixin(_TransferOwnerBase):
|
||||
and saved_info.music_type == MUSIC_ENTITY_RECORDING
|
||||
)
|
||||
if should_discard_identity:
|
||||
# 共享 recording 上下文可能包含错误的专辑、年份等字段；整张丢弃，
|
||||
# 只保留当前文件实际标签，目录级匹配失败时也不会回落到错误身份。
|
||||
# 丢弃选错媒体产生的快照，原始资源标题仍可独立作为无身份的查询线索。
|
||||
file_meta = deepcopy(file_tags)
|
||||
file_meta.org_string = file_path.name
|
||||
file_meta.title = file_meta.title or file_path.stem
|
||||
return file_meta, None
|
||||
return restore_music_resource_meta(download_history, file_path, file_meta), None
|
||||
|
||||
file_meta = deepcopy(saved_meta)
|
||||
# 新旧历史都可能仅在 media 中保留已选专辑；先补缺，再沿用文件标签的
|
||||
|
||||
@@ -8,6 +8,7 @@ from pathlib import Path
|
||||
from typing import Any, Optional, Tuple, Union, cast
|
||||
|
||||
from app.application.audio import AudioMetadataHelper
|
||||
from app.application.history import DownloadHistorySnapshot
|
||||
from app.chain.transfer.contract import _TransferOwnerBase
|
||||
from app.domain.context import MediaInfo, MusicInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
@@ -29,6 +30,56 @@ class MusicBatchContext:
|
||||
resolved_contexts: dict[Tuple[str, str], tuple[MetaMusic, MusicInfo]] = field(default_factory=dict)
|
||||
|
||||
|
||||
def restore_music_resource_meta(
|
||||
history: Optional[DownloadHistorySnapshot],
|
||||
path: Path,
|
||||
file_meta: Optional[MetaMusic] = None,
|
||||
) -> Optional[MetaMusic]:
|
||||
"""从原始种子主副标题补足文件的发行线索，不复用已丢弃的远端身份。
|
||||
|
||||
专辑字段只传到下载根及其碟片目录；全集或更深的独立专辑目录只继承
|
||||
可信艺人。文件曲名、曲序和已有标签不被整包标题覆盖。
|
||||
"""
|
||||
title = str(getattr(history, "torrent_name", None) or "")
|
||||
description = str(getattr(history, "torrent_description", None) or "")
|
||||
history_type = getattr(history, "type", None)
|
||||
if history_type in (MediaType.MOVIE, MediaType.TV, MediaType.MOVIE.value, MediaType.TV.value):
|
||||
return file_meta
|
||||
if not title and not description:
|
||||
return file_meta
|
||||
root_text = getattr(history, "path", None)
|
||||
root = Path(root_text) if root_text else None
|
||||
if root and root != path and not path.is_relative_to(root):
|
||||
return file_meta
|
||||
resource = MetaMusic.parse_resource(title, description)
|
||||
meta = deepcopy(file_meta) if file_meta else (
|
||||
AudioMetadataHelper.read(path) if path.is_file() else AudioMetadataHelper.read_filename(path)
|
||||
)
|
||||
parent = path.parent.parent if MetaMusic.parse_disc_dir(path.parent.name) else path.parent
|
||||
direct_release = bool(root and (root == path or root == parent))
|
||||
collection = bool(re.search(
|
||||
r"全集|专辑合集|專輯合集|\bdiscography\b|\bcomplete\s+(?:albums?|collection)\b|\d+\s*张专辑",
|
||||
f"{title} {description}", re.IGNORECASE,
|
||||
))
|
||||
collective = {music_text_key(value) for value in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人")}
|
||||
artists = [artist for artist in resource.artists if music_text_key(artist) not in collective]
|
||||
if not meta.artists and artists:
|
||||
meta.artists = list(artists)
|
||||
if not meta.album_artist and len(resource.artists) == 1:
|
||||
meta.album_artist = resource.artists[0]
|
||||
if direct_release and not collection:
|
||||
album = resource.album
|
||||
if root != path and not resource.track_number:
|
||||
album = album or resource.title
|
||||
if not meta.album and album:
|
||||
meta.album = album
|
||||
if not meta.year and resource.year:
|
||||
meta.year = resource.year
|
||||
if not meta.version and resource.version:
|
||||
meta.version = resource.version
|
||||
return meta
|
||||
|
||||
|
||||
def _music_directory_consensus(values: list[str]) -> Optional[str]:
|
||||
"""只在同目录至少两个标签高度一致时返回共识值。"""
|
||||
normalized = [
|
||||
|
||||
@@ -763,7 +763,7 @@ flowchart LR
|
||||
| 指标 | 当前值 |
|
||||
|---|---:|
|
||||
| Python 模块 | 1054 |
|
||||
| 内部导入边 | 8,955 |
|
||||
| 内部导入边 | 8,956 |
|
||||
| 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
|
||||
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
|
||||
| Direct egress | 53（债务已清零，53 条精确 containment） |
|
||||
@@ -773,7 +773,7 @@ flowchart LR
|
||||
| Model/Oper 自动事务与自建 Session | 0 |
|
||||
| 组合根外 `SystemConfigOper()` | 0 |
|
||||
|
||||
整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责；宿主依赖边现为 8,955，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。当前 `app.startup.lifecycle` 为 574、`app.factory` 为 586、`app.main` 为 588。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
|
||||
整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责；宿主依赖边现为 8,956，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。当前 `app.startup.lifecycle` 为 574、`app.factory` 为 586、`app.main` 为 588。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
|
||||
|
||||
架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
|
||||
语义门禁，`Check host architecture snapshot` 再执行快照测试及一次
|
||||
|
||||
@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
|
||||
|
||||
| 指标 | 当前值 | 解释 |
|
||||
|---|---:|---|
|
||||
| 宿主 Python 模块 / 内部依赖边 | 1054 / 8,955 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档 |
|
||||
| 宿主 Python 模块 / 内部依赖边 | 1054 / 8,956 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档 |
|
||||
| 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
|
||||
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
|
||||
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
|
||||
|
||||
@@ -1097,8 +1097,8 @@
|
||||
"runtime_only": true
|
||||
}
|
||||
},
|
||||
"edge_count": 8955,
|
||||
"edge_sha256": "b56dbef23e86e017c08b09b354d227a0c286cae52c224f797c254efa956de368",
|
||||
"edge_count": 8956,
|
||||
"edge_sha256": "b6547bb8bb60c5adc0dae7afaf2dddb11a782c3b9644b550e28945bfadbe2daf",
|
||||
"edges": [
|
||||
"app -> app.foundation",
|
||||
"app -> app.foundation.environment",
|
||||
@@ -5582,6 +5582,7 @@
|
||||
"app.chain.transfer.history -> app.schemas.workflow",
|
||||
"app.chain.transfer.music -> app.application",
|
||||
"app.chain.transfer.music -> app.application.audio",
|
||||
"app.chain.transfer.music -> app.application.history",
|
||||
"app.chain.transfer.music -> app.chain",
|
||||
"app.chain.transfer.music -> app.chain.transfer",
|
||||
"app.chain.transfer.music -> app.chain.transfer.contract",
|
||||
|
||||
197
tests/test_music_resource_context.py
Normal file
197
tests/test_music_resource_context.py
Normal file
@@ -0,0 +1,197 @@
|
||||
"""下载种子原始证据在丢弃旧身份后仍参与整专查询，并保持作用范围。"""
|
||||
|
||||
import asyncio
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
from mutagen.flac import FLAC
|
||||
|
||||
from app.application.history import DownloadHistorySnapshot
|
||||
from app.chain.media import MediaChain
|
||||
from app.chain.transfer.facade import TransferChain
|
||||
from app.domain.context import MusicAlbumInfo, MusicInfo
|
||||
from app.domain.meta.metamusic import MetaMusic
|
||||
from app.schemas.types import MediaType
|
||||
from tests.test_transfer_sync_extra_files import make_fileitem
|
||||
|
||||
|
||||
def _history(root: Path, **kwargs) -> DownloadHistorySnapshot:
|
||||
"""构造只有站点原始名称、没有已选媒体身份的下载历史。"""
|
||||
values = dict(
|
||||
id=1, path=str(root), type=MediaType.MUSIC.value, title="Unrecognized",
|
||||
torrent_name="Artist - Actual Album (2003) [FLAC]",
|
||||
torrent_description="音乐专辑", note=None,
|
||||
)
|
||||
values.update(kwargs)
|
||||
return DownloadHistorySnapshot(**values)
|
||||
|
||||
|
||||
def _audio_files(root: Path) -> list[Path]:
|
||||
"""生成无标签但具有真实时长和曲名的整专输入。"""
|
||||
root.mkdir(parents=True)
|
||||
files = [root / "01 - First Song.flac", root / "02 - Second Song.flac"]
|
||||
for path in files:
|
||||
shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", path)
|
||||
return files
|
||||
|
||||
|
||||
@pytest.mark.parametrize("note", [
|
||||
None, {"music": {"version": 99}}, {"music": {"version": 1}},
|
||||
{"music": {"version": 1, "meta": {"type": "电影"}}},
|
||||
])
|
||||
def test_missing_or_invalid_music_snapshot_keeps_torrent_evidence(tmp_path, note):
|
||||
"""旧历史没有可用音乐快照时，原始主副标题仍可补艺人、专辑和年份。"""
|
||||
root = tmp_path / "opaque"
|
||||
paths = _audio_files(root)
|
||||
history = _history(root, note=note)
|
||||
|
||||
meta, info = TransferChain._restore_music_download_context(history, paths[0])
|
||||
|
||||
assert info is None
|
||||
assert meta.title == "First Song"
|
||||
assert meta.track_number == 1
|
||||
assert meta.duration == 3
|
||||
assert meta.artists == ["Artist"]
|
||||
assert meta.album == "Actual Album"
|
||||
assert meta.year == 2003
|
||||
assert meta.media_id is None
|
||||
|
||||
|
||||
def test_discarded_wrong_identity_does_not_discard_original_resource(tmp_path):
|
||||
"""选错的媒体快照可以丢弃，原始种子不是错误身份的一部分。"""
|
||||
root = tmp_path / "opaque"
|
||||
paths = _audio_files(root)
|
||||
wrong = MusicInfo(media_source="musicbrainz", media_id="wrong", title="Wrong Song", album="Wrong Album")
|
||||
history = _history(root, note={"music": {
|
||||
"version": 1, "meta": MetaMusic.from_music_info(wrong).to_dict(), "media": wrong.to_dict(),
|
||||
}})
|
||||
|
||||
meta, info = TransferChain._restore_music_download_context(history, paths[0], discard_saved_identity=True)
|
||||
|
||||
assert info is None
|
||||
assert (meta.title, meta.album) == ("First Song", "Actual Album")
|
||||
assert meta.artists == ["Artist"]
|
||||
assert meta.media_id is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("subdir,torrent", [
|
||||
("2001 - Child Album", "Artist - Actual Album (2003) [FLAC]"),
|
||||
("", "Artist - Discography (2003) [FLAC]"),
|
||||
])
|
||||
def test_package_title_does_not_override_independent_albums(tmp_path, subdir, torrent):
|
||||
"""全集和独立子专辑只接收艺人，不能被顶层下载标题改成同一个专辑。"""
|
||||
root = tmp_path / "opaque"
|
||||
paths = _audio_files(root / subdir if subdir else root)
|
||||
|
||||
meta, _ = TransferChain._restore_music_download_context(_history(root, torrent_name=torrent), paths[0])
|
||||
|
||||
assert meta.artists == ["Artist"]
|
||||
assert meta.album == ("Child Album" if subdir else None)
|
||||
assert meta.year == (2001 if subdir else None)
|
||||
|
||||
|
||||
def test_resource_album_reaches_disc_directory_and_uses_subtitle(tmp_path):
|
||||
"""CD 子目录继承同一发行信息，中文副标题补齐主标题没有的作品与署名。"""
|
||||
root = tmp_path / "opaque"
|
||||
paths = _audio_files(root / "CD2")
|
||||
history = _history(root, torrent_name="FLAC", torrent_description="歌手：周杰伦；专辑：叶惠美；[2003]")
|
||||
|
||||
meta, _ = TransferChain._restore_music_download_context(history, paths[0])
|
||||
|
||||
assert meta.artists == ["周杰伦"]
|
||||
assert meta.album == "叶惠美"
|
||||
assert meta.year == 2003
|
||||
assert meta.disc_number == 2
|
||||
assert meta.title == "First Song"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("outside,media_type", [(True, "音乐"), (False, "电影"), (False, "电视剧")])
|
||||
def test_unrelated_history_cannot_supply_music_identity(tmp_path, outside, media_type):
|
||||
"""原始资源必须属于当前音频，影视任务的附加音轨不能被改为音乐。"""
|
||||
root = tmp_path / "opaque"
|
||||
paths = _audio_files(root)
|
||||
history = _history(tmp_path / "other" if outside else root, type=media_type)
|
||||
|
||||
assert TransferChain._restore_music_download_context(history, paths[0]) == (None, None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def media_chain(monkeypatch):
|
||||
"""使用隔离目录缓存，避免同步、异步测试共享其它用例的识别结果。"""
|
||||
from app.chain.media.cache import AlbumDirectoryCache
|
||||
|
||||
chain = MediaChain()
|
||||
monkeypatch.setattr(chain, "_album_dir_cache", AlbumDirectoryCache(16))
|
||||
return chain
|
||||
|
||||
|
||||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||||
def test_original_resource_is_used_by_album_query(tmp_path, monkeypatch, media_chain, asynchronous):
|
||||
"""资源艺人、专辑、年份进入真实目录查询，并保留各文件曲名和时长。"""
|
||||
root = tmp_path / "opaque"
|
||||
paths = _audio_files(root)
|
||||
meta, _ = TransferChain._restore_music_download_context(_history(root), paths[0])
|
||||
album = MusicAlbumInfo(media_source="musicbrainz", media_id="group", title="Actual Album", tracks=[
|
||||
MusicInfo(media_source="musicbrainz", media_id=f"recording-{index}", title=title,
|
||||
album="Actual Album", artists=["Artist"], year=2003, track_number=index, duration=3)
|
||||
for index, title in enumerate(("First Song", "Second Song"), 1)
|
||||
])
|
||||
source = Mock(match_music_album=Mock(return_value=album), async_match_music_album=AsyncMock(return_value=album))
|
||||
monkeypatch.setattr("app.chain.media.album.MusicBrainzChain", Mock(return_value=source))
|
||||
|
||||
if asynchronous:
|
||||
matched = asyncio.run(media_chain.async_recognize_music_album_directory(root, contextual_meta=meta))
|
||||
query = source.async_match_music_album.call_args
|
||||
else:
|
||||
matched_meta, info = TransferChain._match_music_album_context(make_fileitem(str(paths[0])), paths[0], meta)
|
||||
assert info.media_id == "recording-1"
|
||||
assert matched_meta.title == "First Song"
|
||||
matched = media_chain.recognize_music_album_directory(root, contextual_meta=meta)
|
||||
query = source.match_music_album.call_args
|
||||
|
||||
assert query.args[0].title == query.args[0].album == "Actual Album"
|
||||
assert query.args[0].artists == ["Artist"]
|
||||
assert query.args[0].year == 2003
|
||||
assert [track.duration for track in query.args[1]] == [3, 3]
|
||||
assert [info.title for info in matched.values()] == ["First Song", "Second Song"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("asynchronous", [False, True])
|
||||
def test_album_cache_changes_with_resource_context(tmp_path, monkeypatch, media_chain, asynchronous):
|
||||
"""同一目录和文件签名也必须区分先后提供的不同专辑证据。"""
|
||||
root = tmp_path / "opaque"
|
||||
_audio_files(root)
|
||||
source = Mock(match_music_album=Mock(return_value=None), async_match_music_album=AsyncMock(return_value=None))
|
||||
monkeypatch.setattr("app.chain.media.album.MusicBrainzChain", Mock(return_value=source))
|
||||
for album in ("First Album", "First Album", "Second Album"):
|
||||
context = MetaMusic(album=album, artists=["Artist"])
|
||||
if asynchronous:
|
||||
asyncio.run(media_chain.async_recognize_music_album_directory(root, contextual_meta=context))
|
||||
else:
|
||||
media_chain.recognize_music_album_directory(root, contextual_meta=context)
|
||||
operation = source.async_match_music_album if asynchronous else source.match_music_album
|
||||
|
||||
assert [call.args[0].album for call in operation.call_args_list] == ["First Album", "Second Album"]
|
||||
|
||||
|
||||
def test_local_album_tags_override_conflicting_resource(tmp_path, monkeypatch, media_chain):
|
||||
"""目录里已明确的专辑标签和年份不能被种子候选覆盖。"""
|
||||
root = tmp_path / "opaque"
|
||||
paths = _audio_files(root)
|
||||
for path in paths:
|
||||
audio = FLAC(path)
|
||||
audio.update({"album": ["Tagged Album"], "artist": ["Tagged Artist"], "date": ["2001"]})
|
||||
audio.save()
|
||||
source = Mock(match_music_album=Mock(return_value=None))
|
||||
monkeypatch.setattr("app.chain.media.album.MusicBrainzChain", Mock(return_value=source))
|
||||
|
||||
media_chain.recognize_music_album_directory(root, contextual_meta=MetaMusic(
|
||||
album="Wrong Album", artists=["Wrong Artist"], year=2020,
|
||||
))
|
||||
|
||||
query = source.match_music_album.call_args.args[0]
|
||||
assert query.album == "Tagged Album"
|
||||
assert query.artists == ["Tagged Artist"]
|
||||
assert query.year == 2001
|
||||
@@ -83,7 +83,7 @@ def test_album_context_preserves_classification_and_remote_music_facts(
|
||||
monkeypatch.setattr(
|
||||
MediaChain,
|
||||
"recognize_music_album_directory",
|
||||
lambda _self, _directory: {str(file_path.resolve()): matched_info},
|
||||
lambda _self, _directory, **_kwargs: {str(file_path.resolve()): matched_info},
|
||||
)
|
||||
|
||||
merged_meta, merged_info = FileFilterMixin._match_music_album_context(
|
||||
@@ -160,7 +160,7 @@ def test_album_context_rejects_candidate_conflicting_with_file_tags(
|
||||
monkeypatch.setattr(
|
||||
MediaChain,
|
||||
"recognize_music_album_directory",
|
||||
lambda _self, _directory: {str(file_path.resolve()): wrong_release},
|
||||
lambda _self, _directory, **_kwargs: {str(file_path.resolve()): wrong_release},
|
||||
)
|
||||
|
||||
returned_meta, returned_info = FileFilterMixin._match_music_album_context(
|
||||
@@ -212,7 +212,7 @@ def test_album_context_preserves_local_release_track_position(
|
||||
monkeypatch.setattr(
|
||||
MediaChain,
|
||||
"recognize_music_album_directory",
|
||||
lambda _self, _directory: {str(file_path.resolve()): matched_info},
|
||||
lambda _self, _directory, **_kwargs: {str(file_path.resolve()): matched_info},
|
||||
)
|
||||
|
||||
merged_meta, merged_info = FileFilterMixin._match_music_album_context(
|
||||
|
||||
Reference in New Issue
Block a user