backend/common_fixes.py
Rama master · réplica de lectura del repositorio original.
# backend/common_fixes.py
"""All common subtitle fix functions.
apply_common_fixes(subs, options) applies enabled fixes in sequence and returns
a list of Change objects. subs is a pysubs2.SSAFile modified in place.
"""
from __future__ import annotations
import re
import textwrap
import pysubs2
from subtitle_types import Change, _ms_to_str
_WATERMARK_PATTERNS = [
re.compile(r"downloaded\s+from", re.IGNORECASE),
re.compile(r"synced\s+by", re.IGNORECASE),
re.compile(r"subtitles\s+by", re.IGNORECASE),
re.compile(r"www\.", re.IGNORECASE),
re.compile(r"opensubtitles", re.IGNORECASE),
re.compile(r"subscene", re.IGNORECASE),
re.compile(r"subf2m", re.IGNORECASE),
re.compile(r"yifysubtitles", re.IGNORECASE),
]
_OVERLAP_IS_INTENTIONAL_FORMATS = ("ass", "ssa")
def apply_common_fixes(
subs: pysubs2.SSAFile, options: dict, fmt: str | None = None
) -> list[Change]:
"""Apply all enabled common fixes to subs (in place). Return list of Changes.
``fmt`` is the file's format ("srt", "ass", "ssa"); timing fixes that only
make sense for plain-text cues are skipped for ASS/SSA.
"""
defaults = {
"utf8_conversion": True,
"linebreak_normalization": True,
"whitespace_cleanup": True,
"empty_event_removal": True,
"ocr_fixes": False,
"fix_uppercase": False,
"watermark_removal": True,
"quote_normalization": True,
"apostrophe_normalization": True,
"em_dash_correction": True,
"crocodile_brackets": True,
"multiple_spaces": True,
"interpunction_spacing": True,
"consecutive_dots": True,
"long_line_wrap": False,
"wrap_chars": 42,
"short_line_merge": False,
"overlap_fix": True,
"min_display_time": False,
"min_ms": 800,
}
opts = {**defaults, **options}
changes: list[Change] = []
# Determine if fix_uppercase should apply: >60% of non-empty lines are all-caps
_should_fix_uppercase = False
if opts["fix_uppercase"]:
non_empty = [e.plaintext.strip() for e in subs.events if e.plaintext.strip()]
if non_empty:
caps_count = sum(1 for t in non_empty if t == t.upper() and any(c.isalpha() for c in t))
_should_fix_uppercase = (caps_count / len(non_empty)) > 0.6
to_remove: list[int] = []
for i, event in enumerate(subs.events):
if event.is_comment:
continue
original = event.text
text = original
# Watermark removal
if opts["watermark_removal"]:
plain = event.plaintext.strip()
if any(p.search(plain) for p in _WATERMARK_PATTERNS):
ts = f"{_ms_to_str(event.start)} --> {_ms_to_str(event.end)}"
changes.append(Change(i, ts, plain, "", "common_fixes"))
to_remove.append(i)
continue
# Em-dash correction: -- → —
if opts["em_dash_correction"]:
text = text.replace("--", "\u2014")
# Quote normalization: curly → straight
if opts["quote_normalization"]:
text = text.replace("\u201c", '"').replace("\u201d", '"')
text = text.replace("\u2018", "'").replace("\u2019", "'")
# Apostrophe normalization: '' → "
if opts["apostrophe_normalization"]:
text = text.replace("''", '"')
# Crocodile brackets: >> at line start
if opts["crocodile_brackets"]:
text = re.sub(r"^>>\s*", "", text, flags=re.MULTILINE)
# Multiple spaces
if opts["multiple_spaces"]:
text = re.sub(r" +", " ", text)
# Consecutive dots: .... → ...
if opts["consecutive_dots"]:
text = re.sub(r"\.{4,}", "...", text)
# Interpunction spacing: remove space before , . ! ?
if opts["interpunction_spacing"]:
text = re.sub(r" ([,\.!\?])", r"\1", text)
# Fix uppercase: apply sentence case
if _should_fix_uppercase:
plain = event.plaintext.strip()
if plain and plain == plain.upper() and any(c.isalpha() for c in plain):
text = ". ".join(s.strip().capitalize() for s in plain.split(". "))
# Long line wrap
if opts["long_line_wrap"]:
wrap_chars = opts["wrap_chars"]
plain = event.plaintext.strip()
if len(plain) > wrap_chars:
wrapped = textwrap.fill(plain, width=wrap_chars, break_long_words=False)
# Use \n as line separator (works for both SRT and ASS)
text = wrapped.replace("\n", "\n")
# Record ONE change per event with the final modified_text
if text != original:
ts = f"{_ms_to_str(event.start)} --> {_ms_to_str(event.end)}"
plain_original = event.plaintext.strip()
event.text = text
plain_modified = event.plaintext.strip()
changes.append(Change(i, ts, plain_original, plain_modified, "common_fixes"))
# Remove watermarked events in reverse order
for i in sorted(to_remove, reverse=True):
del subs.events[i]
# Short line merge: combine adjacent short events (within 500ms gap, both under 20 chars)
if opts["short_line_merge"]:
i = 0
while i < len(subs.events) - 1:
a = subs.events[i]
b = subs.events[i + 1]
gap = b.start - a.end
if gap <= 500 and len(a.plaintext.strip()) < 20 and len(b.plaintext.strip()) < 20:
ts = f"{_ms_to_str(a.start)} --> {_ms_to_str(a.end)}"
merged_text = a.text.rstrip() + "\n" + b.text.lstrip()
changes.append(Change(i, ts, a.plaintext.strip(), merged_text, "common_fixes"))
a.text = merged_text
a.end = b.end
del subs.events[i + 1]
else:
i += 1
# Overlap fix: adjust end time of event[i] when it overlaps event[i+1].
# Not for ASS/SSA — overlaps there are deliberate (karaoke syllables, layered
# signs), and clamping them collapses typeset releases to zero-duration events.
if opts["overlap_fix"] and fmt not in _OVERLAP_IS_INTENTIONAL_FORMATS:
for i in range(len(subs.events) - 1):
if subs.events[i].end > subs.events[i + 1].start:
original_end = subs.events[i].end
subs.events[i].end = subs.events[i + 1].start - 1
ts = f"{_ms_to_str(subs.events[i].start)} --> {_ms_to_str(original_end)}"
changes.append(
Change(
i,
ts,
f"end={original_end}ms",
f"end={subs.events[i].end}ms",
"common_fixes",
)
)
# Minimum display time
if opts["min_display_time"]:
min_ms = opts["min_ms"]
for i, event in enumerate(subs.events):
duration = event.end - event.start
if duration < min_ms:
ts = f"{_ms_to_str(event.start)} --> {_ms_to_str(event.end)}"
changes.append(
Change(
i,
ts,
f"duration={duration}ms",
f"duration={min_ms}ms",
"common_fixes",
)
)
event.end = event.start + min_ms
return changes