mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
"""
|
|
影视识别模块共享的附加信息能力实现。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Optional
|
|
|
|
from app.domain.context import MediaInfo
|
|
from app.domain.media import is_media_source_enabled
|
|
from app.domain.meta.metabase import MetaBase
|
|
from app.domain.metainfo import MetaInfo
|
|
from app.schemas.types import MediaSource, MediaSourceSelection, MediaType
|
|
|
|
|
|
class MediaAuxiliaryProviderMixin:
|
|
"""
|
|
把媒体识别接口适配为可聚合的影视附加信息能力。
|
|
|
|
:ivar auxiliary_media_source: 当前模块提供附加信息时使用的数据源
|
|
"""
|
|
|
|
auxiliary_media_source: MediaSource
|
|
|
|
@staticmethod
|
|
def _build_auxiliary_meta(
|
|
mediainfo: MediaInfo,
|
|
metainfo: Optional[MetaBase] = None,
|
|
) -> MetaBase:
|
|
"""
|
|
根据主识别结果构造不携带原来源身份的标题查询参数。
|
|
"""
|
|
# MetaInfo 是工厂函数，实际返回 MetaBase 及其子类实例
|
|
meta = MetaInfo(mediainfo.title or getattr(metainfo, "name", None) or "")
|
|
if not meta.en_name:
|
|
meta.en_name = mediainfo.en_title or getattr(metainfo, "en_name", None)
|
|
meta.type = mediainfo.type or getattr(metainfo, "type", None) or MediaType.UNKNOWN
|
|
meta.begin_season = (
|
|
mediainfo.season
|
|
if mediainfo.season is not None
|
|
else getattr(metainfo, "begin_season", None)
|
|
)
|
|
season_year = None
|
|
if meta.begin_season is not None and mediainfo.season_years:
|
|
season_year = (
|
|
mediainfo.season_years.get(meta.begin_season)
|
|
or mediainfo.season_years.get(int(meta.begin_season))
|
|
)
|
|
meta.year = season_year or mediainfo.year or getattr(metainfo, "year", None)
|
|
return meta
|
|
|
|
def _auxiliary_recognize_kwargs(
|
|
self,
|
|
mediainfo: MediaInfo,
|
|
metainfo: Optional[MetaBase] = None,
|
|
) -> dict[str, object]:
|
|
"""
|
|
构造当前来源的识别参数，同源媒体优先使用来源原生 ID。
|
|
"""
|
|
same_source = mediainfo.media_source == self.auxiliary_media_source
|
|
return {
|
|
"meta": self._build_auxiliary_meta(mediainfo, metainfo),
|
|
"mtype": mediainfo.type,
|
|
"media_source": self.auxiliary_media_source,
|
|
"media_id": mediainfo.media_id if same_source else None,
|
|
"episode_group": mediainfo.episode_group,
|
|
"cache": True,
|
|
}
|
|
|
|
def get_media_auxiliary_info(
|
|
self,
|
|
mediainfo: MediaInfo,
|
|
media_source: Optional[MediaSourceSelection] = None,
|
|
metainfo: Optional[MetaBase] = None,
|
|
) -> list[MediaInfo]:
|
|
"""
|
|
使用当前来源同步获取影视别名等附加信息。
|
|
|
|
:param mediainfo: 已识别的主媒体信息，仅处理电影和电视剧
|
|
:param media_source: 允许使用的数据源集合；不包含当前来源时不发起识别
|
|
:param metainfo: 可补充标题、年份和季号的名称解析结果
|
|
:return: 当前来源的单条补充结果；不适用或识别失败时返回空列表
|
|
"""
|
|
if (
|
|
not mediainfo
|
|
or mediainfo.type not in (MediaType.MOVIE, MediaType.TV)
|
|
or not is_media_source_enabled(media_source, self.auxiliary_media_source)
|
|
):
|
|
return []
|
|
recognize: Callable[..., Optional[MediaInfo]] = getattr(self, "recognize_media")
|
|
result = recognize(**self._auxiliary_recognize_kwargs(mediainfo, metainfo))
|
|
return [result] if result else []
|
|
|
|
async def async_get_media_auxiliary_info(
|
|
self,
|
|
mediainfo: MediaInfo,
|
|
media_source: Optional[MediaSourceSelection] = None,
|
|
metainfo: Optional[MetaBase] = None,
|
|
) -> list[MediaInfo]:
|
|
"""
|
|
使用当前来源异步获取影视别名等附加信息。
|
|
|
|
:param mediainfo: 已识别的主媒体信息，仅处理电影和电视剧
|
|
:param media_source: 允许使用的数据源集合；不包含当前来源时不发起识别
|
|
:param metainfo: 可补充标题、年份和季号的名称解析结果
|
|
:return: 等待识别后得到的单条补充结果；不适用或识别失败时返回空列表
|
|
"""
|
|
if (
|
|
not mediainfo
|
|
or mediainfo.type not in (MediaType.MOVIE, MediaType.TV)
|
|
or not is_media_source_enabled(media_source, self.auxiliary_media_source)
|
|
):
|
|
return []
|
|
recognize: Callable[..., Awaitable[Optional[MediaInfo]]] = getattr(
|
|
self,
|
|
"async_recognize_media",
|
|
)
|
|
result = await recognize(**self._auxiliary_recognize_kwargs(mediainfo, metainfo))
|
|
return [result] if result else []
|