llm-ingestion-okf/tools/okf_asset_census.py

#!/usr/bin/env python3
"""What happens to every image this machine can reach, one row each.

The known-positive instrument for a change to the asset path. A round that
converts one format has to show that it moved NOTHING else, and "nothing else"
is a statement about a corpus, not about a fixture. This script produces the
per-image row that makes the claim checkable: the source's own sha256, the
format read off its bytes, and what `assets.read_image` did with it -- carried
under some digest, or refused with a code.

Run it from two pinned trees and diff the rows. A JPEG whose carried digest
moved is a regression with a name; a BMP whose carried digest moved from a
`.bmp` row to a `.png` row is the round working.

IT IS NOT PART OF THE PACKAGE. It lives outside `src/`, never enters a wheel,
and a consumer's install surface is unchanged by its existence.

Two rules it is built around, both learned in this repository:

- **A count with no denominator is not a measurement.** Every summary line
  carries `of N`, and a run that found no image says so rather than printing
  an empty table that reads like a clean result.
- **A failed read is never an empty result.** An unreadable file is its own
  row with its own reason, never a silent absence from the census.

Usage:

    python3 tools/okf_asset_census.py <path> [<path> ...] [--pdf] [--rows]

`<path>` is a file or a directory walked recursively. `--pdf` additionally
opens every `.pdf` found and runs the PDF image reader over it, which is the
slow half and needs the `[extract]` extra. `--rows` prints one line per image;
without it only the summary is printed.
"""

from __future__ import annotations

import argparse
import hashlib
import sys
from collections import Counter
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))

from llm_ingestion_okf import assets as assets_module
from llm_ingestion_okf.errors import ExtractionError

#: Suffixes worth opening as a standalone image. The type is still SNIFFED --
#: this only decides which files are picked up off the disk, and a file whose
#: name lies is reported under the format its bytes declare.
IMAGE_SUFFIXES = frozenset(
    {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".tif", ".tiff", ".webp", ".jp2", ".j2k"}
)


def _digest(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _row(origin: str, data: bytes) -> tuple[str, str, str]:
    """`(origin, source digest, outcome)` for one image's bytes."""
    before = _digest(data)
    sniffed = assets_module.sniff(data)
    kind = sniffed[0] if sniffed else "not-an-image"
    try:
        image = assets_module.read_image(data, name=origin)
    except ExtractionError as exc:
        return origin, before, f"REJECT {exc.code} in={kind}"
    return (
        origin,
        before,
        f"CARRY {image.media_type}{image.suffix} in={kind} out={_digest(image.data)}",
    )


def _walk(paths: list[Path]) -> list[Path]:
    found: list[Path] = []
    for path in paths:
        if path.is_file():
            found.append(path)
        elif path.is_dir():
            found += [p for p in path.rglob("*") if p.is_file()]
        else:
            print(f"census: {path} is neither a file nor a directory", file=sys.stderr)
    return sorted(found)


def _pdf_rows(path: Path) -> list[tuple[str, str, str]]:
    """Every image object the PDF reader reaches on this file.

    The reader is asked for the images it CARRIES and the ones it refuses, so
    a document whose pictures all fail is a set of coded rows and not a blank.
    """
    import pdfplumber

    from llm_ingestion_okf import extract as extract_module

    rows: list[tuple[str, str, str]] = []
    with pdfplumber.open(str(path)) as document:
        for number, page in enumerate(document.pages, start=1):
            # THE PAGE, not `page.page_obj`. `_pdf_images` reads `page.images`
            # through `getattr(..., [])`, so handing it the wrong object
            # returns an empty tuple and no error: measured 2026-09-19, that
            # mistake reported 0 images over 67 PDFs and read like a clean
            # result. The attribute is asserted rather than assumed.
            if not hasattr(page, "images"):
                raise RuntimeError(
                    f"{path}: the page object exposes no `images` attribute; this census "
                    "would report an absence it never measured"
                )
            carried, rejected = extract_module._pdf_images(page)
            for image in carried:
                rows.append(
                    (
                        f"{path}#p{number}:{image.name}",
                        _digest(image.data),
                        f"CARRY {image.media_type}{image.suffix} out={_digest(image.data)}",
                    )
                )
            for rejection in rejected:
                rows.append((f"{path}#p{number}:{rejection.name}", "-", f"REJECT {rejection.code}"))
    return rows


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("paths", nargs="+", type=Path)
    parser.add_argument("--pdf", action="store_true", help="also open every .pdf found")
    parser.add_argument("--rows", action="store_true", help="print one line per image")
    args = parser.parse_args(argv)

    files = _walk(args.paths)
    rows: list[tuple[str, str, str]] = []
    pdfs = 0
    unreadable = 0
    for path in files:
        if path.suffix.lower() in IMAGE_SUFFIXES:
            try:
                data = path.read_bytes()
            except OSError as exc:
                unreadable += 1
                rows.append((str(path), "-", f"UNREADABLE {exc.__class__.__name__}"))
                continue
            rows.append(_row(str(path), data))
        elif args.pdf and path.suffix.lower() == ".pdf":
            pdfs += 1
            try:
                rows += _pdf_rows(path)
            except Exception as exc:  # a broken PDF is a row, never a blank
                unreadable += 1
                rows.append((str(path), "-", f"UNREADABLE {exc.__class__.__name__}"))

    if args.rows:
        for origin, before, outcome in rows:
            print(f"{before}\t{outcome}\t{origin}")

    total = len(rows)
    print(f"files walked: {len(files)}; pdfs opened: {pdfs}; image rows: {total}")
    if total == 0:
        print("NO IMAGE MEASURED -- this is an absence, not a clean result")
        return 3
    summary = Counter(outcome.split(" out=")[0] for _, _, outcome in rows)
    for outcome, count in sorted(summary.items()):
        print(f"  {count} of {total}\t{outcome}")
    carried = sum(count for outcome, count in summary.items() if outcome.startswith("CARRY"))
    print(f"carried: {carried} of {total}; refused: {total - carried} of {total}")
    return 0


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