fix: Patched a reproducible text-detection bug

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"
 

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()

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.