backend/subtitle_dialogue.py
Rama master · réplica de lectura del repositorio original.
"""Is a subtitle plausibly the full dialogue track, or only signs?
Prod 2026-11 (Slime S03E24): AnimeTosho's "English Signs [Official]" track,
7 events in style "Default" with actor "Sign", passed every check. The style
classifier reads "Default" as dialogue, and the cues spanned 68 % of the
episode, so the coverage check passed too. The file replaced a 405-cue SRT
as an upgrade and was translated to a "German subtitle" without one line of
dialogue.
This module counts what a viewer would read as dialogue and judges that
count against the video length. The thresholds are deliberately low: a sparse
but genuine subtitle (a near-silent episode, a three-minute short) must pass,
because the verdict blocks downloads and translations unattended.
"""
from __future__ import annotations
import logging
import os
import re
import threading
from dataclasses import dataclass
logger = logging.getLogger(__name__)
#: Fewer dialogue lines than this are never a full subtitle — unless the video
#: is known to be shorter than MIN_RATED_DURATION_S (a one-minute short may
#: hold seven lines; only an empty one is judged there).
MIN_DIALOGUE_EVENTS = 10
#: With a known length of at least MIN_RATED_DURATION_S: fewer than
#: RATED_DIALOGUE_EVENTS lines AND fewer than MIN_EVENTS_PER_MINUTE is sparse.
#: A 24-minute episode holds 250-400 lines; on the reference library every
#: full-named sidecar under 40 lines for such an episode was a signs track.
#: The 40 matches the dubtitle detector's _MIN_FULLTEXT_CUES.
RATED_DIALOGUE_EVENTS = 40
MIN_EVENTS_PER_MINUTE = 2.0
MIN_RATED_DURATION_S = 180.0
#: A replacement holding less than this share of the old file's lines lost
#: the dialogue — if the old file had at least DROP_MIN_OLD_EVENTS (below
#: that a few lines more or less are noise, not a lost track).
DROP_MAX_SHARE = 0.25
DROP_MIN_OLD_EVENTS = 8
# Actor names typesetters use for sign events. Anchored: a character named
# "Signe" or "Tsuki" must not match.
_SIGN_ACTOR = re.compile(r"^\s*(?:signs?|typeset(?:ting)?|ts)\s*$", re.IGNORECASE)
# No nested "{": an unclosed block stops at the next one, keeping this linear.
_CLOSED_BLOCK = re.compile(r"\{[^{}]*\}")
@dataclass(frozen=True)
class DialogueAssessment:
dialogue_events: int
total_events: int
per_minute: float | None
sparse_reason: str | None
def _plain(text: str) -> str:
"""The readable text: tags and drawing geometry removed (linear lexer).
The lexer refuses malformed events (an unclosed override block); those fall
back to dropping closed blocks, which still counts the text they hold.
"""
from ass_utils import text_outside_drawing
try:
visible = text_outside_drawing(text)
except Exception: # noqa: BLE001 — ASSNeedsReviewError and friends: count, don't fail
visible = _CLOSED_BLOCK.sub("", text)
return visible.replace("\\N", " ").replace("\\n", " ").replace("\\h", " ").strip()
def _visible_events(subs, dialog_styles):
"""Yield (event, plain text, is_dialogue) for every event a viewer sees."""
for event in subs.events:
if event.is_comment:
continue
plain = _plain(event.text)
if not plain:
continue
is_dialogue = (
dialog_styles is None or event.style in dialog_styles
) and not _SIGN_ACTOR.match(event.name or "")
yield event, plain, is_dialogue
def _dialogue_lines(subs, dialog_styles) -> tuple[int, int]:
"""(dialogue lines, visible events): layers and repeats counted once."""
seen: set[tuple[int, int, str]] = set()
visible = 0
for event, plain, is_dialogue in _visible_events(subs, dialog_styles):
visible += 1
if is_dialogue:
seen.add((event.start, event.end, plain))
return len(seen), visible
def assess(subs, duration_s: float | None, *, is_ass: bool) -> DialogueAssessment:
"""Judge a parsed subtitle. ``is_ass`` selects style-aware counting."""
dialog_styles = None
if is_ass:
from ass_utils import classify_styles
dialog_styles, _signs = classify_styles(subs)
lines, visible = _dialogue_lines(subs, dialog_styles)
return judge(lines, visible, duration_s)
def judge(lines: int, visible: int, duration_s: float | None) -> DialogueAssessment:
"""The sparse verdict for a dialogue-line count against the video length.
Shared by :func:`assess` and the sidecar check, which caches the counts.
"""
per_minute = lines / (duration_s / 60) if duration_s and duration_s > 0 else None
reason = None
short = duration_s is not None and 0 < duration_s < MIN_RATED_DURATION_S
if lines == 0 or (lines < MIN_DIALOGUE_EVENTS and not short):
reason = f"{lines} dialogue lines of {visible} events"
elif (
per_minute is not None
and duration_s >= MIN_RATED_DURATION_S
and lines < RATED_DIALOGUE_EVENTS
and per_minute < MIN_EVENTS_PER_MINUTE
):
reason = f"{lines} dialogue lines in {duration_s / 60:.0f} min ({per_minute:.1f}/min)"
return DialogueAssessment(lines, visible, per_minute, reason)
def sparse_source_reason(subs, video_path: str | None, *, is_ass: bool) -> str | None:
"""Why ``subs`` is no full source for translating ``video_path``, or None.
Judged only against a known video length: without one there is no way to
tell a signs track from a short. Never raises — a check that cannot run
must not block a translation.
"""
try:
import subtitle_plausibility
duration = subtitle_plausibility.video_duration_s(video_path) if video_path else None
if not duration:
return None
return assess(subs, duration, is_ass=is_ass).sparse_reason
except Exception as exc: # noqa: BLE001
logger.debug("Sparse-source check skipped for %s: %s", video_path, exc)
return None
def _parse(content: bytes):
"""(subs, is_ass) or None when the content is no readable subtitle."""
if not content:
return None
try:
import pysubs2
from subtitle_encoding import ensure_utf8, is_wide_unicode
if is_wide_unicode(content):
content = ensure_utf8(content)
text = content.decode("utf-8-sig", errors="replace")
subs = pysubs2.SSAFile.from_string(text)
except Exception: # noqa: BLE001 — unreadable: no verdict, other checks decide
return None
if not subs.events:
return None
return subs, subs.format in ("ass", "ssa")
def assess_content(content: bytes, duration_s: float | None) -> DialogueAssessment | None:
"""Judge raw subtitle bytes; None when they cannot be read."""
parsed = _parse(content)
if parsed is None:
return None
subs, is_ass = parsed
try:
return assess(subs, duration_s, is_ass=is_ass)
except Exception as exc: # noqa: BLE001 — no verdict rather than a failed download
logger.debug("Dialogue assessment skipped: %s", exc)
return None
def dialogue_drop(old_content: bytes, new_content: bytes) -> str | None:
"""Why ``new_content`` would lose the dialogue ``old_content`` holds, or None.
Cue counts differ between releases (merged or split lines), so only a
collapse counts: the new file keeps under a quarter of the old lines.
"""
old = assess_content(old_content, None)
new = assess_content(new_content, None)
if old is None or new is None or old.dialogue_events < DROP_MIN_OLD_EVENTS:
return None
if new.dialogue_events < old.dialogue_events * DROP_MAX_SHARE:
return (
f"replacement holds {new.dialogue_events} dialogue lines, "
f"the existing subtitle {old.dialogue_events}"
)
return None
#: Largest sidecar read for the on-disk check; a real subtitle is far smaller.
_MAX_SIDECAR_BYTES = 8 * 1024 * 1024
#: One wanted cleanup reads every sidecar of the library (prod: ~10,000), so
#: the cache must hold all of them or each pass parses them again (rc.2:
#: 172 s instead of 59 s). An entry is a few hundred bytes.
_COUNT_CACHE_MAX = 65_536
_count_cache: dict[tuple[str, int, int], tuple[int, int] | None] = {}
_count_lock = threading.Lock()
def _sidecar_counts(path: str) -> tuple[int, int] | None:
"""(dialogue lines, visible events) of a sidecar, cached per path, mtime and size."""
try:
st = os.stat(path)
except OSError:
return None
key = (path, st.st_mtime_ns, st.st_size)
with _count_lock:
if key in _count_cache:
return _count_cache[key]
counts = None
if 0 < st.st_size <= _MAX_SIDECAR_BYTES:
try:
with open(path, "rb") as fh:
parsed = _parse(fh.read())
if parsed is not None:
subs, is_ass = parsed
assessment = assess(subs, None, is_ass=is_ass)
counts = (assessment.dialogue_events, assessment.total_events)
except Exception as exc: # noqa: BLE001 — no verdict, never a failed scan
logger.debug("Sparse sidecar check skipped for %s: %s", path, exc)
with _count_lock:
if len(_count_cache) >= _COUNT_CACHE_MAX:
# Drop the oldest eighth, not everything: clearing at the limit
# threw away the entries the same pass had just filled.
for old in list(_count_cache)[: max(1, _COUNT_CACHE_MAX // 8)]:
del _count_cache[old]
_count_cache[key] = counts
return counts
def sparse_sidecar_reason(sidecar_path: str | None, video_path: str | None) -> str | None:
"""Why the full sidecar on disk is signs/credits only, or None.
Prod 2026-11: a German ``.de.ass`` of 10 title-card and credit lines for a
24-minute episode counted as the full subtitle, so the episode was never
searched again. Judged only against a known video length (the cached
stream probe): without one a signs track cannot be told from a short.
Only for full sidecars — the caller passes ``.de.ass``/``.de.srt``, never
a ``.forced`` or signs file. Never raises.
"""
if not sidecar_path or not video_path:
return None
try:
counts = _sidecar_counts(sidecar_path)
if counts is None:
return None
lines, visible = counts
if lines >= RATED_DIALOGUE_EVENTS:
return None # never sparse; skip the duration lookup
import subtitle_plausibility
duration = subtitle_plausibility.video_duration_s(video_path)
if not duration:
return None
return judge(lines, visible, duration).sparse_reason
except Exception as exc: # noqa: BLE001
logger.debug("Sparse sidecar check skipped for %s: %s", sidecar_path, exc)
return None
def dialogue_starts(content: bytes) -> list[float] | None:
"""Sorted start times (seconds) of the dialogue lines; None when unreadable.
Signs, songs and typesetting are left out the way :func:`assess` leaves
them out, so two releases of the same episode compare on what is spoken.
"""
parsed = _parse(content)
if parsed is None:
return None
subs, is_ass = parsed
try:
dialog_styles = None
if is_ass:
from ass_utils import classify_styles
dialog_styles, _signs = classify_styles(subs)
starts = {
event.start
for event, _plain, is_dialogue in _visible_events(subs, dialog_styles)
if is_dialogue
}
except Exception as exc: # noqa: BLE001 — no timing verdict rather than a failed download
logger.debug("Dialogue start times skipped: %s", exc)
return None
return [ms / 1000.0 for ms in sorted(starts)]