llm-ingestion-okf/tools/okf_cid_measure.py

"""Measure the CID-glyph share of every document in a corpus. Measure, don't build.

Order 20260904T172353Z-6290714297-from-.claude. `docs/2026-09-04-k3-arm-c.md`
found that Bilag 9.1 -- the largest concept in the K2 bundle -- is an
extraction failure, not a segmentation failure: 95.1 % of its extracted text
is `(cid:N)` glyph codes, the shape `pdfminer.six` (behind `pdfplumber`, this
library's PDF reader) emits when a font carries no usable ToUnicode mapping.
This instrument answers whether Bilag 9.1 is alone in the corpus, or whether
the CID failure reaches other documents -- a K1-arm (door-level) question, not
a K3 one. It builds nothing: no CID-mapped extraction path, no new extractor.

Runs the SAME extraction call the door makes (`extract_text`), never a second
parser -- a number from a different code path would answer a different
question than "what does this repository actually persist."

`(cid:N)` and "four or more letters" are THIS instrument's own definitions,
not upstream terms; both are named here rather than assumed.
"""

from __future__ import annotations

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

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from llm_ingestion_okf.errors import ExtractionError
from llm_ingestion_okf.extract import extract_text

#: What "a CID glyph code" means here -- pdfminer.six's placeholder for a
#: character its font's encoding cannot map to Unicode. Not an upstream term:
#: this is the shape observed in the extracted text, not a spec.
_CID = re.compile(r"\(cid:\d+\)")

#: "An alphabetic word of four or more letters" -- the same measure used on
#: Bilag 9.1 in `docs/2026-09-04-k3-arm-c.md`. Unicode letters (so æøå count),
#: digits and `_` excluded, so a `(cid:12)` code's "cid" (3 letters) never
#: qualifies and the CID codes cannot inflate this count.
_WORD = re.compile(r"[^\W\d_]{4,}", re.UNICODE)


@dataclass(frozen=True)
class CidMeasurement:
    total_chars: int
    cid_chars: int
    word_count: int

    @property
    def pct(self) -> float:
        return 100.0 * self.cid_chars / self.total_chars if self.total_chars else 0.0


def measure(text: str) -> CidMeasurement:
    cid_chars = sum(len(match.group(0)) for match in _CID.finditer(text))
    word_count = len(_WORD.findall(text))
    return CidMeasurement(total_chars=len(text), cid_chars=cid_chars, word_count=word_count)


@dataclass(frozen=True)
class Row:
    """One corpus file's result: either measured, or a stated reason it was not."""

    name: str
    measurement: CidMeasurement | None
    skip_reason: str | None

    def line(self) -> str:
        if self.measurement is None:
            return f"| {self.name} | -- | -- | -- | -- | not measured: {self.skip_reason} |"
        m = self.measurement
        return (
            f"| {self.name} | {m.total_chars} | {m.cid_chars} | {m.pct:.1f} % | {m.word_count} | |"
        )


def run(corpus: Path) -> list[Row]:
    rows: list[Row] = []
    for path in sorted(corpus.iterdir(), key=lambda p: p.name):
        if not path.is_file():
            continue
        try:
            text = extract_text(path.name, path.read_bytes())
        except ExtractionError as exc:
            rows.append(Row(name=path.name, measurement=None, skip_reason=f"{exc.code}: {exc}"))
            continue
        rows.append(Row(name=path.name, measurement=measure(text), skip_reason=None))
    return rows


def render(corpus: Path, rows: list[Row]) -> str:
    measured = [row for row in rows if row.measurement is not None]
    over_10 = sum(
        1 for row in measured if row.measurement is not None and row.measurement.pct > 10.0
    )
    over_50 = sum(
        1 for row in measured if row.measurement is not None and row.measurement.pct > 50.0
    )
    lines = [
        f"# CID-glyph share, {corpus}",
        "",
        f"N = {len(rows)} (corpus directory file count). Measured: {len(measured)}/{len(rows)}.",
        f"Over 10 %: {over_10}/{len(measured)}. Over 50 %: {over_50}/{len(measured)}.",
        "",
        "| file | total chars | cid chars | cid share | words (4+ letters) | |",
        "|---|---|---|---|---|---|",
    ]
    lines.extend(row.line() for row in rows)
    return "\n".join(lines) + "\n"


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--corpus", type=Path, required=True, help="the directory to measure")
    parser.add_argument("--report", type=Path, required=True, help="where to write the table")
    args = parser.parse_args(argv)
    if not args.corpus.is_dir():
        print(f"okf-cid-measure: FAILED - no corpus directory at {args.corpus}", file=sys.stderr)
        return 1
    rows = run(args.corpus)
    args.report.write_text(render(args.corpus, rows), encoding="utf-8")
    print(f"okf-cid-measure: wrote {args.report}")
    return 0


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