scripts/validate_covt_references.py

#!/usr/bin/env python3
"""Verify that every CoVT image reference resolves, with no extra image files."""

from __future__ import annotations

import json
import os
import re
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
IMAGE_FIELD = re.compile(r'"image"\s*:\s*"([^"\\]*(?:\\.[^"\\]*)*)"')


def main() -> None:
    covt = ROOT / "data" / "covt"
    stats = json.loads((covt / "stats.json").read_text(encoding="utf-8"))

    reference_count = 0
    referenced: set[str] = set()
    with (covt / "data.json").open(encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, 1):
            match = IMAGE_FIELD.search(line)
            if match is None:
                continue
            try:
                image_name = json.loads(f'"{match.group(1)}"')
            except json.JSONDecodeError as exc:
                raise ValueError(
                    f"invalid escaped image name at data.json line {line_number}"
                ) from exc
            reference_count += 1
            referenced.add(image_name)

    # The materialized image directory is contractually flat. os.listdir avoids
    # hundreds of thousands of remote per-file stat calls on the Weka mount.
    actual = set(os.listdir(covt / "images"))
    missing = sorted(referenced - actual)
    extra = sorted(actual - referenced)

    print(f"record_image_refs={reference_count}")
    print(f"unique_referenced_images={len(referenced)}")
    print(f"actual_directory_entries={len(actual)}")
    print(f"duplicate_reference_count={reference_count - len(referenced)}")
    print(f"missing_references={len(missing)}")
    print(f"unreferenced_entries={len(extra)}")
    if missing:
        print(f"first_missing={missing[:10]}")
    if extra:
        print(f"first_unreferenced={extra[:10]}")

    if reference_count != int(stats["rows"]):
        raise AssertionError(
            f"image-bearing row count {reference_count} != {stats['rows']}"
        )
    if len(referenced) != int(stats["image_count"]):
        raise AssertionError(
            f"unique references {len(referenced)} != {stats['image_count']}"
        )
    if missing or extra:
        raise AssertionError("CoVT image reference closure is not exact")
    print("covt_reference_closure=PASS")


if __name__ == "__main__":
    main()