nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-10-31 12:24:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(music): organize trusted audio tags without online recognition

Browse Source
This commit is contained in:
jxxghp
2026-09-29 21:28:35 +08:00
parent 2dd1813015
commit ee3fe99d55
18 changed files with 670 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
app/application/classification/execution.py
View File

@@ -81,8 +81,9 @@ class ClassificationExecutionPort(Protocol):
| None = None,
effective_override: ClassificationSelection | None = None,
refresh: bool = False,
allow_enrichment: bool = True,
) -> ClassificationSubject:
"""分类一个完整媒体对象，并按需强制刷新已存在的同 revision 结果。"""
"""分类完整媒体；离线整理可禁止外部事实补充，仍执行当前用户分类策略。"""
...
async def async_finalize(
@@ -96,6 +97,7 @@ class ClassificationExecutionPort(Protocol):
| None = None,
effective_override: ClassificationSelection | None = None,
refresh: bool = False,
allow_enrichment: bool = True,
) -> ClassificationSubject:
"""异步分类完整媒体对象，并允许有界补充缺失标准事实。"""
...
@@ -193,17 +195,19 @@ class ClassificationExecutionService:
| None = None,
effective_override: ClassificationSelection | None = None,
refresh: bool = False,
allow_enrichment: bool = True,
) -> ClassificationSubject:
"""复制并分类完整识别结果，来源缓存中的旧结果永远不作为真值。"""
"""复制并分类完整结果；离线调用可禁用外部补充，但仍应用当前策略。"""
finalized, policy, facts, effective_override = self._prepare(
media,
extensions=extensions,
effective_override=effective_override,
refresh=refresh,
allow_local_music=not allow_enrichment,
)
if policy is None or facts is None:
return finalized
if self._enrichment is not None:
if allow_enrichment and self._enrichment is not None:
try:
facts = self._enrichment.enrich(policy, facts, finalized)
except Exception: # noqa: BLE001 分类补充不得成为识别硬依赖
@@ -226,17 +230,19 @@ class ClassificationExecutionService:
| None = None,
effective_override: ClassificationSelection | None = None,
refresh: bool = False,
allow_enrichment: bool = True,
) -> ClassificationSubject:
"""异步复制并分类完整结果，补充失败时继续使用原始事实。"""
"""异步分类完整结果，支持只使用本地事实，补充失败时保留原始事实。"""
finalized, policy, facts, effective_override = self._prepare(
media,
extensions=extensions,
effective_override=effective_override,
refresh=refresh,
allow_local_music=not allow_enrichment,
)
if policy is None or facts is None:
return finalized
if self._enrichment is not None:
if allow_enrichment and self._enrichment is not None:
try:
facts = await self._enrichment.async_enrich(
policy,
@@ -260,6 +266,7 @@ class ClassificationExecutionService:
effective_override: ClassificationSelection | None,
refresh: bool,
policy_override: ClassificationPolicy | None = None,
allow_local_music: bool = False,
) -> tuple[
ClassificationSubject,
ClassificationPolicy | None,
@@ -287,6 +294,7 @@ class ClassificationExecutionService:
facts = build_classification_facts(
finalized,
extensions=merged_extensions,
allow_local_music=allow_local_music,
)
except ValueError:
finalized.set_library_category("")

8
app/chain/_recognition.py
View File

@@ -305,8 +305,9 @@ class _RecognitionFinalizationOwner:
*,
effective_override: ClassificationSelection | None = None,
refresh: bool = False,
allow_enrichment: bool = True,
) -> Optional[_ClassificationSubjectT]:
"""在唯一应用服务中复制并分类完整识别结果。"""
"""统一复制并分类结果，离线整理可禁止外部补充而不绕过用户分类规则。"""
if mediainfo is None:
return None
service = getattr(cast(ChainRuntimeMixinHost, self), "classification_service", None)
@@ -318,6 +319,7 @@ class _RecognitionFinalizationOwner:
mediainfo,
effective_override=effective_override,
refresh=refresh,
**({"allow_enrichment": False} if not allow_enrichment else {}),
),
)
@@ -327,8 +329,9 @@ class _RecognitionFinalizationOwner:
*,
effective_override: ClassificationSelection | None = None,
refresh: bool = False,
allow_enrichment: bool = True,
) -> Optional[_ClassificationSubjectT]:
"""通过异步应用服务复制、补充并分类完整识别结果。"""
"""异步复制并分类结果，保留调用方禁止外部补充的离线约束。"""
if mediainfo is None:
return None
service = getattr(cast(ChainRuntimeMixinHost, self), "classification_service", None)
@@ -340,6 +343,7 @@ class _RecognitionFinalizationOwner:
mediainfo,
effective_override=effective_override,
refresh=refresh,
**({"allow_enrichment": False} if not allow_enrichment else {}),
),
)

2
app/chain/media/contract.py
View File

@@ -87,6 +87,7 @@ if TYPE_CHECKING:
*,
effective_override: ClassificationSelection | None = None,
refresh: bool = False,
allow_enrichment: bool = True,
) -> Optional[_ClassificationSubjectT]:
"""通过注入的应用服务分类一个完整识别结果。"""
...
@@ -97,6 +98,7 @@ if TYPE_CHECKING:
*,
effective_override: ClassificationSelection | None = None,
refresh: bool = False,
allow_enrichment: bool = True,
) -> Optional[_ClassificationSubjectT]:
"""通过注入的应用服务异步补充并分类完整识别结果。"""
...

354
app/chain/transfer/filter.py
View File

@@ -1,13 +1,9 @@
"""整理文件筛选、音乐上下文与源目录清理判定。"""
import re
import threading
from collections import Counter
from copy import deepcopy
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, Optional, Protocol, Tuple, Union, cast
from typing import Any, Dict, Optional, Protocol, Union, cast
from app.application.audio import AudioMetadataHelper
from app.application.configuration import (
get_chain_runtime_config_snapshot,
)
@@ -23,13 +19,13 @@ 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.domain.context import MediaInfo, MusicAlbumInfo, MusicInfo
from app.domain.media import normalize_music_type
from app.domain.meta.metamusic import MetaMusic
from app.domain.music import (
music_album_matches,
music_artist_matches,
music_text_key,
music_title_matches,
music_version_matches,
music_year_matches,
@@ -90,348 +86,6 @@ def _network_filesystem_snapshot() -> NetworkFilesystemPort:
return port
@dataclass(slots=True)
class _MusicBatchContext:
"""同批音乐文件之间的歌词关联与已识别上下文。"""
related_main_keys: dict[Tuple[str, str], Tuple[str, str]] = field(default_factory=dict)
single_main_keys: set[Tuple[str, str]] = field(default_factory=set)
album_main_keys: set[Tuple[str, str]] = field(default_factory=set)
directory_evidence: dict[Tuple[str, str], MetaMusic] = field(default_factory=dict)
album_evidence_by_main_key: dict[Tuple[str, str], MetaMusic] = field(default_factory=dict)
resolved_contexts: dict[Tuple[str, str], tuple[MetaMusic, MusicInfo]] = field(default_factory=dict)
def _music_directory_consensus(values: list[str]) -> Optional[str]:
"""只在同目录至少两个标签高度一致时返回共识值。"""
normalized = [
(music_text_key(value), value.strip())
for value in values
if value and music_text_key(value)
]
if len(normalized) < 2:
return None
counts = Counter(key for key, _value in normalized)
key, count = counts.most_common(1)[0]
if count < 2 or count / len(normalized) < 0.8:
return None
return next(value for current_key, value in normalized if current_key == key)
def _music_directory_evidence(items: list[FileItem]) -> Optional[MetaMusic]:
"""从同一发行目录的原始音频标签提取可信艺人与专辑共识。"""
artist_values: list[str] = []
album_values: list[str] = []
collective_keys = {
music_text_key(value)
for value in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人")
}
for item in items:
if getattr(item, "storage", "local") != "local" or not item.path:
continue
tag_meta = AudioMetadataHelper.read_tags(Path(item.path))
if not tag_meta:
continue
artist = tag_meta.album_artist
if not artist and len(tag_meta.artists) == 1:
artist = tag_meta.artists[0]
if artist and music_text_key(artist) not in collective_keys:
artist_values.append(artist)
if tag_meta.album:
album_values.append(tag_meta.album)
artist = _music_directory_consensus(artist_values)
album = _music_directory_consensus(album_values)
if not artist and not album:
return None
return MetaMusic(
artists=[artist] if artist else None,
album_artist=artist,
album=album,
)
def _music_tagged_album_groups(
items: list[FileItem],
) -> list[tuple[list[FileItem], MetaMusic]]:
"""按精确的专辑艺人和专辑标签划分同目录内的多碟/多发行分组。"""
grouped: dict[tuple[str, str], tuple[list[FileItem], str, str]] = {}
collective_keys = {
music_text_key(value)
for value in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人")
}
for item in items:
if getattr(item, "storage", "local") != "local" or not item.path:
continue
tag_meta = AudioMetadataHelper.read_tags(Path(item.path))
if not tag_meta or not tag_meta.album:
continue
artist = tag_meta.album_artist
if not artist and len(tag_meta.artists) == 1:
artist = tag_meta.artists[0]
artist_key = music_text_key(artist)
album_key = music_text_key(tag_meta.album)
if not artist or artist_key in collective_keys or not album_key:
continue
key = (artist_key, album_key)
if key not in grouped:
grouped[key] = ([], artist.strip(), tag_meta.album.strip())
grouped[key][0].append(item)
return [
(
grouped_items,
MetaMusic(artists=[artist], album_artist=artist, album=album),
)
for grouped_items, artist, album in grouped.values()
if len(grouped_items) >= 2
]
def _apply_music_directory_evidence(
meta: MetaMusic,
evidence: Optional[MetaMusic],
) -> MetaMusic:
"""仅补齐缺失的作品身份字段，不覆盖单曲原有标签。"""
if not evidence:
return meta
merged = deepcopy(meta)
if not merged.artists and evidence.artists:
merged.artists = list(evidence.artists)
if not merged.album_artist and evidence.album_artist:
merged.album_artist = evidence.album_artist
if not merged.album and evidence.album:
merged.album = evidence.album
return merged
def _apply_music_directory_year(meta: MetaMusic, file_path: Path) -> MetaMusic:
"""用最近发行目录开头的四位年份纠正合集内不可靠的音频年份标签。"""
for parent in list(file_path.parents)[:4]:
match = re.match(r"^((?:19|20)\d{2})(?:\D|$)", parent.name)
if not match:
continue
merged = deepcopy(meta)
merged.year = cast(Any, int(match.group(1)))
return merged
return meta
def _prepare_music_batch_context(
owner: _TransferOwnerBase,
file_items: list[tuple[FileItem, bool]],
batch_mtype: Optional[MediaType],
) -> _MusicBatchContext:
"""建立同目录音轨、歌词和单音轨目录的批次索引。"""
context = _MusicBatchContext()
if batch_mtype != MediaType.MUSIC:
return context
main_items_by_dir: dict[Tuple[str, str], list[FileItem]] = {}
for current_item, current_bluray_dir in file_items:
if not current_bluray_dir and owner._is_media_file(current_item, MediaType.MUSIC):
main_items_by_dir.setdefault(
owner._get_file_parent_key(current_item), []
).append(current_item)
context.single_main_keys = {
owner._get_file_key(items[0])
for items in main_items_by_dir.values()
if len(items) == 1
}
for parent_key, items in main_items_by_dir.items():
evidence = _music_directory_evidence(items)
if evidence:
context.directory_evidence[parent_key] = evidence
# 两首及以上音轨，且目录内艺人和专辑标签均达到高一致性时，足以在
# 远端临时不可用时证明这是一个专辑目录；不能把整个艺术家合集根目录
# 当成一张专辑，也不能仅凭文件夹名称猜测类别。
if evidence and len(items) > 1 and evidence.artists and evidence.album:
context.album_main_keys.update(
owner._get_file_key(item) for item in items
)
for grouped_items, grouped_evidence in _music_tagged_album_groups(items):
for item in grouped_items:
item_key = owner._get_file_key(item)
context.album_main_keys.add(item_key)
context.album_evidence_by_main_key[item_key] = grouped_evidence
for current_item, _current_bluray_dir in file_items:
if not owner._is_music_lyrics_file(current_item):
continue
related_key = owner._get_related_main_file_key(
current_item,
main_items_by_dir.get(owner._get_file_parent_key(current_item), []),
)
if related_key:
context.related_main_keys[owner._get_file_key(current_item)] = related_key
return context
def _resolve_music_batch_file_context(
owner: _TransferOwnerBase,
*,
batch_context: _MusicBatchContext,
file_item: FileItem,
file_path: Path,
file_meta: Any,
selected_tracks: dict[str, MusicInfo],
fallback: Optional[Union[MediaInfo, MusicInfo]],
discard_shared_identity: bool,
multi_track_batch: bool,
release_regions: Optional[list[str]],
release_scripts: Optional[list[str]],
) -> tuple[Any, Optional[Union[MediaInfo, MusicInfo]]]:
"""解析一项音乐上下文，并让同名歌词复用主音轨结果。"""
file_key = owner._get_file_key(file_item)
related_key = batch_context.related_main_keys.get(file_key)
related_context = (
batch_context.resolved_contexts.get(related_key)
if related_key
else None
)
if related_context:
return deepcopy(related_context[0]), deepcopy(related_context[1])
file_meta, task_mediainfo = owner._selected_music_task_context(
file_item,
file_path,
file_meta,
selected_tracks,
fallback,
)
if isinstance(file_meta, MetaMusic):
file_key = owner._get_file_key(file_item)
file_meta = _apply_music_directory_evidence(
file_meta,
batch_context.album_evidence_by_main_key.get(file_key)
or batch_context.directory_evidence.get(owner._get_file_parent_key(file_item)),
)
if discard_shared_identity:
file_meta = _apply_music_directory_year(file_meta, file_path)
if not task_mediainfo and isinstance(file_meta, MetaMusic):
file_meta, task_mediainfo = _recognize_music_batch_file(
owner,
batch_context=batch_context,
file_item=file_item,
file_path=file_path,
file_meta=file_meta,
discard_shared_identity=discard_shared_identity,
multi_track_batch=multi_track_batch,
release_regions=release_regions,
release_scripts=release_scripts,
)
if discard_shared_identity and isinstance(file_meta, MetaMusic):
# 目录年份是当前下载包/发行目录的本地强证据。目录级或逐曲远端识别
# 可能再次用再版年份覆盖前面已经纠正的年份，因此在所有识别层结束后
# 重放一次目录年份约束。显式指定 MusicBrainz 身份时不会进入此分支。
file_meta = _apply_music_directory_year(file_meta, file_path)
if isinstance(task_mediainfo, MusicInfo) and file_meta.year:
task_mediainfo = deepcopy(task_mediainfo)
task_mediainfo.year = file_meta.year
directory_evidence = (
batch_context.album_evidence_by_main_key.get(owner._get_file_key(file_item))
or batch_context.directory_evidence.get(owner._get_file_parent_key(file_item))
)
if (
owner._is_audio_file(file_item)
and isinstance(file_meta, MetaMusic)
and isinstance(task_mediainfo, MusicInfo)
and owner._get_file_key(file_item) in batch_context.album_main_keys
and directory_evidence
):
# 同一物理发行目录只能生成一个专辑目录。逐曲远端识别可能因服务
# 波动而混用正式标题、别名或本地兜底；用已验证的目录标签共识统一
# 专辑和专辑艺人，但保留每首曲目的远端 ID、曲名、封面等信息。
file_meta = deepcopy(file_meta)
task_mediainfo = deepcopy(task_mediainfo)
if directory_evidence.album:
file_meta.album = directory_evidence.album
task_mediainfo.album = directory_evidence.album
if directory_evidence.album_artist:
file_meta.album_artist = directory_evidence.album_artist
task_mediainfo.album_artist = directory_evidence.album_artist
if file_meta.year:
task_mediainfo.year = file_meta.year
if (
owner._is_audio_file(file_item)
and isinstance(task_mediainfo, MusicInfo)
and owner._get_file_key(file_item) in batch_context.album_main_keys
and str(task_mediainfo.album_type or "").casefold()
not in {"ep", "broadcast", "other"}
):
# 曲目可能同时作为 Single 单独发行。整理完整多音轨目录时，目录内
# 高一致性的专辑/专辑艺人标签是当前文件归属的更强证据，不能让某一
# 首歌的单曲发行身份把同一张专辑拆进 Single 分类。
task_mediainfo = deepcopy(task_mediainfo)
task_mediainfo.album_type = "Album"
finalized = owner._finalize_recognition_result(task_mediainfo, refresh=True)
if isinstance(finalized, MusicInfo):
task_mediainfo = finalized
task_mediainfo.set_library_category("Album")
if (
owner._is_audio_file(file_item)
and isinstance(file_meta, MetaMusic)
and isinstance(task_mediainfo, MusicInfo)
):
batch_context.resolved_contexts[owner._get_file_key(file_item)] = (
deepcopy(file_meta),
deepcopy(task_mediainfo),
)
return file_meta, task_mediainfo
def _recognize_music_batch_file(
owner: _TransferOwnerBase,
*,
batch_context: _MusicBatchContext,
file_item: FileItem,
file_path: Path,
file_meta: MetaMusic,
discard_shared_identity: bool,
multi_track_batch: bool,
release_regions: Optional[list[str]],
release_scripts: Optional[list[str]],
) -> tuple[MetaMusic, Optional[MusicInfo]]:
"""按专辑、曲目证据和单音轨目录结构依次补齐音乐身份。"""
file_meta, task_mediainfo = owner._match_music_album_context(
file_item,
file_path,
file_meta,
release_regions,
release_scripts,
)
if not task_mediainfo and discard_shared_identity and owner._is_audio_file(file_item):
file_meta, task_mediainfo = owner._match_music_recording_context(
file_item,
file_path,
file_meta,
)
if task_mediainfo or not discard_shared_identity:
return file_meta, task_mediainfo
task_mediainfo = owner._music_info_from_meta(file_meta)
if (
multi_track_batch
and owner._get_file_key(file_item) in batch_context.album_main_keys
):
task_mediainfo.album_type = "Album"
finalized = owner._finalize_recognition_result(task_mediainfo)
if isinstance(finalized, MusicInfo):
task_mediainfo = finalized
if not task_mediainfo.library_category:
task_mediainfo.set_library_category("Album")
return file_meta, task_mediainfo
if owner._get_file_key(file_item) not in batch_context.single_main_keys:
return file_meta, task_mediainfo
# 远端识别均未命中时，仅单音轨子目录可安全按 Single 兜底；
# 多音轨目录仍保持未识别，避免把缺失专辑误判为单曲。
task_mediainfo.album_type = "Single"
finalized = owner._finalize_recognition_result(task_mediainfo)
if isinstance(finalized, MusicInfo):
task_mediainfo = finalized
if not task_mediainfo.library_category:
# 隔离测试可能没有装配分类服务，保留旧分类路径作为测试兼容。
task_mediainfo.set_library_category("Single")
return file_meta, task_mediainfo
class _MusicFileFilterBase(_TransferOwnerBase):
"""提供通用文件识别及 MusicBrainz 专辑、曲目匹配能力。"""
@@ -644,8 +298,8 @@ class FileFilterMixin(_TransferOwnerBase):
"""提供整理文件筛选、音乐批次上下文和源目录清理判定。"""
__mixin_host_protocol__ = TransferMixinHost
_prepare_music_batch_context = _prepare_music_batch_context
_resolve_music_batch_file_context = _resolve_music_batch_file_context
_prepare_music_batch_context = prepare_music_batch_context
_resolve_music_batch_file_context = resolve_music_batch_file_context
_requires_automatic_category = cast(Any, staticmethod(
_MusicFileFilterBase._requires_automatic_category
))

385
app/chain/transfer/music.py Normal file
View File

@@ -0,0 +1,385 @@
"""音乐整理批次的本地标签准入、目录共识和曲目上下文编排。"""
import re
from collections import Counter
from copy import deepcopy
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional, Tuple, Union, cast
from app.application.audio import AudioMetadataHelper
from app.chain.transfer.contract import _TransferOwnerBase
from app.domain.context import MediaInfo, MusicInfo
from app.domain.meta.metamusic import MetaMusic
from app.domain.music import music_tags_are_usable, music_text_key
from app.runtime.log import logger
from app.schemas.file import FileItem
from app.schemas.types import MediaType
@dataclass(slots=True)
class MusicBatchContext:
"""同批音乐文件之间的歌词关联与已识别上下文。"""
related_main_keys: dict[Tuple[str, str], Tuple[str, str]] = field(default_factory=dict)
single_main_keys: set[Tuple[str, str]] = field(default_factory=set)
album_main_keys: set[Tuple[str, str]] = field(default_factory=set)
directory_evidence: dict[Tuple[str, str], MetaMusic] = field(default_factory=dict)
album_evidence_by_main_key: dict[Tuple[str, str], MetaMusic] = field(default_factory=dict)
resolved_contexts: dict[Tuple[str, str], tuple[MetaMusic, MusicInfo]] = field(default_factory=dict)
def _music_directory_consensus(values: list[str]) -> Optional[str]:
"""只在同目录至少两个标签高度一致时返回共识值。"""
normalized = [
(music_text_key(value), value.strip())
for value in values
if value and music_text_key(value)
]
if len(normalized) < 2:
return None
counts = Counter(key for key, _value in normalized)
key, count = counts.most_common(1)[0]
if count < 2 or count / len(normalized) < 0.8:
return None
return next(value for current_key, value in normalized if current_key == key)
def _music_directory_evidence(items: list[FileItem]) -> Optional[MetaMusic]:
"""从同一发行目录的原始音频标签提取可信艺人与专辑共识。"""
artist_values: list[str] = []
album_values: list[str] = []
collective_keys = {
music_text_key(value)
for value in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人")
}
for item in items:
if getattr(item, "storage", "local") != "local" or not item.path:
continue
tag_meta = AudioMetadataHelper.read_tags(Path(item.path))
if not tag_meta:
continue
artist = tag_meta.album_artist
if not artist and len(tag_meta.artists) == 1:
artist = tag_meta.artists[0]
if artist and music_text_key(artist) not in collective_keys:
artist_values.append(artist)
if tag_meta.album:
album_values.append(tag_meta.album)
artist = _music_directory_consensus(artist_values)
album = _music_directory_consensus(album_values)
if not artist and not album:
return None
return MetaMusic(
artists=[artist] if artist else None,
album_artist=artist,
album=album,
)
def _music_tagged_album_groups(
items: list[FileItem],
) -> list[tuple[list[FileItem], MetaMusic]]:
"""按精确的专辑艺人和专辑标签划分同目录内的多碟/多发行分组。"""
grouped: dict[tuple[str, str], tuple[list[FileItem], str, str]] = {}
collective_keys = {
music_text_key(value)
for value in ("Various Artists", "Various", "VA", "群星", "众艺人", "眾藝人")
}
for item in items:
if getattr(item, "storage", "local") != "local" or not item.path:
continue
tag_meta = AudioMetadataHelper.read_tags(Path(item.path))
if not tag_meta or not tag_meta.album:
continue
artist = tag_meta.album_artist
if not artist and len(tag_meta.artists) == 1:
artist = tag_meta.artists[0]
artist_key = music_text_key(artist)
album_key = music_text_key(tag_meta.album)
if not artist or artist_key in collective_keys or not album_key:
continue
key = (artist_key, album_key)
if key not in grouped:
grouped[key] = ([], artist.strip(), tag_meta.album.strip())
grouped[key][0].append(item)
return [
(
grouped_items,
MetaMusic(artists=[artist], album_artist=artist, album=album),
)
for grouped_items, artist, album in grouped.values()
if len(grouped_items) >= 2
]
def _apply_music_directory_evidence(
meta: MetaMusic,
evidence: Optional[MetaMusic],
) -> MetaMusic:
"""仅补齐缺失的作品身份字段，不覆盖单曲原有标签。"""
if not evidence:
return meta
merged = deepcopy(meta)
if not merged.artists and evidence.artists:
merged.artists = list(evidence.artists)
if not merged.album_artist and evidence.album_artist:
merged.album_artist = evidence.album_artist
if not merged.album and evidence.album:
merged.album = evidence.album
return merged
def _apply_music_directory_year(meta: MetaMusic, file_path: Path) -> MetaMusic:
"""用最近发行目录开头的四位年份纠正合集内不可靠的音频年份标签。"""
for parent in list(file_path.parents)[:4]:
match = re.match(r"^((?:19|20)\d{2})(?:\D|$)", parent.name)
if not match:
continue
merged = deepcopy(meta)
merged.year = cast(Any, int(match.group(1)))
return merged
return meta
def _local_music_context(
owner: _TransferOwnerBase,
file_item: FileItem,
file_path: Path,
) -> Optional[tuple[MetaMusic, MusicInfo]]:
"""仅以本地真实标签建立快速整理上下文，不向在线来源确认身份。"""
if file_item.storage != "local" or not owner._is_audio_file(file_item):
return None
tags = AudioMetadataHelper.read_tags(file_path)
if tags is None or not music_tags_are_usable(tags):
return None
meta = tags.apply_path_context(file_path)
logger.info(f"{file_path.name} 使用完整音频标签整理：{meta.artist} - {meta.title}")
return meta, MusicInfo.from_meta(meta)
def prepare_music_batch_context(
owner: _TransferOwnerBase,
file_items: list[tuple[FileItem, bool]],
batch_mtype: Optional[MediaType],
) -> MusicBatchContext:
"""建立同目录音轨、歌词和单音轨目录的批次索引。"""
context = MusicBatchContext()
if batch_mtype != MediaType.MUSIC:
return context
main_items_by_dir: dict[Tuple[str, str], list[FileItem]] = {}
for current_item, current_bluray_dir in file_items:
if not current_bluray_dir and owner._is_media_file(current_item, MediaType.MUSIC):
main_items_by_dir.setdefault(
owner._get_file_parent_key(current_item), []
).append(current_item)
context.single_main_keys = {
owner._get_file_key(items[0])
for items in main_items_by_dir.values()
if len(items) == 1
}
for parent_key, items in main_items_by_dir.items():
evidence = _music_directory_evidence(items)
if evidence:
context.directory_evidence[parent_key] = evidence
# 两首及以上音轨，且目录内艺人和专辑标签均达到高一致性时，足以在
# 远端临时不可用时证明这是一个专辑目录；不能把整个艺术家合集根目录
# 当成一张专辑，也不能仅凭文件夹名称猜测类别。
if evidence and len(items) > 1 and evidence.artists and evidence.album:
context.album_main_keys.update(
owner._get_file_key(item) for item in items
)
for grouped_items, grouped_evidence in _music_tagged_album_groups(items):
for item in grouped_items:
item_key = owner._get_file_key(item)
context.album_main_keys.add(item_key)
context.album_evidence_by_main_key[item_key] = grouped_evidence
for current_item, _current_bluray_dir in file_items:
if not owner._is_music_lyrics_file(current_item):
continue
related_key = owner._get_related_main_file_key(
current_item,
main_items_by_dir.get(owner._get_file_parent_key(current_item), []),
)
if related_key:
context.related_main_keys[owner._get_file_key(current_item)] = related_key
return context
def resolve_music_batch_file_context(
owner: _TransferOwnerBase,
*,
batch_context: MusicBatchContext,
file_item: FileItem,
file_path: Path,
file_meta: Any,
selected_tracks: dict[str, MusicInfo],
fallback: Optional[Union[MediaInfo, MusicInfo]],
discard_shared_identity: bool,
multi_track_batch: bool,
release_regions: Optional[list[str]],
release_scripts: Optional[list[str]],
) -> tuple[Any, Optional[Union[MediaInfo, MusicInfo]]]:
"""解析一项音乐上下文，并让同名歌词复用主音轨结果。"""
file_key = owner._get_file_key(file_item)
related_key = batch_context.related_main_keys.get(file_key)
related_context = (
batch_context.resolved_contexts.get(related_key)
if related_key
else None
)
if related_context:
return deepcopy(related_context[0]), deepcopy(related_context[1])
file_meta, task_mediainfo = owner._selected_music_task_context(
file_item,
file_path,
file_meta,
selected_tracks,
fallback,
)
local_context = (
_local_music_context(owner, file_item, file_path)
if not task_mediainfo and isinstance(file_meta, MetaMusic)
else None
)
if local_context:
file_meta, task_mediainfo = local_context
if isinstance(file_meta, MetaMusic):
file_key = owner._get_file_key(file_item)
file_meta = _apply_music_directory_evidence(
file_meta,
batch_context.album_evidence_by_main_key.get(file_key)
or batch_context.directory_evidence.get(owner._get_file_parent_key(file_item)),
)
if discard_shared_identity and not local_context:
file_meta = _apply_music_directory_year(file_meta, file_path)
if not task_mediainfo and isinstance(file_meta, MetaMusic):
file_meta, task_mediainfo = _recognize_music_batch_file(
owner,
batch_context=batch_context,
file_item=file_item,
file_path=file_path,
file_meta=file_meta,
discard_shared_identity=discard_shared_identity,
multi_track_batch=multi_track_batch,
release_regions=release_regions,
release_scripts=release_scripts,
)
if discard_shared_identity and not local_context and isinstance(file_meta, MetaMusic):
# 目录年份是当前下载包/发行目录的本地强证据。目录级或逐曲远端识别
# 可能再次用再版年份覆盖前面已经纠正的年份，因此在所有识别层结束后
# 重放一次目录年份约束。显式指定 MusicBrainz 身份时不会进入此分支。
file_meta = _apply_music_directory_year(file_meta, file_path)
if isinstance(task_mediainfo, MusicInfo) and file_meta.year:
task_mediainfo = deepcopy(task_mediainfo)
task_mediainfo.year = file_meta.year
directory_evidence = (
batch_context.album_evidence_by_main_key.get(owner._get_file_key(file_item))
or batch_context.directory_evidence.get(owner._get_file_parent_key(file_item))
)
if (
owner._is_audio_file(file_item)
and isinstance(file_meta, MetaMusic)
and isinstance(task_mediainfo, MusicInfo)
and owner._get_file_key(file_item) in batch_context.album_main_keys
and directory_evidence
):
# 同一物理发行目录只能生成一个专辑目录。逐曲远端识别可能因服务
# 波动而混用正式标题、别名或本地兜底；用已验证的目录标签共识统一
# 专辑和专辑艺人，但保留每首曲目的远端 ID、曲名、封面等信息。
file_meta = deepcopy(file_meta)
task_mediainfo = deepcopy(task_mediainfo)
if directory_evidence.album:
file_meta.album = directory_evidence.album
task_mediainfo.album = directory_evidence.album
if directory_evidence.album_artist:
file_meta.album_artist = directory_evidence.album_artist
task_mediainfo.album_artist = directory_evidence.album_artist
if file_meta.year:
task_mediainfo.year = file_meta.year
if (
owner._is_audio_file(file_item)
and isinstance(task_mediainfo, MusicInfo)
and owner._get_file_key(file_item) in batch_context.album_main_keys
and str(task_mediainfo.album_type or "").casefold()
not in {"ep", "broadcast", "other"}
):
# 曲目可能同时作为 Single 单独发行。整理完整多音轨目录时，目录内
# 高一致性的专辑/专辑艺人标签是当前文件归属的更强证据，不能让某一
# 首歌的单曲发行身份把同一张专辑拆进 Single 分类。
task_mediainfo = deepcopy(task_mediainfo)
task_mediainfo.album_type = "Album"
if not local_context:
finalized = owner._finalize_recognition_result(task_mediainfo, refresh=True)
if isinstance(finalized, MusicInfo):
task_mediainfo = finalized
task_mediainfo.set_library_category("Album")
if local_context:
task_mediainfo = owner._finalize_recognition_result(task_mediainfo, allow_enrichment=False)
if (
owner._is_audio_file(file_item)
and isinstance(file_meta, MetaMusic)
and isinstance(task_mediainfo, MusicInfo)
):
batch_context.resolved_contexts[owner._get_file_key(file_item)] = (
deepcopy(file_meta),
deepcopy(task_mediainfo),
)
return file_meta, task_mediainfo
def _recognize_music_batch_file(
owner: _TransferOwnerBase,
*,
batch_context: MusicBatchContext,
file_item: FileItem,
file_path: Path,
file_meta: MetaMusic,
discard_shared_identity: bool,
multi_track_batch: bool,
release_regions: Optional[list[str]],
release_scripts: Optional[list[str]],
) -> tuple[MetaMusic, Optional[MusicInfo]]:
"""按专辑、曲目证据和单音轨目录结构依次补齐音乐身份。"""
file_meta, task_mediainfo = owner._match_music_album_context(
file_item,
file_path,
file_meta,
release_regions,
release_scripts,
)
if not task_mediainfo and discard_shared_identity and owner._is_audio_file(file_item):
file_meta, task_mediainfo = owner._match_music_recording_context(
file_item,
file_path,
file_meta,
)
if task_mediainfo or not discard_shared_identity:
return file_meta, task_mediainfo
task_mediainfo = owner._music_info_from_meta(file_meta)
if (
multi_track_batch
and owner._get_file_key(file_item) in batch_context.album_main_keys
):
task_mediainfo.album_type = "Album"
finalized = owner._finalize_recognition_result(task_mediainfo)
if isinstance(finalized, MusicInfo):
task_mediainfo = finalized
if not task_mediainfo.library_category:
task_mediainfo.set_library_category("Album")
return file_meta, task_mediainfo
if owner._get_file_key(file_item) not in batch_context.single_main_keys:
return file_meta, task_mediainfo
# 远端识别均未命中时，仅单音轨子目录可安全按 Single 兜底；
# 多音轨目录仍保持未识别，避免把缺失专辑误判为单曲。
task_mediainfo.album_type = "Single"
finalized = owner._finalize_recognition_result(task_mediainfo)
if isinstance(finalized, MusicInfo):
task_mediainfo = finalized
if not task_mediainfo.library_category:
# 隔离测试可能没有装配分类服务，保留旧分类路径作为测试兼容。
task_mediainfo.set_library_category("Single")
return file_meta, task_mediainfo

14
app/domain/classification/facts.py
View File

@@ -15,6 +15,8 @@ from app.domain.classification.vocabulary import (
TMDB_GENRE_KEYS as _TMDB_GENRE_KEYS,
)
from app.domain.context import MediaInfo, MusicAlbumInfo, MusicArtistInfo, MusicInfo
from app.domain.meta.metamusic import MetaMusic
from app.domain.music import music_tags_are_usable
from app.schemas.category import (
ClassificationFacts,
ClassificationFactValue,
@@ -32,11 +34,17 @@ def build_classification_facts(
media: ClassificationMedia,
*,
extensions: Mapping[str, Mapping[str, ClassificationFactValue]] | None = None,
allow_local_music: bool = False,
) -> ClassificationFacts:
"""构造规则求值器唯一接收的标准事实，并保留主媒体身份。"""
"""构造标准事实；离线音乐可显式使用完整本地名称，身份保持空值而不伪造 ID。"""
media_source = _enum_text(getattr(media, "media_source", None))
media_id = _optional_text(getattr(media, "media_id", None))
if not media_source or not media_id:
local_music = (
allow_local_music and not media_source and not media_id
and isinstance(media, MusicInfo)
and music_tags_are_usable(MetaMusic.from_music_info(media))
)
if (not media_source or not media_id) and not local_music:
raise ValueError("分类事实要求完整的 media_source 与 media_id")
media_type = _classification_media_type(getattr(media, "type", None))
@@ -58,7 +66,7 @@ def build_classification_facts(
return ClassificationFacts(
identity=ClassificationIdentityFacts(
media_source=media_source,
media_id=media_id,
media_id=media_id or "",
),
media=media_facts,
music=_music_facts(media) if is_music else None,

26
app/domain/music.py
View File

@@ -59,6 +59,32 @@ _ISRC = re.compile(r"[A-Z]{2}[A-Z0-9]{3}[0-9]{7}", re.IGNORECASE | re.ASCII)
_CJK = re.compile(r"[\u3040-\u30ff\u3400-\u9fff\uac00-\ud7af]")
def music_tags_are_usable(meta: Optional[MetaMusic]) -> bool:
"""判断纯标签是否足以确定本地整理路径，不把文件名或目录猜测当作标签。
年份和远端 ID 不影响本地可整理性；占位、乱码或宣传 URL 不能成为
歌曲/专辑身份。调用方必须传入未经路径补写的原始标签证据。
"""
if meta is None or re.fullmatch(r"\d{1,3}", str(meta.title or "").strip()):
return False
artist = meta.album_artist or next(iter(meta.artists), None)
required = [meta.title, meta.album, artist]
return all(_usable_music_tag(value) for value in required)
def _usable_music_tag(value: Optional[str]) -> bool:
"""保守排除占位值及乱码，不误删数字专辑名和包含 Unknown 的正常标题。"""
text = str(value or "").strip()
placeholders = {"unknown", "unknown artist", "unknown album", "untitled", "未知", "未知艺术家", "未知专辑"}
return bool(
text
and text.casefold() not in placeholders
and "\ufffd" not in text
and not re.search(r"https?://|www\.", text, re.IGNORECASE)
and not re.fullmatch(r"(?:track|audio|音轨|曲目)\s*[-._ ]*\d+", text, re.IGNORECASE)
)
@dataclass(frozen=True, slots=True)
class MusicMatch:
"""区分可自动采用的精确命中、仅可人工确认的候选和无关资源。"""

6
docs/architecture-overview.md
View File

@@ -762,8 +762,8 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 1053 |
| 内部导入边 | 8,937 |
| Python 模块 | 1054 |
| 内部导入边 | 8,955 |
| 非平凡 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` 包中的解析展示与订阅协调模块处理；宿主依赖边现为 8,937，音乐交互包未新增 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` 为 573、`app.factory` 为 585、`app.main` 为 587。站点图片域名快照新增 `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,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`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
语义门禁，`Check host architecture snapshot` 再执行快照测试及一次

2
docs/refactor/optimization-checklist.md
View File

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| 指标 | 当前值 | 解释 |
|---|---:|---|
| 宿主 Python 模块 / 内部依赖边 | 1053 / 8,937 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档 |
| 宿主 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` 归档 |
| 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

1
docs/rules/05-architecture.md
View File

@@ -72,6 +72,7 @@ to make the directory tree look symmetrical.
| `app/application/network.py` | System network-test target catalog, immutable public/private projections, URL and redirect admission, response validation and the injected transport Port; startup owns concrete HTTP Adapter assembly |
| `app/application/outbox.py` | Durable intent, transaction-only stager, short-transaction dispatch store, claim fencing and structured post-commit result contracts |
| `app/application/transfer/` | Durable transfer use cases: `workflow.py` owns queue service orchestration; `models.py` owns admission/planning/task contracts; `jobs.py` owns in-process task views; `notifications.py` owns failure aggregation; `projection.py` owns domain projections; `execution.py` owns stable operation identity, step/checkpoint state, retry/manual-review commands and terminal-settlement DTOs; `recovery.py` owns failed/corrupt task cleanup and history detachment through the execution repository; `history.py` projects history write fields and file fingerprints; `feedback.py` owns failure stages, notification snapshots and message text, while Chain owns notification delivery and cleanup side effects |
| `app/chain/transfer/music.py` | 音乐整理的纯标签准入、目录共识、发行分组与曲目上下文编排；`filter.py` 保留通用文件筛选和既有 mixin 调用契约。可信标签在本地完成分类与命名，分类的外部事实补充可由本次用例明确禁用，仍遵守用户策略与人工覆盖 |
| `app/application/plugin/` | Plugin market catalog, installation command, installed-plugin identity contract and startup migration, runtime port, folder operations and dynamic-route use cases; filenames remain single words (`catalog.py`, `identity.py`, `migration.py`, `install.py`, `runtime.py`, `folders.py`, `routes.py`) |
| `app/application/server/` | MoviePilot Server reporting and sharing use cases; local data readers and transport callbacks are injected by startup |
| `app/application/site/` | Configured site catalog, authentication level and index-resource capability; the generated extension and its data bundle stay together here |

27
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -1097,8 +1097,8 @@
"runtime_only": true
}
},
"edge_count": 8937,
"edge_sha256": "971c665cd7d8ca534b7a6f09ce61d0c9e7cf4ba00b3340dd60840c1fe924af8b",
"edge_count": 8955,
"edge_sha256": "b56dbef23e86e017c08b09b354d227a0c286cae52c224f797c254efa956de368",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -5513,7 +5513,6 @@
"app.chain.transfer.facade -> app.runtime",
"app.chain.transfer.facade -> app.runtime.reload",
"app.chain.transfer.filter -> app.application",
"app.chain.transfer.filter -> app.application.audio",
"app.chain.transfer.filter -> app.application.configuration",
"app.chain.transfer.filter -> app.application.directory",
"app.chain.transfer.filter -> app.application.history",
@@ -5526,6 +5525,7 @@
"app.chain.transfer.filter -> app.chain.storage",
"app.chain.transfer.filter -> app.chain.transfer",
"app.chain.transfer.filter -> app.chain.transfer.contract",
"app.chain.transfer.filter -> app.chain.transfer.music",
"app.chain.transfer.filter -> app.domain",
"app.chain.transfer.filter -> app.domain.context",
"app.chain.transfer.filter -> app.domain.media",
@@ -5580,6 +5580,21 @@
"app.chain.transfer.history -> app.schemas.transfer",
"app.chain.transfer.history -> app.schemas.types",
"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.chain",
"app.chain.transfer.music -> app.chain.transfer",
"app.chain.transfer.music -> app.chain.transfer.contract",
"app.chain.transfer.music -> app.domain",
"app.chain.transfer.music -> app.domain.context",
"app.chain.transfer.music -> app.domain.meta",
"app.chain.transfer.music -> app.domain.meta.metamusic",
"app.chain.transfer.music -> app.domain.music",
"app.chain.transfer.music -> app.runtime",
"app.chain.transfer.music -> app.runtime.log",
"app.chain.transfer.music -> app.schemas",
"app.chain.transfer.music -> app.schemas.file",
"app.chain.transfer.music -> app.schemas.types",
"app.chain.transfer.plan -> app.application",
"app.chain.transfer.plan -> app.application.classification",
"app.chain.transfer.plan -> app.application.classification.reference",
@@ -6400,6 +6415,9 @@
"app.domain.classification.facts -> app.domain.classification",
"app.domain.classification.facts -> app.domain.classification.vocabulary",
"app.domain.classification.facts -> app.domain.context",
"app.domain.classification.facts -> app.domain.meta",
"app.domain.classification.facts -> app.domain.meta.metamusic",
"app.domain.classification.facts -> app.domain.music",
"app.domain.classification.facts -> app.schemas",
"app.domain.classification.facts -> app.schemas.category",
"app.domain.classification.fields -> app.domain",
@@ -10038,7 +10056,7 @@
"app.workflow.actions.transfer_file -> app.workflow",
"app.workflow.actions.transfer_file -> app.workflow.actions"
],
"module_count": 1053,
"module_count": 1054,
"modules": [
"app",
"app.adapters",
@@ -10518,6 +10536,7 @@
"app.chain.transfer.filter",
"app.chain.transfer.format",
"app.chain.transfer.history",
"app.chain.transfer.music",
"app.chain.transfer.plan",
"app.chain.transfer.queue",
"app.chain.transfer.records",

6
tests/fixtures/architecture/startup-performance-baseline.json vendored
View File

@@ -6,7 +6,7 @@
"repeat": 3,
"targets": {
"app.startup.lifecycle": {
"loaded_app_module_count": 573,
"loaded_app_module_count": 574,
"max_ms": 1293.338,
"median_ms": 1156.239,
"min_ms": 1102.806,
@@ -17,7 +17,7 @@
]
},
"app.factory": {
"loaded_app_module_count": 585,
"loaded_app_module_count": 586,
"max_ms": 1127.911,
"median_ms": 1122.382,
"min_ms": 1119.221,
@@ -28,7 +28,7 @@
]
},
"app.main": {
"loaded_app_module_count": 587,
"loaded_app_module_count": 588,
"max_ms": 1188.652,
"median_ms": 1183.509,
"min_ms": 1174.522,

4
tests/test_chain_mixin_contracts.py
View File

@@ -14,7 +14,7 @@ from app.chain._messaging import MessageProcessingMixin, NotificationMixin
from app.chain._music import MusicSubscribeMixin
from app.chain._recognition import RecognitionMixin
from app.chain.subscribe.facade import SubscribeChain
from app.chain.transfer import TransferChain
from app.chain.transfer.facade import TransferChain
from app.chain.transfer.filter import FileFilterMixin
from app.chain.transfer.format import EpisodeFormatMixin
from app.chain.transfer.records import FileKeyMixin, HistoryMatchMixin, ManualHistoryMixin
@@ -69,6 +69,7 @@ def test_domain_mixins_keep_concrete_imports_explicit_until_next_migration() ->
for relative in (
"app/chain/_music.py",
"app/chain/transfer/filter.py",
"app/chain/transfer/music.py",
"app/chain/transfer/format.py",
"app/chain/transfer/records.py",
"app/chain/transfer/retry.py",
@@ -80,6 +81,7 @@ def test_domain_mixins_keep_concrete_imports_explicit_until_next_migration() ->
if node.module.startswith("app.chain.") and node.module not in {
"app.chain._contracts",
"app.chain.transfer.contract",
"app.chain.transfer.music",
}:
violations.append(f"{relative}:{node.module}")

22
tests/test_media_classification_execution.py
View File

@@ -513,6 +513,28 @@ def test_execution_sync_and_async_apply_the_same_enriched_facts() -> None:
assert source.library_category == ""
def test_offline_classification_keeps_policy_without_external_enrichment() -> None:
"""离线结果仍遵守用户分类和手工覆盖，但同步异步都不调用外部补充。"""
source = MusicInfo(title="晴天", artists=["周杰伦"], album="叶惠美")
enrichment = _StaticEnrichment()
service = ClassificationExecutionService(_Runtime(_policy()), enrichment=enrichment)
override = ClassificationSelection(
category_path=["收藏"], source="manual", category_id="music.other",
)
sync_result = service.finalize(source, allow_enrichment=False, effective_override=override)
async_result = asyncio.run(service.async_finalize(
source, allow_enrichment=False, effective_override=override,
))
assert enrichment.calls == []
assert sync_result.classification == async_result.classification
assert sync_result.classification.recommended.category_id == "music.other"
assert sync_result.library_category == async_result.library_category == "收藏"
assert source.media_id is None
assert source.library_category == ""
def test_music_metadata_category_is_never_promoted_without_a_rule() -> None:
"""音乐来源描述分类只能形成事实，目录分类必须来自规则或兜底。"""
music = MusicInfo(

21
tests/test_media_classification_facts.py
View File

@@ -151,3 +151,24 @@ def test_incomplete_identity_is_rejected(
with pytest.raises(ValueError, match="media_source"):
build_classification_facts(media)
def test_local_music_facts_require_explicit_opt_in_and_keep_empty_identity() -> None:
"""完整本地音乐可在离线整理中分类，但不伪造在线身份或改变默认校验。"""
media = MusicInfo(title="晴天", artists=["周杰伦"], album="叶惠美")
with pytest.raises(ValueError, match="media_source"):
build_classification_facts(media)
facts = build_classification_facts(media, allow_local_music=True)
assert facts.identity.media_source == facts.identity.media_id == ""
assert facts.media.type == "音乐"
assert facts.media.title == "晴天"
assert facts.music.artists == ["周杰伦"]
def test_local_music_opt_in_does_not_accept_incomplete_or_video_evidence() -> None:
"""离线许可不能让残缺音乐标签或缺身份影视进入已完成分类。"""
for media in (MusicInfo(title="Track 01"), MediaInfo(title="Example", type=MediaType.MOVIE)):
with pytest.raises(ValueError, match="media_source"):
build_classification_facts(media, allow_local_music=True)

136
tests/test_music_local_transfer.py Normal file
View File

@@ -0,0 +1,136 @@
"""真实标签经过整理批次、用户分类和目标命名后无需在线识别。"""
import hashlib
import shutil
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock
import pytest
from mutagen.flac import FLAC
from app.application.classification.execution import ClassificationExecutionService
from app.application.messaging.message import TemplateHelper
from app.chain.media import MediaChain
from app.domain.meta.metamusic import MetaMusic
from app.domain.music import music_tags_are_usable
from app.modules.filemanager.transhandler import TransHandler
from app.runtime.config import ConfigModel
from app.schemas.category import CategoryConfig, ClassificationPolicy
from app.schemas.file import FileItem
from app.schemas.system import TransferDirectoryConf
from app.schemas.transfer import TransferInfo
from app.schemas.types import MediaType
from tests.test_transfer_sync_extra_files import bind_empty_history_repositories, make_fileitem, make_transfer_chain
def _tagged_files(directory: Path) -> list[Path]:
"""生成名字无意义且父目录年份错误的实际双碟 FLAC，身份由标签提供。"""
directory.mkdir()
files = []
for disc, title in ((1, "晴天"), (2, "以父之名")):
path = directory / f"file-{disc}.flac"
shutil.copyfile(Path(__file__).parent / "fixtures/audio/silence.flac", path)
audio = FLAC(path)
audio.update({
"title": [title], "artist": ["周杰伦"], "album": ["叶惠美"],
"albumartist": ["周杰伦"], "date": ["2003"],
"tracknumber": ["1"], "tracktotal": ["1"],
"discnumber": [str(disc)], "disctotal": ["2"],
})
audio.save()
files.append(path)
return files
def _classification_service():
"""提供自定义未分类去向和禁止调用的外部补充端口。"""
policy = ClassificationPolicy.model_validate({
"schema_version": 2, "revision": 1, "enrichment_mode": "enrich_missing",
"categories": [{"id": "music.local", "media_type": "音乐", "name": "收藏", "path": ["收藏"]}],
"rules": [], "fallbacks": {"音乐": "music.local"},
})
runtime = SimpleNamespace(active_policy=lambda: policy, legacy_config=lambda: CategoryConfig())
enrichment = Mock()
enrichment.async_enrich = AsyncMock()
return ClassificationExecutionService(runtime, enrichment=enrichment), enrichment
@pytest.mark.parametrize("mtype", [None, MediaType.MUSIC])
def test_complete_tags_preview_offline_and_preserve_custom_category(tmp_path, monkeypatch, mtype):
"""自动与显式音乐批次均以真实标签完成目标命名，双碟不碰撞且源内容不变。"""
source = tmp_path / "2022-wrong-name"
paths = _tagged_files(source)
hashes = [hashlib.sha256(path.read_bytes()).digest() for path in paths]
items = [make_fileitem(str(path)) for path in paths]
chain = make_transfer_chain()
bind_empty_history_repositories(chain)
chain.eventmanager = SimpleNamespace(send_event=lambda *_args, **_kwargs: None)
service, enrichment = _classification_service()
chain.classification_service = service
monkeypatch.setattr(chain, "_TransferChain__get_trans_fileitems", lambda *_args, **_kwargs: [(item, False) for item in items])
monkeypatch.setattr(chain, "_TransferChain__put_to_jobview", lambda _task: True)
monkeypatch.setattr(chain, "_register_scrape_batch_task", lambda _task: None)
monkeypatch.setattr(chain, "_close_scrape_batch", lambda _batch: None)
monkeypatch.setattr("app.chain.transfer.workflow.StorageChain.get_item", lambda _self, item: item)
monkeypatch.setattr("app.chain.transfer.workflow.get_configured_system_config", lambda: SimpleNamespace(get=lambda _key: None))
monkeypatch.setattr("app.modules.filemanager.transhandler.eventmanager.send_event", Mock(return_value=None))
online = []
for name in ("recognize_music_album_directory", "recognize_music_by_path", "recognize_by_meta", "recognize_media"):
operation = Mock(side_effect=AssertionError("完整标签整理不应请求在线识别"))
monkeypatch.setattr(MediaChain, name, operation)
online.append(operation)
destinations = []
def preview_plan(task, **_kwargs):
"""执行真实模板渲染而不写库或复制文件，检查批次传入的最终音乐上下文。"""
assert task.mediainfo.library_category == "收藏"
assert task.mediainfo.media_id is None
context = TemplateHelper().builder.build(
meta=task.meta, mediainfo=task.mediainfo, file_extension=".flac", include_raw_objects=False,
)
target = TransHandler.get_rename_path(
ConfigModel().MUSIC_RENAME_FORMAT, context, path=tmp_path / "library" / "收藏",
)
destinations.append(target.relative_to(tmp_path / "library").as_posix())
return TransferInfo(success=True, fileitem=task.fileitem, target_item=make_fileitem(str(target)))
monkeypatch.setattr(chain, "_plan_checkpoint_and_execute", preview_plan)
state, preview = chain.do_transfer(
fileitem=FileItem(storage="local", path=str(source) + "/", type="dir", name=source.name),
target_directory=TransferDirectoryConf(
library_path=str(tmp_path / "library"), library_storage="local", library_category_folder=True,
),
mtype=mtype, force=True, preview=True,
)
assert state is True
assert preview["summary"] == {"total": 2, "success": 2, "failed": 0}
assert destinations == [
"收藏/周杰伦/叶惠美 (2003)/Disc 1/01 - 晴天.flac",
"收藏/周杰伦/叶惠美 (2003)/Disc 2/01 - 以父之名.flac",
]
assert [hashlib.sha256(path.read_bytes()).digest() for path in paths] == hashes
for operation in online:
operation.assert_not_called()
enrichment.enrich.assert_not_called()
enrichment.async_enrich.assert_not_called()
@pytest.mark.parametrize("field,value", [
("title", "Track 01"), ("title", "01"), ("title", "Unknown"), ("title", "坏\ufffd名"),
("title", "www.example.test"), ("album", None), ("artists", []), ("album", "Unknown Album"),
])
def test_incomplete_or_placeholder_tags_require_more_evidence(field, value):
"""路径可补字段不能使不完整或占位标签获得直接整理资格。"""
meta = MetaMusic(title="晴天", artists=["周杰伦"], album="叶惠美")
setattr(meta, field, value)
assert music_tags_are_usable(meta) is False
def test_complete_tags_do_not_require_year_or_remote_id():
"""真实名称足以本地整理，数字专辑名与未知年份不构成失败。"""
meta = MetaMusic(title="Welcome to New York", artists=["Taylor Swift"], album="1989")
assert music_tags_are_usable(meta) is True

1
tests/test_transfer_architecture.py
View File

@@ -40,6 +40,7 @@ def test_transfer_package_retires_monolith_and_keeps_thin_facade() -> None:
"filter.py",
"format.py",
"history.py",
"music.py",
"plan.py",
"queue.py",
"records.py",

12
tests/test_transfer_music_album_selection.py
View File

@@ -114,10 +114,16 @@ def _shared_recording() -> MusicInfo:
def _patch_local_music_reads(monkeypatch, local_metas: dict[Path, MetaMusic]) -> None:
"""把批次规划中的本地标签读取固定为测试证据。"""
"""为在线匹配用例提供缺曲名的部分标签，完整曲名由路径解析测试桩补齐。"""
def partial_tags(path):
"""保留专辑分组证据，让本组用例仍实际执行远端候选对齐分支。"""
meta = deepcopy(local_metas[Path(path)])
meta.title = None
return meta
monkeypatch.setattr(
"app.chain.transfer.filter.AudioMetadataHelper.read_tags",
lambda path: deepcopy(local_metas[Path(path)]),
"app.chain.transfer.music.AudioMetadataHelper.read_tags",
partial_tags,
)
monkeypatch.setattr(
MediaChain,