gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ocr): add AI-based embedded text quality check with automatic OCR fallback

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-24 18:41:27 +00:00
parent ebdb332117
commit b03bfb5e02
6 changed files with 1141 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/config.py
+13
View File
@@ -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,
app/tasks/process_document.py
+64
View File
@@ -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,
app/utils/settings_service.py
+13
View File
@@ -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",
app/utils/text_quality.py
+326
View File
@@ -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,
)
docs/ConfigurationGuide.md
+22
View File
@@ -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:
tests/test_text_quality.py
+703
View File
@@ -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<</Type /Catalog /Pages 2 0 R>>\nendobj\n"
b"2 0 obj\n<</Type /Pages /Kids [3 0 R] /Count 1>>\nendobj\n"
b"3 0 obj\n<</Type /Page /Parent 2 0 R /MediaBox [0 0 612 792]>>\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<</Size 4 /Root 1 0 R>>\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"