Add IBM Watson transcription provider

.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:

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)

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"