notification_copilot.py

"""Main orchestration entry point for GitHub Notification Copilot."""

from __future__ import annotations

import logging
import os
from typing import Any, Dict, List, Optional

import yaml
from dotenv import load_dotenv

from actions_executor import ActionsExecutor
from github_api_client import GitHubAPIClient
from llm_classifier import LLMClassifier

load_dotenv()
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s — %(message)s",
)
logger = logging.getLogger(__name__)


def load_config(path: str = "config.yaml") -> Dict[str, Any]:
    if not os.path.exists(path):
        return {}
    with open(path) as f:
        return yaml.safe_load(f) or {}


class NotificationCopilot:
    """Orchestrates notification fetching, classification, and action execution."""

    def __init__(
        self,
        github_token: Optional[str] = None,
        anthropic_api_key: Optional[str] = None,
        config_path: str = "config.yaml",
        dry_run: bool = False,
    ):
        self._config = load_config(config_path)
        token = github_token or os.getenv("GITHUB_TOKEN")
        if not token:
            raise ValueError("GITHUB_TOKEN required (env var or github_token param)")

        self._github = GitHubAPIClient(token=token)
        self._executor = ActionsExecutor(self._github, dry_run=dry_run)
        self._dry_run = dry_run

        llm_key = anthropic_api_key or os.getenv("ANTHROPIC_API_KEY")
        if llm_key:
            try:
                self._classifier: Optional[LLMClassifier] = LLMClassifier(api_key=llm_key)
            except RuntimeError as exc:
                logger.warning("LLM classifier unavailable: %s", exc)
                self._classifier = None
        else:
            logger.warning("No ANTHROPIC_API_KEY — will use rule-based fallback classification")
            self._classifier = None

    def triage(self, limit: int = 50) -> List[Dict]:
        """Fetch, classify, and act on notifications. Returns list of results."""
        notifications = self._github.get_notifications(per_page=limit)
        logger.info("Fetched %d notifications", len(notifications))

        results = []
        for notif in notifications:
            classification = (
                self._classifier.classify(notif)
                if self._classifier
                else self._rule_based_classify(notif)
            )
            action_result = self._executor.execute(notif, classification)
            results.append(
                {
                    "id": notif["id"],
                    "title": notif.get("subject", {}).get("title", ""),
                    "repo": notif.get("repository", {}).get("full_name", ""),
                    **classification,
                    "executed": action_result.get("executed", False),
                }
            )

        summary = self._executor.get_execution_summary()
        logger.info("Triage complete — %s", summary)
        return results

    # ------------------------------------------------------------------
    # Rule-based fallback (no LLM)
    # ------------------------------------------------------------------

    def _rule_based_classify(self, notification: Dict) -> Dict:
        """Simple heuristic classifier used when LLM is unavailable."""
        reason = notification.get("reason", "")
        subject = notification.get("subject", {})
        ntype = subject.get("type", "")
        title = subject.get("title", "")

        # P1: direct review requests or mentions
        if reason in ("review_requested", "assign"):
            return {"priority": "P1", "action": "review_now", "reason": "Direct review request", "summary": title[:50]}

        # P3: bot-authored or automated
        lower_title = title.lower()
        if any(kw in lower_title for kw in ["dependabot", "renovate", "[bot]", "security advisory"]):
            return {"priority": "P3", "action": "mute", "reason": "Automated/bot notification", "summary": title[:50]}

        # P2: everything else
        return {"priority": "P2", "action": "review_later", "reason": f"Reason: {reason}", "summary": title[:50]}


def main():
    import argparse

    parser = argparse.ArgumentParser(description="GitHub Notification Copilot")
    parser.add_argument("--limit", type=int, default=50, help="Max notifications to triage")
    parser.add_argument("--dry-run", action="store_true", help="Print actions without executing")
    args = parser.parse_args()

    copilot = NotificationCopilot(dry_run=args.dry_run)
    results = copilot.triage(limit=args.limit)

    for r in results:
        flag = "🔴" if r["priority"] == "P1" else "🟡" if r["priority"] == "P2" else "⚪"
        print(f"{flag} [{r['priority']}] {r['repo']} — {r['summary']} ({r['action']})")

    print(f"\n✅ Triaged {len(results)} notifications")


if __name__ == "__main__":
    main()