feat: add language detection from subtitle track titles and scanner versioning to invalidate stale cache entries

Add SCANNER_VERSION constant set to 2 and include in run_ffprobe() output. Create language_from_text() helper to infer language from track titles using token matching with German/English/Japanese/Korean/French/Spanish/Italian aliases. Apply language_from_text() to subtitle streams when ffprobe reports "und" language tag by checking title field. Update cached() to validate scanner_version matches
This commit is contained in:
2026-08-21 19:29:25 +02:00
parent 67e588bfc8
commit 34af0a4ad9
+26 -3
View File
@@ -679,6 +679,7 @@ def map_path(path, remote, local):
return path return path
EXTERNAL_SUBTITLE_EXTENSIONS = {".srt", ".ass", ".ssa", ".vtt", ".sub", ".idx", ".sup"} EXTERNAL_SUBTITLE_EXTENSIONS = {".srt", ".ass", ".ssa", ".vtt", ".sub", ".idx", ".sup"}
SCANNER_VERSION = 2
def external_subtitle_files(path): def external_subtitle_files(path):
media = Path(path) media = Path(path)
@@ -707,18 +708,40 @@ def external_subtitle_languages(path):
languages.append(norm_lang("it")) languages.append(norm_lang("it"))
return unique_langs(languages) return unique_langs(languages)
def language_from_text(value):
"""Infer a language from a subtitle/audio track title when the codec tag is `und`."""
tokens = {token for token in re.split(r"[. _()\[\]-]+", str(value or "").lower()) if token}
aliases = {
"de": "de", "deu": "de", "ger": "de", "german": "de", "deutsch": "de",
"en": "en", "eng": "en", "english": "en",
"ja": "ja", "jpn": "ja", "japanese": "ja", "japanisch": "ja",
"ko": "ko", "kor": "ko", "korean": "ko", "koreanisch": "ko",
"fr": "fr", "fra": "fr", "fre": "fr", "french": "fr", "französisch": "fr",
"es": "es", "spa": "es", "spanish": "es", "spanisch": "es",
"it": "it", "ita": "it", "italian": "it", "italienisch": "it",
}
for token in tokens:
if token in aliases:
return norm_lang(aliases[token])
return None
def run_ffprobe(path): def run_ffprobe(path):
proc = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=size:stream=index,codec_type,codec_name,channels,channel_layout:stream_tags=language,title", "-of", "json", path], capture_output=True, text=True, timeout=90) proc = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=size:stream=index,codec_type,codec_name,channels,channel_layout:stream_tags=language,title", "-of", "json", path], capture_output=True, text=True, timeout=90)
if proc.returncode != 0: raise RuntimeError(proc.stderr.strip() or "ffprobe fehlgeschlagen") if proc.returncode != 0: raise RuntimeError(proc.stderr.strip() or "ffprobe fehlgeschlagen")
raw, audios, subs, details, video = json.loads(proc.stdout or "{}"), [], [], [], None raw, audios, subs, details, video = json.loads(proc.stdout or "{}"), [], [], [], None
for stream in raw.get("streams", []): for stream in raw.get("streams", []):
tags = stream.get("tags") or {}; lang = norm_lang(tags.get("language")); kind = stream.get("codec_type") tags = stream.get("tags") or {}; kind = stream.get("codec_type")
lang = norm_lang(tags.get("language"))
# Viele ASS-Spuren werden von ffprobe nur als `und` gemeldet, obwohl
# der Tracktitel z. B. „Deutsch - Undefined - ASS“ enthält.
if kind == "subtitle" and lang["name"] == "Unbekannt":
lang = language_from_text(tags.get("title")) or lang
if kind == "video" and not video: video = stream.get("codec_name") if kind == "video" and not video: video = stream.get("codec_name")
elif kind == "audio": elif kind == "audio":
audios.append(lang); details.append({"language": lang, "codec": stream.get("codec_name"), "channels": stream.get("channels"), "layout": stream.get("channel_layout"), "title": tags.get("title")}) audios.append(lang); details.append({"language": lang, "codec": stream.get("codec_name"), "channels": stream.get("channels"), "layout": stream.get("channel_layout"), "title": tags.get("title")})
elif kind == "subtitle": subs.append(lang) elif kind == "subtitle": subs.append(lang)
subtitles = unique_langs(subs + external_subtitle_languages(path)) subtitles = unique_langs(subs + external_subtitle_languages(path))
return {"audio_languages": unique_langs(audios), "subtitle_languages": subtitles, "external_subtitle_files": [str(file) for file in external_subtitle_files(path)], "audio_details": details, "video_codec": video or "", "size": int((raw.get("format") or {}).get("size") or 0)} return {"scanner_version": SCANNER_VERSION, "audio_languages": unique_langs(audios), "subtitle_languages": subtitles, "external_subtitle_files": [str(file) for file in external_subtitle_files(path)], "audio_details": details, "video_codec": video or "", "size": int((raw.get("format") or {}).get("size") or 0)}
def cached(path): def cached(path):
p = Path(path) p = Path(path)
@@ -727,7 +750,7 @@ def cached(path):
if row and float(row["mtime"]) == float(mtime): if row and float(row["mtime"]) == float(mtime):
data = json.loads(row["data"]) data = json.loads(row["data"])
current_subtitles = [str(file) for file in external_subtitle_files(path)] current_subtitles = [str(file) for file in external_subtitle_files(path)]
if "external_subtitle_files" in data and data["external_subtitle_files"] == current_subtitles: if data.get("scanner_version") == SCANNER_VERSION and "external_subtitle_files" in data and data["external_subtitle_files"] == current_subtitles:
return data return data
return None return None