145 lines
4.2 KiB
Python
145 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""CHECK: re-derive the drops catalog from the source folder and diff it
|
|
against the committed artifact drops-catalog.csv.
|
|
|
|
This is a real regeneration test, not a tautology: it re-walks
|
|
/root/stackchain-library-updates/drops from scratch, recomputes every field,
|
|
and compares row-for-row (path-sorted) against the committed CSV.
|
|
|
|
Exit code 0 = commit is correct; nonzero = wrong.
|
|
"""
|
|
import csv
|
|
import os
|
|
import sys
|
|
from collections import Counter
|
|
|
|
DROPS = "/root/stackchain-library-updates/drops"
|
|
ARTIFACT = "/root/troupe-standup/wizard-work/ledger/drops-catalog.csv"
|
|
|
|
# Same categories the builder uses.
|
|
_IMAGE_EXTS = (".jpg", ".jpeg", ".png", ".gif", ".webp", ".webm")
|
|
|
|
|
|
def classify(name):
|
|
low = name.lower()
|
|
if low.endswith(".mp4"):
|
|
return "mp4"
|
|
if low.endswith(".json"):
|
|
return "json"
|
|
if low.endswith(".txt"):
|
|
return "txt"
|
|
if low.endswith(".md"):
|
|
return "md"
|
|
if low.endswith(_IMAGE_EXTS):
|
|
return "jpg"
|
|
return "other"
|
|
|
|
|
|
def rederive():
|
|
"""Independently re-derive the same 42 rows the builder writes.
|
|
|
|
Re-walks /root/stackchain-library-updates/drops from scratch: every
|
|
top-level entry becomes either a dir-row (recursive file walk) or a
|
|
file-row (size only). No second pass over the file entries, so the row
|
|
count stays 39 dirs + 3 files = 42 (mirrors the builder's dedupe).
|
|
"""
|
|
rows = []
|
|
for name in sorted(os.listdir(DROPS)):
|
|
p = os.path.join(DROPS, name)
|
|
if os.path.isfile(p):
|
|
rows.append(_file_row(p, name))
|
|
else:
|
|
rows.append(_dir_row(p, name))
|
|
return rows
|
|
|
|
|
|
def _file_row(path, name):
|
|
size = 0
|
|
try:
|
|
size = os.path.getsize(path)
|
|
except OSError:
|
|
pass
|
|
return {
|
|
"path": name, "kind": "file", "total_files": 1,
|
|
"mp4": 0, "json": 0, "txt": 0, "md": 0, "jpg": 0, "other": 0,
|
|
"size_bytes": size,
|
|
}
|
|
|
|
|
|
def _dir_row(path, name):
|
|
total = 0
|
|
counts = Counter()
|
|
size = 0
|
|
for root, _dirs, files in os.walk(path):
|
|
for fn in files:
|
|
fp = os.path.join(root, fn)
|
|
total += 1
|
|
counts[classify(fn)] += 1
|
|
try:
|
|
size += os.path.getsize(fp)
|
|
except OSError:
|
|
pass
|
|
return {
|
|
"path": name, "kind": "dir", "total_files": total,
|
|
"mp4": counts["mp4"], "json": counts["json"], "txt": counts["txt"],
|
|
"md": counts["md"], "jpg": counts["jpg"], "other": counts["other"],
|
|
"size_bytes": size,
|
|
}
|
|
|
|
|
|
def read_csv():
|
|
with open(ARTIFACT, encoding="utf-8", newline="") as f:
|
|
reader = csv.DictReader(f)
|
|
return list(reader)
|
|
|
|
|
|
def to_native(row):
|
|
"""Normalise every cell to a JSON-serialisable python type for safe compare."""
|
|
return {k: (v.strip() if isinstance(v, str) else v) for k, v in row.items()}
|
|
|
|
|
|
def main():
|
|
errors = []
|
|
|
|
try:
|
|
actual = read_csv()
|
|
except Exception as exc:
|
|
print("CHECK FAILED: cannot read artifact %s: %s" % (ARTIFACT, exc))
|
|
sys.exit(1)
|
|
|
|
expected = rederive()
|
|
|
|
if len(actual) != len(expected):
|
|
errors.append("row count: artifact=%d re-derived=%d"
|
|
% (len(actual), len(expected)))
|
|
|
|
for i, (a, e) in enumerate(zip(actual, expected), 1):
|
|
a = to_native(a)
|
|
for key in ("path", "kind", "total_files", "mp4", "json", "txt",
|
|
"md", "jpg", "other", "size_bytes"):
|
|
if key not in a:
|
|
errors.append("row %d: missing field %r" % (i, key))
|
|
elif a[key] != str(e[key]):
|
|
errors.append("row %d field %r: artifact=%r re-derived=%r"
|
|
% (i, key, a[key], e[key]))
|
|
|
|
# Ordering check: artifact rows must be sorted by path the same way as re-derived.
|
|
actual_paths = [to_native(r)["path"] for r in actual]
|
|
expected_paths = [r["path"] for r in expected]
|
|
if actual_paths != expected_paths:
|
|
errors.append("row order/sort mismatch")
|
|
|
|
if errors:
|
|
print("CHECK FAILED:")
|
|
for e in errors:
|
|
print(" -", e)
|
|
sys.exit(1)
|
|
|
|
print("CHECK OK: %d rows re-derived from %s and identical to the committed "
|
|
"artifact." % (len(actual), DROPS))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|