src/knowledge_extraction/genre.py

"""Step 5 — what KIND of document is this, decided before anything is spent.

The pipeline was calibrated on one genre (an Indonesian mining standard) and is
now handed three. The failure is not subtle: on the Komatsu shop manual the
glossary branch burns ~104 calls to abstain 93 times, because a troubleshooting
manual states symptoms and remedies, not definitions. No amount of prompt work
reaches that — it is the wrong question asked of the wrong document.

Four signals separate the three genres cleanly, and all four are already
computed for free before the paid stage:

                        heading coverage   tabular ratio   abbrev pairs   rule density
    BUMA standard              41%              25%             5             0.50
    Komatsu shop manual         1%              69%             0             0.01
    Open Pit textbook          63%               5%             0             1.10

They do not merely differ, they differ in DIFFERENT DIRECTIONS, which is what
makes a rule-based classifier honest here rather than a model call: a shop
manual is mostly tables with no headings, a textbook is mostly headed prose, a
standard sits between them and is the only one that defines its abbreviations.

**What this module does NOT do yet, and why.** Classification is on; acting on
it is not. Both levers below default to off because neither can be justified by
a measurement today, and shipping an unmeasured recall change is how the
clustering defect survived for months:

* `ROUTE_LABELS` — switching the GLiNER label set per genre. The `diagnostic`
  and `conceptual` sets produce far more mentions on their genres (2.5x on
  Komatsu), but only BUMA has a gold set, so "more mentions" cannot be shown to
  be "better recall". Unlocked by: a gold set for a second document.
* `ROUTE_BRANCHES` — skipping branches that do not fit. Measured on exactly one
  document (Komatsu: 0 definitions across two paid runs, as predicted), which
  is a prediction confirmed once, not a rule. Unlocked by: a second diagnostic
  document behaving the same way, or Rifqi accepting one document as enough.

Until then this is observability: the verdict and its warnings are recorded, an
operator can see why a document is expensive and unproductive, and turning
either lever on is a one-line change with the reasoning already written down.
"""

from __future__ import annotations

from dataclasses import dataclass, field

# Both default OFF. See the module docstring for what unlocks each.
ROUTE_LABELS = False
ROUTE_BRANCHES = False

# Thresholds sit between the measured clusters rather than on top of any one of
# them, so a fourth document of a known genre is likely to land correctly.
# Widely separated by construction — see the table above.
HEADING_COVERAGE_LOW = 0.10
HEADING_COVERAGE_HIGH = 0.50
TABULAR_RATIO_HIGH = 0.50
TABULAR_RATIO_LOW = 0.20
RULE_DENSITY_LOW = 0.10

# Below this, the document's own structure is missing and every heading-derived
# signal downstream degrades: `source_wording` (read off headings), the
# `term_in_heading` ranking weight, and `DocumentContext.outline`. Komatsu has
# exactly ONE headed chunk of 85, and nothing warned about it.
HEADING_COVERAGE_WARN = 0.10

GENRE_LABELS = {
    "standard": "broad",
    "diagnostic": "diagnostic",
    "conceptual": "conceptual",
    "unknown": "broad",
}

# Which branches earn their spend on each genre. `unknown` runs everything —
# not knowing the genre is not a reason to extract less.
GENRE_BRANCHES: dict[str, tuple[str, ...]] = {
    "standard": ("glossary", "rule", "formula", "summary"),
    # A shop manual defines nothing. What it holds is a symptom x cause matrix
    # across 59 tables, and the branch that reads one does not exist yet
    # (Step 8) — so today the honest answer is "almost nothing here fits".
    "diagnostic": ("summary",),
    "conceptual": ("glossary", "rule", "summary"),
    "unknown": ("glossary", "rule", "formula", "summary"),
}


@dataclass
class GenreSignals:
    n_chunks: int
    heading_coverage: float
    tabular_ratio: float
    abbrev_pairs: int
    rule_density: float

    def as_dict(self) -> dict:
        return {
            "n_chunks": self.n_chunks,
            "heading_coverage": round(self.heading_coverage, 3),
            "tabular_ratio": round(self.tabular_ratio, 3),
            "abbrev_pairs": self.abbrev_pairs,
            "rule_density": round(self.rule_density, 3),
        }


@dataclass
class GenreVerdict:
    genre: str
    signals: GenreSignals
    reasons: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)

    @property
    def labels_variant(self) -> str:
        """The label set this genre wants — advisory unless `ROUTE_LABELS`."""
        return GENRE_LABELS.get(self.genre, "broad")

    @property
    def branches(self) -> tuple[str, ...]:
        """The branches that fit — advisory unless `ROUTE_BRANCHES`."""
        return GENRE_BRANCHES.get(self.genre, GENRE_BRANCHES["unknown"])

    @property
    def well_served(self) -> bool:
        """Is this a document the pipeline can actually do something with?

        False is not a failure — it is the honest answer for a shop manual
        today, and it is worth saying BEFORE spending ~104 calls to find out.
        """
        return self.genre != "diagnostic"

    def as_dict(self) -> dict:
        return {
            "genre": self.genre,
            "labels_variant": self.labels_variant,
            "branches": list(self.branches),
            "well_served": self.well_served,
            "routing_applied": {"labels": ROUTE_LABELS, "branches": ROUTE_BRANCHES},
            "signals": self.signals.as_dict(),
            "reasons": self.reasons,
            "warnings": self.warnings,
        }


def measure(doc, filtered) -> GenreSignals:
    """Four free signals off the parsed artifact and the free filter stage."""
    chunks = doc.chunks or []
    n = len(chunks) or 1
    return GenreSignals(
        n_chunks=len(chunks),
        heading_coverage=sum(1 for c in chunks if (c.heading or "").strip()) / n,
        tabular_ratio=sum(1 for c in chunks if c.is_tabular) / n,
        abbrev_pairs=len(filtered.abbrev_pairs),
        rule_density=len(filtered.rule_candidates) / n,
    )


def classify(doc, filtered) -> GenreVerdict:
    """Name the genre, or say `unknown` rather than guess.

    Deliberately rule-based and deliberately conservative. A misclassification
    that routes a document to the wrong label set costs recall silently, which
    is the one failure mode this pipeline has repeatedly proven bad at noticing
    — so an unclear document gets `unknown`, which routes to exactly today's
    behaviour and spends nothing extra.
    """
    signals = measure(doc, filtered)
    reasons: list[str] = []

    if (
        signals.tabular_ratio >= TABULAR_RATIO_HIGH
        and signals.heading_coverage <= HEADING_COVERAGE_LOW
        and signals.rule_density <= RULE_DENSITY_LOW
    ):
        genre = "diagnostic"
        reasons.append(
            f"{signals.tabular_ratio:.0%} tabular with {signals.heading_coverage:.0%} "
            "headed and almost no rule language — a symptom/remedy matrix, not prose"
        )
    elif (
        signals.heading_coverage >= HEADING_COVERAGE_HIGH
        and signals.tabular_ratio <= TABULAR_RATIO_LOW
    ):
        genre = "conceptual"
        reasons.append(
            f"{signals.heading_coverage:.0%} of chunks are headed and only "
            f"{signals.tabular_ratio:.0%} tabular — structured expository prose"
        )
    elif signals.abbrev_pairs >= 3 and signals.heading_coverage > HEADING_COVERAGE_LOW:
        genre = "standard"
        reasons.append(
            f"{signals.abbrev_pairs} abbreviation definitions in headed prose — "
            "a document that names its own terms"
        )
    else:
        genre = "unknown"
        reasons.append(
            "no genre signature matched; routing is unchanged and nothing is skipped"
        )

    warnings: list[str] = []
    if signals.heading_coverage < HEADING_COVERAGE_WARN:
        warnings.append(
            f"heading coverage is {signals.heading_coverage:.0%} "
            f"({round(signals.heading_coverage * signals.n_chunks)} of "
            f"{signals.n_chunks} chunks): `source_wording`, the term_in_heading "
            "ranking weight and the document outline all degrade together, and "
            "nothing downstream can tell that from a document with no terms"
        )
    if genre == "diagnostic":
        warnings.append(
            "diagnostic documents define almost nothing — the glossary branch is "
            "expected to abstain on nearly every cluster. The content is in the "
            "tables, and the branch that reads them is not built (Step 8)"
        )

    return GenreVerdict(genre=genre, signals=signals, reasons=reasons, warnings=warnings)