1"""Audio/video -> text with Groq's Whisper API: ffmpeg prepares speech-quality chunks, Groq transcribes them."""
2from __future__ import annotations
3
4import json
5import math
6import os
7import subprocess
8import time
9
10import httpx
11
12GROQ_URL = 'https://api.groq.com/openai/v1/audio/transcriptions'
13MODELS = {'accurate': 'whisper-large-v3', 'fast': 'whisper-large-v3-turbo'}
14CHUNK_SECONDS = 600
15MAX_RETRIES = 6
16
17
18class TranscriptionError(Exception):
19 pass
20
21
22def probe_duration(path: str) -> float:
23 """Duration of the first audio stream in seconds; raises if the file has no audio."""
24 out = subprocess.run(['ffprobe', '-v', 'error', '-select_streams', 'a:0', '-show_entries', 'stream=duration:format=duration',
25 '-of', 'json', path], capture_output=True, text=True, timeout=120)
26 try:
27 info = json.loads(out.stdout or '{}')
28 except json.JSONDecodeError:
29 info = {}
30 if not info.get('streams'):
31 raise TranscriptionError('no audio track found (not an audio/video file, damaged or unsupported)')
32 for value in (info['streams'][0].get('duration'), info.get('format', {}).get('duration')):
33 try:
34 if value and float(value) > 0:
35 return float(value)
36 except ValueError:
37 pass
38 raise TranscriptionError('cannot read the audio duration')
39
40
41def make_chunks(path: str, outdir: str, duration: float) -> list[tuple[str, float]]:
42 """Speech-quality mono MP3 chunks (16 kHz, 32 kbit/s). Returns (chunk path, start offset)."""
43 chunks = []
44 for i in range(math.ceil(duration / CHUNK_SECONDS)):
45 start = i * CHUNK_SECONDS
46 dest = os.path.join(outdir, f'chunk-{i:04d}.mp3')
47 length = min(CHUNK_SECONDS, duration - start)
48 res = subprocess.run(['ffmpeg', '-v', 'error', '-y', '-ss', str(start), '-t', f'{length:.3f}', '-i', path,
49 '-vn', '-ac', '1', '-ar', '16000', '-b:a', '32k', dest],
50 capture_output=True, text=True, timeout=900)
51 if res.returncode != 0 or not os.path.exists(dest):
52 raise TranscriptionError(f'cannot decode the audio: {res.stderr.strip()[:200]}')
53 if os.path.getsize(dest) > 1000:
54 chunks.append((dest, float(start)))
55 if not chunks:
56 raise TranscriptionError('the audio track is empty')
57 return chunks
58
59
60def transcribe_chunk(client: httpx.Client, path: str, *, model: str, language: str | None, prompt: str | None) -> dict:
61 data = {'model': model, 'response_format': 'verbose_json', 'temperature': '0'}
62 if language:
63 data['language'] = language
64 if prompt:
65 data['prompt'] = prompt[:800]
66 headers = {}
67 if os.environ.get('GROQ_API_KEY'):
68 headers['Authorization'] = 'Bearer ' + os.environ['GROQ_API_KEY']
69 for attempt in range(MAX_RETRIES):
70 with open(path, 'rb') as f:
71 r = client.post(GROQ_URL, data=data, files={'file': (os.path.basename(path), f, 'audio/mpeg')},
72 headers=headers)
73 if r.status_code == 200:
74 return r.json()
75 if r.status_code in (429, 500, 502, 503, 504) and attempt < MAX_RETRIES - 1:
76 wait = _retry_after(r.headers) or min(60, 5 * 2 ** attempt)
77 if wait > 300:
78 raise TranscriptionError('the speech service is at its usage limit right now; try again later')
79 time.sleep(wait)
80 continue
81 if r.status_code in (401, 403):
82 raise TranscriptionError('speech service authorisation failed (Actor configuration problem)')
83 raise TranscriptionError(f'speech service error HTTP {r.status_code}: {r.text[:200]}')
84 raise TranscriptionError('the speech service did not respond, try again later')
85
86
87def _retry_after(headers) -> float | None:
88 value = headers.get('retry-after')
89 try:
90 return float(value) if value else None
91 except ValueError:
92 return None
93
94
95def transcribe(path: str, workdir: str, *, quality: str = 'accurate', language: str | None = None,
96 prompt: str | None = None, max_minutes: int | None = None) -> dict:
97 """Returns {'text', 'segments', 'language', 'duration', 'minutes', 'truncated'}."""
98 duration = probe_duration(path)
99 truncated = False
100 if max_minutes is not None and duration > max_minutes * 60:
101 duration, truncated = max_minutes * 60.0, True
102 model = MODELS.get(quality, MODELS['accurate'])
103 segments, texts, detected = [], [], None
104 with httpx.Client(timeout=httpx.Timeout(300, connect=30)) as client:
105 for chunk, offset in make_chunks(path, workdir, duration):
106 res = transcribe_chunk(client, chunk, model=model, language=language,
107 prompt=prompt if not texts else (prompt or '') + ' ' + texts[-1][-400:])
108 detected = detected or res.get('language')
109 texts.append((res.get('text') or '').strip())
110 for s in res.get('segments') or []:
111 text = (s.get('text') or '').strip()
112 if text and offset + float(s['start']) < duration:
113 segments.append({'start': round(offset + float(s['start']), 2),
114 'end': round(min(offset + float(s['end']), duration), 2), 'text': text})
115 return {'text': '\n'.join(t for t in texts if t), 'segments': segments, 'language': detected,
116 'duration': round(duration, 2), 'minutes': max(1, math.ceil(duration / 60)), 'truncated': truncated,
117 'model': model}
118
119
120def _ts(seconds: float, sep: str) -> str:
121 ms = int(round(seconds * 1000))
122 h, ms = divmod(ms, 3_600_000)
123 m, ms = divmod(ms, 60_000)
124 s, ms = divmod(ms, 1000)
125 return f'{h:02d}:{m:02d}:{s:02d}{sep}{ms:03d}'
126
127
128def to_srt(segments: list[dict]) -> str:
129 return '\n'.join(f"{i}\n{_ts(s['start'], ',')} --> {_ts(s['end'], ',')}\n{s['text']}\n"
130 for i, s in enumerate(segments, 1))
131
132
133def to_vtt(segments: list[dict]) -> str:
134 return 'WEBVTT\n\n' + '\n'.join(f"{_ts(s['start'], '.')} --> {_ts(s['end'], '.')}\n{s['text']}\n" for s in segments)
