risk-nexus/tools/risk.py

#!/usr/bin/env python3
"""Read-only query CLI over the register: list, show, waits, escalations, policies.

    python3 tools/risk.py list [--all] [--severity high] [--system qonto-assistant]
    python3 tools/risk.py show RISK-F-0012
    python3 tools/risk.py waits
    python3 tools/risk.py escalations
    python3 tools/risk.py policies

The verbs are generic on purpose (list/show over records), so that a future
common nexus interface can map onto them. Writing stays with record_check.py.
"""
from __future__ import annotations

import argparse
import sys

import register_lib as lib

SHOW_FIELDS = ("title", "status", "severity", "impact", "likelihood", "disclosure",
               "system", "fix_owner", "fix_tracking", "closure_condition", "escalation",
               "escalation_status", "cadence", "next_check", "ruling")


def cmd_list(a) -> int:
    fs = lib.findings()
    if not a.all:
        fs = [f for f in fs if lib.watched(f.get("status"))]
    if a.severity:
        fs = [f for f in fs if f.get("severity") == a.severity]
    if a.system:
        fs = [f for f in fs if f.get("system") == a.system]
    fs.sort(key=lambda f: (lib.sev_rank(f.get("severity")), f["id"]))
    if not fs:
        print("no matching findings")
    for f in fs:
        print(f"{f['id']}  {str(f.get('severity')):8} {str(f.get('status')):9} "
              f"{str(f.get('fix_owner')):20} {f.get('title')}")
    return 0


def cmd_show(a) -> int:
    rec = next((r for r in lib.findings() + lib.regulatory() if r.get("id") == a.id), None)
    if rec is None:
        print(f"{a.id}: not found")
        return 1
    print(f"{a.id}  ({rec['_path'].relative_to(lib.REPO)})")
    for key in SHOW_FIELDS:
        if rec.get(key) not in (None, ""):
            print(f"  {key:18} {rec[key]}")
    for w in rec.get("waiting_on") or []:
        print(f"  waiting on {w.get('who')} until {w.get('default_at')}: {w.get('what')}")
    return 0


def cmd_waits(_a) -> int:
    rows = []
    for r in lib.records_with_waits():
        for w in r["waiting_on"]:
            rows.append((str(w.get("default_at")), r.get("id"), w.get("who"), w.get("what")))
    for when, rid, who, what in sorted(rows):
        print(f"{when}  {rid:14} {who:20} {what}")
    if not rows:
        print("nobody is being waited on")
    return 0


def cmd_escalations(_a) -> int:
    rows = [f for f in lib.findings() if f.get("escalation") not in (None, "none")]
    for f in rows:
        print(f"{f['id']}  trigger {f.get('escalation_trigger')}  {f.get('escalation_status')}"
              f"  sent {f.get('escalation_sent', '-')}  {f.get('title')}")
    if not rows:
        print("no escalations recorded")
    return 0


def cmd_policies(_a) -> int:
    for r in lib.regulatory():
        print(f"{r.get('id'):14} {str(r.get('status')):22} next {str(r.get('next_check'))[:16]:16} "
              f"{r.get('title')}")
    return 0


def main(argv: list[str]) -> int:
    p = argparse.ArgumentParser(prog="risk", description=__doc__.splitlines()[0])
    sub = p.add_subparsers(dest="cmd", required=True)
    ls = sub.add_parser("list", help="findings, most severe first (live ones unless --all)")
    ls.add_argument("--all", action="store_true", help="include fixed and withdrawn")
    ls.add_argument("--severity", choices=lib.SEVERITIES)
    ls.add_argument("--system")
    ls.set_defaults(fn=cmd_list)
    sh = sub.add_parser("show", help="one finding or regulatory record")
    sh.add_argument("id")
    sh.set_defaults(fn=cmd_show)
    for name, fn, hlp in (("waits", cmd_waits, "everything owed to or by the register, by deadline"),
                          ("escalations", cmd_escalations, "escalations and their delivery state"),
                          ("policies", cmd_policies, "regulatory records and full policies")):
        sub.add_parser(name, help=hlp).set_defaults(fn=fn)
    a = p.parse_args(argv)
    return a.fn(a)


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))