Files
movie_sync/rename_audio.py

368 lines
15 KiB
Python

#!/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()