darkplex-core/cortex/triage.py

#!/usr/bin/env python3
"""Priority Triage System — score and rank tasks by urgency, importance, effort, readiness."""

import json
import re
import sys
from dataclasses import dataclass, asdict

# --- Keyword banks (DE + EN) ---

URGENCY_KEYWORDS = {
    # word -> weight (0.0-1.0 contribution)
    "jetzt": 0.9, "sofort": 0.9, "asap": 0.8, "urgent": 0.8, "dringend": 0.8,
    "notfall": 1.0, "emergency": 1.0, "down": 0.9, "broken": 0.7, "kaputt": 0.7,
    "fehler": 0.6, "error": 0.6, "critical": 0.9, "alarm": 0.8, "outage": 0.9,
    "crash": 0.8, "timeout": 0.5, "offline": 0.7, "immediately": 0.8,
    "deadline": 0.7, "frist": 0.7, "heute": 0.5, "today": 0.5,
    "blocker": 0.8, "p0": 1.0, "p1": 0.8, "sev1": 0.9, "sev2": 0.7,
    "morgen": 0.6, "tomorrow": 0.6, "nächste woche": 0.5, "next week": 0.5,
    "fällig": 0.7, "overdue": 0.8, "überfällig": 0.8,
    "beschwer": 0.6, "complaint": 0.6, "problem": 0.4,
}

IMPORTANCE_KEYWORDS = {
    "production": 0.9, "prod": 0.9, "security": 0.9, "sicherheit": 0.9,
    "geld": 0.8, "money": 0.8, "finance": 0.8, "finanzen": 0.8,
    "kunde": 0.7, "customer": 0.7, "kunden": 0.7, "bafin": 1.0,
    "regulatory": 0.9, "compliance": 0.9, "gateway": 0.7, "dns": 0.7,
    "albert": 0.6, "chef": 0.6, "boss": 0.6,
    "database": 0.6, "datenbank": 0.6, "backup": 0.7,
    "ssl": 0.7, "certificate": 0.7, "zertifikat": 0.7,
    "auth": 0.7, "login": 0.6, "password": 0.7, "passwort": 0.7,
    "data loss": 1.0, "datenverlust": 1.0,
    # Security keywords
    "breach": 0.95, "compromised": 0.9, "hacked": 0.95, "vulnerability": 0.8,
    "exploit": 0.9, "unauthorized": 0.85, "ssh keys": 0.85,
    # Compliance keywords
    "audit": 0.8, "iso": 0.7, "compliance": 0.9, "regulatory": 0.9,
    "certification": 0.7, "zertifizierung": 0.7,
    # Business keywords
    "revenue": 0.8, "umsatz": 0.8, "zahlung": 0.7, "payment": 0.7,
    "beschwer": 0.7,
}

LOW_EFFORT_KEYWORDS = {
    "quick": 0.8, "schnell": 0.8, "einfach": 0.8, "simple": 0.8,
    "typo": 0.9, "tippfehler": 0.9, "config": 0.6, "toggle": 0.9,
    "restart": 0.7, "neustart": 0.7, "one-liner": 0.9, "fix": 0.5,
}

HIGH_EFFORT_KEYWORDS = {
    "komplett": 0.8, "refactor": 0.9, "migration": 0.9, "redesign": 0.9,
    "rewrite": 0.9, "umschreiben": 0.9, "architecture": 0.8, "architektur": 0.8,
    "von grund auf": 0.9, "from scratch": 0.9, "multi-step": 0.6,
    "complex": 0.7, "komplex": 0.7, "projekt": 0.5, "project": 0.5,
}

NOT_READY_KEYWORDS = {
    "blocked": 0.9, "blockiert": 0.9, "warte auf": 0.8, "waiting for": 0.8,
    "depends on": 0.8, "abhängig von": 0.8, "brauche erst": 0.8, "need first": 0.8,
    "pending": 0.6, "ausstehend": 0.6, "not yet": 0.5, "noch nicht": 0.5,
}


def _scan_keywords(text: str, keywords: dict[str, float]) -> float:
    """Scan text for keywords, return max-based score with diminishing boost."""
    t = text.lower()
    hits = []
    for kw, weight in keywords.items():
        if kw in t:
            hits.append(weight)
    if not hits:
        return 0.0
    hits.sort(reverse=True)
    score = hits[0]
    for h in hits[1:]:
        score = min(1.0, score + h * 0.15)  # diminishing returns
    return round(min(1.0, score), 3)


@dataclass
class TriageScore:
    text: str
    urgency: float
    importance: float
    effort: float  # 0=trivial, 1=massive
    readiness: float  # 1=ready, 0=blocked
    priority: float

    def to_dict(self):
        d = asdict(self)
        d["priority"] = round(d["priority"], 3)
        return d


def score_task(text: str) -> TriageScore:
    """Score a single task/message."""
    urgency = _scan_keywords(text, URGENCY_KEYWORDS)
    importance = _scan_keywords(text, IMPORTANCE_KEYWORDS)

    low_effort = _scan_keywords(text, LOW_EFFORT_KEYWORDS)
    high_effort = _scan_keywords(text, HIGH_EFFORT_KEYWORDS)
    # effort: 0=trivial, 1=massive. Default 0.5 (medium)
    if low_effort > high_effort:
        effort = max(0.1, 0.5 - low_effort * 0.4)
    elif high_effort > low_effort:
        effort = min(1.0, 0.5 + high_effort * 0.4)
    else:
        effort = 0.5

    not_ready = _scan_keywords(text, NOT_READY_KEYWORDS)
    readiness = round(1.0 - not_ready, 3)

    # priority_score = (urgency * 0.4) + (importance * 0.35) + (1/effort_norm * 0.15) + (ready * 0.1)
    # 1/effort: invert so low effort = high score. Scale effort 0.1-1.0 -> 1.0-10.0 inverted to 0.1-1.0
    effort_inv = round(1.0 - effort, 3)  # simple inversion: low effort -> high score
    priority = (urgency * 0.4) + (importance * 0.35) + (effort_inv * 0.15) + (readiness * 0.1)

    return TriageScore(
        text=text,
        urgency=round(urgency, 3),
        importance=round(importance, 3),
        effort=round(effort, 3),
        readiness=readiness,
        priority=round(priority, 3),
    )


def rank_tasks(texts: list[str]) -> list[TriageScore]:
    """Score and rank multiple tasks by priority (highest first)."""
    scores = [score_task(t) for t in texts]
    scores.sort(key=lambda s: s.priority, reverse=True)
    return scores


def analyze_message(text: str) -> dict:
    """Deep analysis of a message — return signals found."""
    score = score_task(text)
    t = text.lower()

    signals = {"urgency": [], "importance": [], "effort": [], "readiness": []}
    for kw in URGENCY_KEYWORDS:
        if kw in t:
            signals["urgency"].append(kw)
    for kw in IMPORTANCE_KEYWORDS:
        if kw in t:
            signals["importance"].append(kw)
    for kw in LOW_EFFORT_KEYWORDS:
        if kw in t:
            signals["effort"].append(f"{kw} (low)")
    for kw in HIGH_EFFORT_KEYWORDS:
        if kw in t:
            signals["effort"].append(f"{kw} (high)")
    for kw in NOT_READY_KEYWORDS:
        if kw in t:
            signals["readiness"].append(f"{kw} (blocker)")

    # Classify
    p = score.priority
    if p >= 0.8:
        classification = "🔴 CRITICAL — handle immediately"
    elif p >= 0.6:
        classification = "🟠 HIGH — handle soon"
    elif p >= 0.4:
        classification = "🟡 MEDIUM — schedule"
    elif p >= 0.2:
        classification = "🔵 LOW — backlog"
    else:
        classification = "⚪ MINIMAL — ignore"

    return {
        "score": score.to_dict(),
        "signals": signals,
        "classification": classification,
    }


def main():
    if len(sys.argv) < 2:
        print("Usage: python3 triage.py <score|rank|analyze> <text...>")
        sys.exit(1)

    cmd = sys.argv[1]

    if cmd == "score":
        if len(sys.argv) < 3:
            print("Usage: python3 triage.py score \"task description\"")
            sys.exit(1)
        result = score_task(sys.argv[2])
        print(json.dumps(result.to_dict(), indent=2, ensure_ascii=False))

    elif cmd == "rank":
        if len(sys.argv) < 3:
            print("Usage: python3 triage.py rank \"task1\" \"task2\" ...")
            sys.exit(1)
        tasks = sys.argv[2:]
        ranked = rank_tasks(tasks)
        output = [{"rank": i + 1, **s.to_dict()} for i, s in enumerate(ranked)]
        print(json.dumps(output, indent=2, ensure_ascii=False))

    elif cmd == "analyze":
        if len(sys.argv) < 3:
            print("Usage: python3 triage.py analyze \"message\"")
            sys.exit(1)
        result = analyze_message(sys.argv[2])
        print(json.dumps(result, indent=2, ensure_ascii=False))

    else:
        print(f"Unknown command: {cmd}")
        sys.exit(1)


if __name__ == "__main__":
    main()