From b43af389648e3edeccefb1a84e8eb440de68a472 Mon Sep 17 00:00:00 2001 From: vladd-bit Date: Thu, 15 Oct 2026 12:31:53 +0100 Subject: [PATCH] Patched a reproducible text-detection bug --- .../ocr_service/processor/converter.py | 5 +- .../ocr_service/tests/test_plain_text.py | 83 +++++++++++++++++++ apps/ocr-service/ocr_service/utils/utils.py | 53 ++++++++---- 3 files changed, 122 insertions(+), 19 deletions(-) create mode 100644 apps/ocr-service/ocr_service/tests/test_plain_text.py diff --git a/apps/ocr-service/ocr_service/processor/converter.py b/apps/ocr-service/ocr_service/processor/converter.py index 32ebe3a..2cef58f 100755 --- a/apps/ocr-service/ocr_service/processor/converter.py +++ b/apps/ocr-service/ocr_service/processor/converter.py @@ -24,6 +24,7 @@ from ocr_service.settings import settings from ocr_service.utils.utils import ( INPUT_FILTERS, + decode_plain_text, delete_tmp_files, is_encrypted_office_document, terminate_hanging_process, @@ -101,7 +102,7 @@ def _extract_text_fallback(self, self.log.warning("Failed to parse RTF during fallback; using raw decode") if not text: - text = stream.decode("utf-8", "ignore") + text = decode_plain_text(stream) or stream.decode("utf-8", "ignore") return unescape(text) @@ -518,7 +519,7 @@ def prepare(self, ctx: ProcessContext) -> None: self.log.info( "Unknown text-like content; treating as plain text, skipping unoserver/LO conversion" ) - ctx.output_text = ctx.stream.decode("utf-8", "ignore") + ctx.output_text = decode_plain_text(ctx.stream) or "" ctx.metadata["pages"] = 1 ctx.metadata["content-type"] = "text/plain" diff --git a/apps/ocr-service/ocr_service/tests/test_plain_text.py b/apps/ocr-service/ocr_service/tests/test_plain_text.py new file mode 100644 index 0000000..2fb5acc --- /dev/null +++ b/apps/ocr-service/ocr_service/tests/test_plain_text.py @@ -0,0 +1,83 @@ +import codecs +import unittest +from unittest.mock import Mock, patch + +from ocr_service.dto.process_context import ProcessContext +from ocr_service.processor.converter import DocumentConverter +from ocr_service.settings import settings +from ocr_service.utils.utils import is_file_content_plain_text, normalise_file_name_with_ext + + +class TestPlainText(unittest.TestCase): + def test_unicode_text_is_extracted_without_libreoffice(self): + text = "Résumé café — Ελληνικά 中文 Привет 😀\nSecond line." + payloads = [ + text.encode("utf-8"), + text.encode("utf-8-sig"), + codecs.BOM_UTF16_LE + text.encode("utf-16-le"), + codecs.BOM_UTF16_BE + text.encode("utf-16-be"), + codecs.BOM_UTF32_LE + text.encode("utf-32-le"), + codecs.BOM_UTF32_BE + text.encode("utf-32-be"), + ] + for mode in ("OCR", "NO_OCR"): + for stream in payloads: + for original_name in ("request-id", "request-id.txt"): + with self.subTest(mode=mode, prefix=stream[:4], name=original_name), patch.object( + settings, "OCR_SERVICE_OPERATION_MODE", mode + ): + name = normalise_file_name_with_ext(original_name, stream) + converter = DocumentConverter(Mock(), {}) + converter._preprocess_doc = Mock(return_value=b"") + ctx = ProcessContext(stream=stream, file_name=name, file_type=None) + + converter.prepare(ctx) + + self.assertEqual(name, "request-id.txt") + self.assertEqual(ctx.output_text, text) + self.assertEqual(ctx.metadata["content-type"], "text/plain") + self.assertEqual(ctx.metadata["pages"], 1) + self.assertNotIn("fallback_reason", ctx.metadata) + converter._preprocess_doc.assert_not_called() + + def test_utf8_character_across_sample_boundary_is_not_treated_as_binary(self): + stream = ("a" * 4095 + "é" + " rest").encode("utf-8") + self.assertTrue(is_file_content_plain_text(stream)) + + def test_binary_content_is_not_mislabeled_as_text(self): + payloads = [ + b"", + b"\x00\x01\x02\x03", + b"a" * 100 + b"\x00", + b"\xff\xfe\x00", # Truncated UTF-16. + b"a" * 4096 + b"\xff", # Invalid UTF-8 after the sample. + ] + for stream in payloads: + with self.subTest(prefix=stream[:8], size=len(stream)): + self.assertFalse(is_file_content_plain_text(stream)) + self.assertEqual(normalise_file_name_with_ext("request-id", stream), "request-id") + + def test_ascii_text_and_whitespace_still_work(self): + self.assertTrue(is_file_content_plain_text(b"Normal text\twith\r\nwhitespace\f\v")) + self.assertTrue(is_file_content_plain_text(b" \t\n")) + + def test_text_suffix_does_not_force_binary_content_through_text_extraction(self): + converter = DocumentConverter(Mock(), {}) + converter._preprocess_doc = Mock(return_value=b"") + stream = b"\x00\x01\x02\x03" + ctx = ProcessContext(stream=stream, file_name="request-id.txt", file_type=None) + + converter.prepare(ctx) + + converter._preprocess_doc.assert_called_once_with(stream, file_name="request-id.txt") + self.assertEqual(ctx.output_text, "") + + def test_text_fallback_preserves_unicode_and_strips_byte_order_mark(self): + converter = DocumentConverter(Mock(), {}) + text = "Résumé — 中文" + for encoding in ("utf-8-sig", "utf-16", "utf-32"): + with self.subTest(encoding=encoding): + self.assertEqual(converter._extract_text_fallback(text.encode(encoding)), text) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/ocr-service/ocr_service/utils/utils.py b/apps/ocr-service/ocr_service/utils/utils.py index f5f789c..8c863f7 100755 --- a/apps/ocr-service/ocr_service/utils/utils.py +++ b/apps/ocr-service/ocr_service/utils/utils.py @@ -6,13 +6,13 @@ converter kept for reference. """ +import codecs import contextlib import fcntl import json import logging import os import shutil -import string import sys import xml.sax import zipfile @@ -32,7 +32,6 @@ logger = logging.getLogger(__name__) -PRINTABLE = set(bytes(string.printable, "ascii")) | {9, 10, 13} OLE_SIGNATURE = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" ODF_MIME_EXTENSIONS: dict[str, str] = { @@ -207,29 +206,49 @@ def delete_tmp_files(file_paths: list[str]) -> None: if os.path.exists(file_path): os.remove(file_path) -def is_file_content_plain_text(stream: bytes, threshold: float = 0.95) -> bool: - """Heuristic to determine whether a byte stream is likely plain text. +def decode_plain_text(stream: bytes, threshold: float = 0.95) -> str | None: + """Decode Unicode text, returning None for binary or invalid input. - Args: - stream: Raw bytes to inspect. - threshold: Ratio of printable ASCII bytes required to treat as text. - - Returns: - bool: True if the stream appears to be text-like. + BOMs identify UTF-16/32; otherwise require UTF-8. Validate the whole + stream before checking printable Unicode characters in a bounded sample. + This avoids both splitting a multibyte character and silently dropping + invalid bytes beyond the sample. """ if not stream: - return False + return None - sample = stream[:4096] + # UTF-32 LE starts with the UTF-16 LE BOM, so check UTF-32 first. + if stream.startswith((codecs.BOM_UTF32_LE, codecs.BOM_UTF32_BE)): + encoding = "utf-32" + elif stream.startswith((codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE)): + encoding = "utf-16" + else: + encoding = "utf-8-sig" - # If it can't be decoded as UTF-8 at all, treat as binary try: - sample.decode("utf-8") + text = stream.decode(encoding) except UnicodeDecodeError: - return False + return None + + if not text or "\x00" in text: + return None - printable = sum(1 for b in sample if b in PRINTABLE) - return printable / len(sample) >= threshold + sample = text[:4096] + printable = sum(char.isprintable() or char in "\t\n\r\v\f" for char in sample) + return text if printable / len(sample) >= threshold else None + + +def is_file_content_plain_text(stream: bytes, threshold: float = 0.95) -> bool: + """Heuristic to determine whether a byte stream is likely Unicode text. + + Args: + stream: Raw bytes to inspect. + threshold: Ratio of printable Unicode characters required to treat as text. + + Returns: + bool: True if the stream appears to be text-like. + """ + return decode_plain_text(stream, threshold) is not None def is_file_type_html(stream: bytes) -> bool: """Detect HTML content by scanning the head of the stream.