diff --git a/docs/changelog.rst b/docs/changelog.rst
index b6025644..fc374bb9 100644
--- a/docs/changelog.rst
+++ b/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
diff --git a/docs/conf.py b/docs/conf.py
index 5bc3c371..7cf9d12d 100644
--- a/docs/conf.py
+++ b/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.
diff --git a/pycaption/base.py b/pycaption/base.py
index 62965b04..5cafdb21 100644
--- a/pycaption/base.py
+++ b/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,9 +82,7 @@ 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:]
@@ -89,8 +90,23 @@ def _decode_content(content):
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,
):
"""
diff --git a/setup.py b/setup.py
index b409890c..398f31eb 100644
--- a/setup.py
+++ b/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",
diff --git a/tests/test_bytes_input.py b/tests/test_bytes_input.py
new file mode 100644
index 00000000..2d292fe4
--- /dev/null
+++ b/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
+
+
diff --git a/tests/test_double_encoding.py b/tests/test_double_encoding.py
new file mode 100644
index 00000000..2cad3d72
--- /dev/null
+++ b/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 = (
+ '\n'
+ '\n'
+ " \n"
+ f'
'
+ f"{garbled}
\n"
+ "
{garbled_dash} Hello
\n"
+ " \n"
+ " \n"
+ "
\n"
+ " \n"
+ "\n"
+ "\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