mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
fix(parse): warn when PDF parse yields no text layer (scanned PDFs) (#1021)
* fix(parse): import email.message and repair pdfplumber test mock - email_parser.py uses email.message.Message at class-definition time but only did 'import email', so 'import semantica.parse' fails in a fresh Python process unless something else imported email.message first - test_pdf_parser patched semantica.parse.pdf_parser.pdfplumber, which never exists as a module attribute (pdfplumber is imported inside PDFParser.parse); inject a fake module via sys.modules instead * fix(parse): warn when PDF parse yields no text layer (scanned PDFs) Scanned (image-only) PDFs parsed via the default pdfplumber route returned an empty full_text with progress status 'completed' - no error, no warning - so the failure only surfaced far downstream. Warn in PDFParser.parse() when every parsed page yields no text (and extract_text is enabled), pointing users to method='docling' with enable_ocr=True. * fix(parse): improve scanned PDF detection --------- Co-authored-by: shanyu910 <208111055+shanyu910@users.noreply.github.com> Co-authored-by: Sameer Kadam <sskadam6305@gmail.com>
This commit is contained in:
co-authored by
shanyu910
Sameer Kadam
parent
65e6dcfef5
commit
7f928f9f8e
@@ -33,6 +33,7 @@ License: MIT
|
||||
"""
|
||||
|
||||
import email
|
||||
import email.message
|
||||
from dataclasses import dataclass, field
|
||||
from email import message_from_bytes, message_from_string
|
||||
from email.header import decode_header
|
||||
|
||||
@@ -143,15 +143,39 @@ class PDFParser:
|
||||
)
|
||||
pages.append(page_data)
|
||||
|
||||
full_text = "\n\n".join(page.text for page in pages)
|
||||
|
||||
# Scanned/image-only PDFs have no text layer; warn so the
|
||||
# failure surfaces at parse time instead of downstream.
|
||||
# Check any() over page texts to avoid a temporary stripped
|
||||
# copy of the full concatenation for large documents.
|
||||
no_text = (
|
||||
options.get("extract_text", True)
|
||||
and pages
|
||||
and not any(page.text.strip() for page in pages)
|
||||
)
|
||||
if no_text:
|
||||
self.logger.warning(
|
||||
f"PDF {file_path.name}: parsed {len(pages)} page(s) but "
|
||||
f"extracted no text. This is likely a scanned "
|
||||
f"(image-only) PDF. Retry with "
|
||||
f"parse_pdf(..., method='docling', enable_ocr=True) "
|
||||
f"for OCR-based extraction."
|
||||
)
|
||||
|
||||
self.progress_tracker.stop_tracking(
|
||||
tracking_id,
|
||||
status="completed",
|
||||
message=f"Parsed {len(pages)} pages",
|
||||
message=(
|
||||
f"Parsed {len(pages)} page(s) (no text layer detected)"
|
||||
if no_text
|
||||
else f"Parsed {len(pages)} page(s)"
|
||||
),
|
||||
)
|
||||
return {
|
||||
"metadata": metadata.__dict__,
|
||||
"pages": [page.__dict__ for page in pages],
|
||||
"full_text": "\n\n".join(page.text for page in pages),
|
||||
"full_text": full_text,
|
||||
"total_pages": len(pdf.pages),
|
||||
}
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import tempfile
|
||||
import os
|
||||
import json
|
||||
import csv
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@@ -137,29 +138,32 @@ class TestParseComprehensive(unittest.TestCase):
|
||||
|
||||
# --- Document Parser Tests ---
|
||||
|
||||
@patch('semantica.parse.pdf_parser.pdfplumber')
|
||||
def test_pdf_parser(self, mock_pdfplumber):
|
||||
def test_pdf_parser(self):
|
||||
parser = PDFParser()
|
||||
# pdfplumber is imported inside PDFParser.parse, so inject a fake
|
||||
# module into sys.modules instead of patching a module attribute
|
||||
mock_pdfplumber = MagicMock()
|
||||
mock_pdf = MagicMock()
|
||||
mock_page = MagicMock()
|
||||
mock_page.extract_text.return_value = "Page text"
|
||||
mock_pdf.pages = [mock_page]
|
||||
# Ensure metadata is a dict, not a property object if that's an issue
|
||||
mock_pdf.metadata = {"Title": "Test PDF"}
|
||||
|
||||
|
||||
# Setup the context manager
|
||||
mock_context_manager = MagicMock()
|
||||
mock_context_manager.__enter__.return_value = mock_pdf
|
||||
mock_context_manager.__exit__.return_value = None
|
||||
mock_pdfplumber.open.return_value = mock_context_manager
|
||||
|
||||
|
||||
# We don't need a real file if we mock open, but the parser likely checks file existence
|
||||
with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.pdf') as tmp:
|
||||
tmp.write(b"dummy pdf content")
|
||||
tmp_path = tmp.name
|
||||
|
||||
|
||||
try:
|
||||
result = parser.parse(tmp_path)
|
||||
with patch.dict(sys.modules, {"pdfplumber": mock_pdfplumber}):
|
||||
result = parser.parse(tmp_path)
|
||||
# Returns dict with full_text
|
||||
self.assertIn("Page text", result["full_text"])
|
||||
self.assertEqual(result["metadata"].get("title"), "Test PDF")
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
"""
|
||||
Tests for scanned-PDF (empty text layer) detection in PDFParser.
|
||||
|
||||
Covers:
|
||||
- Warning when all parsed pages lack a text layer (scanned PDFs)
|
||||
- No warning when text is present or text extraction is disabled
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from semantica.parse.pdf_parser import PDFParser
|
||||
|
||||
|
||||
class TestScannedPdfWarning(unittest.TestCase):
|
||||
"""PDFParser warns when a PDF has pages but no extractable text."""
|
||||
|
||||
def setUp(self):
|
||||
self.mock_logger = MagicMock()
|
||||
logger_patcher = patch(
|
||||
"semantica.parse.pdf_parser.get_logger", return_value=self.mock_logger
|
||||
)
|
||||
tracker_patcher = patch("semantica.parse.pdf_parser.get_progress_tracker")
|
||||
logger_patcher.start()
|
||||
tracker_patcher.start()
|
||||
self.addCleanup(logger_patcher.stop)
|
||||
self.addCleanup(tracker_patcher.stop)
|
||||
|
||||
def _make_pdf_path(self):
|
||||
with tempfile.NamedTemporaryFile(mode="wb", delete=False, suffix=".pdf") as tmp:
|
||||
tmp.write(b"dummy pdf content")
|
||||
return tmp.name
|
||||
|
||||
def _mock_pdfplumber(self, page_text):
|
||||
"""Build a fake pdfplumber module whose single page yields page_text."""
|
||||
mock_page = MagicMock()
|
||||
mock_page.extract_text.return_value = page_text
|
||||
mock_page.extract_tables.return_value = []
|
||||
mock_page.width = 612
|
||||
mock_page.height = 792
|
||||
|
||||
mock_pdf = MagicMock()
|
||||
mock_pdf.pages = [mock_page]
|
||||
mock_pdf.metadata = {}
|
||||
|
||||
mock_pdfplumber = MagicMock()
|
||||
context_manager = MagicMock()
|
||||
context_manager.__enter__.return_value = mock_pdf
|
||||
mock_pdfplumber.open.return_value = context_manager
|
||||
return mock_pdfplumber
|
||||
|
||||
def test_warns_when_no_text_layer(self):
|
||||
mock_pdfplumber = self._mock_pdfplumber(page_text=None) # scanned page
|
||||
parser = PDFParser()
|
||||
path = self._make_pdf_path()
|
||||
try:
|
||||
with patch.dict(sys.modules, {"pdfplumber": mock_pdfplumber}):
|
||||
result = parser.parse(path)
|
||||
self.assertEqual(result["full_text"], "")
|
||||
self.mock_logger.warning.assert_called_once()
|
||||
message = self.mock_logger.warning.call_args[0][0]
|
||||
self.assertIn("scanned", message)
|
||||
self.assertIn("enable_ocr", message)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_no_warning_when_text_present(self):
|
||||
mock_pdfplumber = self._mock_pdfplumber(page_text="Real digital text")
|
||||
parser = PDFParser()
|
||||
path = self._make_pdf_path()
|
||||
try:
|
||||
with patch.dict(sys.modules, {"pdfplumber": mock_pdfplumber}):
|
||||
result = parser.parse(path)
|
||||
self.assertIn("Real digital text", result["full_text"])
|
||||
self.mock_logger.warning.assert_not_called()
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_no_warning_when_extract_text_disabled(self):
|
||||
mock_pdfplumber = self._mock_pdfplumber(page_text=None)
|
||||
parser = PDFParser()
|
||||
path = self._make_pdf_path()
|
||||
try:
|
||||
with patch.dict(sys.modules, {"pdfplumber": mock_pdfplumber}):
|
||||
result = parser.parse(path, extract_text=False, extract_tables=False)
|
||||
self.assertEqual(result["full_text"], "")
|
||||
self.mock_logger.warning.assert_not_called()
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_warns_when_pages_have_only_whitespace(self):
|
||||
"""Pages that return only whitespace count as no extractable text."""
|
||||
mock_pdfplumber = self._mock_pdfplumber(page_text=" \n\t ")
|
||||
parser = PDFParser()
|
||||
path = self._make_pdf_path()
|
||||
try:
|
||||
with patch.dict(sys.modules, {"pdfplumber": mock_pdfplumber}):
|
||||
parser.parse(path)
|
||||
self.mock_logger.warning.assert_called_once()
|
||||
message = self.mock_logger.warning.call_args[0][0]
|
||||
self.assertIn("scanned", message)
|
||||
self.assertIn("enable_ocr", message)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_no_warning_for_mixed_pdf(self):
|
||||
"""A PDF where at least one page has real text must not warn."""
|
||||
# Build a two-page PDF: page 1 has text, page 2 is image-only
|
||||
mock_page_text = MagicMock()
|
||||
mock_page_text.extract_text.return_value = "Actual content"
|
||||
mock_page_text.extract_tables.return_value = []
|
||||
mock_page_text.width = 612
|
||||
mock_page_text.height = 792
|
||||
|
||||
mock_page_image = MagicMock()
|
||||
mock_page_image.extract_text.return_value = None
|
||||
mock_page_image.extract_tables.return_value = []
|
||||
mock_page_image.width = 612
|
||||
mock_page_image.height = 792
|
||||
|
||||
mock_pdf = MagicMock()
|
||||
mock_pdf.pages = [mock_page_text, mock_page_image]
|
||||
mock_pdf.metadata = {}
|
||||
|
||||
mock_pdfplumber = MagicMock()
|
||||
context_manager = MagicMock()
|
||||
context_manager.__enter__.return_value = mock_pdf
|
||||
mock_pdfplumber.open.return_value = context_manager
|
||||
|
||||
parser = PDFParser()
|
||||
path = self._make_pdf_path()
|
||||
try:
|
||||
with patch.dict(sys.modules, {"pdfplumber": mock_pdfplumber}):
|
||||
result = parser.parse(path)
|
||||
self.assertIn("Actual content", result["full_text"])
|
||||
self.mock_logger.warning.assert_not_called()
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user