import concurrent.futures import json import os import re import sys import threading import time from datetime import datetime, timezone from pathlib import Path import requests BRIEFS_DIR = Path(sys.argv[1]) if len(sys.argv) > 1 else None if not BRIEFS_DIR: raise SystemExit("Pass the briefs directory as the first argument.") API_KEY = os.environ.get("OPENAI_API_KEY") if not API_KEY: raise SystemExit("OPENAI_API_KEY is not set.") MODEL = "gpt-4o-mini-tts-2025-12-15" VOICE = "marin" CONCURRENCY = 24 MAX_ATTEMPTS = 8 thread_local = threading.local() def utc_now(): return datetime.now(timezone.utc).isoformat() def session(): if not hasattr(thread_local, "session"): thread_local.session = requests.Session() return thread_local.session def parse_brief(source_name, markdown): sections = re.split(r"^##\s+", markdown, flags=re.MULTILINE)[1:] activities = [] for section in sections: lines = section.splitlines() activity_id = lines[0].strip() body = "\n".join(lines[1:]) instruction_match = re.search(r"^Instruction/scene track:\s*(.+)$", body, re.MULTILINE) answer_match = re.search(r"^Answer/model track:\s*(.+)$", body, re.MULTILINE) if not activity_id or not instruction_match or not answer_match: raise ValueError(f"Could not parse {source_name}, section {activity_id or 'unknown'}") activities.append( { "activity_id": activity_id, "instruction": instruction_match.group(1).strip(), "answer": answer_match.group(1).strip(), } ) return activities def valid_audio(file_path): try: if file_path.stat().st_size < 500: return False with file_path.open("rb") as stream: header = stream.read(12) return header[:3] == b"ID3" or (len(header) >= 2 and header[0] == 0xFF and header[1] & 0xE0 == 0xE0) except OSError: return False def create_speech(task): output_path = task["output_path"] if valid_audio(output_path): return "skipped", task, None if task["track_type"] == "instruction": instructions = ( "Speak in a warm, clear, patient educational voice. Use natural pacing and careful pronunciation. " "Do not add, omit, or paraphrase words. No music or sound effects." ) else: instructions = ( "Speak in a warm, clear model-answer voice. Use natural pacing and careful pronunciation. " "Do not add, omit, or paraphrase words. No music or sound effects." ) payload = { "model": MODEL, "voice": VOICE, "input": task["text"], "instructions": instructions, "response_format": "mp3", } for attempt in range(1, MAX_ATTEMPTS + 1): try: response = session().post( "https://api.openai.com/v1/audio/speech", headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}, json=payload, timeout=(15, 180), ) if response.ok: if len(response.content) < 500: raise RuntimeError(f"Audio response was unexpectedly small ({len(response.content)} bytes)") partial_path = output_path.with_suffix(output_path.suffix + ".partial") partial_path.write_bytes(response.content) partial_path.replace(output_path) return "created", task, None retryable = response.status_code == 429 or response.status_code >= 500 detail = response.text[:500] if not retryable: return "failed", task, f"HTTP {response.status_code}: {detail}" if attempt == MAX_ATTEMPTS: return "failed", task, f"HTTP {response.status_code}: {detail}" retry_after = response.headers.get("retry-after") delay = float(retry_after) if retry_after else min(30, 2**attempt) time.sleep(delay) except Exception as exc: if attempt == MAX_ATTEMPTS: return "failed", task, str(exc) time.sleep(min(30, 2**attempt)) return "failed", task, "Maximum attempts exhausted" brief_files = sorted( BRIEFS_DIR.glob("AUD-L*-U*.md"), key=lambda item: [int(value) for value in re.findall(r"\d+", item.name)], ) tasks = [] packs = [] for brief_file in brief_files: brief_id = brief_file.stem activities = parse_brief(brief_file.name, brief_file.read_text(encoding="utf-8")) pack_dir = BRIEFS_DIR / brief_id pack_dir.mkdir(parents=True, exist_ok=True) manifest = { "id": brief_id, "source_brief": brief_file.name, "model": MODEL, "voice": VOICE, "format": "mp3", "generated_at": utc_now(), "tracks": [], } for activity in activities: for track_type in ("instruction", "answer"): filename = f"{activity['activity_id']}-{track_type}.mp3" text = activity[track_type] manifest["tracks"].append( { "activity_id": activity["activity_id"], "type": track_type, "file": filename, "text": text, } ) tasks.append( { "output_path": pack_dir / filename, "text": text, "track_type": track_type, } ) packs.append((pack_dir, manifest)) stats = {"created": 0, "skipped": 0, "failed": 0} failures = [] print(f"starting briefs={len(brief_files)} tracks={len(tasks)} model={MODEL} voice={VOICE}", flush=True) with concurrent.futures.ThreadPoolExecutor(max_workers=CONCURRENCY) as executor: futures = [executor.submit(create_speech, task) for task in tasks] for finished, future in enumerate(concurrent.futures.as_completed(futures), 1): result, task, error = future.result() stats[result] += 1 if error: failures.append({"output": str(task["output_path"]), "error": error}) if finished % 25 == 0 or finished == len(tasks): print( f"progress {finished}/{len(tasks)} created={stats['created']} " f"skipped={stats['skipped']} failed={stats['failed']}", flush=True, ) for pack_dir, manifest in packs: manifest["completed_at"] = utc_now() manifest["complete"] = all(valid_audio(pack_dir / track["file"]) for track in manifest["tracks"]) (pack_dir / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") failure_path = BRIEFS_DIR / "AUD-generation-failures.json" if failures: failure_path.write_text(json.dumps(failures, indent=2) + "\n", encoding="utf-8") print(f"completed with {len(failures)} failures", file=sys.stderr, flush=True) raise SystemExit(1) failure_path.unlink(missing_ok=True) print(f"completed tracks={len(tasks)} created={stats['created']} skipped={stats['skipped']}", flush=True)