48 lines
2.0 KiB
Python
48 lines
2.0 KiB
Python
"""Factory + реестр для реализаций плагинов Transcriber/Summarizer."""
|
||
|
||
from core.plugins.config import SummarizerConfig, TranscriberConfig
|
||
from core.plugins.summarizer import Summarizer
|
||
from core.plugins.transcriber import Transcriber
|
||
|
||
|
||
class PluginError(Exception):
|
||
"""Базовая ошибка при сбое реестра/factory плагинов."""
|
||
|
||
|
||
class UnknownProviderError(PluginError):
|
||
"""Вызывается, когда запрошенный `provider` плагина не зарегистрирован."""
|
||
|
||
|
||
_TRANSCRIBERS: dict[str, type[Transcriber]] = {}
|
||
_SUMMARIZERS: dict[str, type[Summarizer]] = {}
|
||
|
||
|
||
def register_transcriber[T: type[Transcriber]](cls: T) -> T:
|
||
"""Зарегистрировать подкласс `Transcriber` под его ключом `provider`."""
|
||
_TRANSCRIBERS[cls.provider] = cls
|
||
return cls
|
||
|
||
|
||
def register_summarizer[S: type[Summarizer]](cls: S) -> S:
|
||
"""Зарегистрировать подкласс `Summarizer` под его ключом `provider`."""
|
||
_SUMMARIZERS[cls.provider] = cls
|
||
return cls
|
||
|
||
|
||
def create_transcriber(cfg: TranscriberConfig) -> Transcriber:
|
||
"""Инстанцировать `Transcriber`, зарегистрированный для `cfg.provider`."""
|
||
try:
|
||
cls = _TRANSCRIBERS[cfg.provider]
|
||
except KeyError as exc:
|
||
raise UnknownProviderError(f"Неизвестный провайдер transcriber: {cfg.provider!r}") from exc
|
||
return cls(model=cfg.model, language=cfg.language, **cfg.options) # type: ignore[call-arg]
|
||
|
||
|
||
def create_summarizer(cfg: SummarizerConfig) -> Summarizer:
|
||
"""Инстанцировать `Summarizer`, зарегистрированный для `cfg.provider`."""
|
||
try:
|
||
cls = _SUMMARIZERS[cfg.provider]
|
||
except KeyError as exc:
|
||
raise UnknownProviderError(f"Неизвестный провайдер summarizer: {cfg.provider!r}") from exc
|
||
return cls(model=cfg.model, chunk_minutes=cfg.chunk_minutes, **cfg.options) # type: ignore[call-arg]
|