Add Wit.ai transcription provider

README.md

@@ -1,11 +1,11 @@
 # Video Transcription Tool
 
-This tool automates the process of transcribing video files using multiple transcription services: Azure OpenAI, Groq, and OpenAI APIs. It converts video files to MP3 format, transcribes the audio, and saves the transcription as a text file.
+This tool automates the process of transcribing video files using multiple transcription services, including Azure OpenAI, Groq, OpenAI, and additional provider plugins. It converts video files to MP3 format, transcribes the audio, and saves the transcription as a text file.
 
 ## Features
 
 - Converts video files to MP3 format using ffmpeg
-- Supports transcription using Azure OpenAI, Groq, and OpenAI APIs
+- Supports transcription using Azure OpenAI, Groq, OpenAI, Wit.ai, and other provider plugins
 - Supports processing of individual video files or entire directories
 - Cleans up temporary MP3 files after transcription
 - Provides flexibility in selecting the transcription service via configuration
@@ -18,6 +18,7 @@ This tool automates the process of transcribing video files using multiple trans
   - Azure OpenAI API
   - Groq Cloud API
   - OpenAI API
+  - Wit.ai API
 
 ## Installation
 
@@ -53,6 +54,11 @@ This tool automates the process of transcribing video files using multiple trans
    OPENAI_MODEL=whisper-1
    OPENAI_API_ENDPOINT=https://api.openai.com/v1/audio/transcriptions
    OPENAI_MODEL_NAME_CHAT=gpt-4o
+
+   # Wit.ai
+   WITAI_ACCESS_TOKEN=your_witai_server_access_token_here
+   WITAI_API_VERSION=20240304
+   WITAI_CONTENT_TYPE=audio/mpeg
    ```
 
 ## Building and Installing the Package
@@ -102,24 +108,27 @@ This tool automates the process of transcribing video files using multiple trans
 Run the script with a video file or directory as an argument:
 
 ```
-sapat <video_file_or_directory> [--language <language>] [--prompt <prompt>] [--temperature <temperature>]
+sapat <video_file_or_directory> [--language <language>] [--transcription-prompt <prompt>] [--temperature <temperature>]
 ```
 
 ### Options
 
 - `--language`: Specify the language of the audio (default: "en").
-- `--prompt`: Optional prompt to guide the model's transcription.
+- `--transcription-prompt`: Optional prompt to guide the model's transcription.
 - `--temperature`: The sampling temperature, between 0 and 1 (default: 0).
 - `--quality`: Quality of the MP3 audio: 'L' for low, 'M' for medium, and 'H' for high (default: 'M').
-- `--api`: Specify the API to use for transcription.
-   - `--api azure` for Azure OpenAI API
-   - `--api groq` for Groq Cloud API
-   - `--api openai` for OpenAI API
+- `--provider`: Specify the provider to use for transcription.
+   - `--provider azure` for Azure OpenAI API
+   - `--provider groq` for Groq Cloud API
+   - `--provider openai` for OpenAI API
+   - `--provider witai` for Wit.ai Speech API
 
 Example:
 
 ```
-sapat my_video.mp4 --quality H --language es --prompt "This is a test prompt" --temperature 0.5 --api groq
+sapat my_video.mp4 --quality H --language es --transcription-prompt "This is a test prompt" --temperature 0.5 --provider groq
+
+sapat my_video.mp4 --quality M --provider witai
 ```
 
 - If a file is provided, it will process that single file.

sapat/providers/witai.py

@@ -0,0 +1,136 @@
+# ABOUTME: Wit.ai transcription provider
+# ABOUTME: Uses Wit.ai's /speech HTTP API for binary audio transcription
+
+import json
+import os
+from pathlib import Path
+from typing import Any, Optional
+
+import requests
+
+from sapat.providers import register
+from sapat.providers.base import (
+    AudioFormat,
+    ProviderConfig,
+    TranscriptionProvider,
+    TranscriptionResult,
+)
+
+
+@register
+class WitAIProvider(TranscriptionProvider):
+    name = "witai"
+    config = ProviderConfig(
+        required_env_vars=["WITAI_ACCESS_TOKEN"],
+        max_file_size_mb=25.0,
+        preferred_format=AudioFormat.MP3,
+        supports_correction=False,
+        default_model="speech",
+    )
+
+    def transcribe(
+        self,
+        audio_file: str,
+        model: str,
+        language: str = "en",
+        prompt: Optional[str] = None,
+        temperature: float = 0,
+        **kwargs,
+    ) -> TranscriptionResult:
+        token = os.getenv("WITAI_ACCESS_TOKEN", "")
+        endpoint = os.getenv("WITAI_API_ENDPOINT", "https://api.wit.ai/speech")
+        api_version = os.getenv("WITAI_API_VERSION", "20240304")
+        content_type = os.getenv(
+            "WITAI_CONTENT_TYPE",
+            self._content_type(audio_file),
+        )
+
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": content_type,
+        }
+
+        with open(audio_file, "rb") as f:
+            audio_data = f.read()
+
+        try:
+            response = requests.post(
+                endpoint,
+                params={"v": api_version},
+                headers=headers,
+                data=audio_data,
+                timeout=90,
+            )
+        except requests.RequestException as exc:
+            raise RuntimeError(f"Wit.ai transcription request failed: {exc}") from exc
+
+        if response.status_code != 200:
+            raise RuntimeError(
+                f"Wit.ai transcription failed ({response.status_code}): "
+                f"{response.text}"
+            )
+
+        payload = self._decode_response(response)
+        text = self._extract_text(payload)
+        return TranscriptionResult(text=text, raw_response=payload)
+
+    def resolve_model(self, model_alias: str) -> str:
+        # Wit.ai selects the speech model from the app configured for the token.
+        return model_alias
+
+    @staticmethod
+    def _content_type(audio_file: str) -> str:
+        extension = Path(audio_file).suffix.lower()
+        content_types = {
+            ".flac": "audio/flac",
+            ".mp3": "audio/mpeg",
+            ".ogg": "audio/ogg",
+            ".wav": "audio/wav",
+            ".webm": "audio/webm",
+        }
+        return content_types.get(extension, "application/octet-stream")
+
+    @classmethod
+    def _decode_response(cls, response) -> Any:
+        try:
+            return response.json()
+        except ValueError:
+            return cls._decode_json_stream(response.text)
+
+    @staticmethod
+    def _decode_json_stream(body: str) -> Any:
+        text = body.strip()
+        if not text:
+            raise RuntimeError("Wit.ai response was empty.")
+
+        decoder = json.JSONDecoder()
+        values = []
+        index = 0
+        while index < len(text):
+            while index < len(text) and text[index].isspace():
+                index += 1
+            if index >= len(text):
+                break
+            try:
+                value, index = decoder.raw_decode(text, index)
+            except json.JSONDecodeError as exc:
+                raise RuntimeError(
+                    f"Wit.ai response was not valid JSON: {exc}"
+                ) from exc
+            values.append(value)
+
+        if not values:
+            raise RuntimeError("Wit.ai response did not include a JSON payload.")
+        return values[-1]
+
+    @staticmethod
+    def _extract_text(payload: Any) -> str:
+        if isinstance(payload, dict):
+            for key in ("text", "_text", "speech", "message"):
+                value = payload.get(key)
+                if isinstance(value, str) and value.strip():
+                    return value.strip()
+        elif isinstance(payload, str) and payload.strip():
+            return payload.strip()
+
+        raise RuntimeError("Wit.ai response did not include transcript text.")

tests/providers/test_witai.py

@@ -0,0 +1,128 @@
+# ABOUTME: Tests for the Wit.ai transcription provider
+# ABOUTME: Verifies auth, binary upload, response parsing, and registry behavior
+
+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
+        self.text = text
+
+    def json(self):
+        if isinstance(self.payload, Exception):
+            raise self.payload
+        return self.payload
+
+
+@pytest.fixture
+def audio_file(tmp_path):
+    path = tmp_path / "sample.mp3"
+    path.write_bytes(b"fake audio data")
+    return str(path)
+
+
+class TestWitAIProvider:
+    @patch.dict(os.environ, {"WITAI_ACCESS_TOKEN": "test-token"}, clear=True)
+    def test_available_with_access_token(self):
+        from sapat.providers.witai import WitAIProvider
+
+        assert WitAIProvider.is_available() is True
+
+    @patch.dict(os.environ, {}, clear=True)
+    def test_not_available_without_access_token(self):
+        from sapat.providers.witai import WitAIProvider
+
+        assert WitAIProvider.is_available() is False
+
+    @patch.dict(os.environ, {"WITAI_ACCESS_TOKEN": "test-token"}, clear=True)
+    @patch("sapat.providers.witai.requests.post")
+    def test_transcribe_posts_binary_audio(self, mock_post, audio_file):
+        mock_post.return_value = FakeResponse(payload={"text": "hello from wit"})
+
+        from sapat.providers.witai import WitAIProvider
+
+        provider = WitAIProvider()
+        result = provider.transcribe(audio_file, model="speech")
+
+        assert isinstance(result, TranscriptionResult)
+        assert result.text == "hello from wit"
+
+        mock_post.assert_called_once()
+        url = mock_post.call_args.args[0]
+        kwargs = mock_post.call_args.kwargs
+
+        assert url == "https://api.wit.ai/speech"
+        assert kwargs["params"] == {"v": "20240304"}
+        assert kwargs["headers"]["Authorization"] == "Bearer test-token"
+        assert kwargs["headers"]["Content-Type"] == "audio/mpeg"
+        assert kwargs["data"] == b"fake audio data"
+
+    @patch.dict(
+        os.environ,
+        {
+            "WITAI_ACCESS_TOKEN": "test-token",
+            "WITAI_API_ENDPOINT": "https://example.test/speech",
+            "WITAI_API_VERSION": "20250101",
+            "WITAI_CONTENT_TYPE": "audio/wav",
+        },
+        clear=True,
+    )
+    @patch("sapat.providers.witai.requests.post")
+    def test_respects_endpoint_version_and_content_type(self, mock_post, audio_file):
+        mock_post.return_value = FakeResponse(payload={"text": "configured"})
+
+        from sapat.providers.witai import WitAIProvider
+
+        result = WitAIProvider().transcribe(audio_file, model="speech")
+
+        assert result.text == "configured"
+        assert mock_post.call_args.args[0] == "https://example.test/speech"
+        assert mock_post.call_args.kwargs["params"] == {"v": "20250101"}
+        assert mock_post.call_args.kwargs["headers"]["Content-Type"] == "audio/wav"
+
+    @patch.dict(os.environ, {"WITAI_ACCESS_TOKEN": "test-token"}, clear=True)
+    @patch("sapat.providers.witai.requests.post")
+    def test_parses_streaming_json_response(self, mock_post, audio_file):
+        mock_post.return_value = FakeResponse(
+            payload=ValueError("not single json"),
+            text='{"text": "partial"}\n{"text": "final transcript"}',
+        )
+
+        from sapat.providers.witai import WitAIProvider
+
+        result = WitAIProvider().transcribe(audio_file, model="speech")
+
+        assert result.text == "final transcript"
+        assert result.raw_response == {"text": "final transcript"}
+
+    @patch.dict(os.environ, {"WITAI_ACCESS_TOKEN": "test-token"}, clear=True)
+    @patch("sapat.providers.witai.requests.post")
+    def test_transcribe_api_error_raises(self, mock_post, audio_file):
+        mock_post.return_value = FakeResponse(status_code=401, text="invalid token")
+
+        from sapat.providers.witai import WitAIProvider
+
+        with pytest.raises(RuntimeError, match="Wit.ai transcription failed"):
+            WitAIProvider().transcribe(audio_file, model="speech")
+
+    def test_decode_json_stream_handles_concatenated_objects(self):
+        from sapat.providers.witai import WitAIProvider
+
+        payload = WitAIProvider._decode_json_stream(
+            '{"text": "first"}{"_text": "second"}'
+        )
+
+        assert payload == {"_text": "second"}
+
+    def test_extract_text_rejects_empty_payload(self):
+        from sapat.providers.witai import WitAIProvider
+
+        with pytest.raises(RuntimeError, match="transcript text"):
+            WitAIProvider._extract_text({"entities": {}})

tests/test_registry.py

@@ -24,14 +24,17 @@ class FakeProvider(TranscriptionProvider):
     name = "fake_test"
     config = ProviderConfig(required_env_vars=[])
 
-    def transcribe(self, audio_file, model, language="en", prompt=None, temperature=0, **kwargs):
+    def transcribe(
+        self, audio_file, model, language="en", prompt=None, temperature=0, **kwargs
+    ):
         return TranscriptionResult(text="fake")
 
 
 @pytest.fixture(autouse=True)
 def reset_registry():
     """Reset the registry before each test."""
     import sapat.providers as reg
+
     reg._registry.clear()
     reg._discovered = False
     yield
@@ -92,21 +95,33 @@ def transcribe(self, **kw):
 
 class TestAutoDiscovery:
     def test_discovers_azure_when_env_set(self):
-        with patch.dict(os.environ, {
-            "AZURE_OPENAI_API_KEY": "test",
-            "AZURE_OPENAI_ENDPOINT": "https://test.openai.azure.com",
-            "AZURE_OPENAI_STT_API_VERSION": "2024-02-01",
-        }):
+        with patch.dict(
+            os.environ,
+            {
+                "AZURE_OPENAI_API_KEY": "test",
+                "AZURE_OPENAI_ENDPOINT": "https://test.openai.azure.com",
+                "AZURE_OPENAI_STT_API_VERSION": "2024-02-01",
+            },
+        ):
             from sapat.providers.azure import AzureProvider
+
             register(AzureProvider)
             assert "azure" in _registry
 
     def test_discovers_groq_when_env_set(self):
         with patch.dict(os.environ, {"GROQ_API_KEY": "test"}):
             from sapat.providers.groq import GroqProvider
+
             register(GroqProvider)
             assert "groq" in _registry
 
+    def test_discovers_witai_when_env_set(self):
+        with patch.dict(os.environ, {"WITAI_ACCESS_TOKEN": "test"}):
+            from sapat.providers.witai import WitAIProvider
+
+            register(WitAIProvider)
+            assert "witai" in _registry
+
     def test_no_discovery_without_env(self):
         with patch.dict(os.environ, {}, clear=True):
             _discover_providers()