Setting the file. One moment.
Transcribe · Transcribing Slack Audio · PostHog/skills · Skills Docs
Back to the top of the page scripts/ transcribe.py
Python · 305 lines · 11 KB
2 bad arguments
17 3 ffmpeg, ffprobe, whisper-cli, or the model is missing
18 """
19
20 from __future__ import annotations
21
22 import os
23 import re
24 import sys
25 import json
26 import time
27 import shutil
28 import argparse
29 import tempfile
30 import subprocess
31 from pathlib import Path
32 from dataclasses import dataclass
33
34 # Matches TASK_RUN_ARTIFACT_MAX_SIZE_BYTES, the largest attachment that reaches the sandbox.
35 # The duration and processing limits below are the real bounds. This cap only stops a file
36 # that could not have come from an attachment.
37 MAX_FILE_BYTES = 30 * 1024 * 1024
38 DEFAULT_MAX_DURATION_SECONDS = 5 * 60
39 DEFAULT_TIMEOUT_SECONDS = 5 * 60
40 PROBE_TIMEOUT_SECONDS = 30
41 DEFAULT_MODEL_PATH = "/opt/whisper/models/ggml-small-q5_1.bin"
42 MODEL_PATH_ENV = "WHISPER_MODEL_PATH"
43
44 EXIT_OK = 0
45 EXIT_FAILED = 1
46 EXIT_USAGE = 2
47 EXIT_MISSING_TOOLS = 3
48
49 # whisper.cpp writes these markers for silence or noise instead of words.
50 _NON_SPEECH_MARKER = re.compile( r " \[ (?: BLANK_AUDIO | MUSIC | NOISE | SILENCE | INAUDIBLE ) \] | \( (?: silence | music | noise ) \) " , re.I)
51
52
53 class TranscriptionError ( Exception ):
54 pass
55
56
57 @dataclass ( frozen = True , kw_only = True , slots = True )
58 class Transcript :
59 text: str
60 language: str | None
61 duration_seconds: float
62
63
64 def _error (message: str ) -> None :
65 print ( f "error: { message } " , file = sys.stderr)
66
67
68 def _run (command: list[ str ], * , timeout: float , limit_seconds: int , step: str ) -> subprocess.CompletedProcess[ str ]:
69 """Run one tool. `subprocess.run` kills the child when the timeout expires."""
70 stopped = TranscriptionError( f "processing passed the { limit_seconds } second limit during { step } and was stopped" )
71 if timeout <= 0 :
72 raise stopped
73 try :
74 return subprocess.run(command, capture_output = True , text = True , timeout = timeout, check = False , stdin = subprocess. DEVNULL )
75 except subprocess.TimeoutExpired:
76 raise stopped
77
78
79 def available_cpus () -> int :
80 """CPUs this process may use, capped at 8 because whisper.cpp gains little past that.
81
82 `os.cpu_count()` reports the host. A container usually gets fewer CPUs through its
83 affinity mask or a cgroup quota, and more threads than CPUs makes whisper.cpp slower.
84 """
85 count = len (os.sched_getaffinity( 0 )) if hasattr (os, "sched_getaffinity" ) else os.cpu_count() or 1
86 try :
87 quota, period = Path( "/sys/fs/cgroup/cpu.max" ).read_text().split()[: 2 ]
88 if quota != "max" :
89 count = min (count, max ( 1 , - ( - int (quota) // int (period))))
90 except ( OSError , ValueError ):
91 pass
92 return max ( 1 , min (count, 8 ))
93
94
95 def _tail (text: str , lines: int = 5 ) -> str :
96 return " | " .join(line.strip() for line in text.strip().splitlines()[ - lines:] if line.strip())
97
98
99 def missing_tools (model_path: Path) -> list[ str ]:
100 missing = [tool for tool in ( "ffprobe" , "ffmpeg" , "whisper-cli" ) if shutil.which(tool) is None ]
101 if not model_path.is_file():
102 missing.append( f "whisper model at { model_path } (set { MODEL_PATH_ENV } to another ggml model)" )
103 return missing
104
105
106 def probe_duration_seconds (audio: Path, * , deadline: float , timeout_seconds: int ) -> float :
107 result = _run(
108 [
109 "ffprobe" ,
110 "-v" ,
111 "error" ,
112 "-select_streams" ,
113 "a" ,
114 "-show_entries" ,
115 "stream=index:format=duration" ,
116 "-of" ,
117 "json" ,
118 str (audio),
119 ],
120 timeout = min ( PROBE_TIMEOUT_SECONDS , deadline - time.monotonic()),
121 limit_seconds = timeout_seconds,
122 step = "reading the audio metadata" ,
123 )
124 if result.returncode != 0 :
125 raise TranscriptionError( f "ffprobe could not read the file as audio ( { _tail(result.stderr) or 'no details' } )" )
126 try :
127 info = json.loads(result.stdout or " {} " )
128 except json.JSONDecodeError:
129 raise TranscriptionError( "ffprobe returned unreadable metadata" )
130 if not info.get( "streams" ):
131 raise TranscriptionError( "the file has no audio stream" )
132 try :
133 return float (info[ "format" ][ "duration" ])
134 except ( KeyError , TypeError , ValueError ):
135 raise TranscriptionError( "the recording length is unknown, so the duration limit cannot be checked" )
136
137
138 def transcribe_file (
139 audio: Path,
140 * ,
141 model_path: Path,
142 language: str ,
143 max_duration_seconds: int ,
144 timeout_seconds: int ,
145 threads: int ,
146 ) -> Transcript:
147 if not audio.is_file():
148 raise TranscriptionError( "file not found" )
149 size = audio.stat().st_size
150 if size == 0 :
151 raise TranscriptionError( "the file is empty" )
152 if size > MAX_FILE_BYTES :
153 raise TranscriptionError( f "the file is { size / 1024 / 1024 :.1f} MB, over the { MAX_FILE_BYTES // 1024 // 1024 } MB limit" )
154
155 deadline = time.monotonic() + timeout_seconds
156 duration = probe_duration_seconds(audio, deadline = deadline, timeout_seconds = timeout_seconds)
157 if duration > max_duration_seconds:
158 raise TranscriptionError(
159 f "the recording is { duration / 60 :.1f} minutes long, over the { max_duration_seconds / 60 :g} minute limit"
160 )
161
162 with tempfile.TemporaryDirectory( prefix = "slack-audio-" ) as workdir:
163 wav = Path(workdir) / "audio.wav"
164 result = _run(
165 [
166 "ffmpeg" ,
167 "-nostdin" ,
168 "-hide_banner" ,
169 "-loglevel" ,
170 "error" ,
171 "-y" ,
172 "-i" ,
173 str (audio),
174 # The probe can underreport a damaged file, so the conversion cuts at the limit too.
175 "-t" ,
176 str (max_duration_seconds),
177 "-vn" ,
178 "-ac" ,
179 "1" ,
180 "-ar" ,
181 "16000" ,
182 "-c:a" ,
183 "pcm_s16le" ,
184 str (wav),
185 ],
186 timeout = deadline - time.monotonic(),
187 limit_seconds = timeout_seconds,
188 step = "converting the audio" ,
189 )
190 if result.returncode != 0 or not wav.is_file():
191 raise TranscriptionError( f "ffmpeg could not decode the audio ( { _tail(result.stderr) or 'no details' } )" )
192
193 output_base = Path(workdir) / "transcript"
194 result = _run(
195 [
196 "whisper-cli" ,
197 "--model" ,
198 str (model_path),
199 "--file" ,
200 str (wav),
201 "--language" ,
202 language,
203 "--threads" ,
204 str (threads),
205 # Greedy decoding. Beam search is several times slower on CPU for a small gain.
206 "--beam-size" ,
207 "1" ,
208 "--best-of" ,
209 "1" ,
210 "--no-prints" ,
211 "--output-json" ,
212 "--output-file" ,
213 str (output_base),
214 ],
215 timeout = deadline - time.monotonic(),
216 limit_seconds = timeout_seconds,
217 step = "transcription" ,
218 )
219 json_path = output_base.with_suffix( ".json" )
220 if result.returncode != 0 or not json_path.is_file():
221 raise TranscriptionError( f "whisper-cli failed ( { _tail(result.stderr) or 'no details' } )" )
222 try :
223 data = json.loads(json_path.read_text( encoding = "utf-8" , errors = "replace" ))
224 except json.JSONDecodeError:
225 raise TranscriptionError( "whisper-cli wrote an unreadable transcript" )
226
227 segments = [ str (segment.get( "text" ) or "" ) for segment in data.get( "transcription" ) or []]
228 text = " " .join( " " .join( _NON_SPEECH_MARKER .sub( " " , segment).split()) for segment in segments).strip()
229 text = " " .join(text.split())
230 if not text:
231 raise TranscriptionError( "no speech was detected in the recording" )
232 detected = (data.get( "result" ) or {}).get( "language" )
233 return Transcript(
234 text = text,
235 language = detected if isinstance (detected, str ) and detected else None ,
236 duration_seconds = duration,
237 )
238
239
240 def main (argv: list[ str ] | None = None ) -> int :
241 parser = argparse.ArgumentParser( description = "Transcribe audio files locally with whisper.cpp." )
242 parser.add_argument( "files" , nargs = "+" , type = Path, help = "audio files to transcribe" )
243 parser.add_argument(
244 "--language" ,
245 default = "auto" ,
246 help = "spoken language as an ISO 639-1 code such as en or de, or auto to detect it (default: auto)" ,
247 )
248 parser.add_argument(
249 "--max-duration-seconds" ,
250 type = int ,
251 default = DEFAULT_MAX_DURATION_SECONDS ,
252 help = f "refuse longer recordings (default and maximum: { DEFAULT_MAX_DURATION_SECONDS } )" ,
253 )
254 parser.add_argument(
255 "--timeout-seconds" ,
256 type = int ,
257 default = DEFAULT_TIMEOUT_SECONDS ,
258 help = f "processing time allowed per file (default and maximum: { DEFAULT_TIMEOUT_SECONDS } )" ,
259 )
260 args = parser.parse_args(argv)
261
262 if not re.fullmatch( r "auto | [ a-z ] {2,3} " , args.language):
263 _error( f "--language must be auto or a two- or three-letter language code, not { args.language !r} " )
264 return EXIT_USAGE
265 # The limits can be lowered but not raised, so a caller cannot opt out of them.
266 if not 0 < args.max_duration_seconds <= DEFAULT_MAX_DURATION_SECONDS :
267 _error( f "--max-duration-seconds must be between 1 and { DEFAULT_MAX_DURATION_SECONDS } " )
268 return EXIT_USAGE
269 if not 0 < args.timeout_seconds <= DEFAULT_TIMEOUT_SECONDS :
270 _error( f "--timeout-seconds must be between 1 and { DEFAULT_TIMEOUT_SECONDS } " )
271 return EXIT_USAGE
272
273 model_path = Path(os.environ.get( MODEL_PATH_ENV ) or DEFAULT_MODEL_PATH )
274 missing = missing_tools(model_path)
275 if missing:
276 _error( f "local transcription is not available in this environment. Missing: { ', ' .join(missing) } " )
277 return EXIT_MISSING_TOOLS
278
279 threads = available_cpus()
280 failed = False
281 for audio in args.files:
282 try :
283 transcript = transcribe_file(
284 audio,
285 model_path = model_path,
286 language = args.language,
287 max_duration_seconds = args.max_duration_seconds,
288 timeout_seconds = args.timeout_seconds,
289 threads = threads,
290 )
291 except TranscriptionError as error:
292 failed = True
293 _error( f " { audio.name } : { error } " )
294 continue
295 print (
296 f " { audio.name } : transcribed { transcript.duration_seconds :.0f} s of audio, "
297 f "language { transcript.language or args.language } " ,
298 file = sys.stderr,
299 )
300 print ( f "=== { audio.name } === \n{ transcript.text }\n " )
301 return EXIT_FAILED if failed else EXIT_OK
302
303
304 if __name__ == "__main__" :
305 sys.exit(main())