"""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