160 lines
4.9 KiB
Python
160 lines
4.9 KiB
Python
"""PDF validation and text extraction.
|
|
|
|
Uploads are read once into memory and parsed from ``io.BytesIO``. Nothing is written
|
|
to disk, so no shared predictable path exists to race on.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import logging
|
|
import re
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from pathlib import PurePosixPath, PureWindowsPath
|
|
|
|
from pypdf import PdfReader
|
|
|
|
from app.core.errors import (
|
|
EncryptedPDFError,
|
|
InvalidPDFError,
|
|
PDFTextUnavailableError,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PDF_SIGNATURE = b"%PDF-"
|
|
# Some real-world PDFs carry a few junk bytes before the header.
|
|
_SIGNATURE_SEARCH_WINDOW = 1024
|
|
|
|
# Below this, extraction produced nothing a reviewer could act on -- almost always a
|
|
# scanned/image-only PDF.
|
|
_MIN_USABLE_CHARS = 30
|
|
|
|
_UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"|?*\x00-\x1f]')
|
|
_CONTROL_CHARS = re.compile(r"[\x01-\x08\x0b\x0c\x0e-\x1f\x7f]")
|
|
_HORIZONTAL_RUNS = re.compile(r"[ \t]{2,}")
|
|
_TRAILING_SPACE = re.compile(r"[ \t]+\n")
|
|
_BLANK_RUNS = re.compile(r"\n{3,}")
|
|
_ALPHANUMERIC = re.compile(r"[A-Za-z0-9]")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExtractedResume:
|
|
"""One resume that survived extraction and is ready to score."""
|
|
|
|
filename: str
|
|
candidate_id: str
|
|
text: str
|
|
page_count: int
|
|
truncated: bool
|
|
|
|
|
|
def sanitize_filename(raw: str | None) -> str:
|
|
"""Reduce an uploaded filename to a bare, safe basename.
|
|
|
|
``PurePosixPath(...).name`` alone is not enough: on POSIX it leaves a
|
|
Windows-style ``..\\..\\evil.pdf`` fully intact. ``PureWindowsPath`` treats both
|
|
``/`` and ``\\`` as separators, so it is applied first.
|
|
"""
|
|
if not raw:
|
|
return "resume.pdf"
|
|
|
|
# Strip control characters before path parsing so pathlib never sees a NUL.
|
|
cleaned = _UNSAFE_FILENAME_CHARS.sub("_", raw)
|
|
name = PureWindowsPath(cleaned).name
|
|
name = PurePosixPath(name).name
|
|
name = name.strip().strip(".")
|
|
|
|
if not name:
|
|
return "resume.pdf"
|
|
return name[:255]
|
|
|
|
|
|
def _normalize_text(text: str) -> str:
|
|
"""Strip NULs and control characters, collapse runs, keep meaningful line breaks."""
|
|
text = text.replace("\x00", "")
|
|
text = text.replace("\r\n", "\n").replace("\r", "\n")
|
|
text = _CONTROL_CHARS.sub("", text)
|
|
text = _HORIZONTAL_RUNS.sub(" ", text)
|
|
text = _TRAILING_SPACE.sub("\n", text)
|
|
text = _BLANK_RUNS.sub("\n\n", text)
|
|
return text.strip()
|
|
|
|
|
|
def _truncate(text: str, max_chars: int) -> tuple[str, bool]:
|
|
"""Cut at a line boundary near the limit rather than mid-word."""
|
|
if len(text) <= max_chars:
|
|
return text, False
|
|
|
|
window = text[:max_chars]
|
|
boundary = window.rfind("\n")
|
|
if boundary >= int(max_chars * 0.8):
|
|
window = window[:boundary]
|
|
return window.rstrip(), True
|
|
|
|
|
|
def extract_resume(data: bytes, filename: str, max_chars: int) -> ExtractedResume:
|
|
"""Validate and extract one PDF.
|
|
|
|
Raises :class:`~app.core.errors.ATSError` subclasses; callers turn those into
|
|
per-candidate failures so one bad file never aborts a batch.
|
|
"""
|
|
if data[:_SIGNATURE_SEARCH_WINDOW].find(PDF_SIGNATURE) == -1:
|
|
raise InvalidPDFError("missing %PDF- signature")
|
|
|
|
try:
|
|
reader = PdfReader(io.BytesIO(data))
|
|
except Exception as exc: # pypdf raises a wide family of parse errors
|
|
raise InvalidPDFError("pypdf failed to open the document") from exc
|
|
|
|
if reader.is_encrypted:
|
|
# Password handling is deliberately out of scope.
|
|
raise EncryptedPDFError("document is encrypted")
|
|
|
|
try:
|
|
pages = list(reader.pages)
|
|
except Exception as exc:
|
|
raise InvalidPDFError("pypdf failed to enumerate pages") from exc
|
|
|
|
if not pages:
|
|
raise InvalidPDFError("document has no pages")
|
|
|
|
page_texts: list[str] = []
|
|
for page in pages:
|
|
try:
|
|
raw_text = page.extract_text() or ""
|
|
except Exception: # a single bad page must not sink the whole document
|
|
raw_text = ""
|
|
page_texts.append(_normalize_text(raw_text))
|
|
|
|
body = "\n".join(chunk for chunk in page_texts if chunk)
|
|
if len(body) < _MIN_USABLE_CHARS or not _ALPHANUMERIC.search(body):
|
|
raise PDFTextUnavailableError("extracted text was empty or unusable")
|
|
|
|
# Page separators are added only after the usability check, so the markers can
|
|
# never make an image-only PDF look like it contained text.
|
|
marked = "\n\n".join(
|
|
f"[Page {index}]\n{chunk}" for index, chunk in enumerate(page_texts, start=1) if chunk
|
|
)
|
|
text, truncated = _truncate(marked, max_chars)
|
|
|
|
resume = ExtractedResume(
|
|
filename=filename,
|
|
candidate_id=uuid.uuid4().hex,
|
|
text=text,
|
|
page_count=len(pages),
|
|
truncated=truncated,
|
|
)
|
|
logger.info(
|
|
"pdf_extracted",
|
|
extra={
|
|
"file_name": resume.filename,
|
|
"candidate_id": resume.candidate_id,
|
|
"page_count": resume.page_count,
|
|
"extracted_chars": len(resume.text),
|
|
"truncated": resume.truncated,
|
|
},
|
|
)
|
|
return resume
|