feat: transcribe chunks with retries and resume

4 days ago · 1af9ea5e6e
2 changed files with 233 additions and 30 deletions
--- a/src/voice_transcriptor/services/transcription.py
+++ b/src/voice_transcriptor/services/transcription.py
@ -1,13 +1,160 @@
-from voice_transcriptor.models import AppSettings, MediaInfo
+from __future__ import annotations

+import random
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable

-class TranscriptionNotImplementedError(RuntimeError):
-    """Raised because transcription is outside the milestone 1 scope."""
+import openai
+
+from voice_transcriptor.models import AppSettings
+from voice_transcriptor.services.job_manifest import ChunkStatus, JobManifestRepository
+from voice_transcriptor.services.preprocessing import CancellationToken, PreprocessingCancelled
+
+
+BRAZILIAN_PROMPT = (
+    "Conversa em português brasileiro. Preserve a língua falada; não traduza. "
+    "Preserve ortografia e pontuação brasileiras, números, nomes próprios, "
+    "termos técnicos e siglas com máxima fidelidade."
+)
+
+
+class TranscriptionError(RuntimeError): pass
+class PermanentTranscriptionError(TranscriptionError): pass
+class RetryExhaustedError(TranscriptionError): pass
+
+
+@dataclass(frozen=True, slots=True)
+class RetryPolicy:
+    max_attempts: int = 5
+    initial_delay_seconds: float = 1.0
+    max_delay_seconds: float = 30.0
+    jitter_ratio: float = 0.2
+
+
+@dataclass(frozen=True, slots=True)
+class TranscriptionProgress:
+    completed: int
+    total: int
+    current_chunk: int | None
+    elapsed_seconds: float
+    phase: str
+    message: str
+    api_error: str | None = None
+
+
+def normalize_language(value: str) -> str:
+    normalized = value.strip().replace("_", "-")
+    return normalized.split("-", 1)[0].lower()
+
+
+def build_prompt(context: str) -> str:
+    extra = context.strip()
+    return f"{BRAZILIAN_PROMPT}\n\n{extra}" if extra else BRAZILIAN_PROMPT
+
+
+class OpenAITranscriptionClient:
+    def __init__(self, client) -> None: self.client = client
+
+    def transcribe(self, path: Path, model: str, language: str, prompt: str) -> str:
+        with path.open("rb") as audio_file:
+            response = self.client.audio.transcriptions.create(
+                file=audio_file,
+                model=model,
+                language=normalize_language(language),
+                prompt=prompt,
+                response_format="json",
+            )
+        text = getattr(response, "text", None)
+        if not isinstance(text, str):
+            raise PermanentTranscriptionError("The transcription API returned no text.")
+        return text


 class TranscriptionService:
-    def transcribe(self, media: MediaInfo, settings: AppSettings) -> None:
-        del media, settings
-        raise TranscriptionNotImplementedError(
-            "Transcription is not implemented in milestone 1."
-        )
+    def __init__(
+        self,
+        client: OpenAITranscriptionClient,
+        manifests: JobManifestRepository,
+        retry_policy: RetryPolicy | None = None,
+        sleep: Callable[[float], None] = time.sleep,
+        monotonic: Callable[[], float] = time.monotonic,
+        random_value: Callable[[], float] = random.random,
+    ) -> None:
+        self.client = client; self.manifests = manifests
+        self.retry_policy = retry_policy or RetryPolicy()
+        self.sleep = sleep; self.monotonic = monotonic; self.random_value = random_value
+
+    def run(
+        self,
+        manifest_path: Path,
+        settings: AppSettings,
+        token: CancellationToken | None = None,
+        progress: Callable[[TranscriptionProgress], None] | None = None,
+    ) -> Path:
+        token = token or CancellationToken(); started = self.monotonic()
+        try:
+            token.raise_if_cancelled()
+            manifest = self.manifests.recover_for_resume(manifest_path)
+            total = manifest["total_chunks"]
+            for item in manifest["chunks"]:
+                if item["status"] == ChunkStatus.COMPLETED: continue
+                token.raise_if_cancelled(); index = item["index"]
+                text = self._transcribe_with_retry(manifest_path, item, settings, token, progress, started, total)
+                self.manifests.mark_completed(manifest_path, index, text)
+                self.manifests.assemble_transcript(manifest_path)
+                completed = self.manifests.load(manifest_path)["completed_chunks"]
+                self._emit(progress, completed, total, index + 1, started, "transcribing", f"Completed chunk {index + 1} of {total}")
+            self.manifests.mark_job_state(manifest_path, "completed")
+            return self.manifests.assemble_transcript(manifest_path)
+        except PreprocessingCancelled:
+            self.manifests.mark_job_state(manifest_path, "cancelled")
+            raise
+
+    def _transcribe_with_retry(self, manifest_path: Path, item: dict, settings: AppSettings, token: CancellationToken, progress, started: float, total: int) -> str:
+        policy = self.retry_policy; index = item["index"]
+        for attempt in range(1, policy.max_attempts + 1):
+            token.raise_if_cancelled(); self.manifests.mark_processing(manifest_path, index)
+            self._emit(progress, self.manifests.load(manifest_path)["completed_chunks"], total, index + 1, started, "transcribing", f"Transcribing chunk {index + 1} of {total}")
+            try:
+                return self.client.transcribe(manifest_path.parent / item["path"], settings.model, settings.language, build_prompt(settings.context_vocabulary))
+            except PreprocessingCancelled: raise
+            except Exception as exc:
+                status = getattr(exc, "status_code", None)
+                retryable = self._retryable(exc, status)
+                safe = self._safe_error(status, retryable)
+                if not retryable:
+                    self.manifests.mark_failed(manifest_path, index, safe)
+                    raise PermanentTranscriptionError(safe) from None
+                if attempt >= policy.max_attempts:
+                    self.manifests.mark_failed(manifest_path, index, safe)
+                    raise RetryExhaustedError(f"{safe} Retry limit reached.") from None
+                delay = self._delay(exc, attempt)
+                self._emit(progress, self.manifests.load(manifest_path)["completed_chunks"], total, index + 1, started, "retrying", f"API temporarily unavailable; retrying in {delay:g} seconds.", safe)
+                token.raise_if_cancelled(); self.sleep(delay); token.raise_if_cancelled()
+        raise AssertionError("unreachable")
+
+    def _delay(self, exc: Exception, attempt: int) -> float:
+        headers = getattr(getattr(exc, "response", None), "headers", {}) or {}
+        retry_after = headers.get("retry-after") or headers.get("Retry-After")
+        try: server_delay = float(retry_after)
+        except (TypeError, ValueError): server_delay = 0
+        base = max(server_delay, self.retry_policy.initial_delay_seconds * (2 ** (attempt - 1)))
+        base = min(base, self.retry_policy.max_delay_seconds)
+        jitter = base * self.retry_policy.jitter_ratio * ((self.random_value() * 2) - 1)
+        return max(0, min(self.retry_policy.max_delay_seconds, base + jitter))
+
+    @staticmethod
+    def _retryable(exc: Exception, status: int | None) -> bool:
+        transient_types = (openai.RateLimitError, openai.APIConnectionError, openai.APITimeoutError)
+        return isinstance(exc, transient_types) or status in (408, 409, 429) or (isinstance(status, int) and status >= 500)
+
+    @staticmethod
+    def _safe_error(status: int | None, retryable: bool) -> str:
+        kind = "transient" if retryable else "permanent"
+        suffix = f" (HTTP {status})" if isinstance(status, int) else ""
+        return f"OpenAI API {kind} error{suffix}."
+
+    def _emit(self, callback, completed: int, total: int, current: int | None, started: float, phase: str, message: str, api_error: str | None = None) -> None:
+        if callback: callback(TranscriptionProgress(completed, total, current, max(0, self.monotonic() - started), phase, message, api_error))
--- a/tests/test_transcription.py
+++ b/tests/test_transcription.py
@ -1,26 +1,82 @@
 from pathlib import Path
+from types import SimpleNamespace

 import pytest

-from voice_transcriptor.models import AppSettings, MediaInfo
-from voice_transcriptor.services.transcription import (
-    TranscriptionNotImplementedError,
-    TranscriptionService,
-)
-
-
-def test_transcribe_raises_milestone_exception_with_user_readable_message() -> None:
-    media = MediaInfo(
-        path=Path("sample.mp3"),
-        size_bytes=1,
-        duration_seconds=1.0,
-        audio_codec="mp3",
-    )
-    settings = AppSettings(
-        model="gpt-4o-mini-transcribe",
-        language="en",
-        output_directory=Path("output"),
-    )
-
-    with pytest.raises(TranscriptionNotImplementedError, match="(?i)not implemented"):
-        TranscriptionService().transcribe(media, settings)
+from voice_transcriptor.models import AppSettings
+from voice_transcriptor.services.job_manifest import JobManifestRepository
+from voice_transcriptor.services.preprocessing import CancellationToken, PreprocessingCancelled
+from voice_transcriptor.services.transcription import OpenAITranscriptionClient, PermanentTranscriptionError, RetryPolicy, TranscriptionService, build_prompt, normalize_language
+
+
+class Endpoint:
+    def __init__(self, responses): self.responses = list(responses); self.calls = []
+    def create(self, **kwargs):
+        self.calls.append(kwargs); result = self.responses.pop(0)
+        if isinstance(result, Exception): raise result
+        return SimpleNamespace(text=result)
+
+
+class StatusFailure(Exception):
+    def __init__(self, status_code: int):
+        super().__init__(f"sensitive sk-leaked-key status {status_code}")
+        self.status_code = status_code; self.response = SimpleNamespace(headers={})
+
+
+def make_job(tmp_path: Path) -> tuple[JobManifestRepository, Path]:
+    job = tmp_path / "job"; (job / "chunks").mkdir(parents=True)
+    for index in range(2): (job / "chunks" / f"chunk-{index:05d}.m4a").write_bytes(b"audio")
+    repository = JobManifestRepository()
+    created = repository.create(job, {"path": str(tmp_path / "source.mp3")}, {"model": "gpt-transcribe"}, [
+        {"index": 0, "path": "chunks/chunk-00000.m4a", "source_start_seconds": "0", "source_end_seconds": "10", "duration_seconds": "10"},
+        {"index": 1, "path": "chunks/chunk-00001.m4a", "source_start_seconds": "9", "source_end_seconds": "20", "duration_seconds": "11"},
+    ])
+    return repository, Path(created["manifest_path"])
+
+
+def test_brazilian_language_and_prompt_preserve_spoken_language() -> None:
+    prompt = build_prompt("  AWS, PostgreSQL, Brasília  ")
+    assert normalize_language("pt-BR") == "pt"; assert normalize_language("pt_BR") == "pt"; assert normalize_language("en-US") == "en"
+    assert "não traduza" in prompt.lower(); assert "números" in prompt.lower(); assert prompt.endswith("AWS, PostgreSQL, Brasília")
+
+
+def test_adapter_calls_verified_audio_transcriptions_interface(tmp_path: Path) -> None:
+    audio = tmp_path / "chunk.m4a"; audio.write_bytes(b"audio")
+    endpoint = Endpoint(["Olá, Brasília."]); client = SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))
+    text = OpenAITranscriptionClient(client).transcribe(audio, "future-compatible-model", "pt-BR", "vocabulário")
+    assert text == "Olá, Brasília."
+    call = endpoint.calls[0]
+    assert call["model"] == "future-compatible-model"; assert call["language"] == "pt"; assert call["prompt"] == "vocabulário"; assert call["response_format"] == "json"; assert call["file"].closed is True
+
+
+def test_run_saves_each_chunk_and_resume_skips_completed(tmp_path: Path) -> None:
+    repository, manifest_path = make_job(tmp_path); first_endpoint = Endpoint(["Primeiro", "Segundo"])
+    service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=first_endpoint))), repository)
+    settings = AppSettings("gpt-transcribe", "pt-BR", tmp_path, context_vocabulary="Pix")
+    transcript = service.run(manifest_path, settings)
+    assert transcript.read_text(encoding="utf-8") == "Primeiro\nSegundo\n"
+    manifest = repository.load(manifest_path)
+    assert manifest["state"] == "completed"; assert [item["status"] for item in manifest["chunks"]] == ["completed", "completed"]; assert manifest["chunks"][1]["source_start_seconds"] == "9"
+    resume_endpoint = Endpoint([])
+    TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=resume_endpoint))), repository).run(manifest_path, settings)
+    assert resume_endpoint.calls == []
+
+
+def test_transient_failure_retries_with_exponential_backoff(tmp_path: Path) -> None:
+    repository, manifest_path = make_job(tmp_path); endpoint = Endpoint([StatusFailure(429), StatusFailure(503), "Primeiro", "Segundo"]); delays = []
+    service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository, retry_policy=RetryPolicy(max_attempts=3, initial_delay_seconds=1, max_delay_seconds=10, jitter_ratio=0), sleep=delays.append)
+    service.run(manifest_path, AppSettings("gpt-transcribe", "pt-BR", tmp_path))
+    assert delays == [1, 2]; assert len(endpoint.calls) == 4
+
+
+def test_permanent_api_failure_is_not_retried_and_is_sanitized(tmp_path: Path) -> None:
+    repository, manifest_path = make_job(tmp_path); endpoint = Endpoint([StatusFailure(400)])
+    service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository)
+    with pytest.raises(PermanentTranscriptionError) as caught: service.run(manifest_path, AppSettings("bad-model", "pt-BR", tmp_path))
+    assert len(endpoint.calls) == 1; assert "sk-leaked-key" not in str(caught.value); assert "sk-leaked-key" not in manifest_path.read_text(encoding="utf-8")
+
+
+def test_cancelled_job_makes_no_api_request(tmp_path: Path) -> None:
+    repository, manifest_path = make_job(tmp_path); endpoint = Endpoint(["unexpected"]); token = CancellationToken(); token.cancel()
+    with pytest.raises(PreprocessingCancelled): TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository).run(manifest_path, AppSettings("gpt-transcribe", "pt-BR", tmp_path), token=token)
+    assert endpoint.calls == []; assert repository.load(manifest_path)["state"] == "cancelled"