#!/usr/bin/env python3 """ Detect audio stream languages via Whisper and set MKV metadata. Usage: python rename_audio.py /path/to/movie.mkv python rename_audio.py /path/to/movie.mkv --apply python rename_audio.py /path/to/movie.mkv --apply --stream 0,2 Features: - Probes all audio streams in the file - Runs Whisper language detection on each (showing confidence scores) - Detects uncertain results (low confidence / top language < 80%) - Sets language code + updates title with detected language name - No re-encode — metadata only via ffmpeg -map_metadata + -metadata Requires: pip install openai-whisper (or faster-whisper) """ from __future__ import annotations import argparse import json import os import shutil import subprocess import sys import tempfile try: import whisper HAS_WHISPER = True except ImportError: HAS_WHISPER = False # ─── Language code ↔ name mapping (ISO 639-1 / ISO 639-2 for MKV) ─── LANG_NAMES = { "en": "English", "de": "German", "fr": "French", "es": "Spanish", "it": "Italian", "pt": "Portuguese", "nl": "Dutch", "ru": "Russian", "ja": "Japanese", "ko": "Korean", "zh": "Chinese", "ar": "Arabic", "hi": "Hindi", "tr": "Turkish", "pl": "Polish", "sv": "Swedish", "da": "Danish", "no": "Norwegian", "fi": "Finnish", "cs": "Czech", "ro": "Romanian", "hu": "Hungarian", "uk": "Ukrainian", "el": "Greek", "he": "Hebrew", "th": "Thai", "vi": "Vietnamese", "id": "Indonesian", "ms": "Malay", "tl": "Filipino", "ca": "Catalan", "eu": "Basque", "gl": "Galician", "af": "Afrikaans", "bg": "Bulgarian", "hr": "Croatian", "sk": "Slovak", "sl": "Slovenian", "sr": "Serbian", "lt": "Lithuanian", "lv": "Latvian", "et": "Estonian", "sw": "Swahili", "ta": "Tamil", "te": "Telugu", "bn": "Bengali", "ur": "Urdu", "fa": "Persian", "ga": "Irish", "cy": "Welsh", "is": "Icelandic", "mk": "Macedonian", "sq": "Albanian", "bs": "Bosnian", "mt": "Maltese", "hy": "Armenian", "ka": "Georgian", "km": "Khmer", "my": "Burmese", "lo": "Lao", "ne": "Nepali", "si": "Sinhala", "kk": "Kazakh", "uz": "Uzbek", "mn": "Mongolian", "ps": "Pashto", "sd": "Sindhi", "gu": "Gujarati", "pa": "Punjabi", "mr": "Marathi", "kn": "Kannada", "ml": "Malayalam", "or": "Odia", "as": "Assamese", "ha": "Hausa", "yo": "Yoruba", "zu": "Zulu", "ig": "Igbo", "so": "Somali", "am": "Amharic", "rw": "Kinyarwanda", "sn": "Shona", "mg": "Malagasy", "wo": "Wolof", "ny": "Chichewa", "st": "Southern Sotho", "tn": "Tswana", "ts": "Tsonga", "ve": "Venda", "xh": "Xhosa", "ab": "Abkhaz", "aa": "Afar", "ak": "Akan", "an": "Aragonese", "as": "Assamese", "av": "Avaric", "ay": "Aymara", "az": "Azerbaijani", "be": "Belarusian", "bh": "Bihari", "bi": "Bislama", "bm": "Bambara", "bo": "Tibetan", "br": "Breton", "ch": "Chamorro", "co": "Corsican", "cr": "Cree", "cu": "Old Church Slavonic", "cv": "Chuvash", "dv": "Divehi", "dz": "Dzongkha", "ee": "Ewe", "fo": "Faroese", "fj": "Fijian", "fy": "Western Frisian", "gd": "Scottish Gaelic", "gn": "Guarani", "gv": "Manx", "ht": "Haitian", "hz": "Herero", "ia": "Interlingua", "ie": "Interlingue", "ik": "Inupiaq", "io": "Ido", "jv": "Javanese", "ka": "Georgian", "kg": "Kongo", "ki": "Kikuyu", "kj": "Kwanyama", "kl": "Kalaallisut", "kr": "Kanuri", "ku": "Kurdish", "ky": "Kyrgyz", "lb": "Luxembourgish", "lg": "Ganda", "li": "Limburgish", "ln": "Lingala", "lt": "Lithuanian", "lu": "Luba-Katanga", "lv": "Latvian", "mg": "Malagasy", "mh": "Marshall", "mi": "Maori", "mk": "Macedonian", "mn": "Mongolian", "mo": "Moldavian", "mr": "Marathi", "ms": "Malay", "mt": "Maltese", "na": "Nauru", "nb": "Norwegian Bokmal", "nd": "North Ndebele", "ne": "Nepali", "ng": "Ndonga", "nn": "Norwegian Nynorsk", "nr": "South Ndebele", "nv": "Navajo", "oc": "Occitan", "oj": "Ojibwa", "om": "Oromo", "or": "Odia", "os": "Ossetic", "pi": "Pali", "ps": "Pashto", "qu": "Quechua", "rm": "Romansh", "rn": "Rundi", "rw": "Kinyarwanda", "sa": "Sanskrit", "sc": "Sardinian", "se": "Northern Sami", "sg": "Sango", "si": "Sinhala", "sm": "Samoan", "sn": "Shona", "sd": "Sindhi", "so": "Somali", "sr": "Serbian", "ss": "Swati", "su": "Sundanese", "sw": "Swahili", "ty": "Tahitian", "ug": "Uyghur", "uz": "Uzbek", "ve": "Venda", "vo": "Volapuk", "wa": "Walloon", "wo": "Wolof", "xh": "Xhosa", "yi": "Yiddish", "yo": "Yoruba", "za": "Zhuang", "zu": "Zulu", } # ISO 639-2/B codes for MKV (3-letter) LANG_CODE_TO_3 = { "en": "eng", "de": "ger", "fr": "fre", "es": "spa", "it": "ita", "pt": "por", "nl": "dut", "ru": "rus", "ja": "jpn", "ko": "kor", "zh": "chi", "ar": "ara", "hi": "hin", "tr": "tur", "pl": "pol", "sv": "swe", "da": "dan", "no": "nor", "fi": "fin", "cs": "cze", "ro": "rum", "hu": "hun", "uk": "ukr", "el": "gre", "he": "heb", "th": "tha", "vi": "vie", "id": "ind", "ms": "may", "tl": "fil", "ca": "cat", "eu": "baq", "gl": "gal", "af": "afr", "bg": "bul", "hr": "hrv", "sk": "slo", "sl": "slv", "sr": "srp", "lt": "lit", "lv": "lav", "et": "est", "sw": "swa", "ta": "tam", "te": "tel", "bn": "ben", "ur": "urd", "fa": "per", "ga": "gle", "cy": "wel", "is": "ice", "mk": "mac", "sq": "alb", "bs": "bos", "mt": "mlt", "hy": "arm", "ka": "geo", "km": "khm", "my": "bur", "lo": "lao", "ne": "nep", "si": "sin", "kk": "kaz", "uz": "uzb", "mn": "mon", "ps": "pus", "sd": "snd", "gu": "guj", "pa": "pan", "as": "asm", "ha": "hau", "yo": "yor", "zu": "zul", "ig": "ibo", "so": "som", "am": "amh", "rw": "kin", "sn": "sna", "ny": "nya", "wo": "wol", } def probe_audio_streams(path: str) -> list[dict]: """Probe all audio streams in a file.""" try: r = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=index,codec_name,channels,sample_rate,bit_rate," "stream_tags=language,title", "-of", "json", path], capture_output=True, text=True, timeout=10 ) if r.returncode != 0: return [] data = json.loads(r.stdout) streams = [] for s in data.get("streams", []): tags = {k.lower(): v for k, v in (s.get("tags") or {}).items()} sr = s.get("sample_rate") streams.append({ "index": s.get("index", 0), "codec": s.get("codec_name", ""), "channels": s.get("channels", 0), "sample_rate": int(sr) if sr else 0, "language": tags.get("language", ""), "title": tags.get("title", ""), }) return streams except Exception: return [] def extract_audio_segment(video_path: str, stream_index: int, out_wav: str, start_sec: float = 0, duration_sec: float = 60) -> bool: """Extract a segment of an audio stream for language detection.""" try: r = subprocess.run( ["ffmpeg", "-y", "-ss", str(start_sec), "-i", video_path, "-map", f"0:a:{stream_index}", "-t", str(duration_sec), "-acodec", "pcm_f32le", "-ar", "16000", "-ac", "1", "-v", "error", out_wav], capture_output=True, timeout=30 ) return r.returncode == 0 and os.path.exists(out_wav) and os.path.getsize(out_wav) > 0 except Exception: return False def detect_language_whisper(wav_path: str, model_name: str = "base") -> tuple[str, float, dict]: """ Detect language using Whisper. Returns (lang_code, top_probability, full_probs). """ if not HAS_WHISPER: raise RuntimeError("openai-whisper not installed. Run: pip install openai-whisper") model = whisper.load_model(model_name) audio = whisper.load_audio(wav_path) # trim to 30 seconds max for speed audio = audio[:30 * 16000] result = model.detect_language(audio) lang = result["language"] probs = result.get("probabilities", {}) # Build full probability dict full_probs = {} if probs is not None: for i, p in enumerate(probs): lang_token = whisper.tokenizer._lang_code_to_token.get(i, "") if lang_token: full_probs[lang_token] = float(p) # Top probability top_prob = full_probs.get(lang, 0.0) return lang, top_prob, full_probs def get_lang_name(code: str) -> str: """Get human-readable language name from ISO 639-1 code.""" return LANG_NAMES.get(code, code.upper()) def main(): ap = argparse.ArgumentParser(description="Detect audio languages via Whisper and set MKV metadata") ap.add_argument("file", help="MKV/MP4 file to process") ap.add_argument("--apply", action="store_true", help="Actually apply changes (default: dry run)") ap.add_argument("--stream", default=None, help="Process specific stream indices (comma-separated, e.g. 0,2)") ap.add_argument("--model", default="base", help="Whisper model (tiny/base/small/medium/large)") ap.add_argument("--segment", type=float, default=60, help="Seconds of audio to analyze (default: 60)") ap.add_argument("--start", type=float, default=0, help="Start offset for analysis (seconds)") args = ap.parse_args() if not os.path.exists(args.file): print(f"ERROR: file not found: {args.file}", file=sys.stderr) sys.exit(1) if not HAS_WHISPER: print("ERROR: openai-whisper not installed.", file=sys.stderr) print("Install with: pip install openai-whisper", file=sys.stderr) print("Or for faster detection: pip install faster-whisper", file=sys.stderr) sys.exit(1) streams = probe_audio_streams(args.file) if not streams: print("No audio streams found.", file=sys.stderr) sys.exit(1) print(f"File: {args.file}") print(f"Audio streams: {len(streams)}") print() # Filter streams if --stream specified if args.stream: selected = [int(x.strip()) for x in args.stream.split(",")] streams = [s for s in streams if s["index"] in selected] # Create temp dir for extracted segments with tempfile.TemporaryDirectory(prefix="rename_audio_") as tmp_dir: changes = [] for s in streams: idx = s["index"] cur_lang = s["language"] cur_title = s["title"] print(f"{'=' * 60}") print(f"Stream [{idx}]: {s['codec'].upper()} {s['channels']}ch") print(f" Current language: {cur_lang or '(none)'}") print(f" Current title: {cur_title or '(none)'}") # Extract audio segment wav_path = os.path.join(tmp_dir, f"stream_{idx}.wav") print(f" Extracting {args.segment}s from {args.start}s...") if not extract_audio_segment(args.file, idx, wav_path, args.start, args.segment): print(f" SKIP: failed to extract audio", file=sys.stderr) continue # Detect language print(f" Detecting language with Whisper ({args.model})...") try: lang, top_prob, all_probs = detect_language_whisper(wav_path, args.model) except Exception as e: print(f" ERROR: detection failed: {e}", file=sys.stderr) continue lang_name = get_lang_name(lang) print(f"\n Detected: {lang_name} ({lang}) — confidence: {top_prob:.1%}") # Show top 5 probabilities sorted_probs = sorted(all_probs.items(), key=lambda x: -x[1])[:5] print(f" Top languages:") for l, p in sorted_probs: marker = " <<<" if l == lang else "" print(f" {get_lang_name(l):20s} ({l}) {p:.1%}{marker}") # Confidence assessment if top_prob >= 0.80: confidence = "HIGH" elif top_prob >= 0.50: confidence = "MEDIUM (review recommended)" else: confidence = "LOW (likely unsupported language)" print(f" Confidence: {confidence}") # Determine new metadata lang_3 = LANG_CODE_TO_3.get(lang, lang) needs_change = False if cur_lang != lang_3: needs_change = True print(f"\n → Language will change: {cur_lang or '(none)'} → {lang_3}") else: print(f"\n → Language already correct: {lang_3}") # Build new title: prepend detected language name if not already present new_title = cur_title if lang_name.lower() not in (cur_title or "").lower(): if cur_title: new_title = f"{lang_name} {cur_title}" else: new_title = lang_name if new_title != cur_title: needs_change = True print(f" → Title will change: \"{cur_title or '(none)'}\" → \"{new_title}\"") if not needs_change: print(f" → No changes needed") continue # Build ffmpeg command cmd_str = apply_metadata(args.file, idx, lang_3, new_title, dry_run=True) print(f"\n Command: {cmd_str}") changes.append({ "index": idx, "old_lang": cur_lang, "new_lang": lang_3, "old_title": cur_title, "new_title": new_title, "detected": lang, "confidence": top_prob, "cmd": cmd_str, }) print(f"\n{'=' * 60}") if not changes: print("No changes to apply.") return print(f"Changes to apply: {len(changes)}") for c in changes: print(f" Stream [{c['index']}]: lang {c['old_lang'] or '(none)'} → {c['new_lang']}, " f"title → \"{c['new_title']}\" ({c['confidence']:.0%} confidence)") if not args.apply: print(f"\nDry run complete. Use --apply to apply changes.") return # Apply changes print(f"\nApplying changes...") for c in changes: lang_3 = c["new_lang"] title = c["new_title"] idx = c["index"] meta_key = f"metadata:s:a:{idx}" base, ext = os.path.splitext(args.file) tmp_out = f"{base}_tmp{ext}" cmd = [ "ffmpeg", "-y", "-i", args.file, "-map", "0", "-c", "copy", f"{meta_key}:language={lang_3}", f"{meta_key}:title={title}", tmp_out ] print(f" Stream [{idx}]: {' '.join(cmd[:8])}...") r = subprocess.run(cmd, capture_output=True, timeout=300) if r.returncode != 0: print(f" ERROR: ffmpeg failed: {r.stderr.decode()[-200:]}", file=sys.stderr) continue shutil.move(tmp_out, args.file) print(f" ✓ Stream [{idx}] updated") print(f"\nDone. File: {args.file}") if __name__ == "__main__": main()