File size: 2,323 Bytes
bca45b0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | #!/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()
|