tegwick
eff75a303c
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
79 lines
3.5 KiB
Python
79 lines
3.5 KiB
Python
"""Latest activities across all runs and reviews, newest first. Derived from run records; nothing extra is stored."""
|
|
from __future__ import annotations
|
|
|
|
from . import maturity as M, run as R
|
|
|
|
|
|
def _step_title(run, sid: str) -> str:
|
|
try:
|
|
return R.Run.step_def(run, sid).get("title", sid)
|
|
except Exception:
|
|
return sid
|
|
|
|
|
|
def _gate_title(run, gid: str) -> str:
|
|
try:
|
|
return run.pack.data and next(g["title"] for st in _all_stages(run.pack.data["stages"]) for g in st.get("gates", []) if g["id"] == gid)
|
|
except (StopIteration, KeyError):
|
|
return gid
|
|
|
|
|
|
def _all_stages(stages):
|
|
for st in stages:
|
|
yield st
|
|
yield from _all_stages(st.get("stages", []))
|
|
|
|
|
|
def events_for(pack, pack_index: int | None) -> list[dict]:
|
|
out = []
|
|
for rid in R.list_runs(pack):
|
|
run = R.load(pack, rid)
|
|
d = run.data
|
|
link = f"/run/{pack_index}/{rid}" if pack_index is not None else None
|
|
who = d["operator"]
|
|
base = {"pack": pack.data.get("title", pack.id), "run": rid, "link": link, "project": (d.get("context") or {}).get("project")}
|
|
|
|
def ev(at, text, who=who):
|
|
if at:
|
|
out.append({**base, "at": at, "who": who, "text": text})
|
|
|
|
ev(d["created"], f"started a run ({d.get('mode', 'manual')} mode)")
|
|
for sid, st in d["steps"].items():
|
|
title = _step_title(run, sid)
|
|
if st.get("finished"):
|
|
verb = {"done": "finished", "failed": "failed", "pushed": "pushed through"}.get(st["state"])
|
|
if verb:
|
|
ev(st["finished"], f"{verb} step “{title}”")
|
|
for gid, gs in d.get("gates", {}).items():
|
|
when = max([r["at"] for r in gs.get("results", {}).values()] or [""])
|
|
if gs["state"] in ("passed", "failed") and when:
|
|
ev(when, f"gate “{_gate_title(run, gid)}” {gs['state']}")
|
|
if gs["state"] == "waived":
|
|
ev(gs["waiver"]["at"], f"waived gate “{_gate_title(run, gid)}”: {gs['waiver']['reason']}")
|
|
for dv in d.get("deviations", []):
|
|
if dv["kind"] in ("override", "changed-after-use"):
|
|
ev(dv["at"], f"{dv['kind'].replace('-', ' ')}: {dv['item']} ({dv['reason']})", dv.get("who", who))
|
|
for fb in d.get("feedback", []):
|
|
ev(fb["at"], f"rated the run {fb['rating']}" + (f": {fb['comment']}" if fb["comment"] else ""))
|
|
if d.get("outcome"):
|
|
ev(d.get("finished"), f"finished the run as {d['outcome']}")
|
|
for rv in M._reviews(pack):
|
|
out.append({"pack": pack.data.get("title", pack.id), "run": None, "link": f"/pack/{pack_index}" if pack_index is not None else None,
|
|
"project": None, "at": rv["at"], "who": rv["by"], "text": "reviewed the runbook" + (f": {rv['note']}" if rv.get("note") else "")})
|
|
return out
|
|
|
|
|
|
def project_run_events(pr) -> list[dict]:
|
|
return [{"pack": pr.data["title"], "run": None, "link": f"/project-run/{pr.id}", "project": pr.data["project"],
|
|
"at": ev["at"], "who": ev["who"], "text": "project run: " + ev["text"]} for ev in pr.data["events"]]
|
|
|
|
|
|
def latest(packs: list, limit: int = 15, project_runs: list | None = None) -> list[dict]:
|
|
events: list[dict] = []
|
|
for i, p in enumerate(packs):
|
|
if p.ok and p.data.get("engine") == "native":
|
|
events += events_for(p, i)
|
|
for pr in project_runs or []:
|
|
events += project_run_events(pr)
|
|
events.sort(key=lambda x: x["at"], reverse=True)
|
|
return events[:limit]
|