106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""CHECK -- extract: Hermes profile directory census.
|
|
|
|
Single bare command whose exit code is the verdict (0 = correct, nonzero = wrong):
|
|
python3 /root/.hermes/wizard-outputs/ledger/verify_wizardwork_census.py
|
|
|
|
Recomputes the content census straight from the live filesystem (os.stat + os.walk +
|
|
hashlib.sha256) and diffs cell-by-cell against the committed TSV. A changed or added
|
|
content file, a missing file, or a wrong size/size makes this exit 1. Volatile
|
|
runtime dirs, `*.pyc`, and this verifier's own outputs are excluded so a fresh run
|
|
stays green in the same directory.
|
|
|
|
This is a real test (any real content change flips a value and fails), not a
|
|
tautology.
|
|
|
|
Exit codes:
|
|
0 committed TSV matches live content census
|
|
1 mismatch (row count, relpath, size, or sha256)
|
|
2 ungradable (source unreadable or malformed)
|
|
"""
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
from datetime import date
|
|
|
|
ROOT = "/root/.hermes/profiles/ledger"
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
TSV = os.path.join(HERE, "wizard-work-census-%s.tsv" % date.today().isoformat())
|
|
MD = os.path.join(HERE, "wizard-work-census-%s.md" % date.today().isoformat())
|
|
VOLATILE_DIRS = {
|
|
"__pycache__", "cache", "logs", "image_cache", "audio_cache",
|
|
"plugin-update-checks", "source-checks", "runtime", "work", "workspace",
|
|
"sessions",
|
|
}
|
|
EXCLUDE_SUFFIXES = (".pyc", ".pyo")
|
|
EXCLUDE_NAMES = {".DS_Store"}
|
|
VERIFY_BASENAME = "verify_wizardwork_census.py"
|
|
|
|
|
|
def is_artifact(fn, rel):
|
|
if fn in EXCLUDE_NAMES or fn.endswith(EXCLUDE_SUFFIXES):
|
|
return True
|
|
if fn == VERIFY_BASENAME:
|
|
return True
|
|
if rel.startswith(HERE + os.sep):
|
|
return True
|
|
return False
|
|
|
|
|
|
def expected_rows():
|
|
out = []
|
|
for dirpath, dirnames, filenames in os.walk(ROOT):
|
|
dirnames[:] = [d for d in dirnames if d not in VOLATILE_DIRS]
|
|
for fn in filenames:
|
|
if is_artifact(fn, os.path.relpath(os.path.join(dirpath, fn), ROOT)):
|
|
continue
|
|
p = os.path.join(dirpath, fn)
|
|
rel = os.path.relpath(p, ROOT)
|
|
out.append((rel, os.stat(p).st_size, hashlib.sha256(open(p, "rb").read()).hexdigest()))
|
|
out.sort()
|
|
return out
|
|
|
|
|
|
def read_committed():
|
|
rows = []
|
|
with open(TSV, encoding="utf-8", newline="") as f:
|
|
for line in f:
|
|
line = line.rstrip("\n")
|
|
if not line or line.startswith("relpath"):
|
|
continue
|
|
parts = line.split("\t")
|
|
if len(parts) != 3:
|
|
raise ValueError("malformed line: %r" % line)
|
|
rows.append((parts[0], int(parts[1]), parts[2]))
|
|
return rows
|
|
|
|
|
|
def main():
|
|
try:
|
|
expected = expected_rows()
|
|
except Exception as exc: # noqa: BLE001
|
|
print("UNGRADABLE: %r" % exc)
|
|
return 2
|
|
try:
|
|
committed = read_committed()
|
|
except Exception as exc: # noqa: BLE001
|
|
print("UNGRADABLE: cannot read committed TSV: %r" % exc)
|
|
return 2
|
|
|
|
if len(expected) != len(committed):
|
|
print("FAIL: row count live=%d committed=%d" % (len(expected), len(committed)))
|
|
return 1
|
|
|
|
for (rel, size, sha), (c_rel, c_size, c_sha) in zip(expected, committed):
|
|
if rel != c_rel or size != c_size or sha != c_sha:
|
|
print("FAIL: %s live=(size=%s sha=%s...) committed=(size=%s sha=%s...)"
|
|
% (rel, size, sha[:12], c_size, c_sha[:12]))
|
|
return 1
|
|
|
|
print("PASS: %d files, committed TSV matches live content census" % len(expected))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|