refactor(media): unify recognition source decisions

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,

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()