From 514f468ac667ff94f7b8f6646d553d9928756bc2 Mon Sep 17 00:00:00 2001 From: attaboy11 Date: Sat, 30 May 2026 06:52:42 +0100 Subject: [PATCH 1/3] Add Rev AI transcription provider Signed-off-by: attaboy11 --- sapat/providers/revai.py | 93 +++++++++++++++++++++++ tests/providers/test_revai.py | 138 ++++++++++++++++++++++++++++++++++ 2 files changed, 231 insertions(+) create mode 100644 sapat/providers/revai.py create mode 100644 tests/providers/test_revai.py diff --git a/sapat/providers/revai.py b/sapat/providers/revai.py new file mode 100644 index 0000000..8489628 --- /dev/null +++ b/sapat/providers/revai.py @@ -0,0 +1,93 @@ +# ABOUTME: Rev AI asynchronous transcription provider +# ABOUTME: Submits local audio, polls job status, and fetches plain-text transcripts + +import os + +import requests + +from sapat.providers import register +from sapat.providers.async_poll import AsyncPollProvider +from sapat.providers.base import ( + AudioFormat, + ProviderConfig, + TranscriptionResult, +) + + +@register +class RevAIProvider(AsyncPollProvider): + """Rev AI asynchronous Speech-to-Text API provider.""" + + name = "revai" + config = ProviderConfig( + required_env_vars=["REVAI_ACCESS_TOKEN"], + max_file_size_mb=2048.0, + preferred_format=AudioFormat.MP3, + supports_correction=False, + default_model="async", + ) + + def __init__(self): + super().__init__() + self.access_token = os.getenv("REVAI_ACCESS_TOKEN", "") + self.base_url = os.getenv( + "REVAI_API_BASE_URL", + "https://api.rev.ai/speechtotext/v1", + ).rstrip("/") + self.poll_interval = float(os.getenv("REVAI_JOB_POLL_INTERVAL_SECONDS", "5")) + self.max_poll_time = float(os.getenv("REVAI_JOB_TIMEOUT_SECONDS", "3600")) + + def _headers(self, accept: str = "application/json") -> dict: + return { + "Authorization": f"Bearer {self.access_token}", + "Accept": accept, + } + + def _upload(self, audio_file: str, model: str, language: str, **kwargs) -> str: + data = {} + if language and language.lower() != "auto": + data["language"] = language + + with open(audio_file, "rb") as f: + response = requests.post( + f"{self.base_url}/jobs", + headers=self._headers(), + data=data, + files={"media": (os.path.basename(audio_file), f)}, + timeout=120, + ) + + if response.status_code not in (200, 201): + raise RuntimeError(f"Rev AI job creation failed: {response.text}") + + job_id = response.json().get("id") + if not job_id: + raise RuntimeError("Rev AI job creation response did not include an id.") + return job_id + + def _poll(self, job_id: str) -> str: + response = requests.get( + f"{self.base_url}/jobs/{job_id}", + headers=self._headers(), + timeout=30, + ) + if response.status_code != 200: + raise RuntimeError(f"Rev AI status check failed: {response.text}") + + status = response.json().get("status") + if status == "transcribed": + return "completed" + if status == "failed": + return "failed" + return "pending" + + def _fetch_result(self, job_id: str) -> TranscriptionResult: + response = requests.get( + f"{self.base_url}/jobs/{job_id}/transcript", + headers=self._headers("text/plain"), + timeout=60, + ) + if response.status_code != 200: + raise RuntimeError(f"Rev AI transcript retrieval failed: {response.text}") + + return TranscriptionResult(text=response.text, raw_response=response.text) diff --git a/tests/providers/test_revai.py b/tests/providers/test_revai.py new file mode 100644 index 0000000..8cffb39 --- /dev/null +++ b/tests/providers/test_revai.py @@ -0,0 +1,138 @@ +# ABOUTME: Tests for the Rev AI asynchronous transcription provider +# ABOUTME: Verifies upload, polling, transcript fetch, and availability behavior + +import os +from unittest.mock import patch + +import pytest + +from sapat.providers.base import TranscriptionResult + + +class FakeResponse: + """Minimal requests.Response stand-in for mocked HTTP calls.""" + + 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.mp3" + path.write_bytes(b"fake audio bytes") + return str(path) + + +class TestRevAIProvider: + @patch.dict( + os.environ, + { + "REVAI_ACCESS_TOKEN": "test-token", + "REVAI_API_BASE_URL": "https://revai.test/speechtotext/v1", + }, + clear=False, + ) + @patch("sapat.providers.revai.requests.get") + @patch("sapat.providers.revai.requests.post") + def test_transcribe_submits_polls_and_fetches_text( + self, mock_post, mock_get, audio_file + ): + mock_post.return_value = FakeResponse( + status_code=201, + payload={"id": "job-123", "status": "in_progress"}, + ) + mock_get.side_effect = [ + FakeResponse(status_code=200, payload={"status": "in_progress"}), + FakeResponse(status_code=200, payload={"status": "transcribed"}), + FakeResponse(status_code=200, text="Transcript from Rev AI."), + ] + + from sapat.providers.revai import RevAIProvider + + provider = RevAIProvider() + provider.poll_interval = 0.01 + result = provider.transcribe(audio_file, model="async", language="en") + + assert isinstance(result, TranscriptionResult) + assert result.text == "Transcript from Rev AI." + + mock_post.assert_called_once() + post_url = mock_post.call_args.args[0] + post_kwargs = mock_post.call_args.kwargs + assert post_url == "https://revai.test/speechtotext/v1/jobs" + assert post_kwargs["headers"]["Authorization"] == "Bearer test-token" + assert post_kwargs["data"]["language"] == "en" + assert "media" in post_kwargs["files"] + + assert mock_get.call_args_list[0].args[0].endswith("/jobs/job-123") + assert ( + mock_get.call_args_list[2].args[0] + == "https://revai.test/speechtotext/v1/jobs/job-123/transcript" + ) + assert mock_get.call_args_list[2].kwargs["headers"]["Accept"] == "text/plain" + + @patch.dict(os.environ, {"REVAI_ACCESS_TOKEN": "test-token"}, clear=False) + def test_default_model(self): + from sapat.providers.revai import RevAIProvider + + assert RevAIProvider.config.default_model == "async" + + @patch.dict(os.environ, {"REVAI_ACCESS_TOKEN": "test-token"}, clear=False) + def test_available_with_access_token(self): + from sapat.providers.revai import RevAIProvider + + assert RevAIProvider.is_available() is True + + @patch.dict(os.environ, {}, clear=True) + def test_not_available_without_access_token(self): + from sapat.providers.revai import RevAIProvider + + assert RevAIProvider.is_available() is False + + @patch.dict( + os.environ, + { + "REVAI_ACCESS_TOKEN": "test-token", + "REVAI_API_BASE_URL": "https://revai.test/speechtotext/v1", + }, + clear=False, + ) + @patch("sapat.providers.revai.requests.post") + def test_upload_requires_job_id(self, mock_post, audio_file): + mock_post.return_value = FakeResponse(status_code=201, payload={}) + + from sapat.providers.revai import RevAIProvider + + provider = RevAIProvider() + with pytest.raises(RuntimeError, match="did not include an id"): + provider.transcribe(audio_file, model="async") + + @patch.dict( + os.environ, + { + "REVAI_ACCESS_TOKEN": "test-token", + "REVAI_API_BASE_URL": "https://revai.test/speechtotext/v1", + }, + clear=False, + ) + @patch("sapat.providers.revai.requests.get") + @patch("sapat.providers.revai.requests.post") + def test_failed_job_raises(self, mock_post, mock_get, audio_file): + mock_post.return_value = FakeResponse( + status_code=201, payload={"id": "job-123"} + ) + mock_get.return_value = FakeResponse( + status_code=200, payload={"status": "failed"} + ) + + from sapat.providers.revai import RevAIProvider + + provider = RevAIProvider() + provider.poll_interval = 0.01 + with pytest.raises(RuntimeError, match="failed"): + provider.transcribe(audio_file, model="async") From c86dbf12a9bb8f769c7b81d05e2b354e964c78b5 Mon Sep 17 00:00:00 2001 From: attaboy11 Date: Sat, 30 May 2026 22:22:05 +0100 Subject: [PATCH 2/3] Document Rev AI provider wiring --- README.md | 25 +++++++++++++++++-------- tests/test_registry.py | 27 +++++++++++++++++++++------ 2 files changed, 38 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index e5b7168..4c72bf8 100644 --- a/README.md +++ b/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: Azure OpenAI, Groq, OpenAI, and Rev AI APIs. 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, and Rev AI APIs - 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 + - Rev AI API ## Installation @@ -53,6 +54,13 @@ 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 + + # Rev AI + REVAI_ACCESS_TOKEN=your_rev_ai_access_token_here + # Optional overrides: + # REVAI_API_BASE_URL=https://api.rev.ai/speechtotext/v1 + # REVAI_JOB_POLL_INTERVAL_SECONDS=5 + # REVAI_JOB_TIMEOUT_SECONDS=3600 ``` ## Building and Installing the Package @@ -111,15 +119,16 @@ sapat [--language ] [--prompt ] [--t - `--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 revai` for Rev AI 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 revai ``` - If a file is provided, it will process that single file. @@ -129,7 +138,7 @@ The script will create a `.txt` file with the same name as the input video file, ## Note -This tool is designed for use with multiple APIs (Azure OpenAI, Groq, and OpenAI). Ensure you have valid API credentials configured in the `.env` file and the necessary permissions and credits for the API service you plan to use. +This tool is designed for use with multiple APIs (Azure OpenAI, Groq, OpenAI, and Rev AI). Ensure you have valid API credentials configured in the `.env` file and the necessary permissions and credits for the API service you plan to use. ## License diff --git a/tests/test_registry.py b/tests/test_registry.py index 3e9be8b..0f506db 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -24,7 +24,9 @@ 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") @@ -32,6 +34,7 @@ def transcribe(self, audio_file, model, language="en", prompt=None, temperature= 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_revai_when_env_set(self): + with patch.dict(os.environ, {"REVAI_ACCESS_TOKEN": "test"}): + from sapat.providers.revai import RevAIProvider + + register(RevAIProvider) + assert "revai" in _registry + def test_no_discovery_without_env(self): with patch.dict(os.environ, {}, clear=True): _discover_providers() From 498da2f7958460f9864ac7c7a8c151c2819d03f5 Mon Sep 17 00:00:00 2001 From: attaboy11 Date: Sat, 30 May 2026 22:57:43 +0100 Subject: [PATCH 3/3] Harden Rev AI provider multipart options --- README.md | 5 +++-- sapat/providers/revai.py | 18 ++++++++++++++---- tests/providers/test_revai.py | 23 ++++++++++++++++++++++- 3 files changed, 39 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 4c72bf8..5b7c10f 100644 --- a/README.md +++ b/README.md @@ -110,15 +110,16 @@ 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 [--language ] [--prompt ] [--temperature ] +sapat [--language ] [--transcription-prompt ] [--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'). +- `--model`: Provider-specific model or mode. For Rev AI, use `fusion`, `low_cost`, `machine`, or `human` to set Rev AI's transcriber option; omit it for the default machine transcription path. - `--provider`: Specify the provider to use for transcription. - `--provider azure` for Azure OpenAI API - `--provider groq` for Groq Cloud API diff --git a/sapat/providers/revai.py b/sapat/providers/revai.py index 8489628..284bedd 100644 --- a/sapat/providers/revai.py +++ b/sapat/providers/revai.py @@ -1,6 +1,8 @@ # ABOUTME: Rev AI asynchronous transcription provider # ABOUTME: Submits local audio, polls job status, and fetches plain-text transcripts +import json +import mimetypes import os import requests @@ -43,17 +45,25 @@ def _headers(self, accept: str = "application/json") -> dict: "Accept": accept, } - def _upload(self, audio_file: str, model: str, language: str, **kwargs) -> str: - data = {} + def _job_options(self, model: str, language: str) -> dict: + options = {} if language and language.lower() != "auto": - data["language"] = language + options["language"] = language + if model and model != self.config.default_model: + options["transcriber"] = model + return options + + def _upload(self, audio_file: str, model: str, language: str, **kwargs) -> str: + options = self._job_options(model, language) + data = {"options": json.dumps(options)} if options else {} + media_type = mimetypes.guess_type(audio_file)[0] or "application/octet-stream" with open(audio_file, "rb") as f: response = requests.post( f"{self.base_url}/jobs", headers=self._headers(), data=data, - files={"media": (os.path.basename(audio_file), f)}, + files={"media": (os.path.basename(audio_file), f, media_type)}, timeout=120, ) diff --git a/tests/providers/test_revai.py b/tests/providers/test_revai.py index 8cffb39..a388414 100644 --- a/tests/providers/test_revai.py +++ b/tests/providers/test_revai.py @@ -1,6 +1,7 @@ # ABOUTME: Tests for the Rev AI asynchronous transcription provider # ABOUTME: Verifies upload, polling, transcript fetch, and availability behavior +import json import os from unittest.mock import patch @@ -66,8 +67,9 @@ def test_transcribe_submits_polls_and_fetches_text( post_kwargs = mock_post.call_args.kwargs assert post_url == "https://revai.test/speechtotext/v1/jobs" assert post_kwargs["headers"]["Authorization"] == "Bearer test-token" - assert post_kwargs["data"]["language"] == "en" + assert json.loads(post_kwargs["data"]["options"]) == {"language": "en"} assert "media" in post_kwargs["files"] + assert post_kwargs["files"]["media"][2] == "audio/mpeg" assert mock_get.call_args_list[0].args[0].endswith("/jobs/job-123") assert ( @@ -82,6 +84,25 @@ def test_default_model(self): assert RevAIProvider.config.default_model == "async" + @patch.dict(os.environ, {"REVAI_ACCESS_TOKEN": "test-token"}, clear=False) + def test_job_options_omit_auto_language_and_default_model(self): + from sapat.providers.revai import RevAIProvider + + provider = RevAIProvider() + + assert provider._job_options("async", "auto") == {} + + @patch.dict(os.environ, {"REVAI_ACCESS_TOKEN": "test-token"}, clear=False) + def test_job_options_map_model_to_transcriber(self): + from sapat.providers.revai import RevAIProvider + + provider = RevAIProvider() + + assert provider._job_options("fusion", "en-gb") == { + "language": "en-gb", + "transcriber": "fusion", + } + @patch.dict(os.environ, {"REVAI_ACCESS_TOKEN": "test-token"}, clear=False) def test_available_with_access_token(self): from sapat.providers.revai import RevAIProvider