| 1 | """Caption-free transcription: compress -> chunk -> provider-fallback. |
| 2 | |
| 3 | When a video/audio item has no captions, this turns the media into text via a |
| 4 | Whisper API. Source-agnostic: the input is a media URL or local path, the output |
| 5 | is transcript text. The pipeline: |
| 6 | |
| 7 | 1. Acquire audio (yt-dlp for a URL, or use a local file as-is). |
| 8 | 2. Re-encode to a low-bitrate mono stream so most clips fit under the provider |
| 9 | upload limit. |
| 10 | 3. If still over the limit, split into bounded-duration chunks. |
| 11 | 4. Transcribe each chunk through an ordered provider list (Groq free tier |
| 12 | first, OpenAI paid backstop), with per-chunk fallback, then join. |
| 13 | |
| 14 | Never raises. Returns a typed :class:`TranscriptResult`; missing prerequisites |
| 15 | (no ffmpeg, no provider key) yield a degraded result with a reason, consumed by |
| 16 | the source-health layer, rather than a crash. |
| 17 | """ |
| 18 | |
| 19 | from __future__ import annotations |
| 20 | |
| 21 | import os |
| 22 | import shutil |
| 23 | import tempfile |
| 24 | from dataclasses import dataclass, field |
| 25 | from typing import Optional |
| 26 | |
| 27 | from . import env, health, subproc |
| 28 | |
| 29 | # Whisper's documented upload ceiling. We compress to stay under it and chunk |
| 30 | # when a single clip still exceeds it. |
| 31 | MAX_UPLOAD_BYTES = 25 * 1024 * 1024 |
| 32 | CHUNK_SECONDS = 600 # 10-minute chunks when splitting is required |
| 33 | |
| 34 | _PROVIDER_ENDPOINTS = { |
| 35 | "groq": "https://api.groq.com/openai/v1/audio/transcriptions", |
| 36 | "openai": "https://api.openai.com/v1/audio/transcriptions", |
| 37 | } |
| 38 | _PROVIDER_MODELS = { |
| 39 | "groq": "whisper-large-v3", |
| 40 | "openai": "whisper-1", |
| 41 | } |
| 42 | |
| 43 | |
| 44 | @dataclass |
| 45 | class TranscriptResult: |
| 46 | text: str = "" |
| 47 | ok: bool = False |
| 48 | reason: str = "" |
| 49 | provider: str = "" |
| 50 | chunks: int = 0 |
| 51 | health: Optional[health.SourceHealth] = field(default=None) |
| 52 | |
| 53 | |
| 54 | def is_available(config: dict) -> bool: |
| 55 | """True when ffmpeg is present AND at least one Whisper provider key is set.""" |
| 56 | return bool(shutil.which("ffmpeg")) and bool(env.transcription_providers(config)) |
| 57 | |
| 58 | |
| 59 | def transcribe_media( |
| 60 | source: str, |
| 61 | config: dict, |
| 62 | timeout: float = 120.0, |
| 63 | ) -> TranscriptResult: |
| 64 | """Transcribe a media URL or local path. Never raises. |
| 65 | |
| 66 | Returns a degraded result (ok=False, reason set) when prerequisites are |
| 67 | missing or every provider fails, so callers can report the gap honestly. |
| 68 | """ |
| 69 | if not shutil.which("ffmpeg"): |
| 70 | return _degraded("ffmpeg not installed", health.MISSING) |
| 71 | providers = env.transcription_providers(config) |
| 72 | if not providers: |
| 73 | return _degraded( |
| 74 | "no transcription provider key (set GROQ_API_KEY or OPENAI_API_KEY)", |
| 75 | health.MISSING, |
| 76 | ) |
| 77 | |
| 78 | workdir = tempfile.mkdtemp(prefix="l30d-transcribe-") |
| 79 | try: |
| 80 | audio_path = _acquire_audio(source, workdir, timeout=timeout) |
| 81 | if not audio_path: |
| 82 | return _degraded("could not acquire/compress audio", health.ERROR) |
| 83 | |
| 84 | chunk_paths = _chunk_audio(audio_path, workdir) |
| 85 | if not chunk_paths: |
| 86 | return _degraded("audio chunking produced no segments", health.ERROR) |
| 87 | |
| 88 | texts: list[str] = [] |
| 89 | used_provider = "" |
| 90 | for chunk in chunk_paths: |
| 91 | chunk_text, provider = _transcribe_chunk(chunk, providers, timeout=timeout) |
| 92 | if chunk_text is None: |
| 93 | return _degraded( |
| 94 | f"all providers failed on a chunk ({len(texts)}/{len(chunk_paths)} done)", |
| 95 | health.ERROR, |
| 96 | ) |
| 97 | texts.append(chunk_text) |
| 98 | used_provider = provider or used_provider |
| 99 | |
| 100 | joined = "\n".join(t.strip() for t in texts if t.strip()) |
| 101 | if not joined: |
| 102 | return _degraded("transcription produced empty text", health.DEGRADED) |
| 103 | return TranscriptResult( |
| 104 | text=joined, |
| 105 | ok=True, |
| 106 | provider=used_provider, |
| 107 | chunks=len(chunk_paths), |
| 108 | health=health.SourceHealth(name="transcribe", state=health.OK), |
| 109 | ) |
| 110 | finally: |
| 111 | shutil.rmtree(workdir, ignore_errors=True) |
| 112 | |
| 113 | |
| 114 | def _degraded(reason: str, state: str) -> TranscriptResult: |
| 115 | return TranscriptResult( |
| 116 | ok=False, |
| 117 | reason=reason, |
| 118 | health=health.SourceHealth(name="transcribe", state=state, reason=reason), |
| 119 | ) |
| 120 | |
| 121 | |
| 122 | def _acquire_audio(source: str, workdir: str, timeout: float) -> Optional[str]: |
| 123 | """Produce a compressed mono/16kHz/low-bitrate audio file, or None. |
| 124 | |
| 125 | For a URL, extract audio with yt-dlp; for a local path, transcode it. The |
| 126 | re-encode keeps most clips under the upload ceiling. |
| 127 | """ |
| 128 | raw = source |
| 129 | if source.startswith("http"): |
| 130 | raw = os.path.join(workdir, "raw.m4a") |
| 131 | if not _run([ |
| 132 | "yt-dlp", "-f", "bestaudio", "-o", raw, "--no-playlist", source |
| 133 | ], timeout=timeout): |
| 134 | return None |
| 135 | if not os.path.exists(raw): |
| 136 | return None |
| 137 | elif not os.path.exists(source): |
| 138 | return None |
| 139 | |
| 140 | out = os.path.join(workdir, "audio.mp3") |
| 141 | # Mono, 16kHz, 32kbps keeps speech intelligible while shrinking the file. |
| 142 | if not _run([ |
| 143 | "ffmpeg", "-y", "-i", raw, "-ac", "1", "-ar", "16000", "-b:a", "32k", out |
| 144 | ], timeout=timeout): |
| 145 | return None |
| 146 | return out if os.path.exists(out) else None |
| 147 | |
| 148 | |
| 149 | def _chunk_audio(audio_path: str, workdir: str) -> list[str]: |
| 150 | """Return [audio_path] when small enough, else ffmpeg-segmented chunk paths.""" |
| 151 | try: |
| 152 | size = os.path.getsize(audio_path) |
| 153 | except OSError: |
| 154 | return [] |
| 155 | if size <= MAX_UPLOAD_BYTES: |
| 156 | return [audio_path] |
| 157 | |
| 158 | pattern = os.path.join(workdir, "chunk_%03d.mp3") |
| 159 | if not _run([ |
| 160 | "ffmpeg", "-y", "-i", audio_path, "-f", "segment", |
| 161 | "-segment_time", str(CHUNK_SECONDS), "-c", "copy", pattern |
| 162 | ]): |
| 163 | # Fall back to the single (oversized) file; the provider may still accept it. |
| 164 | return [audio_path] |
| 165 | chunks = sorted( |
| 166 | os.path.join(workdir, f) for f in os.listdir(workdir) if f.startswith("chunk_") |
| 167 | ) |
| 168 | return chunks or [audio_path] |
| 169 | |
| 170 | |
| 171 | def _transcribe_chunk( |
| 172 | path: str, |
| 173 | providers: list[tuple[str, str]], |
| 174 | timeout: float, |
| 175 | ) -> tuple[Optional[str], str]: |
| 176 | """Try each provider in order; return (text, provider) or (None, '').""" |
| 177 | for name, key in providers: |
| 178 | try: |
| 179 | text = _post_audio(name, path, key, timeout=timeout) |
| 180 | except Exception: # noqa: BLE001 - any provider failure -> try the next |
| 181 | text = None |
| 182 | if text is not None: |
| 183 | return text, name |
| 184 | return None, "" |
| 185 | |
| 186 | |
| 187 | def _post_audio(provider: str, path: str, api_key: str, timeout: float) -> Optional[str]: |
| 188 | """POST one audio file to a Whisper-compatible endpoint; return text or None.""" |
| 189 | import json |
| 190 | import urllib.request |
| 191 | |
| 192 | endpoint = _PROVIDER_ENDPOINTS[provider] |
| 193 | model = _PROVIDER_MODELS[provider] |
| 194 | boundary = "----l30dTranscribeBoundary" |
| 195 | with open(path, "rb") as fh: |
| 196 | audio = fh.read() |
| 197 | |
| 198 | parts: list[bytes] = [] |
| 199 | parts.append(f"--{boundary}\r\n".encode()) |
| 200 | parts.append(b'Content-Disposition: form-data; name="model"\r\n\r\n') |
| 201 | parts.append(f"{model}\r\n".encode()) |
| 202 | parts.append(f"--{boundary}\r\n".encode()) |
| 203 | parts.append( |
| 204 | b'Content-Disposition: form-data; name="file"; filename="audio.mp3"\r\n' |
| 205 | b"Content-Type: audio/mpeg\r\n\r\n" |
| 206 | ) |
| 207 | parts.append(audio) |
| 208 | parts.append(f"\r\n--{boundary}--\r\n".encode()) |
| 209 | body = b"".join(parts) |
| 210 | |
| 211 | req = urllib.request.Request( |
| 212 | endpoint, |
| 213 | data=body, |
| 214 | headers={ |
| 215 | "Authorization": f"Bearer {api_key}", |
| 216 | "Content-Type": f"multipart/form-data; boundary={boundary}", |
| 217 | }, |
| 218 | method="POST", |
| 219 | ) |
| 220 | with urllib.request.urlopen(req, timeout=timeout) as resp: |
| 221 | data = json.loads(resp.read().decode("utf-8")) |
| 222 | return data.get("text") |
| 223 | |
| 224 | |
| 225 | def _run(command: list[str], timeout: float = 120.0) -> bool: |
| 226 | """Run a subprocess; return True on exit 0, False on any failure. No raise. |
| 227 | |
| 228 | Uses subproc.run_with_timeout so a timed-out yt-dlp/ffmpeg is killed at the |
| 229 | process-group level (os.setsid/killpg) instead of orphaning child trees. |
| 230 | """ |
| 231 | try: |
| 232 | result = subproc.run_with_timeout(command, timeout=int(timeout)) |
| 233 | except (subproc.SubprocTimeout, FileNotFoundError, OSError): |
| 234 | return False |
| 235 | return result.returncode == 0 |
| 236 |