#!/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 [ ...] [--pdf] [--rows] `` 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())