Classifying Site Documents with a Rules-First Pipeline

This is a working build of the classifier described in Document Classification & Routing for Clinical Site Packets, part of Automated Document Ingestion & Validation Workflows. Where the parent guide covers the architecture and the reasoning behind a rules-first design, this walkthrough builds the concrete module: a rule set loaded from an external, version-controlled configuration file, a scorer that combines filename, PDF metadata, and keyword signals, and a threshold gate that either routes a document to its validator or drops it into a manual-review queue. Every decision this module makes is written to the append-only audit log before the document moves on.

Why naive approaches fail

The obvious first cut at this problem is a chain of if "1572" in filename.lower(): statements scattered through the ingestion code. It works for a demo and fails in production for reasons that only show up months later:

  • Rules become untestable as they multiply. A chain of string checks embedded in control flow cannot be unit-tested in isolation from the surrounding pipeline, and it cannot be reviewed as a diff by someone who does not read Python.
  • No confidence signal. A boolean “did it match” collapses a document that hit three strong signals and a document that barely matched one weak keyword into the same outcome. Both get auto-routed, and the weak match is the one that quietly lands on the wrong validator.
  • Silent acceptance of a corrupted file. If the classifier does not deliberately catch a malformed PDF and instead lets an exception propagate from deep inside a parsing call, the document either crashes the batch or, worse, gets partially classified from whatever fragment of metadata happened to be readable.
  • Hardcoded rules can’t be changed without a deploy. When the rule table lives inline in application code, adding a document type or fixing a rule that fires on the wrong file requires a code review and release cycle instead of a configuration change that regulatory affairs can review on its own schedule.

Architecture overview

The module is a short pipeline: load the rule set once at process start, extract the three signal sources for a given file, score every rule against its signal, and emit a decision plus an audit record.

Rules-first classifier build pipeline Five stages left to right: load rules from configuration, extract signals from the file, score matched rules, apply the confidence threshold, and emit the decision. Each stage feeds the next with a single arrow. Load rules Extract signals Score matches Threshold Emit decision

Setup and configuration

The only external dependency is a maintained PDF library for reading document metadata and AcroForm fields; everything else is the standard library.

pip install "pypdf>=4.0"

Configuration — where the rule file lives and how strict the threshold is — comes entirely from the environment, so the same module runs against a looser threshold in a sandbox and a stricter one in production without a code change:

"""Environment-driven configuration for the document classifier."""
from __future__ import annotations

import os
from dataclasses import dataclass


@dataclass(frozen=True)
class ClassifierConfig:
    rules_path: str
    confidence_threshold: float

    @classmethod
    def from_env(cls) -> "ClassifierConfig":
        def required(name: str) -> str:
            value = os.environ.get(name)
            if not value:
                raise RuntimeError(f"Missing required environment variable: {name}")
            return value

        return cls(
            rules_path=required("CLASSIFIER_RULES_PATH"),
            confidence_threshold=float(os.environ.get("CLASSIFIER_CONFIDENCE_THRESHOLD", "0.75")),
        )

Full working implementation

The rule set is authored as plain JSON so a regulatory-affairs reviewer can read and change it without touching Python, and the loader validates every rule’s regex and document type before the classifier ever runs against real files.

"""Load and validate the classification rule set from external configuration."""
from __future__ import annotations

import json
import re
from dataclasses import dataclass
from enum import Enum
from pathlib import Path


class DocType(str, Enum):
    FDA_1572 = "FDA_1572"
    INVESTIGATOR_CV = "INVESTIGATOR_CV"
    IRB_APPROVAL_LETTER = "IRB_APPROVAL_LETTER"
    FINANCIAL_DISCLOSURE = "FINANCIAL_DISCLOSURE"
    DELEGATION_LOG = "DELEGATION_LOG"
    UNKNOWN = "UNKNOWN"


class Signal(str, Enum):
    FILENAME = "filename"
    PDF_METADATA = "pdf_metadata"
    KEYWORD_TEXT = "keyword_text"


@dataclass(frozen=True)
class Rule:
    rule_id: str
    doc_type: DocType
    signal: Signal
    pattern: str
    weight: float


def load_rules(rules_path: str) -> list[Rule]:
    """Parse and validate the rule set; raise on any malformed entry.

    Validation happens once, at load time, so a broken rule fails the
    deployment rather than silently never matching at runtime.
    """
    raw = json.loads(Path(rules_path).read_text(encoding="utf-8"))
    rules: list[Rule] = []
    for entry in raw["rules"]:
        re.compile(entry["pattern"])  # raises re.error on a malformed pattern
        rules.append(
            Rule(
                rule_id=entry["rule_id"],
                doc_type=DocType(entry["doc_type"]),
                signal=Signal(entry["signal"]),
                pattern=entry["pattern"],
                weight=float(entry["weight"]),
            )
        )
    if not rules:
        raise ValueError(f"No rules loaded from {rules_path!r}")
    return rules

An example of the JSON a reviewer would actually edit:

{
  "rules": [
    {"rule_id": "FDA1572-FN-01", "doc_type": "FDA_1572", "signal": "filename", "pattern": "1572", "weight": 0.4},
    {"rule_id": "FDA1572-MD-01", "doc_type": "FDA_1572", "signal": "pdf_metadata", "pattern": "statement of investigator", "weight": 0.4},
    {"rule_id": "FDA1572-KW-01", "doc_type": "FDA_1572", "signal": "keyword_text", "pattern": "statement of investigator", "weight": 0.3}
  ]
}

The classifier itself extracts the three signals for one file, handling a corrupted PDF as an explicit, quarantinable outcome rather than an unhandled exception:

"""Classify one document against a loaded rule set, with explicit error handling."""
from __future__ import annotations

from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path

from pypdf import PdfReader
from pypdf.errors import PdfReadError


@dataclass(frozen=True)
class Decision:
    doc_id: str
    doc_type: DocType
    confidence: float
    matched_rule_ids: tuple[str, ...]
    outcome: str  # "routed" | "manual_review" | "quarantined"


def _pdf_signals(path: Path) -> dict[str, str]:
    """Return combined metadata text; raises PdfReadError for a corrupt file."""
    reader = PdfReader(str(path))
    info = reader.metadata or {}
    fields = reader.get_fields() or {}
    parts = [str(v) for v in info.values() if v]
    parts += [str(f.value) for f in fields.values() if f.value]
    return {"pdf_metadata": " ".join(parts)}


def classify_document(
    doc_id: str,
    path: Path,
    body_text: str,
    rules: list[Rule],
    config: ClassifierConfig,
) -> Decision:
    """Classify a single document, quarantining it if its PDF structure is unreadable."""
    try:
        pdf_signals = _pdf_signals(path)
    except PdfReadError as exc:
        return Decision(doc_id, DocType.UNKNOWN, 0.0, (), f"quarantined:{exc}")

    haystacks = {
        Signal.FILENAME: path.stem.lower(),
        Signal.PDF_METADATA: pdf_signals["pdf_metadata"],
        Signal.KEYWORD_TEXT: body_text,
    }

    votes: dict[DocType, float] = defaultdict(float)
    matched: dict[DocType, list[str]] = defaultdict(list)
    for rule in rules:
        if re.search(rule.pattern, haystacks[rule.signal], re.IGNORECASE):
            votes[rule.doc_type] += rule.weight
            matched[rule.doc_type].append(rule.rule_id)

    if not votes:
        return Decision(doc_id, DocType.UNKNOWN, 0.0, (), "manual_review")

    best = max(votes, key=lambda candidate: votes[candidate])
    confidence = min(votes[best], 1.0)
    outcome = "routed" if confidence >= config.confidence_threshold else "manual_review"
    return Decision(doc_id, best, confidence, tuple(matched[best]), outcome)

Finally, every decision is written to the audit trail with the same structured-logging discipline the rest of the platform uses, so the module never needs its own bespoke persistence layer:

"""Emit a structured audit record for a classification decision."""
from __future__ import annotations

import json
import logging

logger = logging.getLogger("ingestion.classification")


def audit_decision(decision: Decision) -> None:
    logger.info(
        json.dumps(
            {
                "action": "document_classified",
                "doc_id": decision.doc_id,
                "doc_type": decision.doc_type.value,
                "confidence": round(decision.confidence, 4),
                "matched_rule_ids": list(decision.matched_rule_ids),
                "outcome": decision.outcome,
            },
            sort_keys=True,
        )
    )

Validation and edge-case handling

A handful of edge cases account for most misclassifications in practice, and each has a specific, intentional handling path above rather than a generic catch-all:

  • A document that matches rules for two different types. classify_document picks the highest-scoring type, but if two types are within a narrow margin of each other, treat that tie as evidence the rule set itself is ambiguous for this kind of file — widen the margin check in the scorer before trusting the raw winner, and route close calls to manual review regardless of the absolute confidence number.
  • A password-protected or encrypted PDF. PdfReader raises on an encrypted file it cannot open without a password; that surfaces as the same PdfReadError branch as any other corrupt file and is quarantined rather than silently classified from the filename alone.
  • An empty or near-empty body_text. A scanned document that failed OCR entirely still has a filename and possibly PDF metadata; the keyword signal contributes nothing, but the other two signals can still produce a confident result. The scorer does not special-case this — an empty haystack simply matches no keyword rules.
  • A rule file with a duplicate rule_id. load_rules does not currently reject duplicates, so add a uniqueness check in code review or a schema-level constraint before deploying a hand-edited rule file; a duplicate ID makes the audit trail’s “which rule matched” answer ambiguous.

Testing and verification

The test suite exercises the three outcomes that matter under audit: a confident automatic route, a low-confidence fallback, and a corrupt file that must never be silently classified.

"""Tests for the rules-first document classifier."""
from __future__ import annotations

from pathlib import Path

import pytest


@pytest.fixture
def rules() -> list[Rule]:
    return [
        Rule("FDA1572-FN-01", DocType.FDA_1572, Signal.FILENAME, "1572", 0.4),
        Rule("FDA1572-KW-01", DocType.FDA_1572, Signal.KEYWORD_TEXT, "statement of investigator", 0.5),
    ]


@pytest.fixture
def config() -> ClassifierConfig:
    return ClassifierConfig(rules_path="unused", confidence_threshold=0.75)


def test_high_confidence_routes_automatically(tmp_path: Path, rules, config) -> None:
    pdf_path = tmp_path / "Form1572_signed.pdf"
    pdf_path.write_bytes(b"%PDF-1.7\n%%EOF")
    decision = classify_document(
        "doc-1", pdf_path, "Statement of Investigator", rules, config,
    )
    assert decision.doc_type is DocType.FDA_1572
    assert decision.outcome == "routed"
    assert decision.confidence >= config.confidence_threshold


def test_low_confidence_falls_back_to_manual_review(tmp_path: Path, rules, config) -> None:
    pdf_path = tmp_path / "scan0042.pdf"
    pdf_path.write_bytes(b"%PDF-1.7\n%%EOF")
    decision = classify_document("doc-2", pdf_path, "", rules, config)
    assert decision.doc_type is DocType.FDA_1572
    assert decision.outcome == "manual_review"


def test_unreadable_pdf_is_quarantined_not_guessed(tmp_path: Path, rules, config) -> None:
    corrupt_path = tmp_path / "broken.pdf"
    corrupt_path.write_bytes(b"not a real pdf")
    decision = classify_document("doc-3", corrupt_path, "irrelevant text", rules, config)
    assert decision.outcome.startswith("quarantined")
    assert decision.doc_type is DocType.UNKNOWN

The first test proves a strongly matching document clears the threshold and routes automatically; the second proves a weak filename-only match on the same document type is held back for review rather than auto-routed; the third proves a structurally broken file is quarantined with a specific reason instead of silently classified from whatever metadata happened to be readable.