mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-15 05:25:12 +00:00
imSp4rky
2dc458f8b4
Live-to-VOD replays carry cue times on the live media clock, which overflowed the SRT conversion and dropped every cue.
1183 lines
49 KiB
Python
1183 lines
49 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import subprocess
|
|
import time
|
|
from collections import defaultdict
|
|
from enum import Enum
|
|
from functools import partial
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterable, Optional, Sequence, Union
|
|
|
|
import click
|
|
import pycaption
|
|
import requests
|
|
from construct import Container
|
|
from langcodes import Language
|
|
from pycaption import Caption, CaptionList, CaptionNode, WebVTTReader
|
|
from pycaption.geometry import Layout
|
|
from pymp4.parser import MP4
|
|
from subby import CommonIssuesFixer, SAMIConverter, SDHStripper
|
|
from subtitle_filter import Subtitles
|
|
|
|
from unshackle.core import binaries
|
|
from unshackle.core.config import config
|
|
from unshackle.core.tracks.track import Track
|
|
from unshackle.core.utilities import try_ensure_utf8, with_script
|
|
from unshackle.core.utils.subprocess import log_tool_run
|
|
from unshackle.core.utils.webvtt import merge_segmented_webvtt, rebase_webvtt
|
|
|
|
# silence srt library INFO logging
|
|
logging.getLogger("srt").setLevel(logging.ERROR)
|
|
|
|
|
|
class Subtitle(Track):
|
|
class Codec(str, Enum):
|
|
SubRip = "SRT" # https://wikipedia.org/wiki/SubRip
|
|
SubStationAlpha = "SSA" # https://wikipedia.org/wiki/SubStation_Alpha
|
|
SubStationAlphav4 = "ASS" # https://wikipedia.org/wiki/SubStation_Alpha#Advanced_SubStation_Alpha=
|
|
TimedTextMarkupLang = "TTML" # https://wikipedia.org/wiki/Timed_Text_Markup_Language
|
|
WebVTT = "VTT" # https://wikipedia.org/wiki/WebVTT
|
|
SAMI = "SMI" # https://wikipedia.org/wiki/SAMI
|
|
MicroDVD = "SUB" # https://wikipedia.org/wiki/MicroDVD
|
|
MPL2 = "MPL2" # MPL2 subtitle format
|
|
TMP = "TMP" # TMP subtitle format
|
|
# MPEG-DASH box-encapsulated subtitle formats
|
|
fTTML = "STPP" # https://www.w3.org/TR/2018/REC-ttml-imsc1.0.1-20180424
|
|
fVTT = "WVTT" # https://www.w3.org/TR/webvtt1
|
|
|
|
@property
|
|
def extension(self) -> str:
|
|
return self.value.lower()
|
|
|
|
@staticmethod
|
|
def from_mime(mime: str) -> Subtitle.Codec:
|
|
mime = mime.lower().strip().split(".")[0]
|
|
if mime == "srt":
|
|
return Subtitle.Codec.SubRip
|
|
elif mime == "ssa":
|
|
return Subtitle.Codec.SubStationAlpha
|
|
elif mime == "ass":
|
|
return Subtitle.Codec.SubStationAlphav4
|
|
elif mime == "ttml":
|
|
return Subtitle.Codec.TimedTextMarkupLang
|
|
elif mime == "vtt":
|
|
return Subtitle.Codec.WebVTT
|
|
elif mime in ("smi", "sami"):
|
|
return Subtitle.Codec.SAMI
|
|
elif mime in ("sub", "microdvd"):
|
|
return Subtitle.Codec.MicroDVD
|
|
elif mime == "mpl2":
|
|
return Subtitle.Codec.MPL2
|
|
elif mime == "tmp":
|
|
return Subtitle.Codec.TMP
|
|
elif mime == "stpp":
|
|
return Subtitle.Codec.fTTML
|
|
elif mime == "wvtt":
|
|
return Subtitle.Codec.fVTT
|
|
raise ValueError(f"The MIME '{mime}' is not a supported Subtitle Codec")
|
|
|
|
@staticmethod
|
|
def from_codecs(codecs: str) -> Subtitle.Codec:
|
|
for codec in codecs.lower().split(","):
|
|
mime = codec.strip().split(".")[0]
|
|
try:
|
|
return Subtitle.Codec.from_mime(mime)
|
|
except ValueError:
|
|
pass
|
|
raise ValueError(f"No MIME types matched any supported Subtitle Codecs in '{codecs}'")
|
|
|
|
@staticmethod
|
|
def from_netflix_profile(profile: str) -> Subtitle.Codec:
|
|
profile = profile.lower().strip()
|
|
if profile.startswith("webvtt"):
|
|
return Subtitle.Codec.WebVTT
|
|
if profile.startswith("dfxp"):
|
|
return Subtitle.Codec.TimedTextMarkupLang
|
|
raise ValueError(f"The Content Profile '{profile}' is not a supported Subtitle Codec")
|
|
|
|
# WebVTT sanitization patterns (compiled once for performance)
|
|
CUE_ID_PATTERN = re.compile(r"^[A-Za-z]+\d+$")
|
|
TIMING_START_PATTERN = re.compile(r"^\d+:\d+[:\.]")
|
|
TIMING_LINE_PATTERN = re.compile(r"^((?:\d+:)?\d+:\d+[.,]\d+)\s*-->\s*((?:\d+:)?\d+:\d+[.,]\d+)(.*)$")
|
|
LINE_POS_PATTERN = re.compile(r"line:(\d+(?:\.\d+)?%?)")
|
|
|
|
def __init__(
|
|
self,
|
|
*args: Any,
|
|
codec: Optional[Subtitle.Codec] = None,
|
|
cc: bool = False,
|
|
sdh: bool = False,
|
|
forced: bool = False,
|
|
**kwargs: Any,
|
|
):
|
|
"""
|
|
Make a new Subtitle track object.
|
|
|
|
Parameters:
|
|
codec: A Subtitle.Codec enum representing the subtitle format.
|
|
If not specified, unshackle uses MediaInfo to get the format
|
|
after it downloads the track.
|
|
cc: Closed Caption.
|
|
- Intended as if you could not hear the audio at all.
|
|
- Can have Sound as well as Dialogue, but does not have to.
|
|
- Original source would be from EIA-CC encoded video. Typically all
|
|
upper-case characters.
|
|
Indicators of it being CC without knowing original source:
|
|
- Extracted with CCExtractor, or
|
|
- >>> (or similar) at the start of some or all lines, or
|
|
- All text is uppercase or at least the majority, or
|
|
- Subtitles use the scrolling-text style (one line appears, oldest
|
|
line then disappears).
|
|
The fact that you downloaded it as a SRT or VTT or such does not mean it
|
|
is not from an EIA-CC stream. And I would not take the streaming services
|
|
(CC) as gospel either as they tend to get it wrong too.
|
|
sdh: Deaf or Hard-of-Hearing. Also known as HOH in the UK (EU?).
|
|
- Intended as if you could not hear the audio at all.
|
|
- MUST have Sound as well as Dialogue to count as SDH.
|
|
- It has no "syntax" or "format", but it is not transmitted in
|
|
archaic forms like EIA-CC streams. It is intended for transmission
|
|
through SubRip (SRT), WebVTT (VTT), TTML, and the other modern
|
|
subtitle formats.
|
|
If you can see important audio/sound transcriptions and not only dialogue
|
|
and it does not have the indicators of CC, then it is most likely SDH.
|
|
If it does not have important audio/sounds transcriptions it might only be
|
|
regular subtitling (you would not mark as CC or SDH). This would be the
|
|
case for most translation subtitles. Like Anime for example.
|
|
forced: Typically used if there is important information at some point in time
|
|
like watching a Dubbed title and an important Sign or Letter is shown
|
|
or someone talking in a different language.
|
|
The Matroska Spec recommends that the player plays Forced tracks if
|
|
the player's current playback audio language matches a subtitle
|
|
marked as "forced".
|
|
However, that does not mean every player works like this but there is
|
|
no other way to reliably work with Forced subtitles where multiple
|
|
forced subtitles may be in the output file. Know what to expect
|
|
with "forced" subtitles.
|
|
|
|
Note: If codec is not specified, unshackle can skip some checks or assume a value.
|
|
Give as much information as possible.
|
|
|
|
Information on Subtitle Types:
|
|
https://bit.ly/2Oe4fLC (3PlayMedia Blog on SUB vs CC vs SDH).
|
|
However, I would not pay much attention to the claims about SDH needing to
|
|
be in the original language. It is logically not true.
|
|
|
|
CC == Closed Captions. Source: Every site.
|
|
SDH = Subtitles for the Deaf or Hard-of-Hearing. Source: Every site.
|
|
HOH = Exact same as SDH. Is a term used in the UK. Source: https://bit.ly/2PGJatz (ICO UK)
|
|
|
|
More in-depth information, examples, and stuff to look for can be found in the Parameter
|
|
explanation list above.
|
|
"""
|
|
super().__init__(*args, **kwargs)
|
|
|
|
if not isinstance(codec, (Subtitle.Codec, type(None))):
|
|
raise TypeError(f"Expected codec to be a {Subtitle.Codec}, not {codec!r}")
|
|
if not isinstance(cc, (bool, int)) or (isinstance(cc, int) and cc not in (0, 1)):
|
|
raise TypeError(f"Expected cc to be a {bool} or bool-like {int}, not {cc!r}")
|
|
if not isinstance(sdh, (bool, int)) or (isinstance(sdh, int) and sdh not in (0, 1)):
|
|
raise TypeError(f"Expected sdh to be a {bool} or bool-like {int}, not {sdh!r}")
|
|
if not isinstance(forced, (bool, int)) or (isinstance(forced, int) and forced not in (0, 1)):
|
|
raise TypeError(f"Expected forced to be a {bool} or bool-like {int}, not {forced!r}")
|
|
|
|
self.codec = codec
|
|
|
|
self.cc = bool(cc)
|
|
self.sdh = bool(sdh)
|
|
self.forced = bool(forced)
|
|
|
|
if self.cc and self.sdh:
|
|
raise ValueError("A text track cannot be both CC and SDH.")
|
|
|
|
if self.forced and (self.cc or self.sdh):
|
|
raise ValueError("A text track cannot be CC/SDH as well as Forced.")
|
|
|
|
# TODO: Migrate to new event observer system
|
|
# Called after Track has been converted to another format
|
|
self.OnConverted: Optional[Callable[[Subtitle.Codec], None]] = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data = super().to_dict()
|
|
data.update(
|
|
{
|
|
"codec": self.codec.name if self.codec else None,
|
|
"cc": self.cc,
|
|
"sdh": self.sdh,
|
|
"forced": self.forced,
|
|
}
|
|
)
|
|
return data
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> Subtitle:
|
|
kwargs = Track.base_kwargs_from_dict(data)
|
|
return cls(
|
|
**kwargs,
|
|
codec=Subtitle.Codec[data["codec"]] if data.get("codec") else None,
|
|
cc=data.get("cc", False),
|
|
sdh=data.get("sdh", False),
|
|
forced=data.get("forced", False),
|
|
)
|
|
|
|
@property
|
|
def mux_language(self) -> Language:
|
|
"""The language tag written to the output file, with the script its region implies."""
|
|
return with_script(self.language)
|
|
|
|
def __str__(self) -> str:
|
|
return " | ".join(
|
|
filter(
|
|
bool,
|
|
["SUB", f"[{self.codec.value}]" if self.codec else None, str(self.language), self.get_track_name()],
|
|
)
|
|
)
|
|
|
|
def get_track_name(self) -> Optional[str]:
|
|
"""Return the base Track Name."""
|
|
track_name = super().get_track_name() or ""
|
|
flag = self.cc and "CC" or self.sdh and "SDH" or self.forced and "Forced"
|
|
if flag:
|
|
if track_name:
|
|
flag = f" ({flag})"
|
|
track_name += flag
|
|
return track_name or None
|
|
|
|
def download(
|
|
self,
|
|
session: requests.Session,
|
|
prepare_drm: partial,
|
|
max_workers: Optional[int] = None,
|
|
progress: Optional[partial] = None,
|
|
*,
|
|
cdm: Optional[object] = None,
|
|
no_proxy_download: bool = False,
|
|
proxy_download: Optional[str] = None,
|
|
adaptive_workers: bool = False,
|
|
download_processes: int = 1,
|
|
):
|
|
super().download(
|
|
session,
|
|
prepare_drm,
|
|
max_workers,
|
|
progress,
|
|
cdm=cdm,
|
|
no_proxy_download=no_proxy_download,
|
|
proxy_download=proxy_download,
|
|
adaptive_workers=adaptive_workers,
|
|
download_processes=download_processes,
|
|
)
|
|
if not self.path:
|
|
return
|
|
|
|
if self.codec == Subtitle.Codec.fTTML:
|
|
self.convert(Subtitle.Codec.TimedTextMarkupLang)
|
|
elif self.codec == Subtitle.Codec.fVTT:
|
|
self.convert(Subtitle.Codec.WebVTT)
|
|
elif self.codec == Subtitle.Codec.WebVTT:
|
|
text = self.path.read_text("utf8")
|
|
text = Subtitle.strip_webvtt_timestamp_map(text)
|
|
if self.descriptor == Track.Descriptor.DASH:
|
|
from unshackle.core.manifests import DASH
|
|
|
|
template = DASH.merge_segment_templates(
|
|
self.data["dash"]["adaptation_set"], self.data["dash"]["representation"]
|
|
)
|
|
if template is not None:
|
|
offset = template.get("presentationTimeOffset", "0")
|
|
text = rebase_webvtt(text, int(offset) * 1000 // int(template.get("timescale", "1")))
|
|
if len(self.data["dash"]["segment_durations"]) > 1:
|
|
text = merge_segmented_webvtt(
|
|
text,
|
|
segment_durations=self.data["dash"]["segment_durations"],
|
|
timescale=self.data["dash"]["timescale"],
|
|
)
|
|
elif self.descriptor == Track.Descriptor.HLS:
|
|
if len(self.data["hls"]["segment_durations"]) > 1:
|
|
text = merge_segmented_webvtt(
|
|
text,
|
|
segment_durations=self.data["hls"]["segment_durations"],
|
|
timescale=1, # ?
|
|
)
|
|
|
|
text = Subtitle.sanitize_webvtt_timestamps(text)
|
|
text = Subtitle.sanitize_webvtt_cue_identifiers(text)
|
|
text = Subtitle.merge_overlapping_webvtt_cues(text)
|
|
text = Subtitle.strip_webvtt_cue_classes(text)
|
|
|
|
preserve_formatting = config.subtitle.get("preserve_formatting", True)
|
|
|
|
if preserve_formatting:
|
|
self.path.write_text(text, encoding="utf8")
|
|
else:
|
|
try:
|
|
caption_set = pycaption.WebVTTReader().read(text)
|
|
Subtitle.merge_same_cues(caption_set)
|
|
Subtitle.filter_unwanted_cues(caption_set)
|
|
subtitle_text = pycaption.WebVTTWriter().write(caption_set)
|
|
self.path.write_text(subtitle_text, encoding="utf8")
|
|
except pycaption.exceptions.CaptionReadNoCaptions:
|
|
# some renditions carry headers but no cues; write them out rather than fail the download
|
|
self.path.write_text(text, encoding="utf8")
|
|
except pycaption.exceptions.CaptionReadSyntaxError:
|
|
text = Subtitle.sanitize_webvtt(text)
|
|
try:
|
|
caption_set = pycaption.WebVTTReader().read(text)
|
|
Subtitle.merge_same_cues(caption_set)
|
|
Subtitle.filter_unwanted_cues(caption_set)
|
|
subtitle_text = pycaption.WebVTTWriter().write(caption_set)
|
|
self.path.write_text(subtitle_text, encoding="utf8")
|
|
except Exception:
|
|
# Keep the sanitized version even if parsing failed
|
|
self.path.write_text(text, encoding="utf8")
|
|
|
|
@staticmethod
|
|
def strip_webvtt_cue_classes(text: str) -> str:
|
|
"""
|
|
Reduce a cue payload to the markup every target format shares: ``<i>``, ``<b>``
|
|
and ``<u>`` without their class list. It removes class spans (``<c.foo>``),
|
|
``<v>``, ``<lang>``, ``<ruby>``/``<rt>`` and karaoke timestamps
|
|
(``<00:00:01.000>``).
|
|
|
|
A class only selects a CSS ``::cue`` rule that no player receives, so it never
|
|
carries styling, but it does break rendering. FFmpeg's SubRip reader treats a tag
|
|
name as a tag only while it matches ``[0-9a-zA-Z_/]``. The dot fails that test, so
|
|
FFmpeg prints ``<i.loud>`` as cue text and then emits the closing italic-off with
|
|
nothing opened. FFmpeg's WebVTT reader matches ``i``/``b``/``u`` exactly and drops
|
|
the rest, so there a class silently loses the styling instead. The tags this
|
|
method removes outright are the ones both readers already discard, and that
|
|
libass would draw as text if a conversion carried them into ASS.
|
|
"""
|
|
text = re.sub(r"</?c(?:\.[^\s>]*)?(?:\s[^>]*)?>", "", text)
|
|
text = re.sub(r"(</?)([ibu])\.[^\s>]*", r"\1\2", text)
|
|
text = re.sub(r"</?(?:v|lang)(?:\.[^\s>]*)?(?:\s[^>]*)?>", "", text)
|
|
text = re.sub(r"</?(?:ruby|rt)>", "", text)
|
|
return re.sub(r"<\d{2,}:\d{2}(?::\d{2})?\.\d{3}>", "", text)
|
|
|
|
@staticmethod
|
|
def strip_webvtt_timestamp_map(text: str) -> str:
|
|
"""
|
|
Remove X-TIMESTAMP-MAP header lines (RFC 8216 §3.5, HLS only).
|
|
|
|
Cue times are already absolute, but SubtitleEdit reads the header and offsets
|
|
every cue by it during conversion.
|
|
"""
|
|
# IGNORECASE: SubtitleEdit matches the header case-insensitively, so must the strip
|
|
return re.sub(r"^[ \t]*X-TIMESTAMP-MAP[^\n]*\n?", "", text, flags=re.MULTILINE | re.IGNORECASE)
|
|
|
|
@staticmethod
|
|
def sanitize_webvtt_timestamps(text: str) -> str:
|
|
"""
|
|
Fix invalid timestamps in WebVTT files, particularly negative timestamps.
|
|
|
|
Parameters:
|
|
text: The WebVTT text as string
|
|
|
|
Returns:
|
|
Sanitized WebVTT text
|
|
"""
|
|
return re.sub(r"(-\d+:\d+:\d+\.\d+)", "00:00:00.000", text)
|
|
|
|
@staticmethod
|
|
def has_webvtt_cue_identifiers(text: str) -> bool:
|
|
"""
|
|
Examine WebVTT text for cue identifiers that need removal.
|
|
|
|
Parameters:
|
|
text: The WebVTT text as string
|
|
|
|
Returns:
|
|
True if the text has cue identifiers, False otherwise
|
|
"""
|
|
lines = text.split("\n")
|
|
|
|
for i, line in enumerate(lines):
|
|
line = line.strip()
|
|
if Subtitle.CUE_ID_PATTERN.match(line):
|
|
j = i + 1
|
|
while j < len(lines) and not lines[j].strip():
|
|
j += 1
|
|
if j < len(lines) and ("-->" in lines[j] or Subtitle.TIMING_START_PATTERN.match(lines[j].strip())):
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def sanitize_webvtt_cue_identifiers(text: str) -> str:
|
|
"""
|
|
Remove WebVTT cue identifiers that can confuse subtitle parsers.
|
|
|
|
Some services use cue identifiers such as "Q0" and "Q1"
|
|
that appear on their own line before the timing line. These can be
|
|
incorrectly parsed as part of the previous cue's text by
|
|
some parsers (like pysubs2).
|
|
|
|
Parameters:
|
|
text: The WebVTT text as string
|
|
|
|
Returns:
|
|
Sanitized WebVTT text with cue identifiers removed
|
|
"""
|
|
if not Subtitle.has_webvtt_cue_identifiers(text):
|
|
return text
|
|
|
|
lines = text.split("\n")
|
|
sanitized_lines = []
|
|
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
|
|
if Subtitle.CUE_ID_PATTERN.match(line):
|
|
j = i + 1
|
|
while j < len(lines) and not lines[j].strip():
|
|
j += 1
|
|
if j < len(lines) and ("-->" in lines[j] or Subtitle.TIMING_START_PATTERN.match(lines[j].strip())):
|
|
i += 1
|
|
continue
|
|
|
|
sanitized_lines.append(lines[i])
|
|
i += 1
|
|
|
|
return "\n".join(sanitized_lines)
|
|
|
|
@staticmethod
|
|
def parse_vtt_time(t: str) -> int:
|
|
"""Parse WebVTT timestamp to milliseconds. Returns 0 for malformed input."""
|
|
try:
|
|
t = t.replace(",", ".")
|
|
parts = t.split(":")
|
|
if len(parts) == 2:
|
|
m, s = parts
|
|
h = "0"
|
|
elif len(parts) >= 3:
|
|
h, m, s = parts[:3]
|
|
else:
|
|
return 0
|
|
sec_parts = s.split(".")
|
|
secs = int(sec_parts[0])
|
|
# Handle variable millisecond digits (e.g., .5 = 500ms, .50 = 500ms, .500 = 500ms)
|
|
ms = int(sec_parts[1].ljust(3, "0")[:3]) if len(sec_parts) > 1 else 0
|
|
return int(h) * 3600000 + int(m) * 60000 + secs * 1000 + ms
|
|
except (ValueError, IndexError):
|
|
return 0
|
|
|
|
@staticmethod
|
|
def has_overlapping_webvtt_cues(text: str) -> bool:
|
|
"""
|
|
Examine WebVTT text for overlapping cues that need merging.
|
|
|
|
Detects cues with start times within 50ms of each other and the same end time,
|
|
which indicates multi-line subtitles split into separate cues.
|
|
|
|
Parameters:
|
|
text: The WebVTT text as string
|
|
|
|
Returns:
|
|
True if the text has overlapping cues, False otherwise
|
|
"""
|
|
timings = []
|
|
for line in text.split("\n"):
|
|
match = Subtitle.TIMING_LINE_PATTERN.match(line)
|
|
if match:
|
|
start_str, end_str = match.group(1), match.group(2)
|
|
timings.append((Subtitle.parse_vtt_time(start_str), Subtitle.parse_vtt_time(end_str)))
|
|
|
|
for i in range(len(timings) - 1):
|
|
curr_start, curr_end = timings[i]
|
|
next_start, next_end = timings[i + 1]
|
|
if abs(curr_start - next_start) <= 50 and curr_end == next_end:
|
|
return True
|
|
|
|
return False
|
|
|
|
@staticmethod
|
|
def merge_overlapping_webvtt_cues(text: str) -> str:
|
|
"""
|
|
Merge WebVTT cues that have overlapping/near-identical times but different line positions.
|
|
|
|
Some services use separate cues for each line of a multi-line subtitle, with
|
|
slightly different start times (1ms apart) and different line: positions.
|
|
This merges them into single cues with proper line ordering based on the
|
|
line: position (lower percentage = higher on screen = first line).
|
|
|
|
Parameters:
|
|
text: The WebVTT text as string
|
|
|
|
Returns:
|
|
WebVTT text with overlapping cues merged
|
|
"""
|
|
if not Subtitle.has_overlapping_webvtt_cues(text):
|
|
return text
|
|
|
|
lines = text.split("\n")
|
|
cues = []
|
|
header_lines = []
|
|
in_header = True
|
|
i = 0
|
|
|
|
while i < len(lines):
|
|
line = lines[i]
|
|
|
|
if in_header:
|
|
if "-->" in line:
|
|
in_header = False
|
|
else:
|
|
header_lines.append(line)
|
|
i += 1
|
|
continue
|
|
|
|
match = Subtitle.TIMING_LINE_PATTERN.match(line)
|
|
if match:
|
|
start_str, end_str, settings = match.groups()
|
|
line_pos = 100.0 # Default to bottom
|
|
line_match = Subtitle.LINE_POS_PATTERN.search(settings)
|
|
if line_match:
|
|
pos_str = line_match.group(1).rstrip("%")
|
|
line_pos = float(pos_str)
|
|
|
|
content_lines = []
|
|
i += 1
|
|
while i < len(lines) and lines[i].strip() and "-->" not in lines[i]:
|
|
content_lines.append(lines[i])
|
|
i += 1
|
|
|
|
cues.append(
|
|
{
|
|
"start_ms": Subtitle.parse_vtt_time(start_str),
|
|
"end_ms": Subtitle.parse_vtt_time(end_str),
|
|
"start_str": start_str,
|
|
"end_str": end_str,
|
|
"line_pos": line_pos,
|
|
"content": "\n".join(content_lines),
|
|
"settings": settings,
|
|
}
|
|
)
|
|
else:
|
|
i += 1
|
|
|
|
# Merge overlapping cues (within 50ms of each other with same end time)
|
|
merged_cues = []
|
|
i = 0
|
|
while i < len(cues):
|
|
current = cues[i]
|
|
group = [current]
|
|
|
|
j = i + 1
|
|
while j < len(cues):
|
|
other = cues[j]
|
|
if abs(current["start_ms"] - other["start_ms"]) <= 50 and current["end_ms"] == other["end_ms"]:
|
|
group.append(other)
|
|
j += 1
|
|
else:
|
|
break
|
|
|
|
if len(group) > 1:
|
|
group.sort(key=lambda x: x["line_pos"])
|
|
earliest = min(group, key=lambda x: x["start_ms"])
|
|
merged_cues.append(
|
|
{
|
|
"start_str": earliest["start_str"],
|
|
"end_str": group[0]["end_str"],
|
|
"content": "\n".join(c["content"] for c in group),
|
|
"settings": "",
|
|
}
|
|
)
|
|
else:
|
|
merged_cues.append(
|
|
{
|
|
"start_str": current["start_str"],
|
|
"end_str": current["end_str"],
|
|
"content": current["content"],
|
|
"settings": current["settings"],
|
|
}
|
|
)
|
|
|
|
i = j if len(group) > 1 else i + 1
|
|
|
|
result_lines = header_lines[:]
|
|
if result_lines and result_lines[-1].strip():
|
|
result_lines.append("")
|
|
|
|
for cue in merged_cues:
|
|
result_lines.append(f"{cue['start_str']} --> {cue['end_str']}{cue['settings']}")
|
|
result_lines.append(cue["content"])
|
|
result_lines.append("")
|
|
|
|
return "\n".join(result_lines)
|
|
|
|
@staticmethod
|
|
def sanitize_webvtt(text: str) -> str:
|
|
"""
|
|
More thorough sanitization of WebVTT files that corrects multiple potential issues.
|
|
|
|
This is lossy, so use it only as a fallback once normal parsing has failed. This method discards
|
|
everything before the WEBVTT header line and reduces the header itself to a bare "WEBVTT" line.
|
|
Negative timestamps become 00:00:00.000, and it pads timestamps that have no hours field to
|
|
HH:MM:SS.mmm.
|
|
|
|
Parameters:
|
|
text: The WebVTT text as string
|
|
|
|
Returns:
|
|
Sanitized WebVTT text
|
|
"""
|
|
# Make sure we have a proper WEBVTT header
|
|
if not text.strip().startswith("WEBVTT"):
|
|
text = "WEBVTT\n\n" + text
|
|
|
|
lines = text.split("\n")
|
|
sanitized_lines = []
|
|
timestamp_pattern = re.compile(r"^((?:\d+:)?\d+:\d+\.\d+)\s+-->\s+((?:\d+:)?\d+:\d+\.\d+)")
|
|
|
|
header_done = False
|
|
for line in lines:
|
|
if not header_done:
|
|
if line.startswith("WEBVTT"):
|
|
sanitized_lines.append("WEBVTT")
|
|
header_done = True
|
|
continue
|
|
|
|
if "-" in line and "-->" in line:
|
|
line = re.sub(r"(-\d+:\d+:\d+\.\d+)", "00:00:00.000", line)
|
|
|
|
match = timestamp_pattern.match(line)
|
|
if match:
|
|
start_time = match.group(1)
|
|
end_time = match.group(2)
|
|
|
|
if start_time.count(":") == 1:
|
|
start_time = f"00:{start_time}"
|
|
if end_time.count(":") == 1:
|
|
end_time = f"00:{end_time}"
|
|
|
|
line = f"{start_time} --> {end_time}"
|
|
|
|
sanitized_lines.append(line)
|
|
|
|
return "\n".join(sanitized_lines)
|
|
|
|
def convert(self, codec: Subtitle.Codec, *, forced: bool = False) -> Path:
|
|
"""
|
|
Convert this Subtitle to another format.
|
|
|
|
Backend selection is data-driven (see ``tracks/subtitle_convert.py``): unshackle
|
|
uses the best available backend that can convert source->target, and falls back
|
|
through the capability chain on failure. The ``conversion_method`` config key
|
|
(``auto`` | ``subby`` | ``pysubs2`` | ``subtitleedit`` | ``pycaption``) pins the
|
|
backend, and ``preferred_conversion_method`` nudges it per-service. An explicit
|
|
config value always wins.
|
|
|
|
``forced`` marks an explicit user request (``--sub-format``). unshackle skips lossy
|
|
downconverts of styled formats (SSA/ASS -> SRT) unless ``forced`` is True.
|
|
"""
|
|
from unshackle.core.tracks.subtitle_convert import run_conversion
|
|
|
|
if not self.path or not self.path.exists():
|
|
raise ValueError("You must download the subtitle track first.")
|
|
|
|
method = (
|
|
config.subtitle.get("conversion_method") or getattr(self, "preferred_conversion_method", None) or "auto"
|
|
)
|
|
pin = None if method == "auto" else method
|
|
return run_conversion(self, codec, pin=pin, forced=forced)
|
|
|
|
@staticmethod
|
|
def extract_fonts(text: str) -> set[str]:
|
|
"""
|
|
Font names referenced by an ASS/SSA subtitle.
|
|
|
|
Covers both sources that need attaching for correct rendering:
|
|
- the ``Fontname`` column of every ``Style:`` line in ``[V4+ Styles]``/``[V4 Styles]``
|
|
(column located from the section's ``Format:`` line, not assumed by index), and
|
|
- inline ``\\fn`` font overrides inside ``Dialogue`` override blocks.
|
|
|
|
This method removes a leading ``@`` (vertical-writing prefix), de-duplicates the
|
|
names case-insensitively, and prefers a mixed-case spelling over an all-lowercase one.
|
|
"""
|
|
names: set[str] = set()
|
|
name_index = 1 # ASS default Style order: Name, Fontname, ...
|
|
in_styles = False
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith("["):
|
|
in_styles = stripped.lower() in ("[v4+ styles]", "[v4 styles]")
|
|
continue
|
|
if not in_styles:
|
|
continue
|
|
if stripped.lower().startswith("format:"):
|
|
columns = [c.strip().lower() for c in stripped.split(":", 1)[1].split(",")]
|
|
if "fontname" in columns:
|
|
name_index = columns.index("fontname")
|
|
elif stripped.lower().startswith("style:"):
|
|
fields = stripped.split(":", 1)[1].split(",")
|
|
if len(fields) > name_index:
|
|
names.add(fields[name_index].strip())
|
|
|
|
names.update(match.strip() for match in re.findall(r"\\fn([^\\}]+)", text))
|
|
|
|
canonical: dict[str, str] = {}
|
|
for name in (raw.lstrip("@").strip() for raw in names):
|
|
if not name:
|
|
continue
|
|
key = name.lower()
|
|
if key not in canonical or (name != name.lower() and canonical[key] == canonical[key].lower()):
|
|
canonical[key] = name
|
|
return set(canonical.values())
|
|
|
|
@staticmethod
|
|
def parse(data: bytes, codec: Subtitle.Codec) -> pycaption.CaptionSet:
|
|
if not isinstance(data, bytes):
|
|
raise ValueError(f"Subtitle data must be parsed as bytes data, not {type(data).__name__}")
|
|
|
|
try:
|
|
if codec == Subtitle.Codec.SubRip:
|
|
text = try_ensure_utf8(data).decode("utf8")
|
|
caption_set = pycaption.SRTReader().read(text)
|
|
elif codec == Subtitle.Codec.fTTML:
|
|
caption_lists: dict[str, pycaption.CaptionList] = defaultdict(pycaption.CaptionList)
|
|
for segment in (
|
|
Subtitle.parse(box.data, Subtitle.Codec.TimedTextMarkupLang)
|
|
for box in MP4.parse_stream(BytesIO(data))
|
|
if box.type == b"mdat"
|
|
):
|
|
for lang in segment.get_languages():
|
|
caption_lists[lang].extend(segment.get_captions(lang))
|
|
caption_set: pycaption.CaptionSet = pycaption.CaptionSet(caption_lists)
|
|
elif codec == Subtitle.Codec.TimedTextMarkupLang:
|
|
text = try_ensure_utf8(data).decode("utf8")
|
|
text = text.replace("tt:", "")
|
|
# negative size values aren't allowed in TTML/DFXP spec, replace with 0
|
|
text = re.sub(r"-(\d+(?:\.\d+)?)(px|em|%|c|pt)", r"0\2", text)
|
|
caption_set = pycaption.DFXPReader().read(text)
|
|
elif codec == Subtitle.Codec.fVTT:
|
|
caption_lists: dict[str, pycaption.CaptionList] = defaultdict(pycaption.CaptionList)
|
|
caption_list, language = Subtitle.merge_segmented_wvtt(data)
|
|
caption_lists[language] = caption_list
|
|
caption_set: pycaption.CaptionSet = pycaption.CaptionSet(caption_lists)
|
|
elif codec == Subtitle.Codec.WebVTT:
|
|
text = try_ensure_utf8(data).decode("utf8")
|
|
text = Subtitle.sanitize_broken_webvtt(text)
|
|
text = Subtitle.space_webvtt_headers(text)
|
|
caption_set = pycaption.WebVTTReader().read(text)
|
|
elif codec == Subtitle.Codec.SAMI:
|
|
# Use subby for SAMI parsing
|
|
converter = SAMIConverter()
|
|
srt_subtitles = converter.from_bytes(data)
|
|
# Convert SRT back to CaptionSet for compatibility
|
|
srt_text = str(srt_subtitles).encode("utf8")
|
|
caption_set = Subtitle.parse(srt_text, Subtitle.Codec.SubRip)
|
|
else:
|
|
raise ValueError(f'Unknown Subtitle format "{codec}"...')
|
|
except pycaption.exceptions.CaptionReadSyntaxError as e:
|
|
raise SyntaxError(f'A syntax error has occurred when reading the "{codec}" subtitle: {e}')
|
|
except pycaption.exceptions.CaptionReadNoCaptions:
|
|
return pycaption.CaptionSet({"en": []})
|
|
|
|
# remove empty caption lists or some code breaks, especially if it's the first list
|
|
for language in caption_set.get_languages():
|
|
if not caption_set.get_captions(language):
|
|
# noinspection PyProtectedMember
|
|
del caption_set._captions[language]
|
|
|
|
return caption_set
|
|
|
|
@staticmethod
|
|
def sanitize_broken_webvtt(text: str) -> str:
|
|
"""
|
|
Remove or fix corrupted WebVTT lines, particularly those with invalid timestamps.
|
|
|
|
Parameters:
|
|
text: The WebVTT text as string
|
|
|
|
Returns:
|
|
Sanitized WebVTT text with corrupted lines removed
|
|
"""
|
|
lines = text.splitlines()
|
|
sanitized_lines = []
|
|
|
|
i = 0
|
|
while i < len(lines):
|
|
if not lines[i].strip():
|
|
sanitized_lines.append(lines[i])
|
|
i += 1
|
|
continue
|
|
|
|
if "-->" in lines[i]:
|
|
timestamp_parts = lines[i].split("-->")
|
|
if len(timestamp_parts) != 2 or not timestamp_parts[1].strip() or timestamp_parts[1].strip() == "0":
|
|
j = i + 1
|
|
while j < len(lines) and "-->" not in lines[j] and lines[j].strip():
|
|
j += 1
|
|
i = j
|
|
continue
|
|
|
|
sanitized_lines.append(lines[i])
|
|
else:
|
|
sanitized_lines.append(lines[i])
|
|
|
|
i += 1
|
|
|
|
return "\n".join(sanitized_lines)
|
|
|
|
@staticmethod
|
|
def space_webvtt_headers(data: Union[str, bytes]):
|
|
"""
|
|
Space out the WEBVTT Headers from Captions.
|
|
|
|
Segmented VTT when merged may have the WEBVTT headers part of the next caption
|
|
as they were not separated far enough from the previous caption and ended up
|
|
as caption text rather than the header for the next segment.
|
|
"""
|
|
if isinstance(data, bytes):
|
|
data = try_ensure_utf8(data).decode("utf8")
|
|
elif not isinstance(data, str):
|
|
raise ValueError(f"Expecting data to be a str, not {data!r}")
|
|
|
|
text = (
|
|
data.replace("WEBVTT", "\n\nWEBVTT").replace("\r", "").replace("\n\n\n", "\n \n\n").replace("\n\n<", "\n<")
|
|
)
|
|
|
|
return text
|
|
|
|
@staticmethod
|
|
def merge_same_cues(caption_set: pycaption.CaptionSet):
|
|
"""Merge captions with the same timecodes and text as one in-place."""
|
|
for lang in caption_set.get_languages():
|
|
captions = caption_set.get_captions(lang)
|
|
last_caption = None
|
|
concurrent_captions = pycaption.CaptionList()
|
|
merged_captions = pycaption.CaptionList()
|
|
for caption in captions:
|
|
if last_caption:
|
|
if (caption.start, caption.end) == (last_caption.start, last_caption.end):
|
|
if caption.get_text() != last_caption.get_text():
|
|
concurrent_captions.append(caption)
|
|
last_caption = caption
|
|
continue
|
|
else:
|
|
merged_captions.append(pycaption.base.merge(concurrent_captions))
|
|
concurrent_captions = [caption]
|
|
last_caption = caption
|
|
|
|
if concurrent_captions:
|
|
merged_captions.append(pycaption.base.merge(concurrent_captions))
|
|
if merged_captions:
|
|
caption_set.set_captions(lang, merged_captions)
|
|
|
|
@staticmethod
|
|
def filter_unwanted_cues(caption_set: pycaption.CaptionSet):
|
|
"""
|
|
Filter out subtitle cues containing only or whitespace.
|
|
"""
|
|
for lang in caption_set.get_languages():
|
|
captions = caption_set.get_captions(lang)
|
|
filtered_captions = pycaption.CaptionList()
|
|
|
|
for caption in captions:
|
|
text = caption.get_text().strip()
|
|
if not text or text == " " or all(c in " \t\n\r\xa0" for c in text.replace(" ", "\xa0")):
|
|
continue
|
|
|
|
filtered_captions.append(caption)
|
|
|
|
caption_set.set_captions(lang, filtered_captions)
|
|
|
|
@staticmethod
|
|
def merge_segmented_wvtt(data: bytes, period_start: float = 0.0) -> tuple[CaptionList, Optional[str]]:
|
|
"""
|
|
Convert Segmented DASH WebVTT cues into a pycaption Caption List.
|
|
Also returns an ISO 639-2 alpha-3 language tag if available.
|
|
|
|
Code ported originally by xhlove to Python from shaka-player.
|
|
rlaphoenix has since improved it with pymp4 and
|
|
pycaption functions.
|
|
"""
|
|
captions = CaptionList()
|
|
|
|
# init:
|
|
saw_wvtt_box = False
|
|
timescale = None
|
|
language = None
|
|
|
|
# media:
|
|
# > tfhd
|
|
default_duration = None
|
|
# > tfdt
|
|
saw_tfdt_box = False
|
|
base_time = 0
|
|
# > trun
|
|
saw_trun_box = False
|
|
samples = []
|
|
|
|
def flatten_boxes(box: Container) -> Iterable[Container]:
|
|
for child in box:
|
|
if hasattr(child, "children"):
|
|
yield from flatten_boxes(child.children)
|
|
del child["children"]
|
|
if hasattr(child, "entries"):
|
|
yield from flatten_boxes(child.entries)
|
|
del child["entries"]
|
|
# some boxes (mainly within 'entries') uses format not type
|
|
child["type"] = child.get("type") or child.get("format")
|
|
yield child
|
|
|
|
for box in flatten_boxes(MP4.parse_stream(BytesIO(data))):
|
|
# init
|
|
if box.type == b"mdhd":
|
|
timescale = box.timescale
|
|
language = box.language
|
|
|
|
if box.type == b"wvtt":
|
|
saw_wvtt_box = True
|
|
|
|
# media
|
|
if box.type == b"styp":
|
|
# styp opens a new segment, so reset per-segment fragment state
|
|
default_duration = None
|
|
saw_tfdt_box = False
|
|
base_time = 0
|
|
saw_trun_box = False
|
|
samples = []
|
|
|
|
if box.type == b"tfhd":
|
|
if box.flags.default_sample_duration_present:
|
|
default_duration = box.default_sample_duration
|
|
|
|
if box.type == b"tfdt":
|
|
saw_tfdt_box = True
|
|
base_time = box.baseMediaDecodeTime
|
|
|
|
if box.type == b"trun":
|
|
saw_trun_box = True
|
|
samples = box.sample_info
|
|
|
|
if box.type == b"mdat":
|
|
if not timescale:
|
|
raise ValueError("Timescale was not found in the Segmented WebVTT.")
|
|
if not saw_wvtt_box:
|
|
raise ValueError("The WVTT box was not found in the Segmented WebVTT.")
|
|
if not saw_tfdt_box:
|
|
raise ValueError("The TFDT box was not found in the Segmented WebVTT.")
|
|
if not saw_trun_box:
|
|
raise ValueError("The TRUN box was not found in the Segmented WebVTT.")
|
|
|
|
vttc_boxes = MP4.parse_stream(BytesIO(box.data))
|
|
current_time = base_time + period_start
|
|
|
|
for sample, vttc_box in zip(samples, vttc_boxes):
|
|
duration = sample.sample_duration or default_duration
|
|
if sample.sample_composition_time_offsets:
|
|
current_time += sample.sample_composition_time_offsets
|
|
|
|
start_time = current_time
|
|
end_time = current_time + (duration or 0)
|
|
current_time = end_time
|
|
|
|
if vttc_box.type == b"vtte":
|
|
# vtte is a vttc that's empty, skip
|
|
continue
|
|
|
|
layout: Optional[Layout] = None
|
|
nodes: list[CaptionNode] = []
|
|
|
|
for cue_box in vttc_box.children:
|
|
if cue_box.type == b"vsid":
|
|
continue
|
|
if cue_box.type == b"sttg":
|
|
layout = Layout(webvtt_positioning=cue_box.settings)
|
|
elif cue_box.type == b"payl":
|
|
nodes.extend(
|
|
[
|
|
node
|
|
for line in cue_box.cue_text.split("\n")
|
|
for node in [
|
|
CaptionNode.create_text(WebVTTReader()._decode(line)),
|
|
CaptionNode.create_break(),
|
|
]
|
|
]
|
|
)
|
|
nodes.pop()
|
|
|
|
if nodes:
|
|
caption = Caption(
|
|
start=start_time * timescale, # as microseconds
|
|
end=end_time * timescale,
|
|
nodes=nodes,
|
|
layout_info=layout,
|
|
)
|
|
p_caption = captions[-1] if captions else None
|
|
if p_caption and caption.start == p_caption.end and str(caption.nodes) == str(p_caption.nodes):
|
|
# it's a duplicate, but lets take its end time
|
|
p_caption.end = caption.end
|
|
continue
|
|
captions.append(caption)
|
|
|
|
return captions, language
|
|
|
|
def sample_slice(
|
|
self, starts: Sequence[float], timescale: float = 1, window: Optional[tuple[int, int]] = None
|
|
) -> slice:
|
|
"""Return the segments of ``window`` plus one segment on each side.
|
|
|
|
Cue times in a segment are absolute, so ``sample_start`` stays zero and ``trim`` makes the cut.
|
|
A track that ends before the window keeps only its last segment.
|
|
"""
|
|
try:
|
|
keep = super().sample_slice(starts, timescale, window)
|
|
except click.ClickException:
|
|
keep = slice(len(starts) - 1, None)
|
|
self.sample_start = 0.0
|
|
if keep.start is None:
|
|
return keep
|
|
return slice(max(keep.start - 1, 0), None if keep.stop is None else keep.stop + 1)
|
|
|
|
def trim(self, start: float, end: float) -> bool:
|
|
"""Keep the cues between ``start`` and ``end`` seconds and move them to start at zero.
|
|
|
|
Returns False and leaves the file whole when pysubs2 cannot read or write its format.
|
|
"""
|
|
import pysubs2
|
|
|
|
low, high = round(start * 1000), round(end * 1000)
|
|
try:
|
|
subs = pysubs2.load(str(self.path), encoding="utf-8")
|
|
subs.events = [event for event in subs.events if event.end > low and event.start < high]
|
|
for event in subs.events:
|
|
event.start = max(event.start, low) - low
|
|
event.end = min(event.end, high) - low
|
|
subs.save(str(self.path))
|
|
except (pysubs2.exceptions.Pysubs2Error, UnicodeDecodeError):
|
|
return False
|
|
self.sample_start = start
|
|
return True
|
|
|
|
def strip_hearing_impaired(self) -> None:
|
|
"""
|
|
Remove the SDH captions.
|
|
|
|
The 'sdh_method' setting in config gives the SDH stripping method:
|
|
- 'auto' (default): Tries subby first, then SubtitleEdit, then filter-subs
|
|
- 'subby': Uses subby's SDHStripper
|
|
- 'subtitleedit': Uses SubtitleEdit when available
|
|
- 'filter-subs': Uses subtitle-filter library
|
|
"""
|
|
if not self.path or not self.path.exists():
|
|
raise ValueError("You must download the subtitle track first.")
|
|
|
|
sdh_method = config.subtitle.get("sdh_method", "auto")
|
|
|
|
if sdh_method == "subby" and self.codec == Subtitle.Codec.SubRip:
|
|
fixer = CommonIssuesFixer()
|
|
stripper = SDHStripper()
|
|
srt, _ = fixer.from_file(self.path)
|
|
stripped, status = stripper.from_srt(srt)
|
|
if status is True:
|
|
stripped.save(self.path)
|
|
return
|
|
elif sdh_method == "subtitleedit" and binaries.SubtitleEdit:
|
|
pass # Continue to SubtitleEdit section below
|
|
elif sdh_method == "auto":
|
|
if self.codec == Subtitle.Codec.SubRip:
|
|
try:
|
|
fixer = CommonIssuesFixer()
|
|
stripper = SDHStripper()
|
|
srt, _ = fixer.from_file(self.path)
|
|
stripped, status = stripper.from_srt(srt)
|
|
if status is True:
|
|
stripped.save(self.path)
|
|
return
|
|
except Exception as e: # Fall through to other methods
|
|
logging.getLogger("Subtitle").debug(f"subby SDH strip failed, falling back: {e!r}")
|
|
|
|
conversion_method = config.subtitle.get("conversion_method", "auto")
|
|
use_subtitleedit = sdh_method == "subtitleedit" or (
|
|
sdh_method == "auto" and conversion_method in ("auto", "subtitleedit")
|
|
)
|
|
|
|
if binaries.SubtitleEdit and use_subtitleedit:
|
|
from unshackle.core.tracks.subtitle_convert import SUBTITLE_EDIT_FORMATS, subtitleedit_args
|
|
|
|
output_format = SUBTITLE_EDIT_FORMATS.get(self.codec, self.codec.name.lower())
|
|
se_start = time.monotonic()
|
|
subprocess.run(
|
|
subtitleedit_args(
|
|
binaries.SubtitleEdit, self.path, output_format, output_folder=self.path.parent, remove_hi=True
|
|
),
|
|
check=True,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
log_tool_run(
|
|
"SubtitleEdit strip SDH",
|
|
"SubtitleEdit",
|
|
0,
|
|
duration_ms=round((time.monotonic() - se_start) * 1000, 1),
|
|
format=output_format,
|
|
)
|
|
else:
|
|
if config.subtitle.get("convert_before_strip", True) and self.codec != Subtitle.Codec.SubRip:
|
|
# Filter reads SRT only; force the (possibly lossy) conversion so content and
|
|
# codec label stay in sync. convert() relabels path/codec only on real conversion.
|
|
self.convert(Subtitle.Codec.SubRip, forced=True)
|
|
|
|
try:
|
|
sub = Subtitles(self.path)
|
|
try:
|
|
sub.filter(
|
|
rm_fonts=True, rm_ast=True, rm_music=True, rm_effects=True, rm_names=True, rm_author=True
|
|
)
|
|
except ValueError as e:
|
|
if "too many values to unpack" in str(e):
|
|
# Retry without name removal if the error is due to multiple colons in time references
|
|
# This can happen with lines like "at 10:00 and 2:00"
|
|
sub = Subtitles(self.path)
|
|
sub.filter(
|
|
rm_fonts=True, rm_ast=True, rm_music=True, rm_effects=True, rm_names=False, rm_author=True
|
|
)
|
|
else:
|
|
raise
|
|
sub.save()
|
|
except (IOError, OSError) as e:
|
|
if "is not valid subtitle file" in str(e):
|
|
log = logging.getLogger("Subtitle")
|
|
log.warning(f"Failed to strip SDH from {self.path.name}, keeping it as is: {e}")
|
|
else:
|
|
raise
|
|
|
|
def reverse_rtl(self) -> None:
|
|
"""
|
|
Reverse RTL (Right to Left) Start/End on Captions.
|
|
Use this to fix the positioning of sentence-ending characters.
|
|
"""
|
|
if not self.path or not self.path.exists():
|
|
raise ValueError("You must download the subtitle track first.")
|
|
|
|
if not binaries.SubtitleEdit:
|
|
raise EnvironmentError("SubtitleEdit executable not found...")
|
|
|
|
from unshackle.core.tracks.subtitle_convert import SUBTITLE_EDIT_FORMATS, subtitleedit_args
|
|
|
|
output_format = SUBTITLE_EDIT_FORMATS.get(self.codec, self.codec.name.lower())
|
|
|
|
se_start = time.monotonic()
|
|
subprocess.run(
|
|
subtitleedit_args(
|
|
binaries.SubtitleEdit, self.path, output_format, output_folder=self.path.parent, reverse_rtl=True
|
|
),
|
|
check=True,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
log_tool_run(
|
|
"SubtitleEdit reverse RTL",
|
|
"SubtitleEdit",
|
|
0,
|
|
duration_ms=round((time.monotonic() - se_start) * 1000, 1),
|
|
format=output_format,
|
|
)
|
|
|
|
|
|
__all__ = ("Subtitle",)
|