152 lines
4.1 KiB
Python
152 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Triage classifier for the sol lane.
|
|
|
|
Classifies a seed of inbox pending items into (category, decision) pairs and
|
|
emits a JSON artifact with per-item decisions plus aggregate counts.
|
|
|
|
Categories: archive_fix | app_pr | general
|
|
Decisions: approve | table | reject
|
|
|
|
Rules are keyword/phrase regexes, scored per category; highest score wins and
|
|
ties break by fixed priority. Decision is then chosen by a separate keyword
|
|
pass over the evidence text. Stdlib only, no third-party deps.
|
|
"""
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
DEFAULT_SEED = HERE / "seed-2026-10-29.json"
|
|
DEFAULT_ART = HERE / "Triage-2026-10-29.json"
|
|
|
|
CATS = ["archive_fix", "app_pr", "general"]
|
|
PRIORITY = {"archive_fix": 3, "app_pr": 2, "general": 1}
|
|
|
|
CATEGORY_RULES = {
|
|
"archive_fix": [
|
|
r"never shipped",
|
|
r"still renders",
|
|
r"lacks",
|
|
r"\black\b",
|
|
r"omit",
|
|
r"misattribute",
|
|
r"restats",
|
|
r"drops",
|
|
r"wrong",
|
|
r"fix",
|
|
],
|
|
"app_pr": [
|
|
r"publish",
|
|
r"release notes",
|
|
r"pull request",
|
|
r"endpoint",
|
|
r"submit",
|
|
r"healthz",
|
|
r"readme",
|
|
],
|
|
"general": [
|
|
r"standup",
|
|
r"digest",
|
|
r"handover",
|
|
r"under checksum",
|
|
r"independent QC",
|
|
],
|
|
}
|
|
|
|
DECISION_RULES = {
|
|
"table": [
|
|
r"standup",
|
|
r"digest",
|
|
r"hand over",
|
|
r"under checksum",
|
|
r"QC",
|
|
],
|
|
}
|
|
|
|
|
|
def _score(text, rules):
|
|
low = text.lower()
|
|
scores, matched = {}, {}
|
|
for cat, patterns in rules.items():
|
|
s = 0
|
|
ms = []
|
|
for p in patterns:
|
|
m = re.search(p, low)
|
|
if m:
|
|
s += 1
|
|
ms.append(m.group(0))
|
|
scores[cat] = s
|
|
matched[cat] = ms
|
|
return scores, matched
|
|
|
|
|
|
def classify(text):
|
|
"""Return (category, score, reason) from keyword rules."""
|
|
scores, matched = _score(text, CATEGORY_RULES)
|
|
best = max(scores.values())
|
|
if best == 0:
|
|
return "general", 0, "no category rule matched; defaulted to general"
|
|
winners = [c for c in CATS if scores[c] == best]
|
|
cat = max(winners, key=lambda c: PRIORITY[c])
|
|
parts = [f"{c}:{scores[c]}" for c in winners]
|
|
reason = "matched " + ", ".join(sorted(set(x for c in winners for x in matched[c])))
|
|
return cat, best, reason
|
|
|
|
|
|
def decide(text):
|
|
"""Return (decision, reason)."""
|
|
low = text.lower()
|
|
for cname, patterns in DECISION_RULES.items():
|
|
for p in patterns:
|
|
if re.search(p, low):
|
|
return "table", f"decision rule '{p}' matched"
|
|
return "approve", "no table keyword matched; approved"
|
|
|
|
|
|
def main(argv=None):
|
|
import argparse
|
|
ap = argparse.ArgumentParser(description="sol-lane triage classifier")
|
|
ap.add_argument("--seed", default=str(DEFAULT_SEED))
|
|
ap.add_argument("--output", default=str(DEFAULT_ART))
|
|
args = ap.parse_args()
|
|
|
|
seed = json.loads(Path(args.seed).read_text(encoding="utf-8"))
|
|
pending = seed.get("pending", [])
|
|
if not pending:
|
|
print("FAIL: seed has no pending items", file=sys.stderr)
|
|
return 1
|
|
|
|
items = []
|
|
for d in pending:
|
|
text = f"{d.get('evidence', '')} {d.get('proposed_action', '')}".lower()
|
|
cat, score, reason = classify(text)
|
|
dec, dec_reason = decide(text)
|
|
items.append({
|
|
"id": d["id"],
|
|
"category": cat,
|
|
"decision": dec,
|
|
"reason": f"{reason}; {dec_reason}",
|
|
})
|
|
|
|
cats = {c: 0 for c in CATS}
|
|
dec_counts = {"approve": 0, "table": 0, "reject": 0}
|
|
for it in items:
|
|
cats[it["category"]] += 1
|
|
dec_counts[it["decision"]] += 1
|
|
|
|
art = {
|
|
"seed": Path(args.seed).name,
|
|
"date": "2026-10-29",
|
|
"categories": cats,
|
|
"decision_counts": dec_counts,
|
|
"items": items,
|
|
}
|
|
Path(args.output).write_text(json.dumps(art, indent=2) + "\n", encoding="utf-8")
|
|
print(f"PASS: classified {len(items)} items -> {json.dumps(dec_counts)} / {json.dumps(cats)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|