diff --git a/app/chain/media/plugin.py b/app/chain/media/plugin.py index 63b67e2fc..607fed0a0 100644 --- a/app/chain/media/plugin.py +++ b/app/chain/media/plugin.py @@ -66,6 +66,16 @@ class _RecognitionSelection: return helped if self.is_recognized(helped) else native +@dataclass(frozen=True, slots=True) +class _VideoHelpElements: + """保存影视辅助识别事件解析出的规范化标题要素。""" + + title: str + year: Optional[str] + season: Optional[int] + episode: Optional[int] + + def _normalize_music_year(value: object) -> Optional[int]: """把本地或插件音乐年份统一为正整数，空值和非法文本返回空。""" text = str(value or "").strip() @@ -281,6 +291,55 @@ class MediaPluginOwner(_MediaPluginOwnerBase): text = str(value).strip() return int(text) if text.isdigit() else None + @staticmethod + def _video_help_elements( + event_data: Mapping[str, object], + ) -> Optional[_VideoHelpElements]: + """规范化影视辅助识别结果，标题为空或未知时拒绝该结果。""" + recognized_title: Optional[str] = None + recognized_year: Optional[str] = None + if event_data.get("name"): + recognized_title = ( + str(event_data["name"]).split("/")[0].strip().replace(".", " ") + ) + if not recognized_title or recognized_title == "Unknown": + return None + if event_data.get("year"): + recognized_year = str(event_data["year"]).split("/")[0].strip() + if not str(recognized_year).isdigit(): + recognized_year = None + return _VideoHelpElements( + title=recognized_title, + year=recognized_year, + season=MediaPluginOwner._parse_recognize_event_number( + event_data.get("season") + ), + episode=MediaPluginOwner._parse_recognize_event_number( + event_data.get("episode") + ), + ) + + @staticmethod + def _prepare_video_help_meta( + org_meta: MetaBase, + event_data: Mapping[str, object], + ) -> Optional[MetaBase]: + """统一校验影视辅助要素并把有效变化应用到本次元数据。""" + elements = MediaPluginOwner._video_help_elements(event_data) + if elements is None: + return None + if elements.title == org_meta.name and elements.year == org_meta.year: + logger.info("辅助识别与原始识别结果一致，无需重新识别媒体信息") + return None + logger.info("辅助识别结果与原始识别结果不一致，重新匹配媒体信息 ...") + org_meta.name = elements.title + org_meta.year = elements.year + org_meta.begin_season = elements.season + org_meta.begin_episode = elements.episode + if org_meta.begin_season is not None or org_meta.begin_episode is not None: + org_meta.type = MediaType.TV + return org_meta + def recognize_help( self, title: str, @@ -318,38 +377,13 @@ class MediaPluginOwner(_MediaPluginOwnerBase): ) if not result: return None - # 获取返回事件数据 event_data = result.event_data or {} logger.info(f"获取到辅助识别结果：{event_data}") - # 处理数据格式 - recognized_title: Optional[str] = None - recognized_year: Optional[str] = None - if event_data.get("name"): - recognized_title = str(event_data["name"]).split("/")[0].strip().replace(".", " ") - if event_data.get("year"): - recognized_year = str(event_data["year"]).split("/")[0].strip() - season_number = self._parse_recognize_event_number(event_data.get("season")) - episode_number = self._parse_recognize_event_number(event_data.get("episode")) - if not recognized_title: + rematch_meta = MediaPluginOwner._prepare_video_help_meta(org_meta, event_data) + if rematch_meta is None: return None - if recognized_title == "Unknown": - return None - if not str(recognized_year).isdigit(): - recognized_year = None - # 结果赋值 - if recognized_title == org_meta.name and recognized_year == org_meta.year: - logger.info("辅助识别与原始识别结果一致，无需重新识别媒体信息") - return None - logger.info("辅助识别结果与原始识别结果不一致，重新匹配媒体信息 ...") - org_meta.name = recognized_title - org_meta.year = recognized_year - org_meta.begin_season = season_number - org_meta.begin_episode = episode_number - if org_meta.begin_season is not None or org_meta.begin_episode is not None: - org_meta.type = MediaType.TV - # 重新识别 return self.recognize_media( - meta=org_meta, + meta=rematch_meta, media_source=media_source, share_meta=share_meta, episode_group=episode_group, @@ -372,25 +406,52 @@ class MediaPluginOwner(_MediaPluginOwnerBase): :param media_source: 请求级识别数据源 :param music_type: 音乐实体类型 """ - # 发送音乐名称识别事件，等待插件返回标题要素 result = self.eventmanager.send_event( ChainEventType.MusicNameRecognize, - { - "title": title, - "artist": org_meta.artist, - "album": org_meta.album, - "year": org_meta.year, - "music_type": music_type, - }, + MediaPluginOwner._music_help_payload(title, org_meta, music_type), ) if not result: return None event_data = result.event_data or {} logger.info(f"获取到音乐辅助识别结果：{event_data}") - name, artist, album, year = self._parse_music_recognize_event(event_data) + new_meta = MediaPluginOwner._prepare_music_help_meta(org_meta, event_data) + if new_meta is None: + return None + # 重新识别，仅采信取得远端身份的结果，否则由选择流程保留原生兜底 + mediainfo = self.recognize_media( + meta=new_meta, + media_source=media_source, + share_meta=share_meta, + music_type=music_type, + ) + return mediainfo if mediainfo and mediainfo.media_source else None + + @staticmethod + def _music_help_payload( + title: str, + org_meta: MetaMusic, + music_type: Optional[str], + ) -> Mapping[str, object]: + """为同步与异步音乐辅助识别生成完全相同的事件载荷。""" + return { + "title": title, + "artist": org_meta.artist, + "album": org_meta.album, + "year": org_meta.year, + "music_type": music_type, + } + + @staticmethod + def _prepare_music_help_meta( + org_meta: MetaMusic, + event_data: Mapping[str, object], + ) -> Optional[MetaMusic]: + """统一校验音乐辅助要素，并仅为真实变化构造重新匹配元数据。""" + name, artist, album, year = MediaPluginOwner._parse_music_recognize_event( + event_data + ) if not name: return None - # 辅助识别要素与原始一致时无需重新匹配 if ( name == org_meta.title and (not artist or artist in org_meta.artists) @@ -400,21 +461,13 @@ class MediaPluginOwner(_MediaPluginOwnerBase): logger.info("音乐辅助识别与原始识别结果一致，无需重新匹配媒体信息") return None logger.info("音乐辅助识别结果与原始识别结果不一致，重新匹配媒体信息 ...") - new_meta = self._build_music_help_meta( + return MediaPluginOwner._build_music_help_meta( org_meta=org_meta, name=name, artist=artist, album=album, year=year, ) - # 重新识别，仅采信取得远端身份的结果，否则由选择流程保留原生兜底 - mediainfo = self.recognize_media( - meta=new_meta, - media_source=media_source, - share_meta=share_meta, - music_type=music_type, - ) - return mediainfo if mediainfo and mediainfo.media_source else None @staticmethod def _parse_music_recognize_event( @@ -541,35 +594,11 @@ class MediaPluginOwner(_MediaPluginOwnerBase): # 获取返回事件数据 event_data = result.event_data or {} logger.info(f"获取到辅助识别结果：{event_data}") - # 处理数据格式 - recognized_title: Optional[str] = None - recognized_year: Optional[str] = None - if event_data.get("name"): - recognized_title = str(event_data["name"]).split("/")[0].strip().replace(".", " ") - if event_data.get("year"): - recognized_year = str(event_data["year"]).split("/")[0].strip() - season_number = self._parse_recognize_event_number(event_data.get("season")) - episode_number = self._parse_recognize_event_number(event_data.get("episode")) - if not recognized_title: + rematch_meta = MediaPluginOwner._prepare_video_help_meta(org_meta, event_data) + if rematch_meta is None: return None - if recognized_title == "Unknown": - return None - if not str(recognized_year).isdigit(): - recognized_year = None - # 结果赋值 - if recognized_title == org_meta.name and recognized_year == org_meta.year: - logger.info("辅助识别与原始识别结果一致，无需重新识别媒体信息") - return None - logger.info("辅助识别结果与原始识别结果不一致，重新匹配媒体信息 ...") - org_meta.name = recognized_title - org_meta.year = recognized_year - org_meta.begin_season = season_number - org_meta.begin_episode = episode_number - if org_meta.begin_season is not None or org_meta.begin_episode is not None: - org_meta.type = MediaType.TV - # 重新识别 return await self.async_recognize_media( - meta=org_meta, + meta=rematch_meta, media_source=media_source, share_meta=share_meta, episode_group=episode_group, @@ -592,41 +621,17 @@ class MediaPluginOwner(_MediaPluginOwnerBase): :param media_source: 请求级识别数据源 :param music_type: 音乐实体类型 """ - # 发送音乐名称识别事件，等待插件返回标题要素 result = await self.eventmanager.async_send_event( ChainEventType.MusicNameRecognize, - { - "title": title, - "artist": org_meta.artist, - "album": org_meta.album, - "year": org_meta.year, - "music_type": music_type, - }, + MediaPluginOwner._music_help_payload(title, org_meta, music_type), ) if not result: return None event_data = result.event_data or {} logger.info(f"获取到音乐辅助识别结果：{event_data}") - name, artist, album, year = self._parse_music_recognize_event(event_data) - if not name: + new_meta = MediaPluginOwner._prepare_music_help_meta(org_meta, event_data) + if new_meta is None: return None - # 辅助识别要素与原始一致时无需重新匹配 - if ( - name == org_meta.title - and (not artist or artist in org_meta.artists) - and (not album or album == org_meta.album) - and (not year or year == _normalize_music_year(org_meta.year)) - ): - logger.info("音乐辅助识别与原始识别结果一致，无需重新匹配媒体信息") - return None - logger.info("音乐辅助识别结果与原始识别结果不一致，重新匹配媒体信息 ...") - new_meta = self._build_music_help_meta( - org_meta=org_meta, - name=name, - artist=artist, - album=album, - year=year, - ) # 重新识别，仅采信取得远端身份的结果，否则由选择流程保留原生兜底 mediainfo = await self.async_recognize_media( meta=new_meta, diff --git a/app/chain/media/recognition.py b/app/chain/media/recognition.py index 2ca91514f..c08dc9425 100644 --- a/app/chain/media/recognition.py +++ b/app/chain/media/recognition.py @@ -1,7 +1,9 @@ """媒体原生路由与同步异步识别回退 owner。""" from copy import deepcopy -from typing import Any, Optional, cast +from dataclasses import dataclass +from enum import Enum +from typing import Any, Mapping, Optional, cast from app.chain.base import ChainBase from app.chain.media.contract import _MediaOwnerBase @@ -21,21 +23,65 @@ from app.schemas.types import ( ) +class _NativeRecognitionAction(Enum): + """标识媒体识别入口下一步需要执行的真实 I/O 动作。""" + + MUSIC = "music" + VIDEO = "video" + NONE = "none" + + +@dataclass(frozen=True, slots=True) +class _NativeRecognitionPlan: + """冻结同步与异步原生识别入口共用的路由和调用参数。""" + + action: _NativeRecognitionAction + kwargs: Mapping[str, Any] + refresh_cache: bool = False + + +@dataclass(frozen=True, slots=True) +class _MetaRecognitionRequest: + """冻结一次按标题识别所需的元数据、副本和来源决策。""" + + title: str + metainfo: MetaBase + share_meta: MetaBase + mtype: Optional[MediaType] + media_source: Optional[MediaSource] + episode_group: Optional[str] + music_type: Optional[str] + is_music: bool + + @property + def plugin_event(self) -> ChainEventType: + """返回与媒体类型匹配的辅助识别事件。""" + return ( + ChainEventType.MusicNameRecognize + if self.is_music + else ChainEventType.NameRecognize + ) + + class MediaRecognitionOwner(_MediaOwnerBase): """媒体原生路由与同步异步识别回退 owner。""" - def _run_native_media_recognize( + def _native_recognition_plan( self, - module_kwargs: dict[str, Any], + module_kwargs: Mapping[str, Any], cache: bool, - ) -> Optional[MediaInfo]: - """统一同步媒体识别路由，未指定来源时影视和音乐只使用各自主数据源。""" + ) -> _NativeRecognitionPlan: + """把媒体类型、显式来源和默认来源归并为唯一原生识别计划。""" meta = module_kwargs.get("meta") mtype = module_kwargs.get("mtype") media_source = module_kwargs.get("media_source") - if isinstance(meta, MetaMusic) or mtype == MediaType.MUSIC or is_music_media_source(media_source): + if ( + isinstance(meta, MetaMusic) + or mtype == MediaType.MUSIC + or is_music_media_source(media_source) + ): if media_source: - recognize_kwargs = { + recognize_kwargs: dict[str, Any] = { "media_source": media_source, "meta": meta if isinstance(meta, MetaMusic) else None, "media_id": module_kwargs.get("media_id"), @@ -43,29 +89,52 @@ class MediaRecognitionOwner(_MediaOwnerBase): } if "music_type" in module_kwargs: recognize_kwargs["music_type"] = module_kwargs["music_type"] - with fresh(not cache): - return cast( - Optional[MediaInfo], - self.recognize_music_from_source(**recognize_kwargs), - ) + return _NativeRecognitionPlan( + action=_NativeRecognitionAction.MUSIC, + kwargs=recognize_kwargs, + refresh_cache=not cache, + ) if isinstance(meta, MetaMusic): + return _NativeRecognitionPlan( + action=_NativeRecognitionAction.MUSIC, + kwargs={ + "media_source": self._music_primary_source, + "meta": meta, + "cache": cache, + "music_type": MUSIC_ENTITY_RECORDING, + }, + ) + return _NativeRecognitionPlan( + action=_NativeRecognitionAction.NONE, + kwargs={}, + ) + video_kwargs = dict(module_kwargs) + if not media_source and isinstance(meta, MetaBase): + video_kwargs["media_source"] = self._video_primary_source + return _NativeRecognitionPlan( + action=_NativeRecognitionAction.VIDEO, + kwargs=video_kwargs, + ) + + def _run_native_media_recognize( + self, + module_kwargs: dict[str, Any], + cache: bool, + ) -> Optional[MediaInfo]: + """统一同步媒体识别路由，未指定来源时影视和音乐只使用各自主数据源。""" + plan = MediaRecognitionOwner._native_recognition_plan( + self, module_kwargs, cache + ) + if plan.action is _NativeRecognitionAction.NONE: + return None + if plan.action is _NativeRecognitionAction.MUSIC: + with fresh(plan.refresh_cache): return cast( Optional[MediaInfo], - self.recognize_music_from_source( - media_source=self._music_primary_source, - meta=meta, - cache=cache, - music_type=MUSIC_ENTITY_RECORDING, - ), + self.recognize_music_from_source(**plan.kwargs), ) - return None - if not media_source and isinstance(meta, MetaBase): - module_kwargs = { - **module_kwargs, - "media_source": self._video_primary_source, - } return ChainBase._run_native_media_recognize( - cast(ChainBase, self), module_kwargs, cache + cast(ChainBase, self), dict(plan.kwargs), cache ) async def _async_run_native_media_recognize( @@ -74,44 +143,61 @@ class MediaRecognitionOwner(_MediaOwnerBase): cache: bool, ) -> Optional[MediaInfo]: """统一异步媒体识别路由，未指定来源时影视和音乐只使用各自主数据源。""" - meta = module_kwargs.get("meta") - mtype = module_kwargs.get("mtype") - media_source = module_kwargs.get("media_source") - if isinstance(meta, MetaMusic) or mtype == MediaType.MUSIC or is_music_media_source(media_source): - if media_source: - recognize_kwargs = { - "media_source": media_source, - "meta": meta if isinstance(meta, MetaMusic) else None, - "media_id": module_kwargs.get("media_id"), - "cache": cache, - } - if "music_type" in module_kwargs: - recognize_kwargs["music_type"] = module_kwargs["music_type"] - async with async_fresh(not cache): - return cast( - Optional[MediaInfo], - await self.async_recognize_music_from_source(**recognize_kwargs), - ) - if isinstance(meta, MetaMusic): + plan = MediaRecognitionOwner._native_recognition_plan( + self, module_kwargs, cache + ) + if plan.action is _NativeRecognitionAction.NONE: + return None + if plan.action is _NativeRecognitionAction.MUSIC: + async with async_fresh(plan.refresh_cache): return cast( Optional[MediaInfo], - await self.async_recognize_music_from_source( - media_source=self._music_primary_source, - meta=meta, - cache=cache, - music_type=MUSIC_ENTITY_RECORDING, - ), + await self.async_recognize_music_from_source(**plan.kwargs), ) - return None - if not media_source and isinstance(meta, MetaBase): - module_kwargs = { - **module_kwargs, - "media_source": self._video_primary_source, - } return await ChainBase._async_run_native_media_recognize( - cast(ChainBase, self), module_kwargs, cache + cast(ChainBase, self), dict(plan.kwargs), cache ) + @staticmethod + def _meta_recognition_request( + metainfo: MetaBase, + *, + mtype: Optional[MediaType], + media_source: Optional[MediaSource], + episode_group: Optional[str], + music_type: Optional[str], + ) -> _MetaRecognitionRequest: + """为同步与异步标题识别冻结相同的元数据副本和来源语义。""" + return _MetaRecognitionRequest( + title=metainfo.title, + metainfo=metainfo, + share_meta=deepcopy(metainfo), + mtype=mtype, + media_source=media_source, + episode_group=episode_group, + music_type=music_type, + is_music=mtype == MediaType.MUSIC or isinstance(metainfo, MetaMusic), + ) + + @staticmethod + def _has_remote_identity(result: Optional[MediaInfo]) -> bool: + """音乐识别仅在取得远端来源身份后视为完整命中。""" + return bool(result and result.media_source) + + @staticmethod + def _accepted_recognition( + request: _MetaRecognitionRequest, + mediainfo: Optional[MediaInfo], + ) -> Optional[MediaInfo]: + """统一识别结果的空值判定与成功日志，图片补充留给各 I/O 外壳。""" + if not mediainfo: + return None + logger.info( + f"{request.title} 识别到媒体信息：" + f"{mediainfo.type.value} {mediainfo.title_year}" + ) + return mediainfo + def recognize_by_meta( self, metainfo: MetaBase, @@ -165,53 +251,54 @@ class MediaRecognitionOwner(_MediaOwnerBase): """ if not metainfo: return None - title = metainfo.title - share_meta = deepcopy(metainfo) - # 音乐原生兜底结果无远端身份，需按是否取得身份判定，才会请求辅助识别 - is_music = mtype == MediaType.MUSIC or isinstance(metainfo, MetaMusic) - is_recognized = (lambda result: bool(result and result.media_source)) if is_music else None + request = MediaRecognitionOwner._meta_recognition_request( + metainfo, + mtype=mtype, + media_source=media_source, + episode_group=episode_group, + music_type=music_type, + ) def native_recognize() -> Optional[MediaInfo]: """使用请求级数据源执行原生识别。""" return self.recognize_media( - meta=metainfo, - mtype=mtype, - media_source=media_source, - share_meta=share_meta, - episode_group=episode_group, - music_type=music_type, + meta=request.metainfo, + mtype=request.mtype, + media_source=request.media_source, + share_meta=request.share_meta, + episode_group=request.episode_group, + music_type=request.music_type, ) def plugin_recognize() -> Optional[MediaInfo]: """执行辅助识别并保持请求级数据源约束。""" - if is_music and not isinstance(metainfo, MetaMusic): + if request.is_music and not isinstance(request.metainfo, MetaMusic): return None return self.recognize_help( - title=title, - org_meta=metainfo, - share_meta=share_meta, - media_source=media_source, - episode_group=episode_group, - music_type=music_type, + title=request.title, + org_meta=request.metainfo, + share_meta=request.share_meta, + media_source=request.media_source, + episode_group=request.episode_group, + music_type=request.music_type, ) # 按 config 中设置的识别顺序识别，影视与音乐共用同一选择流程 mediainfo = self.select_recognize_source( - log_name=title, - log_context=title, + log_name=request.title, + log_context=request.title, native_fn=native_recognize, plugin_fn=plugin_recognize, - is_recognized=is_recognized, - plugin_event=( - ChainEventType.MusicNameRecognize - if is_music - else ChainEventType.NameRecognize + is_recognized=( + MediaRecognitionOwner._has_remote_identity + if request.is_music + else None ), + plugin_event=request.plugin_event, ) - if not mediainfo: + mediainfo = MediaRecognitionOwner._accepted_recognition(request, mediainfo) + if mediainfo is None: return None - # 识别成功 - logger.info(f"{title} 识别到媒体信息：{mediainfo.type.value} {mediainfo.title_year}") if obtain_images: self.obtain_images(mediainfo=mediainfo) return mediainfo @@ -270,52 +357,54 @@ class MediaRecognitionOwner(_MediaOwnerBase): """ if not metainfo: return None - title = metainfo.title - share_meta = deepcopy(metainfo) - # 音乐原生兜底结果无远端身份，需按是否取得身份判定，才会请求辅助识别 - is_music = mtype == MediaType.MUSIC or isinstance(metainfo, MetaMusic) - is_recognized = (lambda result: bool(result and result.media_source)) if is_music else None + request = MediaRecognitionOwner._meta_recognition_request( + metainfo, + mtype=mtype, + media_source=media_source, + episode_group=episode_group, + music_type=music_type, + ) async def native_recognize() -> Optional[MediaInfo]: """异步使用请求级数据源执行原生识别。""" return await self.async_recognize_media( - meta=metainfo, - mtype=mtype, - media_source=media_source, - share_meta=share_meta, - episode_group=episode_group, - music_type=music_type, + meta=request.metainfo, + mtype=request.mtype, + media_source=request.media_source, + share_meta=request.share_meta, + episode_group=request.episode_group, + music_type=request.music_type, ) async def plugin_recognize() -> Optional[MediaInfo]: """异步执行辅助识别并保持请求级数据源约束。""" - if is_music and not isinstance(metainfo, MetaMusic): + if request.is_music and not isinstance(request.metainfo, MetaMusic): return None return await self.async_recognize_help( - title=title, - org_meta=metainfo, - share_meta=share_meta, - media_source=media_source, - episode_group=episode_group, - music_type=music_type, + title=request.title, + org_meta=request.metainfo, + share_meta=request.share_meta, + media_source=request.media_source, + episode_group=request.episode_group, + music_type=request.music_type, ) # 按 config 中设置的识别顺序识别，影视与音乐共用同一选择流程 mediainfo = await self.async_select_recognize_source( - log_name=title, - log_context=title, + log_name=request.title, + log_context=request.title, native_fn=native_recognize, plugin_fn=plugin_recognize, - is_recognized=is_recognized, - plugin_event=( - ChainEventType.MusicNameRecognize - if is_music - else ChainEventType.NameRecognize + is_recognized=( + MediaRecognitionOwner._has_remote_identity + if request.is_music + else None ), + plugin_event=request.plugin_event, ) - if not mediainfo: + mediainfo = MediaRecognitionOwner._accepted_recognition(request, mediainfo) + if mediainfo is None: return None - logger.info(f"{title} 识别到媒体信息：{mediainfo.type.value} {mediainfo.title_year}") if obtain_images: await self.async_obtain_images(mediainfo=mediainfo) return mediainfo