mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 12:24:39 +08:00
* fix(media): preserve recognition and rule mutation identities * test: cover typed classification facts with alias policies --------- Co-authored-by: jxxghp <jxxghp@gmail.com>
2461 lines
94 KiB
Python
2461 lines
94 KiB
Python
"""
|
||
聚合媒体、音乐、种子和字幕的领域上下文、序列化契约及来源元数据投影入口。
|
||
"""
|
||
|
||
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,
|
||
}
|