diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 4dbd204f..d7fba702 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -136,6 +136,8 @@ _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) + "document_classify_enabled": True, # Observability "metrics_enabled": True, "metrics_port": 9090, @@ -727,6 +729,9 @@ 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. + document_classify_enabled: bool = True # Observability settings metrics_enabled: bool = True @@ -1339,6 +1344,7 @@ def get_settings() -> Settings: "document_pdf_graphics_limit": "DOCUMENT_PDF_GRAPHICS_LIMIT", "document_parse_timeout_seconds": "DOCUMENT_PARSE_TIMEOUT_SECONDS", "document_parse_mem_limit_mb": "DOCUMENT_PARSE_MEM_LIMIT_MB", + "document_classify_enabled": "DOCUMENT_CLASSIFY_ENABLED", # Observability settings "metrics_enabled": "METRICS_ENABLED", "metrics_port": "METRICS_PORT", diff --git a/nextcloud_mcp_server/document_processors/classifier.py b/nextcloud_mcp_server/document_processors/classifier.py new file mode 100644 index 00000000..22e34d7b --- /dev/null +++ b/nextcloud_mcp_server/document_processors/classifier.py @@ -0,0 +1,169 @@ +"""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. + +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): + + * 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) + +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. +""" + +import logging +import re +from dataclasses import dataclass, field + +logger = logging.getLogger(__name__) + +# Page-sampling: classify at most this many pages on large docs (evenly spaced) +# so the pass stays bounded regardless of page count. +MAX_SAMPLED_PAGES = 24 + +# A page counts as "scanned-like" when a raster image covers most of it. +IMAGE_COVERAGE_SCANNED = 0.80 +# Text-quality score below which the layer is treated as junk (mashed tokens). +MIN_TEXT_QUALITY = 0.45 +# Fraction of sampled pages that must look scanned/bad for a doc->ocr verdict. +OCR_PAGE_FRACTION = 0.5 + +_WORD_RE = re.compile(r"\S+") + + +@dataclass +class PageSignals: + page_no: int + char_count: int + image_coverage: float # 0..1 of page area covered by images + text_quality: float # 0..1; low = mashed/space-less/garbage layer + needs_ocr: bool # scanned or unusable text layer + + +@dataclass +class DocClassification: + page_count: int + sampled_pages: int + total_chars: int + mean_text_quality: float + ocr_page_fraction: float # fraction of sampled pages flagged needs_ocr + recommended_tier: str # "fast" | "ocr" + flags: set[str] = field( + default_factory=set + ) # scanned | bad_text_layer | image_heavy + pages: list[PageSignals] = field(default_factory=list) + + +def _text_quality(text: str) -> float: + """Score a text layer's usability in ``[0, 1]`` (1 = clean prose). + + Penalises the two hallmarks of a junk/OCR-mangled layer: too little + whitespace (words mashed together) and very long tokens. Empty text scores + 0 -- "no usable layer". + """ + if not text: + return 0.0 + tokens = _WORD_RE.findall(text) + if not tokens: + return 0.0 + whitespace_ratio = sum(c.isspace() for c in text) / len(text) + mean_token_len = sum(len(t) for t in tokens) / len(tokens) + overlong_frac = sum(len(t) > 20 for t in tokens) / len(tokens) + # Clean prose: ~15-20% whitespace, mean token ~4-6 chars, ~no overlong tokens. + ws_score = min(whitespace_ratio / 0.12, 1.0) + len_score = ( + 1.0 if mean_token_len <= 10 else max(0.0, 1.0 - (mean_token_len - 10) / 15) + ) + overlong_score = max(0.0, 1.0 - overlong_frac * 5) + return round(ws_score * len_score * overlong_score, 3) + + +def _sample_indices(page_count: int) -> list[int]: + if page_count <= MAX_SAMPLED_PAGES: + return list(range(page_count)) + # Evenly spaced sample across the document. + step = page_count / MAX_SAMPLED_PAGES + return sorted({int(i * step) for i in range(MAX_SAMPLED_PAGES)}) + + +def classify_pdf(content: bytes) -> DocClassification: + """Classify a PDF from its bytes. + + May raise (e.g. ``pymupdf`` errors) if the bytes can't be opened as a PDF; + callers run it in a guarded context (shadow mode swallows failures) so a + bad file never breaks indexing. + """ + import pymupdf # noqa: PLC0415 -- keep the heavy import lazy / off module load + + doc = pymupdf.open("pdf", content) + try: + page_count = doc.page_count + indices = _sample_indices(page_count) + pages: list[PageSignals] = [] + for n in indices: + page = doc.load_page(n) + text = page.get_text("text") + quality = _text_quality(text) + page_area = abs(page.rect.width * page.rect.height) or 1.0 + img_area = 0.0 + for img in page.get_images(full=True): + for rect in page.get_image_rects(img[0]): + img_area += abs(rect.width * rect.height) + coverage = min(img_area / page_area, 1.0) + # A page that is mostly a raster image is a scan/photo: its content + # (handwriting, stamps, figure text) is not fully in any text layer, + # so OCR is needed to capture it -- regardless of whether a partial + # text layer is present. Text quality/char-count are kept as + # diagnostic signals (flags + tuning metrics), not the trigger, + # because OCR only helps when there is an image to read. + needs_ocr = coverage >= IMAGE_COVERAGE_SCANNED + pages.append( + PageSignals(n, len(text), round(coverage, 3), quality, needs_ocr) + ) + finally: + doc.close() + + 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 + ) + ocr_frac = (sum(p.needs_ocr for p in pages) / sampled) if sampled else 0.0 + + flags: set[str] = set() + if any(p.image_coverage >= IMAGE_COVERAGE_SCANNED for p in pages): + flags.add("image_heavy") + if ( + ocr_frac >= OCR_PAGE_FRACTION + and total_chars + and mean_quality < MIN_TEXT_QUALITY + ): + flags.add("bad_text_layer") + if ocr_frac >= OCR_PAGE_FRACTION and total_chars == 0: + flags.add("scanned") + + recommended = "ocr" if ocr_frac >= OCR_PAGE_FRACTION else "fast" + + return DocClassification( + page_count=page_count, + 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/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 7cd40e4f..2ed3c5a1 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -272,6 +272,30 @@ document_parse_failed_total = Counter( ["reason"], # reason: timeout | oom | error ) +# --- Tier-0 classifier (shadow mode) ----------------------------------------- +# +# The classifier runs a cheap pre-pass per PDF and recommends a starting tier. +# In shadow mode it changes no routing -- these metrics gather the per-tenant +# doc-mix needed to tune the thresholds before routing is enabled. + +document_classified_total = Counter( + "astrolabe_document_classified_total", + "Documents classified by tier-0, by recommended starting tier", + ["recommended_tier"], # fast | ocr +) + +document_classifier_flag_total = Counter( + "astrolabe_document_classifier_flag_total", + "Tier-0 classifier flags raised on documents", + ["flag"], # image_heavy | scanned | bad_text_layer +) + +document_text_quality = Histogram( + "astrolabe_document_text_quality", + "Tier-0 mean text-layer quality per document (0=junk, 1=clean prose)", + buckets=(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0), +) + # --- Embedding stages --------------------------------------------------------- embedding_duration_seconds = Histogram( @@ -637,6 +661,20 @@ def record_document_parse_failed(reason: str) -> None: document_parse_failed_total.labels(reason=reason).inc() +def record_document_classification( + recommended_tier: str, flags: set[str], mean_text_quality: float +) -> None: + """Record a tier-0 classification result (shadow mode -- observability only). + + Primitive args (not the DocClassification object) keep the observability + layer free of a dependency on document_processors. + """ + document_classified_total.labels(recommended_tier=recommended_tier).inc() + for flag in flags: + document_classifier_flag_total.labels(flag=flag).inc() + document_text_quality.observe(mean_text_quality) + + def record_embedding( kind: str, provider: str, diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 73a8c0c2..f53a7418 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -17,10 +17,12 @@ 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, @@ -164,6 +166,29 @@ 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: + logger.debug("Tier-0 classification failed for %s", file_path, exc_info=True) + + async def process_document( doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3 ): @@ -534,6 +559,10 @@ 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 registry = get_registry() diff --git a/tests/unit/test_doc_classifier.py b/tests/unit/test_doc_classifier.py new file mode 100644 index 00000000..24bacb0f --- /dev/null +++ b/tests/unit/test_doc_classifier.py @@ -0,0 +1,86 @@ +"""Unit tests for the tier-0 document classifier. + +Pins the routing decisions and the text-quality heuristic that drive which +extraction tier a PDF starts in: + * a clean born-digital PDF (text, no full-page images) -> ``fast`` (tier 1); + * a full-page-image scan -> ``ocr`` (tier 3), since handwriting/stamps aren't + in any text layer; + * the text-quality score distinguishes clean prose from mashed/space-less junk. +""" + +import pymupdf +import pytest + +from nextcloud_mcp_server.document_processors import classifier as clf + +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 _full_page_image_pdf(pages: int = 2) -> bytes: + # A page whose entire area is a raster image -> looks scanned. + doc = pymupdf.open() + pix = pymupdf.Pixmap(pymupdf.csRGB, pymupdf.IRect(0, 0, 600, 850)) + pix.clear_with(255) + img = pix.tobytes("png") + for _ in range(pages): + page = doc.new_page(width=595, height=842) + page.insert_image(page.rect, stream=img) + data: bytes = doc.tobytes() + doc.close() + return data + + +# --- text-quality heuristic -------------------------------------------------- + + +def test_text_quality_clean_prose_scores_high(): + assert clf._text_quality("the quick brown fox jumps over the lazy dog") > 0.8 + + +def test_text_quality_mashed_tokens_scores_low(): + # space-less / mashed layer (the "Student 147" failure mode) + mashed = "01322234567mobileoutstandingresilienceacademicachievementhurdles" + assert clf._text_quality(mashed) < clf.MIN_TEXT_QUALITY + + +def test_text_quality_empty_is_zero(): + assert clf._text_quality("") == 0.0 + + +# --- routing ----------------------------------------------------------------- + + +def test_digital_pdf_routes_fast(): + c = clf.classify_pdf(_digital_pdf()) + assert c.recommended_tier == "fast" + assert c.ocr_page_fraction == 0.0 + assert "image_heavy" not in c.flags + assert c.mean_text_quality > 0.8 + + +def test_full_page_image_routes_ocr(): + c = clf.classify_pdf(_full_page_image_pdf()) + assert c.recommended_tier == "ocr" + assert c.ocr_page_fraction == 1.0 + assert "image_heavy" in c.flags + + +# --- sampling bounds large docs ---------------------------------------------- + + +def test_large_doc_is_sampled(): + c = clf.classify_pdf(_digital_pdf(pages=120)) + assert c.page_count == 120 + assert c.sampled_pages <= clf.MAX_SAMPLED_PAGES diff --git a/tests/unit/test_shadow_classify.py b/tests/unit/test_shadow_classify.py new file mode 100644 index 00000000..c0110379 --- /dev/null +++ b/tests/unit/test_shadow_classify.py @@ -0,0 +1,62 @@ +"""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()