|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
from pypdf import PdfReader
|
|
from pypdf.errors import PdfReadError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAX_PDF_BYTES = 15_000_000
|
|
MAX_PDF_PAGES = 50
|
|
PDF_MAGIC = b"%PDF-"
|
|
WHITESPACE = re.compile(r"[ \t]+")
|
|
BLANK_LINES = re.compile(r"\n{3,}")
|
|
|
|
|
|
def is_pdf(content_type: str, url: str, head: bytes) -> bool:
|
|
if "application/pdf" in content_type or "application/x-pdf" in content_type:
|
|
return True
|
|
if urlsplit(url).path.lower().endswith(".pdf"):
|
|
return True
|
|
return head.startswith(PDF_MAGIC)
|
|
|
|
|
|
def _normalize(text: str) -> str:
|
|
text = text.replace("\r\n", "\n").replace("\r", "\n")
|
|
text = WHITESPACE.sub(" ", text)
|
|
text = BLANK_LINES.sub("\n\n", text)
|
|
return text.strip()
|
|
|
|
|
|
def extract_pdf_text(raw: bytes) -> tuple[str, str]:
|
|
if not raw.startswith(PDF_MAGIC):
|
|
logger.debug("deepsearch pdf payload missing magic header (%d bytes)", len(raw))
|
|
return "", ""
|
|
handle = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False)
|
|
staged = Path(handle.name)
|
|
try:
|
|
handle.write(raw)
|
|
handle.close()
|
|
logger.info("deepsearch staged pdf at %s (%d bytes)", staged, len(raw))
|
|
reader = PdfReader(str(staged))
|
|
title = ""
|
|
if reader.metadata and reader.metadata.title:
|
|
title = str(reader.metadata.title).strip()
|
|
parts: list[str] = []
|
|
for index, page in enumerate(reader.pages):
|
|
if index >= MAX_PDF_PAGES:
|
|
logger.info("deepsearch pdf truncated at %d pages", MAX_PDF_PAGES)
|
|
break
|
|
extracted = page.extract_text() or ""
|
|
if extracted.strip():
|
|
parts.append(extracted)
|
|
text = _normalize("\n\n".join(parts))
|
|
logger.info("deepsearch extracted %d chars from pdf (%d pages)", len(text), len(parts))
|
|
return title, text
|
|
except (PdfReadError, ValueError, OSError) as exc:
|
|
logger.info("deepsearch pdf parse failed: %s", exc)
|
|
return "", ""
|
|
finally:
|
|
staged.unlink(missing_ok=True)
|