cc-ci-orchestrator/cc-ci-plan/recipe-report.py

#!/usr/bin/env python3
"""recipe-report — data + HTML helper for the weekly "Recipe Report" (/recipe-report skill).

Three subcommands; the /recipe-report agent runs them around its own review/classification:
  survey [DATE]            print structured JSON of the run + every recipe's open PRs + CI verdict
                           (DATE defaults to today UTC; reads the dated /upgrade-all summary if present)
  render SPEC.json OUT.html   render the agent's report spec -> a self-contained HTML page
  publish OUT.html DATE    copy the page to cc-ci:/var/lib/cc-ci-reports/week-DATE.html and regen index

The agent: `survey` -> review the data, write a SPEC (see SPEC SHAPE below) -> `render` -> `publish`.

SPEC SHAPE (the agent writes this JSON):
  {"date":"YYYY-MM-DD","subtitle":"Week of <human date>","headline":"one line",
   "needs_attention":[{"title":"...","body":"<plain text>","links":[{"text":"PR #4","url":"..."}]}],
   "routine":[ {same shape} ],
   "table":[{"recipe":"x","change":"a → b","status":"GREEN","ci":"level 8 ✓","ci_url":"...",
             "pr":"#4","pr_url":"...","notes":"..."}]}
PUBLIC PAGE — include only public-safe data (no secrets/tokens/raw logs).
"""
import html, json, os, subprocess, sys, urllib.request, urllib.parse
from datetime import datetime, timezone

LOGDIR = "/srv/cc-ci/.cc-ci-logs"
TESTENV = "/srv/cc-ci/.testenv"
INFRA = {"cc-ci", "cc-ci-orchestrator", "cc-ci-secrets"}
HOST_REPORTS = "/var/lib/cc-ci-reports"


def _env():
    e = {}
    try:
        for ln in open(TESTENV):
            ln = ln.strip()
            if "=" in ln and not ln.startswith("#"):
                k, v = ln.split("=", 1)
                e[k] = v.strip().strip('"').strip("'")
    except FileNotFoundError:
        pass
    return e


def _gitea_get(path, env):
    url = f"https://{env['GITEA_URL']}/api/v1{path}"
    auth = "%s:%s" % (env["GITEA_USERNAME"], env["GITEA_PASSWORD"])
    import base64
    req = urllib.request.Request(url, headers={"Authorization": "Basic " + base64.b64encode(auth.encode()).decode()})
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            return json.load(r)
    except Exception:
        return None


def survey(date):
    env = _env()
    out = {"date": date, "generated": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")}
    # The /upgrade-all dated summary (raw markdown; the agent reads its narrative).
    sm = os.path.join(LOGDIR, "upgrades", f"upgrade-all-{date}.md")
    out["upgrade_summary_md"] = open(sm).read() if os.path.exists(sm) else None
    # Every recipe mirror -> open PRs + the cc-ci/testme verdict on each PR head.
    repos = _gitea_get("/orgs/recipe-maintainers/repos?limit=100", env) or []
    recipes = sorted(r["name"] for r in repos if r["name"] not in INFRA and not r["name"].startswith("archived-"))
    out["recipes"] = []
    for r in recipes:
        prs = _gitea_get(f"/repos/recipe-maintainers/{r}/pulls?state=open&limit=50", env) or []
        rec = {"recipe": r, "open_prs": []}
        for p in prs:
            sha = p["head"]["sha"]
            st = _gitea_get(f"/repos/recipe-maintainers/{r}/commits/{sha}/status", env) or {}
            ci = {"state": st.get("state")}
            for s in (st.get("statuses") or []):
                if s.get("context") == "cc-ci/testme":
                    ci = {"state": s.get("state"), "url": s.get("target_url"), "desc": s.get("description")}
            rec["open_prs"].append({"number": p["number"], "title": p["title"],
                                    "url": p["html_url"], "head": sha[:10], "ci": ci})
        out["recipes"].append(rec)
    print(json.dumps(out, indent=2))


# ---------- HTML rendering ----------
CSS = """
:root{--fg:#1a1a1a;--mut:#666;--bd:#e3e3e3;--bg:#fafafa;--card:#fff;--accent:#c9760a;--red:#c0392b;--green:#218a4e}
*{box-sizing:border-box}body{font:16px/1.6 system-ui,-apple-system,Segoe UI,sans-serif;color:var(--fg);
background:var(--bg);margin:0;padding:2.5rem 1rem 4rem}.wrap{max-width:56rem;margin:0 auto}
h1{font-size:2.4rem;margin:0;letter-spacing:-.02em}.sub{color:var(--mut);font-size:1.15rem;margin:.2rem 0 .2rem}
.headline{font-size:1.1rem;margin:.6rem 0 2rem;padding:.7rem 1rem;background:#fff7ec;border-left:4px solid var(--accent);border-radius:4px}
h2{font-size:1.45rem;margin:2.4rem 0 .8rem;border-bottom:2px solid var(--bd);padding-bottom:.3rem}
.item{background:var(--card);border:1px solid var(--bd);border-radius:8px;padding:.9rem 1.1rem;margin:.7rem 0}
.item.attn{border-left:4px solid var(--red)}.item .t{font-weight:600}.item .b{color:#333;margin:.25rem 0 .4rem}
.links a{margin-right:.8rem;font-size:.92rem}a{color:#0b62c4;text-decoration:none}a:hover{text-decoration:underline}
table{border-collapse:collapse;width:100%;font-size:.93rem;margin-top:.6rem}
th,td{text-align:left;padding:.5rem .6rem;border-bottom:1px solid var(--bd);vertical-align:top}
th{color:var(--mut);font-weight:600;font-size:.82rem;text-transform:uppercase;letter-spacing:.03em}
.s-GREEN{color:var(--green);font-weight:600}.s-RED,.s-FAILED{color:var(--red);font-weight:600}
.s-STALE{color:var(--accent);font-weight:600}.s-SKIPPED,.s-UPTODATE{color:var(--mut)}
footer{margin-top:3rem;padding-top:1rem;border-top:1px solid var(--bd);color:var(--mut);font-size:.9rem}
.idx li{margin:.4rem 0}.idx .d{color:var(--mut);font-size:.9rem}
"""


def _esc(s):
    return html.escape(str(s or ""))


def _links(links):
    if not links:
        return ""
    a = " · ".join(f'<a href="{_esc(l["url"])}">{_esc(l["text"])}</a>' for l in links)
    return f'<div class="links">{a}</div>'


def _items(items, attn=False):
    if not items:
        return "<p><em>None.</em></p>"
    out = []
    for it in items:
        out.append(
            f'<div class="item{" attn" if attn else ""}">'
            f'<div class="t">{_esc(it.get("title"))}</div>'
            f'<div class="b">{_esc(it.get("body"))}</div>{_links(it.get("links"))}</div>')
    return "\n".join(out)


def _table(rows):
    if not rows:
        return ""
    head = "<tr><th>Recipe</th><th>Change</th><th>Status</th><th>CI</th><th>PR</th><th>Notes</th></tr>"
    trs = []
    for r in rows:
        scls = "s-" + str(r.get("status", "")).upper().replace("-", "").replace(" ", "")
        ci = _esc(r.get("ci"))
        if r.get("ci_url"):
            ci = f'<a href="{_esc(r["ci_url"])}">{ci}</a>'
        pr = _esc(r.get("pr"))
        if r.get("pr_url"):
            pr = f'<a href="{_esc(r["pr_url"])}">{pr}</a>'
        trs.append(f"<tr><td>{_esc(r.get('recipe'))}</td><td>{_esc(r.get('change'))}</td>"
                   f'<td class="{scls}">{_esc(r.get("status"))}</td><td>{ci}</td><td>{pr}</td>'
                   f"<td>{_esc(r.get('notes'))}</td></tr>")
    return f"<table>{head}{''.join(trs)}</table>"


def _page(title, body):
    return (f'<!doctype html><html lang="en"><head><meta charset="utf-8">'
            f'<meta name="viewport" content="width=device-width,initial-scale=1">'
            f"<title>{_esc(title)}</title><style>{CSS}</style></head><body><div class=wrap>{body}</div></body></html>")


def render(spec_path, out_path):
    s = json.load(open(spec_path))
    gen = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
    body = (
        f'<h1>🌻 The Recipe Report</h1>'
        f'<div class="sub">{_esc(s.get("subtitle", "Week of " + s["date"]))}</div>'
        f'<div class="headline">{_esc(s.get("headline"))}</div>'
        f'<h2>⚑ Needs attention</h2>{_items(s.get("needs_attention"), attn=True)}'
        f'<h2>Routine</h2>{_items(s.get("routine"))}'
        f'<h2>All recipes</h2>{_table(s.get("table"))}'
        f'<footer>Generated {gen} · '
        f'<a href="https://ci.commoninternet.net/">dashboard</a> · '
        f'<a href="./">all reports</a></footer>')
    open(out_path, "w").write(_page("The Recipe Report — " + s["date"], body))
    print("wrote", out_path)


def publish(html_path, date):
    # Copy the page to the host, then rebuild index.html from the host's week-*.html list.
    page = f"week-{date}.html"
    data = open(html_path, "rb").read()
    subprocess.run(["ssh", "cc-ci", f"cat > {HOST_REPORTS}/{page}"], input=data, check=True)
    # Rebuild index: list week-*.html, newest first, with each page's <title>/subtitle hint = date.
    listing = subprocess.run(["ssh", "cc-ci", f"ls -1 {HOST_REPORTS}/week-*.html 2>/dev/null"],
                             capture_output=True, text=True).stdout.split()
    dates = sorted({os.path.basename(p)[5:-5] for p in listing}, reverse=True)
    lis = "\n".join(f'<li><a href="week-{d}.html">Week of {d}</a> <span class="d">— {d}</span></li>' for d in dates)
    idx = _page("The Recipe Report", f'<h1>🌻 The Recipe Report</h1>'
                f'<div class="sub">Weekly review of Co-op Cloud recipe upgrades + CI</div>'
                f'<ul class="idx">{lis or "<li><em>No reports yet.</em></li>"}</ul>')
    subprocess.run(["ssh", "cc-ci", f"cat > {HOST_REPORTS}/index.html"], input=idx.encode(), check=True)
    print(f"published https://report.ci.commoninternet.net/{page}  (+ index)")


def main():
    a = sys.argv[1:]
    cmd = a[0] if a else "survey"
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    if cmd == "survey":
        survey(a[1] if len(a) > 1 else today)
    elif cmd == "render":
        render(a[1], a[2])
    elif cmd == "publish":
        publish(a[1], a[2])
    else:
        print(__doc__); sys.exit(2)


if __name__ == "__main__":
    main()