112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier for classify-messages.json.
|
|
|
|
Exit 0 iff the committed artifact re-derives every inbox message by the
|
|
triage rule (re-read from raw inputs, no import of the generator) and the
|
|
aggregate counts match.
|
|
|
|
Run: python3 verify_classify_messages.py
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO = Path("/root/troupe-standup")
|
|
INBOX = REPO / "wizard-work" / "sol" / "inbox"
|
|
ART = REPO / "wizard-work" / "sol" / "classify-messages.json"
|
|
|
|
# The exact rule the artifact is supposed to implement (copy, not import).
|
|
LANE_RULES = [
|
|
("ops", ["cron", "deploy", "disk", "backup", "server", "fail", "uptime"]),
|
|
("finance", ["invoice", "payment", "expense", "salary", "bitcoin", "fee"]),
|
|
("person", ["alexander", "sol", "wizard", "troupe", "meeting", "alex"]),
|
|
]
|
|
LANES = ["ops", "finance", "person", "other"]
|
|
|
|
|
|
def rederive():
|
|
"""Re-derive the classification of every inbox message straight from the
|
|
raw files, using only the rule above."""
|
|
items = []
|
|
for path in sorted(INBOX.glob("*.txt")):
|
|
subject, body = "", ""
|
|
for raw in path.read_text(encoding="utf-8").splitlines():
|
|
if raw.startswith("subject:"):
|
|
subject = raw.split(":", 1)[1].strip()
|
|
elif raw.startswith("body:"):
|
|
body = raw.split(":", 1)[1].strip()
|
|
text = (subject + " " + body).lower()
|
|
lane = "other"
|
|
for name, kws in LANE_RULES:
|
|
if any(k in text for k in kws):
|
|
lane = name
|
|
break
|
|
items.append({"file": path.name, "lane": lane})
|
|
return items
|
|
|
|
|
|
problems = []
|
|
|
|
|
|
def check(cond, msg):
|
|
if not cond:
|
|
problems.append(msg)
|
|
|
|
|
|
def main():
|
|
if not INBOX.is_dir():
|
|
problems.append("inbox dir missing")
|
|
print_fail()
|
|
return 1
|
|
if not ART.exists():
|
|
problems.append("artifact missing: " + str(ART))
|
|
print_fail()
|
|
return 1
|
|
|
|
expected = rederive()
|
|
expected_names = {e["file"] for e in expected}
|
|
got_names = set()
|
|
art = json.loads(ART.read_text(encoding="utf-8"))
|
|
|
|
# ---- 1. every input re-derives -------------------------------------
|
|
for e in expected:
|
|
got_names.add(e["file"])
|
|
a = next((i for i in art.get("items", []) if i.get("file") == e["file"]),
|
|
None)
|
|
check(a is not None, "no report item for " + e["file"])
|
|
if a is not None:
|
|
check(a.get("lane") == e["lane"],
|
|
"lane mismatch for %r: got %r want %r" % (e["file"], a.get("lane"), e["lane"]))
|
|
|
|
# ---- 2. every report item came from an input ------------------------
|
|
for a in art.get("items", []):
|
|
check(a.get("file") in expected_names,
|
|
"report item not in inbox: " + a.get("file", "?"))
|
|
|
|
# ---- 3. counts ------------------------------------------------------
|
|
exp_counts = {l: 0 for l in LANES}
|
|
for e in expected:
|
|
exp_counts[e["lane"]] += 1
|
|
if art.get("counts") != exp_counts:
|
|
problems.append("counts %r != expected %r" % (art.get("counts"), exp_counts))
|
|
if art.get("total") != len(expected):
|
|
problems.append("total %r != expected %d" % (art.get("total"), len(expected)))
|
|
|
|
if problems:
|
|
print_fail()
|
|
return 1
|
|
print("PASS: classify-messages.json re-derives %d/%d inbox messages by the "
|
|
"triage rule; counts=%s" % (len(expected), len(expected), art.get("counts")))
|
|
return 0
|
|
|
|
|
|
def print_fail():
|
|
print("FAIL")
|
|
for p in problems:
|
|
print(" -", p)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|