runbook-tutorials/rtut/__main__.py

"""CLI: python -m rtut {list,validate,show,render} ..."""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

from . import pack as P


def _packs(root: str) -> list[P.Pack]:
    if Path(root).is_file():
        return [P.load(Path(root))]
    return [P.load(p) for p in P.discover(Path(root))]


def _run(pk, a) -> int:
    from . import run as R
    try:
        if a.action == "new":
            r = R.create(pk, a.operator, dict(kv.split("=", 1) for kv in a.param))
            print(r.data["run_id"])
            return 0
        if a.action == "list":
            print("\n".join(R.list_runs(pk)))
            return 0
        r = R.load(pk, a.run_id)
        if a.action == "status":
            for sid, st in r.data["steps"].items():
                v = st.get("verification", {})
                print(f"{st['state']:8} {sid}  {'verified' if v.get('passed') else ''}")
            return 0
        if a.action == "cmd":
            print(R.command_for(r, a.step)); return 0
        if a.action == "start":
            R.start(r, a.step); print(R.command_for(r, a.step) or ""); return 0
        if a.action == "verify":
            res = R.verify(r, a.step, confirm=a.yes, manual_ok=a.passed or None)
            print("passed" if res["passed"] else "FAILED"); return 0 if res["passed"] else 1
        if a.action == "done":
            R.complete(r, a.step, dict(kv.split("=", 1) for kv in a.evidence)); return 0
        if a.action == "fail":
            R.fail(r, a.step, a.note); return 0
        if a.action == "retry":
            R.retry(r, a.step); return 0
        if a.action == "skip":
            R.skip(r, a.step, a.note); return 0
        if a.action == "finalize":
            R.finalize(r, a.outcome or "", a.note)
            if r.data["outcome"] == "succeeded" and a.apply:
                print("recorded:", R.apply_exercise(r))
            print(r.data["outcome"]); return 0
        if a.action == "capture":
            R.add_capture(r, a.step, sys.stdin.read(), a.command); print("captured"); return 0
        if a.action == "feedback":
            R.add_feedback(r, a.rating or "", a.note); print("recorded"); return 0
        old, new = R.exercise_update(r)
        import difflib
        print("".join(difflib.unified_diff(old.splitlines(True), new.splitlines(True), "pack.yaml", "pack.yaml")))
        if a.apply:
            pk.path.write_text(new); print("written")
        return 0
    except (R.RunError, KeyError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 1


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(prog="rtut")
    sub = ap.add_subparsers(dest="cmd", required=True)
    for name in ("list", "validate"):
        sp = sub.add_parser(name)
        sp.add_argument("root", nargs="?", default=".")
    sp = sub.add_parser("run", help="record a run of a native pack")
    sp.add_argument("action", choices=["new", "list", "status", "cmd", "start", "verify", "done", "fail", "retry", "skip", "finish", "feedback", "finalize", "capture"])
    sp.add_argument("manifest")
    sp.add_argument("run_id", nargs="?")
    sp.add_argument("step", nargs="?")
    sp.add_argument("--operator", "-o", default="")
    sp.add_argument("--param", "-p", action="append", default=[], help="name=value")
    sp.add_argument("--evidence", "-e", action="append", default=[], help="name=value (non-secret)")
    sp.add_argument("--note", default="")
    sp.add_argument("--outcome", choices=["succeeded", "failed", "abandoned"], help="finalize: how the run ended")
    sp.add_argument("--command", default="", help="capture: the command you actually ran; output is read from stdin")
    sp.add_argument("--rating", choices=["good", "neutral", "poor"], help="feedback: how the runbook felt")
    sp.add_argument("--yes", action="store_true", help="confirm running a verify command")
    sp.add_argument("--pass", dest="passed", action="store_true", help="manual check: operator confirms it passed")
    sp.add_argument("--apply", action="store_true", help="finish: write the exercise-status change")
    sp = sub.add_parser("ui", help="serve the local UI on 127.0.0.1")
    sp.add_argument("roots", nargs="*", default=["."])
    sp.add_argument("--port", type=int, default=8877)
    sp = sub.add_parser("score", help="how often each runbook was tried and how it went")
    sp.add_argument("root", nargs="?", default=".")
    sp = sub.add_parser("feedback", help="list the latest feedback of every run of a pack")
    sp.add_argument("manifest")
    sp = sub.add_parser("launch"); sp.add_argument("manifest")
    sp.add_argument("--timeout", type=float, default=30.0)
    sp = sub.add_parser("show"); sp.add_argument("manifest")
    sp = sub.add_parser("render"); sp.add_argument("manifest"); sp.add_argument("step")
    sp.add_argument("--param", "-p", action="append", default=[], help="name=value")
    a = ap.parse_args(argv)

    if a.cmd == "ui":
        from . import ui
        srv = ui.serve([Path(r) for r in a.roots], a.port)
        print(f"http://127.0.0.1:{a.port}  (Ctrl-C stops)")
        try:
            srv.serve_forever()
        except KeyboardInterrupt:
            pass
        return 0
    if a.cmd == "score":
        from . import run as R
        for p in _packs(a.root):
            if p.ok and p.data.get("engine") == "native" and p.spec == "runbook-pack/v0.1":
                c = R.scorecard(p)
                print(f"{p.id}\ttried {c['tried']}\tsucceeded {c['succeeded']}\tfailed {c['failed']}\tabandoned {c['abandoned']}\topen {c['open']}\t[{p.data['exercise_status']}]")
        return 0
    if a.cmd in ("list", "validate"):
        packs = _packs(a.root)
        if not packs:
            print("no packs found", file=sys.stderr)
            return 1
        bad = 0
        for p in packs:
            d = p.data
            if a.cmd == "list":
                print(f"{p.id}\t{d.get('engine')}\t{d.get('exercise_status')}\t{d.get('title')}")
            else:
                print(f"{'ok  ' if p.ok else 'FAIL'} {p.id}")
                for e in p.errors:
                    print(f"     - {e}")
            bad += not p.ok
        return 1 if bad and a.cmd == "validate" else 0
    pk = P.load(Path(a.manifest))
    if not pk.ok:
        print("\n".join(pk.errors), file=sys.stderr)
        return 1
    if a.cmd == "feedback":
        from . import run as R
        for f in R.all_feedback(pk):
            print(f"{f['at']}  {R.RATINGS[f['rating']]}  {f['operator']}  {f['run_id']}  {f['comment']}")
        return 0
    if a.cmd == "run":
        return _run(pk, a)
    if a.cmd == "launch":
        from . import legacy
        try:
            proc = legacy.launch(pk, a.timeout)
        except (ValueError, TimeoutError) as e:
            print(f"error: {e}", file=sys.stderr)
            return 1
        print(f"running: {pk.data['legacy'].get('url', '(no url)')}  (Ctrl-C stops it)")
        try:
            return proc.wait()
        except KeyboardInterrupt:
            proc.terminate()
            return 0
    if a.cmd == "show":
        d = pk.data
        print(f"{d['title']}  [{d['exercise_status']}]  engine={d['engine']}")
        for i, s in enumerate(d.get("steps", []), 1):
            print(f"{i}. [{s['owner']}] {s['title']}  ({s.get('risk', 'read-only')})")
            print(f"     done when: {s['verify']['done_when']}")
        return 0
    values = dict(kv.split("=", 1) for kv in a.param)
    step = next((s for s in pk.data.get("steps", []) if s["id"] == a.step), None)
    if not step or "command" not in step:
        print("no such step or step has no command", file=sys.stderr)
        return 1
    try:
        print(P.render(step["command"], values, pk))
    except (KeyError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())