Claude
754 lines
28 KiB
Python
754 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""pragent pilot — read-mostly dashboard.
|
|
|
|
Stdlib HTTP server (mirrors `webhook_server.py`'s BaseHTTPRequestHandler +
|
|
ThreadingHTTPServer shape) that renders three views off the feedback SQLite:
|
|
|
|
GET / overview
|
|
GET /r/<owner>/<name> repo summary + edit form
|
|
GET /r/<owner>/<name>/<index> one PR's findings
|
|
GET /r/<owner>/<name>/<index>/raw raw Markdown body (via Gitea)
|
|
GET /static/style.css CSS
|
|
POST /r/<owner>/<name>/edit mutate .pr-review.json (Tasks C+D)
|
|
|
|
Auth: oauth2-proxy fronts this service in-cluster. Every route except
|
|
`/static/*` requires the `X-Forwarded-User` header (set by oauth2-proxy
|
|
once the user has logged in via Logto). Missing header → 401 +
|
|
`WWW-Authenticate: Basic realm="pragent-dashboard"` so oauth2-proxy
|
|
intercepts the response.
|
|
|
|
DB: `PRAGENT_FEEDBACK_DB` points at the SQLite file the webhook server
|
|
also writes. Per-request open (SQLite is cheap, no concurrency hazard,
|
|
no stale-conn surprise after the file rotates).
|
|
|
|
All HTML is rendered via `string.Template` and every dynamic value is
|
|
escaped with `html.escape(..., quote=True)`. No `.format`, no f-string
|
|
templates — see `_render_*` for the discipline.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import datetime
|
|
import html
|
|
import json
|
|
import os
|
|
import secrets
|
|
import string
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from pilot import dashboard_data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
FEEDBACK_DB = "" # legacy; readers should call _feedback_db()
|
|
PORT = int(os.environ.get("DASHBOARD_PORT", "8081"))
|
|
|
|
GITEA_API = "" # legacy; readers should call _gitea_api()
|
|
BOT_TOKEN = "" # legacy; readers should call _bot_token()
|
|
|
|
# CSRF secret for the edit form. Regenerated per process (each Python
|
|
# interpreter launch). Behind oauth2-proxy this is enough — only an
|
|
# already-authenticated same-tab request can read this and echo it back.
|
|
_CSRF_SECRET: str = secrets.token_urlsafe(24)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lazy config readers — tests set env after import, so each request re-reads.
|
|
# Production: env is fixed for the process lifetime; the per-request lookup is
|
|
# a dict access, not a syscall.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _feedback_db() -> str:
|
|
return os.environ.get("PRAGENT_FEEDBACK_DB", "")
|
|
|
|
|
|
def _bot_token() -> str:
|
|
return os.environ.get("PRAGENT_BOT_TOKEN", "")
|
|
|
|
|
|
def _gitea_api() -> str:
|
|
return os.environ.get("GITEA_API", "http://gitea-http.gitea.svc.cluster.local:3000")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stylesheet — small, dark-mode-friendly, deliberately under 100 lines
|
|
# ---------------------------------------------------------------------------
|
|
|
|
STYLE_CSS = """
|
|
:root { color-scheme: light dark; }
|
|
* { box-sizing: border-box; }
|
|
body {
|
|
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", system-ui, sans-serif;
|
|
margin: 0; padding: 0;
|
|
background: #0f1115; color: #e6e6e6;
|
|
line-height: 1.5;
|
|
}
|
|
header {
|
|
background: #1a1d23; padding: 12px 20px;
|
|
border-bottom: 1px solid #2a2f38;
|
|
display: flex; align-items: center; gap: 18px;
|
|
}
|
|
header h1 { font-size: 18px; margin: 0; }
|
|
header nav a {
|
|
color: #8ab4f8; text-decoration: none; margin-right: 12px;
|
|
}
|
|
header nav a:hover { text-decoration: underline; }
|
|
main { padding: 20px; max-width: 1100px; margin: 0 auto; }
|
|
h2 { margin-top: 24px; font-size: 16px; color: #c9d1d9; }
|
|
.metric-row { display: flex; gap: 16px; flex-wrap: wrap; margin-bottom: 16px; }
|
|
.metric {
|
|
background: #1a1d23; padding: 14px 18px; border-radius: 8px;
|
|
min-width: 140px; border: 1px solid #2a2f38;
|
|
}
|
|
.metric .v { font-size: 28px; font-weight: 600; }
|
|
.metric .l { font-size: 12px; color: #8b949e; text-transform: uppercase; letter-spacing: 0.04em; }
|
|
table { width: 100%; border-collapse: collapse; margin: 8px 0 16px; font-size: 14px; }
|
|
th, td { text-align: left; padding: 6px 10px; border-bottom: 1px solid #2a2f38; }
|
|
th { color: #8b949e; font-weight: 500; text-transform: uppercase; font-size: 11px; letter-spacing: 0.04em; }
|
|
tr:hover td { background: #161922; }
|
|
.sev-critical { color: #ff7b72; font-weight: 600; }
|
|
.sev-high { color: #f0883e; }
|
|
.sev-medium { color: #d29922; }
|
|
.sev-low { color: #8b949e; }
|
|
.muted { color: #8b949e; font-size: 12px; }
|
|
.sparkline { font-family: ui-monospace, "SF Mono", monospace; letter-spacing: 1px; }
|
|
form { background: #1a1d23; padding: 14px 18px; border-radius: 8px; border: 1px solid #2a2f38; margin: 12px 0; }
|
|
form label { display: block; margin: 8px 0 4px; color: #c9d1d9; font-size: 13px; }
|
|
form input[type=text], form textarea, form select {
|
|
background: #0f1115; color: #e6e6e6; border: 1px solid #2a2f38;
|
|
border-radius: 4px; padding: 6px 8px; font-family: inherit; font-size: 14px;
|
|
width: 100%;
|
|
}
|
|
form textarea { min-height: 80px; }
|
|
form .row { display: flex; gap: 8px; align-items: center; margin-top: 12px; }
|
|
form button {
|
|
background: #2ea043; color: white; border: none; border-radius: 4px;
|
|
padding: 6px 14px; font-size: 14px; cursor: pointer;
|
|
}
|
|
form button:hover { background: #3fb950; }
|
|
.flash { background: #3d1e1e; color: #ff7b72; padding: 8px 12px; border-radius: 4px; margin-bottom: 12px; }
|
|
code { background: #161922; padding: 1px 4px; border-radius: 3px; font-size: 13px; }
|
|
pre { background: #161922; padding: 12px; border-radius: 6px; overflow-x: auto; }
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Templates — string.Template so dynamic values are always escaped explicitly
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_BASE = string.Template("""<!doctype html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<title>${title}</title>
|
|
<link rel="stylesheet" href="/static/style.css">
|
|
</head>
|
|
<body>
|
|
<header>
|
|
<h1>pragent dashboard</h1>
|
|
<nav>
|
|
<a href="/">Home</a>
|
|
<a href="/r/${repos_first}">repos</a>
|
|
</nav>
|
|
<span class="muted" style="margin-left:auto">${db_status}</span>
|
|
</header>
|
|
<main>
|
|
${body}
|
|
</main>
|
|
</body>
|
|
</html>""")
|
|
|
|
|
|
_OVERVIEW = string.Template("""<h2>Overview</h2>
|
|
<div class="metric-row">
|
|
<div class="metric"><div class="v">${total_reviews}</div><div class="l">reviews</div></div>
|
|
<div class="metric"><div class="v">${total_findings}</div><div class="l">findings</div></div>
|
|
<div class="metric"><div class="v">${total_repos}</div><div class="l">repos</div></div>
|
|
<div class="metric"><div class="v">${last_30d_reviews}</div><div class="l">last 30d</div></div>
|
|
</div>
|
|
|
|
<h2>Last 7 days</h2>
|
|
<div class="sparkline">${sparkline}</div>
|
|
<div class="muted">total cost: $${total_cost_usd} — no per-review cost logged</div>
|
|
|
|
<h2>Top repos</h2>
|
|
${top_repos_table}
|
|
""")
|
|
|
|
|
|
_REPO = string.Template("""<h2>Repo: <code>${repo}</code></h2>
|
|
<div class="metric-row">
|
|
<div class="metric"><div class="v">${total_runs}</div><div class="l">runs</div></div>
|
|
<div class="metric"><div class="v">${sev_critical}</div><div class="l sev-critical">critical</div></div>
|
|
<div class="metric"><div class="v">${sev_high}</div><div class="l sev-high">high</div></div>
|
|
<div class="metric"><div class="v">${sev_medium}</div><div class="l sev-medium">medium</div></div>
|
|
<div class="metric"><div class="v">${sev_low}</div><div class="l sev-low">low</div></div>
|
|
</div>
|
|
|
|
<h2>Edit .pr-review.json</h2>
|
|
${flash}
|
|
<form method="post" action="/r/${repo_url}/edit">
|
|
<input type="hidden" name="_csrf" value="${csrf}">
|
|
<label for="static_message">Static banner message (max 400 chars)</label>
|
|
<textarea id="static_message" name="static_message" maxlength="400">${current_static_message}</textarea>
|
|
<label for="model">Model (PRICES keys)</label>
|
|
<select id="model" name="model">${model_options}</select>
|
|
<div class="row">
|
|
<button type="submit">Save</button>
|
|
<span class="muted">posted via the bot identity; one commit on the base branch</span>
|
|
</div>
|
|
</form>
|
|
|
|
<h2>Top findings (by occurrence)</h2>
|
|
${top_findings_table}
|
|
|
|
<h2>Runs by day (last 30d)</h2>
|
|
${runs_by_day_table}
|
|
|
|
<h2>Reviews</h2>
|
|
${reviews_table}
|
|
""")
|
|
|
|
|
|
_PR = string.Template("""<h2>PR <code>${repo}</code> #${pr}</h2>
|
|
<div class="muted">head sha: <code>${head_sha}</code></div>
|
|
<div class="muted">posted_at: ${posted_at_iso}</div>
|
|
<div class="muted">review_id_gitea: ${review_id_gitea} · body_comment_id: ${body_comment_id}</div>
|
|
|
|
<h2>Findings</h2>
|
|
${findings_table}
|
|
|
|
<p><a href="/r/${repo_url}/${pr}/raw">raw review body (Markdown)</a></p>
|
|
""")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Small helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _esc(s) -> str:
|
|
"""HTML-escape any value to a string."""
|
|
return html.escape(str(s), quote=True)
|
|
|
|
|
|
def _ts_iso(ts: int) -> str:
|
|
if not ts:
|
|
return "—"
|
|
return datetime.datetime.fromtimestamp(int(ts), tz=datetime.timezone.utc).isoformat()
|
|
|
|
|
|
def _sparkline(buckets: list[dict]) -> str:
|
|
"""7-bucket sparkline as unicode bars."""
|
|
bars = "▁▂▃▄▅▆▇█"
|
|
if not buckets:
|
|
return ""
|
|
mx = max((b.get("count", 0) for b in buckets), default=0) or 1
|
|
out = []
|
|
for b in buckets:
|
|
n = b.get("count", 0)
|
|
idx = min(len(bars) - 1, int(round(n / mx * (len(bars) - 1))))
|
|
out.append(bars[idx])
|
|
return "".join(out)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Renderers — one per page
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _overview_body(data: dict) -> str:
|
|
top_rows = "".join(
|
|
f"<tr><td><a href=\"/r/{_esc(r['repo'])}\">{_esc(r['repo'])}</a></td>"
|
|
f"<td>{int(r['run_count'])}</td>"
|
|
f"<td class=\"muted\">{_ts_iso(int(r['last_seen']))}</td></tr>"
|
|
for r in data.get("top_repos", [])
|
|
) or "<tr><td class=\"muted\">no reviews yet</td></tr>"
|
|
top_table = f"<table><thead><tr><th>repo</th><th>runs</th><th>last seen</th></tr></thead><tbody>{top_rows}</tbody></table>"
|
|
return _OVERVIEW.substitute(
|
|
total_reviews=_esc(data.get("total_reviews", 0)),
|
|
total_findings=_esc(data.get("total_findings", 0)),
|
|
total_repos=_esc(data.get("total_repos", 0)),
|
|
last_30d_reviews=_esc(data.get("last_30d_reviews", 0)),
|
|
sparkline=_esc(_sparkline(data.get("daily", []))),
|
|
total_cost_usd=f"{float(data.get('total_cost_usd', 0.0)):.2f}",
|
|
top_repos_table=top_table,
|
|
)
|
|
|
|
|
|
def _repo_body(data: dict, *, repo_url: str, csrf: str, current_model: str,
|
|
current_static_message: str, flash: str = "") -> str:
|
|
fbs = data.get("findings_by_severity", {})
|
|
tf = data.get("top_findings", [])
|
|
|
|
# Top findings table.
|
|
if tf:
|
|
rows = "".join(
|
|
f"<tr><td><code>{_esc(f['path'])}:{_esc(f['line'])}</code></td>"
|
|
f"<td class=\"sev-{_esc(f.get('severity', 'low').lower())}\">{_esc(f.get('severity', ''))}</td>"
|
|
f"<td>{_esc(f.get('problem', ''))}</td>"
|
|
f"<td>{int(f.get('occurrences', 0))}</td>"
|
|
f"<td>+{int(f.get('upvotes', 0))} / -{int(f.get('downvotes', 0))}</td>"
|
|
f"<td>{'resolved' if int(f.get('resolved', 0)) else 'open'}</td>"
|
|
f"<td>{int(f.get('reply_count', 0))}</td></tr>"
|
|
for f in tf
|
|
)
|
|
top_findings_table = (
|
|
"<table><thead><tr><th>location</th><th>severity</th>"
|
|
"<th>problem</th><th>occurrences</th><th>votes</th>"
|
|
"<th>state</th><th>replies</th></tr></thead><tbody>"
|
|
f"{rows}</tbody></table>"
|
|
)
|
|
else:
|
|
top_findings_table = "<p class=\"muted\">no findings yet</p>"
|
|
|
|
# Runs by day.
|
|
runs = data.get("runs_by_day", [])
|
|
if runs:
|
|
rows = "".join(
|
|
f"<tr><td>{_esc(r['date'])}</td><td>{int(r.get('count', 0))}</td></tr>"
|
|
for r in runs
|
|
)
|
|
runs_by_day_table = (
|
|
"<table><thead><tr><th>date</th><th>runs</th></tr></thead>"
|
|
f"<tbody>{rows}</tbody></table>"
|
|
)
|
|
else:
|
|
runs_by_day_table = "<p class=\"muted\">no runs in the last 30 days</p>"
|
|
|
|
# Reviews list — derived from finding timestamps; cheap because we
|
|
# just enumerate the repo's review rows.
|
|
reviews_table = _repo_reviews_table(repo_url, data.get("recent_reviews", []))
|
|
|
|
# Model select (Task D) — sorted PRICES keys + "keep current".
|
|
from cost_model import PRICES # local: pilot-only dep
|
|
model_options = (
|
|
f"<option value=\"\">— keep current ({_esc(current_model or 'unset')}) —</option>"
|
|
+ "".join(
|
|
f"<option value=\"{_esc(k)}\" {'selected' if k == current_model else ''}>{_esc(k)}</option>"
|
|
for k in sorted(PRICES)
|
|
)
|
|
)
|
|
|
|
return _REPO.substitute(
|
|
repo=_esc(data.get("repo", "")),
|
|
repo_url=_esc(repo_url),
|
|
total_runs=_esc(data.get("total_runs", 0)),
|
|
sev_critical=_esc(fbs.get("critical", 0)),
|
|
sev_high=_esc(fbs.get("high", 0)),
|
|
sev_medium=_esc(fbs.get("medium", 0)),
|
|
sev_low=_esc(fbs.get("low", 0)),
|
|
csrf=_esc(csrf),
|
|
current_static_message=_esc(current_static_message),
|
|
model_options=model_options,
|
|
flash=_esc(flash),
|
|
top_findings_table=top_findings_table,
|
|
runs_by_day_table=runs_by_day_table,
|
|
reviews_table=reviews_table,
|
|
)
|
|
|
|
|
|
def _repo_reviews_table(repo_url: str, rows: list[dict]) -> str:
|
|
if not rows:
|
|
return "<p class=\"muted\">no reviews yet</p>"
|
|
out = "<table><thead><tr><th>PR</th><th>head sha</th><th>posted</th></tr></thead><tbody>"
|
|
for r in rows:
|
|
out += (
|
|
f"<tr><td><a href=\"/r/{_esc(repo_url)}/{int(r['pr'])}\">#{int(r['pr'])}</a></td>"
|
|
f"<td><code>{_esc(r['head_sha'][:10])}</code></td>"
|
|
f"<td class=\"muted\">{_ts_iso(int(r.get('posted_at', 0)))}</td></tr>"
|
|
)
|
|
out += "</tbody></table>"
|
|
return out
|
|
|
|
|
|
def _pr_body(data: dict, *, repo_url: str) -> str:
|
|
findings = data.get("findings", [])
|
|
if findings:
|
|
rows = "".join(
|
|
f"<tr><td><code>{_esc(f['path'])}:{_esc(f['line'])}</code></td>"
|
|
f"<td class=\"sev-{_esc(f.get('severity', 'low').lower())}\">{_esc(f.get('severity', ''))}</td>"
|
|
f"<td>{_esc(f.get('problem', ''))}</td>"
|
|
f"<td>{_esc(f.get('fix', ''))}</td>"
|
|
f"<td>{_esc(f.get('suggestion', ''))}</td>"
|
|
f"<td>+{int(f.get('upvotes', 0))} / -{int(f.get('downvotes', 0))}</td>"
|
|
f"<td>{'resolved' if int(f.get('resolved', 0)) else 'open'}</td>"
|
|
f"<td>{int(f.get('reply_count', 0))}</td></tr>"
|
|
for f in findings
|
|
)
|
|
findings_table = (
|
|
"<table><thead><tr><th>location</th><th>severity</th>"
|
|
"<th>problem</th><th>fix</th><th>suggestion</th>"
|
|
"<th>votes</th><th>state</th><th>replies</th></tr></thead>"
|
|
f"<tbody>{rows}</tbody></table>"
|
|
)
|
|
else:
|
|
findings_table = "<p class=\"muted\">no findings</p>"
|
|
|
|
return _PR.substitute(
|
|
repo=_esc(data.get("repo", "")),
|
|
repo_url=_esc(repo_url),
|
|
pr=_esc(data.get("pr", 0)),
|
|
head_sha=_esc(data.get("head_sha", "")),
|
|
posted_at_iso=_ts_iso(int(data.get("posted_at", 0))),
|
|
review_id_gitea=_esc(data.get("review_id_gitea", "") or "—"),
|
|
body_comment_id=_esc(data.get("body_comment_id", "") or "—"),
|
|
findings_table=findings_table,
|
|
)
|
|
|
|
|
|
def _page(title: str, body: str, *, repos_first: str = "") -> str:
|
|
db_status = _feedback_db() or "(no DB configured)"
|
|
return _BASE.substitute(
|
|
title=_esc(title),
|
|
body=body,
|
|
repos_first=_esc(repos_first),
|
|
db_status=_esc(db_status),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gitea HTTP helper — minimal, used by the raw body fetch and the edit endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _http(method: str, url: str, *, token: str = "", body: dict | None = None,
|
|
raw_body: bytes | None = None) -> tuple[int, bytes]:
|
|
"""Like ai_review._http but local: this module is stdlib-only and doesn't
|
|
depend on the ai_review import (which pulls in a 1700-line reviewer)."""
|
|
headers = {"Accept": "application/json"}
|
|
data: bytes | None = None
|
|
if raw_body is not None:
|
|
data = raw_body
|
|
headers["Content-Type"] = "application/json"
|
|
elif body is not None:
|
|
data = json.dumps(body).encode()
|
|
headers["Content-Type"] = "application/json"
|
|
if token:
|
|
headers["Authorization"] = f"token {token}"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return r.status, r.read()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read()
|
|
except urllib.error.URLError as e:
|
|
raise RuntimeError(f"network error: {e.reason}") from e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _is_authed(headers) -> bool:
|
|
"""True when oauth2-proxy forwarded a verified user.
|
|
|
|
oauth2-proxy sets `X-Forwarded-User` (and friends) only after a
|
|
successful Logto login + email allowlist check. Unauthenticated
|
|
requests never see the header, so the dashboard never has to know
|
|
about cookies, secrets, or Logto's token shape.
|
|
"""
|
|
return bool((headers.get("X-Forwarded-User") or "").strip())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _route_overview() -> bytes:
|
|
data = dashboard_data.overview(_feedback_db())
|
|
body = _overview_body(data)
|
|
# nav: first repo if any
|
|
repos_first = ""
|
|
if data.get("top_repos"):
|
|
repos_first = data["top_repos"][0]["repo"]
|
|
return _page("Overview", body, repos_first=repos_first).encode()
|
|
|
|
|
|
def _route_repo(owner: str, name: str) -> bytes:
|
|
repo_url = f"{owner}/{name}"
|
|
data = dashboard_data.repo_summary(_feedback_db(), repo_url)
|
|
# Pull current .pr-review.json (best-effort) so the form fields prefill.
|
|
current_static_message, current_model, flash = "", "", ""
|
|
cfg, err = _fetch_pr_review_json(repo_url)
|
|
if cfg:
|
|
current_static_message = cfg.get("static_message", "")
|
|
current_model = cfg.get("model", "")
|
|
elif err and err != "404":
|
|
flash = f"could not read .pr-review.json: {err}"
|
|
body = _repo_body(
|
|
data,
|
|
repo_url=repo_url,
|
|
csrf=_CSRF_SECRET,
|
|
current_model=current_model,
|
|
current_static_message=current_static_message,
|
|
flash=flash,
|
|
)
|
|
return _page(f"repo {repo_url}", body, repos_first=repo_url).encode()
|
|
|
|
|
|
def _route_pr(owner: str, name: str, index: int) -> bytes:
|
|
repo_url = f"{owner}/{name}"
|
|
data = dashboard_data.pr_summary(_feedback_db(), repo_url, int(index))
|
|
body = _pr_body(data, repo_url=repo_url)
|
|
return _page(f"PR {repo_url}#{index}", body, repos_first=repo_url).encode()
|
|
|
|
|
|
def _route_pr_raw(owner: str, name: str, index: int) -> tuple[int, bytes]:
|
|
repo_url = f"{owner}/{name}"
|
|
data = dashboard_data.pr_summary(_feedback_db(), repo_url, int(index))
|
|
body_comment_id = data.get("body_comment_id")
|
|
if not body_comment_id:
|
|
return 404, b"no body_comment_id"
|
|
status, raw = _http(
|
|
"GET",
|
|
f"{_gitea_api()}/api/v1/repos/{repo_url}/issues/{index}/comments/{body_comment_id}",
|
|
token=_bot_token(),
|
|
)
|
|
if status != 200:
|
|
return 404, f"Gitea returned {status}".encode()
|
|
try:
|
|
parsed = json.loads(raw)
|
|
md = parsed.get("body", "")
|
|
except (json.JSONDecodeError, ValueError):
|
|
return 404, b"could not parse Gitea response"
|
|
return 200, md.encode()
|
|
|
|
|
|
def _route_static_css() -> bytes:
|
|
return STYLE_CSS.encode()
|
|
|
|
|
|
def _route_edit(owner: str, name: str, form: dict) -> tuple[int, dict, bytes]:
|
|
"""Mutate .pr-review.json via the Gitea contents API (Tasks C+D)."""
|
|
repo_url = f"{owner}/{name}"
|
|
csrf = form.get("_csrf", "")
|
|
if csrf != _CSRF_SECRET:
|
|
return 302, {"Location": f"/r/{repo_url}"}, b""
|
|
static_message = (form.get("static_message") or "").strip()[:400]
|
|
model = (form.get("model") or "").strip()
|
|
|
|
# Validate model against PRICES.
|
|
from cost_model import PRICES
|
|
if model and model not in PRICES:
|
|
flash = urllib.parse.quote(f"unknown model {model!r}; not saved")
|
|
return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b""
|
|
|
|
cfg, err = _fetch_pr_review_json(repo_url)
|
|
if err and err != "404":
|
|
flash = urllib.parse.quote(f"could not read .pr-review.json: {err}")
|
|
return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b""
|
|
if cfg is None:
|
|
cfg = {}
|
|
|
|
if static_message:
|
|
cfg["static_message"] = static_message
|
|
elif "static_message" in cfg and not static_message:
|
|
# Empty submission clears the banner.
|
|
del cfg["static_message"]
|
|
if model:
|
|
cfg["model"] = model
|
|
elif "model" in cfg and not model:
|
|
del cfg["model"]
|
|
|
|
payload = json.dumps(cfg, indent=2, sort_keys=True).encode()
|
|
b64 = base64.b64encode(payload).decode()
|
|
body = {"content": b64, "message": "pragent dashboard: update .pr-review.json"}
|
|
if err == "404":
|
|
# File didn't exist — Gitea contents PUT still creates the file when
|
|
# `sha` is omitted, but only on certain versions; passing sha=None is
|
|
# safer.
|
|
pass
|
|
else:
|
|
# GET returned a sha — include it so Gitea enforces optimistic lock.
|
|
# The sha lives in cfg's wrapper: re-fetch once to capture it.
|
|
_, raw = _http(
|
|
"GET",
|
|
f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json",
|
|
token=_bot_token(),
|
|
)
|
|
try:
|
|
existing = json.loads(raw)
|
|
sha = existing.get("sha")
|
|
if sha:
|
|
body["sha"] = sha
|
|
except (json.JSONDecodeError, ValueError):
|
|
pass
|
|
|
|
status, _ = _http(
|
|
"PUT",
|
|
f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json",
|
|
token=_bot_token(),
|
|
body=body,
|
|
)
|
|
if status not in (200, 201):
|
|
flash = urllib.parse.quote(f"Gitea PUT failed: status {status}")
|
|
return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b""
|
|
return 302, {"Location": f"/r/{repo_url}"}, b""
|
|
|
|
|
|
def _fetch_pr_review_json(repo_url: str) -> tuple[dict | None, str | None]:
|
|
"""Return (cfg, None) on success, (None, None) when the file doesn't exist,
|
|
(None, 'reason') on error."""
|
|
if not _bot_token():
|
|
return None, "PRAGENT_BOT_TOKEN not set"
|
|
status, raw = _http(
|
|
"GET",
|
|
f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json",
|
|
token=_bot_token(),
|
|
)
|
|
if status == 404:
|
|
return None, "404"
|
|
if status != 200:
|
|
return None, f"status {status}"
|
|
try:
|
|
wrapper = json.loads(raw)
|
|
content_b64 = wrapper.get("content", "").replace("\n", "")
|
|
decoded = base64.b64decode(content_b64).decode("utf-8", errors="replace")
|
|
cfg = json.loads(decoded)
|
|
except (json.JSONDecodeError, ValueError) as e:
|
|
return None, f"parse error: {e}"
|
|
if not isinstance(cfg, dict):
|
|
return None, "not a JSON object"
|
|
return cfg, None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Handler
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def _send(self, status: int, body: bytes, *, content_type: str = "text/html; charset=utf-8",
|
|
extra_headers: dict | None = None) -> None:
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
if extra_headers:
|
|
for k, v in extra_headers.items():
|
|
self.send_header(k, v)
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _redirect(self, location: str) -> None:
|
|
body = b""
|
|
self.send_response(302)
|
|
self.send_header("Location", location)
|
|
self.send_header("Content-Length", "0")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _unauthorized(self) -> None:
|
|
"""401 + Basic challenge so oauth2-proxy intercepts and redirects to Logto."""
|
|
body = b"unauthorized\n"
|
|
self.send_response(401)
|
|
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("WWW-Authenticate", 'Basic realm="pragent-dashboard"')
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
# --- GET -----------------------------------------------------------------
|
|
|
|
def do_GET(self):
|
|
path = self.path
|
|
# Static is exempt from auth (also unauthenticated browser fingerprinting
|
|
# noise, but it's the same CSS regardless of viewer).
|
|
if path == "/static/style.css":
|
|
self._send(200, _route_static_css(), content_type="text/css; charset=utf-8")
|
|
return
|
|
if not _is_authed(self.headers):
|
|
self._unauthorized()
|
|
return
|
|
|
|
if path == "/" or path == "":
|
|
self._send(200, _route_overview())
|
|
return
|
|
|
|
# /r/<owner>/<name> → repo
|
|
# /r/<owner>/<name>/<index> → PR
|
|
# /r/<owner>/<name>/<index>/raw → raw Markdown
|
|
m = _REPO_PR_RAW_RE.match(path)
|
|
if m:
|
|
owner, name, idx, raw = m.group(1), m.group(2), m.group(3), m.group(4)
|
|
if raw:
|
|
status, body = _route_pr_raw(owner, name, int(idx))
|
|
self._send(status, body,
|
|
content_type="text/plain; charset=utf-8" if status == 200 else "text/plain")
|
|
return
|
|
if idx:
|
|
self._send(200, _route_pr(owner, name, int(idx)))
|
|
return
|
|
self._send(200, _route_repo(owner, name))
|
|
return
|
|
|
|
self._send(404, b"not found", content_type="text/plain")
|
|
|
|
# --- POST ----------------------------------------------------------------
|
|
|
|
def do_POST(self):
|
|
path = self.path
|
|
if not _is_authed(self.headers):
|
|
self._unauthorized()
|
|
return
|
|
# /r/<owner>/<name>/edit
|
|
m = _EDIT_RE.match(path)
|
|
if m:
|
|
owner, name = m.group(1), m.group(2)
|
|
length = int(self.headers.get("Content-Length", "0") or "0")
|
|
raw = self.rfile.read(length) if length else b""
|
|
form = urllib.parse.parse_qs(raw.decode("utf-8", errors="replace"))
|
|
# Collapse lists to single values.
|
|
form_single = {k: v[0] for k, v in form.items()}
|
|
status, extra, body = _route_edit(owner, name, form_single)
|
|
self._send(status, body, content_type="text/plain", extra_headers=extra)
|
|
return
|
|
self._send(404, b"not found", content_type="text/plain")
|
|
|
|
def log_message(self, fmt, *args):
|
|
print(f"pragent-dashboard: {self.address_string()} {fmt % args}", flush=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routing regexes (compiled at import time)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import re # noqa: E402
|
|
|
|
_REPO_PR_RAW_RE = re.compile(
|
|
r"^/r/([^/]+)/([^/]+)(?:/(\d+)(?:/(raw))?)?/?$"
|
|
)
|
|
_EDIT_RE = re.compile(r"^/r/([^/]+)/([^/]+)/edit/?$")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def main() -> int:
|
|
if not _feedback_db():
|
|
print("pragent-dashboard: WARNING: PRAGENT_FEEDBACK_DB not set; dashboard will be empty",
|
|
flush=True)
|
|
print("pragent-dashboard: auth via oauth2-proxy (X-Forwarded-User required)", flush=True)
|
|
server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
|
|
print(f"pragent-dashboard: listening on :{PORT}", flush=True)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main()) |