| """ |
| subtitle_parser.py β Subtitle Parsing & Writing |
| Handles .srt / .vtt / .ass via pysubs2. |
| """ |
|
|
| import copy |
| import logging |
| import os |
| import re |
| import shutil |
| import tempfile |
|
|
| import pysubs2 |
|
|
| logger = logging.getLogger(__name__) |
|
|
| WORKSPACE_DIR = os.path.join(tempfile.gettempdir(), "subsync_workspace") |
| os.makedirs(WORKSPACE_DIR, exist_ok=True) |
|
|
| |
| SKIP_STYLES = { |
| "EDR", "Romaji", "JP", "Karaoke", "KAR", |
| "Signs", "Lyrics_JP", "Romaji2", "OP_Romaji", "ED_Romaji", |
| } |
|
|
| _EXT_MAP = {".srt": "srt", ".vtt": "vtt", ".ass": "ass", ".ssa": "ass"} |
|
|
| _ASS_TAG_RE = re.compile(r"\{[^}]*\}") |
| _HTML_TAG_RE = re.compile(r"<[^>]+>") |
| _LEADING_TAGS_RE = re.compile(r"^(\{[^}]*\})+") |
|
|
|
|
| def is_skip_style(style_name: str) -> bool: |
| """Return True if this ASS style should be skipped (not translated).""" |
| if not style_name: |
| return False |
| s = style_name.strip() |
| if s in SKIP_STYLES: |
| return True |
| sl = s.lower() |
| return any(k in sl for k in ["romaji", "karaoke", "kar_", "_kar", "furigana"]) |
|
|
|
|
| def _detect_format(filename: str) -> str: |
| ext = os.path.splitext(filename.lower())[1] |
| fmt = _EXT_MAP.get(ext) |
| if not fmt: |
| raise ValueError( |
| f"'{ext}' format ααα― support ααα―ααΊαα« β .srt / .vtt / .ass αα¬ αααΊααΆαααΊ" |
| ) |
| return fmt |
|
|
|
|
| def _strip_tags(text: str) -> str: |
| text = _ASS_TAG_RE.sub("", text) |
| text = _HTML_TAG_RE.sub("", text) |
| return text.strip() |
|
|
|
|
| def parse_subtitle(file_bytes: bytes, filename: str): |
| """Parse subtitle bytes β (SSAFile, fmt_string).""" |
| fmt = _detect_format(filename) |
|
|
| text = None |
| for enc in ("utf-8-sig", "utf-8", "latin-1", "cp1252"): |
| try: |
| text = file_bytes.decode(enc) |
| break |
| except (UnicodeDecodeError, LookupError): |
| continue |
| if text is None: |
| raise ValueError("ααα―ααΊ encoding αααΊα αααα«") |
|
|
| text = text.lstrip("\ufeff") |
|
|
| try: |
| subs = pysubs2.SSAFile.from_string(text) |
| logger.info("Parsed %d events from '%s' (%s)", len(subs), filename, fmt) |
| return subs, fmt |
| except Exception as e1: |
| logger.warning("from_string failed: %s", e1) |
|
|
| try: |
| suffix = os.path.splitext(filename)[1] or ("." + fmt) |
| fd, tmp = tempfile.mkstemp(suffix=suffix) |
| try: |
| os.write(fd, file_bytes) |
| finally: |
| os.close(fd) |
| subs = pysubs2.SSAFile.load(tmp) |
| os.unlink(tmp) |
| return subs, fmt |
| except Exception as e2: |
| raise ValueError( |
| f"Parse αααα«:\nAttempt 1: {e1}\nAttempt 2: {e2}" |
| ) from e2 |
|
|
|
|
| def get_plain_lines(ssafile) -> list: |
| """ |
| Return list[str] same length as ssafile.events. |
| - Comment events β "" (keep original, not displayed) |
| - Skip styles β "" (EDR/romaji/karaoke etc.) |
| - Normal events β clean text (tags stripped) |
| Empty string β replace_lines keeps original text unchanged. |
| """ |
| lines = [] |
| for event in ssafile.events: |
| |
| if getattr(event, "is_comment", False): |
| lines.append("") |
| continue |
|
|
| |
| if is_skip_style(event.style): |
| lines.append("") |
| continue |
|
|
| raw = event.text.replace("\\N", "\n").replace("\\n", "\n") |
| clean = _strip_tags(raw) |
| lines.append(clean) |
| return lines |
|
|
|
|
| def replace_lines(ssafile, new_lines: list): |
| """ |
| Return new SSAFile with translated text. Timestamps never touched. |
| Leading ASS override tags {\\pos(...)\\be5} preserved from original. |
| Empty translation β keep original event.text. |
| """ |
| n_events = len(ssafile.events) |
| n_lines = len(new_lines) |
|
|
| if n_lines != n_events: |
| raise ValueError( |
| f"TIMESTAMP SAFETY: translated lines ({n_lines}) " |
| f"!= subtitle events ({n_events})" |
| ) |
|
|
| result = copy.deepcopy(ssafile) |
|
|
| for i, event in enumerate(result.events): |
| translated = str(new_lines[i]).strip() |
|
|
| |
| if not translated: |
| continue |
|
|
| |
| original_text = ssafile.events[i].text |
| leading_m = _LEADING_TAGS_RE.match(original_text) |
| tag_prefix = leading_m.group(0) if leading_m else "" |
|
|
| event.text = tag_prefix + translated.replace("\n", "\\N") |
|
|
| return result |
|
|
|
|
| def write_subtitle(ssafile, fmt: str) -> bytes: |
| try: |
| return ssafile.to_string(fmt).encode("utf-8") |
| except Exception as exc: |
| raise RuntimeError(f"Subtitle write αααα«: {exc}") from exc |
|
|
|
|
| def output_filename(original_name: str) -> str: |
| base, ext = os.path.splitext(original_name) |
| return f"{base}.my{ext}" |
|
|
|
|
| def clean_workspace() -> None: |
| try: |
| if os.path.exists(WORKSPACE_DIR): |
| shutil.rmtree(WORKSPACE_DIR) |
| os.makedirs(WORKSPACE_DIR, exist_ok=True) |
| except Exception as exc: |
| raise RuntimeError(f"Workspace ααΎααΊαΈαααα«: {exc}") from exc |
|
|
|
|
| def workspace_size_mb() -> float: |
| total = 0 |
| for dirpath, _, filenames in os.walk(WORKSPACE_DIR): |
| for f in filenames: |
| try: |
| total += os.path.getsize(os.path.join(dirpath, f)) |
| except OSError: |
| pass |
| return round(total / (1024 * 1024), 2) |
|
|