fix(subs): stop segmented WebVTT merge from stripping formatting

Merge cues at the string level instead of round-tripping through
pycaption, which dropped inline tags and cue settings. STYLE and
REGION blocks are kept, duplicate boundary cues splice correctly,
and a bad timing line now skips the cue instead of crashing.

Also rank pysubs2 above SubtitleEdit for ASS to TTML/VTT since
SubtitleEdit flattens inline styling there. Documented the measured
backend fidelity table and added regression tests for the merge.

docs/guide/subtitles.md

@@ -226,6 +226,47 @@ With `conversion_method: auto`, unshackle ranks these automatically per conversi
 Setting it to a specific value pins that backend as the first choice, falling back to
 others only if the pin cannot handle the pair.
 
+#### What each backend actually preserves
+
+The table below comes from round-tripping a subtitle that carries italics, bold,
+underline, positioning, and colour through every backend. Each cell lists the
+styling that survives; `—` means the backend cannot handle that pair.
+
+| Conversion | `subtitleedit` | `pysubs2` | `subby` | `pycaption` |
+| --- | --- | --- | --- | --- |
+| WebVTT → SRT | italic, bold, underline, position | italic, underline | italic, position | *(strips all)* |
+| WebVTT → ASS | italic, bold, underline, position | italic, bold, underline | — | — |
+| WebVTT → TTML | italic, bold, underline, position | italic, bold, underline | position | *(strips all)* |
+| ASS/SSA → SRT | all (+ colour) | italic, underline | — | — |
+| ASS/SSA → TTML | position only | italic, bold, underline | — | — |
+| ASS/SSA → VTT | position only | italic, underline | — | — |
+
+Reading the table:
+
+- **Keep the original**: leaving `--sub-format` unset (or set to `original`) never
+  round-trips the file, so every style survives. Only convert when your player cannot
+  read the source format.
+- **`subtitleedit`** (SubtitleEdit / `seconv`): the best choice for anything → SRT. It
+  is the only backend that carries colour, and it embeds `{\an8}` tags to preserve
+  positioning. When writing TTML or WebVTT from ASS it flattens inline styling to plain
+  text and keeps only positioning, so avoid it there.
+- **`pysubs2`**: keeps inline italic/underline on every pair, and bold except when
+  writing SRT or WebVTT. It never carries positioning or colour. SSA/ASS is its native
+  model, which makes it the best pick for SSA↔ASS.
+- **`subby`**: reads only WebVTT/fVTT/SAMI (never ASS) and is tuned for → SRT, where it
+  uniquely converts WebVTT cue settings into `{\an8}` positioning. Its
+  `CommonIssuesFixer` may also drop near-duplicate cues.
+- **`pycaption`**: strips all styling; last-resort fallback only.
+
+!!! tip "What `auto` picks"
+    `auto` prefers `subtitleedit` first when it is installed, then falls back to `subby`
+    for → SRT and `pysubs2` otherwise. The one exception comes from the table above:
+    for ASS/SSA → TTML or WebVTT it picks `pysubs2` even when SubtitleEdit is
+    installed, since SubtitleEdit flattens inline styling on those pairs. Most installs
+    do not ship SubtitleEdit, so in practice `auto` means `subby` (→ SRT) or `pysubs2`.
+    Install [SubtitleEdit / `seconv`](#subtitle-configuration) if you need colour or the
+    highest → SRT fidelity.
+
 !!! note "Config wins over a service's preference"
     A service may set a `preferred_conversion_method` on its own tracks (for example when
     it ships subtitles that a particular backend handles best). An explicit

tests/tracks/test_subtitle_convert.py

@@ -87,6 +87,18 @@ def test_subtitleedit_ranks_first_when_available(monkeypatch):
     assert chain[0] == "subtitleedit"
 
 
+def test_styled_ass_to_ttml_vtt_prefers_pysubs2_over_subtitleedit(monkeypatch):
+    # Measured: SE flattens ASS inline styling to TTML/VTT; pysubs2 keeps i/b/u.
+    monkeypatch.setattr(binaries, "SubtitleEdit", "/usr/bin/seconv")
+    for target in (Codec.TimedTextMarkupLang, Codec.WebVTT):
+        chain = [b.name for b in sc.resolve_backends(Codec.SubStationAlphav4, target)]
+        assert chain[0] == "pysubs2", f"expected pysubs2 first for ASS->{target.name}, got {chain}"
+        assert "subtitleedit" in chain  # still a fallback
+    # ASS->SRT keeps SE first (only backend that keeps colour there).
+    chain = [b.name for b in sc.resolve_backends(Codec.SubStationAlphav4, Codec.SubRip)]
+    assert chain[0] == "subtitleedit"
+
+
 def test_pin_then_fallback_orders_pin_first():
     chain = [b.name for b in sc.resolve_backends(Codec.WebVTT, Codec.SubRip, pin="pysubs2")]
     assert chain[0] == "pysubs2"

tests/tracks/test_webvtt_merge.py

@@ -0,0 +1,113 @@
+"""Regression tests for segmented-WebVTT merging (``core/utils/webvtt.py``).
+
+The merge used to round-trip cues through pycaption's reader/writer (and a pysubs2
+pre-normalization step), both of which silently dropped inline formatting despite
+``preserve_formatting=True``. It is now a pure string-level splice: cue
+timing lines, settings, and payload are kept verbatim, so span tags
+(``<i>``/``<u>``/``<b>``), ASS-style overrides (``{\\an8}``), and entities all
+survive, while duplicate headers collapse and boundary-duplicate cues dedupe.
+"""
+
+from __future__ import annotations
+
+from unshackle.core.utils.webvtt import merge_segmented_webvtt
+
+# Two segments, each a full WebVTT document with its own X-TIMESTAMP-MAP header,
+# as they arrive concatenated from a DASH/HLS subtitle stream. The first cue carries
+# span tags, an {\an8} override, a cue setting, and a literal &lt; entity.
+SEGMENTED_VTT = (
+    "WEBVTT\n"
+    "X-TIMESTAMP-MAP=MPEGTS:900000,LOCAL:00:00:00.000\n\n"
+    "00:00:01.000 --> 00:00:03.000 line:10% align:center\n"
+    "<i>Hello</i> {\\an8}world <u>und</u> <b>bold</b> 5 &lt; 6\n\n"
+    "WEBVTT\n"
+    "X-TIMESTAMP-MAP=MPEGTS:1800000,LOCAL:00:00:00.000\n\n"
+    "00:00:04.000 --> 00:00:06.000\n"
+    "<i>Second</i> cue\n"
+)
+
+
+def test_merge_preserves_inline_formatting():
+    out = merge_segmented_webvtt(SEGMENTED_VTT, segment_durations=[900000, 900000], timescale=1)
+    for tag in ("<i>", "</i>", "<u>", "</u>", "<b>", "</b>", "{\\an8}"):
+        assert tag in out, f"{tag!r} was stripped from merged output:\n{out}"
+
+
+def test_merge_preserves_cue_settings_and_entities():
+    out = merge_segmented_webvtt(SEGMENTED_VTT, segment_durations=[900000, 900000], timescale=1)
+    assert "line:10% align:center" in out  # cue positioning survives
+    assert "&lt;" in out  # a literal entity is not turned into a stray tag
+
+
+def test_merge_collapses_headers_into_single_document():
+    out = merge_segmented_webvtt(SEGMENTED_VTT, segment_durations=[900000, 900000], timescale=1)
+    assert out.count("WEBVTT") == 1
+    assert "X-TIMESTAMP-MAP" not in out
+
+
+def test_merge_dedupes_boundary_duplicate_cues():
+    # A cue that spans a segment boundary is emitted in both segments; the merge
+    # should collapse the adjacent identical pair into one.
+    dup = SEGMENTED_VTT + (
+        "\nWEBVTT\n"
+        "X-TIMESTAMP-MAP=MPEGTS:2700000,LOCAL:00:00:00.000\n\n"
+        "00:00:04.000 --> 00:00:06.000\n"
+        "<i>Second</i> cue\n"
+    )
+    out = merge_segmented_webvtt(dup, segment_durations=[900000, 900000, 900000], timescale=1)
+    assert out.count("<i>Second</i> cue") == 1, f"boundary duplicate not deduped:\n{out}"
+
+
+def test_merge_keeps_timing_lines_verbatim():
+    # String-level merge must not re-render timestamps (pycaption shortened
+    # 00:00:01.000 -> 00:01.000) and must keep the settings tail untouched.
+    out = merge_segmented_webvtt(SEGMENTED_VTT, segment_durations=[900000, 900000], timescale=1)
+    assert "00:00:01.000 --> 00:00:03.000 line:10% align:center" in out
+    assert "00:00:04.000 --> 00:00:06.000" in out
+
+
+def test_merge_splices_triple_duplicate_into_one():
+    # A cue spanning two boundaries appears in three consecutive segments; the
+    # kept cue must be extended each time (latent bug: extending the dropped one).
+    vtt = (
+        "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nSame cue\n\n"
+        "WEBVTT\n\n00:00:02.000 --> 00:00:03.000\nSame cue\n\n"
+        "WEBVTT\n\n00:00:03.000 --> 00:00:04.000\nSame cue\n"
+    )
+    out = merge_segmented_webvtt(vtt)
+    assert out.count("Same cue") == 1
+    assert "00:00:01.000 --> 00:00:04.000" in out
+
+
+def test_merge_drops_cue_ids_and_notes_keeps_style():
+    vtt = (
+        "WEBVTT\n\n"
+        "STYLE\n::cue { color: red }\n\n"
+        "NOTE a comment\n\n"
+        "some-cue-id-42\n00:00:01.000 --> 00:00:02.000\nHello\n"
+    )
+    out = merge_segmented_webvtt(vtt)
+    assert "Hello" in out
+    assert "some-cue-id-42" not in out
+    assert "NOTE" not in out
+    assert "STYLE\n::cue { color: red }" in out  # kept; players without STYLE support ignore it
+
+
+def test_merge_dedupes_repeated_style_blocks_before_cues():
+    # Every segment repeats the same STYLE block; keep exactly one, before the first cue.
+    style = "WEBVTT\n\nSTYLE\n::cue { color: red }\n\n"
+    vtt = style + "00:00:01.000 --> 00:00:02.000\nOne\n\n" + style + "00:00:03.000 --> 00:00:04.000\nTwo\n"
+    out = merge_segmented_webvtt(vtt)
+    assert out.count("STYLE") == 1
+    assert out.index("STYLE") < out.index("-->")
+
+
+def test_merge_skips_malformed_timing_line():
+    vtt = (
+        "WEBVTT\n\n"
+        "garbage --> not-a-time\nBroken\n\n"
+        "00:00:01.000 --> 00:00:02.000\nGood\n"
+    )
+    out = merge_segmented_webvtt(vtt)
+    assert "Good" in out
+    assert "Broken" not in out

unshackle/core/tracks/subtitle_convert.py

@@ -154,6 +154,11 @@ class SubtitleEditBackend:
         return source in self.reads and target in self.writes
 
     def rank(self, source: Codec, target: Codec) -> int:
+        if source in (Codec.SubStationAlpha, Codec.SubStationAlphav4) and target in (
+            Codec.TimedTextMarkupLang,
+            Codec.WebVTT,
+        ):
+            return 3
         return 0
 
     def convert(self, source: Codec, src: Path, target: Codec, out: Path) -> None:

unshackle/core/utils/webvtt.py

@@ -1,225 +1,74 @@
 import re
-import sys
-import typing
 from typing import Optional
 
-import pysubs2
+# Timing line: optional hours, verbatim settings tail.
-from pycaption import Caption, CaptionList, CaptionNode, CaptionReadError, WebVTTReader, WebVTTWriter
+_TIMING = re.compile(r"^((?:\d+:)?\d{2}:\d{2}\.\d{3})[ \t]+-->[ \t]+((?:\d+:)?\d{2}:\d{2}\.\d{3})(.*)$")
-
-from unshackle.core.config import config
 
 
-class CaptionListExt(CaptionList):
+def _timestamp_ms(ts: str) -> int:
-    @typing.no_type_check
+    hms, ms = ts.rsplit(".", 1)
-    def __init__(self, iterable=None, layout_info=None):
+    parts = [int(p) for p in hms.split(":")]
-        self.first_segment_mpegts = 0
+    while len(parts) < 3:
-        super().__init__(iterable, layout_info)
+        parts.insert(0, 0)
-
+    hours, minutes, seconds = parts
-
+    return ((hours * 60 + minutes) * 60 + seconds) * 1000 + int(ms)
-class CaptionExt(Caption):
-    @typing.no_type_check
-    def __init__(self, start, end, nodes, style=None, layout_info=None, segment_index=0, mpegts=0, cue_time=0.0):
-        style = style or {}
-        self.segment_index: int = segment_index
-        self.mpegts: float = mpegts
-        self.cue_time: float = cue_time
-        super().__init__(start, end, nodes, style, layout_info)
-
-
-class WebVTTReaderExt(WebVTTReader):
-    # HLS extension support <https://datatracker.ietf.org/doc/html/rfc8216#section-3.5>
-    RE_TIMESTAMP_MAP = re.compile(r"X-TIMESTAMP-MAP.*")
-    RE_MPEGTS = re.compile(r"MPEGTS:(\d+)")
-    RE_LOCAL = re.compile(r"LOCAL:((?:(\d{1,}):)?(\d{2}):(\d{2})\.(\d{3}))")
-
-    def _parse(self, lines: list[str]) -> CaptionList:
-        captions = CaptionListExt()
-        start = None
-        end = None
-        nodes: list[CaptionNode] = []
-        layout_info = None
-        found_timing = False
-        segment_index = -1
-        mpegts = 0
-        cue_time = 0.0
-
-        # The first segment MPEGTS is needed to calculate the rest. It is possible that
-        # the first segment contains no cue and is ignored by pycaption, this acts as a fallback.
-        captions.first_segment_mpegts = 0
-
-        for i, line in enumerate(lines):
-            if "-->" in line:
-                found_timing = True
-                timing_line = i
-                last_start_time = captions[-1].start if captions else 0
-                try:
-                    start, end, layout_info = self._parse_timing_line(line, last_start_time)
-                except CaptionReadError as e:
-                    new_msg = f"{e.args[0]} (line {timing_line})"
-                    tb = sys.exc_info()[2]
-                    raise type(e)(new_msg).with_traceback(tb) from None
-
-            elif "" == line:
-                if found_timing and nodes:
-                    found_timing = False
-                    caption = CaptionExt(
-                        start,
-                        end,
-                        nodes,
-                        layout_info=layout_info,
-                        segment_index=segment_index,
-                        mpegts=mpegts,
-                        cue_time=cue_time,
-                    )
-                    captions.append(caption)
-                    nodes = []
-
-            elif "WEBVTT" in line:
-                # Merged segmented VTT doesn't have index information, track manually.
-                segment_index += 1
-                mpegts = 0
-                cue_time = 0.0
-            elif m := self.RE_TIMESTAMP_MAP.match(line):
-                if r := self.RE_MPEGTS.search(m.group()):
-                    mpegts = int(r.group(1))
-
-                cue_time = self._parse_local(m.group())
-
-                # Early assignment in case the first segment contains no cue.
-                if segment_index == 0:
-                    captions.first_segment_mpegts = mpegts
-
-            else:
-                if found_timing:
-                    if nodes:
-                        nodes.append(CaptionNode.create_break())
-                    nodes.append(CaptionNode.create_text(self._decode(line)))
-                else:
-                    # it's a comment or some metadata; ignore it
-                    pass
-
-        # Add a last caption if there are remaining nodes
-        if nodes:
-            caption = CaptionExt(start, end, nodes, layout_info=layout_info, segment_index=segment_index, mpegts=mpegts)
-            captions.append(caption)
-
-        return captions
-
-    @staticmethod
-    def _parse_local(string: str) -> float:
-        """
-        Parse WebVTT LOCAL time and convert it to seconds.
-        """
-        m = WebVTTReaderExt.RE_LOCAL.search(string)
-        if not m:
-            return 0
-
-        parsed = m.groups()
-        if not parsed:
-            return 0
-        hours = int(parsed[1])
-        minutes = int(parsed[2])
-        seconds = int(parsed[3])
-        milliseconds = int(parsed[4])
-        return (milliseconds / 1000) + seconds + (minutes * 60) + (hours * 3600)
 
 
 def merge_segmented_webvtt(vtt_raw: str, segment_durations: Optional[list[int]] = None, timescale: int = 1) -> str:
     """
-    Merge Segmented WebVTT data.
+    Merge concatenated Segmented WebVTT documents into a single document.
 
-    Parameters:
+    Strips the per-segment WEBVTT and X-TIMESTAMP-MAP headers and splices cues
-        vtt_raw: The concatenated WebVTT files to merge. All WebVTT headers must be
+    repeated across a segment boundary (identical payload, <=1ms gap or overlap).
-            appropriately spaced apart, or it may produce unwanted effects like
+    Timing lines, cue settings and payload are kept verbatim, so inline formatting
-            considering headers as captions, timestamp lines, etc.
+    (span tags, entities, ASS-style overrides) survives untouched.
-        segment_durations: A list of each segment's duration. If not provided it will try
-            to get it from the X-TIMESTAMP-MAP headers, specifically the MPEGTS number.
-        timescale: The number of time units per second.
 
-    This parses the X-TIMESTAMP-MAP data to compute new absolute timestamps, replacing
+    No timestamp offset is applied: segments are expected to carry absolute cue
-    the old start and end timestamp values. All X-TIMESTAMP-MAP header information will
+    times already (same guard as shaka-player and N_m3u8DL-RE; re-offsetting
-    be removed from the output as they are no longer of concern. Consider this function
+    double-shifts such streams). segment_durations/timescale are accepted for
-    the opposite of a WebVTT Segmenter, a WebVTT Joiner of sorts.
+    call-site compatibility and only matter if offsetting is ever reintroduced.
 
-    Algorithm borrowed from N_m3u8DL-RE and shaka-player.
+    Unique STYLE/REGION blocks are kept (players that don't support them ignore
+    them); cue identifiers and NOTE blocks are dropped; cues with a malformed
+    timing line are skipped (downstream sanitizers catch worse).
     """
-    MPEG_TIMESCALE = 90_000
+    # each kept cue: [start, end, settings, payload, normalized payload for dedupe]
+    cues: list[list[str]] = []
+    headers: list[str] = []  # unique STYLE/REGION blocks, first-seen order
+    prev: Optional[list[str]] = None
+    prev_end_ms = 0
 
-    # Check config for conversion method preference
+    for block in re.split(r"\n[ \t]*\n", vtt_raw.replace("\r\n", "\n").replace("\r", "\n")):
-    conversion_method = config.subtitle.get("conversion_method", "auto")
+        lines = [
-    use_pysubs2 = conversion_method in ("pysubs2", "auto")
+            line
+            for line in block.split("\n")
+            if not (line.startswith("WEBVTT") or line.lstrip().startswith("X-TIMESTAMP-MAP"))
+        ]
+        timing_i = next((i for i, line in enumerate(lines) if "-->" in line), None)
+        if timing_i is None:  # NOTE or other metadata block
+            first = next((line.strip() for line in lines if line.strip()), "")
+            if first.startswith(("STYLE", "REGION")):
+                header = "\n".join(lines).strip("\n")
+                if header not in headers:  # segments repeat the same block
+                    headers.append(header)
+            continue
+        timing = _TIMING.match(lines[timing_i].strip())
+        if not timing:
+            continue
+        start, end, settings = timing.groups()
+        payload = [line for line in lines[timing_i + 1 :] if line.strip()]
+        if not payload:
+            continue
 
-    if use_pysubs2:
+        normalized = "\n".join(line.strip() for line in payload)
-        # Try using pysubs2 first for more lenient parsing
+        if prev is not None and _timestamp_ms(start) - prev_end_ms <= 1 and normalized == prev[4]:
-        try:
+            prev[1] = end  # splice: extend the kept cue, drop the duplicate
-            # Use pysubs2 to parse and normalize the VTT
+            prev_end_ms = _timestamp_ms(end)
-            subs = pysubs2.SSAFile.from_string(vtt_raw)
+            continue
-            # Convert back to WebVTT string for pycaption processing
-            normalized_vtt = subs.to_string("vtt")
-            vtt = WebVTTReaderExt().read(normalized_vtt)
-        except Exception:
-            # Fall back to direct pycaption parsing
-            vtt = WebVTTReaderExt().read(vtt_raw)
-    else:
-        # Use pycaption directly
-        vtt = WebVTTReaderExt().read(vtt_raw)
-    for lang in vtt.get_languages():
-        prev_caption = None
-        duplicate_index: list[int] = []
-        captions = vtt.get_captions(lang)
 
-        # Some providers can produce "segment_index" values that are
+        prev = [start, end, settings, "\n".join(payload), normalized]
-        # outside the provided segment_durations list after normalization/merge.
+        prev_end_ms = _timestamp_ms(end)
-        # This used to crash with IndexError and abort the entire download.
+        cues.append(prev)
-        if segment_durations and captions:
-            max_idx = max(getattr(c, "segment_index", 0) for c in captions)
-            if max_idx >= len(segment_durations):
-                # Pad with the last known duration (or 0 if empty) so indexing is safe.
-                pad_val = segment_durations[-1] if segment_durations else 0
-                segment_durations = segment_durations + [pad_val] * (max_idx - len(segment_durations) + 1)
 
-        if captions[0].segment_index == 0:
+    blocks = headers + [f"{c[0]} --> {c[1]}{c[2]}\n{c[3]}" for c in cues]
-            first_segment_mpegts = captions[0].mpegts
+    return "WEBVTT\n\n" + "\n\n".join(blocks) + "\n"
-        else:
-            first_segment_mpegts = segment_durations[0] if segment_durations else captions.first_segment_mpegts
-
-        caption: CaptionExt
-        for i, caption in enumerate(captions):
-            # DASH WebVTT doesn't have MPEGTS timestamp like HLS. Instead,
-            # calculate the timestamp from SegmentTemplate/SegmentList duration.
-            likely_dash = first_segment_mpegts == 0 and caption.mpegts == 0
-            if likely_dash and segment_durations:
-                # Defensive: segment_index can still be out of range if captions are malformed.
-                if caption.segment_index < 0 or caption.segment_index >= len(segment_durations):
-                    continue
-                duration = segment_durations[caption.segment_index]
-                caption.mpegts = MPEG_TIMESCALE * (duration / timescale)
-
-            if caption.mpegts == 0:
-                continue
-
-            # Commeted to fix DSNP subs being out of sync and mistimed.
-            # seconds = (caption.mpegts - first_segment_mpegts) / MPEG_TIMESCALE - caption.cue_time
-            # offset = seconds * 1_000_000  # pycaption use microseconds
-
-            # if caption.start < offset:
-            #    caption.start += offset
-            #    caption.end += offset
-
-            # If the difference between current and previous captions is <=1ms
-            # and the payload is equal then splice.
-            if (
-                prev_caption
-                and not caption.is_empty()
-                and (caption.start - prev_caption.end) <= 1000  # 1ms in microseconds
-                and caption.get_text() == prev_caption.get_text()
-            ):
-                prev_caption.end = caption.end
-                duplicate_index.append(i)
-
-            prev_caption = caption
-
-        # Remove duplicate
-        captions[:] = [c for c_index, c in enumerate(captions) if c_index not in set(duplicate_index)]
-
-    return WebVTTWriter().write(vtt)