From ebdb3321174ba0422445d6a93b3ef90bb9f0a54c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 24 Feb 2026 18:25:53 +0000 Subject: [PATCH 1/3] Initial plan From b03bfb5e026abd02b9284537c4c6257a5da418f3 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 24 Feb 2026 18:41:27 +0000 Subject: [PATCH 2/3] feat(ocr): add AI-based embedded text quality check with automatic OCR fallback Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/config.py | 13 + app/tasks/process_document.py | 64 ++++ app/utils/settings_service.py | 13 + app/utils/text_quality.py | 326 ++++++++++++++++ docs/ConfigurationGuide.md | 22 ++ tests/test_text_quality.py | 703 ++++++++++++++++++++++++++++++++++ 6 files changed, 1141 insertions(+) create mode 100644 app/utils/text_quality.py create mode 100644 tests/test_text_quality.py diff --git a/app/config.py b/app/config.py index aaf0c067..12d3329b 100644 --- a/app/config.py +++ b/app/config.py @@ -276,6 +276,19 @@ class Settings(BaseSettings): ), ) + # Text quality check - AI-based assessment of embedded PDF text + enable_text_quality_check: bool = Field( + default=True, + description=( + "Enable AI-based quality check for embedded PDF text. " + "When enabled, text extracted from non-digital PDFs is evaluated by the AI model. " + "If the text is poor quality (OCR artefacts, typos, incoherence), the file is " + "re-processed with OCR instead of using the embedded text. " + "Digitally-created PDFs (Word, LibreOffice, LaTeX, etc.) are always trusted and " + "bypass the check. Default: True (enabled)." + ), + ) + # Processing step timeout - prevents files from getting stuck in "in_progress" state step_timeout: int = Field( default=600, diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index 0c994d38..4226f0b7 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -17,6 +17,7 @@ from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt from app.tasks.process_with_ocr import process_with_ocr from app.tasks.retry_config import BaseTaskWithRetry from app.utils import get_unique_filepath_with_counter, hash_file, log_task_progress +from app.utils.text_quality import check_text_quality, detect_pdf_text_source logger = logging.getLogger(__name__) @@ -409,6 +410,69 @@ def process_document( file_id=file_id, ) + # ---------------------------------------------------------------- + # AI-based text quality check + # Digitally-created PDFs are always trusted; OCR-sourced or unknown + # PDFs are validated. Poor-quality text triggers automatic re-OCR. + # ---------------------------------------------------------------- + if settings.enable_text_quality_check: + text_source = detect_pdf_text_source(new_local_path) + logger.info(f"[{task_id}] Detected PDF text source: {text_source.value}") + + quality_result = check_text_quality(extracted_text, text_source) + logger.info( + f"[{task_id}] Text quality check result: " + f"good={quality_result.is_good_quality}, score={quality_result.quality_score}, " + f"source={quality_result.text_source.value}, feedback={quality_result.feedback!r}" + ) + + if not quality_result.is_good_quality: + # Poor quality: discard embedded text and re-OCR instead. + issues_str = ", ".join(quality_result.issues) if quality_result.issues else "unspecified" + detail_msg = ( + f"Text quality check FAILED – score={quality_result.quality_score}/100, " + f"source={quality_result.text_source.value}, issues=[{issues_str}].\n" + f"AI feedback: {quality_result.feedback}\n" + f"Embedded text will be ignored; re-running OCR." + ) + logger.warning(f"[{task_id}] {detail_msg}") + log_task_progress( + task_id, + "check_text_quality", + "failure", + f"Poor quality text (score={quality_result.quality_score}/100); queuing OCR", + file_id=file_id, + detail=detail_msg, + ) + log_task_progress( + task_id, + "process_document", + "success", + "Queued for OCR (text quality too low)", + file_id=file_id, + ) + process_with_ocr.delay(new_filename, file_id) + return { + "file": new_local_path, + "status": "Queued for OCR (poor embedded text quality)", + "file_id": file_id, + } + + # Good quality: record the result and proceed with local extraction. + detail_msg = ( + f"Text quality check PASSED – score={quality_result.quality_score}/100, " + f"source={quality_result.text_source.value}.\n" + f"AI feedback: {quality_result.feedback}" + ) + log_task_progress( + task_id, + "check_text_quality", + "success", + f"Text quality OK (score={quality_result.quality_score}/100)", + file_id=file_id, + detail=detail_msg, + ) + # Mark OCR as skipped since we extracted text locally log_task_progress( task_id, diff --git a/app/utils/settings_service.py b/app/utils/settings_service.py index bfd53acb..15df25df 100644 --- a/app/utils/settings_service.py +++ b/app/utils/settings_service.py @@ -1057,6 +1057,19 @@ SETTING_METADATA = { "required": False, "restart_required": False, }, + "enable_text_quality_check": { + "category": "Processing", + "description": ( + "Enable AI-based quality check for embedded PDF text. " + "When enabled, text extracted from non-digital PDFs is evaluated by the AI model. " + "Poor-quality text (OCR artefacts, typos, incoherence) triggers automatic re-OCR. " + "Digitally-created PDFs (Word, LibreOffice, LaTeX, etc.) are always trusted and skip the check." + ), + "type": "boolean", + "sensitive": False, + "required": False, + "restart_required": False, + }, # Notifications Settings "notification_urls": { "category": "Notifications", diff --git a/app/utils/text_quality.py b/app/utils/text_quality.py new file mode 100644 index 00000000..3901633b --- /dev/null +++ b/app/utils/text_quality.py @@ -0,0 +1,326 @@ +"""Utility module for assessing the quality of embedded text in PDF documents. + +This module provides functionality to: + +- Detect whether a PDF's embedded text came from a digital creation process + (e.g., exported from Word, LibreOffice, LaTeX) or a previous OCR pass. +- Assess the quality of extracted text using an AI model. +- Log detailed feedback for debugging and continuous improvement. + +**Rationale** + +Some files contain embedded text that is of poor quality — characterised by +excessive typos, nonsensical content, or textual fragments that do not reflect +the meaning of the document. The most common cause is that the PDF was +previously processed by an OCR engine of varying quality. + +If the embedded text is from a digitally created PDF, the quality is assumed to +be good and no AI check is performed. If the text appears to come from a prior +OCR pass (or the source is unknown), the AI quality check is performed. Poor +quality text triggers automatic re-OCR so that the downstream pipeline operates +on the best available text. +""" + +import json +import logging +import re +from dataclasses import dataclass, field +from enum import Enum +from typing import Optional + +from app.config import settings +from app.utils.ai_provider import get_ai_provider + +logger = logging.getLogger(__name__) + +# Maximum characters of text forwarded to the AI for quality assessment. +_TEXT_SAMPLE_MAX_CHARS = 3000 + +# --------------------------------------------------------------------------- +# Text source detection +# --------------------------------------------------------------------------- + +# Keywords (lower-cased) in /Producer or /Creator that indicate a prior OCR pass. +_OCR_PRODUCER_KEYWORDS: list[str] = [ + "tesseract", + "ocrmypdf", + "abbyy", + "nuance", + "readiris", + "omnipage", + "recognita", + "recogniform", + "acrobat capture", + "pdf ocr", + "exactscan", + "iris ocr", + "prizmo", + "pdfsandwich", + "pdf2searchable", +] + +# Keywords (lower-cased) in /Producer or /Creator that indicate digital authoring. +_DIGITAL_PRODUCER_KEYWORDS: list[str] = [ + "microsoft", + "libreoffice", + "openoffice", + "indesign", + "photoshop", + "quarkxpress", + "latex", + "pdftex", + "pdflatex", + "xetex", + "lualatex", + "word", + "excel", + "powerpoint", + "pages", + "keynote", + "numbers", + "scribus", + "affinity", + "canva", + "fpdf", + "reportlab", + "itext", + "fpdf2", + "wkhtmltopdf", + "google docs", + "chromium", + "chrome", + "webkit", + "prawn", + "cairo", + "pango", + "ghostscript", + "inkscape", +] + + +class TextSource(str, Enum): + """Indicates the origin of text embedded in a PDF.""" + + DIGITAL = "digital" # Created by a digital authoring tool (Word, LibreOffice, LaTeX…) + OCR_PREVIOUS = "ocr" # Previously run through an OCR engine + UNKNOWN = "unknown" # Source cannot be determined + + +# --------------------------------------------------------------------------- +# Result data class +# --------------------------------------------------------------------------- + + +@dataclass +class TextQualityResult: + """Result of an embedded-text quality assessment.""" + + is_good_quality: bool + quality_score: int # 0-100; 0 = completely garbled, 100 = perfect + text_source: TextSource + feedback: str + issues: list[str] = field(default_factory=list) + ai_response_raw: Optional[str] = None + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def detect_pdf_text_source(pdf_path: str) -> TextSource: + """Detect whether a PDF's text layer was created digitally or via OCR. + + Inspects the ``/Producer`` and ``/Creator`` metadata fields for known OCR + or digital-authoring-tool names. + + Args: + pdf_path: Absolute path to the PDF file. + + Returns: + :class:`TextSource` indicating the likely origin of the embedded text. + """ + try: + import pypdf + + with open(pdf_path, "rb") as f: + reader = pypdf.PdfReader(f) + info = reader.metadata or {} + + producer = str(info.get("/Producer", "") or "").lower() + creator = str(info.get("/Creator", "") or "").lower() + combined = f"{producer} {creator}" + + logger.debug(f"[text_quality] PDF metadata – Producer: {producer!r}, Creator: {creator!r}") + + for keyword in _OCR_PRODUCER_KEYWORDS: + if keyword in combined: + logger.info( + f"[text_quality] Detected OCR-origin PDF " + f"(keyword={keyword!r}, producer={producer!r}, creator={creator!r})" + ) + return TextSource.OCR_PREVIOUS + + for keyword in _DIGITAL_PRODUCER_KEYWORDS: + if keyword in combined: + logger.info( + f"[text_quality] Detected digitally-created PDF " + f"(keyword={keyword!r}, producer={producer!r}, creator={creator!r})" + ) + return TextSource.DIGITAL + + logger.info( + f"[text_quality] Could not determine PDF text source " + f"(producer={producer!r}, creator={creator!r}); treating as UNKNOWN" + ) + return TextSource.UNKNOWN + + except Exception as exc: + logger.warning(f"[text_quality] Failed to read PDF metadata from {pdf_path}: {exc}") + return TextSource.UNKNOWN + + +def check_text_quality(text: str, text_source: TextSource) -> TextQualityResult: + """Assess the quality of embedded PDF text using an AI model. + + Digitally-originated text is assumed to be correct and is **not** forwarded + to the AI. Text from a previous OCR pass, or of unknown origin, is + assessed for: + + - Excessive typos and OCR character-substitution artefacts. + - Lack of semantic coherence. + - Garbage characters or symbol soup. + + The text sample and the full AI feedback are logged at DEBUG / INFO level + to aid debugging and continuous quality improvement. + + Args: + text: The extracted text content to evaluate. + text_source: Where the text came from (digital, OCR, or unknown). + + Returns: + :class:`TextQualityResult` with the quality assessment. + """ + # 1. Digital PDFs are assumed good – skip the AI call entirely. + if text_source == TextSource.DIGITAL: + logger.info( + "[text_quality] Skipping quality check for digitally-created PDF " + "(source detected as digital; text quality assumed correct)." + ) + return TextQualityResult( + is_good_quality=True, + quality_score=100, + text_source=text_source, + feedback="Digitally-created PDF – text quality assumed correct; no AI check performed.", + ) + + # 2. Trivial case: empty or whitespace-only text. + stripped = text.strip() + if not stripped: + logger.info("[text_quality] Text is empty; marking as poor quality.") + return TextQualityResult( + is_good_quality=False, + quality_score=0, + text_source=text_source, + feedback="No text content to evaluate.", + issues=["empty_text"], + ) + + sample = stripped[:_TEXT_SAMPLE_MAX_CHARS] + logger.info( + f"[text_quality] Assessing text quality " + f"(source={text_source.value}, sample_chars={len(sample)}, total_chars={len(stripped)})" + ) + logger.debug(f"[text_quality] Text sample forwarded to AI:\n{sample}") + + prompt = ( + "You are a document quality assessor. Your task is to evaluate whether the " + "text extracted from a PDF is high-quality and semantically meaningful, or " + "whether it looks like garbled OCR output with typos, garbage characters, or " + "nonsensical fragments.\n\n" + "Evaluate the following text and return a JSON object with exactly these fields:\n" + ' "quality_score": integer 0-100 (0=completely garbled, 100=perfect text)\n' + ' "is_good_quality": boolean (true if quality_score >= 65)\n' + ' "feedback": one-sentence summary of your assessment\n' + ' "issues": list of issues found (e.g. ["excessive_typos", "garbage_characters", ' + '"incoherent_text", "fragmented_sentences"]); empty list if none\n\n' + "Criteria for POOR quality (score < 65):\n" + "- Excessive typos, misspellings, or letter substitutions typical of OCR errors\n" + "- Garbage characters (%, @, #, symbols mixed randomly into words)\n" + "- Incoherent or nonsensical sentences that carry no meaning\n" + "- Sequences of random characters or numbers without context\n" + "- Heavy fragmentation (isolated letters or words without sentence structure)\n\n" + "Criteria for GOOD quality (score >= 65):\n" + "- Mostly readable text with at most minor imperfections\n" + "- Coherent sentences and/or paragraphs\n" + "- Recognisable language (any language accepted)\n\n" + f"Text to evaluate:\n---\n{sample}\n---\n\n" + "Return only the JSON object, no markdown fences." + ) + + response_text: Optional[str] = None + try: + provider = get_ai_provider() + model = settings.ai_model or settings.openai_model + response_text = provider.chat_completion( + messages=[ + { + "role": "system", + "content": "You are a document quality assessor. Respond only with valid JSON.", + }, + {"role": "user", "content": prompt}, + ], + model=model, + temperature=0, + ) + + logger.info(f"[text_quality] AI quality check raw response: {response_text[:500]}") + + # Strip optional markdown code fences before parsing. + clean = re.sub(r"```(?:json)?\s*", "", response_text).strip().rstrip("`").strip() + parsed: dict = json.loads(clean) + + quality_score = int(parsed.get("quality_score", 0)) + is_good = bool(parsed.get("is_good_quality", quality_score >= 65)) + feedback = str(parsed.get("feedback", "")) + issues = list(parsed.get("issues", [])) + + logger.info( + f"[text_quality] Quality assessment complete – " + f"score={quality_score}, good={is_good}, issues={issues}, feedback={feedback!r}" + ) + + return TextQualityResult( + is_good_quality=is_good, + quality_score=quality_score, + text_source=text_source, + feedback=feedback, + issues=issues, + ai_response_raw=response_text, + ) + + except json.JSONDecodeError as exc: + logger.warning( + f"[text_quality] Could not parse AI quality response as JSON: {exc}. " + "Treating text as acceptable quality to avoid false negatives." + ) + return TextQualityResult( + is_good_quality=True, + quality_score=50, + text_source=text_source, + feedback=f"AI response could not be parsed as JSON ({exc}); assuming acceptable quality.", + ai_response_raw=response_text, + ) + + except Exception as exc: + logger.error( + f"[text_quality] AI quality check failed: {exc}. " + "Treating text as acceptable quality to avoid false negatives." + ) + return TextQualityResult( + is_good_quality=True, + quality_score=50, + text_source=text_source, + feedback=f"Quality check could not be performed ({exc}); assuming acceptable quality.", + ai_response_raw=response_text, + ) diff --git a/docs/ConfigurationGuide.md b/docs/ConfigurationGuide.md index 1a63a1f6..6c7124a9 100644 --- a/docs/ConfigurationGuide.md +++ b/docs/ConfigurationGuide.md @@ -461,6 +461,28 @@ DocuElevate supports multiple OCR engines that can be used individually or in co When multiple providers are listed, all run in parallel and their results are merged according to `OCR_MERGE_STRATEGY`. +#### Embedded Text Quality Check + +DocuElevate can automatically assess whether the text already embedded in a PDF is of sufficient quality before deciding to skip OCR. This prevents poor OCR output from a previous scan being silently used for downstream processing. + +| **Variable** | **Description** | **Default** | +|--------------------------------|---------------------------------------------------------------------------------|-------------| +| `ENABLE_TEXT_QUALITY_CHECK` | Enable AI-based quality assessment of embedded PDF text. | `true` | + +**How it works:** + +1. When a PDF with embedded text is received, DocuElevate first examines the PDF metadata (`/Producer`, `/Creator`). +2. If the PDF was **digitally created** (e.g., exported from Word, LibreOffice, LaTeX, or any modern authoring tool), the embedded text is considered trustworthy and the quality check is skipped — digital text cannot be improved by re-OCRing. +3. If the PDF was **previously OCR'd** (Tesseract, ABBYY, ocrmypdf, etc.) or the origin is **unknown**, an AI model evaluates a sample of the extracted text for: + - Excessive typos and character-substitution artefacts typical of OCR + - Garbage characters or symbol soup + - Incoherent or nonsensical sentences + - Heavy fragmentation +4. If the quality score falls **below 65/100**, the embedded text is discarded and the file is sent to the configured OCR providers for a fresh scan. +5. All quality decisions (score, source, AI feedback) are recorded in the processing log for review. + +> **Tip**: Set `ENABLE_TEXT_QUALITY_CHECK=false` to disable the check entirely and always use embedded text as-is. This is useful when the AI provider is unavailable or when processing speed is more important than text accuracy. + #### Searchable PDF Text Layer Not all OCR providers embed a searchable text layer in the output PDF. The table below summarises each provider's behaviour and how DocuElevate handles it: diff --git a/tests/test_text_quality.py b/tests/test_text_quality.py new file mode 100644 index 00000000..aa151c1d --- /dev/null +++ b/tests/test_text_quality.py @@ -0,0 +1,703 @@ +""" +Comprehensive tests for app/utils/text_quality.py. + +Covers: +- detect_pdf_text_source: digital, OCR, and unknown PDF metadata +- check_text_quality: digital bypass, good text, poor text, empty text +- AI failure and JSON-parse error handling +- Integration with process_document: quality check disabled, good quality, + poor quality (triggers re-OCR), digital source bypass +""" + +from unittest.mock import MagicMock, patch + +import pytest + +from app.utils.text_quality import ( + TextQualityResult, + TextSource, + check_text_quality, + detect_pdf_text_source, +) + +# --------------------------------------------------------------------------- +# Helpers / sample texts +# --------------------------------------------------------------------------- + +# Well-formed English invoice text – should pass quality checks. +GOOD_TEXT = """ +INVOICE #2024-0042 +Date: 15 January 2024 + +Bill To: + Acme Corporation + 123 Main Street + Springfield, IL 62701 + +Description Qty Unit Price Total +Widget A 10 $12.50 $125.00 +Widget B 5 $22.00 $110.00 + Subtotal: $235.00 + Tax: $17.63 + Total: $252.63 + +Payment due within 30 days. Thank you for your business. +""" + +# Garbled OCR-artefact text with heavy character substitution – poor quality. +POOR_OCR_TEXT = """ +lnv0|c3 #2@24-@@42 +D@t3: l5 J@nu@ry 2@24 + +Bi|l T0: + Acm3 C0rp0r@ti0n + l23 M@in Str33t + Springf|3|d, lL 62701 + +D3scripti0n Qty Unit Pric3 T0t@l +Widg3t A l0 $l2.5@ $l25.@@ +Widg3t B 5 $22.@@ $ll@.@@ + Subr0t@l: $235.@@ + T@x: $l7.63 + T0t@l: $252.63 + +P@ym3nt du3 with|n 30 d@ys. Th@nk y0u f0r y0ur busin3ss. +""" + +# Complete garbage – random symbol soup. +GARBAGE_TEXT = "ÿÿÿÿÿÿÿ @@@ %%% !!! *** ### ^^^ &&&" * 20 + +# Text so fragmented it carries no meaning. +FRAGMENTED_TEXT = "a b c d e f g h i j k l m n o p q r s t u v w x y z " * 10 + +# --------------------------------------------------------------------------- +# Minimal valid PDF bytes used when we need to patch pypdf.PdfReader +# --------------------------------------------------------------------------- +_MINIMAL_PDF = ( + b"%PDF-1.4\n" + b"1 0 obj\n<>\nendobj\n" + b"2 0 obj\n<>\nendobj\n" + b"3 0 obj\n<>\nendobj\n" + b"xref\n0 4\n" + b"0000000000 65535 f \n" + b"0000000009 00000 n \n" + b"0000000058 00000 n \n" + b"0000000115 00000 n \n" + b"trailer\n<>\n" + b"startxref\n190\n%%EOF\n" +) + + +def _pdf_with_metadata(tmp_path, producer: str = "", creator: str = "") -> str: + """Write a minimal PDF file and return its path (metadata is mocked later).""" + p = tmp_path / "test.pdf" + p.write_bytes(_MINIMAL_PDF) + return str(p) + + +# --------------------------------------------------------------------------- +# detect_pdf_text_source +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestDetectPdfTextSource: + """Tests for detect_pdf_text_source().""" + + def _mock_metadata(self, producer: str, creator: str = "") -> MagicMock: + """Build a mock PdfReader whose .metadata dict contains /Producer and /Creator.""" + meta = {} + if producer: + meta["/Producer"] = producer + if creator: + meta["/Creator"] = creator + reader = MagicMock() + reader.metadata = meta + return reader + + @pytest.mark.parametrize( + "producer,creator,expected", + [ + # OCR producers + ("Tesseract OCR 5.3.0", "", TextSource.OCR_PREVIOUS), + ("ocrmypdf 14.0", "", TextSource.OCR_PREVIOUS), + ("ABBYY FineReader 15", "", TextSource.OCR_PREVIOUS), + ("Nuance PDF Converter", "", TextSource.OCR_PREVIOUS), + ("ReadIris 17", "", TextSource.OCR_PREVIOUS), + ("OmniPage 19", "", TextSource.OCR_PREVIOUS), + # Digital producers + ("Microsoft Word 365", "", TextSource.DIGITAL), + ("LibreOffice 7.5", "", TextSource.DIGITAL), + ("pdflatex", "", TextSource.DIGITAL), + ("xetex", "", TextSource.DIGITAL), + ("ReportLab PDF Library", "", TextSource.DIGITAL), + ("wkhtmltopdf 0.12.6", "", TextSource.DIGITAL), + ("Chromium 120", "", TextSource.DIGITAL), + ("Google Docs", "", TextSource.DIGITAL), + # Creator field + ("", "Microsoft Excel 2021", TextSource.DIGITAL), + ("", "Tesseract-OCR", TextSource.OCR_PREVIOUS), + # Unknown + ("Adobe Acrobat", "", TextSource.UNKNOWN), + ("", "", TextSource.UNKNOWN), + ], + ) + def test_source_detection(self, tmp_path, producer, creator, expected): + """Producer/Creator metadata maps to the correct TextSource.""" + pdf_path = _pdf_with_metadata(tmp_path) + reader_mock = self._mock_metadata(producer, creator) + + with patch("pypdf.PdfReader", return_value=reader_mock): + result = detect_pdf_text_source(pdf_path) + + assert result == expected + + def test_read_error_returns_unknown(self, tmp_path): + """If pypdf raises an exception, return UNKNOWN (safe fallback).""" + pdf_path = _pdf_with_metadata(tmp_path) + + with patch("pypdf.PdfReader", side_effect=Exception("corrupt PDF")): + result = detect_pdf_text_source(pdf_path) + + assert result == TextSource.UNKNOWN + + def test_none_metadata_returns_unknown(self, tmp_path): + """If reader.metadata is None, return UNKNOWN.""" + pdf_path = _pdf_with_metadata(tmp_path) + reader_mock = MagicMock() + reader_mock.metadata = None + + with patch("pypdf.PdfReader", return_value=reader_mock): + result = detect_pdf_text_source(pdf_path) + + assert result == TextSource.UNKNOWN + + +# --------------------------------------------------------------------------- +# check_text_quality – digital bypass +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestCheckTextQualityDigitalBypass: + """Digital-origin PDFs must skip the AI call and return 100/good.""" + + def test_digital_source_skips_ai(self): + """No AI provider call is made for DIGITAL source.""" + with patch("app.utils.text_quality.get_ai_provider") as mock_provider: + result = check_text_quality(GOOD_TEXT, TextSource.DIGITAL) + + mock_provider.assert_not_called() + assert result.is_good_quality is True + assert result.quality_score == 100 + assert result.text_source == TextSource.DIGITAL + + def test_digital_source_poor_looking_text_still_trusted(self): + """Even if the text looks poor, digital origin is always trusted.""" + with patch("app.utils.text_quality.get_ai_provider") as mock_provider: + result = check_text_quality(POOR_OCR_TEXT, TextSource.DIGITAL) + + mock_provider.assert_not_called() + assert result.is_good_quality is True + + +# --------------------------------------------------------------------------- +# check_text_quality – empty text +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestCheckTextQualityEmpty: + """Empty / whitespace text must fail immediately without an AI call.""" + + @pytest.mark.parametrize("empty_text", ["", " ", "\n\t\n"]) + def test_empty_text_fails_without_ai(self, empty_text): + with patch("app.utils.text_quality.get_ai_provider") as mock_provider: + result = check_text_quality(empty_text, TextSource.OCR_PREVIOUS) + + mock_provider.assert_not_called() + assert result.is_good_quality is False + assert result.quality_score == 0 + assert "empty_text" in result.issues + + def test_empty_text_unknown_source(self): + with patch("app.utils.text_quality.get_ai_provider") as mock_provider: + result = check_text_quality("", TextSource.UNKNOWN) + + mock_provider.assert_not_called() + assert result.is_good_quality is False + + +# --------------------------------------------------------------------------- +# check_text_quality – AI-backed assessments +# --------------------------------------------------------------------------- + + +def _make_ai_response(quality_score: int, is_good: bool, feedback: str, issues: list) -> str: + """Build a JSON string mimicking the AI response format.""" + import json as _json + + return _json.dumps( + { + "quality_score": quality_score, + "is_good_quality": is_good, + "feedback": feedback, + "issues": issues, + } + ) + + +@pytest.mark.unit +class TestCheckTextQualityAI: + """Tests for the AI-backed quality assessment.""" + + def _mock_provider(self, response: str) -> MagicMock: + """Return a mock AI provider whose chat_completion returns *response*.""" + provider = MagicMock() + provider.chat_completion.return_value = response + return provider + + def test_good_text_passes(self): + """A high-quality AI response marks text as good.""" + ai_resp = _make_ai_response(90, True, "Well-structured invoice text.", []) + provider = self._mock_provider(ai_resp) + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = "gpt-4o-mini" + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(GOOD_TEXT, TextSource.OCR_PREVIOUS) + + assert result.is_good_quality is True + assert result.quality_score == 90 + assert result.issues == [] + + def test_poor_ocr_text_fails(self): + """A low-quality AI response marks text as poor.""" + ai_resp = _make_ai_response( + 25, False, "Severe OCR artefacts with character substitutions.", ["excessive_typos", "garbage_characters"] + ) + provider = self._mock_provider(ai_resp) + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = "gpt-4o-mini" + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(POOR_OCR_TEXT, TextSource.OCR_PREVIOUS) + + assert result.is_good_quality is False + assert result.quality_score == 25 + assert "excessive_typos" in result.issues + assert "garbage_characters" in result.issues + + def test_garbage_text_fails(self): + """Complete garbage text is scored very low.""" + ai_resp = _make_ai_response(5, False, "Random symbol soup – no readable content.", ["garbage_characters"]) + provider = self._mock_provider(ai_resp) + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = "gpt-4o-mini" + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(GARBAGE_TEXT, TextSource.UNKNOWN) + + assert result.is_good_quality is False + assert result.quality_score <= 20 + + def test_fragmented_text_fails(self): + """Fragmented text is scored low.""" + ai_resp = _make_ai_response(30, False, "Highly fragmented, no coherent sentences.", ["fragmented_sentences"]) + provider = self._mock_provider(ai_resp) + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = None + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(FRAGMENTED_TEXT, TextSource.OCR_PREVIOUS) + + assert result.is_good_quality is False + + def test_borderline_score_uses_is_good_quality_field(self): + """The is_good_quality field from the AI takes precedence over the threshold.""" + # Score 64 but AI explicitly says True + ai_resp = _make_ai_response(64, True, "Mostly readable despite minor issues.", []) + provider = self._mock_provider(ai_resp) + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = None + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(GOOD_TEXT, TextSource.OCR_PREVIOUS) + + assert result.is_good_quality is True + assert result.quality_score == 64 + + def test_markdown_fences_stripped_before_parse(self): + """The parser handles AI responses wrapped in markdown code fences.""" + import json as _json + + inner = _json.dumps({"quality_score": 80, "is_good_quality": True, "feedback": "Fine.", "issues": []}) + fenced = f"```json\n{inner}\n```" + provider = self._mock_provider(fenced) + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = None + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(GOOD_TEXT, TextSource.UNKNOWN) + + assert result.is_good_quality is True + assert result.quality_score == 80 + + def test_raw_ai_response_stored_in_result(self): + """The raw AI response is preserved in TextQualityResult.ai_response_raw.""" + ai_resp = _make_ai_response(88, True, "Good text.", []) + provider = self._mock_provider(ai_resp) + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = None + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(GOOD_TEXT, TextSource.OCR_PREVIOUS) + + assert result.ai_response_raw == ai_resp + + def test_text_truncated_to_sample_max(self): + """Text longer than _TEXT_SAMPLE_MAX_CHARS is truncated before sending to AI.""" + from app.utils.text_quality import _TEXT_SAMPLE_MAX_CHARS + + long_text = "a" * (_TEXT_SAMPLE_MAX_CHARS + 5000) + ai_resp = _make_ai_response(85, True, "Fine.", []) + provider = self._mock_provider(ai_resp) + captured_prompts: list[str] = [] + + def _capture(messages, model, temperature=0, **kw): + captured_prompts.append(messages[-1]["content"]) + return ai_resp + + provider.chat_completion.side_effect = _capture + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = None + mock_settings.openai_model = "gpt-4o-mini" + check_text_quality(long_text, TextSource.UNKNOWN) + + assert len(captured_prompts) == 1 + # The prompt should NOT contain more than _TEXT_SAMPLE_MAX_CHARS "a"s + count_a = captured_prompts[0].count("a" * 100) + assert "a" * (_TEXT_SAMPLE_MAX_CHARS + 1) not in captured_prompts[0] + + +# --------------------------------------------------------------------------- +# check_text_quality – error / edge cases +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestCheckTextQualityErrorHandling: + """Tests for failure modes that must not crash the pipeline.""" + + def test_json_parse_error_returns_good_quality(self): + """Unparseable AI response defaults to good quality (avoids false negatives).""" + provider = MagicMock() + provider.chat_completion.return_value = "This is not JSON at all." + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = None + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(GOOD_TEXT, TextSource.UNKNOWN) + + assert result.is_good_quality is True + assert result.quality_score == 50 + + def test_ai_provider_exception_returns_good_quality(self): + """If the AI provider raises an exception, default to good quality.""" + provider = MagicMock() + provider.chat_completion.side_effect = RuntimeError("API timeout") + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = None + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(GOOD_TEXT, TextSource.OCR_PREVIOUS) + + assert result.is_good_quality is True + assert result.quality_score == 50 + + def test_ai_provider_exception_stores_none_raw_response(self): + """ai_response_raw should be None when the provider raises before returning.""" + provider = MagicMock() + provider.chat_completion.side_effect = ConnectionError("no internet") + + with ( + patch("app.utils.text_quality.get_ai_provider", return_value=provider), + patch("app.utils.text_quality.settings") as mock_settings, + ): + mock_settings.ai_model = None + mock_settings.openai_model = "gpt-4o-mini" + result = check_text_quality(GOOD_TEXT, TextSource.UNKNOWN) + + assert result.ai_response_raw is None + + +# --------------------------------------------------------------------------- +# TextQualityResult dataclass +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestTextQualityResult: + """Tests for the TextQualityResult dataclass.""" + + def test_default_issues_is_empty_list(self): + result = TextQualityResult( + is_good_quality=True, + quality_score=90, + text_source=TextSource.DIGITAL, + feedback="Good.", + ) + assert result.issues == [] + assert result.ai_response_raw is None + + def test_issues_field(self): + result = TextQualityResult( + is_good_quality=False, + quality_score=20, + text_source=TextSource.OCR_PREVIOUS, + feedback="Bad.", + issues=["excessive_typos"], + ) + assert result.issues == ["excessive_typos"] + + +# --------------------------------------------------------------------------- +# Integration: process_document task with text quality check +# --------------------------------------------------------------------------- + +# Build a minimal but real PDF with embedded text so pypdf.PdfReader works. +_EMBEDDED_TEXT_PDF = b"""%PDF-1.4 +1 0 obj +<< +/Type /Catalog +/Pages 2 0 R +>> +endobj +2 0 obj +<< +/Type /Pages +/Kids [3 0 R] +/Count 1 +>> +endobj +3 0 obj +<< +/Type /Page +/Parent 2 0 R +/MediaBox [0 0 612 792] +/Resources << +/Font << +/F1 << +/Type /Font +/Subtype /Type1 +/BaseFont /Helvetica +>> +>> +>> +/Contents 4 0 R +>> +endobj +4 0 obj +<< +/Length 44 +>> +stream +BT +/F1 12 Tf +100 700 Td +(Invoice total is $252.63) Tj +ET +endstream +endobj +xref +0 5 +0000000000 65535 f +0000000009 00000 n +0000000058 00000 n +0000000115 00000 n +0000000306 00000 n +trailer +<< +/Size 5 +/Root 1 0 R +>> +startxref +404 +%%EOF +""" + + +@pytest.mark.unit +@pytest.mark.requires_db +class TestProcessDocumentTextQuality: + """Integration tests verifying quality check in process_document.""" + + def _write_pdf(self, tmp_path, name: str = "doc.pdf") -> str: + p = tmp_path / name + p.write_bytes(_EMBEDDED_TEXT_PDF) + return str(p) + + def test_quality_check_disabled_skips_ai(self, db_session, tmp_path): + """When enable_text_quality_check=False, the AI is never called.""" + from app.tasks.process_document import process_document + + pdf_path = self._write_pdf(tmp_path) + + with ( + patch("app.tasks.process_document.SessionLocal") as mock_sl, + patch("app.tasks.process_document.settings") as mock_settings, + patch("app.tasks.process_document.log_task_progress"), + patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_gpt, + patch("app.tasks.process_document.detect_pdf_text_source") as mock_detect, + patch("app.tasks.process_document.check_text_quality") as mock_check, + ): + mock_sl.return_value.__enter__.return_value = db_session + mock_sl.return_value.__exit__.return_value = None + mock_settings.workdir = str(tmp_path) + mock_settings.enable_deduplication = False + mock_settings.show_deduplication_step = False + mock_settings.enable_text_quality_check = False + + result = process_document.run(pdf_path) + + mock_detect.assert_not_called() + mock_check.assert_not_called() + mock_gpt.delay.assert_called_once() + assert result["status"] == "Text extracted locally" + + def test_quality_check_good_text_proceeds_to_gpt(self, db_session, tmp_path): + """When quality check passes, metadata extraction is queued normally.""" + from app.tasks.process_document import process_document + + pdf_path = self._write_pdf(tmp_path) + + good_quality = TextQualityResult( + is_good_quality=True, + quality_score=90, + text_source=TextSource.OCR_PREVIOUS, + feedback="Good readable text.", + ) + + with ( + patch("app.tasks.process_document.SessionLocal") as mock_sl, + patch("app.tasks.process_document.settings") as mock_settings, + patch("app.tasks.process_document.log_task_progress"), + patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_gpt, + patch("app.tasks.process_document.process_with_ocr") as mock_ocr, + patch("app.tasks.process_document.detect_pdf_text_source", return_value=TextSource.OCR_PREVIOUS), + patch("app.tasks.process_document.check_text_quality", return_value=good_quality), + ): + mock_sl.return_value.__enter__.return_value = db_session + mock_sl.return_value.__exit__.return_value = None + mock_settings.workdir = str(tmp_path) + mock_settings.enable_deduplication = False + mock_settings.show_deduplication_step = False + mock_settings.enable_text_quality_check = True + + result = process_document.run(pdf_path) + + mock_gpt.delay.assert_called_once() + mock_ocr.delay.assert_not_called() + assert result["status"] == "Text extracted locally" + + def test_quality_check_poor_text_triggers_ocr(self, db_session, tmp_path): + """When quality check fails, OCR is queued instead of GPT extraction.""" + from app.tasks.process_document import process_document + + pdf_path = self._write_pdf(tmp_path) + + poor_quality = TextQualityResult( + is_good_quality=False, + quality_score=20, + text_source=TextSource.OCR_PREVIOUS, + feedback="Severe OCR artefacts.", + issues=["excessive_typos", "garbage_characters"], + ) + + with ( + patch("app.tasks.process_document.SessionLocal") as mock_sl, + patch("app.tasks.process_document.settings") as mock_settings, + patch("app.tasks.process_document.log_task_progress"), + patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_gpt, + patch("app.tasks.process_document.process_with_ocr") as mock_ocr, + patch("app.tasks.process_document.detect_pdf_text_source", return_value=TextSource.OCR_PREVIOUS), + patch("app.tasks.process_document.check_text_quality", return_value=poor_quality), + ): + mock_sl.return_value.__enter__.return_value = db_session + mock_sl.return_value.__exit__.return_value = None + mock_settings.workdir = str(tmp_path) + mock_settings.enable_deduplication = False + mock_settings.show_deduplication_step = False + mock_settings.enable_text_quality_check = True + + result = process_document.run(pdf_path) + + mock_ocr.delay.assert_called_once() + mock_gpt.delay.assert_not_called() + assert "OCR" in result["status"] + + def test_quality_check_digital_source_skips_ai_call(self, db_session, tmp_path): + """Digital-origin PDFs bypass the AI and proceed directly to GPT.""" + from app.tasks.process_document import process_document + + pdf_path = self._write_pdf(tmp_path) + + digital_result = TextQualityResult( + is_good_quality=True, + quality_score=100, + text_source=TextSource.DIGITAL, + feedback="Digitally-created PDF – text quality assumed correct; no AI check performed.", + ) + + with ( + patch("app.tasks.process_document.SessionLocal") as mock_sl, + patch("app.tasks.process_document.settings") as mock_settings, + patch("app.tasks.process_document.log_task_progress"), + patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_gpt, + patch("app.tasks.process_document.process_with_ocr") as mock_ocr, + patch("app.tasks.process_document.detect_pdf_text_source", return_value=TextSource.DIGITAL), + patch("app.tasks.process_document.check_text_quality", return_value=digital_result), + ): + mock_sl.return_value.__enter__.return_value = db_session + mock_sl.return_value.__exit__.return_value = None + mock_settings.workdir = str(tmp_path) + mock_settings.enable_deduplication = False + mock_settings.show_deduplication_step = False + mock_settings.enable_text_quality_check = True + + result = process_document.run(pdf_path) + + mock_gpt.delay.assert_called_once() + mock_ocr.delay.assert_not_called() + assert result["status"] == "Text extracted locally" From 0c500e1ec659e88b7b638380e338c4f2f7985308 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 24 Feb 2026 18:44:20 +0000 Subject: [PATCH 3/3] fix(ocr): address code review: add model fallback default, remove unused variable Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/utils/text_quality.py | 2 +- tests/test_text_quality.py | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/app/utils/text_quality.py b/app/utils/text_quality.py index 3901633b..e2200092 100644 --- a/app/utils/text_quality.py +++ b/app/utils/text_quality.py @@ -261,7 +261,7 @@ def check_text_quality(text: str, text_source: TextSource) -> TextQualityResult: response_text: Optional[str] = None try: provider = get_ai_provider() - model = settings.ai_model or settings.openai_model + model = settings.ai_model or settings.openai_model or "gpt-4o-mini" response_text = provider.chat_completion( messages=[ { diff --git a/tests/test_text_quality.py b/tests/test_text_quality.py index aa151c1d..fc2338c8 100644 --- a/tests/test_text_quality.py +++ b/tests/test_text_quality.py @@ -401,7 +401,6 @@ class TestCheckTextQualityAI: assert len(captured_prompts) == 1 # The prompt should NOT contain more than _TEXT_SAMPLE_MAX_CHARS "a"s - count_a = captured_prompts[0].count("a" * 100) assert "a" * (_TEXT_SAMPLE_MAX_CHARS + 1) not in captured_prompts[0]