diff --git a/bin/__pycache__/nexus_watchdog.cpython-311.pyc b/bin/__pycache__/nexus_watchdog.cpython-311.pyc new file mode 100644 index 0000000..9c63227 Binary files /dev/null and b/bin/__pycache__/nexus_watchdog.cpython-311.pyc differ diff --git a/bin/ezra_weekly_report.py b/bin/ezra_weekly_report.py new file mode 100644 index 0000000..4383621 --- /dev/null +++ b/bin/ezra_weekly_report.py @@ -0,0 +1,487 @@ +#!/usr/bin/env python3 +""" +Ezra Weekly Wizard Performance Report + +Runs weekly (via cron) and reports wizard fleet performance to the +Timmy Time Telegram group. Surfaces problems before Alexander has to ask. + +Metrics reported: + - Issues opened/closed per wizard (7-day window) + - Unassigned issue count + - Overloaded wizards (>15 open assignments) + - Idle wizards (0 closes in 7 days) + +USAGE +===== + # One-shot report + python bin/ezra_weekly_report.py + + # Dry-run (print to stdout, don't send Telegram) + python bin/ezra_weekly_report.py --dry-run + + # Crontab entry (every Monday at 09:00) + 0 9 * * 1 cd /path/to/the-nexus && python bin/ezra_weekly_report.py + +ENVIRONMENT +=========== + GITEA_URL Gitea base URL (default: http://143.198.27.163:3000) + GITEA_TOKEN Gitea API token + NEXUS_REPO Repository slug (default: Timmy_Foundation/the-nexus) + TELEGRAM_BOT_TOKEN Telegram bot token for delivery + TELEGRAM_CHAT_ID Telegram chat/group ID for delivery + +ZERO DEPENDENCIES +================= +Pure stdlib. No pip installs. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import os +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)-7s %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +logger = logging.getLogger("ezra.weekly_report") + +# ── Configuration ──────────────────────────────────────────────────── + +GITEA_URL = os.environ.get("GITEA_URL", "http://143.198.27.163:3000") +GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "") +GITEA_REPO = os.environ.get("NEXUS_REPO", "Timmy_Foundation/the-nexus") +TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "") +TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "") + +OVERLOAD_THRESHOLD = 15 # open assignments above this = overloaded +WINDOW_DAYS = 7 # look-back window for opened/closed counts +PAGE_LIMIT = 50 # Gitea items per page + + +# ── Data types ──────────────────────────────────────────────────────── + +@dataclass +class WizardStats: + """Per-wizard performance data for the reporting window.""" + login: str + opened: int = 0 # issues opened in the window + closed: int = 0 # issues closed in the window + open_assignments: int = 0 # currently open issues assigned to this wizard + + @property + def is_overloaded(self) -> bool: + return self.open_assignments > OVERLOAD_THRESHOLD + + @property + def is_idle(self) -> bool: + return self.closed == 0 + + +@dataclass +class WeeklyReport: + """Aggregate weekly performance report.""" + generated_at: float + window_days: int + wizard_stats: Dict[str, WizardStats] = field(default_factory=dict) + unassigned_count: int = 0 + + @property + def overloaded(self) -> List[WizardStats]: + return [s for s in self.wizard_stats.values() if s.is_overloaded] + + @property + def idle(self) -> List[WizardStats]: + """Wizards with open assignments but zero closes in the window.""" + return [ + s for s in self.wizard_stats.values() + if s.is_idle and s.open_assignments > 0 + ] + + def to_markdown(self) -> str: + """Format the report as Telegram-friendly markdown.""" + ts = datetime.fromtimestamp(self.generated_at, tz=timezone.utc) + ts_str = ts.strftime("%Y-%m-%d %H:%M UTC") + window = self.window_days + + lines = [ + f"📊 *Ezra Weekly Wizard Report* — {ts_str}", + f"_{window}-day window_", + "", + ] + + # ── Per-wizard throughput table ────────────────────────────── + lines.append("*Wizard Throughput*") + lines.append("```") + lines.append(f"{'Wizard':<18} {'Opened':>6} {'Closed':>6} {'Open':>6}") + lines.append("-" * 40) + + sorted_wizards = sorted( + self.wizard_stats.values(), + key=lambda s: s.closed, + reverse=True, + ) + for s in sorted_wizards: + flag = " ⚠️" if s.is_overloaded else (" 💤" if s.is_idle and s.open_assignments > 0 else "") + lines.append( + f"{s.login:<18} {s.opened:>6} {s.closed:>6} {s.open_assignments:>6}{flag}" + ) + + lines.append("```") + lines.append("") + + # ── Summary ────────────────────────────────────────────────── + total_opened = sum(s.opened for s in self.wizard_stats.values()) + total_closed = sum(s.closed for s in self.wizard_stats.values()) + lines.append( + f"*Fleet totals:* {total_opened} opened · {total_closed} closed · " + f"{self.unassigned_count} unassigned" + ) + lines.append("") + + # ── Alerts ─────────────────────────────────────────────────── + alerts = [] + if self.overloaded: + names = ", ".join(s.login for s in self.overloaded) + alerts.append( + f"🔴 *Overloaded* (>{OVERLOAD_THRESHOLD} open): {names}" + ) + if self.idle: + names = ", ".join(s.login for s in self.idle) + alerts.append(f"💤 *Idle* (0 closes in {window}d): {names}") + if self.unassigned_count > 0: + alerts.append( + f"📭 *Unassigned issues:* {self.unassigned_count} waiting for triage" + ) + + if alerts: + lines.append("*Alerts*") + lines.extend(alerts) + else: + lines.append("✅ No alerts — fleet running clean.") + + lines.append("") + lines.append("_— Ezra, archivist-wizard_") + return "\n".join(lines) + + +# ── Gitea API ───────────────────────────────────────────────────────── + +def _gitea_request( + method: str, + path: str, + params: Optional[Dict[str, Any]] = None, + data: Optional[dict] = None, +) -> Any: + """Make a Gitea API request. Returns parsed JSON or None on failure.""" + url = f"{GITEA_URL.rstrip('/')}/api/v1{path}" + if params: + url = f"{url}?{urllib.parse.urlencode(params)}" + + body = json.dumps(data).encode() if data else None + req = urllib.request.Request(url, data=body, method=method) + if GITEA_TOKEN: + req.add_header("Authorization", f"token {GITEA_TOKEN}") + req.add_header("Content-Type", "application/json") + req.add_header("Accept", "application/json") + + try: + with urllib.request.urlopen(req, timeout=15) as resp: + raw = resp.read().decode() + return json.loads(raw) if raw.strip() else {} + except urllib.error.HTTPError as e: + logger.warning("Gitea HTTP %d at %s: %s", e.code, path, e.read().decode()[:200]) + return None + except Exception as e: + logger.warning("Gitea request failed (%s): %s", path, e) + return None + + +def _fetch_all_issues(state: str = "open", since: Optional[str] = None) -> List[dict]: + """Fetch all issues from the repo, paginating through results. + + Args: + state: "open" or "closed" + since: ISO 8601 timestamp — only issues updated at or after this time + """ + all_issues: List[dict] = [] + page = 1 + + while True: + params: Dict[str, Any] = { + "state": state, + "type": "issues", + "limit": PAGE_LIMIT, + "page": page, + } + if since: + params["since"] = since + + items = _gitea_request("GET", f"/repos/{GITEA_REPO}/issues", params=params) + if not items or not isinstance(items, list): + break + all_issues.extend(items) + if len(items) < PAGE_LIMIT: + break + page += 1 + + return all_issues + + +def _iso_since(days: int) -> str: + """Return an ISO 8601 timestamp for N days ago (UTC).""" + ts = time.time() - days * 86400 + dt = datetime.fromtimestamp(ts, tz=timezone.utc) + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + + +# ── Report assembly ─────────────────────────────────────────────────── + +def _collect_opened_in_window(window_days: int) -> Dict[str, int]: + """Count issues opened per wizard in the window.""" + since_str = _iso_since(window_days) + since_ts = time.time() - window_days * 86400 + + # All open issues updated since the window (may have been opened before) + all_open = _fetch_all_issues(state="open", since=since_str) + # All closed issues updated since the window + all_closed = _fetch_all_issues(state="closed", since=since_str) + + counts: Dict[str, int] = {} + + for issue in all_open + all_closed: + created_at = issue.get("created_at", "") + if not created_at: + continue + try: + dt = datetime.fromisoformat(created_at.replace("Z", "+00:00")) + created_ts = dt.timestamp() + except (ValueError, AttributeError): + continue + + if created_ts < since_ts: + continue # opened before the window + + poster = (issue.get("user") or {}).get("login", "") + if poster: + counts[poster] = counts.get(poster, 0) + 1 + + return counts + + +def _collect_closed_in_window(window_days: int) -> Dict[str, int]: + """Count issues closed per wizard (the assignee at close time).""" + since_str = _iso_since(window_days) + since_ts = time.time() - window_days * 86400 + + closed_issues = _fetch_all_issues(state="closed", since=since_str) + + counts: Dict[str, int] = {} + + for issue in closed_issues: + closed_at = issue.get("closed_at") or issue.get("updated_at", "") + if not closed_at: + continue + try: + dt = datetime.fromisoformat(closed_at.replace("Z", "+00:00")) + closed_ts = dt.timestamp() + except (ValueError, AttributeError): + continue + + if closed_ts < since_ts: + continue # closed before the window + + # Credit the assignee; fall back to issue poster + assignees = issue.get("assignees") or [] + if assignees: + for assignee in assignees: + login = (assignee or {}).get("login", "") + if login: + counts[login] = counts.get(login, 0) + 1 + else: + poster = (issue.get("user") or {}).get("login", "") + if poster: + counts[poster] = counts.get(poster, 0) + 1 + + return counts + + +def _collect_open_assignments() -> Dict[str, int]: + """Count currently open issues per assignee.""" + open_issues = _fetch_all_issues(state="open") + counts: Dict[str, int] = {} + + for issue in open_issues: + assignees = issue.get("assignees") or [] + for assignee in assignees: + login = (assignee or {}).get("login", "") + if login: + counts[login] = counts.get(login, 0) + 1 + + return counts + + +def _count_unassigned() -> int: + """Count open issues with no assignee.""" + open_issues = _fetch_all_issues(state="open") + return sum( + 1 for issue in open_issues + if not (issue.get("assignees") or []) + ) + + +def build_report(window_days: int = WINDOW_DAYS) -> WeeklyReport: + """Fetch data from Gitea and assemble the weekly report.""" + logger.info("Fetching wizard performance data (window: %d days)", window_days) + + opened = _collect_opened_in_window(window_days) + logger.info("Opened counts: %s", opened) + + closed = _collect_closed_in_window(window_days) + logger.info("Closed counts: %s", closed) + + open_assignments = _collect_open_assignments() + logger.info("Open assignments: %s", open_assignments) + + unassigned = _count_unassigned() + logger.info("Unassigned issues: %d", unassigned) + + # Merge all wizard logins into a unified stats dict + all_logins = set(opened) | set(closed) | set(open_assignments) + wizard_stats: Dict[str, WizardStats] = {} + for login in sorted(all_logins): + wizard_stats[login] = WizardStats( + login=login, + opened=opened.get(login, 0), + closed=closed.get(login, 0), + open_assignments=open_assignments.get(login, 0), + ) + + return WeeklyReport( + generated_at=time.time(), + window_days=window_days, + wizard_stats=wizard_stats, + unassigned_count=unassigned, + ) + + +# ── Telegram delivery ───────────────────────────────────────────────── + +def send_telegram(text: str, bot_token: str, chat_id: str) -> bool: + """Send a message to a Telegram chat via the Bot API. + + Returns True on success, False on failure. + """ + url = f"https://api.telegram.org/bot{bot_token}/sendMessage" + data = json.dumps({ + "chat_id": chat_id, + "text": text, + "parse_mode": "Markdown", + }).encode() + + req = urllib.request.Request(url, data=data, method="POST") + req.add_header("Content-Type", "application/json") + + try: + with urllib.request.urlopen(req, timeout=15) as resp: + raw = resp.read().decode() + result = json.loads(raw) + if result.get("ok"): + logger.info("Telegram delivery: OK (message_id=%s)", result.get("result", {}).get("message_id")) + return True + logger.error("Telegram API error: %s", result.get("description", "unknown")) + return False + except urllib.error.HTTPError as e: + logger.error("Telegram HTTP %d: %s", e.code, e.read().decode()[:200]) + return False + except Exception as e: + logger.error("Telegram delivery failed: %s", e) + return False + + +# ── CLI ─────────────────────────────────────────────────────────────── + +def main() -> None: + parser = argparse.ArgumentParser( + description="Ezra Weekly Wizard Performance Report", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Print the report to stdout instead of sending to Telegram", + ) + parser.add_argument( + "--window", + type=int, + default=WINDOW_DAYS, + help=f"Look-back window in days (default: {WINDOW_DAYS})", + ) + parser.add_argument( + "--json", + action="store_true", + dest="output_json", + help="Output report data as JSON (for integration with other tools)", + ) + args = parser.parse_args() + + if not GITEA_TOKEN and not args.dry_run: + logger.warning("GITEA_TOKEN not set — Gitea API calls will be unauthenticated") + + report = build_report(window_days=args.window) + markdown = report.to_markdown() + + if args.output_json: + data = { + "generated_at": report.generated_at, + "window_days": report.window_days, + "unassigned_count": report.unassigned_count, + "wizards": { + login: { + "opened": s.opened, + "closed": s.closed, + "open_assignments": s.open_assignments, + "overloaded": s.is_overloaded, + "idle": s.is_idle, + } + for login, s in report.wizard_stats.items() + }, + "alerts": { + "overloaded": [s.login for s in report.overloaded], + "idle": [s.login for s in report.idle], + }, + } + print(json.dumps(data, indent=2)) + return + + if args.dry_run: + print(markdown) + return + + if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: + logger.error( + "TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID must be set for delivery. " + "Use --dry-run to print without sending." + ) + sys.exit(1) + + success = send_telegram(markdown, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID) + if not success: + logger.error("Failed to deliver report to Telegram") + sys.exit(1) + + logger.info("Weekly report delivered successfully") + + +if __name__ == "__main__": + main()