#!/usr/bin/env python3 """Whisper language-ID spot-check of what each video's audio ACTUALLY is. Nothing else listens to the audio: `media_stack.audio` trusts the language TAG and dispositions, and release names and tags can be wrong (a "Dual Spanish" release imported Spanish-only; a Japanese dub flagged `default`). This samples short whisper LID windows of each non-commentary audio stream and cross-checks them against the stream's tag (MISLABELED) and Radarr/Sonarr's `originalLanguage` (`media_stack.arrlang`; NO_ENGLISH_AUDIO). A false alert costs the operator's time, so anything not determined with confidence is UNVERIFIED. CLASSIFICATION: A FOREIGN TAG NEVER OUTWEIGHS A CONFIDENT DETECTION Every decision looks at ALL sampled windows against RAW probabilities (0.7996 is not 0.8). A stream counts as English when it is English-tagged and not MISLABELED, OR any window detects English at >= 0.6 whatever its tag. It is ruled OUT only when EVERY window is confident (>= 0.6) and none says English. NO_ENGLISH_AUDIO fires only when the title's original language is English AND every non-commentary stream is ruled out. A mixed or inconclusive stream with nothing confirmed English, or a file with no non-commentary audio, is UNVERIFIED (no finding, no alert). MISLABELED needs the SAME non-English language at >= 0.8 in two windows starting >= 60 s apart, so one hallucination or foreign insert cannot flag a correctly tagged track. The scan (25/50/10/75% of the file, subject to the 60 s spacing) stops early ONLY on a confident English window; otherwise every planned window is sampled. A file under 3 minutes cannot offer two such windows: UNVERIFIED. So is a file with no usable duration (container, audio stream, then a bounded ffprobe call). NEVER DELETES, BLOCKLISTS, OR RE-GRABS: Sonarr/Radarr will not import a same-quality replacement while the file exists, and deleting first risks leaving nothing. Findings go to Telegram, the state file and the JSONL log. PLAYBACK: FAILS CLOSED, BOUNDED (`media_stack.playback`) `None` from `JellyfinClient.now_playing_sessions` means "someone is watching". No library file is read (ffprobe included) before the gate says idle, and every ffmpeg extraction checks playback before starting and while running, aborting and retrying. Waits are bounded by `AUDIO_CHECK_MAX_WAIT_MIN` cumulatively PER FILE; past it the whole run ends cleanly (exit 0, next hourly tick retries). A real run without `JELLYFIN_API_KEY` refuses to start (exit 1); `--dry-run` is exempt. CANDIDATES `.mkv`/`.mp4` under `MEDIA_ROOT`'s `movies/` and `tv/`, skipping hidden names (including `.dupe-recycle`) and files modified < 10 minutes ago (still being written). Newest mtime first, so a fresh import surfaces within the hour while the backlog drains. A file is re-checked only when its `[st_ino, st_size, st_mtime_ns]` changes. A title added to Radarr/Sonarr AFTER its file was checked is not revisited until the file changes. STATE: ONE FILE, VALIDATED WHOLE `var/state/audio-lang-check.state.json` is the only file read back. Per resolved path: identity [st_ino, st_size, st_mtime_ns] the conclusion was made on result "CHECKED" or "UNVERIFIED" findings MISLABELED / NO_ENGLISH_AUDIO dicts (may be empty) first_found, alerted only with findings checked_at failed failed analyses of that identity: identity, attempts, last_result, last_error Keys of older versions (`streams`, `original_status`, `original_language`, `original_rechecked_at`, and `retry`, whose counter also counted outages) are ignored on load, so a leftover `retry` never rules a file out, and dropped whenever the entry is rewritten. Every update is one `mutate_state` (flock), so a conclusion, findings, alert flag and identity are recorded together or not at all. The whole structure is validated on load and the script refuses to WRITE an entry the loader would reject. An unreadable file or malformed entry is a blinding error (exit 1 before any state or log write), never "treat as empty". FAILURES ARE RETRIED A FEW TIMES, NOT CACHED PROBE_FAILED and ERROR are tried again on every run until the same identity has failed `MAX_FAILED_ATTEMPTS` times (counted in `failed`); after that the file is skipped until it changes. ARR_UNAVAILABLE (Radarr/Sonarr could not be asked, as opposed to no usable language) records nothing (no state, no JSONL line) and is summarised in one log line: an outage never uses up a file's attempts. DEPENDENCY_MISSING (no `faster_whisper`) records nothing and ends the run with exit 1. FINDINGS ARE BOUND TO A FILE IDENTITY: an alert only goes out for an entry whose identity equals the file's current one; entries for a gone or replaced file are never sent. If the file changed during analysis, the result is discarded and a later run redoes it. ONE FILE AT A TIME, IDLE PRIORITY: `AUDIO_CHECK_BUDGET_MIN` stops STARTING new files once spent. Temp WAVs live in a per-file workdir under `WORKDIR_ROOT` (local disk, never the media pool). `WHISPER_THREADS` defaults to `min(2, cpu_count)` and cron runs it under `nice -n19 ionice -c3`; the pause-aware extraction is the real protection. ALERTING: EACH NEW FINDING IS SENT ONCE At the end of every real run (also after a playback timeout or a missing `faster_whisper`), findings that are new or changed (`alerted` not true, file still at the identity they were made on) go to Telegram, oldest first, at most 15 items and 3800 characters per message (Telegram rejects > 4096 UTF-16 units), each item cut to 600, as many messages as that takes. Every finding is marked `alerted` once its message was attempted, so a failed send is NOT retried: it is logged to stderr and as an `ALERT_FAILED` JSONL record naming the files. Credentials come from `TELEGRAM_BOT_TOKEN`/`TELEGRAM_CHAT_ID` in the process environment, else the Claude Telegram plugin's `.env` (never this repo's `.env`); with both in the environment the plugin file is not read, and a failure reading it is logged and fails soft (the send then fails and is logged as above). Exit codes: 0 ok, including nothing to do, a second cron instance finding the lock held (exits quietly), and a playback wait that exceeded AUDIO_CHECK_MAX_WAIT_MIN 1 blinded: no JELLYFIN_API_KEY on a real run, unreadable or malformed state, an English-only (`.en`) WHISPER_LID_MODEL, or `faster_whisper` missing 2 usage error (e.g. `--dry-run` with `--path`) 75 `--path` only (EX_TEMPFAIL): another run holds the lock; stdout carries {"path": ..., "result": "BUSY", "reason": ...} `--path FILE` checks one file and prints JSON. It records NOTHING (no state, JSONL or Telegram), so it can never suppress the cron alert for the same file. It cannot be combined with `--dry-run` (exit 2). `--dry-run` makes no WRITES: no `var/` directories, no lock file (importing the script writes nothing; `main()` creates the directories for a run that writes). """ from __future__ import annotations import argparse import json import math import os import re import signal import sys import tempfile import time from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Iterator sys.path.insert(0, str(Path(__file__).resolve().parent)) from media_stack.arrlang import ORIGINAL_UNAVAILABLE, original_language_lookup # noqa: E402 from media_stack.audio import is_commentary_audio # noqa: E402 from media_stack.clients.jellyfin import JellyfinClient # noqa: E402 from media_stack.clients.telegram import send as telegram_send # noqa: E402 from media_stack.envconfig import env_float, env_int # noqa: E402 from media_stack.lang import is_english_tag # noqa: E402 from media_stack.locking import ( # noqa: E402 acquire_singleton_lock, handle_sigterm as _handle_sigterm, ) from media_stack.paths import ( # noqa: E402 MEDIA_LAN_IP, MEDIA_ROOT, MEDIA_STACK_ROOT, VAR_DIR, VAR_LOG, VAR_RUN, VAR_STATE, ensure_var_dirs, load_env_file, ) from media_stack.playback import ( # noqa: E402 PlaybackWaitBudget, PlaybackWaitTimeout, call_with_playback_pause, wait_while_playing, ) from media_stack.probe import ( # noqa: E402 audio_stream_duration, parse_duration, probe, probe_duration, ) from media_stack.state import mutate_state # noqa: E402 from media_stack.subverify import detect_language # noqa: E402 # No `ensure_var_dirs()` at import: `--dry-run` writes nothing. STATE_FILE = VAR_STATE / "audio-lang-check.state.json" LOCK_FILE = VAR_RUN / "audio-lang-check.lock" JSONL_FILE = VAR_LOG / "audio-lang-check.jsonl" # Shared with sub-fallback.py (distinct per-file prefixes). WORKDIR_ROOT = VAR_DIR / "tmp" # Younger files are probably still being written or renamed. MIN_FILE_AGE_S = 10 * 60 LID_WINDOW_S = 30.0 # 0.75 lets a file just over MIN_MISLABEL_DURATION_S have a window far enough from the 25% one. LID_FRACTIONS = (0.25, 0.50, 0.10, 0.75) LID_PROBABILITY_FLOOR = 0.6 MISLABEL_PROBABILITY_FLOOR = 0.8 MIN_WINDOW_GAP_S = 60.0 # windows are independent evidence only this far apart # Below this, two such windows cannot be placed: MISLABELED is skipped. MIN_MISLABEL_DURATION_S = 180.0 # Results that are never a conclusion. Only the first two are counted against a file. FAILED_RESULTS = frozenset({"PROBE_FAILED", "ERROR"}) MAX_FAILED_ATTEMPTS = 3 CONCLUSION_RESULTS = frozenset({"CHECKED", "UNVERIFIED"}) FINDING_TYPES = frozenset({"MISLABELED", "NO_ENGLISH_AUDIO"}) MAX_ALERT_ITEMS = 15 # Under Telegram's 4096 UTF-16 units. TELEGRAM_BUDGET_CHARS = 3800 MAX_ALERT_LINE_CHARS = 600 _MAX_MINUTES_SETTING = 1440.0 # sysexits.h EX_TEMPFAIL. EXIT_BUSY = 75 @dataclass class Config: jellyfin_url: str jellyfin_key: str budget_min: float max_wait_min: float whisper_lid_model: str whisper_threads: int def load_config(env: dict[str, str]) -> Config: # The default must respect the bound env_int enforces on overrides. cpu_bound = os.cpu_count() or 64 default_threads = min(2, cpu_bound) return Config( jellyfin_url=env.get("JELLYFIN_URL") or f"http://{MEDIA_LAN_IP}:8096", jellyfin_key=env.get("JELLYFIN_API_KEY", ""), budget_min=env_float(env, "AUDIO_CHECK_BUDGET_MIN", 20.0, _MAX_MINUTES_SETTING, log=log), max_wait_min=env_float(env, "AUDIO_CHECK_MAX_WAIT_MIN", 15.0, _MAX_MINUTES_SETTING, log=log), whisper_lid_model=env.get("WHISPER_LID_MODEL") or "small", whisper_threads=env_int(env, "WHISPER_THREADS", default_threads, cpu_bound, log=log), ) # `--path` prints JSON on stdout, so logs go to stderr in that mode. _LOG_TO_STDERR = False _LONE_SURROGATE_RE = re.compile("[\ud800-\udfff]") def _clean_text(text: str) -> str: """Replace lone surrogates with U+FFFD: a file name with invalid UTF-8 arrives with `surrogateescape` surrogates and cannot be encoded, so path-derived text goes through here.""" return _LONE_SURROGATE_RE.sub("�", text) def log(msg: str, *, err: bool = False) -> None: """Timestamped line to stdout (stderr with `err`, and always in `--path` mode).""" print(_clean_text(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}"), flush=True, file=sys.stderr if err or _LOG_TO_STDERR else sys.stdout) def _utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="microseconds") def iter_media_files(media_root: Path) -> Iterator[Path]: """`.mkv`/`.mp4` under `movies/` and `tv/`, skipping hidden names (e.g. `.dupe-recycle`).""" for sub in ("movies", "tv"): root = media_root / sub if not root.is_dir(): continue for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if not d.startswith(".")] for name in filenames: if name.startswith("."): continue if name.lower().endswith((".mkv", ".mp4")): yield Path(dirpath) / name def _identity(path: Path, now_s: float | None = None) -> list[int] | None: """`[st_ino, st_size, st_mtime_ns]`, or `None` if gone (or, given `now_s`, modified inside the `MIN_FILE_AGE_S` settle window). Deliberately not `media_stack.fileid`'s four fields: ctime also moves when Radarr/Sonarr add or remove a hardlink, which would re-run whisper on and re-alert about unchanged audio. A same-size rewrite with mtime restored is missed: a stale verdict, never a wrong write.""" try: st = path.stat() except OSError: return None if now_s is not None and now_s - st.st_mtime < MIN_FILE_AGE_S: return None return [st.st_ino, st.st_size, st.st_mtime_ns] def _finite_number(value: object) -> bool: """Finite int/float, never a bool.""" if isinstance(value, bool) or not isinstance(value, (int, float)): return False try: return math.isfinite(value) except OverflowError: return False def _failed_attempts(entry: object, identity: list[int]) -> int: """`PROBE_FAILED`/`ERROR` analyses recorded for exactly this version of the file.""" failed = entry.get("failed") if isinstance(entry, dict) else None attempts = failed.get("attempts") if isinstance(failed, dict) and failed.get("identity") == identity else 0 return attempts if _is_index(attempts) else 0 def select_candidates(state: dict, media_root: Path, now_s: float) -> list[Path]: """Settled files whose identity is not already concluded, or failed `MAX_FAILED_ATTEMPTS` times; newest mtime first.""" found: list[tuple[int, Path]] = [] for video in iter_media_files(media_root): identity = _identity(video, now_s) if identity is None: continue cached = state.get(str(video.resolve())) if isinstance(cached, dict) and cached.get("identity") == identity: continue if _failed_attempts(cached, identity) >= MAX_FAILED_ATTEMPTS: continue found.append((identity[2], video)) found.sort(key=lambda t: t[0], reverse=True) return [p for _, p in found] def _mislabeled_language(windows: list[dict]) -> tuple[str, float] | None: """Two windows with the SAME non-English language at >= `MISLABEL_PROBABILITY_FLOOR`, >= `MIN_WINDOW_GAP_S` apart: `(language, lower raw probability)`, else `None`.""" by_lang: dict[str, list[dict]] = {} for w in windows: if w["language"] != "en" and w["probability"] >= MISLABEL_PROBABILITY_FLOOR: by_lang.setdefault(w["language"], []).append(w) for lang, ws in by_lang.items(): for i, a in enumerate(ws): for b in ws[i + 1:]: if abs(a["start"] - b["start"]) >= MIN_WINDOW_GAP_S: return lang, min(a["probability"], b["probability"]) return None def _lid_scan( video: Path, audio_index: int, duration: float, cfg: Config, jellyfin: JellyfinClient, budget: PlaybackWaitBudget, workdir: Path, ) -> list[dict]: """Whisper LID over a stream's windows in `LID_FRACTIONS` order, stopping early only on a confident ENGLISH window. A fraction within `MIN_WINDOW_GAP_S` of a sampled one is skipped. Returns windows (`fraction`, `start`, `language`, RAW `probability`), at least one. Only `PlaybackWaitTimeout` propagates.""" windows: list[dict] = [] for frac in LID_FRACTIONS: start = max(0.0, duration * frac) if any(abs(start - w["start"]) < MIN_WINDOW_GAP_S for w in windows): continue lang, prob = call_with_playback_pause( detect_language, jellyfin, cfg.jellyfin_key, cfg.max_wait_min, budget, video, audio_index, start, LID_WINDOW_S, cfg.whisper_lid_model, cfg.whisper_threads, workdir, log=log, ) windows.append({"fraction": frac, "start": start, "language": lang, "probability": prob}) if lang == "en" and prob >= LID_PROBABILITY_FLOOR: break return windows def _classify_windows(windows: list[dict]) -> str: """`"english"` (any window English at >= `LID_PROBABILITY_FLOOR`), `"foreign"` (every window confident, none English; the only reading that rules a stream out) or `"unclear"`.""" if any(w["language"] == "en" and w["probability"] >= LID_PROBABILITY_FLOOR for w in windows): return "english" if windows and all(w["probability"] >= LID_PROBABILITY_FLOOR and w["language"] != "en" for w in windows): return "foreign" return "unclear" def _program_indices(streams: list[dict]) -> set[int]: """Indices of the scanned streams.""" return {s["index"] for s in streams if isinstance(s.get("index"), int)} def _duration_from_info(info: dict, streams: list[dict]) -> float | None: """Container duration, else an audio stream's (never video or cover art).""" d = parse_duration((info.get("format") or {}).get("duration")) if d is not None: return d first = streams[0].get("index") if streams else None return audio_stream_duration(streams, first, _program_indices(streams)) def analyze_candidate( video: Path, info: dict, original: str | None, cfg: Config, jellyfin: JellyfinClient, budget: PlaybackWaitBudget, workdir: Path, ) -> tuple[list[dict], list[dict], str, str | None]: """Analyze every non-commentary audio stream; `original` is the Arr original-language code ("eng"/"jpn"/"kor"/`None`). Returns `(findings, stream_records, result, note)`; `result` is `"CHECKED"` or `"UNVERIFIED"` (`note` says why). Rules: module docstring. UNVERIFIED covers no audio, no usable duration, and a file shorter than `MIN_MISLABEL_DURATION_S` with an English-tagged stream (not scanned).""" streams = [ s for s in info.get("streams", []) if s.get("codec_type") == "audio" and not is_commentary_audio(s) ] if not streams: return [], [], "UNVERIFIED", "no non-commentary audio stream to check" duration = _duration_from_info(info, streams) if duration is None: # Another read of the library file: same gate, same budget. wait_while_playing(jellyfin, cfg.jellyfin_key, cfg.max_wait_min, budget, log=log) duration = probe_duration(video, streams[0].get("index"), _program_indices(streams)) if duration is None: return [], [], "UNVERIFIED", "no usable duration; no window sampled" mislabel_checkable = duration >= MIN_MISLABEL_DURATION_S findings: list[dict] = [] stream_records: list[dict] = [] mislabel_skipped = False for s in streams: idx = s.get("index") raw_tag = ((s.get("tags") or {}).get("language") or "").strip().lower() tagged_english = is_english_tag(raw_tag) if tagged_english and not mislabel_checkable: # English by tag; the only check that could overturn it can't run. mislabel_skipped = True stream_records.append({ "index": idx, "tag": raw_tag, "detected_language": None, "probability": None, "windows": [], "counts_as_english": True, "reading": None, "mislabel_skipped": True, }) continue windows = _lid_scan(video, idx, duration, cfg, jellyfin, budget, workdir) best = max(windows, key=lambda w: w["probability"]) # display only reading = _classify_windows(windows) # Stored probabilities are rounded; decisions use the raw windows. rec = { "index": idx, "tag": raw_tag or None, "detected_language": best["language"], "probability": round(best["probability"], 3), "windows": [{**w, "probability": round(w["probability"], 3)} for w in windows], "reading": reading, } # A confident English window, even after two foreign ones, rules out MISLABELED. mislabel = (_mislabeled_language(windows) if tagged_english and reading != "english" else None) if mislabel is not None: lang, prob = mislabel rec["mislabeled_as"] = lang findings.append({ "type": "MISLABELED", "stream_index": idx, "tag": raw_tag or None, "detected_language": lang, "probability": round(prob, 3), }) rec["counts_as_english"] = (tagged_english and mislabel is None) or reading == "english" stream_records.append(rec) findings, result, note = _decide(original, stream_records, findings, mislabel_skipped) return findings, stream_records, result, note def _decide( original: str | None, stream_records: list[dict], mislabeled: list[dict], mislabel_skipped: bool, ) -> tuple[list[dict], str, str | None]: """The ONE place the NO_ENGLISH_AUDIO / UNVERIFIED rules live. `mislabeled` is the MISLABELED findings; a stream's `reading` is `None` when its MISLABELED check was skipped.""" findings = [dict(f) for f in mislabeled] any_english = any(r.get("counts_as_english") for r in stream_records) any_unclear = any(not r.get("counts_as_english") and r.get("reading") != "foreign" for r in stream_records) note: str | None = None result = "CHECKED" if original == "eng" and not any_english: if any_unclear: note = "a stream's language could not be determined" if not findings: result = "UNVERIFIED" else: findings.append({ "type": "NO_ENGLISH_AUDIO", "original_language": original, "streams": [{"index": r["index"], "tag": r["tag"], "detected_language": r["detected_language"], "probability": r["probability"]} for r in stream_records], }) if mislabel_skipped and result == "CHECKED": result, note = "UNVERIFIED", ( f"shorter than {MIN_MISLABEL_DURATION_S:g}s: MISLABELED needs two independent windows") return findings, result, note def process_candidate( video: Path, cfg: Config, jellyfin: JellyfinClient, workdir: Path, ) -> dict: """Original-language lookup, playback gate, probe, analysis of ONE file. Raises only `PlaybackWaitTimeout`; failures come back as a `result` of `PROBE_FAILED`, `ERROR`, `ARR_UNAVAILABLE` or `DEPENDENCY_MISSING`. The Arr lookup touches no media so it goes first: no whisper time on an untrustworthy result.""" rec: dict = {"path": str(video), "findings": []} budget = PlaybackWaitBudget(cfg.max_wait_min) try: original, status = original_language_lookup(video) rec["original_language"] = original rec["original_status"] = status if status == ORIGINAL_UNAVAILABLE: rec["result"] = "ARR_UNAVAILABLE" return rec wait_while_playing(jellyfin, cfg.jellyfin_key, cfg.max_wait_min, budget, log=log) info = probe(video) if info is None: rec["result"] = "PROBE_FAILED" return rec findings, stream_records, result, note = analyze_candidate( video, info, original, cfg, jellyfin, budget, workdir) rec["result"] = result rec["findings"] = findings rec["streams"] = stream_records if note: rec["note"] = note return rec except PlaybackWaitTimeout: raise except ImportError as exc: rec["result"] = "DEPENDENCY_MISSING" rec["error"] = f"{type(exc).__name__}: {exc}" return rec except Exception as exc: rec["result"] = "ERROR" rec["error"] = f"{type(exc).__name__}: {exc}" return rec def _is_identity(value: object) -> bool: return (isinstance(value, list) and len(value) == 3 and all(isinstance(x, int) and not isinstance(x, bool) for x in value)) def _is_str_or_none(value: object) -> bool: return value is None or isinstance(value, str) def _is_probability(value: object) -> bool: return _finite_number(value) and 0.0 <= value <= 1.0 # type: ignore[operator] def _is_index(value: object) -> bool: return isinstance(value, int) and not isinstance(value, bool) def _nonempty_str(value: object) -> bool: return isinstance(value, str) and bool(value) _STREAM_FIELDS = {"index": _is_index, "tag": _is_str_or_none, "detected_language": _nonempty_str, "probability": _is_probability} _FINDING_FIELDS: dict[str, dict] = { "MISLABELED": {"stream_index": _is_index, "tag": _is_str_or_none, "detected_language": _nonempty_str, "probability": _is_probability}, "NO_ENGLISH_AUDIO": {"original_language": _nonempty_str}, } def _finding_problem(finding: object) -> str | None: """Why `finding` is malformed, else `None`, so alert formatting never meets a bad value. MISLABELED: stream index, tag, language, probability in [0, 1]; NO_ENGLISH_AUDIO: original language and a non-empty list of ruled-out streams.""" if not isinstance(finding, dict): return "finding is not an object" kind = finding.get("type") if not isinstance(kind, str) or kind not in FINDING_TYPES: return f"unknown finding type {kind!r}" for field, ok in _FINDING_FIELDS[kind].items(): if field not in finding or not ok(finding[field]): return f"{kind}.{field} is missing or has the wrong type" if kind == "NO_ENGLISH_AUDIO": streams = finding.get("streams") if not isinstance(streams, list) or not streams: return "NO_ENGLISH_AUDIO.streams is missing or not a non-empty list" for s in streams: if not isinstance(s, dict): return "NO_ENGLISH_AUDIO.streams has a non-object item" for field, ok in _STREAM_FIELDS.items(): if field not in s or not ok(s[field]): return f"NO_ENGLISH_AUDIO.streams[].{field} is missing or has the wrong type" return None def _failed_problem(failed: object) -> str | None: if not isinstance(failed, dict): return "failed is not an object" attempts = failed.get("attempts") checks = { "identity": _is_identity(failed.get("identity")), "attempts": _is_index(attempts) and attempts >= 1, "last_result": isinstance(failed.get("last_result"), str) and failed["last_result"] in FAILED_RESULTS, "last_error": _is_str_or_none(failed.get("last_error")), } for field, ok in checks.items(): if not ok: return f"failed.{field} is missing or has the wrong type" return None def _entry_problem(entry: object) -> str | None: """Why `entry` is malformed, else `None`. Fields may be absent (a failure-only entry has no `identity`) but every present one must be valid; failure outcomes appear only inside `failed`. Keys this version does not use (module docstring) are not looked at.""" if not isinstance(entry, dict): return "entry is not an object" if "result" in entry and not (isinstance(entry["result"], str) and entry["result"] in CONCLUSION_RESULTS): return (f"result {entry['result']!r} is not a conclusion " f"({', '.join(sorted(CONCLUSION_RESULTS))})") findings = entry.get("findings", []) if not isinstance(findings, list): return "findings is not a list" for f in findings: problem = _finding_problem(f) if problem: return problem checks = { "identity": _is_identity, "first_found": lambda v: isinstance(v, str), "alerted": lambda v: isinstance(v, bool), "checked_at": lambda v: isinstance(v, str), } for field, ok in checks.items(): if field in entry and not ok(entry[field]): return f"{field} has the wrong type" if "failed" in entry: return _failed_problem(entry["failed"]) return None def entry_problem(entry: object) -> str | None: """`_entry_problem`, never raising.""" try: return _entry_problem(entry) except Exception as exc: return f"could not be validated ({type(exc).__name__})" def _load_state(path: Path) -> dict | None: """The state, `{}` if absent, `None` (logged) if unreadable or malformed; `main()` then exits 1 rather than guess, since wrong guesses lose or repeat alerts.""" if not path.exists(): return {} try: data = json.loads(path.read_text()) except Exception as exc: log(f"ERROR: {path} could not be read/parsed ({type(exc).__name__}: {exc})") return None if not isinstance(data, dict): log(f"ERROR: {path} is not a JSON object") return None for key, entry in data.items(): problem = entry_problem(entry) if problem: log(f"ERROR: {path}: malformed entry for {key!r}: {problem}") return None return data def _conclusion_entry(existing: dict, identity: list[int], rec: dict, checked_at: str) -> dict: """The whole entry for a conclusion at `identity`; anything unlisted (a `failed` record or a key of an older version too) is superseded. The same file with the same findings keeps `alerted` and `first_found`; new findings or a CHANGED identity start `alerted=False`.""" findings = rec.get("findings") or [] same_file = existing.get("identity") == identity entry: dict = {"identity": identity, "result": rec.get("result"), "findings": findings, "checked_at": checked_at} if findings: entry["first_found"] = (existing["first_found"] if same_file and isinstance(existing.get("first_found"), str) else checked_at) entry["alerted"] = bool(same_file and existing.get("findings") == findings and existing.get("alerted") is True) return entry def _record_result(video: Path, identity: list[int], rec: dict, checked_at: str) -> bool: """Record a conclusion in one locked update. `False` on a write failure or an entry the loader would reject; the file is then redone next run.""" key = str(video.resolve()) def _upsert(state: dict) -> dict: existing = state.get(key) entry = _conclusion_entry(existing if isinstance(existing, dict) else {}, identity, rec, checked_at) # Never write what the loader would reject: it would blind every later run. problem = entry_problem(entry) if problem: raise ValueError(f"refusing to record a malformed entry ({problem})") state[key] = entry return state try: mutate_state(STATE_FILE, _upsert) except Exception as exc: log(f"WARNING: {STATE_FILE} update failed for {video} " f"({type(exc).__name__}: {exc}); NOT recorded, will re-check next run") return False return True def _record_failure(video: Path, identity: list[int], rec: dict) -> None: """Count a failed analysis in a `failed` record INSIDE the entry, leaving any earlier conclusion, findings and alert flag untouched; the count restarts when the identity changes. A leftover `retry` of an older version is dropped. Write failures are logged and swallowed.""" key = str(video.resolve()) def _upsert(state: dict) -> dict: entry = state.get(key) entry = entry if isinstance(entry, dict) else {} entry.pop("retry", None) entry["failed"] = {"identity": identity, "attempts": _failed_attempts(entry, identity) + 1, "last_result": rec.get("result"), "last_error": rec.get("error")} problem = _failed_problem(entry["failed"]) if problem: raise ValueError(f"refusing to record a malformed failure record ({problem})") state[key] = entry return state try: mutate_state(STATE_FILE, _upsert) except Exception as exc: log(f"WARNING: {STATE_FILE} failure record failed for {video} " f"({type(exc).__name__}: {exc}); will retry next run") def _append_jsonl(path: Path, rec: dict) -> None: try: with path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(rec) + "\n") except OSError as exc: log(f"WARNING: {path} append failed ({type(exc).__name__}: {exc})") def _title_for(video: Path) -> str: """`" / "`, or the bare filename outside the `movies/`/`tv/` layout.""" try: rel = video.resolve().relative_to(MEDIA_ROOT.resolve()) except (ValueError, OSError): return _clean_text(video.name) parts = rel.parts if len(parts) >= 2 and parts[0].lower() in ("movies", "tv"): folder = parts[1] return _clean_text(f"{folder} / {video.stem}" if len(parts) > 2 else folder) return _clean_text(video.name) def _format_one_finding(f: dict) -> str: """`f` is a validated finding (`_finding_problem`).""" if f["type"] == "MISLABELED": return (f"stream {f['stream_index']} tagged {f['tag'] or 'und'} but detected " f"{f['detected_language']} (p={f['probability']:.2f})") heard = ", ".join(f"{s['index']} tagged {s['tag'] or 'und'} heard {s['detected_language']} " f"(p={s['probability']:.2f})" for s in f["streams"]) return f"no confirmed English audio (original language: English); streams: {heard}" def _format_finding_line(path_str: str, findings: list[dict]) -> str: """One `- title: finding; finding` line.""" body = "; ".join(_format_one_finding(f) for f in findings) return _clean_text(f"- {_title_for(Path(path_str))}: {body}") def _tg_len(text: str) -> int: """UTF-16 code units, as Telegram counts.""" return len(_clean_text(text).encode("utf-16-le")) // 2 def _truncate(text: str, limit: int) -> str: """Cut to `limit` Telegram characters, ending in an ellipsis if cut.""" text = _clean_text(text) if _tg_len(text) <= limit: return text cut = text[: max(limit - 1, 0)] while cut and _tg_len(cut) + 1 > limit: cut = cut[:-1] return cut + "…" def build_alert_message(pending: list[tuple[str, dict]]) -> tuple[str, int]: """One message naming the first items of `pending` (oldest first) and how many it names: as many as fit `MAX_ALERT_ITEMS` and `TELEGRAM_BUDGET_CHARS`. Lines are cut to `MAX_ALERT_LINE_CHARS`, so the first always fits; the caller sends the rest in further messages.""" text = "audio-lang-check: new audio-language findings" named = 0 for path_str, entry in pending[:MAX_ALERT_ITEMS]: line = _truncate(_format_finding_line(path_str, entry["findings"]), MAX_ALERT_LINE_CHARS) if _tg_len(text) + 1 + _tg_len(line) > TELEGRAM_BUDGET_CHARS: break text += "\n" + line named += 1 return text, named def _telegram_credentials() -> tuple[str, str]: """`(token, chat_id)`, either possibly empty. Environment wins (both set: `.env` not read); else `TELEGRAM_ENV` (default `~/.claude/channels/telegram/.env`, never this repo's) fills the gaps. A read failure is logged and fails soft.""" token = os.environ.get("TELEGRAM_BOT_TOKEN", "") chat = os.environ.get("TELEGRAM_CHAT_ID", "") if token and chat: return token, chat env_path = None try: env_path = Path(os.environ.get("TELEGRAM_ENV") or Path.home() / ".claude" / "channels" / "telegram" / ".env") vals = load_env_file(env_path) except Exception as exc: log(f"notify: could not read the Telegram .env ({env_path}): " f"{type(exc).__name__}: {exc}") return token, chat return token or vals.get("TELEGRAM_BOT_TOKEN", ""), chat or vals.get("TELEGRAM_CHAT_ID", "") def _notify(message: str) -> str | None: """Send `message`: `None` on success, else why not. Only the exception TYPE is reported (a message could carry the token).""" try: token, chat = _telegram_credentials() if not token or not chat: return "no Telegram credentials" if telegram_send(token, chat, _clean_text(message)): return None return "Telegram did not accept the message" except Exception as exc: return f"send raised {type(exc).__name__}" def _alert_failed(reason: str, paths: list[str]) -> None: log(f"ALERT NOT DELIVERED, NOT RETRIED ({reason}); findings for: {', '.join(paths)}", err=True) _append_jsonl(JSONL_FILE, {"event": "ALERT_FAILED", "reason": reason, "paths": paths, "at": _utc_now_iso()}) def send_alerts() -> None: """One send attempt per new finding, oldest first, in as many messages as the length bounds need. PENDING: findings present, `alerted` not true, file still at the identity they were made on, checked again before each message (a gone or replaced file's findings are never sent). Every entry named in a message is then marked `alerted`, under the lock and only if unchanged since, whether or not Telegram took the message: a failed send is logged, never retried.""" state = _load_state(STATE_FILE) if state is None: log("not alerting this run (state unreadable)") return rest = [(k, v) for k, v in state.items() if v.get("findings") and v.get("alerted") is not True and v.get("identity") is not None] rest.sort(key=lambda kv: kv[1].get("first_found") or "") attempted: dict[str, dict] = {} while True: # Checked again before every message: a file replaced while earlier ones were sending # is stale now. rest = [(k, v) for k, v in rest if _identity(Path(k)) == v["identity"]] if not rest: break text, named = build_alert_message(rest) reason = _notify(text) if reason: _alert_failed(reason, [k for k, _ in rest[:named]]) else: log(f"alert sent for {named} file(s)") attempted.update(rest[:named]) rest = rest[named:] if not attempted: return snapshot = attempted def _mark(current: dict) -> dict: for k, seen in snapshot.items(): cur = current.get(k) # a rewritten entry is a new finding, not yet alerted if (isinstance(cur, dict) and cur.get("findings") == seen.get("findings") and cur.get("identity") == seen.get("identity")): cur["alerted"] = True return current try: mutate_state(STATE_FILE, _mark) except Exception as exc: log(f"WARNING: failed to record alert delivery ({type(exc).__name__}: {exc}); " f"will alert again next run") def _process_one( video: Path, cfg: Config, jellyfin: JellyfinClient, ) -> dict | None: """`process_candidate` in a fresh workdir; `None` if `PlaybackWaitTimeout` ended it.""" WORKDIR_ROOT.mkdir(parents=True, exist_ok=True) try: with tempfile.TemporaryDirectory(prefix="audio-lang-check_", dir=WORKDIR_ROOT) as wd: return process_candidate(video, cfg, jellyfin, Path(wd)) except PlaybackWaitTimeout as exc: log(f"{exc}; ending cleanly") return None def _check_candidates(candidates: list[Path], cfg: Config) -> int: """Analyze `candidates` one at a time within the run budget and record each result. Exit status for `main()`.""" if not candidates: log("nothing to do (0 candidates)") return 0 jellyfin = JellyfinClient(cfg.jellyfin_url) deadline = time.monotonic() + cfg.budget_min * 60 started = arr_unavailable = 0 for video in candidates: if time.monotonic() >= deadline: log(f"budget ({cfg.budget_min:g} min) reached; " f"deferring {len(candidates) - started} remaining file(s) to the next run") break # Earlier candidates take minutes: the file may be gone or touched since the scan. before = _identity(video, time.time()) if before is None: continue started += 1 log(f"start ({started}/{len(candidates)}): {video}") rec = _process_one(video, cfg, jellyfin) if rec is None: return 0 if _identity(video, time.time()) != before: log(f"DISCARDED (file changed or was touched during analysis): {video}") continue checked_at = _utc_now_iso() rec["checked_at"] = checked_at result = rec.get("result") if result == "ARR_UNAVAILABLE": arr_unavailable += 1 continue if result == "DEPENDENCY_MISSING": _append_jsonl(JSONL_FILE, rec) log("ERROR: faster_whisper is not installed; every file would fail the " "same way -- ending the run") return 1 if result in FAILED_RESULTS: _record_failure(video, before, rec) _append_jsonl(JSONL_FILE, rec) log(f"{result} (at most {MAX_FAILED_ATTEMPTS} attempts per file version): {video}" + (f" [{rec['error']}]" if rec.get("error") else "")) continue if not _record_result(video, before, rec, checked_at): continue _append_jsonl(JSONL_FILE, rec) log(f"{result} ({len(rec['findings'])} finding(s)): {video}") if arr_unavailable: log(f"{arr_unavailable} file(s) not checked: Radarr/Sonarr could not be asked; " f"the next run tries again") log(f"done: {started}/{len(candidates)} file(s) started this run") return 0 def main() -> int: signal.signal(signal.SIGTERM, _handle_sigterm) ap = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--dry-run", action="store_true", help="list candidates and do nothing else (no ASR, no state writes, " "no Telegram); cannot be combined with --path") ap.add_argument("--path", metavar="FILE", help="check a single file and print the result as JSON (still honours " "the playback gate); records NOTHING: no state, no findings, no " "Telegram, so it can never suppress the cron job's own alert") args = ap.parse_args() if args.dry_run and args.path: ap.error("--dry-run cannot be combined with --path (--path analyzes the file; " "--dry-run promises not to)") global _LOG_TO_STDERR _LOG_TO_STDERR = bool(args.path) lock_fp = None if not args.dry_run: ensure_var_dirs() lock_fp = acquire_singleton_lock(LOCK_FILE) if lock_fp is None: if args.path: print(json.dumps({ "path": str(Path(args.path).resolve()), "result": "BUSY", "reason": "another audio-lang-check run holds the lock"}, indent=2)) return EXIT_BUSY return 0 # cron: another instance is running; exit quietly, not an error env = {**load_env_file(MEDIA_STACK_ROOT / ".env"), **os.environ} cfg = load_config(env) real_run = not args.dry_run if cfg.whisper_lid_model.endswith(".en"): log(f"ERROR: WHISPER_LID_MODEL={cfg.whisper_lid_model!r} is an English-only model " f"and cannot detect language -- it would always report 'en' with high confidence, " f"silently defeating the whole point of this check. Use a multilingual model " f"(e.g. 'small', 'base', 'medium').") if real_run: return 1 if real_run and not cfg.jellyfin_key: log("ERROR: JELLYFIN_API_KEY is not set; a real run cannot fail closed on " "playback without it -- refusing to start blinded. --dry-run still works.") return 1 if args.path: video = Path(args.path).resolve() if not video.is_file(): log(f"ERROR: --path {video} is not a file") return 1 rec = _process_one(video, cfg, JellyfinClient(cfg.jellyfin_url)) if rec is None: print(json.dumps({"path": str(video), "result": "PLAYBACK_WAIT_TIMEOUT"}, indent=2)) return 0 # Never recorded: the cron run would skip this file and suppress its alert. print(json.dumps(rec, indent=2)) return 0 state = _load_state(STATE_FILE) if state is None: log("run blinded: the state is unreadable or malformed (nothing was written)") return 1 candidates = select_candidates(state, MEDIA_ROOT, time.time()) if args.dry_run: log(f"DRY RUN: {len(candidates)} candidate(s) would be checked " f"(nothing probed, nothing transcribed, nothing written)") for p in candidates: log(f" would check: {p}") return 0 try: return _check_candidates(candidates, cfg) finally: try: send_alerts() except Exception as exc: log(f"WARNING: alerting failed ({type(exc).__name__}: {exc})", err=True) if __name__ == "__main__": sys.exit(main())