WIP: Add GitHub notification summary Cursor command

.cursor/commands/fetch_notifications.py

@@ -0,0 +1,387 @@
+#!/usr/bin/env python3
+#   Copyright 2026 - 2026 The PyMC Labs Developers
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+#!/usr/bin/env python3
+"""
+Fetch and enrich GitHub notifications.
+
+This script fetches notifications, enriches them with PR/Issue details,
+assigns tags and categories, and saves to JSON. It does NOT generate summaries -
+those are generated by the Cursor agent.
+
+Usage:
+    python .cursor/commands/fetch_notifications.py [--days 7]
+
+Output:
+    .scratch/notifications-enriched.json - Ready for agent summary generation
+"""
+
+import argparse
+import json
+import subprocess
+import sys
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+from notification_config import get_bot_usernames, get_default_days, get_repo
+
+# Load configuration (auto-detects from git remote or config file)
+REPO = get_repo()
+BOT_USERNAMES = get_bot_usernames()
+
+
+def run_gh_command(args):
+    """Run a gh command and return parsed JSON"""
+    try:
+        result = subprocess.run(
+            ["gh"] + args, capture_output=True, text=True, timeout=60
+        )
+        if result.returncode != 0:
+            return None
+        return json.loads(result.stdout) if result.stdout else None
+    except Exception as e:
+        print(f"Warning: gh command failed: {e}", file=sys.stderr)
+        return None
+
+
+def get_relative_time(updated_at_str):
+    """Convert ISO timestamp to relative time string"""
+    updated = datetime.fromisoformat(updated_at_str.replace("Z", "+00:00"))
+    now = datetime.now(UTC)
+    diff = now - updated
+
+    if diff.days == 0:
+        hours = diff.seconds // 3600
+        if hours == 0:
+            minutes = diff.seconds // 60
+            return f"{minutes} min ago" if minutes != 1 else "1 min ago"
+        return f"{hours}h ago" if hours != 1 else "1h ago"
+    elif diff.days == 1:
+        return "yesterday"
+    elif diff.days < 7:
+        return f"{diff.days}d ago"
+    else:
+        return f"{diff.days // 7}w ago"
+
+
+def get_number_from_url(url):
+    """Extract issue/PR number from API URL"""
+    if not url:
+        return None
+    return url.rstrip("/").split("/")[-1]
+
+
+def get_html_url(notification):
+    """Generate the GitHub HTML URL for a notification"""
+    url = notification.get("url")
+    ntype = notification["type"]
+
+    if not url:
+        if ntype in ["WorkflowRun", "CheckSuite"]:
+            return f"https://github.com/{REPO}/actions"
+        return f"https://github.com/{REPO}"
+
+    number = get_number_from_url(url)
+    if ntype == "PullRequest":
+        return f"https://github.com/{REPO}/pull/{number}"
+    else:
+        return f"https://github.com/{REPO}/issues/{number}"
+
+
+def get_tags(notification, details):
+    """Determine smart tags for a notification"""
+    tags = []
+    reason = notification.get("reason", "")
+    ntype = notification.get("type", "")
+
+    # Check resolved status first
+    state = details.get("state", "")
+    is_merged = details.get("mergedAt") is not None
+    is_resolved = state in ["CLOSED", "MERGED", "closed", "merged"] or is_merged
+
+    # Status from details
+    if ntype == "PullRequest":
+        if is_merged:
+            tags.append(("🎉", "PR MERGED"))
+        elif state in ["CLOSED", "closed"]:
+            tags.append(("⛔", "PR CLOSED"))
+
+        if details.get("isDraft"):
+            tags.append(("📝", "DRAFT PR"))
+
+        # Reviews
+        reviews = details.get("reviews", [])
+        if reviews:
+            states = [r.get("state", "") for r in reviews]
+            if "APPROVED" in states:
+                tags.append(("✅", "PR APPROVED"))
+            if "CHANGES_REQUESTED" in states:
+                tags.append(("🔶", "CHANGES REQUESTED"))
+
+        # CI status
+        checks = details.get("statusCheckRollup", [])
+        if checks:
+            conclusions = [c.get("conclusion", "") for c in checks if c]
+            if "FAILURE" in conclusions:
+                tags.append(("🔴", "CI FAILED"))
+            elif "SUCCESS" in conclusions and "FAILURE" not in conclusions:
+                tags.append(("🟢", "CI PASSING"))
+            elif "PENDING" in conclusions or None in conclusions:
+                tags.append(("🟡", "CI PENDING"))
+
+    elif ntype == "Issue":
+        if state in ["CLOSED", "closed"]:
+            tags.append(("✅", "ISSUE CLOSED"))
+
+    # Check for human comments (non-bot activity)
+    comments = details.get("comments", [])
+    reviews = details.get("reviews", [])
+    human_comments = [
+        c for c in comments if c.get("author", {}).get("login", "") not in BOT_USERNAMES
+    ]
+    human_reviews = [
+        r
+        for r in reviews
+        if r.get("author", {}).get("login", "") not in BOT_USERNAMES and r.get("body")
+    ]
+
+    has_human_feedback = bool(human_comments or human_reviews)
+
+    # Reason-based tags
+    if reason == "mention" and not is_resolved:
+        tags.append(("🔴", "NEEDS RESPONSE"))
+    if reason == "review_requested" and not is_resolved:
+        tags.append(("🔴", "NEEDS RESPONSE"))
+    if reason == "author":
+        tags.append(("👤", "YOUR PR" if ntype == "PullRequest" else "YOUR ISSUE"))
+        if has_human_feedback:
+            tags.append(("💬", "NEW REVIEW COMMENT"))
+    if reason == "comment":
+        tags.append(("💬", "NEW COMMENT"))
+    if reason == "subscribed":
+        tags.append(("👀", "FYI ONLY"))
+
+    if is_resolved:
+        tags.append(("🏁", "RESOLVED"))
+
+    return tags
+
+
+def categorize_notification(notification, tags):
+    """Determine the category for a notification"""
+    reason = notification.get("reason", "")
+    ntype = notification.get("type", "")
+    tag_names = [t[1] for t in tags]
+
+    # Resolved items go to resolved category
+    if "RESOLVED" in tag_names:
+        return "resolved"
+
+    # CI workflow runs go to hidden
+    if ntype in ["WorkflowRun", "CheckSuite"]:
+        return "hidden"
+
+    # Requires action
+    if reason in ["mention", "approval_requested"]:
+        return "action"
+    if reason == "review_requested":
+        return "action"
+    if "CHANGES REQUESTED" in tag_names:
+        return "action"
+    if reason == "author" and any(
+        t in tag_names for t in ["NEW COMMENT", "NEW REVIEW COMMENT"]
+    ):
+        return "action"
+
+    # By type
+    if ntype == "PullRequest":
+        return "pr"
+    elif ntype == "Issue":
+        return "issue"
+
+    return "watching"
+
+
+def fetch_notifications(days=7):
+    """Fetch notifications from GitHub API"""
+    since_date = (datetime.now(UTC) - timedelta(days=days)).strftime(
+        "%Y-%m-%dT%H:%M:%SZ"
+    )
+
+    print(f"📡 Fetching notifications since {since_date}...")
+
+    result = subprocess.run(
+        [
+            "gh",
+            "api",
+            "/notifications",
+            "-X",
+            "GET",
+            "-f",
+            f"since={since_date}",
+            "--paginate",
+        ],
+        capture_output=True,
+        text=True,
+    )
+
+    if result.returncode != 0:
+        print(f"Error fetching notifications: {result.stderr}", file=sys.stderr)
+        return []
+
+    try:
+        all_notifications = json.loads(result.stdout)
+    except json.JSONDecodeError:
+        print("Error parsing notifications JSON", file=sys.stderr)
+        return []
+
+    # Filter to CausalPy
+    notifications = [
+        n for n in all_notifications if n.get("repository", {}).get("full_name") == REPO
+    ]
+
+    print(f"📬 Found {len(notifications)} CausalPy notifications")
+    return notifications
+
+
+def enrich_notification(notification):
+    """Enrich a single notification with PR/Issue details"""
+    ntype = notification.get("subject", {}).get("type", "")
+    url = notification.get("subject", {}).get("url", "")
+    number = get_number_from_url(url) if url else None
+
+    # Extract basic fields
+    enriched = {
+        "id": notification["id"],
+        "reason": notification.get("reason", ""),
+        "type": ntype,
+        "title": notification.get("subject", {}).get("title", ""),
+        "url": url,
+        "updated_at": notification.get("updated_at", ""),
+        "unread": notification.get("unread", False),
+        "number": number,
+    }
+
+    # Fetch details
+    details = {}
+    if number:
+        print(f"   📋 Enriching #{number}...")
+        if ntype == "PullRequest":
+            details = (
+                run_gh_command(
+                    [
+                        "pr",
+                        "view",
+                        number,
+                        "--repo",
+                        REPO,
+                        "--json",
+                        "state,isDraft,author,reviews,comments,statusCheckRollup,mergedAt,body",
+                    ]
+                )
+                or {}
+            )
+        elif ntype == "Issue":
+            details = (
+                run_gh_command(
+                    [
+                        "issue",
+                        "view",
+                        number,
+                        "--repo",
+                        REPO,
+                        "--json",
+                        "state,author,labels,comments,body",
+                    ]
+                )
+                or {}
+            )
+
+    # Compute derived fields
+    tags = get_tags(enriched, details)
+    category = categorize_notification(enriched, tags)
+
+    enriched.update(
+        {
+            "details": details,
+            "tags": tags,
+            "category": category,
+            "html_url": get_html_url(enriched),
+            "relative_time": get_relative_time(enriched["updated_at"]),
+            "summary": "",  # Will be filled by agent
+        }
+    )
+
+    return enriched
+
+
+def main():
+    default_days = get_default_days()
+    parser = argparse.ArgumentParser(
+        description=f"Fetch GitHub notifications for {REPO}"
+    )
+    parser.add_argument(
+        "--days",
+        type=int,
+        default=default_days,
+        help=f"Number of days to look back (default: {default_days})",
+    )
+    args = parser.parse_args()
+
+    # Create scratch directory
+    Path(".scratch").mkdir(exist_ok=True)
+
+    # Fetch notifications
+    notifications = fetch_notifications(args.days)
+
+    if not notifications:
+        print("No notifications found.")
+        # Save empty file
+        with open(".scratch/notifications-enriched.json", "w") as f:
+            json.dump([], f)
+        return
+
+    # Enrich each notification
+    print("🔍 Enriching notifications with details...")
+    enriched = []
+    for n in notifications:
+        enriched.append(enrich_notification(n))
+
+    # Save to JSON
+    output_file = ".scratch/notifications-enriched.json"
+    with open(output_file, "w") as f:
+        json.dump(enriched, f, indent=2)
+
+    print(f"\n✅ Saved {len(enriched)} enriched notifications to {output_file}")
+    print("\n📊 Summary by category:")
+
+    categories = {}
+    for n in enriched:
+        cat = n["category"]
+        categories[cat] = categories.get(cat, 0) + 1
+
+    for cat, count in sorted(categories.items(), key=lambda x: -x[1]):
+        emoji = {
+            "action": "🔥",
+            "pr": "📋",
+            "issue": "📝",
+            "watching": "👀",
+            "resolved": "🏁",
+            "hidden": "🙈",
+        }.get(cat, "")
+        print(f"   {emoji} {cat}: {count}")
+
+
+if __name__ == "__main__":
+    main()