diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index d7fba702..5e806912 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -136,8 +136,19 @@ _DEFAULTS: dict[str, Any] = { "document_pdf_graphics_limit": 1000, "document_parse_timeout_seconds": 120.0, "document_parse_mem_limit_mb": 1536, - # Tier-0 classifier (shadow mode: emits metrics, no routing change) + # Tier-0 classifier (records classification metrics on the tiered path) "document_classify_enabled": True, + # Tiered PDF pipeline: pypdfium2 is the default/only hot-path extractor; + # "pymupdf" is a deprecated rollback escape hatch. OCR (tier-3) is the only + # escalation target and is off by default (no provider wired yet). + "document_tier1_engine": "pypdfium2", + "document_ocr_enabled": False, + # OCR backend: "auto" picks gateway (if EMBEDDING_GATEWAY_URL) else mistral + # (if MISTRAL_API_KEY); "gateway"/"mistral" force one; "none" disables. + "document_ocr_provider": "auto", + # Provider-namespaced OCR model id (gateway routes on the prefix; the direct + # mistral backend strips it). + "document_ocr_model": "mistral/mistral-ocr-latest", # Observability "metrics_enabled": True, "metrics_port": 9090, @@ -290,7 +301,8 @@ _dynaconf = Dynaconf( Validator("DOCUMENT_CHUNK_OVERLAP", gte=0), # Non-empty strings Validator("VECTOR_SYNC_PDF_TAG", len_min=1), - # Enum constraints + # Enum constraints (document_* enums are validated + normalized in + # __post_init__ via _enum_fields instead, for case-insensitive input). Validator("LOG_FORMAT", is_in=["text", "json"]), Validator( "LOG_LEVEL", @@ -729,9 +741,22 @@ class Settings: # RLIMIT_AS in the parse subprocess (below the pod limit). Applied once per # worker for its lifetime, so changing it needs a pod restart. document_parse_mem_limit_mb: int = 1536 - # Tier-0 classifier. Shadow mode for now: runs a cheap pre-pass over each PDF - # and emits classification metrics, but does NOT change routing yet. + # Tier-0 classifier. Records classification metrics (recommended_tier, + # text-quality) on the tiered path, derived from the tier-1 extraction. document_classify_enabled: bool = True + # PDF extraction engine for the ``fast`` tier. "pypdfium2" (default, + # permissive license, no find_tables) is the hot path; "pymupdf" is a + # deprecated rollback to pymupdf4llm (AGPL, graphics-limited) for one corpus. + document_tier1_engine: str = "pypdfium2" + # Route scanned/no-text-layer PDFs to the tier-3 OCR provider. Off by + # default; when off, the fast tier is terminal. + document_ocr_enabled: bool = False + # OCR backend selection: "auto" | "gateway" | "mistral" | "none". + document_ocr_provider: str = "auto" + # Provider-namespaced OCR model id (e.g. "mistral/mistral-ocr-latest"). The + # gateway routes on the "/" prefix; the direct mistral backend + # strips it. + document_ocr_model: str = "mistral/mistral-ocr-latest" # Observability settings metrics_enabled: bool = True @@ -859,6 +884,8 @@ class Settings: "embedding_provider": {"autodetect", "gateway"}, "mcp_role": {"api", "worker", "all"}, "collection_metadata_source": {"qdrant", "api"}, + "document_tier1_engine": {"pypdfium2", "pymupdf"}, + "document_ocr_provider": {"auto", "gateway", "mistral", "none"}, } for _field, _allowed in _enum_fields.items(): _val = (getattr(self, _field) or "").strip().lower() @@ -1345,6 +1372,10 @@ def get_settings() -> Settings: "document_parse_timeout_seconds": "DOCUMENT_PARSE_TIMEOUT_SECONDS", "document_parse_mem_limit_mb": "DOCUMENT_PARSE_MEM_LIMIT_MB", "document_classify_enabled": "DOCUMENT_CLASSIFY_ENABLED", + "document_tier1_engine": "DOCUMENT_TIER1_ENGINE", + "document_ocr_enabled": "DOCUMENT_OCR_ENABLED", + "document_ocr_provider": "DOCUMENT_OCR_PROVIDER", + "document_ocr_model": "DOCUMENT_OCR_MODEL", # Observability settings "metrics_enabled": "METRICS_ENABLED", "metrics_port": "METRICS_PORT", diff --git a/nextcloud_mcp_server/document_processors/__init__.py b/nextcloud_mcp_server/document_processors/__init__.py index 19978832..622090ba 100644 --- a/nextcloud_mcp_server/document_processors/__init__.py +++ b/nextcloud_mcp_server/document_processors/__init__.py @@ -1,12 +1,20 @@ """Document processing plugins for extracting text from various file formats.""" from .base import DocumentProcessor, ProcessingResult, ProcessorError +from .ocr import OcrProcessor from .pymupdf import PyMuPDFProcessor +from .pypdfium2_fast import Pypdfium2FastProcessor from .registry import ProcessorRegistry, get_registry -# Register processors at module initialization +# Register processors at module initialization. The tiered PDF pipeline selects +# by tier (not priority): Pypdfium2FastProcessor is the ``fast`` tier, +# PyMuPDFProcessor the ``structured`` rollback, and OcrProcessor the ``ocr`` +# escalation target (reached only when document_ocr_enabled). OcrProcessor gets +# the lowest priority so it is never the non-tiered default for PDFs. _registry = get_registry() +_registry.register(Pypdfium2FastProcessor(), priority=20) _registry.register(PyMuPDFProcessor(), priority=10) +_registry.register(OcrProcessor(), priority=1) __all__ = [ "DocumentProcessor", @@ -15,4 +23,6 @@ __all__ = [ "ProcessorRegistry", "get_registry", "PyMuPDFProcessor", + "Pypdfium2FastProcessor", + "OcrProcessor", ] diff --git a/nextcloud_mcp_server/document_processors/classifier.py b/nextcloud_mcp_server/document_processors/classifier.py index f4883ae1..7ad041d4 100644 --- a/nextcloud_mcp_server/document_processors/classifier.py +++ b/nextcloud_mcp_server/document_processors/classifier.py @@ -1,32 +1,31 @@ """Tier-0 document classifier. -A cheap (<~1s), local pre-pass over a PDF that decides which extraction tier a -document should start in, BEFORE the expensive parse. It runs in *shadow mode* -first: emit the signals as metrics, change no routing, and gather per-tenant -data to tune the thresholds. +Decides which extraction tier a PDF should escalate to, from cheap signals: + * text_quality -- is the text layer usable, or mashed/space-less junk? (the + "Student 147" lesson: a text layer can exist yet be unusable, e.g. + "01322234567mobile") + * image_coverage -- a page that is mostly a raster image is a scan/photo whose + content isn't fully in any text layer. + * no text layer -- the strongest OCR signal available from text alone. -Signals (all cheap; no get_drawings, which is itself slow on the graphics-heavy -pages we'd want to flag -- the parse-time ``graphics_limit`` already makes those -safe, and the tier-1 quality gate catches unrecovered tables post-extraction): +Two entry points: + * ``classify_from_text(text, page_boundaries)`` -- the HOT PATH. Derives the + text-quality/no-text-layer signal from the text the registry's tier-1 step + already extracted, so it adds ~no cost. No image analysis. + * ``classify_pdf(content)`` -- a standalone/diagnostic pass that re-opens the + PDF and adds image-coverage analysis. More expensive; used off the hot path. - * text_layer_chars -- extractable text per page - * text_quality -- is the text layer usable, or mashed/space-less junk? - (the "Student 147" lesson: a text layer can exist yet - be unusable, e.g. "01322234567mobile") - * image_coverage -- fraction of the page covered by raster images - (full-page image + poor text => scanned) +Recommended tier: + * ``ocr`` -- scanned / no-usable-text-layer (route to tier 3, when enabled) + * ``fast`` -- a usable digital text layer (stay on tier 1) -From these it picks a recommended starting tier: - * ``ocr`` -- scanned / image-only / bad-text-layer (route to tier 3) - * ``fast`` -- a usable digital text layer (route to tier 1) - -``structured`` (tier 2 / docling) is intentionally not produced here -- that tier -is a separate service and is reached via the tier-1 quality gate, not tier-0. +``structured`` (tier 2 / docling) is a separate service, not produced here. """ import logging import re from dataclasses import dataclass, field +from typing import Any logger = logging.getLogger(__name__) @@ -40,6 +39,8 @@ IMAGE_COVERAGE_SCANNED = 0.80 MIN_TEXT_QUALITY = 0.45 # Fraction of sampled pages that must look scanned/bad for a doc->ocr verdict. OCR_PAGE_FRACTION = 0.5 +# A page with fewer extracted chars than this has effectively no text layer. +MIN_PAGE_CHARS = 16 _WORD_RE = re.compile(r"\S+") @@ -177,3 +178,61 @@ def classify_pdf(content: bytes) -> DocClassification: flags=flags, pages=pages, ) + + +def classify_from_text( + full_text: str, page_boundaries: list[dict[str, Any]] +) -> DocClassification: + """Classify from text already extracted by tier-1 -- no PDF re-open. + + The hot-path classifier: it derives the text-quality signal from the + extraction the registry already ran, so it adds ~no cost (vs ``classify_pdf``, + which re-opens the PDF and re-extracts). It does NOT do image analysis, so it + cannot distinguish a scanned-with-text-layer page (that needs the image pass, + which only matters once OCR routing is enabled). A page with effectively no + text layer is the one OCR-worthy signal available from text alone. + + ``page_boundaries`` are ``{page, start_offset, end_offset}`` indexing into + ``full_text`` (the tier-1/pdf_highlighter contract). + """ + pages: list[PageSignals] = [] + for b in page_boundaries: + seg = full_text[b["start_offset"] : b["end_offset"]] + needs_ocr = len(seg.strip()) < MIN_PAGE_CHARS + pages.append( + PageSignals(b["page"], len(seg), 0.0, _text_quality(seg), needs_ocr) + ) + + sampled = len(pages) + total_chars = sum(p.char_count for p in pages) + mean_quality = ( + round(sum(p.text_quality for p in pages) / sampled, 3) if sampled else 0.0 + ) + # No pages (empty/corrupt PDF) => no OCR evidence => "fast" (the registry's + # page_count guard also skips escalation; defaulting to 0.0 keeps the + # recorded classification metric accurate rather than a misleading "ocr"). + ocr_frac = (sum(p.needs_ocr for p in pages) / sampled) if sampled else 0.0 + + # Flags gated on ocr_frac >= OCR_PAGE_FRACTION (matching classify_pdf): a + # doc that routes "fast" must not carry a junk-layer flag just because a few + # isolated pages are bad -- otherwise the classification metric diverges + # between this hot path and the standalone classify_pdf. + flags: set[str] = set() + if sampled and ocr_frac >= OCR_PAGE_FRACTION: + if total_chars == 0: + flags.add("no_text_layer") + elif mean_quality < MIN_TEXT_QUALITY: + flags.add("bad_text_layer") + + recommended = "ocr" if ocr_frac >= OCR_PAGE_FRACTION else "fast" + + return DocClassification( + page_count=len(page_boundaries), + sampled_pages=sampled, + total_chars=total_chars, + mean_text_quality=mean_quality, + ocr_page_fraction=round(ocr_frac, 3), + recommended_tier=recommended, + flags=flags, + pages=pages, + ) diff --git a/nextcloud_mcp_server/document_processors/ocr.py b/nextcloud_mcp_server/document_processors/ocr.py new file mode 100644 index 00000000..6178ab6c --- /dev/null +++ b/nextcloud_mcp_server/document_processors/ocr.py @@ -0,0 +1,277 @@ +"""Tier-3 OCR processor. + +Routes scanned / no-text-layer PDFs (the tier-0 classifier's ``ocr`` verdict) to +an OCR backend that returns per-page markdown. Two interchangeable backends, +selected by ``document_ocr_provider``: + + * ``gateway`` -- POST the document to the Astrolabe Cloud model gateway's + ``POST /v1/ocr`` (the same M2M-authenticated gateway as embeddings; NO + provider keys in the pod). The platform default. + * ``mistral`` -- call the Mistral OCR API directly from the pod + (``MISTRAL_API_KEY``), for self-hosters / deployments without the gateway. + +``auto`` prefers the gateway (if ``EMBEDDING_GATEWAY_URL`` is set) then direct +Mistral (if ``MISTRAL_API_KEY``). Both return GitHub-flavoured markdown + exact +``page_boundaries``; bbox is re-derived from the PDF bytes + boundaries by +``search/pdf_highlighter``, as for the other tiers. +""" + +import base64 +import logging +from abc import ABC, abstractmethod +from collections.abc import Awaitable, Callable +from typing import Any + +import anyio +import httpx + +from nextcloud_mcp_server.config import Settings, get_settings + +from .base import DocumentProcessor, ProcessingResult + +logger = logging.getLogger(__name__) + +_OCR_TIMEOUT_SECONDS = 180.0 + + +def _pages_to_text( + pages: list[tuple[int, str]], +) -> tuple[str, list[dict[str, Any]]]: + """Join per-page markdown (ordered by index) into one string + boundaries. + + Pages are joined with a blank line. Boundaries are kept CONTIGUOUS (each + page owns its leading ``\\n\\n`` separator) so they index exactly into the + returned text and ``boundaries[-1]["end_offset"] == len(text)`` -- the + ``search/pdf_highlighter`` contract. Consequence: a page's range starts at + its separator, not its first glyph (the fast pypdfium2 path joins with no + separator, so its ranges are glyph-tight). The 2-char offset is immaterial + to page-level chunk attribution. + """ + sep = "\n\n" + parts: list[str] = [] + boundaries: list[dict[str, Any]] = [] + offset = 0 + for i, (index, markdown) in enumerate(sorted(pages, key=lambda p: p[0])): + chunk = markdown if i == 0 else sep + markdown + start = offset + offset += len(chunk) + parts.append(chunk) + boundaries.append( + {"page": index + 1, "start_offset": start, "end_offset": offset} + ) + return "".join(parts), boundaries + + +class _OcrBackend(ABC): + @abstractmethod + async def ocr( + self, content: bytes, mime_type: str + ) -> tuple[str, list[dict[str, Any]]]: ... + + +class _GatewayOcrBackend(_OcrBackend): + """Calls the model gateway's ``POST /v1/ocr`` (key-isolated, M2M-authed).""" + + def __init__(self, base_url: str, model: str, token_provider: Any = None): + base = base_url.rstrip("/") + if not base.endswith("/v1"): + base = f"{base}/v1" + self._url = f"{base}/ocr" + self._model = model + self._token_provider = token_provider + + async def ocr( + self, content: bytes, mime_type: str + ) -> tuple[str, list[dict[str, Any]]]: + headers: dict[str, str] = {} + if self._token_provider is not None: + headers["Authorization"] = ( + f"Bearer {await self._token_provider.get_token()}" + ) + payload = { + "model": self._model, + "document_b64": base64.b64encode(content).decode("ascii"), + "mime_type": mime_type, + } + async with httpx.AsyncClient( + timeout=httpx.Timeout(_OCR_TIMEOUT_SECONDS, connect=10.0) + ) as client: + resp = await client.post(self._url, json=payload, headers=headers) + resp.raise_for_status() + body = resp.json() + pages = [(p["index"], p.get("markdown", "")) for p in body.get("pages", [])] + return _pages_to_text(pages) + + +class _MistralOcrBackend(_OcrBackend): + """Calls the Mistral OCR API directly (provider key lives in the pod).""" + + def __init__(self, api_key: str, model: str, base_url: str | None = None): + from mistralai.client import Mistral # noqa: PLC0415 -- lazy SDK import + + self._client = Mistral(api_key=api_key, server_url=base_url) + # The gateway-namespaced "mistral/" id strips down to the bare + # upstream model the SDK expects. + self._model = model.split("/", 1)[-1] + + async def ocr( + self, content: bytes, mime_type: str + ) -> tuple[str, list[dict[str, Any]]]: + data_url = ( + f"data:{mime_type};base64,{base64.b64encode(content).decode('ascii')}" + ) + resp = await self._client.ocr.process_async( + model=self._model, + document={"type": "document_url", "document_url": data_url}, + ) + pages = [(p.index, p.markdown or "") for p in (resp.pages or [])] + return _pages_to_text(pages) + + +def build_ocr_backend(settings: Settings) -> _OcrBackend | None: + """Select an OCR backend from settings, or None when none is available.""" + provider = settings.document_ocr_provider + if provider == "none": + return None + + if provider in ("gateway", "auto") and settings.embedding_gateway_url: + token_provider = None + if settings.embedding_gateway_client_id: + # Lazy import avoids a document_processors -> embedding cycle at load. + from ..embedding.gateway_client import ( # noqa: PLC0415 + GatewayTokenProvider, + ) + + # Explicit (not assert -- assert is stripped under `python -O`): the + # M2M triple is all-or-nothing. + if not settings.embedding_gateway_token_url: + raise ValueError( + "EMBEDDING_GATEWAY_TOKEN_URL is required when " + "EMBEDDING_GATEWAY_CLIENT_ID is set" + ) + if not settings.embedding_gateway_client_secret: + raise ValueError( + "EMBEDDING_GATEWAY_CLIENT_SECRET is required when " + "EMBEDDING_GATEWAY_CLIENT_ID is set" + ) + token_provider = GatewayTokenProvider( + token_url=settings.embedding_gateway_token_url, + client_id=settings.embedding_gateway_client_id, + client_secret=settings.embedding_gateway_client_secret, + scope=settings.embedding_gateway_scope, + ) + return _GatewayOcrBackend( + settings.embedding_gateway_url, settings.document_ocr_model, token_provider + ) + + if provider in ("mistral", "auto") and settings.mistral_api_key: + return _MistralOcrBackend( + settings.mistral_api_key, + settings.document_ocr_model, + settings.mistral_base_url, + ) + + # An EXPLICIT provider that's missing its config is an operator error -- warn + # loudly (once, since the backend is resolved+cached) rather than silently + # disabling OCR. "auto"/"none" fall through to None quietly by design. + if provider == "gateway": + logger.warning( + "DOCUMENT_OCR_PROVIDER=gateway but EMBEDDING_GATEWAY_URL is unset; " + "OCR is disabled" + ) + elif provider == "mistral": + logger.warning( + "DOCUMENT_OCR_PROVIDER=mistral but MISTRAL_API_KEY is unset; " + "OCR is disabled" + ) + return None + + +class OcrProcessor(DocumentProcessor): + """Tier-3 OCR processor (gateway or direct Mistral backend).""" + + def __init__(self) -> None: + # Resolve the backend once and reuse it: rebuilding per call would create + # a fresh GatewayTokenProvider each time (discarding its M2M-token cache + # -> a token fetch per document) and a new Mistral SDK client per call. + # A config change needs a pod restart anyway, so caching for the pod's + # lifetime is safe. + self._backend_resolved = False + self._backend: _OcrBackend | None = None + # Serialise first-call resolution so a burst of concurrent OCR requests + # doesn't each build a backend (and fetch its own M2M token). Lazy-init: + # anyio primitives must not be created at import time. + self._backend_lock: anyio.Lock | None = None + + @property + def name(self) -> str: + return "ocr" + + @property + def tier(self) -> str: + return "ocr" + + @property + def supported_mime_types(self) -> set[str]: + return {"application/pdf"} + + async def process( + self, + content: bytes, + content_type: str, + filename: str | None = None, + options: dict[str, Any] | None = None, + progress_callback: ( + Callable[[float, float | None, str | None], Awaitable[None]] | None + ) = None, + ) -> ProcessingResult: + settings = get_settings() + if not self._backend_resolved: + if self._backend_lock is None: + self._backend_lock = anyio.Lock() + async with self._backend_lock: + if not self._backend_resolved: # double-checked + self._backend = build_ocr_backend(settings) + self._backend_resolved = True + backend = self._backend + if backend is None: + logger.warning( + "OCR requested for %s but no backend is configured (provider=%s)", + filename or "", + settings.document_ocr_provider, + ) + return ProcessingResult( + text="", + metadata={"parse_failed_reason": "unsupported"}, + processor=self.name, + success=False, + error="no OCR backend configured", + ) + try: + text, boundaries = await backend.ocr( + content, content_type.split(";")[0].strip().lower() + ) + except Exception as e: + logger.warning("OCR failed for %s: %s", filename or "", e) + return ProcessingResult( + text="", + metadata={"parse_failed_reason": "error"}, + processor=self.name, + success=False, + error=f"{type(e).__name__}: {e}", + ) + return ProcessingResult( + text=text, + metadata={ + "page_count": len(boundaries), + "page_boundaries": boundaries, + "file_size": len(content), + }, + processor=self.name, + ) + + async def health_check(self) -> bool: + # Backends are resolved lazily (and configured per tenant), so there is + # nothing to probe here without making a billable upstream call -- the + # processor reports healthy and surfaces a real failure per-document. + return True diff --git a/nextcloud_mcp_server/document_processors/pymupdf.py b/nextcloud_mcp_server/document_processors/pymupdf.py index d1262b02..fa4386d9 100644 --- a/nextcloud_mcp_server/document_processors/pymupdf.py +++ b/nextcloud_mcp_server/document_processors/pymupdf.py @@ -73,6 +73,13 @@ class PyMuPDFProcessor(DocumentProcessor): def name(self) -> str: return "pymupdf" + @property + def tier(self) -> str: + # pymupdf4llm recovers markdown structure (headings, lists, tables) via + # the expensive graphics-limited table detection -- it is the + # ``structured`` escalation target above the pypdfium2 ``fast`` tier. + return "structured" + @property def supported_mime_types(self) -> set[str]: return self.SUPPORTED_TYPES diff --git a/nextcloud_mcp_server/document_processors/pypdfium2_fast.py b/nextcloud_mcp_server/document_processors/pypdfium2_fast.py new file mode 100644 index 00000000..2096d97e --- /dev/null +++ b/nextcloud_mcp_server/document_processors/pypdfium2_fast.py @@ -0,0 +1,128 @@ +"""Tier-1 fast PDF text extractor (pypdfium2). + +A permissively-licensed (Apache/BSD-2) fast path that extracts a PDF's text +layer + page boundaries WITHOUT pymupdf4llm's expensive O(n^2) table/graphics +analysis. For born-digital PDFs (the tier-0 classifier's ``fast`` verdict) this +returns clean text in well under a second -- including the form/table PDFs that +timed out under pymupdf4llm (e.g. ``Student 1a.pdf``: 120s timeout -> ~1s here). + +bbox is re-derived from the PDF bytes + ``page_boundaries`` by +``search/pdf_highlighter``, so this processor only needs to emit ``text`` and +``metadata["page_boundaries"]`` for chunk highlighting to keep working. + +It deliberately does NOT recover tables/layout; a low-quality result is meant to +escalate to the ``structured`` tier (pymupdf4llm, graphics_limit-guarded) via the +registry (B2 escalation wiring). +""" + +import logging +from collections.abc import Awaitable, Callable +from typing import Any + +import anyio + +from .base import DocumentProcessor, ProcessingResult + +logger = logging.getLogger(__name__) + + +def _extract(content: bytes) -> tuple[str, dict[str, Any]]: + """Extract concatenated text + metadata from a PDF (runs in a worker thread). + + ``page_boundaries`` offsets index into the returned text, which is the page + texts joined with no separator so the offsets stay exact (the contract + ``search/pdf_highlighter`` and the chunker rely on). + """ + import pypdfium2 as pdfium # noqa: PLC0415 -- keep the native import lazy + + pdf = pdfium.PdfDocument(content) + try: + page_texts: list[str] = [] + for i in range(len(pdf)): + page = pdf[i] + textpage = page.get_textpage() + try: + page_texts.append(textpage.get_text_bounded() or "") + finally: + textpage.close() + page.close() + doc_meta = pdf.get_metadata_dict() or {} + finally: + pdf.close() + + page_boundaries: list[dict[str, Any]] = [] + offset = 0 + for n, text in enumerate(page_texts, start=1): + page_boundaries.append( + {"page": n, "start_offset": offset, "end_offset": offset + len(text)} + ) + offset += len(text) + + full_text = "".join(page_texts) + metadata: dict[str, Any] = { + "page_count": len(page_texts), + "page_boundaries": page_boundaries, + } + title = doc_meta.get("Title") + if title: + metadata["title"] = title + return full_text, metadata + + +class Pypdfium2FastProcessor(DocumentProcessor): + """Tier-1 fast PDF text extractor backed by pypdfium2.""" + + @property + def name(self) -> str: + return "pypdfium2_fast" + + @property + def tier(self) -> str: + return "fast" + + @property + def supported_mime_types(self) -> set[str]: + return {"application/pdf"} + + async def process( + self, + content: bytes, + content_type: str, + filename: str | None = None, + options: dict[str, Any] | None = None, + progress_callback: ( + Callable[[float, float | None, str | None], Awaitable[None]] | None + ) = None, + ) -> ProcessingResult: + if progress_callback: + await progress_callback(0, 100, "Extracting text (pypdfium2)") + try: + full_text, metadata = await anyio.to_thread.run_sync( # type: ignore[attr-defined] + _extract, content + ) + except Exception as e: + # Fast path is best-effort: a failure here escalates rather than + # crashing the pipeline. pypdfium2 has no O(n^2) bomb, so this is a + # genuinely malformed PDF, not a resource blowup. + logger.warning( + "pypdfium2 fast extract failed for %s: %s", filename or "", e + ) + return ProcessingResult( + text="", + metadata={"parse_failed_reason": "error"}, + processor=self.name, + success=False, + error=f"{type(e).__name__}: {e}", + ) + metadata["file_size"] = len(content) + if progress_callback: + await progress_callback(100, 100, "Done") + return ProcessingResult(text=full_text, metadata=metadata, processor=self.name) + + async def health_check(self) -> bool: + try: + import pypdfium2 # noqa: F401, PLC0415 -- availability probe + + return True + except Exception: + return False diff --git a/nextcloud_mcp_server/document_processors/registry.py b/nextcloud_mcp_server/document_processors/registry.py index 545513de..db412624 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/nextcloud_mcp_server/document_processors/registry.py @@ -5,10 +5,16 @@ import time from collections.abc import Awaitable, Callable from typing import Any -from nextcloud_mcp_server.observability.metrics import record_document_parse +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.observability.metrics import ( + record_document_classification, + record_document_escalation, + record_document_parse, +) from nextcloud_mcp_server.observability.tracing import trace_operation from .base import DocumentProcessor, ProcessingResult, ProcessorError +from .classifier import classify_from_text logger = logging.getLogger(__name__) @@ -140,7 +146,7 @@ class ProcessorRegistry: Raises: ProcessorError: If no processor found or processing fails """ - # Find processor + # Forced processor bypasses tiering. if processor_name: processor = self.get_processor(processor_name) if not processor: @@ -148,14 +154,169 @@ class ProcessorRegistry: f"Processor '{processor_name}' not found. " f"Available: {', '.join(self.list_processors())}" ) - else: + return await self._run_processor( + processor, content, content_type, filename, options, progress_callback + ) + + # PDFs go through the tiered pipeline (tier-0 classify -> tier-1 fast -> + # tier-3 OCR escalation). Everything else uses priority selection. + if content_type.split(";")[0].strip().lower() == "application/pdf": + return await self._process_pdf( + content, content_type, filename, options, progress_callback + ) + + processor = self.find_processor(content_type) + if not processor: + raise ProcessorError( + f"No processor found for type: {content_type}. " + f"Registered processors: {', '.join(self.list_processors())}" + ) + return await self._run_processor( + processor, content, content_type, filename, options, progress_callback + ) + + def _pdf_processor_for_tier(self, tier: str) -> DocumentProcessor | None: + """First registered processor of ``tier`` that handles PDFs.""" + for name in self._priority_order: + processor = self._processors[name][0] + if processor.tier == tier and processor.supports("application/pdf"): + return processor + return None + + async def _process_pdf( + self, + content: bytes, + content_type: str, + filename: str | None, + options: dict[str, Any] | None, + progress_callback: ( + Callable[[float, float | None, str | None], Awaitable[None]] | None + ), + ) -> ProcessingResult: + """Tiered PDF pipeline. + + pypdfium2 ``fast`` extracts first; classification is then derived from + that text (no PDF re-open), and a scanned/no-text-layer doc escalates to + the ``ocr`` tier when enabled. ``document_tier1_engine="pymupdf"`` is a + deprecated rollback that pins the structured engine instead. + """ + settings = get_settings() + + if settings.document_tier1_engine == "pymupdf": + processor = self._pdf_processor_for_tier("structured") + if processor is None: + # The rollback was set to opt OUT of pypdfium2, so falling back + # to it (the highest-priority PDF processor) silently would + # defeat that intent -- warn loudly. + processor = self.find_processor(content_type) + if processor is None: + raise ProcessorError("No PDF processor registered") + logger.warning( + "document_tier1_engine=pymupdf but no 'structured' processor " + "is registered; falling back to '%s'", + processor.name, + ) + return await self._run_processor( + processor, content, content_type, filename, options, progress_callback + ) + + fast = self._pdf_processor_for_tier("fast") + if fast is None: processor = self.find_processor(content_type) - if not processor: - raise ProcessorError( - f"No processor found for type: {content_type}. " - f"Registered processors: {', '.join(self.list_processors())}" + if processor is None: + raise ProcessorError("No PDF processor registered") + return await self._run_processor( + processor, content, content_type, filename, options, progress_callback + ) + + result = await self._run_processor( + fast, content, content_type, filename, options, progress_callback + ) + + # Tier-0 classification from the extraction (cheap: no PDF re-open). + classification = None + if settings.document_classify_enabled and result.success: + try: + classification = classify_from_text( + result.text, result.metadata.get("page_boundaries") or [] + ) + record_document_classification( + classification.recommended_tier, + classification.flags, + classification.mean_text_quality, + ) + except Exception: + logger.warning( + "Tier-0 classification failed for %s", + filename or "", + exc_info=True, ) + # Escalate scanned / no-text-layer PDFs to OCR (tier-3) when enabled and + # a provider is registered. The fast tier is terminal otherwise. Note: a + # fast FAILURE (encrypted/corrupt -- result.success False, no + # classification) is NOT escalated; a PDF pypdfium2 can't open is treated + # as a hard failure (OCR reads the same bytes and would usually fail + # too). The page_count guard skips a zero-page (empty/corrupt) PDF, which + # OCR can't help either. + if ( + classification is not None + and classification.recommended_tier == "ocr" + and classification.page_count > 0 + and settings.document_ocr_enabled + ): + ocr = self._pdf_processor_for_tier("ocr") + if ocr is not None: + reason = ( + "empty_text" + if classification.total_chars == 0 + else "low_confidence" + ) + record_document_escalation("fast", "ocr", reason) + logger.info( + "Escalating %s fast->ocr (reason=%s)", + filename or "", + reason, + ) + ocr_result = await self._run_processor( + ocr, + content, + content_type, + filename, + options, + progress_callback, + escalated=True, + ) + # OCR is an enhancement, not a gate: if it can't run (no backend + # configured / API down) or returns nothing, keep the tier-1 + # result rather than failing the document. Otherwise an operator + # who sets DOCUMENT_OCR_ENABLED=true without credentials would + # make scanned docs fail entirely -- strictly worse than off. + if ocr_result.success: + return ocr_result + logger.warning( + "OCR escalation did not succeed for %s (%s); keeping the " + "tier-1 result", + filename or "", + ocr_result.metadata.get("parse_failed_reason", "error"), + ) + + return result + + async def _run_processor( + self, + processor: DocumentProcessor, + content: bytes, + content_type: str, + filename: str | None = None, + options: dict[str, Any] | None = None, + progress_callback: ( + Callable[[float, float | None, str | None], Awaitable[None]] | None + ) = None, + *, + escalated: bool = False, + ) -> ProcessingResult: + """Run one processor with the per-processor span + parse metrics.""" tier = processor.tier logger.info( "Processing with '%s' processor", @@ -167,11 +328,6 @@ class ProcessorRegistry: }, ) - # Process (instrumented: per-processor span + parse metrics). - # NOTE: when the tiered pipeline (docling/OCR/LLM) lands, escalation - # decisions are recorded here via record_document_escalation() and an - # add_span_event("document.escalation", ...) -- the escalated=False - # attribute and the metric are wired ahead of that. byte_size = len(content) start_time = time.time() with trace_operation( @@ -181,7 +337,7 @@ class ProcessorRegistry: "processor.tier": tier, "mime_type": content_type, "byte_size": byte_size, - "escalated": False, + "escalated": escalated, }, record_exception=True, ) as span: @@ -217,6 +373,10 @@ class ProcessorRegistry: raise duration = time.time() - start_time + # Record the tier that actually produced this result so downstream + # (Qdrant payload pipeline_tier, analytics) reflects escalation + # instead of a hardcoded "fast". + result.metadata.setdefault("pipeline_tier", tier) pages = int(result.metadata.get("page_count", 0) or 0) chars = len(result.text) status = "success" if result.success else "error" diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 6ae2291c..a0669a71 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -17,12 +17,10 @@ from nextcloud_mcp_server.acl_hash import compute_acl_hash from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.document_processors import get_registry -from nextcloud_mcp_server.document_processors.classifier import classify_pdf from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service from nextcloud_mcp_server.models.deck import DeckCard from nextcloud_mcp_server.observability.metrics import ( record_document_chunks, - record_document_classification, record_document_parse_failed, record_embedding, record_qdrant_operation, @@ -167,36 +165,6 @@ async def processor_task( logger.info("Processor %s stopped", worker_id) -async def _shadow_classify(content: bytes, content_type: str, file_path: str) -> None: - """Tier-0 classification in SHADOW mode: emit metrics, change no routing. - - Best-effort and out of the indexing critical path -- it must never block or - fail indexing. PDFs only (the classifier is PDF-specific). The cheap pre-pass - runs in a worker thread so it doesn't stall the event loop. - """ - if content_type != "application/pdf": - return - try: - c = await anyio.to_thread.run_sync(classify_pdf, content) # type: ignore[attr-defined] - record_document_classification(c.recommended_tier, c.flags, c.mean_text_quality) - logger.debug( - "Tier-0 classified %s: tier=%s flags=%s quality=%s", - file_path, - c.recommended_tier, - sorted(c.flags), - c.mean_text_quality, - ) - except Exception: - # Best-effort: shadow classification must never break indexing, but log - # at WARNING (not DEBUG) so a systematic failure -- a pymupdf bug, memory - # pressure on every PDF -- stays visible at the production LOG_LEVEL=INFO. - logger.warning( - "Tier-0 classification failed for %s (shadow mode, indexing unaffected)", - file_path, - exc_info=True, - ) - - async def process_document( doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3 ): @@ -571,11 +539,8 @@ async def _index_document( "vector_sync.file_size": len(content_bytes), }, ): - # Tier-0 shadow classification (observability only; no routing change). - if settings.document_classify_enabled: - await _shadow_classify(content_bytes, content_type, file_path) - - # Use document processor registry to extract text + # The registry runs the tiered PDF pipeline (tier-0 classify -> + # tier-1 fast -> OCR escalation) and records classification metrics. registry = get_registry() try: @@ -632,20 +597,12 @@ async def _index_document( # Diagnostic: Log page boundary information if available if "page_boundaries" in file_metadata: page_boundaries = file_metadata["page_boundaries"] - logger.info( + logger.debug( "Page boundaries for %s: %s pages, text length: %s", file_path, len(page_boundaries), len(content), ) - # Log first 3 page boundaries for debugging - for boundary in page_boundaries[:3]: - logger.debug( - " Page %s: offsets [%s:%s]", - boundary["page"], - boundary["start_offset"], - boundary["end_offset"], - ) # Verify last boundary matches text length if page_boundaries: last_boundary = page_boundaries[-1] @@ -695,23 +652,13 @@ async def _index_document( # Diagnostic: Verify page number assignment assigned_count = sum(1 for c in chunks if c.page_number is not None) - logger.info( + logger.debug( "Assigned page numbers to %s/%s chunks for %s", assigned_count, len(chunks), file_path, ) - # Log first 3 chunks to see their page assignments - for i, chunk in enumerate(chunks[:3]): - logger.debug( - " Chunk %s: page=%s, offsets=[%s:%s]", - i, - chunk.page_number, - chunk.start_offset, - chunk.end_offset, - ) - # Warning if NO page numbers were assigned if assigned_count == 0: logger.warning( @@ -969,7 +916,11 @@ async def _index_document( # Decomposition payload keys (design ยง10.2), additive. payload_keys.PROCESSOR_VERSION: "monolith-v1", payload_keys.PARSED_AT: indexed_at, - payload_keys.PIPELINE_TIER: "fast", + # Actual tier that produced this doc (registry stamps it on + # the result metadata); non-PDF doc types stay "fast". + payload_keys.PIPELINE_TIER: file_metadata.get( + "pipeline_tier", "fast" + ), payload_keys.EMBEDDING_IDENTITY: _embedding_identity, payload_keys.ACL_HASH: _acl_hash, # File-specific metadata (PDF, etc.) diff --git a/pyproject.toml b/pyproject.toml index 96791e7e..ab4ed9f1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,7 @@ dependencies = [ "dynaconf>=3.2.13,<4.0", "mistralai>=2.4.5", "sqlalchemy[asyncio]>=2.0", + "pypdfium2>=5.9.0", ] classifiers = [ "Development Status :: 4 - Beta", diff --git a/tests/unit/test_decomposition_config.py b/tests/unit/test_decomposition_config.py index 7f4b31a1..b51343d6 100644 --- a/tests/unit/test_decomposition_config.py +++ b/tests/unit/test_decomposition_config.py @@ -29,9 +29,13 @@ class TestDecompositionDefaults: s = Settings( collection_metadata_source=" QDRANT ", mcp_role=" API ", + document_tier1_engine=" PyPDFium2 ", + document_ocr_provider=" Gateway ", ) assert s.collection_metadata_source == "qdrant" assert s.mcp_role == "api" + assert s.document_tier1_engine == "pypdfium2" + assert s.document_ocr_provider == "gateway" class TestEnumValidation: @@ -41,6 +45,8 @@ class TestEnumValidation: ("embedding_provider", "openai"), ("mcp_role", "leader"), ("collection_metadata_source", "redis"), + ("document_tier1_engine", "mupdf"), + ("document_ocr_provider", "gatway"), ], ) def test_invalid_enum_rejected(self, field, value): diff --git a/tests/unit/test_doc_classifier.py b/tests/unit/test_doc_classifier.py index cc86c80d..bc5306d1 100644 --- a/tests/unit/test_doc_classifier.py +++ b/tests/unit/test_doc_classifier.py @@ -156,3 +156,32 @@ def test_image_heavy_flag_without_ocr_routing(): assert "image_heavy" in c.flags assert c.recommended_tier == "fast" assert c.ocr_page_fraction < clf.OCR_PAGE_FRACTION + + +# --- classify_from_text (hot-path, derived from tier-1 extraction) ----------- + + +def test_classify_from_text_clean_routes_fast(): + txt = "the quick brown fox jumps over the lazy dog " * 3 + c = clf.classify_from_text( + txt, [{"page": 1, "start_offset": 0, "end_offset": len(txt)}] + ) + assert c.recommended_tier == "fast" + assert c.mean_text_quality > 0.8 + assert c.flags == set() + + +def test_classify_from_text_empty_routes_ocr(): + c = clf.classify_from_text("", [{"page": 1, "start_offset": 0, "end_offset": 0}]) + assert c.recommended_tier == "ocr" + assert "no_text_layer" in c.flags + assert c.total_chars == 0 + + +def test_classify_from_text_no_pages_routes_fast(): + # An empty/corrupt PDF (no page boundaries) is not OCR evidence -> "fast", + # so the recorded classification metric isn't a misleading "ocr". + c = clf.classify_from_text("", []) + assert c.recommended_tier == "fast" + assert c.ocr_page_fraction == pytest.approx(0.0) + assert c.flags == set() diff --git a/tests/unit/test_ocr_processor.py b/tests/unit/test_ocr_processor.py new file mode 100644 index 00000000..07883027 --- /dev/null +++ b/tests/unit/test_ocr_processor.py @@ -0,0 +1,117 @@ +"""Unit tests for the tier-3 OCR processor + backend selection.""" + +from types import SimpleNamespace +from typing import Any + +import pytest + +from nextcloud_mcp_server.document_processors import ocr + +pytestmark = pytest.mark.unit + + +def _settings(**kw) -> Any: # a Settings stand-in (only the read fields matter) + base = dict( + document_ocr_provider="auto", + document_ocr_model="mistral/mistral-ocr-latest", + embedding_gateway_url=None, + embedding_gateway_client_id=None, + embedding_gateway_client_secret=None, + embedding_gateway_token_url=None, + embedding_gateway_scope=None, + mistral_api_key=None, + mistral_base_url=None, + ) + base.update(kw) + return SimpleNamespace(**base) + + +# --- _pages_to_text ---------------------------------------------------------- + + +def test_pages_to_text_orders_and_exact_boundaries(): + text, boundaries = ocr._pages_to_text([(1, "B"), (0, "A")]) # out of order + assert text == "A\n\nB" + assert boundaries[0] == {"page": 1, "start_offset": 0, "end_offset": 1} + assert boundaries[1]["page"] == 2 + # contiguous + offsets index exactly into the text + assert boundaries[0]["end_offset"] <= boundaries[1]["start_offset"] + assert boundaries[-1]["end_offset"] == len(text) + + +# --- backend selection ------------------------------------------------------- + + +def test_build_backend_none(): + assert ocr.build_ocr_backend(_settings(document_ocr_provider="none")) is None + + +def test_build_backend_gateway(): + b = ocr.build_ocr_backend( + _settings(document_ocr_provider="gateway", embedding_gateway_url="http://gw") + ) + assert isinstance(b, ocr._GatewayOcrBackend) + + +def test_build_backend_mistral(): + b = ocr.build_ocr_backend( + _settings(document_ocr_provider="mistral", mistral_api_key="k") + ) + assert isinstance(b, ocr._MistralOcrBackend) + + +def test_build_backend_auto_prefers_gateway(): + b = ocr.build_ocr_backend( + _settings(embedding_gateway_url="http://gw", mistral_api_key="k") + ) + assert isinstance(b, ocr._GatewayOcrBackend) + + +def test_build_backend_auto_none_configured(): + assert ocr.build_ocr_backend(_settings()) is None + + +def test_gateway_backend_url_normalization(): + b = ocr._GatewayOcrBackend("http://gw", "mistral/mistral-ocr-latest") + assert b._url == "http://gw/v1/ocr" + b2 = ocr._GatewayOcrBackend("http://gw/v1/", "m") + assert b2._url == "http://gw/v1/ocr" + + +# --- OcrProcessor ------------------------------------------------------------ + + +async def test_processor_unsupported_when_no_backend(monkeypatch): + monkeypatch.setattr( + ocr, "get_settings", lambda: _settings(document_ocr_provider="none") + ) + monkeypatch.setattr(ocr, "build_ocr_backend", lambda s: None) + r = await ocr.OcrProcessor().process(b"%PDF-1.7", "application/pdf") + assert r.success is False + assert r.metadata["parse_failed_reason"] == "unsupported" + + +async def test_processor_success(monkeypatch): + class _FakeBackend: + async def ocr(self, content, mime_type): + return "hello world", [{"page": 1, "start_offset": 0, "end_offset": 11}] + + monkeypatch.setattr(ocr, "get_settings", lambda: _settings()) + monkeypatch.setattr(ocr, "build_ocr_backend", lambda s: _FakeBackend()) + r = await ocr.OcrProcessor().process(b"%PDF-1.7", "application/pdf") + assert r.success is True + assert r.text == "hello world" + assert r.metadata["page_count"] == 1 + assert r.processor == "ocr" + + +async def test_processor_backend_error_returns_success_false(monkeypatch): + class _BoomBackend: + async def ocr(self, content, mime_type): + raise RuntimeError("api down") + + monkeypatch.setattr(ocr, "get_settings", lambda: _settings()) + monkeypatch.setattr(ocr, "build_ocr_backend", lambda s: _BoomBackend()) + r = await ocr.OcrProcessor().process(b"%PDF-1.7", "application/pdf") + assert r.success is False + assert r.metadata["parse_failed_reason"] == "error" diff --git a/tests/unit/test_pypdfium2_fast.py b/tests/unit/test_pypdfium2_fast.py new file mode 100644 index 00000000..04f13864 --- /dev/null +++ b/tests/unit/test_pypdfium2_fast.py @@ -0,0 +1,59 @@ +"""Unit tests for the tier-1 pypdfium2 fast PDF extractor.""" + +import pymupdf +import pytest + +from nextcloud_mcp_server.document_processors.pypdfium2_fast import ( + Pypdfium2FastProcessor, +) + +pytestmark = pytest.mark.unit + + +def _digital_pdf( + pages: int = 3, body: str = "Hello world this is clean text. " +) -> bytes: + doc = pymupdf.open() + for _ in range(pages): + page = doc.new_page(width=595, height=842) + page.insert_text((50, 60), body * 8) + data: bytes = doc.tobytes() + doc.close() + return data + + +def test_processor_identity(): + p = Pypdfium2FastProcessor() + assert p.name == "pypdfium2_fast" + assert p.tier == "fast" + assert "application/pdf" in p.supported_mime_types + + +async def test_extract_text_and_exact_page_boundaries(): + p = Pypdfium2FastProcessor() + result = await p.process(_digital_pdf(pages=3), "application/pdf", filename="t.pdf") + + assert result.success is True + assert "Hello world" in result.text + assert result.metadata["page_count"] == 3 + + boundaries = result.metadata["page_boundaries"] + assert len(boundaries) == 3 + assert boundaries[0]["start_offset"] == 0 + # Offsets must index exactly into the returned text (pdf_highlighter contract). + assert boundaries[-1]["end_offset"] == len(result.text) + for prev, nxt in zip(boundaries, boundaries[1:]): + assert prev["end_offset"] == nxt["start_offset"] + + +async def test_malformed_pdf_returns_success_false(): + p = Pypdfium2FastProcessor() + result = await p.process(b"not a pdf at all", "application/pdf", filename="bad.pdf") + + assert result.success is False + assert result.text == "" + assert result.metadata["parse_failed_reason"] == "error" + + +async def test_health_check(): + assert await Pypdfium2FastProcessor().health_check() is True diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py new file mode 100644 index 00000000..769c2c4b --- /dev/null +++ b/tests/unit/test_registry_tiering.py @@ -0,0 +1,184 @@ +"""Unit tests for the tiered PDF routing in ProcessorRegistry. + +Covers: default fast-tier routing, the pymupdf rollback toggle, classification +recording derived from the extraction, and OCR escalation (on/off). +""" + +from unittest.mock import MagicMock + +import pytest + +from nextcloud_mcp_server.document_processors import registry as reg_mod +from nextcloud_mcp_server.document_processors.base import ( + DocumentProcessor, + ProcessingResult, +) +from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry + +pytestmark = pytest.mark.unit + + +class _Fake(DocumentProcessor): + def __init__( + self, + name: str, + tier: str, + text: str = "clean text here", + success=True, + pages: int = 1, + ): + self._name = name + self._tier = tier + self._text = text + self._success = success + self._pages = pages + + @property + def name(self) -> str: + return self._name + + @property + def tier(self) -> str: + return self._tier + + @property + def supported_mime_types(self) -> set[str]: + return {"application/pdf"} + + async def process( + self, content, content_type, filename=None, options=None, progress_callback=None + ): + boundaries = ( + [{"page": 1, "start_offset": 0, "end_offset": len(self._text)}] + if self._pages + else [] + ) + return ProcessingResult( + text=self._text, + metadata={"page_count": self._pages, "page_boundaries": boundaries}, + processor=self._name, + success=self._success, + ) + + async def health_check(self) -> bool: + return True + + +class _Settings: + def __init__(self, engine="pypdfium2", classify=True, ocr=False): + self.document_tier1_engine = engine + self.document_classify_enabled = classify + self.document_ocr_enabled = ocr + + +def _registry(*procs: tuple[DocumentProcessor, int]) -> ProcessorRegistry: + r = ProcessorRegistry() + for proc, prio in procs: + r.register(proc, priority=prio) + return r + + +async def test_pdf_routes_to_fast_tier(monkeypatch): + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings()) + r = _registry((_Fake("fast", "fast"), 20), (_Fake("structured", "structured"), 10)) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "fast" + + +async def test_engine_rollback_uses_structured(monkeypatch): + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(engine="pymupdf")) + r = _registry((_Fake("fast", "fast"), 20), (_Fake("structured", "structured"), 10)) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "structured" + + +async def test_engine_rollback_warns_when_no_structured(monkeypatch, caplog): + # pymupdf rollback with no structured processor registered: it falls back to + # the fast processor but must warn (it silently used what the user opted out + # of otherwise). + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(engine="pymupdf")) + r = _registry((_Fake("fast", "fast"), 20)) + with caplog.at_level( + "WARNING", logger="nextcloud_mcp_server.document_processors.registry" + ): + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "fast" + assert any("no 'structured' processor" in rec.message for rec in caplog.records) + + +async def test_records_classification(monkeypatch): + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings()) + rec = MagicMock() + monkeypatch.setattr(reg_mod, "record_document_classification", rec) + r = _registry((_Fake("fast", "fast"), 20)) + await r.process(b"%PDF-1.7", "application/pdf") + rec.assert_called_once() + + +async def test_classify_disabled_skips_recording(monkeypatch): + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(classify=False)) + rec = MagicMock() + monkeypatch.setattr(reg_mod, "record_document_classification", rec) + r = _registry((_Fake("fast", "fast"), 20)) + await r.process(b"%PDF-1.7", "application/pdf") + rec.assert_not_called() + + +async def test_ocr_escalation_on_empty_text(monkeypatch): + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(ocr=True)) + esc = MagicMock() + monkeypatch.setattr(reg_mod, "record_document_escalation", esc) + r = _registry( + (_Fake("fast", "fast", text=""), 20), + (_Fake("ocr", "ocr", text="ocr text"), 5), + ) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "ocr" + esc.assert_called_once() + + +async def test_zero_page_pdf_does_not_escalate(monkeypatch): + # An empty/corrupt PDF (no pages) classifies "ocr" but must NOT escalate -- + # OCR can't help and it would be wasteful. + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(ocr=True)) + esc = MagicMock() + monkeypatch.setattr(reg_mod, "record_document_escalation", esc) + r = _registry( + (_Fake("fast", "fast", text="", pages=0), 20), + (_Fake("ocr", "ocr"), 5), + ) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "fast" + esc.assert_not_called() + + +async def test_pipeline_tier_stamped_on_metadata(monkeypatch): + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings()) + r = _registry((_Fake("fast", "fast"), 20)) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.metadata["pipeline_tier"] == "fast" + + +async def test_ocr_failure_falls_back_to_fast(monkeypatch): + # OCR enabled but the backend can't run (no creds / API down) -> keep the + # tier-1 result instead of failing the document. + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(ocr=True)) + monkeypatch.setattr(reg_mod, "record_document_escalation", MagicMock()) + r = _registry( + (_Fake("fast", "fast", text=""), 20), + (_Fake("ocr", "ocr", text="", success=False), 5), + ) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "fast" + assert res.success is True + + +async def test_no_ocr_escalation_when_disabled(monkeypatch): + monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(ocr=False)) + r = _registry( + (_Fake("fast", "fast", text=""), 20), + (_Fake("ocr", "ocr"), 5), + ) + res = await r.process(b"%PDF-1.7", "application/pdf") + # Fast tier is terminal when OCR is disabled. + assert res.processor == "fast" diff --git a/tests/unit/test_shadow_classify.py b/tests/unit/test_shadow_classify.py deleted file mode 100644 index c0110379..00000000 --- a/tests/unit/test_shadow_classify.py +++ /dev/null @@ -1,62 +0,0 @@ -"""Tests for the tier-0 shadow-classification wiring in the processor. - -Shadow mode = observability only: it emits classification metrics but must never -block or fail indexing, and only applies to PDFs. -""" - -from unittest.mock import MagicMock - -import pytest - -from nextcloud_mcp_server.document_processors.classifier import DocClassification -from nextcloud_mcp_server.vector import processor as proc - -pytestmark = pytest.mark.unit - - -def _classification() -> DocClassification: - return DocClassification( - page_count=2, - sampled_pages=2, - total_chars=100, - mean_text_quality=0.9, - ocr_page_fraction=0.0, - recommended_tier="fast", - flags={"image_heavy"}, - ) - - -async def test_shadow_classify_records_metrics(monkeypatch): - monkeypatch.setattr(proc, "classify_pdf", lambda content: _classification()) - rec = MagicMock() - monkeypatch.setattr(proc, "record_document_classification", rec) - - await proc._shadow_classify(b"%PDF-1.7", "application/pdf", "f.pdf") - - rec.assert_called_once_with("fast", {"image_heavy"}, 0.9) - - -async def test_shadow_classify_skips_non_pdf(monkeypatch): - called = MagicMock() - monkeypatch.setattr(proc, "classify_pdf", called) - rec = MagicMock() - monkeypatch.setattr(proc, "record_document_classification", rec) - - await proc._shadow_classify(b"plain", "text/plain", "f.txt") - - called.assert_not_called() - rec.assert_not_called() - - -async def test_shadow_classify_swallows_errors(monkeypatch): - def boom(content): - raise ValueError("bad pdf") - - monkeypatch.setattr(proc, "classify_pdf", boom) - rec = MagicMock() - monkeypatch.setattr(proc, "record_document_classification", rec) - - # Must not raise -- shadow classification is best-effort, off the index path. - await proc._shadow_classify(b"%PDF-1.7", "application/pdf", "f.pdf") - - rec.assert_not_called() diff --git a/uv.lock b/uv.lock index 7965d208..b220f2f7 100644 --- a/uv.lock +++ b/uv.lock @@ -2215,6 +2215,7 @@ dependencies = [ { name = "pyjwt", extra = ["crypto"] }, { name = "pymupdf" }, { name = "pymupdf4llm" }, + { name = "pypdfium2" }, { name = "python-json-logger" }, { name = "pythonvcard4" }, { name = "qdrant-client" }, @@ -2283,6 +2284,7 @@ requires-dist = [ { name = "pyjwt", extras = ["crypto"], specifier = ">=2.8.0" }, { name = "pymupdf", specifier = ">=1.26.6" }, { name = "pymupdf4llm", specifier = ">=0.2.2" }, + { name = "pypdfium2", specifier = ">=5.9.0" }, { name = "python-json-logger", specifier = ">=3.2.0" }, { name = "pythonvcard4", specifier = ">=0.2.0" }, { name = "qdrant-client", specifier = ">=1.17.0" }, @@ -3455,6 +3457,35 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c0/1e/5fae75a5dc478e376ab95253c2f611665a4d9e2249667387a975e00fbbcb/pymupdf4llm-0.2.7-py3-none-any.whl", hash = "sha256:3ac6b0344c8bade2c97c3d7ea5eb354c71383a8d1ca177fafc3519dd564273b7", size = 66905, upload-time = "2025-12-07T20:43:12.447Z" }, ] +[[package]] +name = "pypdfium2" +version = "5.9.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b0/98/6b44bf82ddb3c7a3e0249203772aad8981b4491d6227f182685f310faeff/pypdfium2-5.9.0.tar.gz", hash = "sha256:db1274bd27844db6fda17ef1dbcd0026c47d357437058d838e98060c0da9e92e", size = 272455, upload-time = "2026-06-01T15:43:38.08Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8b/d9/59630cb40e5f37e7712e6ea65e9cac633f4195e8b737bb3a46054aa63340/pypdfium2-5.9.0-py3-none-android_23_arm64_v8a.whl", hash = "sha256:91914837c4a4285b3e0724a84eca8079363db7475acbcab405933d1807785664", size = 3407817, upload-time = "2026-06-01T15:42:58.426Z" }, + { url = "https://files.pythonhosted.org/packages/0f/3d/e205708835a3730d5242652b6577ac06ad4721e6fcef77cc7c9d3541c686/pypdfium2-5.9.0-py3-none-android_23_armeabi_v7a.whl", hash = "sha256:90610d352f050b065b703f3a46602a852fce7dd8787300c8c7a472485b644d8f", size = 2862706, upload-time = "2026-06-01T15:43:00.581Z" }, + { url = "https://files.pythonhosted.org/packages/01/47/e843fb895a891438b3f8c6d834fdc9c19183cd60980fc9325429d5c01505/pypdfium2-5.9.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:6c4fbe3a7190b329c526358fb2855d797f7b74b5ecfc61d19657ef20bcebc108", size = 3489945, upload-time = "2026-06-01T15:43:02.542Z" }, + { url = "https://files.pythonhosted.org/packages/35/bd/f5e6afd556f97fcaa2bec4cb04669664c166028fc2a059bd65447c852b43/pypdfium2-5.9.0-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:e93f0cf440169a3e445e6fbd06c803877e7418f3e13254287875cb67f208bb5a", size = 3674186, upload-time = "2026-06-01T15:43:04.496Z" }, + { url = "https://files.pythonhosted.org/packages/6d/4d/5286812216a292d51dfba8e7bff276da198f126508f8c2afa3630bf701dc/pypdfium2-5.9.0-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:1d902e03dff5efd51d93cd23d3e55bde53802fa6207bcd0e455239518859a069", size = 3669571, upload-time = "2026-06-01T15:43:06.571Z" }, + { url = "https://files.pythonhosted.org/packages/ac/c8/822db2c89baa13e6cee321d587fcd42df463a1fc2f7520b3f6814768bc71/pypdfium2-5.9.0-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:6cf38d7ad3575947b82384869f2ab69ba345eb21d83118d25db3e83f967b0421", size = 3400412, upload-time = "2026-06-01T15:43:08.35Z" }, + { url = "https://files.pythonhosted.org/packages/1a/dd/7d09d8cdc28383df13f739a97ac4f1215a704a97a29506dee2bf89d8a350/pypdfium2-5.9.0-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:77f7479a28b43aa658735e3ce79cfd1fccd5d42db035c21bb4c26e8bd7e280e5", size = 3803326, upload-time = "2026-06-01T15:43:10.054Z" }, + { url = "https://files.pythonhosted.org/packages/99/58/3f4e04ffe1ae62b437de07a96da672091cef62b619d0dc78207c1af442e6/pypdfium2-5.9.0-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:07e6ba170d577eabf60dbba701d051c64318dd029d38ca5907d83ae1a66fe779", size = 4216890, upload-time = "2026-06-01T15:43:11.701Z" }, + { url = "https://files.pythonhosted.org/packages/1d/f6/2dde4656750c4a6da99e1f070ca09d2b5a9d68186b42e711a1a3e5b1cb32/pypdfium2-5.9.0-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7ce3a3dd23ec0adaa079d8be54565ba2aa2f6060e76a4989cd42dabc163d74ee", size = 3728830, upload-time = "2026-06-01T15:43:13.329Z" }, + { url = "https://files.pythonhosted.org/packages/d0/ca/f2ff8b9200c7dfc5aee85126edc856eb93c7056085da2454a75ef1e4dbc4/pypdfium2-5.9.0-py3-none-manylinux_2_27_s390x.manylinux_2_28_s390x.whl", hash = "sha256:ae177938f5cf95a275db25a4f8553e2ebd954ecda2f9bc84848ba4b027ce438f", size = 4063322, upload-time = "2026-06-01T15:43:15.158Z" }, + { url = "https://files.pythonhosted.org/packages/64/88/0b587de03c873c28adc59f6ac959de4032d3f3bc946094523b14a192d9c3/pypdfium2-5.9.0-py3-none-manylinux_2_38_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:ffe49edde2ac86f28ca7e58f565255a442f38a7508fff31b79a55f508f25a31e", size = 4039738, upload-time = "2026-06-01T15:43:16.975Z" }, + { url = "https://files.pythonhosted.org/packages/83/4c/fa627f00a954e66465e929077cf43bd012595091fff82758d989486e7bdc/pypdfium2-5.9.0-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:b7b760bc2957ecf73c274af6ed8b168a2dcb328ac0a0f7ed6123cd92f6e7c9c9", size = 4997259, upload-time = "2026-06-01T15:43:18.915Z" }, + { url = "https://files.pythonhosted.org/packages/32/f0/1736d80c5d12d931f74ca6b4213b006ee016ec33c6325fad870234cc240c/pypdfium2-5.9.0-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:7cdc8e5d2f8d82add1e4f70a4fbe5f3b33c17f301ebde38c669fd7f78a7d032c", size = 4537061, upload-time = "2026-06-01T15:43:20.879Z" }, + { url = "https://files.pythonhosted.org/packages/01/00/aa8890dfd385b2e7365034231987029cff15cc7eb4f06e8380da5608738a/pypdfium2-5.9.0-py3-none-musllinux_1_2_i686.whl", hash = "sha256:38a058dbd4929acaf0ab9171179eb86c24d8c6655a6836006796105a9f200890", size = 5232786, upload-time = "2026-06-01T15:43:23.73Z" }, + { url = "https://files.pythonhosted.org/packages/65/12/8f45ea698781a0bed96ac4fbde440060790863273943461f0f160a993d52/pypdfium2-5.9.0-py3-none-musllinux_1_2_ppc64le.whl", hash = "sha256:1894511a0e862e7ec5679f3a6dc43ac72c4ef92c7ca438357203913e8634a643", size = 5170121, upload-time = "2026-06-01T15:43:25.858Z" }, + { url = "https://files.pythonhosted.org/packages/25/bd/9bb6ba375796e1de1d6c1af8d8303dd1781190346871c81a94d4e09eddfd/pypdfium2-5.9.0-py3-none-musllinux_1_2_riscv64.whl", hash = "sha256:040f5513b808db705d4878f57e2bf0b9dc6e6a0ad8d765c36cf62febf3933b28", size = 4663540, upload-time = "2026-06-01T15:43:27.677Z" }, + { url = "https://files.pythonhosted.org/packages/d2/4a/fd103bac197f22038bf70be1f7507ced7519f1214ea0dae137f37803ab8a/pypdfium2-5.9.0-py3-none-musllinux_1_2_s390x.whl", hash = "sha256:f4991ae39bcea757552579bba4aebfaedb71c96dd35c2292f957b8ac9132f1ff", size = 5090619, upload-time = "2026-06-01T15:43:29.522Z" }, + { url = "https://files.pythonhosted.org/packages/22/89/9531fa1e6e004fe522cdca0cd945cd6a9d7338e7125e6b0734d632d31fa6/pypdfium2-5.9.0-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:25ff1a5abd08ff9e87f62e5dac114ea95647c257fbbdbe029be8db71a6d7650b", size = 5050806, upload-time = "2026-06-01T15:43:31.322Z" }, + { url = "https://files.pythonhosted.org/packages/fc/d0/e53c68555ff128b2470e4a468762b320d9c6ae2c914decea3487d923982f/pypdfium2-5.9.0-py3-none-win32.whl", hash = "sha256:b0057dc8c2033584dc3e61afb5f23a135dab52b081695b435e27f9b7b074c605", size = 3670966, upload-time = "2026-06-01T15:43:32.991Z" }, + { url = "https://files.pythonhosted.org/packages/da/0c/22e5fc035ad1594b44f265bc0a59ae34d377bc2ea74a92793e7a674bf96d/pypdfium2-5.9.0-py3-none-win_amd64.whl", hash = "sha256:06508c33b9772cf3878e48364c6e14c70cefc18a3abd6983ac9f338da9305275", size = 3800959, upload-time = "2026-06-01T15:43:34.536Z" }, + { url = "https://files.pythonhosted.org/packages/11/e3/cf1711add7add22a17f7c7633cd795edc92f17ab7bdf1930493ae0f56680/pypdfium2-5.9.0-py3-none-win_arm64.whl", hash = "sha256:565ddfc98795fd2f6054b544ee9791d7b9032f9cf77a57891b6e501fafd0ef3f", size = 3585718, upload-time = "2026-06-01T15:43:36.521Z" }, +] + [[package]] name = "pyreadline3" version = "3.5.4"