Files
MoviePilot/app/domain/context.py
Aqr-K bcbf79f030 fix(media): preserve recognition and rule mutation identities (#7144)
* fix(media): preserve recognition and rule mutation identities

* test: cover typed classification facts with alias policies

---------

Co-authored-by: jxxghp <jxxghp@gmail.com>
2026-10-23 07:33:27 +08:00

2461 lines
94 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
聚合媒体、音乐、种子和字幕的领域上下文、序列化契约及来源元数据投影入口。
"""
import re
import typing
from collections.abc import Mapping
from dataclasses import asdict, dataclass, field, fields
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional, Self, Set, Union
import yaml
from yaml.constructor import ConstructorError
from yaml.events import AliasEvent
from yaml.nodes import MappingNode
from app.domain.meta.metabase import MetaBase
from app.domain.meta.metamusic import (
MetaMusic,
audio_quality_score,
audio_quality_tier,
format_audio_quality,
infer_audio_lossless,
music_credit_values,
normalize_audio_format,
)
from app.domain.metainfo import MetaInfo
from app.foundation import temporal as time_tools
from app.schemas.category import ClassificationFactValue, ClassificationResult
from app.schemas.common import JsonData
from app.schemas.media import normalize_media_source, resolve_media_identity
from app.schemas.types import (
MUSIC_ENTITY_ALBUM,
MUSIC_ENTITY_ARTIST,
MUSIC_ENTITY_RECORDING,
MediaSource,
MediaType,
)
class _LyricsfileSafeLoader(yaml.SafeLoader):
"""
在 SafeLoader 基础上拒绝 Lyricsfile 规范禁止的重复映射键。
"""
def construct_mapping(
self, node: MappingNode, deep: bool = False
) -> dict[Any, Any]:
"""
构造映射并在解析阶段拒绝重复键。
"""
if not isinstance(node, MappingNode):
raise ConstructorError(None, None, "Lyricsfile 映射节点无效", node.start_mark)
keys = set()
for key_node, _value_node in node.value:
key = self.construct_object(key_node, deep=deep)
try:
duplicated = key in keys
except TypeError as err:
raise ConstructorError(None, None, "Lyricsfile 映射键必须可哈希", key_node.start_mark) from err
if duplicated:
raise ConstructorError(None, None, f"Lyricsfile 存在重复键:{key}", key_node.start_mark)
keys.add(key)
return typing.cast(
dict[Any, Any], super().construct_mapping(node, deep=deep)
)
def configure_tmdb_image_url_builder(
builder: Callable[[str], Optional[str]],
) -> None:
"""
兼容旧入口,将 TMDB 图片地址构造器交给来源投影 owner。
"""
from app.domain.projection.tmdb import configure_image_url_builder
configure_image_url_builder(builder)
def _project_tmdb(
state: Mapping[str, Any],
info: Mapping[str, Any],
) -> dict[str, Any]:
"""
按需加载 TMDB 投影 owner,避免领域上下文冷导入全部来源实现。
"""
from app.domain.projection.tmdb import project
return project(state, info)
def _project_douban(
state: Mapping[str, Any],
info: Mapping[str, Any],
) -> dict[str, Any]:
"""
按需加载豆瓣投影 owner,避免领域上下文冷导入全部来源实现。
"""
from app.domain.projection.douban import project
return project(state, info)
def _project_bangumi(
state: Mapping[str, Any],
info: Mapping[str, Any],
) -> dict[str, Any]:
"""
按需加载 Bangumi 投影 owner,避免领域上下文冷导入全部来源实现。
"""
from app.domain.projection.bangumi import project
return project(state, info)
def _project_anilist(
state: Mapping[str, Any],
info: Mapping[str, Any],
) -> dict[str, Any]:
"""
按需加载 AniList 投影 owner,避免领域上下文冷导入全部来源实现。
"""
from app.domain.projection.anilist import project
return project(state, info)
def _validate_music_type(value: object) -> None:
"""
校验音乐模型类型字段,仅接受音乐或空值。
"""
if value in {None, MediaType.MUSIC, MediaType.MUSIC.value, "music"}:
return
raise ValueError(f"不支持的音乐媒体类型:{value}")
def _music_string_list(value: object) -> list[str]:
"""
将音乐标签原始值归一为非空字符串列表,兼容单值、列表与逗号分隔。
"""
if value is None:
return []
if isinstance(value, str):
return [value] if value else []
if isinstance(value, (list, tuple, set)):
return [str(item) for item in value if str(item)]
return [str(value)]
def _music_aligned_list(value: object) -> list[str]:
"""
保留原始位置的字符串列表,用于与艺术家名称按下标对应的 ID 列表。
"""
if value is None:
return []
if isinstance(value, (list, tuple)):
return [str(item or "") for item in value]
return [str(value or "")]
def _music_optional_int(value: object) -> int | None:
"""
将音乐技术参数安全转换为整数,空值与非数字返回 None。
"""
if value in {None, ""}:
return None
try:
return int(str(value))
except (TypeError, ValueError):
return None
def _music_optional_float(value: object) -> float:
"""
将音乐评分安全转换为浮点数,空值与异常返回 0.0。
"""
if value in {None, ""}:
return 0.0
try:
return float(str(value))
except (TypeError, ValueError):
return 0.0
def _music_year_of(release_date: object) -> int | None:
"""
从 MusicBrainz 可变精度日期(YYYY / YYYY-MM / YYYY-MM-DD)提取年份。
"""
text = str(release_date or "")[:4]
return int(text) if text.isdigit() else None
def _music_init_values(model: type, data: dict[str, Any]) -> dict[str, Any]:
"""
按 dataclass 可初始化字段过滤字典,避免传入非法构造参数。
"""
init_names = {item.name for item in fields(model) if item.init}
return {key: value for key, value in data.items() if key in init_names}
def _classification_result(value: object) -> ClassificationResult | None:
"""
将分类结果字典恢复为独立模型,避免领域对象共享可变标签列表。
"""
if value is None:
return None
if isinstance(value, ClassificationResult):
return value.model_copy(deep=True)
if isinstance(value, Mapping):
return ClassificationResult.model_validate(dict(value))
raise TypeError("分类结果必须是 ClassificationResult、字典或空值")
def _classification_payload(value: ClassificationResult | None) -> dict[str, Any] | None:
"""
把分类结果转换为可持久化字典。
"""
return value.model_dump(mode="json") if value else None
def _music_category_tokens(value: object) -> list[str]:
"""
拆分并规范化音乐来源分类文本,供旧载荷语义判定。
"""
return [
re.sub(r"[\s_-]+", " ", part).strip().casefold()
for part in re.split(r"[/\|,;、,]+", str(value or ""))
if part.strip()
]
def _legacy_music_category_is_metadata(
category: object,
*,
album_type: object = None,
secondary_types: object = None,
genres: object = None,
tags: object = None,
classification: ClassificationResult | None = None,
) -> bool:
"""
判断旧音乐 category 是否是来源描述,而不是人工库分类。
旧音乐模型曾把专辑类型、流派写入 category,同时订阅人工分类也可能
覆盖同一字段。优先用分类结果确认库分类;其余仅在文本能由标准音乐
元数据解释时判为来源描述,无法解释时保守保留为库分类。
"""
category_tokens = _music_category_tokens(category)
if not category_tokens:
return False
selection = None
if classification:
selection = classification.effective or classification.recommended
if selection and selection.category_path:
category_text = " / ".join(category_tokens)
path_tokens = _music_category_tokens(" / ".join(selection.category_path))
if category_text in {
" / ".join(path_tokens),
path_tokens[-1] if path_tokens else "",
}:
return False
evidence = [str(album_type or "")]
for value in (secondary_types, genres, tags):
evidence.extend(_music_string_list(value))
evidence_tokens = {
token
for value in evidence
for token in _music_category_tokens(value)
}
category_text = " ".join(category_tokens)
if any(
token and (token in category_text or category_text in token)
for token in evidence_tokens
):
return True
source_descriptors = {
"album",
"ep",
"single",
"broadcast",
"other",
"live",
"compilation",
"soundtrack",
"spoken word",
"interview",
"audiobook",
"audio book",
"audio drama",
"remix",
"dj mix",
"mixtape",
"street",
"demo",
"person",
"group",
"orchestra",
"choir",
"character",
}
return all(token in source_descriptors for token in category_tokens)
def _resolve_music_categories(
*,
category: object,
library_category: object,
metadata_category: object,
album_type: object = None,
secondary_types: object = None,
genres: object = None,
tags: object = None,
classification: ClassificationResult | None = None,
) -> tuple[str, str]:
"""
解析新旧音乐分类字段,新字段存在时禁止来源描述升级为库分类。
"""
compatible = str(category or "").strip()
library = str(library_category or "").strip()
metadata = str(metadata_category or "").strip()
if library or metadata:
return library or compatible, metadata
if not compatible:
return "", ""
if _legacy_music_category_is_metadata(
compatible,
album_type=album_type,
secondary_types=secondary_types,
genres=genres,
tags=tags,
classification=classification,
):
return "", compatible
return compatible, ""
@dataclass
class MusicLyrics:
"""
标准化单曲歌词候选,保留来源匹配度和 Lyricsfile 原始内容。
:param provider: 歌词提供方名称
:param provider_id: 提供方内的歌词或曲目标识;未提供时为 None
:param instrumental: 是否明确标记为纯音乐
:param plain_lyrics: 不含时间轴的歌词;未提供时为 None
:param synced_lyrics: 带时间轴的歌词;未提供时为 None
:param lyricsfile: 保留提供方的 Lyricsfile 原始内容
:param language: 歌词语言;未知时为 None
:param match_score: 与目标曲目的匹配得分,用于候选排序
:param provider_priority: 提供方排序优先级,用于同分候选选择
"""
provider: str
provider_id: str | None = None
instrumental: bool = False
plain_lyrics: str | None = None
synced_lyrics: str | None = None
lyricsfile: str | None = None
language: str | None = None
match_score: int = 0
provider_priority: int = 0
_lyricsfile_max_bytes = 1024 * 1024
_lyricsfile_max_lines = 10000
_lyricsfile_max_nodes = 50000
_lyricsfile_max_depth = 20
def __post_init__(self) -> None:
"""
补全 Lyricsfile 中可安全降级为 LRC 或纯文本的内容。
"""
if not self.lyricsfile:
return
parsed = self._parse_lyricsfile(self.lyricsfile)
if not parsed:
return
metadata_value = parsed.get("metadata")
metadata: dict[str, Any] = (
metadata_value if isinstance(metadata_value, dict) else {}
)
self.instrumental = self.instrumental or bool(
metadata.get("instrumental", parsed.get("instrumental"))
)
self.language = self.language or self._optional_text(
metadata.get("language", parsed.get("language"))
)
if not self.synced_lyrics:
self.synced_lyrics = self._lyricsfile_to_lrc(parsed)
if not self.plain_lyrics:
self.plain_lyrics = self._lyricsfile_to_plain(parsed)
@property
def content(self) -> str | None:
"""
优先返回同步歌词,不存在时回退到纯文本歌词。
"""
return self.synced_lyrics or self.plain_lyrics
@property
def extension(self) -> str | None:
"""
根据歌词内容返回适合播放器扫描的旁挂文件扩展名。
"""
if self.synced_lyrics:
return ".lrc"
if self.plain_lyrics:
return ".txt"
return None
@property
def quality_rank(self) -> int:
"""
返回可比较的歌词质量等级,逐字同步高于逐行同步和纯文本。
"""
if self.lyricsfile and self._lyricsfile_has_words(self.lyricsfile):
return 4
if self.synced_lyrics:
return 3
if self.plain_lyrics:
return 1
if self.instrumental:
return 1
return 0
@property
def identity_key(self) -> tuple[str, str, str]:
"""
构造候选去重键,兼容未提供来源 ID 的插件结果。
"""
return (
self.provider.casefold(),
self.provider_id or "",
self.content or self.lyricsfile or "",
)
@classmethod
def _parse_lyricsfile(cls, content: str) -> dict[str, Any] | None:
"""
安全解析受大小和行数约束的 Lyricsfile YAML,并拒绝锚点别名。
"""
if len(content.encode("utf-8")) > cls._lyricsfile_max_bytes:
return None
try:
if any(isinstance(event, AliasEvent) for event in yaml.parse(content)):
return None
payload = yaml.load(content, Loader=_LyricsfileSafeLoader)
except yaml.YAMLError:
return None
if not isinstance(payload, dict):
return None
if payload.get("version") != "1.0" or not isinstance(payload.get("metadata"), dict):
return None
if not cls._lyricsfile_value_is_safe(payload):
return None
lines = payload.get("lines")
if isinstance(lines, list) and len(lines) > cls._lyricsfile_max_lines:
return None
return payload if cls._lyricsfile_structure_is_valid(payload) else None
@classmethod
def _lyricsfile_structure_is_valid(cls, payload: dict[str, Any]) -> bool:
"""
校验 Lyricsfile 1.0 的必需元数据、歌词形状和毫秒时间范围。
"""
metadata = payload.get("metadata")
if not isinstance(metadata, dict):
return False
if not cls._optional_text(metadata.get("title")) or not cls._optional_text(metadata.get("artist")):
return False
lines = payload.get("lines")
plain = payload.get("plain")
if lines is not None and not isinstance(lines, list):
return False
if plain is not None and not isinstance(plain, str):
return False
if metadata.get("instrumental") is True:
return not lines and not str(plain or "").strip()
if not lines and not str(plain or "").strip():
return False
return all(cls._lyricsfile_line_is_valid(line) for line in lines or [])
@classmethod
def _lyricsfile_line_is_valid(cls, line: Any) -> bool:
"""
校验单行与逐字时间戳,拒绝布尔值伪装整数和倒序区间。
"""
if not isinstance(line, dict) or not isinstance(line.get("text"), str):
return False
start = line.get("start_ms")
end = line.get("end_ms")
if isinstance(start, bool) or not isinstance(start, int) or start < 0:
return False
if end is not None and (
isinstance(end, bool) or not isinstance(end, int) or end < start
):
return False
words = line.get("words")
if words is None:
return True
if not isinstance(words, list):
return False
for word in words:
if not isinstance(word, dict) or not isinstance(word.get("text"), str):
return False
word_start = word.get("start_ms")
word_end = word.get("end_ms")
if isinstance(word_start, bool) or not isinstance(word_start, int) or word_start < 0:
return False
if word_end is not None and (
isinstance(word_end, bool)
or not isinstance(word_end, int)
or word_end < word_start
):
return False
return True
@classmethod
def _lyricsfile_value_is_safe(cls, value: Any, depth: int = 0, count: Optional[list[int]] = None) -> bool:
"""
限制 YAML 解析后的类型、深度和节点数,避免外部文档消耗过量资源。
"""
if depth > cls._lyricsfile_max_depth:
return False
counter = count if count is not None else [0]
counter[0] += 1
if counter[0] > cls._lyricsfile_max_nodes:
return False
if value is None or isinstance(value, (str, int, bool)):
return True
if isinstance(value, list):
return all(cls._lyricsfile_value_is_safe(item, depth + 1, counter) for item in value)
if isinstance(value, dict):
return all(
isinstance(key, str)
and cls._lyricsfile_value_is_safe(item, depth + 1, counter)
for key, item in value.items()
)
return False
@classmethod
def _lyricsfile_has_words(cls, content: str) -> bool:
"""
判断 Lyricsfile 是否包含逐字时间轴。
"""
payload = cls._parse_lyricsfile(content)
return bool(
payload
and any(
isinstance(line, dict) and isinstance(line.get("words"), list)
for line in payload.get("lines") or []
)
)
@staticmethod
def _optional_text(value: Any) -> str | None:
"""
把可选标量规范化为非空文本。
"""
text = str(value or "").strip()
return text or None
@classmethod
def _lyricsfile_to_plain(cls, payload: dict[str, Any]) -> str | None:
"""
从 Lyricsfile plain 或时间行生成播放器可读纯文本。
"""
plain = cls._optional_text(payload.get("plain"))
if plain:
return plain
texts = []
for line in payload.get("lines") or []:
if not isinstance(line, dict):
continue
text = cls._optional_text(line.get("text"))
if not text and isinstance(line.get("words"), list):
text = "".join(
str(word.get("text") or "")
for word in line["words"]
if isinstance(word, dict)
).strip() or None
if text:
texts.append(text)
return "\n".join(texts) or None
@classmethod
def _lyricsfile_to_lrc(cls, payload: dict[str, Any]) -> str | None:
"""
把 Lyricsfile 行级毫秒时间轴转换为通用 LRC。
"""
output = []
for line in payload.get("lines") or []:
if not isinstance(line, dict):
continue
start_ms = _music_optional_int(line.get("start_ms"))
if start_ms is None:
continue
start_ms = max(start_ms, 0)
text = cls._optional_text(line.get("text"))
if not text and isinstance(line.get("words"), list):
text = "".join(
str(word.get("text") or "")
for word in line["words"]
if isinstance(word, dict)
).strip() or None
if not text:
continue
minutes, remainder = divmod(start_ms, 60000)
seconds = remainder / 1000
output.append(f"[{minutes:02d}:{seconds:05.2f}]{text}")
return "\n".join(output) or None
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Self:
"""
从模块或插件返回字典恢复标准歌词对象。
"""
values = _music_init_values(cls, data)
values["provider"] = str(values.get("provider") or "")
values["provider_id"] = (
str(values["provider_id"])
if values.get("provider_id") is not None
else None
)
values["instrumental"] = bool(values.get("instrumental"))
return cls(**values)
@dataclass
class MusicInfo:
"""
标准化音乐元数据信息。
:ivar type: 固定为音乐媒体类型,不接受初始化参数
:param media_source: 媒体主身份的数据来源,与 media_id 配对使用
:param media_id: 当前来源的原生媒体标识,与 media_source 配对使用
:param music_type: 音乐实体类型,默认表示单曲录音
:param title: 音乐实体的标题;未知时为 None
:param artists: 参与署名的艺术家名称,保留来源顺序
:param artist_ids: 艺术家标准 ID,顺序与 artists 一致,供详情页关联跳转
:param album: 曲目所属专辑名称;未知时为 None
:param album_artist: 专辑层级的艺术家署名;未知时为 None
:param album_id: 所属专辑的 MusicBrainz 发行组标识
:param musicbrainz_release_id: 发行、发行组和发行曲目身份独立于 Recording 主身份,不能相互代用
:param musicbrainz_release_group_id: MusicBrainz 发行组标识,用于关联同一专辑的不同版本
:param musicbrainz_release_track_id: MusicBrainz 发行版本中的曲目标识,不等同于录音标识
:param album_type: 专辑主类型:Album、EP、Single 等
:param secondary_types: 专辑副类型:Live、Compilation、Soundtrack 等
:param year: 展示或匹配使用的年份;未知时为 None
:param original_year: 作品最早发行年份;未知时为 None
:param release_year: 当前发行版本的年份;未知时为 None
:param release_date: 来源提供的发行日期文本;未知时为 None
:param release_status: 发行状态,例如正式发行或宣传版本;未知时为 None
:param disc_number: 曲目所属碟号;未知时为 None
:param track_number: 曲目在所属碟中的序号;未知时为 None
:param total_tracks: 所属发行版本的曲目总数;未知时为 None
:param total_discs: 发行版本的总碟数;未知时为 None
:param duration: 曲目时长,单位为秒
:param isrc: 国际标准录音代码;未知时为 None
:param cover_url: 封面图片链接;未提供封面时为 None
:param lyrics: 曲目的歌词文本;尚未获取时为 None
:param version: 曲目的版本说明,例如现场版或混音版
:param audio_format: 音频编码或容器格式;未知时为 None
:param audio_lossless: 是否为无损音频;未知时为 None
:param bit_depth: 音频采样位深,单位为位
:param sample_rate: 音频采样率,单位为赫兹
:param bitrate: 音频码率,单位为比特每秒
:param category: 兼容字段,过渡期始终与 library_category 同步
:param library_category: 自动分类或人工覆盖后的媒体库相对目录分类
:param metadata_category: 数据源提供的专辑类型、流派等描述性分类
:param classification: 分类规则的判定结果;未执行分类时为 None
:param classification_facts: 插件来源提交的受控扩展分类事实,统一收口时由宿主注册表校验
:param genres: 来源提供的音乐流派名称列表
:param tags: 来源提供的音乐标签列表
:param artist_country: 艺术家所属国家;未知时为 None
:param names: 可用于搜索匹配的名称列表
:param title_aliases: 来源提供的同一实体别名及展示转换前的原文,不混入所属专辑或无关联艺术家
:param album_aliases: 所属专辑的别名,不混入曲目或艺术家别名
:param artist_aliases: 参与艺术家的别名,不混入无关联艺术家名称
:param detail_link: 来源站点的详情页链接;未知时为 None
:param listen_count: 来源记录的收听次数;未知时为 None
:param raw_data: 来源返回的原始元数据,供来源特有字段读取
:param field_sources: 字段名到元数据来源的映射,用于保留合并后的来源依据
:param composers: 古典角色独立于发行署名;乐器/声部映射保留演奏者的具体职责
:param conductors: 指挥姓名列表,与专辑发行署名独立
:param orchestras: 参与演奏的乐团名称列表
:param performers: 乐器或声部到演奏者姓名列表的映射
:param music_layout: 音乐整理采用的目录布局;尚未确定时为 None
:param organization_error: 音乐整理准入失败的原因;未失败时为 None
:cvar recognize_cache_hit: 本地识别缓存命中标记,不参与序列化,默认 False
"""
type: MediaType = field(default=MediaType.MUSIC, init=False)
media_source: MediaSource | None = None
media_id: str | None = None
music_type: str = MUSIC_ENTITY_RECORDING
title: str | None = None
artists: list[str] = field(default_factory=list)
artist_ids: list[str] = field(default_factory=list)
album: str | None = None
album_artist: str | None = None
album_id: str | None = None
musicbrainz_release_id: str | None = field(default=None, kw_only=True)
musicbrainz_release_group_id: str | None = field(default=None, kw_only=True)
musicbrainz_release_track_id: str | None = field(default=None, kw_only=True)
album_type: str | None = None
secondary_types: list[str] = field(default_factory=list)
year: int | None = None
original_year: int | None = field(default=None, kw_only=True)
release_year: int | None = field(default=None, kw_only=True)
release_date: str | None = None
release_status: str | None = None
disc_number: int | None = None
track_number: int | None = None
total_tracks: int | None = None
total_discs: int | None = field(default=None, kw_only=True)
duration: int | None = None
isrc: str | None = None
cover_url: str | None = None
lyrics: str | None = None
version: str | None = None
audio_format: str | None = None
audio_lossless: bool | None = None
bit_depth: int | None = None
sample_rate: int | None = None
bitrate: int | None = None
category: str = ""
library_category: str = ""
metadata_category: str = ""
classification: ClassificationResult | None = None
classification_facts: dict[str, ClassificationFactValue] = field(default_factory=dict)
genres: list[str] = field(default_factory=list)
tags: list[str] = field(default_factory=list)
artist_country: str | None = None
names: list[str] = field(default_factory=list)
title_aliases: list[str] = field(default_factory=list)
album_aliases: list[str] = field(default_factory=list)
artist_aliases: list[str] = field(default_factory=list)
detail_link: str | None = None
listen_count: int | None = None
raw_data: dict[str, Any] = field(default_factory=dict)
field_sources: dict[str, str] = field(default_factory=dict, kw_only=True)
composers: list[str] = field(default_factory=list, kw_only=True)
conductors: list[str] = field(default_factory=list, kw_only=True)
orchestras: list[str] = field(default_factory=list, kw_only=True)
performers: dict[str, list[str]] = field(default_factory=dict, kw_only=True)
music_layout: str | None = field(default=None, kw_only=True)
organization_error: str | None = field(default=None, kw_only=True)
# 内部标记:是否命中本地识别缓存,不参与序列化;与 MediaInfo 保持一致,
# 显式声明以保留 getattr(..., False) 的默认值语义(__getattr__ 兜底会覆盖它)
recognize_cache_hit = False
def __setstate__(self, state: dict[str, Any]) -> None:
"""
恢复旧版本音乐缓存时补齐新增别名字段,避免列表字段被缺失属性兜底为 None。
"""
self.__dict__.update(state)
self.__dict__.update(music_credit_values(state))
for name in ("title_aliases", "album_aliases", "artist_aliases"):
self.__dict__.setdefault(name, [])
self.__dict__.setdefault("field_sources", {})
def __post_init__(self) -> None:
"""
规范化媒体身份,并兼容拆分旧音乐分类字段。
"""
self.field_sources = dict(self.field_sources or {})
self.__dict__.update(music_credit_values(self))
self.year = self.year if self.year is not None else self.release_year or self.original_year
self.media_source, self.media_id = resolve_media_identity(media=self)
classification = _classification_result(self.classification)
library_category, metadata_category = _resolve_music_categories(
category=self.category,
library_category=self.library_category,
metadata_category=self.metadata_category,
album_type=self.album_type,
secondary_types=self.secondary_types,
genres=self.genres,
tags=self.tags,
classification=classification,
)
self.__dict__["classification"] = classification
self.__dict__["library_category"] = library_category
self.__dict__["category"] = library_category
self.__dict__["metadata_category"] = metadata_category
self.__dict__["_category_semantics_ready"] = True
def __setattr__(self, name: str, value: Any) -> None:
"""
兼容旧调用方直接写 category,并在初始化后同步库分类字段。
"""
if (
name in {"category", "library_category"}
and self.__dict__.get("_category_semantics_ready")
):
normalized = str(value or "").strip()
self.__dict__["category"] = normalized
self.__dict__["library_category"] = normalized
return
self.__dict__[name] = value
def __getattr__(self, name: str) -> None:
"""
影视专用字段兜底返回 None:音乐模型不存在这些字段,避免下游逐点安全访问。
兼容影视媒体共享的通用读写路径(通知、整理、历史等),任何缺失字段按
空值处理;真实字段仍由类定义接管,不受影响。dunder 特殊方法除外——
copy/pickle 等机制依赖 hasattr 探测 __setstate__ 等钩子,兜底返回 None
会导致探测误判而调用失败。注意这会掩盖属性名拼写错误,属于权衡取舍。
"""
if name.startswith("__") and name.endswith("__"):
raise AttributeError(name)
return None
@property
def artist(self) -> str:
"""
返回兼容现有展示组件的艺术家文本。
"""
return " / ".join(self.artists)
@property
def audio_quality(self) -> str | None:
"""
返回 hires、lossless 或 lossy 音质等级。
"""
return audio_quality_tier(
self.audio_format, self.audio_lossless, self.bit_depth, self.sample_rate, self.bitrate
)
@property
def audio_quality_score(self) -> int:
"""
返回音乐订阅洗版使用的音质优先级。
"""
return audio_quality_score(
self.audio_format, self.audio_lossless, self.bit_depth, self.sample_rate, self.bitrate
)
@property
def audio_specs(self) -> str | None:
"""
返回识别结果和通知使用的格式化音频参数。
"""
return format_audio_quality(
self.audio_format, self.audio_lossless, self.bit_depth, self.sample_rate, self.bitrate
)
@property
def episode_group(self) -> None:
"""
音乐没有剧集组,兼容现有下载历史字段。
"""
return None
@property
def season(self) -> None:
"""
音乐没有季信息,兼容失败冷却和目录逻辑。
"""
return None
@property
def vote_average(self) -> float:
"""
音乐当前没有评分字段,兼容订阅统计与持久化。
"""
return 0.0
@property
def overview(self) -> str:
"""
返回兼容订阅描述字段的音乐摘要。
"""
parts = [self.artist, self.album, self.version]
return " · ".join(part for part in parts if part)
@property
def title_year(self) -> str:
"""
返回包含年份的展示标题。
"""
if not self.title:
return ""
return f"{self.title} ({self.year})" if self.year else self.title
@property
def poster_path(self) -> str | None:
"""
返回兼容现有媒体卡片的封面地址。
"""
return self.cover_url
@property
def backdrop_path(self) -> str | None:
"""
返回兼容现有下载卡片的背景地址。
"""
return self.cover_url
def get_message_image(self, default: bool | None = None) -> str | None:
"""
返回通知消息使用的音乐封面。
"""
return self.cover_url
def get_poster_image(self, default: bool | None = None) -> str | None:
"""
返回海报位使用的音乐封面。
"""
return self.cover_url
def get_backdrop_image(self, default: bool = False) -> str | None:
"""
返回背景图位使用的音乐封面。
"""
return self.cover_url
def set_library_category(self, category: str | None) -> None:
"""
设置媒体库分类,并同步过渡期兼容 category 字段。
"""
self.library_category = category or ""
def clear(self) -> None:
"""
清理不参与队列展示和持久化的上游原始响应。
"""
self.raw_data = {}
def to_dict(self) -> dict[str, Any]:
"""
转换为统一媒体身份的 Context 外层字典。
"""
payload = asdict(self)
payload["classification"] = _classification_payload(self.classification)
payload["category"] = self.library_category
payload.update(
{
"type": self.type.value,
"media_source": (
self.media_source.value
if isinstance(self.media_source, MediaSource)
else self.media_source
),
"artist": self.artist,
"title_year": self.title_year,
"poster_path": self.poster_path,
"backdrop_path": self.backdrop_path,
"overview": self.overview,
"vote_average": self.vote_average,
"audio_quality": self.audio_quality,
"audio_quality_score": self.audio_quality_score,
"audio_specs": self.audio_specs,
}
)
return payload
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Self:
"""
从字典恢复标准化音乐元数据。
"""
_validate_music_type(data.get("type"))
values = _music_init_values(cls, data)
values["classification"] = _classification_result(values.get("classification"))
values["media_source"] = normalize_media_source(values.get("media_source"))
values["artists"] = _music_string_list(values.get("artists") or data.get("artist"))
values["artist_ids"] = _music_aligned_list(values.get("artist_ids"))
for key in ("secondary_types", "genres", "tags", "title_aliases", "album_aliases", "artist_aliases"):
values[key] = _music_string_list(values.get(key))
values["names"] = _music_string_list(values.get("names"))
values["music_type"] = str(values.get("music_type") or MUSIC_ENTITY_RECORDING)
values["audio_format"] = normalize_audio_format(values.get("audio_format"))
values["audio_lossless"] = infer_audio_lossless(
values.get("audio_format"), values.get("audio_lossless")
)
values["raw_data"] = dict(values.get("raw_data") or {})
values["field_sources"] = dict(values.get("field_sources") or {})
if "library_category" in data or "metadata_category" in data:
library_value = (
data.get("library_category")
if "library_category" in data
else data.get("category")
)
values["library_category"] = str(library_value or "")
values["metadata_category"] = str(data.get("metadata_category") or "")
values["category"] = values["library_category"]
else:
library_category, metadata_category = _resolve_music_categories(
category=data.get("category"),
library_category=None,
metadata_category=None,
album_type=values.get("album_type"),
secondary_types=values.get("secondary_types"),
genres=values.get("genres"),
tags=values.get("tags"),
classification=values.get("classification"),
)
values["category"] = library_category
values["library_category"] = library_category
values["metadata_category"] = metadata_category
for key in (
"year",
"original_year",
"release_year",
"disc_number",
"track_number",
"total_tracks",
"total_discs",
"duration",
"listen_count",
"bit_depth",
"sample_rate",
"bitrate",
):
values[key] = _music_optional_int(values.get(key))
return cls(**values)
@classmethod
def from_meta(cls, meta: MetaMusic) -> Self:
"""
将文件名和音频标签解析结果转换为无远端依赖的标准音乐信息。
"""
return cls(
**music_credit_values(meta),
media_source=normalize_media_source(meta.media_source),
media_id=meta.media_id,
music_type=meta.music_type or MUSIC_ENTITY_RECORDING,
album_type=meta.album_type,
secondary_types=list(meta.secondary_types),
music_layout=meta.music_layout,
organization_error=meta.organization_error,
musicbrainz_release_id=meta.musicbrainz_release_id,
musicbrainz_release_group_id=meta.musicbrainz_release_group_id,
musicbrainz_release_track_id=meta.musicbrainz_release_track_id,
original_year=meta.original_year,
release_year=meta.release_year,
field_sources=dict(meta.field_sources),
title=meta.title,
artists=list(meta.artists),
album=meta.album,
album_artist=meta.album_artist,
year=_music_optional_int(meta.year),
disc_number=meta.disc_number,
track_number=meta.track_number,
total_tracks=meta.total_tracks,
total_discs=meta.total_discs,
duration=meta.duration,
isrc=meta.isrc,
version=meta.version,
audio_format=meta.audio_format,
audio_lossless=meta.audio_lossless,
bit_depth=meta.bit_depth,
sample_rate=meta.sample_rate,
bitrate=meta.bitrate,
names=[name for name in (meta.title, meta.album) if name],
)
@dataclass
class MusicRelease:
"""
音乐专辑下的单个发行版本(MusicBrainz Release)。
:param media_id: MusicBrainz 具体发行版本标识
:param title: 当前发行版本的标题
:param date: 发行日期文本;未知时为 None
:param country: 来源提供的国家代码;未知时为 None
:param status: 发行状态,例如正式发行或宣传版本
:param packaging: 实体发行的包装形式
:param formats: 各介质的格式名称列表
:param track_count: 当前发行版本的总曲目数;未知时为 None
:param cover_url: 封面图片链接;未提供封面时为 None
"""
media_id: str | None = None
title: str | None = None
date: str | None = None
country: str | None = None
status: str | None = None
packaging: str | None = None
formats: list[str] = field(default_factory=list)
track_count: int | None = None
cover_url: str | None = None
@property
def year(self) -> int | None:
"""
返回发行版本年份。
"""
return _music_year_of(self.date)
def to_dict(self) -> dict[str, Any]:
"""
转换为可传输的字典。
"""
payload = asdict(self)
payload["year"] = self.year
return payload
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Self:
"""
从字典恢复发行版本信息。
"""
values = _music_init_values(cls, data)
values["formats"] = _music_string_list(values.get("formats"))
values["track_count"] = _music_optional_int(values.get("track_count"))
return cls(**values)
@dataclass
class MusicAlbumInfo:
"""
标准化音乐专辑信息(MusicBrainz Release Group)。
:ivar type: 固定为音乐媒体类型,不接受初始化参数
:ivar music_type: 固定为专辑实体,不接受初始化参数
:param media_source: 媒体主身份的数据来源,与 media_id 配对使用
:param media_id: 当前来源的原生媒体标识,与 media_source 配对使用
:param musicbrainz_release_id: MusicBrainz 具体发行版本标识,不等同于录音或发行组标识
:param musicbrainz_release_group_id: MusicBrainz 发行组标识,用于关联同一专辑的不同版本
:param original_year: 作品最早发行年份;未知时为 None
:param total_discs: 发行版本的总碟数;未知时为 None
:param title: 音乐实体的标题;未知时为 None
:param artists: 参与署名的艺术家名称,保留来源顺序
:param artist_ids: 艺术家标准标识列表,顺序与 artists 对应
:param title_aliases: 当前音乐实体的别名及展示转换前的原名
:param artist_aliases: 参与艺术家的别名,不混入无关联艺术家名称
:param album_type: 专辑主类型:Album、EP、Single、Broadcast、Other
:param secondary_types: 专辑副类型:Live、Compilation、Soundtrack、Remix 等
:param library_category: 自动分类或人工覆盖后的媒体库相对目录分类
:param metadata_category: 数据源提供的专辑类型和副类型描述
:param classification: 分类规则的判定结果;未执行分类时为 None
:param classification_facts: 来源提交的扩展分类事实,由宿主字段注册表校验
:param release_date: 来源提供的发行日期文本;未知时为 None
:param release_status: 发行状态,例如正式发行或宣传版本;未知时为 None
:param artist_country: 艺术家所属国家;未知时为 None
:param cover_url: 封面图片链接;未提供封面时为 None
:param genres: 来源提供的音乐流派名称列表
:param tags: 来源提供的音乐标签列表
:param rating: 来源给出的评分,默认 0 表示尚无评分
:param rating_votes: 参与评分的人数;未知时为 None
:param detail_link: 来源站点的详情页链接;未知时为 None
:param tracks: 专辑的曲目详情,保留来源顺序
:param releases: 同一发行组下的发行版本列表
:param raw_data: 来源返回的原始元数据,供来源特有字段读取
:param field_sources: 字段名到元数据来源的映射,用于保留合并后的来源依据
:param composers: 古典角色独立于发行署名;乐器/声部映射保留演奏者的具体职责
:param conductors: 指挥姓名列表,与专辑发行署名独立
:param orchestras: 参与演奏的乐团名称列表
:param performers: 乐器或声部到演奏者姓名列表的映射
"""
type: MediaType = field(default=MediaType.MUSIC, init=False)
music_type: str = field(default=MUSIC_ENTITY_ALBUM, init=False)
media_source: MediaSource | None = None
media_id: str | None = None
musicbrainz_release_id: str | None = field(default=None, kw_only=True)
musicbrainz_release_group_id: str | None = field(default=None, kw_only=True)
original_year: int | None = field(default=None, kw_only=True)
total_discs: int | None = field(default=None, kw_only=True)
title: str | None = None
artists: list[str] = field(default_factory=list)
artist_ids: list[str] = field(default_factory=list)
title_aliases: list[str] = field(default_factory=list)
artist_aliases: list[str] = field(default_factory=list)
album_type: str | None = None
secondary_types: list[str] = field(default_factory=list)
library_category: str = ""
metadata_category: str = ""
classification: ClassificationResult | None = None
classification_facts: dict[str, ClassificationFactValue] = field(default_factory=dict)
release_date: str | None = None
release_status: str | None = None
artist_country: str | None = None
cover_url: str | None = None
genres: list[str] = field(default_factory=list)
tags: list[str] = field(default_factory=list)
rating: float = 0.0
rating_votes: int | None = None
detail_link: str | None = None
tracks: list[MusicInfo] = field(default_factory=list)
releases: list[MusicRelease] = field(default_factory=list)
raw_data: dict[str, Any] = field(default_factory=dict)
field_sources: dict[str, str] = field(default_factory=dict, kw_only=True)
composers: list[str] = field(default_factory=list, kw_only=True)
conductors: list[str] = field(default_factory=list, kw_only=True)
orchestras: list[str] = field(default_factory=list, kw_only=True)
performers: dict[str, list[str]] = field(default_factory=dict, kw_only=True)
def __setstate__(self, state: dict[str, Any]) -> None:
"""
恢复旧版专辑缓存时补齐别名列表,保留既有媒体身份及曲目数据。
"""
self.__dict__.update(state)
self.__dict__.update(music_credit_values(state))
for name in ("title_aliases", "artist_aliases"):
self.__dict__.setdefault(name, [])
self.__dict__.setdefault("field_sources", {})
def __post_init__(self) -> None:
"""
规范化媒体身份,并补全专辑描述分类和分类结果。
"""
self.field_sources = dict(self.field_sources or {})
self.__dict__.update(music_credit_values(self))
self.media_source, self.media_id = resolve_media_identity(media=self)
self.library_category = str(self.library_category or "").strip()
self.metadata_category = str(self.metadata_category or "").strip() or (
" / ".join(
part for part in [self.album_type, *self.secondary_types] if part
)
)
self.classification = _classification_result(self.classification)
@property
def artist(self) -> str:
"""
返回兼容现有展示组件的艺术家文本。
"""
return " / ".join(self.artists)
@property
def year(self) -> int | None:
"""
返回专辑首次发行年份。
"""
return _music_year_of(self.release_date)
@property
def category(self) -> str:
"""
返回兼容字段,其语义始终等同媒体库分类。
"""
return self.library_category
@category.setter
def category(self, value: str | None) -> None:
"""
兼容旧调用方直接写 category,并同步媒体库分类。
"""
self.set_library_category(value)
def set_library_category(self, category: str | None) -> None:
"""
设置媒体库分类,来源描述分类保持不变。
"""
self.library_category = str(category or "").strip()
@property
def track_count(self) -> int:
"""
返回专辑内已解析的音乐数量。
"""
return len(self.tracks)
@property
def duration(self) -> int | None:
"""
返回专辑内所有音乐时长之和。
"""
durations = [track.duration for track in self.tracks if track.duration]
return sum(durations) if durations else None
@property
def title_year(self) -> str:
"""
返回包含年份的专辑展示标题。
"""
if not self.title:
return ""
return f"{self.title} ({self.year})" if self.year else self.title
@property
def poster_path(self) -> str | None:
"""
返回兼容现有媒体卡片的封面地址。
"""
return self.cover_url
@property
def backdrop_path(self) -> str | None:
"""
返回兼容现有详情页背景的封面地址。
"""
return self.cover_url
@property
def overview(self) -> str:
"""
返回专辑摘要,供卡片和通知复用。
"""
parts = [
self.artist,
self.metadata_category,
self.release_date,
" / ".join(self.genres[:3]),
]
return " · ".join(part for part in parts if part)
def to_dict(self) -> dict[str, Any]:
"""
转换为兼容前端 MediaInfo 结构的字典。
"""
payload = {
key: value
for key, value in asdict(self).items()
if key not in {"tracks", "releases", "type"}
}
payload["classification"] = _classification_payload(self.classification)
payload.update(
{
"type": self.type.value,
"artist": self.artist,
"album": self.title,
"year": self.year,
"category": self.category,
"duration": self.duration,
"total_tracks": self.track_count,
"title_year": self.title_year,
"poster_path": self.poster_path,
"backdrop_path": self.backdrop_path,
"media_source": (
self.media_source.value
if isinstance(self.media_source, MediaSource)
else self.media_source
),
"overview": self.overview,
"vote_average": self.rating,
"tracks": [track.to_dict() for track in self.tracks],
"releases": [release.to_dict() for release in self.releases],
}
)
return payload
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Self:
"""
从字典恢复标准化专辑信息。
"""
_validate_music_type(data.get("type"))
values = _music_init_values(cls, data)
values["classification"] = _classification_result(values.get("classification"))
values["media_source"] = normalize_media_source(values.get("media_source"))
for key in ("artists", "secondary_types", "genres", "tags", "title_aliases", "artist_aliases"):
values[key] = _music_string_list(values.get(key))
values["artist_ids"] = _music_aligned_list(values.get("artist_ids"))
values["rating"] = _music_optional_float(values.get("rating"))
values["rating_votes"] = _music_optional_int(values.get("rating_votes"))
values["tracks"] = [
item if isinstance(item, MusicInfo) else MusicInfo.from_dict(item)
for item in data.get("tracks") or []
]
values["releases"] = [
item if isinstance(item, MusicRelease) else MusicRelease.from_dict(item)
for item in data.get("releases") or []
]
values["raw_data"] = dict(values.get("raw_data") or {})
values["field_sources"] = dict(values.get("field_sources") or {})
values["original_year"] = _music_optional_int(values.get("original_year"))
values["total_discs"] = _music_optional_int(values.get("total_discs"))
if "library_category" in data or "metadata_category" in data:
library_value = (
data.get("library_category")
if "library_category" in data
else data.get("category")
)
values["library_category"] = str(library_value or "")
values["metadata_category"] = str(data.get("metadata_category") or "")
else:
library_category, metadata_category = _resolve_music_categories(
category=data.get("category"),
library_category=None,
metadata_category=None,
album_type=values.get("album_type"),
secondary_types=values.get("secondary_types"),
genres=values.get("genres"),
tags=values.get("tags"),
classification=values.get("classification"),
)
values["library_category"] = library_category
values["metadata_category"] = metadata_category
return cls(**values)
def to_music_info(self) -> MusicInfo:
"""
转换为专辑卡片使用的音乐信息,供列表接口统一返回。
"""
return MusicInfo(
**music_credit_values(self),
media_source=self.media_source,
media_id=self.media_id,
musicbrainz_release_id=self.musicbrainz_release_id,
musicbrainz_release_group_id=self.musicbrainz_release_group_id,
original_year=self.original_year,
release_year=self.year if self.musicbrainz_release_id else None,
total_discs=self.total_discs,
field_sources=dict(self.field_sources),
music_type=MUSIC_ENTITY_ALBUM,
title=self.title,
artists=list(self.artists),
artist_ids=list(self.artist_ids),
album=self.title,
album_artist=self.artist or None,
album_id=self.media_id,
title_aliases=list(self.title_aliases),
album_aliases=list(self.title_aliases),
artist_aliases=list(self.artist_aliases),
album_type=self.album_type,
secondary_types=list(self.secondary_types),
year=self.year,
release_date=self.release_date,
release_status=self.release_status,
total_tracks=self.track_count or None,
duration=self.duration,
cover_url=self.cover_url,
library_category=self.library_category,
metadata_category=self.metadata_category,
classification=self.classification,
genres=list(self.genres),
tags=list(self.tags),
artist_country=self.artist_country,
names=[name for name in (self.title,) if name],
detail_link=self.detail_link,
)
@dataclass
class MusicArtistInfo:
"""
标准化音乐艺术家信息(MusicBrainz Artist)。
:ivar type: 固定为音乐媒体类型,不接受初始化参数
:ivar music_type: 固定为艺术家实体,不接受初始化参数
:param media_source: 媒体主身份的数据来源,与 media_id 配对使用
:param media_id: 当前来源的原生媒体标识,与 media_source 配对使用
:param name: 艺术家的展示名称;未知时为 None
:param sort_name: 来源提供的艺术家排序名称;未知时为 None
:param disambiguation: MusicBrainz 消歧义说明,同名艺术家依靠该字段区分
:param artist_type: 艺术家类型:Person、Group、Orchestra、Choir、Character、Other
:param gender: 艺术家的性别描述;未知时为 None
:param country: 来源提供的国家代码;未知时为 None
:param area: 艺术家所属地区名称;未知时为 None
:param begin_date: 艺术家的出生或成立日期文本
:param end_date: 艺术家的逝世或解散日期文本
:param ended: 来源是否明确标记艺术家活动已结束
:param genres: 来源提供的音乐流派名称列表
:param tags: 来源提供的音乐标签列表
:param aliases: 艺术家的别名列表
:param library_category: 自动分类或人工覆盖后的媒体库相对目录分类
:param metadata_category: 数据源提供的艺术家类型等描述性分类
:param classification: 分类规则的判定结果;未执行分类时为 None
:param classification_facts: 来源提交的扩展分类事实,由宿主字段注册表校验
:param relation: 关联艺术家场景下的关系文本,例如乐队成员、子团体
:param image_url: 艺术家的展示图片链接
:param detail_link: 来源站点的详情页链接;未知时为 None
:param external_links: 外部站点链接,键为关系类型,值为地址
:param album_count: 来源记录的专辑数量;未知时为 None
:param raw_data: 来源返回的原始元数据,供来源特有字段读取
"""
type: MediaType = field(default=MediaType.MUSIC, init=False)
music_type: str = field(default=MUSIC_ENTITY_ARTIST, init=False)
media_source: MediaSource | None = None
media_id: str | None = None
name: str | None = None
sort_name: str | None = None
disambiguation: str | None = None
artist_type: str | None = None
gender: str | None = None
country: str | None = None
area: str | None = None
begin_date: str | None = None
end_date: str | None = None
ended: bool = False
genres: list[str] = field(default_factory=list)
tags: list[str] = field(default_factory=list)
aliases: list[str] = field(default_factory=list)
library_category: str = ""
metadata_category: str = ""
classification: ClassificationResult | None = None
classification_facts: dict[str, ClassificationFactValue] = field(default_factory=dict)
relation: str | None = None
image_url: str | None = None
detail_link: str | None = None
external_links: dict[str, str] = field(default_factory=dict)
album_count: int | None = None
raw_data: dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
"""
规范化媒体身份,并拆分艺术家描述分类与媒体库分类。
"""
self.media_source, self.media_id = resolve_media_identity(media=self)
self.library_category = str(self.library_category or "").strip()
self.metadata_category = str(
self.metadata_category or self.artist_type or ""
).strip()
self.classification = _classification_result(self.classification)
@property
def category(self) -> str:
"""
返回兼容字段,其语义始终等同媒体库分类。
"""
return self.library_category
@category.setter
def category(self, value: str | None) -> None:
"""
兼容旧调用方直接写 category,并同步媒体库分类。
"""
self.set_library_category(value)
def set_library_category(self, category: str | None) -> None:
"""
设置媒体库分类,来源艺术家类型保持不变。
"""
self.library_category = str(category or "").strip()
@property
def title(self) -> str | None:
"""
返回兼容通用媒体展示组件的标题。
"""
return self.name
@property
def life_span(self) -> str:
"""
返回艺术家活跃时间区间文本。
"""
if not self.begin_date and not self.end_date:
return ""
end = self.end_date or ("" if self.ended else "…")
return f"{self.begin_date or '?'} - {end}" if end else (self.begin_date or "")
@property
def overview(self) -> str:
"""
返回艺术家摘要,供卡片和详情页复用。
"""
parts = [
self.artist_type,
self.disambiguation,
self.area or self.country,
self.life_span,
" / ".join(self.genres[:3]),
]
return " · ".join(part for part in parts if part)
@property
def poster_path(self) -> str | None:
"""
返回兼容现有卡片的艺术家图片地址。
"""
return self.image_url
def to_dict(self) -> dict[str, Any]:
"""
转换为可传输的字典。
"""
payload = {key: value for key, value in asdict(self).items() if key != "type"}
payload["classification"] = _classification_payload(self.classification)
payload.update(
{
"type": self.type.value,
"category": self.category,
"title": self.title,
"life_span": self.life_span,
"overview": self.overview,
"poster_path": self.poster_path,
"media_source": (
self.media_source.value
if isinstance(self.media_source, MediaSource)
else self.media_source
),
}
)
return payload
def to_music_info(self) -> MusicInfo:
"""
转换为统一搜索列表使用的音乐信息,但不赋予下载或订阅语义。
"""
return MusicInfo(
media_source=self.media_source,
media_id=self.media_id,
music_type=MUSIC_ENTITY_ARTIST,
title=self.name,
cover_url=self.image_url,
version=self.disambiguation,
library_category=self.library_category,
metadata_category=self.metadata_category,
classification=self.classification,
classification_facts=dict(self.classification_facts),
genres=list(self.genres),
tags=list(self.tags),
artist_country=self.country,
names=[name for name in [self.name, *self.aliases] if name],
title_aliases=list(self.aliases),
detail_link=self.detail_link,
raw_data=dict(self.raw_data),
)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Self:
"""
从字典恢复标准化艺术家信息。
"""
_validate_music_type(data.get("type"))
values = _music_init_values(cls, data)
values["classification"] = _classification_result(values.get("classification"))
values["media_source"] = normalize_media_source(values.get("media_source"))
for key in ("genres", "tags", "aliases"):
values[key] = _music_string_list(values.get(key))
values["ended"] = bool(values.get("ended"))
values["album_count"] = _music_optional_int(values.get("album_count"))
values["external_links"] = {
str(key): str(value)
for key, value in (values.get("external_links") or {}).items()
if value
}
values["raw_data"] = dict(values.get("raw_data") or {})
values["library_category"] = str(
data.get("library_category") or data.get("category") or ""
)
values["metadata_category"] = str(
data.get("metadata_category") or ""
)
return cls(**values)
@dataclass
class TorrentInfo:
"""
种子搜索结果信息。
:param site: 站点配置标识
:param site_name: 站点名称
:param site_cookie: 访问站点所需的 Cookie
:param site_ua: 访问站点使用的 User-Agent
:param site_proxy: 站点是否使用代理
:param site_order: 站点优先级
:param site_downloader: 站点指定的下载器名称
:param title: 种子名称
:param description: 种子副标题
:param media_source: 种子页面声明的媒体身份来源
:param media_id: 种子页面声明的来源原生媒体标识
:param enclosure: 种子链接
:param page_url: 详情页面
:param size: 种子内容大小,单位为字节
:param seeders: 当前做种人数
:param peers: 当前下载人数
:param grabs: 累计完成下载次数
:param pubdate: 发布时间
:param date_elapsed: 来源提供的距发布时间的可读文本
:param freedate: 免费截止时间
:param uploadvolumefactor: 上传流量计费倍率;未知时为 None
:param downloadvolumefactor: 下载流量计费倍率;0 表示免费,未知时为 None
:param hit_and_run: 是否受到站点下载后保种规则约束
:param labels: 种子标签
:param pri_order: 种子优先级
:param category: 种子分类 电影/电视剧/音乐
:param search_resource_id: 自动订阅分页搜索的内部候选标识,只用于提交结果与候选对应,不进入对外字典
"""
site: Optional[int] = None
site_name: Optional[str] = None
site_cookie: Optional[str] = None
site_ua: Optional[str] = None
site_proxy: bool = False
site_order: int = 0
site_downloader: Optional[str] = None
title: Optional[str] = None
description: Optional[str] = None
media_source: Optional[MediaSource] = None
media_id: Optional[str] = None
enclosure: Optional[str] = None
page_url: Optional[str] = None
size: float = 0.0
seeders: int = 0
peers: int = 0
grabs: int = 0
pubdate: Optional[str] = None
date_elapsed: Optional[str] = None
freedate: Optional[str] = None
uploadvolumefactor: Optional[float] = None
downloadvolumefactor: Optional[float] = None
hit_and_run: bool = False
labels: list[str] = field(default_factory=list)
pri_order: int = 0
category: Optional[str] = None
search_resource_id: Optional[str] = None
def __post_init__(self) -> None:
"""
将种子声明的媒体身份规范化为统一成对字段。
"""
self.media_source, self.media_id = resolve_media_identity(media=self)
def __setattr__(self, name: str, value: Any) -> None:
"""
直接写入种子运行时字段,保留历史动态属性行为。
"""
self.__dict__[name] = value
def __get_properties(self) -> list[str]:
"""
获取属性列表。
"""
property_names = []
for member_name in dir(self.__class__):
member = getattr(self.__class__, member_name)
if isinstance(member, property):
property_names.append(member_name)
return property_names
def from_dict(self, data: dict[str, Any]) -> None:
"""
将字典字段写入当前种子对象,跳过只读属性后规范化媒体身份。
:param data: 待恢复的字段字典,未知字段保留为动态属性
"""
properties = self.__get_properties()
for key, value in data.items():
if key in properties:
continue
setattr(self, key, value)
self.media_source, self.media_id = resolve_media_identity(media=self)
@staticmethod
def get_free_string(
upload_volume_factor: Optional[float],
download_volume_factor: Optional[float],
) -> str:
"""
将上传与下载计费倍率转换为站点促销标签。
:param upload_volume_factor: 上传流量计费倍率
:param download_volume_factor: 下载流量计费倍率,0 表示免费下载
:return: 既有倍率组合对应的标签;空值或未收录组合返回“未知”
"""
if upload_volume_factor is None or download_volume_factor is None:
return "未知"
free_strs = {
"1.00 1.00": "普通",
"1.00 0.00": "免费",
"2.00 1.00": "2X",
"4.00 1.00": "4X",
"2.00 0.00": "2X免费",
"4.00 0.00": "4X免费",
"1.00 0.50": "50%",
"2.00 0.50": "2X 50%",
"1.00 0.70": "70%",
"1.00 0.30": "30%",
"1.00 0.75": "75%",
"1.00 0.25": "25%"
}
return free_strs.get('%.2f %.2f' % (upload_volume_factor, download_volume_factor), "未知")
@property
def volume_factor(self) -> str:
"""
返回促销信息。
"""
return self.get_free_string(self.uploadvolumefactor, self.downloadvolumefactor)
@property
def freedate_diff(self) -> str:
"""
返回免费剩余时间。
"""
if not self.freedate:
return ""
return time_tools.format_remaining(self.freedate)
def pub_minutes(self) -> float:
"""
返回发布时间距离当前时间的分钟数。
"""
if not self.pubdate:
return 0
try:
pub_date = datetime.strptime(self.pubdate, "%Y-%m-%d %H:%M:%S")
now_datetime = datetime.now()
return (now_datetime - pub_date).total_seconds() // 60
except Exception as e:
print(f"种子发布时间获取失败: {e}")
return 0
def to_dict(self) -> dict[str, Any]:
"""
返回字典。
"""
dicts = vars(self).copy()
dicts["media_source"] = (
self.media_source.value
if isinstance(self.media_source, MediaSource)
else self.media_source
)
dicts["media_id"] = str(self.media_id) if self.media_id is not None else None
dicts["volume_factor"] = self.volume_factor
dicts["freedate_diff"] = self.freedate_diff
dicts.pop("search_resource_id", None)
return dicts
@dataclass
class SubtitleInfo:
"""
字幕搜索结果信息。
:param site: 站点配置标识
:param site_name: 站点名称
:param site_cookie: 访问站点所需的 Cookie
:param site_ua: 访问站点使用的 User-Agent
:param site_proxy: 站点是否使用代理
:param site_order: 站点优先级
:param title: 字幕标题
:param description: 字幕描述
:param enclosure: 字幕下载链接
:param page_url: 详情页面
:param language: 语言
:param language_icon: 语言图标
:param size: 字幕文件大小,单位为字节
:param pubdate: 发布时间
:param date_elapsed: 来源提供的距发布时间的可读文本
:param grabs: 点击/下载次数
:param uploader: 上传者
:param report_url: 举报页面
:param torrent_id: 关联种子的站点内标识
:param subtitle_id: 字幕的站点内标识
:param file_name: 下载文件名
"""
site: Optional[int] = None
site_name: Optional[str] = None
site_cookie: Optional[str] = None
site_ua: Optional[str] = None
site_proxy: bool = False
site_order: int = 0
title: Optional[str] = None
description: Optional[str] = None
enclosure: Optional[str] = None
page_url: Optional[str] = None
language: Optional[str] = None
language_icon: Optional[str] = None
size: float = 0.0
pubdate: Optional[str] = None
date_elapsed: Optional[str] = None
grabs: int = 0
uploader: Optional[str] = None
report_url: Optional[str] = None
torrent_id: Optional[str] = None
subtitle_id: Optional[str] = None
file_name: Optional[str] = None
def __build_meta_info(self) -> Optional[dict[str, Any]]:
"""
从字幕标题、文件名和描述中识别可展示的季集信息。
"""
for title in (self.title, self.file_name, self.description):
if not title:
continue
try:
meta_dict: dict[str, Any] = MetaInfo(title=title, subtitle=self.description).to_dict()
except Exception:
continue
if meta_dict.get("season_episode") or meta_dict.get("episode_list"):
return meta_dict
return None
def __setattr__(self, name: str, value: Any) -> None:
"""
直接写入字幕运行时字段,保留历史动态属性行为。
"""
self.__dict__[name] = value
def from_dict(self, data: dict[str, Any]) -> None:
"""
将字典字段写入当前种子对象,跳过只读属性后规范化媒体身份。
:param data: 待恢复的字段字典,未知字段保留为动态属性
"""
for key, value in data.items():
setattr(self, key, value)
def to_dict(self) -> dict[str, Any]:
"""
返回字典。
"""
dicts = vars(self).copy()
meta_info = self.__build_meta_info()
if meta_info:
dicts["meta_info"] = meta_info
dicts["season_episode"] = meta_info.get("season_episode")
dicts["episode_list"] = meta_info.get("episode_list")
return dicts
@dataclass
class MediaInfo:
"""
统一媒体信息,负责聚合各元数据源的标准字段。
:param media_source: 媒体主身份来源
:param media_id: 当前数据源原生 ID
:param scrape_source: 请求级刮削来源;为空时使用系统设置
:param type: 类型 电影、电视剧
:param title: 媒体标题
:param en_title: 英文标题
:param hk_title: 香港标题
:param tw_title: 台湾标题
:param sg_title: 新加坡标题
:param year: 年份
:param season: 季
:param tmdb_id: 来源返回的 TMDB 辅助标识,不参与通用身份传递或持久化
:param imdb_id: 来源返回的 IMDb 辅助标识
:param tvdb_id: 来源返回的 TVDB 辅助标识
:param tvdb_slug: TVDB 详情页的可读路径标识
:param douban_id: 来源返回的豆瓣辅助标识
:param bangumi_id: 来源返回的 Bangumi 辅助标识
:param anilist_id: 来源返回的 AniList 辅助标识
:param anidb_id: 来源返回的 AniDB 辅助标识
:param collection_id: 合集ID
:param original_language: 媒体原语种
:param original_title: 媒体原发行标题
:param release_date: 媒体发行日期
:param backdrop_path: 背景图片
:param poster_path: 海报图片
:param logo_path: 媒体标志图片地址
:param vote_average: 评分
:param overview: 描述
:param genre_ids: 风格ID
:param names: 所有别名和译名
:param seasons: 各季的剧集清单信息
:param season_info: 各季详情;IMDb 的 episodes 保存单集 id 与 episode_number,用于独立 NFO 身份
:param season_years: 各季的年份
:param category: 兼容字段,过渡期始终与 library_category 同步
:param library_category: 自动分类或人工覆盖后的媒体库相对目录分类
:param metadata_category: 数据源提供的描述性分类
:param classification: 分类规则计算及人工覆盖的结果;未执行分类时为 None
:param classification_facts: 插件来源提交的受控扩展分类事实,统一收口时由宿主注册表校验
:param tmdb_info: TMDB 原始详情,用于来源字段投影
:param douban_info: 豆瓣原始详情,用于来源字段投影
:param bangumi_info: Bangumi 原始详情,用于来源字段投影
:param anilist_info: AniList 原始详情,用于来源字段投影
:param directors: 导演
:param actors: 演员
:param adult: 是否成人内容
:param created_by: 创建人
:param episode_run_time: 来源提供的单集时长列表,单位为分钟
:param genres: 风格
:param first_air_date: 首播日期
:param homepage: 首页
:param languages: 语种
:param last_air_date: 最后上映日期
:param networks: 流媒体平台
:param number_of_episodes: 集数
:param number_of_seasons: 季数
:param origin_country: 原产国
:param original_name: 原名
:param production_companies: 出品公司
:param production_countries: 出品国
:param spoken_languages: 语种
:param release_dates: 所有发行日期
:param status: 状态
:param tagline: 标签
:param vote_count: 评价数量
:param popularity: 流行度
:param runtime: 影片时长,单位为分钟
:param next_episode_to_air: 下一集
:param content_rating: 内容分级
:param episode_groups: 全部剧集组
:param episode_group: 剧集组
:cvar recognize_cache_hit: 本地识别缓存命中标记,不参与序列化,默认 False
"""
# 内部标记:是否命中本地识别缓存,不参与序列化
recognize_cache_hit = False
media_source: Optional[MediaSource] = None
media_id: Optional[str] = None
scrape_source: Optional[str] = None
type: Optional[MediaType] = None
title: Optional[str] = None
en_title: Optional[str] = None
hk_title: Optional[str] = None
tw_title: Optional[str] = None
sg_title: Optional[str] = None
year: Optional[str] = None
season: Optional[int] = None
tmdb_id: Optional[int] = None
imdb_id: Optional[str] = None
tvdb_id: Optional[int] = None
tvdb_slug: Optional[str] = None
douban_id: Optional[str] = None
bangumi_id: Optional[int] = None
anilist_id: Optional[int] = None
anidb_id: Optional[int] = None
collection_id: Optional[int] = None
original_language: Optional[str] = None
original_title: Optional[str] = None
release_date: Optional[str] = None
backdrop_path: Optional[str] = None
poster_path: Optional[str] = None
logo_path: Optional[str] = None
vote_average: Optional[float] = None
overview: Optional[str] = None
genre_ids: list[int] = field(default_factory=list)
names: list[str] = field(default_factory=list)
seasons: dict[int, list[int]] = field(default_factory=dict)
season_info: list[dict[str, Any]] = field(default_factory=list)
season_years: dict[int, str] = field(default_factory=dict)
category: str = ""
library_category: str = ""
metadata_category: str = ""
classification: ClassificationResult | None = None
classification_facts: dict[str, ClassificationFactValue] = field(default_factory=dict)
tmdb_info: dict[str, Any] = field(default_factory=dict)
douban_info: dict[str, Any] = field(default_factory=dict)
bangumi_info: dict[str, Any] = field(default_factory=dict)
anilist_info: dict[str, Any] = field(default_factory=dict)
directors: List[dict[str, Any]] = field(default_factory=list)
actors: List[dict[str, Any]] = field(default_factory=list)
adult: Optional[bool] = None
created_by: list[dict[str, Any]] = field(default_factory=list)
episode_run_time: list[int] = field(default_factory=list)
genres: List[dict[str, Any]] = field(default_factory=list)
first_air_date: Optional[str] = None
homepage: Optional[str] = None
languages: list[str] = field(default_factory=list)
last_air_date: Optional[str] = None
networks: list[dict[str, Any]] = field(default_factory=list)
number_of_episodes: Optional[int] = None
number_of_seasons: Optional[int] = None
origin_country: list[str] = field(default_factory=list)
original_name: Optional[str] = None
production_companies: list[dict[str, Any]] = field(default_factory=list)
production_countries: list[dict[str, Any]] = field(default_factory=list)
spoken_languages: list[dict[str, Any]] = field(default_factory=list)
release_dates: list[dict[str, Any]] = field(default_factory=list)
status: Optional[str] = None
tagline: Optional[str] = None
vote_count: Optional[int] = None
popularity: Optional[float] = None
runtime: Optional[int] = None
next_episode_to_air: dict[str, Any] = field(default_factory=dict)
content_rating: Optional[str] = None
episode_groups: List[dict[str, Any]] = field(default_factory=list)
episode_group: Optional[str] = None
def __post_init__(self) -> None:
"""
规范化媒体来源、来源投影和新旧分类字段。
"""
self.media_source, self.media_id = resolve_media_identity(media=self)
if self.tmdb_info:
self.set_tmdb_info(self.tmdb_info)
if self.douban_info:
self.set_douban_info(self.douban_info)
if self.bangumi_info:
self.set_bangumi_info(self.bangumi_info)
if self.anilist_info:
self.set_anilist_info(self.anilist_info)
self.media_source, self.media_id = resolve_media_identity(media=self)
library_category = str(self.library_category or self.category or "").strip()
self.__dict__["library_category"] = library_category
self.__dict__["category"] = library_category
self.__dict__["metadata_category"] = str(
self.metadata_category or ""
).strip()
self.__dict__["classification"] = _classification_result(
self.classification
)
self.__dict__["_category_semantics_ready"] = True
def __setattr__(self, name: str, value: Any) -> None:
"""
写入动态字段,并兼容旧调用方直接覆盖 category。
"""
if (
name in {"category", "library_category"}
and self.__dict__.get("_category_semantics_ready")
):
normalized = str(value or "").strip()
self.__dict__["category"] = normalized
self.__dict__["library_category"] = normalized
return
self.__dict__[name] = value
def __get_properties(self) -> list[str]:
"""
获取属性列表。
"""
property_names = []
for member_name in dir(self.__class__):
member = getattr(self.__class__, member_name)
if isinstance(member, property):
property_names.append(member_name)
return property_names
def from_dict(self, data: dict[str, Any]) -> None:
"""
将字典字段写入当前种子对象,跳过只读属性后规范化媒体身份。
:param data: 待恢复的字段字典,未知字段保留为动态属性
"""
properties = self.__get_properties()
for key, value in data.items():
if key in properties or key in {
"category",
"library_category",
"metadata_category",
"classification",
"_category_semantics_ready",
}:
continue
setattr(self, key, value)
if "metadata_category" in data:
self.metadata_category = str(data.get("metadata_category") or "").strip()
if "classification" in data:
self.classification = _classification_result(data.get("classification"))
if "library_category" in data:
self.set_library_category(data.get("library_category"))
elif "category" in data:
self.set_library_category(data.get("category"))
self.media_source, self.media_id = resolve_media_identity(media=self)
if isinstance(self.type, str):
self.type = MediaType(self.type)
def set_category(self, cat: str) -> None:
"""
兼容旧入口,委托统一媒体库分类写入方法。
"""
self.set_library_category(cat)
def set_library_category(self, category: str | None) -> None:
"""
设置媒体库分类,并同步过渡期兼容 category 字段。
"""
self.library_category = category or ""
def _apply_source_projection(self, values: dict[str, Any]) -> None:
"""
把来源 owner 返回的字段映射应用到当前领域对象。
"""
for name, value in values.items():
setattr(self, name, value)
def set_tmdb_info(self, info: dict[str, Any]) -> None:
"""
通过 TMDB 来源 owner 初始化统一媒体字段。
"""
self._apply_source_projection(_project_tmdb(vars(self), info))
def set_douban_info(self, info: dict[str, Any]) -> None:
"""
通过豆瓣来源 owner 初始化统一媒体字段。
"""
self._apply_source_projection(_project_douban(vars(self), info))
@staticmethod
def get_bangumi_media_type(info: dict[str, Any]) -> MediaType:
"""
兼容旧方法,委托 Bangumi 来源 owner 解析媒体类型。
"""
from app.domain.projection.bangumi import resolve_media_type
return resolve_media_type(info)
def set_bangumi_info(self, info: dict[str, Any]) -> None:
"""
通过 Bangumi 来源 owner 初始化统一媒体字段。
"""
self._apply_source_projection(_project_bangumi(vars(self), info))
@staticmethod
def get_anilist_media_type(info: dict[str, Any]) -> MediaType:
"""
兼容旧方法,委托 AniList 来源 owner 解析媒体类型。
"""
from app.domain.projection.anilist import resolve_media_type
return resolve_media_type(info)
@staticmethod
def _anilist_date(date_info: dict[str, Any]) -> Optional[str]:
"""
兼容旧私有补丁点,委托 AniList owner 格式化模糊日期。
"""
from app.domain.projection.anilist import format_date
return format_date(date_info)
@staticmethod
def _anilist_chinese_title(info: dict[str, Any]) -> Optional[str]:
"""
兼容旧私有补丁点,委托 AniList owner 选择中文标题。
"""
from app.domain.projection.anilist import select_chinese_title
return select_chinese_title(info)
def set_anilist_info(self, info: dict[str, Any]) -> None:
"""
通过 AniList 来源 owner 初始化统一媒体字段。
"""
self._apply_source_projection(_project_anilist(vars(self), info))
@property
def title_year(self) -> str:
"""
返回带年份的媒体标题。
"""
if self.title:
return "%s (%s)" % (self.title, self.year) if self.year else self.title
return ""
@property
def detail_link(self) -> str:
"""
返回当前媒体主身份对应的来源详情页地址。
:return: 来源详情页地址;来源或标识不受支持时返回空字符串
"""
if self.media_source == MediaSource.TMDB and self.media_id:
if self.type == MediaType.MOVIE:
return f"https://www.themoviedb.org/movie/{self.media_id}"
else:
return f"https://www.themoviedb.org/tv/{self.media_id}"
if self.media_source == MediaSource.Douban and self.media_id:
return f"https://movie.douban.com/subject/{self.media_id}"
if self.media_source == MediaSource.Bangumi and self.media_id:
return f"https://bgm.tv/subject/{self.media_id}"
if self.media_source == MediaSource.AniList and self.media_id:
return f"https://anilist.co/anime/{self.media_id}"
if self.media_source == MediaSource.IMDb and self.media_id:
return f"https://www.imdb.com/title/{self.media_id}"
if self.media_source == MediaSource.TVDB and self.media_id:
return f"https://thetvdb.com/search?query={self.media_id}"
return ""
@property
def stars(self) -> str:
"""
返回评分星星个数。
"""
if not self.vote_average:
return ""
return "".rjust(int(self.vote_average), "★")
@property
def vote_star(self) -> str:
"""
返回适合消息展示的评分星级文本。
"""
if self.vote_average:
return "评分:%s" % self.stars
return ""
@typing.overload
def get_backdrop_image(self) -> str:
"""
按调用方指定的默认图策略获取背景图地址。
"""
...
@typing.overload
def get_backdrop_image(self, default: bool) -> Union[str, bool]:
"""
按调用方指定的默认图策略获取背景图地址。
"""
...
def get_backdrop_image(self, default: bool = False) -> Union[str, bool]:
"""
返回背景图片地址。
:param default: 图片缺失时的回退标记,默认返回空字符串
:return: 缩放后的背景地址;缺失时返回回退标记或空字符串
"""
if self.backdrop_path:
return self.backdrop_path.replace("original", "w500")
return default or ""
@typing.overload
def get_message_image(self) -> str:
"""
按调用方指定的默认图策略获取通知消息图片。
"""
...
@typing.overload
def get_message_image(self, default: Optional[bool]) -> Union[str, bool]:
"""
按调用方指定的默认图策略获取通知消息图片。
"""
...
def get_message_image(self, default: Optional[bool] = None) -> Union[str, bool]:
"""
返回消息图片地址,优先使用背景图,再回退到海报。
:param default: 两种图片均缺失时的回退标记,默认返回空字符串
:return: 缩放后的图片地址;缺失时返回回退标记或空字符串
"""
if self.backdrop_path:
return self.backdrop_path.replace("original", "w500")
return self.get_poster_image(default=default)
@typing.overload
def get_poster_image(self) -> str:
"""
按调用方指定的默认图策略获取海报图片。
"""
...
@typing.overload
def get_poster_image(self, default: Optional[bool]) -> Union[str, bool]:
"""
按调用方指定的默认图策略获取海报图片。
"""
...
def get_poster_image(self, default: Optional[bool] = None) -> Union[str, bool]:
"""
返回海报图片地址。
:param default: 图片缺失时的回退标记,默认返回空字符串
:return: 缩放后的海报地址;缺失时返回回退标记或空字符串
"""
if self.poster_path:
return self.poster_path.replace("original", "w500")
return default or ""
def get_overview_string(self, max_len: int = 140) -> str:
"""
按指定长度截取媒体简介。
:param max_len: 简介显示长度预算,默认 140,截断时为省略标记预留位置
:return: 去除首尾空白后的简介,超长时截取并附加省略标记;None 按字符串转换
"""
overview = str(self.overview).strip()
placeholder = ' ...'
max_len = max(len(placeholder), max_len - len(placeholder))
overview = (overview[:max_len] + placeholder) if len(overview) > max_len else overview
return overview
def to_dict(self) -> dict[str, Any]:
"""
返回字典。
"""
dicts = vars(self).copy()
dicts.pop("_category_semantics_ready", None)
dicts["category"] = self.library_category
dicts["library_category"] = self.library_category
dicts["metadata_category"] = self.metadata_category
dicts["classification"] = _classification_payload(self.classification)
dicts["type"] = self.type.value if self.type else None
dicts["detail_link"] = self.detail_link
dicts["title_year"] = self.title_year
dicts["tmdb_info"] = None
dicts["douban_info"] = None
dicts["bangumi_info"] = None
dicts["anilist_info"] = None
dicts["media_source"] = (
self.media_source.value
if isinstance(self.media_source, MediaSource)
else self.media_source
)
dicts["media_id"] = str(self.media_id) if self.media_id is not None else None
return dicts
def clear(self) -> None:
"""
去除多余数据,减小体积。
"""
self.tmdb_info = {}
self.douban_info = {}
self.bangumi_info = {}
self.anilist_info = {}
self.seasons = {}
self.genres = []
self.season_info = []
self.names = []
self.actors = []
self.directors = []
self.production_companies = []
self.production_countries = []
self.spoken_languages = []
self.networks = []
self.next_episode_to_air = {}
self.episode_groups = []
@dataclass
class Context:
"""
聚合候选资源的识别信息、匹配依据和下载选择结果。
:param meta_info: 识别信息
:param media_info: 媒体信息
:param torrent_info: 种子信息
:param media_recognize_fail_count: 媒体识别失败次数
:param resource_source: 候选资源来源:rss、spider、search、unknown
:param match_source: 候选匹配来源:MediaSource 枚举值、title、unknown
:param candidate_recognized: 候选自身是否已经识别出有效媒体 ID
:param media_info_is_target: 当前 media_info 是否为目标媒体回填,而不是候选自身识别结果
:param match_status: 音乐匹配等级;待人工确认的候选不得绑定目标媒体身份
:param match_reason: 音乐匹配结论的依据;尚未匹配时为 None
:param allowed_episodes: 调用方对本候选允许下载的剧集集合,None 表示不限制,空集合表示拒绝交付任何集
:param selected_episodes: 下载链实际提交的剧集集合;None 表示尚未执行下载选择
:param confirmed_full_coverage: 下载层确认候选资源覆盖完整目标范围,供订阅事实写入判断整包资源
:param subscription_download: 本次订阅提交的类型、全集洗版模式及新增和升级集数;非订阅下载为 None
:param music_track_keys: 下载层从种子文件清单提取的稳定音轨键,供专辑订阅跨轮次累计下载进度
"""
meta_info: Optional[MetaBase] = None
media_info: Optional[Union[MediaInfo, MusicInfo]] = None
torrent_info: Optional[TorrentInfo] = None
media_recognize_fail_count: int = 0
resource_source: str = "unknown"
match_source: str = "unknown"
candidate_recognized: bool = False
media_info_is_target: bool = False
match_status: Optional[str] = None
match_reason: Optional[str] = None
allowed_episodes: Optional[Set[int]] = None
selected_episodes: Optional[List[int]] = None
confirmed_full_coverage: bool = False
music_track_keys: Optional[List[str]] = None
subscription_download: Optional[Dict[str, JsonData]] = None
def to_dict(self) -> dict[str, Any]:
"""
转换为字典。
"""
return {
"meta_info": self.meta_info.to_dict() if self.meta_info else None,
"torrent_info": self.torrent_info.to_dict() if self.torrent_info else None,
"media_info": self.media_info.to_dict() if self.media_info else None,
"media_recognize_fail_count": self.media_recognize_fail_count,
"resource_source": self.resource_source,
"match_source": self.match_source,
"candidate_recognized": self.candidate_recognized,
"media_info_is_target": self.media_info_is_target,
"match_status": getattr(self, "match_status", None),
"match_reason": getattr(self, "match_reason", None),
# 保留 None / 空集 / 非空集 三态语义,避免下游误把"显式拒绝"当成"不限制"。
"allowed_episodes": sorted(self.allowed_episodes) if self.allowed_episodes is not None else None,
"selected_episodes": self.selected_episodes,
"confirmed_full_coverage": self.confirmed_full_coverage,
"music_track_keys": self.music_track_keys,
"subscription_download": self.subscription_download,
}