Octo 11545

docs/changelog.rst

@@ -1,5 +1,12 @@
 Changelog
 ---------
+2.3.4
+^^^^^^
+  - Auto-detect and repair double-encoded UTF-8 in all readers. When input
+    text has been misread as CP-1252 and re-encoded (e.g. ♪ stored as ♪),
+    ``BaseReader._decode_content()`` now reverses the corruption and logs a
+    warning. Clean UTF-8 input is never modified.
+
 2.3.3
 ^^^^^^
   - All readers (SCC, DFXP, WebVTT, SAMI, SRT, MicroDVD) now accept

docs/conf.py

@@ -53,9 +53,9 @@
 # built documents.
 #
 # The short X.Y version.
-version = "2.3.3"
+version = "2.3.4.dev1"
 # The full version, including alpha/beta/rc tags.
-release = "2.3.3"
+release = "2.3.4.dev1"
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

pycaption/base.py

@@ -5,12 +5,15 @@
 CaptionConverter orchestrator and base classes for readers/writers.
 """
 
+import logging
 import os
 from datetime import timedelta
 from numbers import Number
 
 from .exceptions import CaptionReadError, CaptionReadTimingError, InvalidInputError
 
+logger = logging.getLogger(__name__)
+
 # `und` a special identifier for an undetermined language according to ISO 639-2
 DEFAULT_LANGUAGE_CODE = os.getenv("PYCAPTION_DEFAULT_LANG", "und")
 
@@ -79,18 +82,31 @@ def _decode_content(content):
             try:
                 content = content.decode("utf-8-sig")
             except UnicodeDecodeError as e:
-                raise InvalidInputError(
-                    f"Content is not valid UTF-8: {e}"
-                ) from e
+                raise InvalidInputError(f"Content is not valid UTF-8: {e}") from e
         elif isinstance(content, str):
             if content.startswith(""):
                 content = content[1:]
         else:
             raise InvalidInputError(
                 "The content must be a unicode string or UTF-8 bytes."
             )
+        content = BaseReader._repair_double_encoding(content)
         return content
 
+    @staticmethod
+    def _repair_double_encoding(text):
+        """Fix double-encoded UTF-8 (bytes misread as CP-1252)."""
+        try:
+            repaired = text.encode("cp1252").decode("utf-8")
+        except (UnicodeEncodeError, UnicodeDecodeError):
+            return text
+        if repaired != text:
+            logger.warning(
+                "Detected and repaired double-encoded UTF-8 in caption content"
+            )
+            return repaired
+        return text
+
     def detect(self, content):
         """Return True if content appears to be in this reader's format.
 
@@ -396,7 +412,11 @@ class CaptionSet:
     """
 
     def __init__(
-        self, captions, styles=None, layout_info=None, regions=None,
+        self,
+        captions,
+        styles=None,
+        layout_info=None,
+        regions=None,
         visual_alignment_default=None,
     ):
         """

setup.py

@@ -20,7 +20,7 @@
 
 setup(
     name="pycaption",
-    version="2.3.3",
+    version="2.3.4.dev1",
     description="Closed caption converter",
     long_description=open(README_PATH).read(),
     author="Joe Norton",

tests/test_bytes_input.py

@@ -0,0 +1,161 @@
+"""Tests for bytes input support across all readers.
+
+All readers should accept bytes (raw file content) and decode as UTF-8
+internally, preventing the double-encoding gibberish that occurs when
+callers decode with the wrong system encoding (e.g., cp1252).
+"""
+
+from pycaption import (
+    DFXPReader,
+    MicroDVDReader,
+    SAMIReader,
+    SCCReader,
+    SRTReader,
+    WebVTTReader,
+)
+
+
+class TestSRTReaderBytes:
+    def setup_class(self):
+        self.reader = SRTReader()
+
+    def test_read_bytes_produces_same_result_as_str(self, sample_srt):
+        from_str = self.reader.read(sample_srt)
+        from_bytes = self.reader.read(sample_srt.encode("utf-8"))
+        str_captions = from_str.get_captions("en-US")
+        bytes_captions = from_bytes.get_captions("en-US")
+        assert len(str_captions) == len(bytes_captions)
+        for s, b in zip(str_captions, bytes_captions):
+            assert s.get_text() == b.get_text()
+            assert s.start == b.start
+            assert s.end == b.end
+
+    def test_read_bytes_preserves_music_notes(self, sample_srt):
+        captions = self.reader.read(sample_srt.encode("utf-8"))
+        texts = [c.get_text() for c in captions.get_captions("en-US")]
+        assert any("♪" in t for t in texts)
+
+    def test_read_bytes_with_bom(self, sample_srt):
+        content_with_bom = b"\xef\xbb\xbf" + sample_srt.encode("utf-8")
+        captions = self.reader.read(content_with_bom)
+        assert len(captions.get_captions("en-US")) == 7
+
+    def test_detect_bytes(self, sample_srt):
+        assert self.reader.detect(sample_srt.encode("utf-8")) is True
+
+
+class TestWebVTTReaderBytes:
+    def setup_class(self):
+        self.reader = WebVTTReader()
+
+    def test_read_bytes_produces_same_result_as_str(self, sample_webvtt):
+        from_str = self.reader.read(sample_webvtt)
+        from_bytes = self.reader.read(sample_webvtt.encode("utf-8"))
+        str_captions = from_str.get_captions("en-US")
+        bytes_captions = from_bytes.get_captions("en-US")
+        assert len(str_captions) == len(bytes_captions)
+        for s, b in zip(str_captions, bytes_captions):
+            assert s.get_text() == b.get_text()
+
+    def test_read_bytes_with_bom(self, sample_webvtt):
+        content_with_bom = b"\xef\xbb\xbf" + sample_webvtt.encode("utf-8")
+        captions = self.reader.read(content_with_bom)
+        assert len(captions.get_captions("en-US")) > 0
+
+    def test_detect_bytes(self, sample_webvtt):
+        assert self.reader.detect(sample_webvtt.encode("utf-8")) is True
+
+
+class TestDFXPReaderBytes:
+    def setup_class(self):
+        self.reader = DFXPReader()
+
+    def test_read_bytes_produces_same_result_as_str(self, sample_dfxp):
+        from_str = self.reader.read(sample_dfxp)
+        from_bytes = self.reader.read(sample_dfxp.encode("utf-8"))
+        for lang in from_str.get_languages():
+            str_captions = from_str.get_captions(lang)
+            bytes_captions = from_bytes.get_captions(lang)
+            assert len(str_captions) == len(bytes_captions)
+            for s, b in zip(str_captions, bytes_captions):
+                assert s.get_text() == b.get_text()
+
+    def test_read_bytes_with_bom(self, sample_dfxp):
+        content_with_bom = b"\xef\xbb\xbf" + sample_dfxp.encode("utf-8")
+        captions = self.reader.read(content_with_bom)
+        assert not captions.is_empty()
+
+    def test_detect_bytes(self, sample_dfxp):
+        assert self.reader.detect(sample_dfxp.encode("utf-8")) is True
+
+
+class TestSAMIReaderBytes:
+    def setup_class(self):
+        self.reader = SAMIReader()
+
+    def test_read_bytes_produces_same_result_as_str(self, sample_sami):
+        from_str = self.reader.read(sample_sami)
+        from_bytes = self.reader.read(sample_sami.encode("utf-8"))
+        for lang in from_str.get_languages():
+            str_captions = from_str.get_captions(lang)
+            bytes_captions = from_bytes.get_captions(lang)
+            assert len(str_captions) == len(bytes_captions)
+            for s, b in zip(str_captions, bytes_captions):
+                assert s.get_text() == b.get_text()
+
+    def test_read_bytes_preserves_music_notes(self, sample_sami):
+        captions = self.reader.read(sample_sami.encode("utf-8"))
+        langs = list(captions.get_languages())
+        texts = [c.get_text() for c in captions.get_captions(langs[0])]
+        assert any("♪" in t for t in texts)
+
+    def test_read_bytes_with_bom(self, sample_sami):
+        content_with_bom = b"\xef\xbb\xbf" + sample_sami.encode("utf-8")
+        captions = self.reader.read(content_with_bom)
+        assert not captions.is_empty()
+
+    def test_detect_bytes(self, sample_sami):
+        assert self.reader.detect(sample_sami.encode("utf-8")) is True
+
+
+class TestSCCReaderBytes:
+    def test_read_bytes_produces_same_result_as_str(self, sample_scc_pop_on):
+        from_str = SCCReader().read(sample_scc_pop_on)
+        from_bytes = SCCReader().read(sample_scc_pop_on.encode("utf-8"))
+        str_captions = from_str.get_captions("en-US")
+        bytes_captions = from_bytes.get_captions("en-US")
+        assert len(str_captions) == len(bytes_captions)
+        for s, b in zip(str_captions, bytes_captions):
+            assert s.get_text() == b.get_text()
+
+    def test_read_bytes_with_bom(self, sample_scc_pop_on):
+        content_with_bom = b"\xef\xbb\xbf" + sample_scc_pop_on.encode("utf-8")
+        captions = SCCReader().read(content_with_bom)
+        assert len(captions.get_captions("en-US")) > 0
+
+    def test_detect_bytes(self, sample_scc_pop_on):
+        assert SCCReader().detect(sample_scc_pop_on.encode("utf-8")) is True
+
+
+class TestMicroDVDReaderBytes:
+    def setup_class(self):
+        self.reader = MicroDVDReader()
+
+    def test_read_bytes_produces_same_result_as_str(self, sample_microdvd):
+        from_str = self.reader.read(sample_microdvd)
+        from_bytes = self.reader.read(sample_microdvd.encode("utf-8"))
+        str_captions = from_str.get_captions("und")
+        bytes_captions = from_bytes.get_captions("und")
+        assert len(str_captions) == len(bytes_captions)
+        for s, b in zip(str_captions, bytes_captions):
+            assert s.get_text() == b.get_text()
+
+    def test_read_bytes_with_bom(self, sample_microdvd):
+        content_with_bom = b"\xef\xbb\xbf" + sample_microdvd.encode("utf-8")
+        captions = self.reader.read(content_with_bom)
+        assert not captions.is_empty()
+
+    def test_detect_bytes(self, sample_microdvd):
+        assert self.reader.detect(sample_microdvd.encode("utf-8")) is True
+
+

tests/test_double_encoding.py

@@ -0,0 +1,89 @@
+import logging
+
+import pytest
+
+from pycaption import SAMIReader, SRTReader
+from pycaption.base import BaseReader
+from pycaption.dfxp import DFXPReader
+
+
+def _double_encode(text):
+    """Simulate double-encoding: UTF-8 bytes misread as CP-1252, re-encoded."""
+    return text.encode("utf-8").decode("cp1252")
+
+
+ORIGINAL_CHARS = ["♪", "—", "’", "é"]
+
+
+class TestRepairDoubleEncoding:
+    @pytest.mark.parametrize("original", ORIGINAL_CHARS)
+    def test_fixes_double_encoded(self, original):
+        garbled = _double_encode(original)
+        assert BaseReader._repair_double_encoding(garbled) == original
+
+    def test_leaves_clean_utf8_alone(self):
+        clean = "♪ This is — perfectly fine é text"
+        assert BaseReader._repair_double_encoding(clean) == clean
+
+    def test_logs_warning_on_repair(self, caplog):
+        garbled = _double_encode("♪")
+        with caplog.at_level(logging.WARNING, logger="pycaption.base"):
+            BaseReader._repair_double_encoding(garbled)
+        assert "double-encoded" in caplog.text.lower()
+
+    def test_no_warning_for_clean_input(self, caplog):
+        with caplog.at_level(logging.WARNING, logger="pycaption.base"):
+            BaseReader._repair_double_encoding("♪ Music ♪")
+        assert caplog.text == ""
+
+
+class TestDoubleEncodingEndToEnd:
+    def test_srt_reader(self):
+        garbled_note = _double_encode("♪")
+        content = (
+            "1\n"
+            "00:00:01,000 --> 00:00:02,000\n"
+            f"{garbled_note} Music {garbled_note}\n"
+        )
+        captions = SRTReader().read(content)
+        nodes = captions.get_captions("en-US")[0].nodes
+        text = "".join(n.content for n in nodes)
+        assert "♪" in text
+        assert garbled_note not in text
+
+    def test_dfxp_reader(self):
+        garbled = _double_encode("élève")
+        content = (
+            '<?xml version="1.0" encoding="UTF-8"?>\n'
+            '<tt xml:lang="en" xmlns="http://www.w3.org/ns/ttml">\n'
+            "  <body><div>\n"
+            f'    <p begin="00:00:01.000" end="00:00:02.000">'
+            f"{garbled}</p>\n"
+            "  </div></body>\n"
+            "</tt>\n"
+        )
+        captions = DFXPReader().read(content)
+        nodes = captions.get_captions("en")[0].nodes
+        text = "".join(n.content for n in nodes)
+        assert "élève" in text
+        assert garbled not in text
+
+    def test_sami_reader(self):
+        garbled_dash = _double_encode("—")
+        content = (
+            "<SAMI>\n"
+            "<Body>\n"
+            "  <Sync Start=1000>\n"
+            f"    <P Class=enCC>{garbled_dash} Hello</P>\n"
+            "  </Sync>\n"
+            "  <Sync Start=2000>\n"
+            "    <P Class=enCC>&nbsp;</P>\n"
+            "  </Sync>\n"
+            "</Body>\n"
+            "</SAMI>\n"
+        )
+        captions = SAMIReader().read(content)
+        lang = list(captions.get_languages())[0]
+        nodes = captions.get_captions(lang)[0].nodes
+        text = "".join(n.content for n in nodes)
+        assert "—" in text