backend/ass_probe.py
Rama master · réplica de lectura del repositorio original.
"""Media-stream probing utilities: ffprobe / mediainfo + stream selection.
Extracted from ass_utils.py. Covers anything that reads the container
(audio/subtitle stream metadata) or writes a subtitle stream out again:
- run_ffprobe / get_media_streams / _run_engine — probe backends
- has_target_language_stream / has_target_language_audio
- get_all_subtitle_streams / select_best_subtitle_stream
- get_subtitle_stream_output_path / extract_subtitle_stream
The ASS tag-manipulation helpers (classify_styles, extract_tags,
restore_tags, fix_line_breaks) stay in ass_utils.py.
"""
import json
import logging
import os
import re
import subprocess
from config import get_settings
# 0.71.0 — SDH detection. Matches the common markers fansubs and studios
# use in track titles. Language codes never trigger this; we only inspect
# the `title` field (word-boundary) plus the ffprobe `hearing_impaired`
# disposition flag.
_SDH_TITLE_RE = re.compile(r"\b(sdh|cc|hi)\b", re.IGNORECASE)
def is_sdh_stream(stream: dict) -> bool:
"""Return True if the given ffprobe subtitle stream looks like SDH/CC/HI.
Detection sources (either is sufficient):
- `tags.title` matches /\b(sdh|cc|hi)\b/i
- `disposition.hearing_impaired` is truthy
"""
if not stream:
return False
tags = stream.get("tags") or {}
title = tags.get("title") or ""
if _SDH_TITLE_RE.search(title):
return True
disposition = stream.get("disposition") or {}
return bool(disposition.get("hearing_impaired"))
logger = logging.getLogger(__name__)
class _ExtractionGateBusyError(RuntimeError):
"""Keep a busy media I/O gate from being mistaken for a copy failure."""
def has_target_language_stream(ffprobe_data, target_language=None, kind="full"):
"""Check if the file has an embedded target language subtitle stream.
Args:
ffprobe_data: dict from ffprobe JSON output
target_language: Language code (e.g., "de"). If None, uses default from settings.
kind: "full" (default) counts dialogue tracks only; "forced" counts
forced and signs & songs tracks only.
Returns:
str or None: "ass" if target lang ASS found, "srt" if SRT found, None otherwise.
ASS takes priority over SRT.
"""
if target_language is None:
settings = get_settings()
target_tags = settings.get_target_lang_tags()
else:
from config import _get_language_tags
target_tags = _get_language_tags(target_language)
from forced_detection import detect_subtitle_type
target_ass = False
target_srt = False
for stream in ffprobe_data.get("streams", []):
if stream.get("codec_type") != "subtitle":
continue
lang = stream.get("tags", {}).get("language", "").lower()
if lang not in target_tags:
continue
# A forced or "Signs & Songs" track is not the dialogue subtitle: it
# must not satisfy the language, or a signs-only track would stop the
# search for the real one (and drop wanted rows since 1.14.3).
is_full = detect_subtitle_type(stream_info=stream)[0] == "full"
if is_full != (kind == "full"):
continue
codec = stream.get("codec_name", "").lower()
if codec in ("ass", "ssa"):
target_ass = True
elif codec in ("subrip", "srt"):
target_srt = True
if target_ass:
return "ass"
if target_srt:
return "srt"
return None
def get_all_subtitle_streams(ffprobe_data: dict, exclude_language: str | None = None) -> list[dict]:
"""Return all embedded subtitle streams as a list of {lang, format} dicts.
Args:
ffprobe_data: dict from get_media_streams / ffprobe JSON output.
exclude_language: ISO-639-1 code to exclude (typically the target language,
already tracked separately in existing_sub). None = return all.
Returns:
Deduplicated list of dicts with 'lang' (raw language tag) and 'format'
('ass' or 'srt'). Unknown codecs (dvd_subtitle, etc.) are skipped.
"""
exclude_tags: set[str] = set()
if exclude_language:
from config import _get_language_tags
exclude_tags = _get_language_tags(exclude_language)
seen: set[tuple] = set()
result: list[dict] = []
for stream in ffprobe_data.get("streams", []):
if stream.get("codec_type") != "subtitle":
continue
lang = stream.get("tags", {}).get("language", "").lower()
if not lang:
continue
if lang in exclude_tags:
continue
codec = stream.get("codec_name", "").lower()
if codec in ("ass", "ssa"):
fmt = "ass"
elif codec in ("subrip", "srt"):
fmt = "srt"
else:
continue
key = (lang, fmt)
if key not in seen:
seen.add(key)
result.append({"lang": lang, "format": fmt})
return result
def has_target_language_audio(ffprobe_data, target_language=None):
"""Check if the file has an audio track in the target language.
Args:
ffprobe_data: dict from ffprobe JSON output
target_language: Language code (e.g., "de"). If None, uses default from settings.
Returns:
bool: True if a target language audio track is found.
"""
if target_language is None:
settings = get_settings()
target_tags = settings.get_target_lang_tags()
else:
from config import _get_language_tags
target_tags = _get_language_tags(target_language)
for stream in ffprobe_data.get("streams", []):
if stream.get("codec_type") != "audio":
continue
lang = stream.get("tags", {}).get("language", "").lower()
if lang in target_tags:
return True
return False
def _source_candidate_tag_sets(settings) -> list[set[str]]:
"""Tag sets for every candidate source language, in preference order."""
from config_language_data import _get_language_tags
from translator._helpers import source_language_candidates
try:
candidates = source_language_candidates(settings)
except Exception: # noqa: BLE001 — a settings hiccup must not break stream selection
candidates = [settings.source_language] if getattr(settings, "source_language", "") else []
return [set(_get_language_tags(code)) for code in candidates if code]
def select_best_subtitle_stream(ffprobe_data, format_filter=None):
"""Select the best source language subtitle stream from ffprobe data.
Priority (ASS preferred over SRT):
1. Source lang ASS with "Full" in title (not Signs/Songs)
2. First source lang ASS without "sign"/"song"
3. First source lang ASS
4. ASS without target lang tag, not "sign"/"song"
5. First source lang SRT (fallback)
6. First SRT without target lang tag
7. None
Args:
ffprobe_data: dict from ffprobe JSON output
format_filter: "ass" to only search ASS, "srt" for only SRT, None for all
Returns:
dict with sub_index, format, language, title — or None
"""
settings = get_settings()
source_tags = settings.get_source_lang_tags()
target_tags = settings.get_target_lang_tags()
# Every language a translation may start from, in preference order, as
# tag sets. A track outside this list is never picked as a source: the
# old "first SRT that is not the target" fallback returned the ARABIC
# track of a Blu-ray remux and it went through the pipeline as German.
candidate_tag_sets = _source_candidate_tag_sets(settings)
candidate_tags = set().union(*candidate_tag_sets) if candidate_tag_sets else set()
def _eligible(language: str) -> bool:
# An untagged track is the norm on fansub MKVs and cannot be judged
# by its tag; anything tagged must be a configured candidate.
return not language or language == "und" or language in candidate_tags
streams = ffprobe_data.get("streams", [])
ass_streams = []
srt_streams = []
sub_index = 0
for stream in streams:
if stream.get("codec_type") != "subtitle":
continue
codec = stream.get("codec_name", "").lower()
title = stream.get("tags", {}).get("title", "").lower()
language = stream.get("tags", {}).get("language", "").lower()
info = {
"sub_index": sub_index,
"stream_index": stream.get("index"),
"title": title,
"language": language,
}
if codec in ("ass", "ssa") and format_filter != "srt":
info["format"] = "ass"
ass_streams.append(info)
elif (
codec in ("subrip", "srt", "mov_text", "webvtt", "text", "microdvd")
and format_filter != "ass"
):
info["format"] = "srt"
srt_streams.append(info)
sub_index += 1
# --- ASS Priority ---
if ass_streams:
# P1: "Full" subtitle (not signs/songs only) in a candidate language
for s in ass_streams:
if (
_eligible(s["language"])
and "full" in s["title"]
and "sign" not in s["title"]
and "song" not in s["title"]
):
logger.info("Selected stream %d: '%s' (Full ASS)", s["sub_index"], s["title"])
return s
# P2: Source language, non-signs
src = [s for s in ass_streams if s["language"] in source_tags]
for s in src:
if "sign" not in s["title"] and "song" not in s["title"]:
logger.info(
"Selected stream %d: '%s' (Source lang ASS, non-signs)",
s["sub_index"],
s["title"],
)
return s
# P3: Any source language ASS
if src:
logger.info(
"Selected stream %d: '%s' (Source lang ASS)", src[0]["sub_index"], src[0]["title"]
)
return src[0]
# P4: Non-signs/songs/OP/ED ASS in a candidate source language,
# candidates in configured order, untagged tracks last
_SKIP_TITLE_PARTS = {"sign", "song", "op", "ed", "karaoke", "credit", "caption"}
for tags in [*candidate_tag_sets, {"", "und"}]:
for s in ass_streams:
if s["language"] in tags and not any(
part in s["title"] for part in _SKIP_TITLE_PARTS
):
logger.info(
"Selected stream %d: '%s' (non-signs ASS)", s["sub_index"], s["title"]
)
return s
# --- SRT Fallback ---
if srt_streams:
# P5: Source language SRT
src_srt = [s for s in srt_streams if s["language"] in source_tags]
if src_srt:
logger.info(
"Selected stream %d: '%s' (Source lang SRT fallback)",
src_srt[0]["sub_index"],
src_srt[0]["title"],
)
return src_srt[0]
# P6: SRT in a candidate source language, candidates in configured
# order, untagged tracks last
for tags in [*candidate_tag_sets, {"", "und"}]:
for s in srt_streams:
if s["language"] in tags and s["language"] not in target_tags:
logger.info(
"Selected stream %d: '%s' (SRT fallback)", s["sub_index"], s["title"]
)
return s
# P7: target-language SRT as last resort (e.g. German dub in MP4)
if srt_streams:
tgt_srt = [s for s in srt_streams if s["language"] in target_tags]
if tgt_srt:
logger.info(
"Selected stream %d: '%s' (Target lang SRT last resort)",
tgt_srt[0]["sub_index"],
tgt_srt[0]["title"],
)
return tgt_srt[0]
# Last resort: any ASS stream in a candidate language (or untagged)
for s in ass_streams:
if _eligible(s["language"]):
logger.warning("No ideal stream, using first eligible ASS: %s", s["title"])
return s
return None
def run_ffprobe(file_path, use_cache=True):
"""Run ffprobe and return parsed JSON data for audio and subtitle streams.
Uses cache if available and file hasn't changed (mtime check).
Args:
file_path: Path to the video file
use_cache: If True, check cache first and store result
Returns:
dict: Parsed ffprobe JSON data with {"streams": [...]} containing both
audio and subtitle streams (codec_type "audio" / "subtitle").
Raises:
RuntimeError: If ffprobe fails, times out, or returns invalid JSON
"""
from db.cache import get_ffprobe_cache, set_ffprobe_cache
# Check cache if enabled
if use_cache:
try:
mtime = os.path.getmtime(file_path)
cached = get_ffprobe_cache(file_path, mtime)
if cached:
logger.debug("Using cached ffprobe data for %s", file_path)
return cached
except (OSError, Exception) as e:
logger.debug("Cache check failed for %s: %s", file_path, e)
# Run ffprobe — no -select_streams filter so both audio and subtitle streams
# are returned. Consumers filter by codec_type themselves. Previously this used
# -select_streams s which caused has_target_language_audio() to never find
# audio streams (bug fix).
#
# Audit Gemini-2026-05-09 R3: wrap ``file_path`` with ``_safe_arg_path``
# so a leading ``-`` in the basename can't be interpreted as a flag by
# ffprobe. Same defence ``extract_subtitle_stream`` already applied.
# Loglevel is ``error``, not ``quiet``: ``quiet`` suppresses stderr entirely,
# so the RuntimeError below interpolated an empty string and every failure
# logged as a bare "ffprobe failed:" with no reason (prod 2026-08-01, where
# the real cause — files deleted from disk — stayed invisible). ``error``
# stays silent on success and only speaks up when something actually broke.
from remux import _safe_arg_path
cmd = [
"ffprobe",
"-v",
"error",
"-print_format",
"json",
"-show_streams",
_safe_arg_path(file_path),
]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
except subprocess.TimeoutExpired:
raise RuntimeError(f"ffprobe timed out after 30s: {file_path}")
if result.returncode != 0:
raise RuntimeError(f"ffprobe failed: {result.stderr}")
try:
probe_data = json.loads(result.stdout)
except json.JSONDecodeError as e:
raise RuntimeError(f"ffprobe returned invalid JSON: {e}")
# Cache the result
if use_cache:
try:
mtime = os.path.getmtime(file_path)
set_ffprobe_cache(file_path, mtime, probe_data)
except (OSError, Exception) as e:
logger.debug("Cache store failed for %s: %s", file_path, e)
return probe_data
def get_media_streams(file_path, use_cache=True):
"""Unified entry point for media stream metadata — engine-agnostic.
Reads scan_metadata_engine from config and routes to ffprobe or mediainfo.
Both engines return the same normalized {"streams": [...]} format, so the
cache is engine-agnostic: an entry written by one engine is valid for another.
Args:
file_path: Path to the video file
use_cache: If True, check cache before probing and store result after
Returns:
dict: {"streams": [...]} with audio and subtitle stream objects.
Raises:
RuntimeError: On probe failure (engine-specific errors propagate).
FileNotFoundError: If engine=mediainfo and mediainfo is not installed.
"""
from db.cache import get_ffprobe_cache, set_ffprobe_cache
settings = get_settings()
engine = getattr(settings, "scan_metadata_engine", "auto")
# Cache check — shared across all engines (same format, engine-agnostic)
if use_cache:
try:
mtime = os.path.getmtime(file_path)
cached = get_ffprobe_cache(file_path, mtime)
if cached:
logger.debug("Cache hit for %s (engine=%s)", file_path, engine)
return cached
except (OSError, Exception) as e:
logger.debug("Cache check failed for %s: %s", file_path, e)
probe_data = _run_engine(file_path, engine)
if use_cache:
try:
mtime = os.path.getmtime(file_path)
set_ffprobe_cache(file_path, mtime, probe_data)
except (OSError, Exception) as e:
logger.debug("Cache store failed for %s: %s", file_path, e)
return probe_data
def _run_engine(file_path, engine):
"""Dispatch to the configured metadata engine with fallback logic for 'auto'."""
from mediainfo_utils import _is_mediainfo_available, run_mediainfo
if engine == "mediainfo":
return run_mediainfo(file_path)
if engine == "auto":
if _is_mediainfo_available():
try:
return run_mediainfo(file_path)
except Exception as e:
logger.warning("MediaInfo failed for %s, falling back to ffprobe: %s", file_path, e)
return run_ffprobe(file_path, use_cache=False)
# engine == "ffprobe" or unknown
return run_ffprobe(file_path, use_cache=False)
def get_subtitle_stream_output_path(media_path: str, stream_info: dict) -> str:
"""Build sidecar output path: {basename}.{lang}.{format} next to media file.
The container tag is ISO 639-2/B ("ger", "eng"); the name uses the
canonical ISO 639-1 code the rest of Sublarr works in. Writing the raw tag
grew a second German subtitle next to every existing ``.de`` one, and the
caller's "already extracted" guard — which keys off this very path — never
recognised the sidecar already sitting there. An unknown tag is passed
through unchanged rather than guessed at.
Uses 'und' (undetermined) if language tag is missing.
"""
from config_language_data import normalize_language_code
lang = normalize_language_code((stream_info.get("language") or "und").lower())
fmt = stream_info.get("format", "ass")
base = os.path.splitext(media_path)[0]
return f"{base}.{lang}.{fmt}"
def get_legacy_subtitle_stream_output_path(media_path: str, stream_info: dict) -> str:
"""The name this sidecar had before the language code was canonicalised.
Installs that ran the older code have their sidecars under the raw Matroska
tag (``.ger.srt``). Callers check this path too, so canonicalising the name
does not make Sublarr blind to a file it wrote itself and extract the same
track a second time.
"""
lang = (stream_info.get("language") or "und").lower()
fmt = stream_info.get("format", "ass")
base = os.path.splitext(media_path)[0]
return f"{base}.{lang}.{fmt}"
def extract_subtitle_stream(mkv_path, stream_info, output_path):
"""Extract a subtitle stream (ASS or SRT) from an MKV file atomically.
Audit G5 + G12: previously passed ``output_path`` straight into
``ffmpeg -y`` which writes incrementally to the destination. A crash,
timeout, or ENOSPC mid-write left a partial sidecar at the final
location that downstream code would treat as a complete file. The
extraction now writes to a same-directory tempfile and only
``os.replace``s into place once ffmpeg exits zero. The temp file is
unlinked on every error path so partial bytes never appear under the
real name.
Args:
mkv_path: Path to the MKV file
stream_info: dict from select_best_subtitle_stream() (needs sub_index, format)
output_path: Path to write the extracted file
Raises:
RuntimeError: If ffmpeg fails. The destination is left untouched
in this case — any pre-existing sidecar at ``output_path``
stays intact.
"""
from remux import _safe_arg_path
from utils.atomic_write import atomic_write_via
from utils.io_timeout import compute_io_timeout
ext = os.path.splitext(output_path)[1].lower().lstrip(".")
_encoder_map = {"srt": "srt", "ass": "ass", "ssa": "ass", "vtt": "webvtt"}
encoder = _encoder_map.get(ext, "copy")
settings = get_settings()
copy_first = ext in ("ass", "ssa") and bool(
getattr(settings, "embedded_extract_ass_stream_copy", False)
)
# ffmpeg reads the whole file to pull one track: a remux needs longer.
_timeout = compute_io_timeout(mkv_path, base=0, floor=getattr(settings, "ffmpeg_timeout", 300))
from services.media_io_gate import MediaGateBusyError, media_io_gate
def _run_ffmpeg(tmp_path: str, subtitle_encoder: str) -> None:
cmd = [
"ffmpeg",
"-y",
"-i",
_safe_arg_path(mkv_path),
"-map",
f"0:s:{stream_info['sub_index']}",
"-c:s",
subtitle_encoder,
_safe_arg_path(tmp_path),
]
try:
with media_io_gate.slot("ffmpeg subtitle extraction"):
result = subprocess.run(
cmd,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=_timeout,
)
except MediaGateBusyError as exc:
raise _ExtractionGateBusyError(str(exc)) from None
if result.returncode != 0:
raise RuntimeError(f"ffmpeg extraction failed: {result.stderr}")
# Refuse zero-byte output — ffmpeg sometimes returns 0 on no-data
# streams or when the disk filled mid-write but the buffer flushed
# cleanly enough to get a successful exit code.
if os.path.getsize(tmp_path) == 0:
raise RuntimeError(
"ffmpeg produced an empty sidecar (likely ENOSPC or unreadable stream)"
)
# A bare mkstemp here made every extracted sidecar 0600 — unreadable by the
# media server — and left "tmpXXXXXXXX.srt" behind whenever the process was
# killed mid-extraction. atomic_write_via relaxes the mode, uses the
# sweepable prefix and reclaims stale temps in the directory first.
def _write_extraction(tmp_path: str) -> None:
if not copy_first:
_run_ffmpeg(tmp_path, encoder)
return
try:
_run_ffmpeg(tmp_path, "copy")
except (_ExtractionGateBusyError, subprocess.TimeoutExpired):
# A copy that timed out stalled on I/O, not on the codec; the
# encoder would wait out a second full timeout on the same file.
raise
except (RuntimeError, OSError) as copy_error:
# A failed ffmpeg attempt may leave partial bytes at the temp path.
# Remove them before asking ffmpeg to encode into the same path.
try:
os.unlink(tmp_path)
except FileNotFoundError:
pass
logger.warning(
"ASS/SSA stream-copy failed; retrying with the %s encoder: %s",
encoder,
copy_error,
)
try:
_run_ffmpeg(tmp_path, encoder)
except _ExtractionGateBusyError:
raise
except (RuntimeError, OSError, subprocess.TimeoutExpired) as fallback_error:
raise RuntimeError(
f"ASS/SSA stream-copy failed ({copy_error}); "
f"encoder fallback failed ({fallback_error})"
) from fallback_error
atomic_write_via(output_path, _write_extraction)
logger.info(
"Extracted %s stream %d to %s",
stream_info.get("format", "?"),
stream_info["sub_index"],
output_path,
)