diff --git a/app/chain/media/plugin.py b/app/chain/media/plugin.py index 6b6fce72a..63b67e2fc 100644 --- a/app/chain/media/plugin.py +++ b/app/chain/media/plugin.py @@ -1,6 +1,18 @@ """插件识别选择、事件解析与辅助识别 owner。""" -from typing import TYPE_CHECKING, Awaitable, Callable, Mapping, Optional, Protocol, Tuple +from dataclasses import dataclass +from enum import Enum +from typing import ( + TYPE_CHECKING, + Awaitable, + Callable, + Generator, + Mapping, + Optional, + Protocol, + Tuple, + cast, +) from app.application.configuration import get_chain_runtime_config_snapshot from app.chain.media.contract import _MediaOwnerBase @@ -21,6 +33,39 @@ AsyncRecognitionCallback = Callable[[], Awaitable[Optional[MediaInfo]]] RecognitionPredicate = Callable[[Optional[MediaInfo]], bool] +class _RecognitionSource(Enum): + """标识一次识别选择流程当前需要调用的真实 I/O 来源。""" + + NATIVE = "native" + PLUGIN = "plugin" + + +@dataclass(frozen=True, slots=True) +class _RecognitionSelection: + """统一同步与异步识别入口的来源顺序、回退和结果采信决策。""" + + plugin_first: bool + plugin_available: bool + is_recognized: RecognitionPredicate + + def flow( + self, + ) -> Generator[_RecognitionSource, Optional[MediaInfo], Optional[MediaInfo]]: + """按相同状态机产出 I/O 动作，并接收各入口自己的调用结果。""" + if self.plugin_first and self.plugin_available: + helped = yield _RecognitionSource.PLUGIN + if self.is_recognized(helped): + return helped + native = yield _RecognitionSource.NATIVE + return native or helped + + native = yield _RecognitionSource.NATIVE + if self.is_recognized(native) or not self.plugin_available: + return native + helped = yield _RecognitionSource.PLUGIN + return helped if self.is_recognized(helped) else native + + def _normalize_music_year(value: object) -> Optional[int]: """把本地或插件音乐年份统一为正整数，空值和非法文本返回空。""" text = str(value or "").strip() @@ -98,6 +143,102 @@ else: class MediaPluginOwner(_MediaPluginOwnerBase): """插件识别选择、事件解析与辅助识别 owner。""" + def _recognition_selection( + self, + plugin_event: ChainEventType, + is_recognized: Optional[RecognitionPredicate], + ) -> _RecognitionSelection: + """冻结本次识别的插件可用性、优先级和结果采信谓词。""" + return _RecognitionSelection( + plugin_first=get_chain_runtime_config_snapshot().recognize_plugin_first, + plugin_available=self.eventmanager.check(plugin_event), + is_recognized=is_recognized or bool, + ) + + @staticmethod + def _log_recognition_action( + action: _RecognitionSource, + *, + plugin_first: bool, + log_name: str, + log_context: str, + ) -> None: + """在状态机请求真实 I/O 前记录统一的来源切换原因。""" + if action is _RecognitionSource.PLUGIN: + if plugin_first: + logger.info( + f"插件识别优先模式已开启。请求辅助识别，标题：{log_name} ..." + ) + else: + logger.info( + f"原生识别未识别到 {log_context} 的媒体信息，尝试使用辅助识别 ..." + ) + return + if plugin_first: + logger.info( + f"辅助识别未识别到 {log_context} 的媒体信息，尝试使用原生识别 ..." + ) + else: + logger.info(f"开始识别标题：{log_name} ...") + + def _run_recognition_selection( + self, + selection: _RecognitionSelection, + *, + log_name: str, + log_context: str, + native_fn: RecognitionCallback, + plugin_fn: RecognitionCallback, + ) -> Optional[MediaInfo]: + """同步驱动共享选择状态机，并只在动作边界调用同步实现。""" + flow = selection.flow() + try: + action = next(flow) + while True: + MediaPluginOwner._log_recognition_action( + action, + plugin_first=selection.plugin_first, + log_name=log_name, + log_context=log_context, + ) + result = ( + plugin_fn() + if action is _RecognitionSource.PLUGIN + else native_fn() + ) + action = flow.send(result) + except StopIteration as completed: + return cast(Optional[MediaInfo], completed.value) + + async def _async_run_recognition_selection( + self, + selection: _RecognitionSelection, + *, + log_name: str, + log_context: str, + native_fn: AsyncRecognitionCallback, + plugin_fn: AsyncRecognitionCallback, + ) -> Optional[MediaInfo]: + """异步驱动共享选择状态机，并只在动作边界等待异步实现。""" + flow = selection.flow() + try: + action = next(flow) + while True: + MediaPluginOwner._log_recognition_action( + action, + plugin_first=selection.plugin_first, + log_name=log_name, + log_context=log_context, + ) + result = ( + await plugin_fn() + if action is _RecognitionSource.PLUGIN + else await native_fn() + ) + action = flow.send(result) + except StopIteration as completed: + return cast(Optional[MediaInfo], completed.value) + def select_recognize_source( self, log_name: str, @@ -118,33 +259,17 @@ class MediaPluginOwner(_MediaPluginOwnerBase): 需视为未识别才会请求辅助识别，影视默认按非空判定 :param plugin_event: 辅助识别对应的链式事件类型，音乐使用音乐名称识别事件 """ - if is_recognized is None: - is_recognized = bool - mediainfo: Optional[MediaInfo] = None - # 插件可用性检查走注入的事件管理器，避免链实例直连全局单例 - plugin_available = self.eventmanager.check(plugin_event) - if get_chain_runtime_config_snapshot().recognize_plugin_first and plugin_available: - # 插件优先 - logger.info(f"插件识别优先模式已开启。请求辅助识别，标题：{log_name} ...") - helped = plugin_fn() - if is_recognized(helped): - mediainfo = helped - else: - logger.info(f"辅助识别未识别到 {log_context} 的媒体信息，尝试使用原生识别 ...") - mediainfo = native_fn() - # 辅助结果不采信时保留原生兜底，避免丢失已有识别结果（音乐原生兜底恒非空） - if helped and not mediainfo: - mediainfo = helped - else: - # 原生优先 - logger.info(f"开始识别标题：{log_name} ...") - mediainfo = native_fn() - if not is_recognized(mediainfo) and plugin_available: - logger.info(f"原生识别未识别到 {log_context} 的媒体信息，尝试使用辅助识别 ...") - helped = plugin_fn() - if is_recognized(helped): - mediainfo = helped - return mediainfo + selection = MediaPluginOwner._recognition_selection( + self, plugin_event, is_recognized + ) + return MediaPluginOwner._run_recognition_selection( + self, + selection, + log_name=log_name, + log_context=log_context, + native_fn=native_fn, + plugin_fn=plugin_fn, + ) @staticmethod def _parse_recognize_event_number(value: object) -> Optional[int]: @@ -364,33 +489,17 @@ class MediaPluginOwner(_MediaPluginOwnerBase): :param is_recognized: 判定识别结果是否有效的谓词，语义同同步版本 :param plugin_event: 辅助识别对应的链式事件类型，音乐使用音乐名称识别事件 """ - if is_recognized is None: - is_recognized = bool - mediainfo: Optional[MediaInfo] = None - # 插件可用性检查走注入的事件管理器，避免链实例直连全局单例 - plugin_available = self.eventmanager.check(plugin_event) - if get_chain_runtime_config_snapshot().recognize_plugin_first and plugin_available: - # 插件优先 - logger.info(f"插件优先模式已开启。请求辅助识别，标题：{log_name} ...") - helped = await plugin_fn() - if is_recognized(helped): - mediainfo = helped - else: - logger.info(f"辅助识别未识别到 {log_context} 的媒体信息，尝试使用原生识别") - mediainfo = await native_fn() - # 辅助结果不采信时保留原生兜底，避免丢失已有识别结果（音乐原生兜底恒非空） - if helped and not mediainfo: - mediainfo = helped - else: - # 原生优先 - logger.info(f"识别标题：{log_name} ...") - mediainfo = await native_fn() - if not is_recognized(mediainfo) and plugin_available: - logger.info(f"原生识别未识别到 {log_context} 的媒体信息，尝试使用辅助识别") - helped = await plugin_fn() - if is_recognized(helped): - mediainfo = helped - return mediainfo + selection = MediaPluginOwner._recognition_selection( + self, plugin_event, is_recognized + ) + return await MediaPluginOwner._async_run_recognition_selection( + self, + selection, + log_name=log_name, + log_context=log_context, + native_fn=native_fn, + plugin_fn=plugin_fn, + ) async def async_recognize_help( self, diff --git a/tests/test_music_plugin_recognize.py b/tests/test_music_plugin_recognize.py index 29247d04f..e72ff6b41 100644 --- a/tests/test_music_plugin_recognize.py +++ b/tests/test_music_plugin_recognize.py @@ -6,6 +6,8 @@ import asyncio from unittest.mock import AsyncMock, MagicMock, Mock, patch +import pytest + from app.chain.base import ChainBase from app.chain.media import MediaChain from app.domain.context import MediaInfo, MusicInfo @@ -32,6 +34,93 @@ def _remote_music() -> MusicInfo: ) +@pytest.mark.parametrize( + ("plugin_first", "native_kind", "plugin_kind", "expected_kind", "order"), + [ + (False, "fallback", "remote", "remote", ["native", "plugin"]), + (False, "fallback", "other", "fallback", ["native", "plugin"]), + (True, "none", "fallback", "fallback", ["plugin", "native"]), + (True, "fallback", "remote", "remote", ["plugin"]), + ], +) +def test_recognize_source_selection_keeps_sync_async_parity( + monkeypatch, + plugin_first, + native_kind, + plugin_kind, + expected_kind, + order, +): + """同步与异步入口必须共享来源顺序、采信规则和无身份回退结果。""" + chain = MediaChain() + candidates = { + "none": None, + "fallback": _fallback_music(title="本地兜底"), + "other": _fallback_music(title="插件兜底"), + "remote": _remote_music(), + } + native = candidates[native_kind] + plugin = candidates[plugin_kind] + expected = candidates[expected_kind] + eventmanager = MagicMock() + eventmanager.check.return_value = True + monkeypatch.setattr(chain, "eventmanager", eventmanager) + + sync_order = [] + + def native_sync(): + """记录同步原生 I/O 动作。""" + sync_order.append("native") + return native + + def plugin_sync(): + """记录同步插件 I/O 动作。""" + sync_order.append("plugin") + return plugin + + async_order = [] + + async def native_async(): + """记录异步原生 I/O 动作。""" + async_order.append("native") + return native + + async def plugin_async(): + """记录异步插件 I/O 动作。""" + async_order.append("plugin") + return plugin + + def has_remote_identity(result): + """仅采信包含远端来源的识别结果。""" + return bool(result and result.media_source) + + with patch( + "app.runtime.config.settings.RECOGNIZE_PLUGIN_FIRST", + plugin_first, + ): + sync_result = chain.select_recognize_source( + log_name="测试标题", + log_context="测试标题", + native_fn=native_sync, + plugin_fn=plugin_sync, + is_recognized=has_remote_identity, + ) + async_result = asyncio.run( + chain.async_select_recognize_source( + log_name="测试标题", + log_context="测试标题", + native_fn=native_async, + plugin_fn=plugin_async, + is_recognized=has_remote_identity, + ) + ) + + assert sync_result is expected + assert async_result is expected + assert sync_order == order + assert async_order == order + + def test_music_recognize_help_sends_event_and_rematches(monkeypatch): """原生识别无远端身份时应发送音乐名称识别事件，并按修正要素重新匹配。""" chain = MediaChain()