From d155dc4deca1bdb20547446eb576f0d7a6ffa2ea Mon Sep 17 00:00:00 2001 From: lucas Date: Sat, 30 May 2026 18:49:01 +0900 Subject: [PATCH] Add IBM Watson transcription provider Signed-off-by: lucas --- .env.example | 5 ++ sapat/providers/ibm_watson.py | 99 ++++++++++++++++++++ tests/providers/test_ibm_watson.py | 140 +++++++++++++++++++++++++++++ 3 files changed, 244 insertions(+) create mode 100644 sapat/providers/ibm_watson.py create mode 100644 tests/providers/test_ibm_watson.py diff --git a/.env.example b/.env.example index 3e70778..d9a7000 100644 --- a/.env.example +++ b/.env.example @@ -37,6 +37,11 @@ LOCALAI_API_KEY= # --- ElevenLabs --- ELEVENLABS_API_KEY= +# --- IBM Watson Speech to Text --- +IBM_WATSON_STT_API_KEY= +IBM_WATSON_STT_URL=https://api.us-south.speech-to-text.watson.cloud.ibm.com/instances/YOUR_INSTANCE_ID +# IBM_WATSON_STT_TIMEOUT=120 + # --- Symbl.ai --- SYMBL_ACCESS_TOKEN= # Or use app credentials instead: diff --git a/sapat/providers/ibm_watson.py b/sapat/providers/ibm_watson.py new file mode 100644 index 0000000..da19696 --- /dev/null +++ b/sapat/providers/ibm_watson.py @@ -0,0 +1,99 @@ +# ABOUTME: IBM Watson Speech to Text transcription provider +# ABOUTME: Uses the synchronous /v1/recognize REST API with API key auth + +import os +from pathlib import Path +from typing import Optional + +import requests + +from sapat.providers import register +from sapat.providers.base import ( + AudioFormat, + ProviderConfig, + TranscriptionProvider, + TranscriptionResult, +) + + +@register +class IBMWatsonProvider(TranscriptionProvider): + name = "ibm_watson" + config = ProviderConfig( + required_env_vars=["IBM_WATSON_STT_API_KEY", "IBM_WATSON_STT_URL"], + max_file_size_mb=100.0, + preferred_format=AudioFormat.WAV, + supports_correction=False, + default_model="en-US_BroadbandModel", + ) + + def resolve_model(self, model_alias: str) -> str: + aliases = { + "en": "en-US_BroadbandModel", + "en-us": "en-US_BroadbandModel", + "default": self.config.default_model, + } + return aliases.get(model_alias.lower(), model_alias) + + def transcribe( + self, + audio_file: str, + model: str, + language: str = "en", + prompt: Optional[str] = None, + temperature: float = 0, + **kwargs, + ) -> TranscriptionResult: + api_key = os.getenv("IBM_WATSON_STT_API_KEY", "") + service_url = os.getenv("IBM_WATSON_STT_URL", "").rstrip("/") + timeout = float(os.getenv("IBM_WATSON_STT_TIMEOUT", "120")) + + params = {"model": model} if model else {} + headers = {"Content-Type": self._content_type(audio_file)} + + with open(audio_file, "rb") as audio: + response = requests.post( + f"{service_url}/v1/recognize", + auth=("apikey", api_key), + headers=headers, + params=params, + data=audio, + timeout=timeout, + ) + + if response.status_code != 200: + raise RuntimeError( + f"IBM Watson transcription failed ({response.status_code}): {response.text}" + ) + + payload = response.json() + text = self._extract_transcript(payload) + return TranscriptionResult(text=text, raw_response=payload) + + @staticmethod + def _content_type(audio_file: str) -> str: + suffix = Path(audio_file).suffix.lower() + return { + ".flac": "audio/flac", + ".mp3": "audio/mp3", + ".mpeg": "audio/mpeg", + ".ogg": "audio/ogg", + ".wav": "audio/wav", + ".webm": "audio/webm", + }.get(suffix, "application/octet-stream") + + @staticmethod + def _extract_transcript(payload: dict) -> str: + transcripts = [] + for result in payload.get("results", []): + alternatives = result.get("alternatives") or [] + if not alternatives: + continue + transcript = (alternatives[0].get("transcript") or "").strip() + if transcript: + transcripts.append(transcript) + + if not transcripts: + raise RuntimeError("IBM Watson transcription returned no transcript text") + + return "\n".join(transcripts) diff --git a/tests/providers/test_ibm_watson.py b/tests/providers/test_ibm_watson.py new file mode 100644 index 0000000..e7fc0cc --- /dev/null +++ b/tests/providers/test_ibm_watson.py @@ -0,0 +1,140 @@ +# ABOUTME: Tests for IBM Watson Speech to Text provider +# ABOUTME: Verifies auth, payload shape, response parsing, and availability gates + +import os +from unittest.mock import patch + +import pytest + +from sapat.providers.base import TranscriptionResult + + +class FakeResponse: + def __init__(self, status_code=200, payload=None, text=""): + self.status_code = status_code + self._payload = payload or {} + self.text = text + + def json(self): + return self._payload + + +@pytest.fixture +def audio_file(tmp_path): + path = tmp_path / "sample.wav" + path.write_bytes(b"fake wav bytes") + return str(path) + + +class TestIBMWatsonProvider: + @patch.dict( + os.environ, + { + "IBM_WATSON_STT_API_KEY": "test-key", + "IBM_WATSON_STT_URL": "https://ibm.example.com/instance", + }, + clear=True, + ) + def test_available_with_required_config(self): + from sapat.providers.ibm_watson import IBMWatsonProvider + + assert IBMWatsonProvider.is_available() is True + + @patch.dict(os.environ, {"IBM_WATSON_STT_API_KEY": "test-key"}, clear=True) + def test_not_available_without_service_url(self): + from sapat.providers.ibm_watson import IBMWatsonProvider + + assert IBMWatsonProvider.is_available() is False + + @patch.dict(os.environ, {}, clear=True) + def test_not_available_without_key(self): + from sapat.providers.ibm_watson import IBMWatsonProvider + + assert IBMWatsonProvider.is_available() is False + + @patch.dict( + os.environ, + { + "IBM_WATSON_STT_API_KEY": "test-key", + "IBM_WATSON_STT_URL": "https://ibm.example.com/instance/", + }, + clear=True, + ) + @patch("sapat.providers.ibm_watson.requests.post") + def test_transcribe_sends_binary_recognize_request(self, mock_post, audio_file): + mock_post.return_value = FakeResponse( + payload={ + "results": [ + {"alternatives": [{"transcript": "hello "}]}, + {"alternatives": [{"transcript": "world"}]}, + ] + } + ) + + from sapat.providers.ibm_watson import IBMWatsonProvider + + provider = IBMWatsonProvider() + result = provider.transcribe(audio_file, model="en-US_BroadbandModel") + + assert isinstance(result, TranscriptionResult) + assert result.text == "hello\nworld" + + assert ( + mock_post.call_args.args[0] + == "https://ibm.example.com/instance/v1/recognize" + ) + kwargs = mock_post.call_args.kwargs + assert kwargs["auth"] == ("apikey", "test-key") + assert kwargs["headers"] == {"Content-Type": "audio/wav"} + assert kwargs["params"] == {"model": "en-US_BroadbandModel"} + assert kwargs["timeout"] == 120.0 + + @patch.dict( + os.environ, + { + "IBM_WATSON_STT_API_KEY": "test-key", + "IBM_WATSON_STT_URL": "https://ibm.example.com/instance", + }, + clear=True, + ) + @patch("sapat.providers.ibm_watson.requests.post") + def test_raises_on_api_error(self, mock_post, audio_file): + mock_post.return_value = FakeResponse(status_code=401, text="unauthorized") + + from sapat.providers.ibm_watson import IBMWatsonProvider + + provider = IBMWatsonProvider() + with pytest.raises(RuntimeError, match="401"): + provider.transcribe(audio_file, model="en-US_BroadbandModel") + + def test_extract_transcript_requires_text(self): + from sapat.providers.ibm_watson import IBMWatsonProvider + + with pytest.raises(RuntimeError, match="no transcript"): + IBMWatsonProvider._extract_transcript({"results": []}) + + def test_content_type_from_suffix(self): + from sapat.providers.ibm_watson import IBMWatsonProvider + + assert IBMWatsonProvider._content_type("clip.mp3") == "audio/mp3" + assert IBMWatsonProvider._content_type("clip.wav") == "audio/wav" + assert ( + IBMWatsonProvider._content_type("clip.unknown") + == "application/octet-stream" + ) + + @patch.dict( + os.environ, + { + "IBM_WATSON_STT_API_KEY": "test-key", + "IBM_WATSON_STT_URL": "https://ibm.example.com/instance", + }, + clear=True, + ) + def test_model_aliases(self): + from sapat.providers.ibm_watson import IBMWatsonProvider + + provider = IBMWatsonProvider() + assert provider.resolve_model("en") == "en-US_BroadbandModel" + assert provider.resolve_model("en-us") == "en-US_BroadbandModel" + assert provider.resolve_model("custom-model") == "custom-model"