llm-ingestion-okf/tools/okf_fidelity.py

"""Measure what survives conversion: the source's own strings, and their pairing.

The instrument behind K2. It exists because the arm A report of 2026-08-29
published `docx` 193/196 and `xlsx` 193/193 without shipping the command that
produced them, so those figures could not be re-measured -- against a new
converter version, against the product path, or at all.

Two questions, both of them the source document's own:

- **coverage** -- of the strings the file itself stores (a `docx` paragraph, an
  `xlsx` cell, a `pptx` or `odt` paragraph, an `rtf` cell), how many appear in
  the converted text. Not character count: a
  converter emitting 2.7x more characters can carry FEWER source strings, which
  arm A measured directly.
- **pairing** -- of the source rows carrying a label and a value, how many keep
  both on one output line. That is the criterion a requirement table is read
  by, and the one that separates a usable conversion from a pile of words.

Comparison is normalised (whitespace collapsed, case folded) because a
converter is entitled to re-wrap; it is not entitled to lose a string.

Run standalone and product side by side. A fall from one to the other is a
question this repository has to answer, not a number to publish: a product
figure alone cannot tell a converter limit from a defect we introduced.
"""

from __future__ import annotations

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

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

from llm_ingestion_okf.extract import extract_text  # noqa: E402

#: A source string shorter than this is not evidence either way -- a lone digit
#: or a stray bullet appears in almost any output by accident, so counting it
#: would inflate coverage for every converter equally and measure nothing.
MIN_STRING = 3

_WS = re.compile(r"\s+")
_XML_TEXT = re.compile(r"<(?:w:t|a:t|t)(?:\s[^>]*)?>(.*?)</(?:w:t|a:t|t)>", re.DOTALL)
_XML_PARA = re.compile(r"<w:p[\s>].*?</w:p>", re.DOTALL)
_XML_SI = re.compile(r"<si>(.*?)</si>", re.DOTALL)
#: A presentation's paragraph. A table cell holds one, so cells and body text
#: are read by the same rule rather than by two that could disagree.
_PPTX_PARA = re.compile(r"<a:p[\s>].*?</a:p>", re.DOTALL)
#: A slide part. Sorted NUMERICALLY on the trailing index, because the
#: lexicographic order of `slide2` and `slide10` is not the deck's order.
_PPTX_SLIDE = re.compile(r"^ppt/slides/slide(\d+)\.xml$")
#: An ODF paragraph or heading. A table cell holds a `text:p`, so the same rule
#: covers cells; the inner markup (`text:span` and friends) is stripped rather
#: than parsed, since the question is which STRING the file stores.
_ODF_PARA = re.compile(r"<text:(p|h)(?:\s[^>]*)?>(.*?)</text:\1>", re.DOTALL)
_TAGS = re.compile(r"<[^>]*>")
_ENTITIES = (("&amp;", "&"), ("&lt;", "<"), ("&gt;", ">"), ("&quot;", '"'), ("&apos;", "'"))


#: Stripped from both sides before comparison. These are the converter's own
#: MARKUP, not the document's content: markdown escapes a `[`, wraps a bold run
#: in `**`, and writes an en dash as `--`. Measured on the corpus -- without
#: this, `docx` coverage reads 149/196 while every one of the 47 "missing"
#: strings is present in the output verbatim, wearing an escape. The question
#: is whether the STRING survived, not whether the markup matches.
_MARKUP = str.maketrans({char: None for char in "\\*_`"})
_DASHES = str.maketrans({"\u2013": "-", "\u2014": "-", "\u00a0": " ", "\u2019": "'"})


def normalise(text: str) -> str:
    for entity, char in _ENTITIES:
        text = text.replace(entity, char)
    text = text.translate(_MARKUP).translate(_DASHES)
    text = text.replace("--", "-")
    return _WS.sub(" ", text).strip().casefold()


def _runs(fragment: str) -> str:
    return "".join(match.group(1) for match in _XML_TEXT.finditer(fragment))


# --- the rtf stream ---------------------------------------------------------
#
# RTF is not a package, so there is no XML part to read the document's own
# strings out of: the stream itself is the format. What follows reads exactly
# the constructs the fixture set uses -- groups, destinations, control words,
# the `\'hh` byte escape and the `\uN` unicode escape -- and it is stated as a
# limit rather than sold as an RTF parser. That limit matters for what the
# measurement can claim: for `pptx` and `odt` the fasit comes from a package
# format neither we nor the converter defined, and for `rtf` it comes from a
# reader written here against a stream written here. The `rtf` row is the
# weakest of the three on that axis and the report says so.

#: Control words that END a paragraph or a cell. Everything the document is
#: divided into for the purpose of "did this string survive".
_RTF_BOUNDARIES = frozenset({"par", "cell", "row", "line", "sect", "nestcell", "nestrow"})

#: Groups whose contents are the file's own bookkeeping rather than its text.
#: Reading them would count a font name as a document string.
_RTF_DESTINATIONS = frozenset(
    {
        "fonttbl",
        "colortbl",
        "stylesheet",
        "info",
        "pict",
        "listtable",
        "generator",
        "filetbl",
    }
)

#: A control word plus its optional numeric parameter. The trailing space is
#: the word's DELIMITER and is consumed with it -- that single rule is what
#: makes `\u248 ?` read as one character followed by a discardable `?`, and it
#: is the rule the vendored converter applies differently (see the fixture
#: README).
_RTF_CONTROL = re.compile(r"\\([a-z]+)(-?[0-9]+)? ?")


def rtf_paragraphs(raw: bytes) -> list[str]:
    """The strings an RTF stream stores, one per paragraph or table cell."""
    text = raw.decode("cp1252", "replace")
    end = len(text)
    out: list[str] = []
    buf: list[str] = []
    position = 0
    depth = 0
    skip_depth: int | None = None
    pending_replacements = 0

    while position < end:
        char = text[position]
        if char == "{":
            depth += 1
            position += 1
            continue
        if char == "}":
            if skip_depth is not None and depth <= skip_depth:
                skip_depth = None
            depth -= 1
            position += 1
            continue
        if char == "\\":
            control = _RTF_CONTROL.match(text, position)
            if control is not None:
                word, parameter = control.group(1), control.group(2)
                position = control.end()
                if skip_depth is not None:
                    continue
                if word == "u" and parameter is not None:
                    buf.append(chr(int(parameter)))
                    pending_replacements = 1
                    continue
                pending_replacements = 0
                if word in _RTF_BOUNDARIES:
                    out.append("".join(buf))
                    buf = []
                elif word in _RTF_DESTINATIONS:
                    skip_depth = depth
                continue
            symbol = text[position + 1] if position + 1 < end else ""
            if symbol == "'":
                if skip_depth is None:
                    buf.append(chr(int(text[position + 2 : position + 4] or "3f", 16)))
                position += 4
                continue
            if symbol == "*":
                skip_depth = depth
                position += 2
                continue
            if skip_depth is None:
                buf.append(symbol)
            position += 2
            continue
        position += 1
        if char in "\r\n" or skip_depth is not None:
            continue
        if pending_replacements:
            pending_replacements -= 1
            continue
        buf.append(char)

    out.append("".join(buf))
    return out


def source_strings(path: Path) -> tuple[str, ...]:
    """The strings the FILE itself stores, deduplicated and in document order.

    Read from the package's own XML rather than from any converter's output:
    a fasit derived from one converter would score that converter on its own
    homework.
    """
    suffix = path.suffix.lower()
    if suffix == ".rtf":
        return _dedupe(rtf_paragraphs(path.read_bytes()))
    with zipfile.ZipFile(path) as archive:
        names = set(archive.namelist())
        if suffix == ".docx":
            body = archive.read("word/document.xml").decode("utf-8", "replace")
            found = [_runs(match.group(0)) for match in _XML_PARA.finditer(body)]
        elif suffix == ".xlsx":
            found = []
            if "xl/sharedStrings.xml" in names:
                shared = archive.read("xl/sharedStrings.xml").decode("utf-8", "replace")
                found = [_runs(match.group(1)) for match in _XML_SI.finditer(shared)]
        elif suffix == ".pptx":
            found = []
            slides = [
                (int(match.group(1)), match.group(0))
                for match in (_PPTX_SLIDE.match(name) for name in names)
                if match is not None
            ]
            for _, name in sorted(slides):
                slide = archive.read(name).decode("utf-8", "replace")
                found.extend(_runs(match.group(0)) for match in _PPTX_PARA.finditer(slide))
        elif suffix == ".odt":
            body = archive.read("content.xml").decode("utf-8", "replace")
            found = [_TAGS.sub("", match.group(2)) for match in _ODF_PARA.finditer(body)]
        else:
            raise ValueError(f"no source-string reader for {suffix!r}")

    return _dedupe(found)


def _dedupe(found: list[str]) -> tuple[str, ...]:
    seen: dict[str, None] = {}
    for item in found:
        value = normalise(item)
        if len(value) >= MIN_STRING:
            seen.setdefault(value, None)
    return tuple(seen)


def label_value_rows(strings: tuple[str, ...]) -> tuple[str, ...]:
    """Source strings that read as a label expecting a value beside it.

    A row is usable for pairing when its own text ends in a colon or is a short
    noun phrase followed by a value elsewhere on the row. Approximated by the
    colon rule alone, which is stated rather than tuned: a threshold fitted to
    make a number look good is not a measurement.
    """
    return tuple(item for item in strings if item.endswith(":"))


@dataclass(frozen=True)
class Fidelity:
    path: str
    covered: int
    total: int
    paired: int
    pairable: int

    def line(self) -> str:
        share = 100.0 * self.covered / self.total if self.total else 0.0
        pair = 100.0 * self.paired / self.pairable if self.pairable else 0.0
        return (
            f"{self.path}: coverage {self.covered}/{self.total} ({share:.1f} %), "
            f"paired {self.paired}/{self.pairable} ({pair:.1f} %)"
        )


def score(path: Path, text: str) -> Fidelity:
    strings = source_strings(path)
    haystack = normalise(text)
    lines = [normalise(line) for line in text.splitlines()]
    covered = sum(1 for item in strings if item in haystack)
    pairable = label_value_rows(strings)
    paired = sum(
        1 for label in pairable if any(label in line and len(line) > len(label) for line in lines)
    )
    return Fidelity(
        path=path.name, covered=covered, total=len(strings), paired=paired, pairable=len(pairable)
    )


def product_text(path: Path) -> str:
    """The text the LIBRARY produces -- the same call the door makes."""
    return extract_text(path.name, path.read_bytes())


def standalone_text(path: Path) -> str:
    """The text the converter produces on its own, at the pinned version."""
    import pypandoc

    from llm_ingestion_okf._pandoc import converter_path

    formats = {".docx": "docx", ".xlsx": "xlsx", ".pptx": "pptx", ".odt": "odt", ".rtf": "rtf"}
    with converter_path():
        return str(
            pypandoc.convert_file(
                str(path),
                "markdown",
                format=formats[path.suffix.lower()],
                extra_args=["--eol=lf", "--wrap=none"],
            )
        )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("files", nargs="+", type=Path)
    args = parser.parse_args(argv)
    for path in args.files:
        print(f"--- {path.name}")
        print(f"  standalone  {score(path, standalone_text(path)).line()}")
        print(f"  product     {score(path, product_text(path)).line()}")
    return 0


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