admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #863 from cbcoutinho/feat/ocr-quality-trigger

Browse Source 
feat: quality + scan OCR escalation trigger (junk-text-layer scans)
This commit is contained in:
Chris Coutinho
2026-06-05 05:22:03 +02:00
committed by GitHub
parent d421bf6953 36209e3160
commit 55f5473752
7 changed files with 332 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/config.py
+25
View File
@@ -149,6 +149,15 @@ _DEFAULTS: dict[str, Any] = {
# Provider-namespaced OCR model id (gateway routes on the prefix; the direct
# mistral backend strips it).
"document_ocr_model": "mistral/mistral-ocr-latest",
# OCR escalation triggers (tier-0). A page is OCR-worthy when its text is
# near-empty (< min_page_chars) OR low-quality (< min_text_quality) OR (when
# detect_scanned) mostly a raster image; a doc escalates when the OCR-worthy
# page fraction reaches page_fraction. Calibrate min_text_quality from the
# astrolabe_document_text_quality histogram per tenant.
"document_ocr_min_text_quality": 0.5,
"document_ocr_page_fraction": 0.5,
"document_ocr_min_page_chars": 16,
"document_ocr_detect_scanned": True,
# Observability
"metrics_enabled": True,
"metrics_port": 9090,
@@ -297,6 +306,10 @@ _dynaconf = Dynaconf(
# >=1: pymupdf4llm treats graphics_limit=0 as "no cap", which would
# re-expose the OOM this guards against.
Validator("DOCUMENT_PDF_GRAPHICS_LIMIT", gte=1),
# OCR escalation thresholds: quality + page-fraction are [0, 1].
Validator("DOCUMENT_OCR_MIN_TEXT_QUALITY", gte=0, lte=1),
Validator("DOCUMENT_OCR_PAGE_FRACTION", gte=0, lte=1),
Validator("DOCUMENT_OCR_MIN_PAGE_CHARS", gte=0),
# Non-negative
Validator("DOCUMENT_CHUNK_OVERLAP", gte=0),
# Non-empty strings
@@ -757,6 +770,14 @@ class Settings:
# gateway routes on the "<provider>/" prefix; the direct mistral backend
# strips it.
document_ocr_model: str = "mistral/mistral-ocr-latest"
# OCR escalation triggers (tier-0), per-tenant tunable. A page is OCR-worthy
# if near-empty (< min_page_chars) OR low text-quality (< min_text_quality)
# OR (when detect_scanned, image-analysis only runs when OCR is enabled)
# mostly a raster image; the doc escalates at >= page_fraction such pages.
document_ocr_min_text_quality: float = 0.5
document_ocr_page_fraction: float = 0.5
document_ocr_min_page_chars: int = 16
document_ocr_detect_scanned: bool = True
# Observability settings
metrics_enabled: bool = True
@@ -1376,6 +1397,10 @@ def get_settings() -> Settings:
"document_ocr_enabled": "DOCUMENT_OCR_ENABLED",
"document_ocr_provider": "DOCUMENT_OCR_PROVIDER",
"document_ocr_model": "DOCUMENT_OCR_MODEL",
"document_ocr_min_text_quality": "DOCUMENT_OCR_MIN_TEXT_QUALITY",
"document_ocr_page_fraction": "DOCUMENT_OCR_PAGE_FRACTION",
"document_ocr_min_page_chars": "DOCUMENT_OCR_MIN_PAGE_CHARS",
"document_ocr_detect_scanned": "DOCUMENT_OCR_DETECT_SCANNED",
# Observability settings
"metrics_enabled": "METRICS_ENABLED",
"metrics_port": "METRICS_PORT",
nextcloud_mcp_server/document_processors/classifier.py
+132 -44
View File
@@ -9,11 +9,14 @@ Decides which extraction tier a PDF should escalate to, from cheap signals:
* no text layer -- the strongest OCR signal available from text alone.
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_from_text(text, page_boundaries, ...)`` -- the HOT PATH. Routes on
text-quality + near-empty pages derived from the tier-1 extraction (~no
cost). When OCR + scan-detection are enabled the registry also passes
per-page ``image_coverage`` (from ``image_coverage_per_page``) so scans are
caught too; that image pass is the only added cost and only OCR-opted-in
tenants pay it. Thresholds come from per-tenant settings.
* ``classify_pdf(content)`` -- a standalone/diagnostic pass that re-opens the
PDF and adds image-coverage analysis. More expensive; used off the hot path.
PDF and does image-coverage analysis inline. Off the hot path.
Recommended tier:
* ``ocr`` -- scanned / no-usable-text-layer (route to tier 3, when enabled)
@@ -36,7 +39,10 @@ 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
# Kept in sync with the DOCUMENT_OCR_MIN_TEXT_QUALITY setting default so the
# module/diagnostic default matches production (the registry always passes the
# setting).
MIN_TEXT_QUALITY = 0.5
# 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.
@@ -83,6 +89,13 @@ def _text_quality(text: str) -> float:
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)
# Word-merging (dropped inter-word spaces) is the dominant junk-text-layer
# failure mode on scanned forms -- the older whitespace/overlong(>20) terms
# miss it, because the merges are 10-20 chars and a few dropped spaces still
# leave whitespace above the 0.12 cap. Clean prose keeps <~3% of tokens above
# 12 chars; merged/OCR-mangled layers push it past 10%. (Measured: the junk
# Student-147 scan scores ~0.20 here vs >=0.9 for clean digital docs.)
long_frac = sum(len(t) > 12 for t in tokens) / len(tokens)
# Caps at 1.0 from 12% whitespace (conservative; clean prose runs 15-20%),
# mean token ~4-6 chars, ~no overlong tokens.
ws_score = min(whitespace_ratio / 0.12, 1.0)
@@ -90,7 +103,8 @@ def _text_quality(text: str) -> float:
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)
merge_score = max(0.0, 1.0 - max(0.0, long_frac - 0.03) / 0.12)
return round(ws_score * len_score * overlong_score * merge_score, 3)
def _sample_indices(page_count: int) -> list[int]:
@@ -105,6 +119,21 @@ def _sample_indices(page_count: int) -> list[int]:
)
def _page_image_coverage(page: Any) -> float:
"""Fraction of a pymupdf page covered by raster images, in ``[0, 1]``.
Approximate: an image placed multiple times (tiled backgrounds) is
double-counted, so the raw area can exceed the page -- the min() caps
coverage at 1.0, which is all the scanned/digital split needs.
"""
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)
return min(img_area / page_area, 1.0)
def classify_pdf(content: bytes) -> DocClassification:
"""Classify a PDF from its bytes.
@@ -122,22 +151,16 @@ def classify_pdf(content: bytes) -> DocClassification:
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)
# Approximate: an image placed multiple times (tiled backgrounds) is
# double-counted, so img_area can exceed page_area -- the min() caps
# coverage at 1.0, which is all the scanned/digital split needs.
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
coverage = _page_image_coverage(page)
# OCR-worthy on the same three signals as classify_from_text (kept in
# sync so an operator reproducing routing offline gets the pipeline's
# answer): a mostly-raster scan, a junk/low-quality text layer (the
# word-merging case), or an effectively empty text layer.
needs_ocr = (
coverage >= IMAGE_COVERAGE_SCANNED
or quality < MIN_TEXT_QUALITY
or len(text.strip()) < MIN_PAGE_CHARS
)
pages.append(
PageSignals(n, len(text), round(coverage, 3), quality, needs_ocr)
)
@@ -180,27 +203,88 @@ def classify_pdf(content: bytes) -> DocClassification:
)
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.
def image_coverage_per_page(content: bytes) -> list[float]:
"""Raster-image coverage in ``[0, 1]`` for every page (document order).
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.
Lets the hot path flag scanned pages whose embedded text layer is junk but
statistically clean-looking. Re-opens the PDF, so the registry calls it only
when OCR + scan detection are enabled (the cost is borne by OCR-opted-in
tenants). Returned list is aligned by index with the leading page boundaries.
Bounded to the first ``MAX_SAMPLED_PAGES`` pages -- the image pass is the
costly part, so a 200-page scan isn't fully rasterised on the hot path. Pages
beyond the cap fall back to the text-quality signal in ``classify_from_text``
(a scanned tail has junk text too), and ``page_fraction`` still gates over
every page.
"""
import pymupdf # noqa: PLC0415 -- keep the heavy import lazy
cov: list[float] = []
with pymupdf.open("pdf", content) as doc:
for n in range(min(doc.page_count, MAX_SAMPLED_PAGES)):
cov.append(_page_image_coverage(doc.load_page(n)))
return cov
def classify_from_text(
full_text: str,
page_boundaries: list[dict[str, Any]],
*,
min_text_quality: float = MIN_TEXT_QUALITY,
min_page_chars: int = MIN_PAGE_CHARS,
page_fraction: float = OCR_PAGE_FRACTION,
image_coverage: list[float] | None = None,
) -> DocClassification:
"""Classify from text already extracted by tier-1 -- no PDF re-open by default.
The hot-path classifier. A page is OCR-worthy when its text is near-empty
(``< min_page_chars``), its text-quality is junk (``< min_text_quality`` --
the word-merging signal), OR (when ``image_coverage`` is supplied, i.e. OCR +
scan-detection are on) the page is mostly a raster image. The doc recommends
``ocr`` once ``ocr_frac >= page_fraction``. Thresholds are passed in by the
registry from per-tenant settings.
``page_boundaries`` are ``{page, start_offset, end_offset}`` indexing into
``full_text`` (the tier-1/pdf_highlighter contract).
``full_text``; ``image_coverage[i]`` (if given) aligns with the i-th boundary.
Note: the ``image_heavy`` flag (and the image-coverage trigger) are only set
when ``image_coverage`` is supplied, so for tenants with scan detection off
that flag is always zero -- the text-quality/empty signals still route.
"""
# image_coverage is expected to be one entry per page, capped at
# MAX_SAMPLED_PAGES (see image_coverage_per_page). Any other length means the
# 1:1 page alignment drifted (e.g. the extractor reordered/skipped pages) --
# log it so a contract break surfaces rather than silently misattributing
# coverage to the wrong pages.
if image_coverage is not None:
expected = min(len(page_boundaries), MAX_SAMPLED_PAGES)
if len(image_coverage) != expected:
logger.debug(
"image_coverage length %s != expected %s for %s boundaries; "
"scan signal may be misaligned",
len(image_coverage),
expected,
len(page_boundaries),
)
pages: list[PageSignals] = []
for b in page_boundaries:
for idx, b in enumerate(page_boundaries):
seg = full_text[b["start_offset"] : b["end_offset"]]
needs_ocr = len(seg.strip()) < MIN_PAGE_CHARS
quality = _text_quality(seg)
# image_coverage is one entry per PDF page, aligned 1:1 with the
# boundaries; the length guard is belt-and-suspenders against a mismatch.
cov = (
image_coverage[idx]
if image_coverage is not None and idx < len(image_coverage)
else 0.0
)
needs_ocr = (
len(seg.strip()) < min_page_chars
or quality < min_text_quality
or cov >= IMAGE_COVERAGE_SCANNED
)
pages.append(
PageSignals(b["page"], len(seg), 0.0, _text_quality(seg), needs_ocr)
PageSignals(b["page"], len(seg), round(cov, 3), quality, needs_ocr)
)
sampled = len(pages)
@@ -213,18 +297,22 @@ def classify_from_text(
# 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 gated on ocr_frac >= 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 metric diverges from classify_pdf.
flags: set[str] = set()
if sampled and ocr_frac >= OCR_PAGE_FRACTION:
if sampled and ocr_frac >= page_fraction:
if total_chars == 0:
flags.add("no_text_layer")
elif mean_quality < MIN_TEXT_QUALITY:
# "scanned" (not "no_text_layer"): same name + meaning as classify_pdf
# so astrolabe_document_classifier_flag_total isn't split across two
# labels for the empty-text-layer case.
flags.add("scanned")
elif mean_quality < min_text_quality:
flags.add("bad_text_layer")
if any(p.image_coverage >= IMAGE_COVERAGE_SCANNED for p in pages):
flags.add("image_heavy")
recommended = "ocr" if ocr_frac >= OCR_PAGE_FRACTION else "fast"
recommended = "ocr" if ocr_frac >= page_fraction else "fast"
return DocClassification(
page_count=len(page_boundaries),
nextcloud_mcp_server/document_processors/registry.py
+28 -3
View File
@@ -14,7 +14,7 @@ from nextcloud_mcp_server.observability.metrics import (
from nextcloud_mcp_server.observability.tracing import trace_operation
from .base import DocumentProcessor, ProcessingResult, ProcessorError
from .classifier import classify_from_text
from .classifier import classify_from_text, image_coverage_per_page
logger = logging.getLogger(__name__)
@@ -233,17 +233,42 @@ class ProcessorRegistry:
fast, content, content_type, filename, options, progress_callback
)
# Tier-0 classification from the extraction (cheap: no PDF re-open).
# Tier-0 classification from the extraction (cheap: text-only, no PDF
# re-open). Scan detection (image analysis, re-opens the PDF) runs only
# when OCR + detect_scanned are enabled, so its cost is paid by
# OCR-opted-in tenants only.
classification = None
if settings.document_classify_enabled and result.success:
try:
image_coverage = None
if (
settings.document_ocr_enabled
and settings.document_ocr_detect_scanned
):
try:
image_coverage = image_coverage_per_page(content)
except Exception:
# Best-effort: fall back to text-only signals. WARNING
# (not DEBUG) so a systematic scan-detection failure on an
# OCR-enabled tenant is visible at LOG_LEVEL=INFO.
logger.warning(
"Scan detection failed for %s; using text-only signals",
filename or "<bytes>",
exc_info=True,
)
classification = classify_from_text(
result.text, result.metadata.get("page_boundaries") or []
result.text,
result.metadata.get("page_boundaries") or [],
min_text_quality=settings.document_ocr_min_text_quality,
min_page_chars=settings.document_ocr_min_page_chars,
page_fraction=settings.document_ocr_page_fraction,
image_coverage=image_coverage,
)
record_document_classification(
classification.recommended_tier,
classification.flags,
classification.mean_text_quality,
classification.ocr_page_fraction,
)
except Exception:
logger.warning(
nextcloud_mcp_server/observability/metrics.py
+21 -3
View File
@@ -299,6 +299,18 @@ document_text_quality = Histogram(
buckets=(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0),
)
# Per-document fraction of OCR-worthy pages (near-empty / junk-quality / scanned).
# This is the value the DOCUMENT_OCR_PAGE_FRACTION threshold acts on, so its
# distribution per tenant is the lever for tuning OCR escalation (quality vs
# cost): how many docs sit just below/above the cutoff. Pair with
# document_text_quality (where to set the per-page quality floor) and
# document_escalation_total (realized OCR volume).
document_ocr_page_fraction = Histogram(
"astrolabe_document_ocr_page_fraction",
"Tier-0 fraction of OCR-worthy pages per document (0=all-clean, 1=all-bad)",
buckets=(0.0, 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(
@@ -665,17 +677,23 @@ def record_document_parse_failed(reason: str) -> None:
def record_document_classification(
recommended_tier: str, flags: set[str], mean_text_quality: float
recommended_tier: str,
flags: set[str],
mean_text_quality: float,
ocr_page_fraction: float = 0.0,
) -> None:
"""Record a tier-0 classification result (shadow mode -- observability only).
"""Record a tier-0 classification result.
Primitive args (not the DocClassification object) keep the observability
layer free of a dependency on document_processors.
layer free of a dependency on document_processors. ``mean_text_quality`` and
``ocr_page_fraction`` feed the two histograms operators use to tune the OCR
escalation thresholds per tenant (quality vs cost).
"""
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)
document_ocr_page_fraction.observe(ocr_page_fraction)
def record_embedding(
tests/unit/test_config.py
+24
View File
@@ -381,6 +381,30 @@ class TestDynaconfValidators:
with pytest.raises(ValidationError, match="LOG_FORMAT"):
_reload_config()
@patch.dict(os.environ, {"DOCUMENT_OCR_MIN_TEXT_QUALITY": "1.5"}, clear=True)
def test_ocr_min_text_quality_out_of_range(self):
"""DOCUMENT_OCR_MIN_TEXT_QUALITY must be in [0, 1]."""
from dynaconf import ValidationError
with pytest.raises(ValidationError, match="DOCUMENT_OCR_MIN_TEXT_QUALITY"):
_reload_config()
@patch.dict(os.environ, {"DOCUMENT_OCR_PAGE_FRACTION": "2"}, clear=True)
def test_ocr_page_fraction_out_of_range(self):
"""DOCUMENT_OCR_PAGE_FRACTION must be in [0, 1]."""
from dynaconf import ValidationError
with pytest.raises(ValidationError, match="DOCUMENT_OCR_PAGE_FRACTION"):
_reload_config()
@patch.dict(os.environ, {"DOCUMENT_OCR_MIN_PAGE_CHARS": "-1"}, clear=True)
def test_ocr_min_page_chars_negative(self):
"""DOCUMENT_OCR_MIN_PAGE_CHARS must be non-negative."""
from dynaconf import ValidationError
with pytest.raises(ValidationError, match="DOCUMENT_OCR_MIN_PAGE_CHARS"):
_reload_config()
@patch.dict(os.environ, {"LOG_LEVEL": "VERBOSE"}, clear=True)
def test_invalid_log_level(self):
"""Test invalid LOG_LEVEL raises ValidationError."""
tests/unit/test_doc_classifier.py
+83 -2
View File
@@ -174,7 +174,7 @@ def test_classify_from_text_clean_routes_fast():
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 "scanned" in c.flags # unified with classify_pdf's flag name
assert c.total_chars == 0
@@ -202,4 +202,85 @@ def test_classify_from_text_junk_layer_flags_bad_text_layer():
assert c.recommended_tier == "ocr"
assert c.total_chars > 0
assert "bad_text_layer" in c.flags
assert "no_text_layer" not in c.flags
assert "scanned" not in c.flags # has text, just junk -> not the empty case
# --- quality + scan escalation triggers (Deck #207) --------------------------
_JUNK = (
"ST. TRINIAN'SSCHOOLSTUDENT RECORDFILE struggledsignificantlywith "
"learningdifficulties demonstrateda positiveattitude academictasks"
)
_CLEAN = "the quick brown fox jumps over the lazy dog and then runs away home"
def _two_page(text_a: str, text_b: str):
na = len(text_a)
return text_a + text_b, [
{"page": 1, "start_offset": 0, "end_offset": na},
{"page": 2, "start_offset": na, "end_offset": na + len(text_b)},
]
def test_classify_from_text_low_quality_routes_ocr():
full, bounds = _two_page(_JUNK, _JUNK)
c = clf.classify_from_text(full, bounds)
assert c.recommended_tier == "ocr"
assert "bad_text_layer" in c.flags
def test_quality_floor_override_disables_trigger():
# min_text_quality=0.0 => quality never trips; text present + not scanned => fast
full, bounds = _two_page(_JUNK, _JUNK)
c = clf.classify_from_text(full, bounds, min_text_quality=0.0)
assert c.recommended_tier == "fast"
def test_scan_signal_routes_ocr_even_with_clean_text():
# clean text but every page is a raster scan -> OCR (the Student-147 case)
full, bounds = _two_page(_CLEAN, _CLEAN)
c = clf.classify_from_text(full, bounds, image_coverage=[1.0, 1.0])
assert c.recommended_tier == "ocr"
assert "image_heavy" in c.flags
def test_scan_signal_ignored_when_coverage_low():
full, bounds = _two_page(_CLEAN, _CLEAN)
c = clf.classify_from_text(full, bounds, image_coverage=[0.1, 0.0])
assert c.recommended_tier == "fast"
def test_page_fraction_override():
# exactly one of two pages is junk -> ocr_frac 0.5
full, bounds = _two_page(_CLEAN, _JUNK)
assert (
clf.classify_from_text(full, bounds, page_fraction=0.5).recommended_tier
== "ocr"
)
assert (
clf.classify_from_text(full, bounds, page_fraction=0.6).recommended_tier
== "fast"
)
def test_image_coverage_per_page():
scan = clf.image_coverage_per_page(_full_page_image_pdf(pages=2))
assert len(scan) == 2 and all(c >= 0.8 for c in scan)
digital = clf.image_coverage_per_page(_digital_pdf(pages=2))
assert len(digital) == 2 and all(c < 0.1 for c in digital)
def test_scan_coverage_shorter_than_pages_falls_back_to_text():
# image_coverage shorter than the boundaries (the MAX_SAMPLED_PAGES cap):
# page 0 is flagged scanned; later pages fall back to the text-quality signal.
n = len(_CLEAN)
full = _CLEAN * 3
bounds = [
{"page": 1, "start_offset": 0, "end_offset": n},
{"page": 2, "start_offset": n, "end_offset": 2 * n},
{"page": 3, "start_offset": 2 * n, "end_offset": 3 * n},
]
c = clf.classify_from_text(full, bounds, image_coverage=[1.0])
assert c.pages[0].needs_ocr is True # scanned (coverage)
assert c.pages[1].needs_ocr is False # clean text, no coverage entry
assert c.recommended_tier == "fast" # only 1/3 pages bad
tests/unit/test_registry_tiering.py
+19 -1
View File
@@ -65,10 +65,23 @@ class _Fake(DocumentProcessor):
class _Settings:
def __init__(self, engine="pypdfium2", classify=True, ocr=False):
def __init__(
self,
engine="pypdfium2",
classify=True,
ocr=False,
min_text_quality=0.5,
page_fraction=0.5,
min_page_chars=16,
detect_scanned=False,
):
self.document_tier1_engine = engine
self.document_classify_enabled = classify
self.document_ocr_enabled = ocr
self.document_ocr_min_text_quality = min_text_quality
self.document_ocr_page_fraction = page_fraction
self.document_ocr_min_page_chars = min_page_chars
self.document_ocr_detect_scanned = detect_scanned
def _registry(*procs: tuple[DocumentProcessor, int]) -> ProcessorRegistry:
@@ -113,6 +126,11 @@ async def test_records_classification(monkeypatch):
r = _registry((_Fake("fast", "fast"), 20))
await r.process(b"%PDF-1.7", "application/pdf")
rec.assert_called_once()
# recommended_tier, flags, mean_text_quality, ocr_page_fraction all threaded
# through (the last two feed the per-tenant tuning histograms).
args = rec.call_args.args
assert len(args) == 4
assert isinstance(args[0], str) and isinstance(args[3], float)
async def test_classify_disabled_skips_recording(monkeypatch):