129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Triage classifier for the sol lane.
|
|
|
|
Classifies a list of demands/requests into triage categories
|
|
(decision | info | action | done) and emits a CSV + JSON payload.
|
|
|
|
Rules are keyword/phrase regexes, scored per category. Highest score wins;
|
|
ties broken by a fixed priority order. No third-party deps, stdlib only.
|
|
"""
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
CATS = ["decision", "info", "action", "done"]
|
|
# fixed priority for tie-breaking (higher = elected first on a draw)
|
|
PRIORITY = {"decision": 4, "action": 3, "info": 2, "done": 1}
|
|
|
|
RULES = {
|
|
"decision": [
|
|
r"decision", r"\bwhich\b", r"\bchoose\b", r"\bpick\b",
|
|
r"\bshould we\b", r"\bapprove\b", r"\bconfirm\b", r"\bagree\b",
|
|
r"\bpolicy\b", r"\breconcile\b", r"\bresolve\b", r"\bftr eve\b",
|
|
r"\barchitect\b", r"\bcommit\b",
|
|
],
|
|
"info": [
|
|
r"what is", r"how does", r"explain", r"overview", r"summary",
|
|
r"define", r"meaning", r"difference", r"compare", r"describe",
|
|
r"purpose", r"why", r"when",
|
|
],
|
|
"action": [
|
|
r"fix", r"implement", r"send", r"create", r"deploy", r"update",
|
|
r"clean", r"setup", r"\brun\b", r"\benable\b", r"\binstall\b",
|
|
r"\bmigrate\b", r"\bbuild\b", r"\bprocess\b", r"\bdispatch\b",
|
|
],
|
|
"done": [
|
|
r"\bdone\b", r"\bclosed\b", r"\bcompleted\b", r"\bresolved\b",
|
|
r"\bfinished\b", r"\bprocessed\b", r"\bsent\b", r"\bok\b",
|
|
r"\bworkaround\b",
|
|
],
|
|
}
|
|
|
|
TIE_PRIORITY = {"decision": 4, "action": 3, "info": 2, "done": 1}
|
|
|
|
|
|
def classify(text):
|
|
"""Return (category, score, matched) for a single demand string."""
|
|
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
|
|
|
|
best = max(scores.values())
|
|
if best == 0:
|
|
cat = "info"
|
|
reason = "no rule matched; defaulted to info"
|
|
else:
|
|
# pick highest score, tie-break by fixed priority
|
|
winners = [c for c in CATS if scores[c] == best]
|
|
cat = max(winners, key=lambda c: TIE_PRIORITY[c])
|
|
parts = []
|
|
for c in winners:
|
|
parts.append(f"{c}:{scores[c]}")
|
|
reason = "matched " + ", ".join(sorted(set(x for c in winners for x in matched[c])))
|
|
return cat, best, reason
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description="Triage classifier")
|
|
ap.add_argument("--input", required=True, help="JSON file with a 'demands' array")
|
|
ap.add_argument("--output", required=True, help="CSV output path")
|
|
ap.add_argument("--pretty", action="store_true", help="also write .json payload")
|
|
args = ap.parse_args()
|
|
|
|
data = json.loads(Path(args.input).read_text(encoding="utf-8"))
|
|
demands = data.get("demands", data if isinstance(data, list) else [])
|
|
|
|
rows = []
|
|
for d in demands:
|
|
text = d.get("text", "")
|
|
priority = d.get("priority")
|
|
cat, score, reason = classify(text)
|
|
if priority is None:
|
|
priority = {"decision": 1, "action": 2, "info": 3, "done": 4}[cat]
|
|
rows.append({
|
|
"id": d.get("id", ""),
|
|
"text": text,
|
|
"category": cat,
|
|
"score": score,
|
|
"priority": priority,
|
|
"reason": reason,
|
|
})
|
|
|
|
fieldnames = ["id", "text", "category", "score", "priority", "reason"]
|
|
with open(args.output, "w", encoding="utf-8", newline="") as f:
|
|
w = csv.DictWriter(f, fieldnames=fieldnames, lineterminator="\n")
|
|
w.writeheader()
|
|
w.writerows(rows)
|
|
|
|
if args.pretty:
|
|
out_json = args.output.rsplit(".", 1)[0] + ".json"
|
|
Path(out_json).write_text(
|
|
json.dumps({"classified": rows, "summary": _summary(rows)}, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
return 0
|
|
|
|
|
|
def _summary(rows):
|
|
counts = {c: 0 for c in CATS}
|
|
for r in rows:
|
|
counts[r["category"]] += 1
|
|
return {"total": len(rows), "by_category": counts}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|