From 1af9ea5e6eb06aa842dae3a953cd95816f949798 Mon Sep 17 00:00:00 2001 From: Yutsuo Date: Sun, 30 Aug 2026 21:17:23 -0300 Subject: [PATCH] feat: transcribe chunks with retries and resume --- .../services/transcription.py | 163 +++++++++++++++++- tests/test_transcription.py | 100 ++++++++--- 2 files changed, 233 insertions(+), 30 deletions(-) diff --git a/src/voice_transcriptor/services/transcription.py b/src/voice_transcriptor/services/transcription.py index 09fe9af..544cf1a 100644 --- 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)) diff --git a/tests/test_transcription.py b/tests/test_transcription.py index c3e907f..a294abc 100644 --- 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"