""" Batch transcription using whisper.cpp. Reads /manifest.json, transcribes each audio file in module order, outputs .txt and .srt files, updates manifest status. Resumable: skips files with existing transcripts. Converts MP3 -> WAV (16kHz mono) via ffmpeg before transcription. Text lectures (type=="text") are skipped — their transcript files are written directly by download.py. """ import argparse import json import logging import os import shutil import subprocess import sys from pathlib import Path from courses import course_paths, get_course, validate_manifest_course # whisper.cpp defaults — override with env vars or CLI args. # Shared across courses (same model + binary). WHISPER_BIN = os.getenv("WHISPER_BIN", r"whisper-cli.exe") WHISPER_MODEL = os.getenv("WHISPER_MODEL", r"models\ggml-medium-q5_0.bin") logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.StreamHandler(), logging.FileHandler("transcribe_errors.log"), ], ) log = logging.getLogger(__name__) def find_ffmpeg() -> str: if shutil.which("ffmpeg"): return "ffmpeg" for p in [Path("ffmpeg.exe"), Path("ffmpeg-bin/ffmpeg.exe")]: if p.exists(): return str(p.resolve()) try: import imageio_ffmpeg return imageio_ffmpeg.get_ffmpeg_exe() except ImportError: pass return "" def convert_to_wav(audio_path: str, wav_cache_dir: Path) -> str: src = Path(audio_path) if src.suffix.lower() == ".wav": return audio_path wav_cache_dir.mkdir(parents=True, exist_ok=True) wav_path = wav_cache_dir / (src.stem + ".wav") if wav_path.exists() and wav_path.stat().st_size > 0: log.info(f" WAV cache hit: {wav_path}") return str(wav_path) ffmpeg = find_ffmpeg() if not ffmpeg: log.warning(" ffmpeg not found, using original file (may cause bad transcription)") return audio_path log.info(f" Converting to WAV: {src.name} -> {wav_path.name}") cmd = [ ffmpeg, "-i", audio_path, "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", "-y", str(wav_path), ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) if result.returncode != 0: log.error(f" ffmpeg failed: {result.stderr[:300]}") return audio_path log.info(f" WAV ready: {wav_path.name} ({wav_path.stat().st_size / 1_048_576:.0f} MB)") return str(wav_path) except FileNotFoundError: log.warning(f" ffmpeg not found at: {ffmpeg}") return audio_path except subprocess.TimeoutExpired: log.error(f" ffmpeg conversion timeout for {audio_path}") return audio_path def load_manifest(manifest_path: Path) -> dict: with open(manifest_path, encoding="utf-8") as f: return json.load(f) def save_manifest(manifest: dict, manifest_path: Path): manifest_path.parent.mkdir(parents=True, exist_ok=True) with open(manifest_path, "w", encoding="utf-8") as f: json.dump(manifest, f, indent=2, ensure_ascii=False) def transcribe_file(audio_path: str, output_base: str) -> bool: cmd = [ WHISPER_BIN, "--model", WHISPER_MODEL, "--language", "ro", "--no-gpu", "--threads", str(os.cpu_count() or 4), "--beam-size", "1", "--best-of", "1", "--max-context", "0", "--entropy-thold", "2.4", "--max-len", "60", "--suppress-nst", "--no-fallback", "--output-txt", "--output-srt", "--output-file", output_base, "--file", audio_path, ] log.info(f" CMD: {' '.join(cmd)}") try: env = os.environ.copy() whisper_dir = str(Path(WHISPER_BIN).resolve().parent) env["PATH"] = whisper_dir + os.pathsep + env.get("PATH", "") result = subprocess.run( cmd, stdout=sys.stdout, stderr=sys.stderr, timeout=7200, env=env, ) if result.returncode != 0: log.error(f" whisper.cpp failed (exit {result.returncode})") return False txt_path = Path(f"{output_base}.txt") srt_path = Path(f"{output_base}.srt") if not txt_path.exists() or txt_path.stat().st_size == 0: log.error(f" Empty or missing transcript: {txt_path}") return False log.info(f" Output: {txt_path.name} ({txt_path.stat().st_size} bytes)") if srt_path.exists(): log.info(f" Output: {srt_path.name} ({srt_path.stat().st_size} bytes)") return True except subprocess.TimeoutExpired: log.error(f" Timeout (>2h) for {audio_path}") return False except FileNotFoundError: log.error(f" whisper.cpp not found at: {WHISPER_BIN}") log.error(f" Set WHISPER_BIN env var or put whisper-cli.exe in PATH") return False except Exception as e: log.error(f" Error: {e}") return False def parse_module_filter(arg: str) -> set[int]: result = set() for part in arg.split(","): part = part.strip() if "-" in part: a, b = part.split("-", 1) result.update(range(int(a), int(b) + 1)) else: result.add(int(part)) return result def parse_args(): p = argparse.ArgumentParser(description="Transcribe lecture audio for a course") p.add_argument("--course", default="master", help="Course key (see courses.py)") p.add_argument("--modules", default=None, help="Module filter, e.g. '1-3' or '2,4,5'") return p.parse_args() def main(): args = parse_args() course = get_course(args.course) paths = course_paths(course) if not paths["manifest"].exists(): log.error(f"{paths['manifest']} not found. Run download.py --course {course['key']} first.") sys.exit(1) module_filter = parse_module_filter(args.modules) if args.modules else None if module_filter: log.info(f"Module filter: {sorted(module_filter)}") manifest = load_manifest(paths["manifest"]) validate_manifest_course(manifest, course["key"]) paths["transcripts_dir"].mkdir(parents=True, exist_ok=True) total = 0 transcribed = 0 skipped = 0 failed = 0 for mod_idx, mod in enumerate(manifest["modules"], 1): if module_filter and mod_idx not in module_filter: log.info(f"\nSkipping module {mod_idx}: {mod['name']}") continue log.info(f"\n{'='*60}") log.info(f"Module: {mod['name']}") log.info(f"{'='*60}") for lec in mod["lectures"]: total += 1 # Text and PDF lectures bypass whisper — transcript written by download.py. if lec.get("type") in ("text", "pdf"): lec["transcribe_status"] = "complete" skipped += 1 log.info(f" Skipping {lec.get('type')}: {lec['title']}") continue if lec.get("download_status") != "complete": log.warning(f" Skipping (not downloaded): {lec['title']}") continue audio_path = lec["audio_path"] # Reuse the stem already recorded in the manifest for backward-compat # with M1-M6 paths (strips ' [Audio]' for aresens filenames). stem = Path(lec["original_filename"]).stem.replace(" [Audio]", "") output_base = str(paths["transcripts_dir"] / stem) txt_path = Path(f"{output_base}.txt") if txt_path.exists() and txt_path.stat().st_size > 0: lec["transcribe_status"] = "complete" skipped += 1 log.info(f" Skipping (exists): {stem}.txt") continue log.info(f" Transcribing: {lec['title']}") log.info(f" File: {audio_path}") wav_path = convert_to_wav(audio_path, paths["wav_cache_dir"]) if transcribe_file(wav_path, output_base): lec["transcribe_status"] = "complete" transcribed += 1 else: lec["transcribe_status"] = "failed" failed += 1 save_manifest(manifest, paths["manifest"]) if mod == manifest["modules"][0] and transcribed > 0: log.info(f"First module complete ({transcribed} files). Continuing automatically...") empty_outputs = [ lec["title"] for mod in manifest["modules"] for lec in mod["lectures"] if lec.get("transcribe_status") == "complete" and lec.get("type") != "text" and not Path(lec.get("transcript_path", "")).exists() ] log.info("\n" + "=" * 60) log.info(f"Transcribed {transcribed}/{total} files, {skipped} skipped, {failed} failures.") log.info(f"No empty outputs: {'PASS' if not empty_outputs else 'FAIL'}") if empty_outputs: for t in empty_outputs: log.error(f" Missing transcript: {t}") log.info("=" * 60) save_manifest(manifest, paths["manifest"]) if failed: sys.exit(1) if __name__ == "__main__": main()