Merge pull request #45 from quanru/feat/recognition-provider-foundation

refactor(asr): add recognition provider registry

Author: quanru

src/doubao_input/app.py

@@ -26,7 +26,6 @@
 from doubao_input.result import RecentResult
 from doubao_input.doubao.params_store import ParamsStore
 from doubao_input.doubao.transcription import TranscriptionManager
-from doubao_input.doubao.asr_client import ASRClient
 from doubao_input.doubao.volcengine_asr_client import VolcengineASRClient
 from doubao_input.doubao.volcengine_credentials import (
     VolcengineCredentials, VolcengineCredentialsStore,
@@ -48,6 +47,7 @@
 from doubao_input.polish_preview import PolishPreview
 from doubao_input.trigger.escape_guard import EscapeGuard
 from doubao_input.diagnostics import DiagnosticTrace, report as diagnostic_report
+from doubao_input.recognition_providers import recognition_provider
 
 logger = logging.getLogger(__name__)
 
@@ -347,29 +347,29 @@ def apply_runtime(value, strict):
         self._control.refresh()
 
     def _new_transcription_manager(self):
-        if self.settings.asr_provider == "volcengine":
-            return TranscriptionManager(self.app_state,
-                asr_client=VolcengineASRClient(),
-                credential_store=VolcengineCredentialsStore,
-                interactive_auth=False, clear_rejected_credentials=False)
-        return TranscriptionManager(self.app_state, asr_client=ASRClient(),
-            credential_store=ParamsStore, interactive_auth=True,
-            clear_rejected_credentials=True)
+        provider = recognition_provider(self.settings.asr_provider)
+        return TranscriptionManager(
+            self.app_state,
+            asr_client=provider.new_client(),
+            credential_store=provider.credential_store,
+            interactive_auth=provider.interactive_auth,
+            clear_rejected_credentials=provider.clear_rejected_credentials,
+        )
 
     def _configure_recognition_backend(self):
-        if self.settings.asr_provider == "volcengine":
-            self._tm.configure_backend(VolcengineASRClient(),
-                VolcengineCredentialsStore, interactive_auth=False,
-                clear_rejected_credentials=False)
-        else:
-            self._tm.configure_backend(ASRClient(), ParamsStore,
-                interactive_auth=True, clear_rejected_credentials=True)
+        provider = recognition_provider(self.settings.asr_provider)
+        self._tm.configure_backend(
+            provider.new_client(),
+            provider.credential_store,
+            interactive_auth=provider.interactive_auth,
+            clear_rejected_credentials=provider.clear_rejected_credentials,
+        )
 
     def _recognition_ready(self):
         try:
-            store = (VolcengineCredentialsStore if self.settings.asr_provider == "volcengine"
-                     else ParamsStore)
-            return store.has_saved()
+            return recognition_provider(
+                self.settings.asr_provider
+            ).credential_store.has_saved()
         except (OSError, ValueError):
             return False
 
@@ -765,10 +765,8 @@ def _summary(self):
                 "microphone_id": self.settings.microphone,
                 "microphone_ok": bool(setup and setup.microphone_ok),
                 "asr_provider": self.settings.asr_provider,
-                "asr_provider_name": tr(
-                    "Volcengine official API", "火山引擎官方 API")
-                    if self.settings.asr_provider == "volcengine" else
-                    tr("Doubao account", "豆包账号"),
+                "asr_provider_name": recognition_provider(
+                    self.settings.asr_provider).name,
                 "voice_test_ok": bool(setup and setup.voice_ok),
                 "onboarding_complete": self.settings.onboarding_complete,
                 "result": self.recent.text, "status": self.recent.status}
@@ -816,9 +814,10 @@ def _sign_out(self):
         if self._login_window:
             self._login_window.destroy()
             self._login_window = None
-        official = getattr(getattr(self, "settings", None), "asr_provider", "doubao") == "volcengine"
-        store = (VolcengineCredentialsStore if official
-                 else ParamsStore)
+        provider = recognition_provider(
+            getattr(getattr(self, "settings", None), "asr_provider", "doubao"))
+        official = provider.id == "volcengine"
+        store = provider.credential_store
         try:
             store.clear()
         except OSError as error:

src/doubao_input/recognition_providers.py

@@ -0,0 +1,100 @@
+"""Recognition provider registry shared by settings, UI, and runtime wiring."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Protocol
+
+from doubao_input.i18n import tr
+
+
+class CredentialStore(Protocol):
+    @classmethod
+    def load(cls): ...
+
+    @classmethod
+    def clear(cls) -> None: ...
+
+    @classmethod
+    def has_saved(cls) -> bool: ...
+
+
+@dataclass(frozen=True)
+class RecognitionProvider:
+    id: str
+    name_en: str
+    name_zh: str
+    client_factory: Callable[[], object]
+    credential_store_factory: Callable[[], type[CredentialStore]]
+    interactive_auth: bool
+    clear_rejected_credentials: bool
+
+    @property
+    def name(self) -> str:
+        return tr(self.name_en, self.name_zh)
+
+    def new_client(self):
+        return self.client_factory()
+
+    @property
+    def credential_store(self) -> type[CredentialStore]:
+        return self.credential_store_factory()
+
+
+def _doubao_client():
+    from doubao_input.doubao.asr_client import ASRClient
+    return ASRClient()
+
+
+def _doubao_store():
+    from doubao_input.doubao.params_store import ParamsStore
+    return ParamsStore
+
+
+def _volcengine_client():
+    from doubao_input.doubao.volcengine_asr_client import VolcengineASRClient
+    return VolcengineASRClient()
+
+
+def _volcengine_store():
+    from doubao_input.doubao.volcengine_credentials import VolcengineCredentialsStore
+    return VolcengineCredentialsStore
+
+
+_PROVIDERS = (
+    RecognitionProvider(
+        id="doubao",
+        name_en="Doubao account",
+        name_zh="豆包账号",
+        client_factory=_doubao_client,
+        credential_store_factory=_doubao_store,
+        interactive_auth=True,
+        clear_rejected_credentials=True,
+    ),
+    RecognitionProvider(
+        id="volcengine",
+        name_en="Volcengine official API",
+        name_zh="火山引擎官方 API",
+        client_factory=_volcengine_client,
+        credential_store_factory=_volcengine_store,
+        interactive_auth=False,
+        clear_rejected_credentials=False,
+    ),
+)
+
+RECOGNITION_PROVIDER_IDS = tuple(provider.id for provider in _PROVIDERS)
+_PROVIDERS_BY_ID = {provider.id: provider for provider in _PROVIDERS}
+
+
+def recognition_providers() -> tuple[RecognitionProvider, ...]:
+    return _PROVIDERS
+
+
+def recognition_provider(provider_id: str) -> RecognitionProvider:
+    try:
+        return _PROVIDERS_BY_ID[provider_id]
+    except KeyError as error:
+        raise ValueError(tr(
+            "Unsupported recognition service",
+            "不支持的语音识别服务",
+        )) from error

src/doubao_input/settings.py

@@ -7,11 +7,12 @@
 import tempfile
 from urllib.parse import urlsplit
 from doubao_input.i18n import LANGUAGES, tr
+from doubao_input.recognition_providers import RECOGNITION_PROVIDER_IDS
 
 KEY_CHOICES = {"Disabled": 0, "Fn": 464, "Ctrl": 29, "Shift": 42,
                "Alt": 56, "Meta": 125, "F8": 66, "F9": 67}
 REMOVED_SETTING_FIELDS = frozenset({"polish_undo_modifier", "polish_prompt"})
-ASR_PROVIDERS = ("doubao", "volcengine")
+ASR_PROVIDERS = RECOGNITION_PROVIDER_IDS
 CAPTURABLE_KEY_CODES = frozenset(range(2, 249)) | {464}
 MODIFIER_KEY_CODES = frozenset({29, 42, 54, 56, 97, 100, 125, 126})
 EQUIVALENT_KEY_GROUPS = (

src/doubao_input/ui/control_window.py

@@ -10,6 +10,7 @@
 from doubao_input.ui.style import apply_window_style
 from doubao_input.product import VERSION
 from doubao_input.settings import ASR_PROVIDERS
+from doubao_input.recognition_providers import recognition_providers
 
 
 class ControlWindow:
@@ -324,10 +325,8 @@ def button(box, text, callback, suggested=False):
         provider_row.append(Gtk.Label(
             label=tr("Recognition service", "语音识别服务"),
             xalign=0, hexpand=True, wrap=True))
-        self._asr_provider = Gtk.DropDown.new_from_strings([
-            tr("Doubao account", "豆包账号"),
-            tr("Volcengine API", "火山引擎 API"),
-        ])
+        self._asr_provider = Gtk.DropDown.new_from_strings(
+            [provider.name for provider in recognition_providers()])
         self._asr_provider.set_selected(ASR_PROVIDERS.index(
             self._actions.summary().get("asr_provider", "doubao")))
         self._asr_provider.connect("notify::selected", self._asr_provider_changed)

src/doubao_input/ui/settings_window.py

@@ -9,6 +9,7 @@
 from doubao_input.settings import INPUT_METHODS, WAVEFORM_STYLES
 from doubao_input.ui.style import apply_window_style
 from doubao_input.settings import ASR_PROVIDERS
+from doubao_input.recognition_providers import recognition_providers
 
 
 class SettingsWindow:
@@ -103,10 +104,8 @@ def run(*_):
             "更改会自动保存；选择“系统默认”可跟随桌面的输入设备。")))
 
         section(tr("Recognition service", "语音识别服务"))
-        self.asr_provider = Gtk.DropDown.new_from_strings([
-            tr("Doubao account", "豆包账号"),
-            tr("Volcengine official API", "火山引擎官方 API"),
-        ])
+        self.asr_provider = Gtk.DropDown.new_from_strings(
+            [provider.name for provider in recognition_providers()])
         self.asr_provider.set_selected(ASR_PROVIDERS.index(settings.asr_provider))
         row(tr("Service", "服务"), self.asr_provider)
         self.asr_details = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)

tests/unit/test_recognition_providers.py

@@ -0,0 +1,38 @@
+from unittest import TestCase
+
+from doubao_input.doubao.asr_client import ASRClient
+from doubao_input.doubao.params_store import ParamsStore
+from doubao_input.doubao.volcengine_asr_client import VolcengineASRClient
+from doubao_input.doubao.volcengine_credentials import VolcengineCredentialsStore
+from doubao_input.recognition_providers import (
+    RECOGNITION_PROVIDER_IDS,
+    recognition_provider,
+    recognition_providers,
+)
+
+
+class RecognitionProviderRegistryTest(TestCase):
+    def test_registry_order_drives_persisted_ids_and_ui_names(self):
+        providers = recognition_providers()
+        self.assertEqual(
+            RECOGNITION_PROVIDER_IDS,
+            tuple(provider.id for provider in providers),
+        )
+        self.assertEqual(
+            [provider.name for provider in providers],
+            ["Doubao account", "Volcengine official API"],
+        )
+
+    def test_registry_owns_runtime_backend_configuration(self):
+        doubao = recognition_provider("doubao")
+        self.assertIsInstance(doubao.new_client(), ASRClient)
+        self.assertIs(doubao.credential_store, ParamsStore)
+        self.assertTrue(doubao.interactive_auth)
+        official = recognition_provider("volcengine")
+        self.assertIsInstance(official.new_client(), VolcengineASRClient)
+        self.assertIs(official.credential_store, VolcengineCredentialsStore)
+        self.assertFalse(official.interactive_auth)
+
+    def test_unknown_provider_is_rejected(self):
+        with self.assertRaisesRegex(ValueError, "Unsupported recognition service"):
+            recognition_provider("missing")