"""`okf quality` -- a per-file-type verdict on one bundle, with the denominator.

**This is not `okf check`, and the separation is the point.** `okf check` reads
a consumption skill and one payload against `docs/consumption-contract.md`: it
answers whether a payload carries what a claim must rest on. Measured
2026-09-10 by `vegnormal-okf` on three arms over one corpus, it returned 0
findings and exit 0 on all three while their hit@k ranged from 6 of 6 to 0 of 6
-- a green contract check says nothing about whether the cut found anything
worth reading. This module asks that second question, and it is a SEPARATE
command rather than a `--quality` flag on the first for exactly that reason: the
two answer different questions and a caller must not be able to read one as the
other.

Three verdicts and no fourth: `PASS`, `FAIL`, `UNMEASURED`. A type with no
measured threshold is never `PASS` -- an unmeasured row that reads as a passing
one is the failure this gate exists to prevent, and it is the same failure
`extract._EVIDENCE` was built to prevent one layer down.

**What this gate can and cannot see.** Every metric here is computed from the
bundle alone: no fasit, no model call, no clock, no network. That bounds it
sharply, and the bound is measured rather than assumed.
`docs/2026-09-12-g37-terskler.md` SS 4 records three candidates measured over
the same four bundles and what became of each: duplicate titles WITHIN a
document (0 of 3 206 on the known-bad arm against 349 of 2 761 on the known-good
one -- the wrong direction) and the share of very short concepts (5.6 % against
14.6 % -- also the wrong direction) are not shipped; duplicate titles across the
WHOLE bundle order the four bundles correctly (37.8 / 16.3 / 12.6 / 5.7 %) and
are still not shipped, because a bar separating them would have to be placed
between the two bundles that define it, which is fitting the bar to the number.
The defect that started this work -- 1 148 of 2 761 declared boundaries
recovered -- needs a fasit and no bundle-only metric reaches it.
"""

from __future__ import annotations

import argparse
import re
import sys
from collections import Counter
from dataclasses import dataclass
from pathlib import Path

from .consume import ConsumeError, enumerate_concepts, read_concept, root_bundle_id_of
from .corpus import LOG_NAME
from .profiles import SEGMENTED_OKF_V0_2, BundleProfile

CLI_ID = "okf quality"

#: The row a concept lands in when it declares no `source_file`. Not a file
#: type and never treated as one: measured 2026-09-12, three of the four
#: evidence corpora (`n100-2023`, `n200-2024`, `n500-2024`) carry the key on 0
#: of 446, 0 of 1 133 and 0 of 270 concepts, because their producer is not this
#: library's Door B. A per-file-type gate has nothing to say about them, and
#: says that.
NO_SOURCE_FILE = "(no source_file)"

#: A threshold needs a denominator big enough that a single document cannot be
#: the rate. FIVE, and the number is this repository's own honesty limit rather
#: than a statistical claim: `docs/2026-09-08-k3-runde2-per-filtype.md` states
#: "Per file type the denominators are 8, 3 and 1. A `1/1` is not a rate", and
#: `docs/2026-09-04-k3-arm-c.md` says of the three office types with no corpus
#: file at all: "Unmeasured, not passing." Below this floor the row is
#: `UNMEASURED` and its numbers are still printed.
#:
#: It binds BOTH denominators -- the threshold's and the bundle's. Found by
#: running the gate rather than by reading it: one PDF cut into 2 182 concepts
#: scored 0 of 1 against the 32-document reference and read as PASS.
MIN_DOCUMENTS_FOR_A_THRESHOLD = 5


@dataclass(frozen=True)
class Threshold:
    """One measured bar, carrying the measurement it was read off.

    The bar is held as the measured PAIR (`limit_null` of `limit_documents`)
    rather than a float, so the comparison is exact integer arithmetic and a
    bundle sitting exactly at the reference cannot fall to a rounding step.
    """

    metric: str
    limit_null: int
    limit_documents: int
    #: Documents behind the measurement. Equal to `limit_documents` today and
    #: kept separate because a threshold ratified over a wider corpus than the
    #: one it is expressed as would need both numbers.
    documents: int
    source: str

    def exceeded_by(self, null: int, documents: int) -> bool:
        """`null/documents` strictly worse than the reference, without floats."""
        return null * self.limit_documents > self.limit_null * documents

    def as_share(self) -> str:
        return f"{self.limit_null}/{self.limit_documents}"


#: The bars, per extension, and there are two of them. Read off the pinned
#: reference bundle `K2-bundle-default-20260912` (the 43-document corpus
#: `~/corpora/okf-telling-20260829/K2/trinn1`, N = 43, 39 merged) on
#: 2026-09-12, and set at the value measured there rather than at a rounder
#: number nearby: this is a REGRESSION bar against a pinned artifact, not a
#: claim that a bundle at the bar is good. `docs/2026-09-12-g37-terskler.md`
#: carries the table, the corpora and what each number does not prove.
#:
#: Every other type is absent on purpose. `.xlsx` (2 documents) and `.xml`
#: (1 document) are below the floor above; `.html` has no bundle measured in
#: this repository; `.md`, `.txt`, `.csv`, `.json`, `.htm`, `.pptx`, `.odt`
#: and `.rtf` have no corpus class in `extract._EVIDENCE` at all.
THRESHOLDS: dict[str, Threshold] = {
    ".pdf": Threshold(
        metric="structure_null_share",
        limit_null=8,
        limit_documents=32,
        documents=32,
        source="K2-bundle-default-20260912 (43-document corpus, 32 pdf documents)",
    ),
    ".docx": Threshold(
        metric="structure_null_share",
        limit_null=2,
        limit_documents=5,
        documents=5,
        source="K2-bundle-default-20260912 (43-document corpus, 5 docx documents)",
    ),
}

#: The one bar that needs no corpus: a concept whose body holds no
#: non-whitespace character. Taken from the harness's own definition of a
#: degenerate merge (`corpus.CorpusReport.render`: "a merge is degenerate when
#: the extracted text is zero characters after stripping whitespace -- a
#: definition, not a threshold"), so it applies to every type INCLUDING one with
#: no threshold. FAIL is reachable for every row; PASS is not.
EMPTY_BODY_LIMIT = 0

PASS = "PASS"
FAIL = "FAIL"
UNMEASURED = "UNMEASURED"

_LOG_LINE = re.compile(
    r"N = (\d+).*?merged = (\d+).*?coded rejections = (\d+)",
    re.DOTALL,
)


@dataclass(frozen=True)
class TypeReport:
    """One file type's numbers and its verdict. Every count carries its own N."""

    extension: str
    documents: int
    concepts: int
    empty: int
    structure_null: int
    verdict: str
    threshold: Threshold | None
    reason: str

    def render(self) -> str:
        """One line, and every count on it carries its own denominator.

        The `NO_SOURCE_FILE` row prints neither a document count nor a
        one-concept share: those concepts all share the same empty
        `source_file`, so grouping by it yields `documents 1` for a bundle of
        446 -- a number that looks measured and means nothing.
        """
        head = f"{self.extension:<18} {self.verdict:<11} "
        if self.extension == NO_SOURCE_FILE:
            return (
                f"{head}concepts {self.concepts:>5}  empty {self.empty}/{self.concepts}  "
                f"-- {self.reason}"
            )
        share = f"{self.structure_null}/{self.documents}"
        bar = f"limit {self.threshold.as_share()}" if self.threshold else "no threshold"
        return (
            f"{head}documents {self.documents:>5}  "
            f"concepts {self.concepts:>5}  empty {self.empty}/{self.concepts}  "
            f"one-concept documents {share} ({bar})  -- {self.reason}"
        )


@dataclass(frozen=True)
class BundleQuality:
    """One bundle's rows, its run log if it has one, and the exit code they imply."""

    bundle_root: Path
    bundle_id: str
    rows: tuple[TypeReport, ...]
    #: The `N`, merged and coded-rejection counts from the bundle's own section
    #: 9 log, or `None` when the bundle carries no log. Never defaulted to zero:
    #: a rejected document leaves NO concept in the bundle, so without the log
    #: the gate cannot know whether a type failed to extract entirely.
    run_log: str | None

    def row(self, extension: str) -> TypeReport:
        for row in self.rows:
            if row.extension == extension:
                return row
        raise KeyError(
            f"{extension} is not a row of this bundle: {[r.extension for r in self.rows]}"
        )

    @property
    def exit_code(self) -> int:
        """0 judged and clean, 1 at least one FAIL, 3 nothing could be judged.

        `2` is reserved for "did not run" and is returned by `main` alone. The
        third code exists because exit 0 over a table of `UNMEASURED` rows would
        be exactly the silent pass this gate was built to stop.
        """
        if any(row.verdict == FAIL for row in self.rows):
            return 1
        if any(row.verdict == PASS for row in self.rows):
            return 0
        return 3

    def render(self) -> str:
        lines = [
            f"# {CLI_ID}: {self.bundle_id}",
            "",
            f"bundle: {self.bundle_root}",
            (
                f"run log: {self.run_log}"
                if self.run_log is not None
                else f"run log: no run log in the bundle ({LOG_NAME} absent) -- the "
                "denominators below are the bundle's own, and a document rejected "
                "at extraction leaves no row here at all"
            ),
            "",
            "## Per file type",
            "",
        ]
        lines.extend(row.render() for row in self.rows)
        lines.extend(
            [
                "",
                "## What this verdict is not",
                "",
                "A regression bar against a pinned reference bundle, per file type.",
                "PASS means no worse than that reference on the metrics below; it is",
                "not a claim that the cut found the document's own structure. Boundary",
                "recall and hit@k need a fasit and are outside a bundle-only gate --",
                "docs/2026-09-12-g37-terskler.md carries the measurement that says so.",
            ]
        )
        return "\n".join(lines) + "\n"


def _extension_of(source_file: str) -> str:
    if not source_file.strip():
        return NO_SOURCE_FILE
    suffix = Path(source_file).suffix.lower()
    return suffix if suffix else NO_SOURCE_FILE


def read_run_log(bundle_root: Path) -> str | None:
    """The bundle's own `N`, merged and coded-rejection counts, or `None`."""
    log = bundle_root / LOG_NAME
    if not log.is_file():
        return None
    matches = _LOG_LINE.findall(log.read_text(encoding="utf-8"))
    if not matches:
        return None
    total, merged, rejected = matches[-1]
    return f"N = {total}, merged = {merged}, coded rejections = {rejected}"


def measure_bundle(
    bundle_root: Path, *, profile: BundleProfile = SEGMENTED_OKF_V0_2
) -> BundleQuality:
    """Every concept the index declares, grouped by the extension it came from.

    Reached through the index tree and never `rglob`: the index is the bundle's
    own statement of what it contains, and `consume.enumerate_concepts` is the
    one walker in this library that reads it. Controlled 2026-09-12 against the
    directory listing on four bundles -- 453, 2 761, 3 206 and 446 concepts
    either way.
    """
    root_bundle_id = root_bundle_id_of(bundle_root, profile=profile)
    concepts_per_extension: Counter[str] = Counter()
    empty_per_extension: Counter[str] = Counter()
    documents: dict[str, Counter[str]] = {}
    for concept_id in enumerate_concepts(bundle_root, profile=profile):
        concept = read_concept(
            bundle_root / f"{concept_id}{profile.paths.concept_suffix}",
            bundle_root=bundle_root,
            root_bundle_id=root_bundle_id,
        )
        extension = _extension_of(concept.source_file)
        concepts_per_extension[extension] += 1
        documents.setdefault(extension, Counter())[concept.source_file] += 1
        if not "".join(concept.body.split()):
            empty_per_extension[extension] += 1
    rows = tuple(
        _verdict(
            extension,
            documents=documents[extension],
            concepts=concepts_per_extension[extension],
            empty=empty_per_extension[extension],
        )
        for extension in sorted(concepts_per_extension)
    )
    return BundleQuality(
        bundle_root=bundle_root,
        bundle_id=root_bundle_id,
        rows=rows,
        run_log=read_run_log(bundle_root),
    )


def _verdict(extension: str, *, documents: Counter[str], concepts: int, empty: int) -> TypeReport:
    document_count = len(documents)
    structure_null = sum(1 for count in documents.values() if count == 1)
    threshold = THRESHOLDS.get(extension)
    if extension == NO_SOURCE_FILE and empty <= EMPTY_BODY_LIMIT:
        return TypeReport(
            extension=extension,
            documents=0,
            concepts=concepts,
            empty=empty,
            structure_null=0,
            verdict=UNMEASURED,
            threshold=None,
            reason=(
                f"no source_file on {concepts} of {concepts} concepts, so this "
                "bundle names no file type at all -- the shape three of the four "
                "evidence corpora arrive in, and nothing per file type can be said"
            ),
        )
    if empty > EMPTY_BODY_LIMIT:
        verdict, reason = (
            FAIL,
            (
                f"{empty} of {concepts} concepts carry no non-whitespace body; the "
                "harness calls a zero-character merge degenerate by definition"
            ),
        )
    elif threshold is None:
        verdict, reason = (
            UNMEASURED,
            (
                "no measured threshold for this type; see "
                "docs/2026-09-12-g37-terskler.md, and never read this row as PASS"
            ),
        )
    elif document_count < MIN_DOCUMENTS_FOR_A_THRESHOLD:
        verdict, reason = (
            UNMEASURED,
            (
                f"{document_count} document(s) of this type in the bundle, below the "
                f"floor of {MIN_DOCUMENTS_FOR_A_THRESHOLD}: a share over that few "
                "documents is not a rate, whatever the threshold says"
            ),
        )
    elif threshold.exceeded_by(structure_null, document_count):
        verdict, reason = (
            FAIL,
            (
                f"{structure_null} of {document_count} documents yielded one concept, "
                f"worse than the reference {threshold.as_share()} ({threshold.source})"
            ),
        )
    else:
        verdict, reason = (
            PASS,
            (f"no worse than the reference {threshold.as_share()} ({threshold.source})"),
        )
    return TypeReport(
        extension=extension,
        documents=document_count,
        concepts=concepts,
        empty=empty,
        structure_null=structure_null,
        verdict=verdict,
        threshold=threshold,
        reason=reason,
    )


def parse_args(argv: list[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog=CLI_ID,
        description=(
            "Judge one bundle per file type, with the denominator. Three verdicts: "
            "PASS (no worse than the pinned reference), FAIL, and UNMEASURED -- "
            "which is never PASS. Exit 0 judged and clean, 1 at least one FAIL, "
            "2 did not run, 3 nothing could be judged."
        ),
    )
    parser.add_argument("bundle", type=Path, help="the OKF bundle to judge")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    if not args.bundle.is_dir():
        print(f"{CLI_ID}: FAILED - no such bundle: {args.bundle}", file=sys.stderr)
        return 2
    try:
        report = measure_bundle(args.bundle)
    except (ConsumeError, OSError, ValueError) as exc:
        print(f"{CLI_ID}: FAILED - {exc}", file=sys.stderr)
        return 2
    print(report.render(), end="")
    return report.exit_code


if __name__ == "__main__":
    raise SystemExit(main())