diff --git a/.cursor/commands/fetch_notifications.py b/.cursor/commands/fetch_notifications.py new file mode 100644 index 000000000..9a39a22ee --- /dev/null +++ b/.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() diff --git a/.cursor/commands/generate_summaries.py b/.cursor/commands/generate_summaries.py new file mode 100644 index 000000000..d38b17ce2 --- /dev/null +++ b/.cursor/commands/generate_summaries.py @@ -0,0 +1,284 @@ +#!/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 +""" +Generate comprehensive AI-style summaries for notifications. + +This creates detailed, actionable summaries that help users decide +which notifications to prioritize. +""" + +import json + +from notification_config import get_bot_usernames + +# Load configuration +BOT_USERNAMES = get_bot_usernames() + + +def generate_summary(n): + """Generate a comprehensive summary for a notification.""" + reason = n.get("reason", "") + category = n.get("category", "") + title = n.get("title", "") + ntype = n.get("type", "") + details = n.get("details", {}) + tags = [t[1] for t in n.get("tags", [])] + + body = details.get("body", "") or "" + comments = details.get("comments", []) + reviews = details.get("reviews", []) + + # Filter to human comments only + 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") + ] + + summary_parts = [] + + # === RESOLVED ITEMS === + if category == "resolved": + state = details.get("state", "") + if details.get("mergedAt"): + summary_parts.append("**Status:** โ Merged") + summary_parts.append("") + summary_parts.append( + "This PR has been successfully merged. No further action required." + ) + if body: + summary_parts.append("") + summary_parts.append("**What it did:**") + body_preview = body[:300].replace("\n", " ").strip() + summary_parts.append(f"> {body_preview}...") + elif state in ["CLOSED", "closed"]: + summary_parts.append("**Status:** โ Closed") + summary_parts.append("") + summary_parts.append( + f"This {ntype.lower()} was closed without merging. No action needed." + ) + else: + summary_parts.append("**Status:** โ Resolved") + return "\n".join(summary_parts) + + # === HIDDEN ITEMS === + if category == "hidden": + return "**Status:** ๐ CI/Workflow Activity\n\nThis is automated CI activity. Usually no action needed unless investigating build failures." + + # === ACTION ITEMS === + if reason == "mention": + summary_parts.append("**๐ You were mentioned**") + summary_parts.append("") + + if human_comments: + last = human_comments[-1] + author = last.get("author", {}).get("login", "Someone") + comment_body = last.get("body", "") + summary_parts.append(f"**{author}** said:") + summary_parts.append(f"> {comment_body[:500]}") + summary_parts.append("") + + # Check if it's a question + if "?" in comment_body or "what do you think" in comment_body.lower(): + summary_parts.append( + "**Action Required:** This appears to be a question directed at you. Please review and respond." + ) + else: + summary_parts.append( + "**Action Required:** You were mentioned - review the context and respond if needed." + ) + + elif human_reviews: + last = human_reviews[-1] + author = last.get("author", {}).get("login", "Someone") + review_body = last.get("body", "") + summary_parts.append(f"**{author}** mentioned you in a review:") + summary_parts.append(f"> {review_body[:500]}") + summary_parts.append("") + summary_parts.append( + "**Action Required:** Review the feedback and respond." + ) + else: + summary_parts.append("You were mentioned in this discussion.") + summary_parts.append("") + summary_parts.append( + "**Action Required:** Check the thread for context and respond if needed." + ) + + elif reason == "review_requested": + summary_parts.append("**๐ Review Requested**") + summary_parts.append("") + + if body: + # Extract key info from PR body + body_lines = body.split("\n") + body_preview = "\n".join(body_lines[:10]) + summary_parts.append("**What this PR does:**") + summary_parts.append(f"> {body_preview[:600]}") + summary_parts.append("") + + # Check CI status + if "CI FAILED" in tags: + summary_parts.append( + "โ ๏ธ **CI is failing** - you may want to wait for fixes before reviewing." + ) + summary_parts.append("") + + if "DRAFT PR" in tags: + summary_parts.append( + "๐ **This is a draft PR** - the author may still be working on it." + ) + summary_parts.append("") + + summary_parts.append( + "**Action Required:** Review this PR and provide feedback (approve, request changes, or comment)." + ) + + elif reason == "author": + summary_parts.append(f"**๐ฃ Activity on your {ntype}**") + summary_parts.append("") + + # Check for changes requested + if "CHANGES REQUESTED" in tags: + summary_parts.append("**๐ถ Changes Requested**") + summary_parts.append("") + for r in human_reviews[-3:]: + if r.get("state") == "CHANGES_REQUESTED": + author = r.get("author", {}).get("login", "") + review_body = r.get("body", "") + if review_body: + summary_parts.append(f"**{author}** requested changes:") + summary_parts.append(f"> {review_body[:400]}") + summary_parts.append("") + summary_parts.append( + "**Action Required:** Address the requested changes and update the PR." + ) + + elif "PR APPROVED" in tags: + summary_parts.append("**โ Your PR has been approved!**") + summary_parts.append("") + approvers = [ + r.get("author", {}).get("login", "") + for r in reviews + if r.get("state") == "APPROVED" + ] + if approvers: + summary_parts.append(f"Approved by: {', '.join(approvers)}") + summary_parts.append("") + summary_parts.append("**Action:** Merge when ready (if CI is passing).") + + elif "CI FAILED" in tags: + summary_parts.append("**๐ด CI Failed**") + summary_parts.append("") + summary_parts.append("Your PR has failing CI checks.") + summary_parts.append("") + summary_parts.append( + "**Action Required:** Check the CI logs and fix the failing tests/checks." + ) + + elif human_comments or human_reviews: + summary_parts.append("**๐ฌ New Feedback**") + summary_parts.append("") + + # Show recent human activity + all_activity = [] + for c in human_comments[-3:]: + all_activity.append(("comment", c)) + for r in human_reviews[-2:]: + if r.get("body"): + all_activity.append(("review", r)) + + for atype, item in all_activity[-3:]: + author = item.get("author", {}).get("login", "Someone") + body_text = item.get("body", "")[:300] + if atype == "review": + state = item.get("state", "") + state_emoji = {"APPROVED": "โ ", "CHANGES_REQUESTED": "๐ถ"}.get( + state, "๐ฌ" + ) + summary_parts.append(f"**{author}** ({state_emoji} review):") + else: + summary_parts.append(f"**{author}** commented:") + summary_parts.append(f"> {body_text}") + summary_parts.append("") + + summary_parts.append( + "**Action:** Review the feedback and respond if needed." + ) + + else: + summary_parts.append( + f"There's activity on your {ntype.lower()}, but no specific feedback to highlight." + ) + + elif reason == "subscribed": + summary_parts.append("**๐ Watching**") + summary_parts.append("") + if body: + body_preview = body[:400].replace("\n", " ") + summary_parts.append(f"> {body_preview}") + summary_parts.append("") + summary_parts.append( + "**FYI only** - you're watching this thread. No action required unless you want to participate." + ) + + else: + # Generic notification + if body: + body_preview = body[:500].replace("\n", " ") + summary_parts.append(body_preview) + else: + summary_parts.append(f"Notification about: {title}") + + return "\n".join(summary_parts) if summary_parts else f"Notification about: {title}" + + +def main(): + from pathlib import Path + + # Get the project root (parent of .cursor/commands) + script_dir = Path(__file__).parent + project_root = script_dir.parent.parent + scratch_dir = project_root / ".scratch" + + # Load enriched notifications + input_file = scratch_dir / "notifications-enriched.json" + with open(input_file) as f: + notifications = json.load(f) + + print( + f"Generating comprehensive summaries for {len(notifications)} notifications..." + ) + + # Generate summaries + for n in notifications: + n["summary"] = generate_summary(n) + preview = n["summary"][:60].replace("\n", " ") + print(f" #{n['number']}: {preview}...") + + # Save with summaries + output_file = scratch_dir / "notifications-with-summaries.json" + with open(output_file, "w") as f: + json.dump(notifications, f, indent=2) + + print(f"\nโ Saved to {output_file}") + + +if __name__ == "__main__": + main() diff --git a/.cursor/commands/notification_bots.yml b/.cursor/commands/notification_bots.yml new file mode 100644 index 000000000..36a46b043 --- /dev/null +++ b/.cursor/commands/notification_bots.yml @@ -0,0 +1,32 @@ +# Bot Blacklist for GitHub Notification Summary +# +# Notifications from these usernames are filtered out (ignored). +# This helps reduce noise from automated bot comments. +# +# Format: List of GitHub usernames (case-insensitive matching) +# Bot accounts typically end with [bot] suffix + +# Code coverage bots +- codecov +- codecov[bot] +- codecov-commenter + +# CI/CD bots +- pre-commit-ci +- pre-commit-ci[bot] +- github-actions +- github-actions[bot] + +# Dependency update bots +- dependabot +- dependabot[bot] +- renovate +- renovate[bot] + +# Review/notebook bots +- review-notebook-app +- review-notebook-app[bot] + +# Add your custom bots below: +# - my-custom-bot +# - another-bot[bot] diff --git a/.cursor/commands/notification_config.py b/.cursor/commands/notification_config.py new file mode 100644 index 000000000..1258c5545 --- /dev/null +++ b/.cursor/commands/notification_config.py @@ -0,0 +1,263 @@ +#!/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. +""" +Configuration for the GitHub notification summary command. + +Provides auto-detection of org/repo from git remote, with optional +config file overrides for customization. + +Configuration files: +- .cursor/commands/notification_bots.yml - Bot blacklist (usernames to IGNORE) +- .cursor/notification_config.json - General settings (optional) +""" + +import json +import re +import subprocess +from pathlib import Path + +# Default settings +DEFAULT_DAYS = 7 +DEFAULT_SERVER_PORT = 8765 + +# Fallback bot blacklist (used if YAML file not found) +FALLBACK_BOT_BLACKLIST = { + "codecov", + "codecov[bot]", + "codecov-commenter", + "pre-commit-ci", + "pre-commit-ci[bot]", + "dependabot", + "dependabot[bot]", + "github-actions", + "github-actions[bot]", + "renovate", + "renovate[bot]", + "review-notebook-app[bot]", + "review-notebook-app", +} + + +def get_project_root() -> Path: + """Get the project root directory (where .git is located).""" + # Start from this file's location and go up + current = Path(__file__).resolve().parent + while current != current.parent: + if (current / ".git").exists(): + return current + current = current.parent + # Fallback to cwd + return Path.cwd() + + +def get_commands_dir() -> Path: + """Get the .cursor/commands directory.""" + return Path(__file__).resolve().parent + + +def load_bot_blacklist() -> set[str]: + """ + Load the bot blacklist from notification_bots.yml. + + This is a BLACKLIST - notifications from these usernames are IGNORED. + Helps filter out noise from automated bot comments (codecov, dependabot, etc.) + + Returns: + Set of usernames to ignore (case-sensitive as GitHub usernames are). + """ + commands_dir = get_commands_dir() + bots_file = commands_dir / "notification_bots.yml" + + if bots_file.exists(): + try: + # Simple YAML list parsing (avoids PyYAML dependency) + bots = set() + with open(bots_file) as f: + for line in f: + line = line.strip() + # Skip comments and empty lines + if not line or line.startswith("#"): + continue + # Parse list items (- username) + if line.startswith("- "): + username = line[2:].strip() + # Remove inline comments + if "#" in username: + username = username.split("#")[0].strip() + if username: + bots.add(username) + if bots: + return bots + except Exception as e: + print(f"Warning: Failed to load bot blacklist from {bots_file}: {e}") + + # Fallback to hardcoded defaults + return FALLBACK_BOT_BLACKLIST.copy() + + +def parse_git_remote(remote_url: str) -> tuple[str, str] | None: + """ + Parse org and repo from a git remote URL. + + Supports: + - https://github.com/org/repo.git + - https://github.com/org/repo + - git@github.com:org/repo.git + - git@github.com:org/repo + """ + # HTTPS format + https_match = re.match( + r"https://github\.com/([^/]+)/([^/]+?)(?:\.git)?$", remote_url + ) + if https_match: + return https_match.group(1), https_match.group(2) + + # SSH format + ssh_match = re.match(r"git@github\.com:([^/]+)/([^/]+?)(?:\.git)?$", remote_url) + if ssh_match: + return ssh_match.group(1), ssh_match.group(2) + + return None + + +def detect_repo_from_git() -> str | None: + """Auto-detect org/repo from git remote origin.""" + try: + result = subprocess.run( + ["git", "remote", "get-url", "origin"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0: + remote_url = result.stdout.strip() + parsed = parse_git_remote(remote_url) + if parsed: + org, repo = parsed + return f"{org}/{repo}" + except Exception: + pass + return None + + +def load_config() -> dict: + """ + Load configuration with the following priority: + 1. .cursor/notification_config.json (if exists) - for repo, days, port + 2. .cursor/commands/notification_bots.yml - for bot blacklist + 3. Auto-detect repo from git remote + 4. Built-in defaults + + Returns a dict with: + - repo: str (e.g., "pymc-labs/CausalPy") + - bot_blacklist: set[str] - usernames to IGNORE + - default_days: int + - server_port: int + """ + project_root = get_project_root() + config_file = project_root / ".cursor" / "notification_config.json" + + # Start with defaults + config = { + "repo": None, + "bot_blacklist": load_bot_blacklist(), + "default_days": DEFAULT_DAYS, + "server_port": DEFAULT_SERVER_PORT, + } + + # Try to load JSON config file for general settings + if config_file.exists(): + try: + with open(config_file) as f: + user_config = json.load(f) + + # Override with user settings + if "repo" in user_config: + config["repo"] = user_config["repo"] + if "default_days" in user_config: + config["default_days"] = user_config["default_days"] + if "server_port" in user_config: + config["server_port"] = user_config["server_port"] + # Legacy support: bot_usernames in JSON extends the blacklist + if "bot_usernames" in user_config: + config["bot_blacklist"].update(user_config["bot_usernames"]) + except Exception as e: + print(f"Warning: Failed to load config file: {e}") + + # Auto-detect repo if not specified + if not config["repo"]: + detected = detect_repo_from_git() + if detected: + config["repo"] = detected + + return config + + +def get_repo() -> str: + """Get the repository in org/repo format.""" + config = load_config() + repo = config.get("repo") + if not repo: + raise ValueError( + "Could not determine repository. Either:\n" + "1. Run from a git repository with a GitHub remote, or\n" + '2. Create .cursor/notification_config.json with: {"repo": "org/repo"}' + ) + return repo + + +def get_bot_blacklist() -> set[str]: + """ + Get the set of bot usernames to IGNORE (blacklist). + + Notifications from these usernames are filtered out to reduce noise. + Edit .cursor/commands/notification_bots.yml to customize. + """ + return load_config()["bot_blacklist"] + + +# Alias for backwards compatibility +def get_bot_usernames() -> set[str]: + """Alias for get_bot_blacklist() - returns usernames to ignore.""" + return get_bot_blacklist() + + +def get_default_days() -> int: + """Get the default number of days to look back.""" + return load_config()["default_days"] + + +def get_server_port() -> int: + """Get the server port.""" + return load_config()["server_port"] + + +if __name__ == "__main__": + # Test the config + print("=" * 60) + print("GitHub Notification Summary - Configuration") + print("=" * 60) + config = load_config() + print(f"\n๐ฆ Repository: {config['repo'] or '(not detected)'}") + print(f"๐ Default days: {config['default_days']}") + print(f"๐ Server port: {config['server_port']}") + print(f"\n๐ค Bot Blacklist ({len(config['bot_blacklist'])} usernames to IGNORE):") + for bot in sorted(config["bot_blacklist"]): + print(f" - {bot}") + print("\n๐ Config files:") + print(f" - Bot blacklist: {get_commands_dir() / 'notification_bots.yml'}") + print( + f" - General config: {get_project_root() / '.cursor/notification_config.json'}" + ) diff --git a/.cursor/commands/notification_server.py b/.cursor/commands/notification_server.py new file mode 100644 index 000000000..7acd576ed --- /dev/null +++ b/.cursor/commands/notification_server.py @@ -0,0 +1,1066 @@ +#!/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 +""" +Local web server for interactive GitHub notification management. + +Serves the notification summary as an HTML page with "Mark Done" buttons +that call the GitHub API to archive notifications. + +Usage: + python .cursor/commands/notification_server.py [--port 8765] [--no-open] + +Prerequisites: + Run fetch_notifications.py first to create the enriched JSON file. + Optionally, have the Cursor agent add AI-generated summaries. + +The server will: +1. Load notifications from .scratch/notifications-with-summaries.json + (or .scratch/notifications-enriched.json as fallback) +2. Serve an interactive HTML page with tabs for each category +3. Handle "Mark Done" API calls via DELETE /notifications/threads/{id} +""" + +import argparse +import json +import subprocess +import sys +import webbrowser +from datetime import UTC, datetime +from http.server import HTTPServer, SimpleHTTPRequestHandler +from pathlib import Path +from urllib.parse import urlparse + +from notification_config import get_bot_usernames, get_repo, get_server_port + +# Load configuration (auto-detects from git remote or config file) +REPO = get_repo() +DEFAULT_PORT = get_server_port() +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=30 + ) + if result.returncode != 0: + return None + return json.loads(result.stdout) if result.stdout else None + except Exception: + 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["reason"] + ntype = notification["type"] + + # Handle CI/Workflow notifications + if ntype in ["WorkflowRun", "CheckSuite"]: + title = notification.get("title", "").lower() + if "failed" in title: + tags.append(("๐ด", "CI FAILED")) + elif "cancelled" in title: + tags.append(("โช", "CI CANCELLED")) + if reason == "approval_requested": + tags.append(("๐ด", "NEEDS RESPONSE")) + else: + tags.append(("๐ง", "CI ACTIVITY")) + return tags + + # Action-required tags + if reason == "mention": + tags.append(("๐ด", "NEEDS RESPONSE")) + if reason == "review_requested": + tags.append(("๐ด", "NEEDS RESPONSE")) + + # Author tag + if reason == "author": + tags.append(("๐ค", "YOUR PR" if ntype == "PullRequest" else "YOUR ISSUE")) + + if ntype == "PullRequest" and details: + # Merged/Closed status + if details.get("mergedAt"): + tags.append(("๐", "PR MERGED")) + tags.append(("๐", "RESOLVED")) + elif details.get("state") == "CLOSED": + tags.append(("โ", "PR CLOSED")) + tags.append(("๐", "RESOLVED")) + + # Review status + reviews = details.get("reviews", []) + if reviews: + if any(r.get("state") == "APPROVED" for r in reviews): + tags.append(("โ ", "PR APPROVED")) + if any(r.get("state") == "CHANGES_REQUESTED" for r in reviews): + tags.append(("๐ถ", "CHANGES REQUESTED")) + + # CI status + checks = details.get("statusCheckRollup", []) + if checks: + conclusions = [c.get("conclusion") or c.get("state") for c in checks] + conclusions = [ + c for c in conclusions if c and c not in ["SKIPPED", "NEUTRAL"] + ] + if conclusions: + if any(c in ["FAILURE", "ERROR"] for c in conclusions): + tags.append(("๐ด", "CI FAILED")) + elif all(c in ["SUCCESS"] for c in conclusions): + tags.append(("๐ข", "CI PASSING")) + + # Human feedback + if reason == "author": + reviews = details.get("reviews", []) + human_reviews = [ + r + for r in reviews + if r.get("author", {}).get("login", "") not in BOT_USERNAMES + ] + if human_reviews and human_reviews[-1].get("state") == "COMMENTED": + tags.append(("๐ฌ", "NEW REVIEW COMMENT")) + + elif ntype == "Issue" and details: + if details.get("state") == "CLOSED": + tags.append(("โ ", "ISSUE CLOSED")) + tags.append(("๐", "RESOLVED")) + + # Comment tag + if reason == "comment": + tags.append(("๐ฌ", "NEW COMMENT")) + + # FYI only + if reason == "subscribed" and not any(t[0] in ["๐ด", "๐ถ"] for t in tags): + tags.append(("๐", "FYI ONLY")) + + return tags + + +def categorize_notification(notification, tags): + """Determine which category a notification belongs to""" + reason = notification["reason"] + ntype = notification["type"] + tag_names = [t[1] for t in tags] + is_resolved = "RESOLVED" in tag_names + + # Hidden + if ntype in ["WorkflowRun", "CheckSuite"] and reason != "approval_requested": + return "hidden" + + # Bot comments go to watching + if "BOT COMMENT" in tag_names: + return "watching" + + # Mentions always need response + if reason == "mention": + return "action" + + # Resolved items + if is_resolved: + return "resolved" + + # Review requests + if reason == "review_requested": + return "action" + + # Approval requests + if reason == "approval_requested": + return "action" + + # Changes requested + if "CHANGES REQUESTED" in tag_names: + return "action" + + # Author with feedback + 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 load_notifications(): + """Load pre-enriched notifications from JSON. + + Expects notifications to be already enriched by fetch_notifications.py + and summaries to be added by the Cursor agent. + """ + # Try the enriched file first (new workflow) + enriched_file = ".scratch/notifications-with-summaries.json" + fallback_file = ".scratch/notifications-enriched.json" + + try: + with open(enriched_file) as f: + notifications = json.load(f) + print(f"๐ Loaded from {enriched_file}") + except FileNotFoundError: + try: + with open(fallback_file) as f: + notifications = json.load(f) + print(f"๐ Loaded from {fallback_file} (no summaries yet)") + except FileNotFoundError: + print("โ No notification file found. Run fetch_notifications.py first.") + return [] + + # Update relative times (they may be stale) + for n in notifications: + if n.get("updated_at"): + n["relative_time"] = get_relative_time(n["updated_at"]) + + return notifications + + +def is_action_required(n): + """Determine if a notification requires action.""" + reason = n.get("reason", "") + tags = [t[1] for t in n.get("tags", [])] + + # Direct action triggers + if reason in ["mention", "review_requested", "approval_requested"]: + return True + if "CHANGES REQUESTED" in tags: + return True + return reason == "author" and any( + t in tags for t in ["NEW COMMENT", "NEW REVIEW COMMENT"] + ) + + +def generate_html(notifications): + """Generate HTML page for notifications""" + # Categorize by TYPE first, then we'll split by action within each type + categories = {"pr": [], "issue": [], "watching": [], "resolved": [], "hidden": []} + + for n in notifications: + cat = n["category"] + # Map old "action" category back to pr/issue based on type + if cat == "action": + if n["type"] == "PullRequest": + categories["pr"].append(n) + elif n["type"] == "Issue": + categories["issue"].append(n) + else: + categories["watching"].append(n) + else: + categories[cat].append(n) + + # Mark each notification as action_required + for cat in categories.values(): + for n in cat: + n["action_required"] = is_action_required(n) + + # Sort by updated_at + for cat in categories.values(): + cat.sort(key=lambda x: x["updated_at"], reverse=True) + + html = f""" + +
+ + +| # | Title | Actions |
|---|---|---|
| #{number} | +{title_truncated} | ++ |