"""ASS tag/style utilities and re-exports of media-probe helpers. This module focuses on in-file ASS operations: - classify_styles — split styles into dialog vs. signs/songs using patterns + positional-tag heuristic - extract_tags / restore_tags — preserve override tags across translation - fix_line_breaks — repair line break syntax that translation may mangle The media-probe side (ffprobe / get_media_streams / stream selection / extract_subtitle_stream) lives in ``ass_probe.py`` and is re-exported here so ``from ass_utils import run_ffprobe / get_media_streams / …`` continues to work unchanged. """ import logging import re from ass_drawing import holds_language from ass_lexer import lex from ass_probe import ( # noqa: F401 — re-exported for back-compat extract_subtitle_stream, get_all_subtitle_streams, get_legacy_subtitle_stream_output_path, get_media_streams, get_subtitle_stream_output_path, has_target_language_audio, has_target_language_stream, run_ffprobe, select_best_subtitle_stream, ) logger = logging.getLogger(__name__) # Patterns for style classification. # # Deliberately narrow on the OP/ED front. Anything landing in the signs bucket # is not only left untranslated but also visible to cleanup_signs and # purge_signs_after_extract, which DELETE what they consider signs — so a # fansub style named after a character ("Ed", "Ed Smith") must never match. # Hence: bare "op"/"ed" only as the whole name, plus the unambiguous karaoke # pairing ("OP Romaji", "ED English"), plus romaji anywhere — transliterated # Japanese is never something to hand a translator. SIGNS_PATTERNS = re.compile( r"sign|^op$|^ed$|romaji" r"|^(?:op|ed)[\s_.-]+(?:english|eng|kanji|jp|jpn|lyrics|karaoke|roma?)\b" # "title" but not "Subtitle": that is a dialogue style (prod 2026-11-08). r"|song|karaoke|(?80% of lines have \pos() or \move() tags lines = style_lines[style_name] if lines: pos_count = sum(1 for line in lines if POS_MOVE_RE.search(line)) if pos_count / len(lines) > 0.8: signs_styles.add(style_name) continue # Default: treat as dialog dialog_styles.add(style_name) logger.info( "Style classification - Dialog: %s, Signs/Songs: %s", dialog_styles, signs_styles, ) return dialog_styles, signs_styles #: The reading of ``\p`` tags lives in ``ass_drawing`` so the security #: sanitizer and this module cannot drift apart again. They did: the sanitizer #: read ``{\P1}`` as an opener and deleted the visible dialogue behind it. def text_outside_drawing(text): """Return only the parts of an ASS event that are not drawing commands. Inside a ``{\\pN}`` span the "text" is vector geometry — ``m 78.43 15.3 b 77.04 …`` — which carries no language. Prod 2026-09-17 spent twelve minutes of one automation tick translating and quality-retrying such events, because :func:`extract_tags` strips the ``{\\p1}`` that identifies them and leaves the coordinates looking like ordinary text. Drawing mode is a per-event affair: override tags reset at every event, so an opener without a closer runs to the end of the line rather than bleeding into the next one. An unclosed opener is the common real shape, which is why its remainder counts as drawing rather than as text. Args: text: Raw ASS event text, override tags included Returns: The concatenated text outside every drawing span, tags removed """ if not text: return "" return "".join(text[start:end] for start, end, drawing in _text_runs(text) if not drawing) def _text_runs(text): """Split an event into its non-tag stretches, each labelled drawing or not. Returns a list of ``(start, end, in_drawing)`` offsets into ``text``. Tag blocks themselves are not returned — only the stretches between them, which is where an event's content lives. Drawing mode is a per-event affair: override tags reset at every event, so an opener without a closer runs to the end of this line rather than bleeding into the next one. """ tokens, _closed = lex(text) return [(t.start, t.end, t.drawing) for t in tokens if t.kind == "content"] def contains_drawing(text): """Whether any part of this event's content renders as vector geometry.""" return any(drawing for _start, _end, drawing in _text_runs(text)) def split_around_drawings(text): """Split a mixed event into (prefix, translatable body, suffix). The prefix and suffix are kept byte for byte — they hold the drawing and the tags that switch it on and off, and neither survives a round trip through a translation model. The body spans from the first piece of dialogue to the last, tags included, so a caption with one italic word stays one translatable sentence rather than three fragments the caller would have to drop. Returns None when the event cannot be put back together afterwards, which is the case when a drawing sits *between* two pieces of dialogue: one translated string cannot be mapped onto two separate runs, and guessing the split point corrupts both. Sandbox VM, 1.14.4-rc.9: without this, ``text_outside_drawing`` was only consulted as a yes/no filter and ``extract_tags`` still received the whole original, so ``{\\p1}m 0 0 l 10 10{\\p0}Good morning.`` sent the coordinates to the model, the restored tags landed inside the German sentence, and the sanitizer cut it down to ``" Morgen."``. """ runs = _text_runs(text) body_runs = [ (start, end) for start, end, drawing in runs if not drawing and holds_language(text[start:end]) ] if not body_runs: return None start = body_runs[0][0] end = body_runs[-1][1] if any(drawing and run_start < end and run_end > start for run_start, run_end, drawing in runs): return None return text[:start], text[start:end], text[end:] def extract_tags(text): """Extract ASS override tags from text, return clean text, tag info, and original length. Args: text: ASS dialog text with potential override tags Returns: tuple: (clean_text, tag_info, original_clean_length) where tag_info is a list of (position, tag_string) tuples for restoration """ if not OVERRIDE_TAG_RE.search(text): return text, [], len(text) tag_info = [] parts = OVERRIDE_TAG_RE.split(text) tags = OVERRIDE_TAG_RE.findall(text) clean_parts = [] current_clean_pos = 0 for i, part in enumerate(parts): if i > 0 and i - 1 < len(tags): tag_info.append((current_clean_pos, tags[i - 1])) clean_parts.append(part) current_clean_pos += len(part) clean_text = "".join(clean_parts) return clean_text, tag_info, len(clean_text) def restore_tags(translated_text, tag_info, original_clean_length=None): """Restore ASS override tags into translated text using proportional positioning. Position-0 tags (prefix) stay at the beginning. Other tags are placed proportionally based on original/translated length ratio, snapped to the nearest word boundary within +/- 3 characters. Args: translated_text: Translated text without tags tag_info: List of (position, tag_string) from extract_tags() original_clean_length: Length of original clean text (for proportional calc) Returns: Text with override tags restored """ if not tag_info: return translated_text result = [] text_pos = 0 trans_len = len(translated_text) orig_len = original_clean_length or trans_len sorted_tags = sorted(tag_info, key=lambda x: x[0]) for pos, tag in sorted_tags: if pos == 0: insert_pos = 0 elif orig_len > 0: ratio = pos / orig_len insert_pos = int(ratio * trans_len) # Snap to nearest word boundary within +/- 3 chars best = insert_pos for offset in range(-3, 4): check = insert_pos + offset if 0 <= check <= trans_len: if check == trans_len or translated_text[check] in (" ", "\\"): best = check break insert_pos = best else: insert_pos = min(pos, trans_len) # Never go backwards insert_pos = max(insert_pos, text_pos) insert_pos = min(insert_pos, trans_len) if insert_pos > text_pos: result.append(translated_text[text_pos:insert_pos]) text_pos = insert_pos result.append(tag) if text_pos < trans_len: result.append(translated_text[text_pos:]) return "".join(result) def fix_line_breaks(text): """Fix line breaks after translation. The model sometimes converts \\N to \\n or literal newlines. """ text = text.replace("\n", "\\N") text = re.sub(r"(?