GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-28 19:47:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(meta): stabilize recognition contract

Browse Source
This commit is contained in:
jxxghp
2026-08-25 08:12:04 +08:00
parent fcf6f0c7df
commit 8e99abd678
6 changed files with 340 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/domain/meta/metabase.py
+100 -32
View File
@@ -1,7 +1,7 @@
import logging
import traceback
from dataclasses import dataclass
from typing import Union, Optional, List, Self
from dataclasses import asdict, dataclass
from typing import Any, Union, Optional, List, Self
import cn2an
import regex as re
@@ -40,6 +40,100 @@ VIDEO_BIT_RE = re.compile(
re.IGNORECASE,
)
_META_OPTIONAL_MERGE_FIELDS = (
"resource_type",
"resource_pix",
"resource_team",
"customization",
"resource_effect",
"web_source",
"video_encode",
"video_bit",
"audio_encode",
"part",
"episode_group",
)
@dataclass(frozen=True, slots=True)
class MetaInfoSnapshot:
"""MetaInfo 解析器的不可变稳定输出，用于跨实现等价校验和安全缓存。"""
kind: str
isfile: bool
title: str
org_string: Optional[str]
subtitle: Optional[str]
type: str
cn_name: Optional[str]
en_name: Optional[str]
original_name: Optional[str]
year: Optional[str]
total_season: int
begin_season: Optional[int]
end_season: Optional[int]
total_episode: int
begin_episode: Optional[int]
end_episode: Optional[int]
part: Optional[str]
resource_type: Optional[str]
resource_effect: Optional[str]
resource_pix: Optional[str]
resource_team: Optional[str]
customization: Optional[str]
web_source: Optional[str]
video_encode: Optional[str]
video_bit: Optional[str]
audio_encode: Optional[str]
apply_words: tuple[str, ...]
media_source: Optional[str]
media_id: Optional[str]
episode_group: Optional[str]
fps: Optional[int]
@classmethod
def from_meta(cls, meta: "MetaBase") -> "MetaInfoSnapshot":
"""从兼容的可变 MetaBase 对象提取不包含临时状态的完整契约。"""
media_type = getattr(meta.type, "value", meta.type)
media_source = getattr(meta.media_source, "value", meta.media_source)
return cls(
kind="anime" if type(meta).__name__ == "MetaAnime" else "video",
isfile=bool(meta.isfile),
title=meta.title or "",
org_string=meta.org_string,
subtitle=meta.subtitle,
type=media_type,
cn_name=meta.cn_name,
en_name=meta.en_name,
original_name=meta.original_name,
year=meta.year,
total_season=meta.total_season,
begin_season=meta.begin_season,
end_season=meta.end_season,
total_episode=meta.total_episode,
begin_episode=meta.begin_episode,
end_episode=meta.end_episode,
part=meta.part,
resource_type=meta.resource_type,
resource_effect=meta.resource_effect,
resource_pix=meta.resource_pix,
resource_team=meta.resource_team,
customization=meta.customization,
web_source=meta.web_source,
video_encode=meta.video_encode,
video_bit=meta.video_bit,
audio_encode=meta.audio_encode,
apply_words=tuple(meta.apply_words or ()),
media_source=media_source,
media_id=meta.media_id,
episode_group=meta.episode_group,
fps=meta.fps,
)
def to_dict(self) -> dict[str, Any]:
"""返回便于差异报告和序列化的独立字典。"""
return asdict(self)
@dataclass
class MetaBase(object):
@@ -652,45 +746,19 @@ class MetaBase(object):
self.begin_episode = meta.begin_episode
self.end_episode = meta.end_episode
self.total_episode = meta.total_episode
# 版本
if not self.resource_type:
self.resource_type = meta.resource_type
# 分辨率
if not self.resource_pix:
self.resource_pix = meta.resource_pix
# 制作组/字幕组
if not self.resource_team:
self.resource_team = meta.resource_team
# 自定义占位符
if not self.customization:
self.customization = meta.customization
# 特效
if not self.resource_effect:
self.resource_effect = meta.resource_effect
# 视频编码
if not self.video_encode:
self.video_encode = meta.video_encode
# 视频位深
if not self.video_bit:
self.video_bit = meta.video_bit
# 音频编码
if not self.audio_encode:
self.audio_encode = meta.audio_encode
# 普通可选字段统一遵循文件优先、父目录补空，新增字段只维护一份策略。
for field_name in _META_OPTIONAL_MERGE_FIELDS:
if not getattr(self, field_name):
setattr(self, field_name, getattr(meta, field_name))
# 帧率信息
if not self.fps:
self.fps = meta.fps
# Part
if not self.part:
self.part = meta.part
# 媒体身份必须原子合并，不能将不同目录层级的来源和ID拼成一对
current_source, current_id = resolve_media_identity(media=self)
if current_source and current_id:
self.media_source, self.media_id = current_source, current_id
else:
self.media_source, self.media_id = resolve_media_identity(media=meta)
# 剧集组
if not self.episode_group and meta.episode_group:
self.episode_group = meta.episode_group
def to_dict(self):
"""
app/domain/meta/metavideo.py
+1
View File
@@ -127,6 +127,7 @@ class MetaVideo(MetaBase):
and title.isdigit() \
and len(title) < 5:
self.begin_episode = int(title)
self.total_episode = 1
self.type = MediaType.TV
return
# 全名为Season xx 及 Sxx 直接返回
app/domain/metainfo.py
+80 -32
View File
@@ -1,8 +1,9 @@
import hashlib
import logging
from dataclasses import dataclass
from pathlib import Path
from functools import lru_cache
from typing import Tuple, List, Optional
from typing import Mapping, Tuple, List, Optional
import regex as re
@@ -88,6 +89,18 @@ _LEGACY_ID_KEYS = (
)
@dataclass(frozen=True, slots=True)
class _PreparedMetaInput:
"""Python 回退解析的阶段输入，保留原文与预处理结果之间的明确边界。"""
original_title: str
parsed_title: str
subtitle: Optional[str]
isfile: bool
apply_words: tuple[str, ...]
explicit_metainfo: Mapping[str, object]
def _empty_metainfo() -> dict:
"""
返回媒体标签的默认结构，避免不同识别请求之间共享可变状态。
@@ -263,37 +276,38 @@ def _find_metainfo_python(title: str) -> Tuple[str, dict]:
return title, _normalize_metainfo_identity(metainfo)
def _build_meta_info(
def _prepare_meta_input(
title: str,
subtitle: Optional[str] = None,
custom_words: List[str] = None,
) -> MetaBase:
) -> _PreparedMetaInput:
"""
根据标题构造元数据
应用识别词、显式标签和文件后缀规则，生成稳定的解析阶段输入。
"""
# 原标题
org_title = title
# 预处理标题
title, apply_words = WordsMatcher().prepare(title, custom_words=custom_words)
# 获取标题中媒体信息
title, metainfo = find_metainfo(title)
# 判断是否处理文件
original_title = title
parsed_title, apply_words = WordsMatcher().prepare(title, custom_words=custom_words)
# 完整 Rust 入口已经失败或被禁用，参考实现不得再次跨边界调用部分 Rust 解析器。
parsed_title, explicit_metainfo = _find_metainfo_python(parsed_title)
media_exts = get_media_extensions()
title_path = Path(title) if title else None
title_path = Path(parsed_title) if parsed_title else None
if title_path and title_path.suffix.lower() in media_exts:
isfile = True
# 去掉后缀
title = title_path.stem
parsed_title = title_path.stem
else:
isfile = False
# 识别
meta = MetaAnime(title, subtitle, isfile) if is_anime(title) else MetaVideo(title, subtitle, isfile)
# 记录原标题
meta.title = org_title
# 记录使用的识别词
meta.apply_words = apply_words or []
# 修正媒体信息
media_source, media_id = resolve_media_identity(media=metainfo)
return _PreparedMetaInput(
original_title=original_title,
parsed_title=parsed_title,
subtitle=subtitle,
isfile=isfile,
apply_words=tuple(apply_words or ()),
explicit_metainfo=explicit_metainfo,
)
def _apply_explicit_metainfo(meta: MetaBase, metainfo: Mapping[str, object]) -> None:
"""以显式标签覆盖推断字段，保持用户声明拥有最高优先级。"""
media_source, media_id = resolve_media_identity(media=dict(metainfo))
if media_source and media_id:
meta.media_source = media_source
meta.media_id = media_id
@@ -313,6 +327,42 @@ def _build_meta_info(
meta.end_episode = metainfo['end_episode']
if metainfo.get('total_episode') is not None:
meta.total_episode = metainfo['total_episode']
def _build_meta_info(
title: str,
subtitle: Optional[str] = None,
custom_words: List[str] = None,
) -> MetaBase:
"""按准备、分类解析、显式覆盖三个阶段构造 Python MetaInfo。"""
prepared = _prepare_meta_input(title, subtitle, custom_words)
meta = MetaAnime(
prepared.parsed_title,
prepared.subtitle,
prepared.isfile,
) if is_anime(prepared.parsed_title) else MetaVideo(
prepared.parsed_title,
prepared.subtitle,
prepared.isfile,
)
meta.title = prepared.original_title
meta.apply_words = list(prepared.apply_words)
_apply_explicit_metainfo(meta, prepared.explicit_metainfo)
return meta
def _build_python_meta_info(
title: str,
subtitle: Optional[str] = None,
custom_words: List[str] = None,
) -> MetaBase:
"""构造并完成 original_name 的纯 Python 参考解析结果。"""
meta = _build_meta_info(title=title, subtitle=subtitle, custom_words=custom_words)
if meta.apply_words:
original_meta = _build_meta_info(title=title, subtitle=subtitle)
meta.original_name = original_meta.name or meta.name
else:
meta.original_name = meta.name or None
return meta
@@ -473,13 +523,11 @@ def MetaInfo(title: str, subtitle: Optional[str] = None, custom_words: List[str]
)
if rust_meta:
return rust_meta
meta = _build_meta_info(title=title, subtitle=subtitle, custom_words=custom_words)
if meta.apply_words:
original_meta = _build_meta_info(title=title, subtitle=subtitle)
meta.original_name = original_meta.name or meta.name
else:
meta.original_name = meta.name
return meta
return _build_python_meta_info(
title=title,
subtitle=subtitle,
custom_words=custom_words,
)
def MetaInfoPath(path: Path, custom_words: List[str] = None, force_video: bool = False) -> MetaBase:
@@ -512,16 +560,16 @@ def MetaInfoPath(path: Path, custom_words: List[str] = None, force_video: bool =
if rust_meta:
return rust_meta
# 文件元数据，不包含后缀
file_meta = MetaInfo(title=path.name, custom_words=custom_words)
file_meta = _build_python_meta_info(title=path.name, custom_words=custom_words)
if should_use_parent_title_for_file_stem(path.stem, path.parent.name, file_meta):
clear_parsed_title_for_parent_merge(file_meta)
# 上级目录元数据
dir_meta = MetaInfo(title=path.parent.name, custom_words=custom_words)
dir_meta = _build_python_meta_info(title=path.parent.name, custom_words=custom_words)
if file_meta.type == MediaType.TV or dir_meta.type != MediaType.TV:
# 合并元数据
file_meta.merge(dir_meta)
# 上上级目录元数据
root_meta = MetaInfo(title=path.parent.parent.name, custom_words=custom_words)
root_meta = _build_python_meta_info(title=path.parent.parent.name, custom_words=custom_words)
if file_meta.type == MediaType.TV or root_meta.type != MediaType.TV:
# 合并元数据
file_meta.merge(root_meta)