summarize_activity.py

#!/usr/bin/env python3
"""summarize_activity.py

DEPRECATED: Use summarize_reports.py instead.

Generate a Slack-ready markdown summary of GitHub & ClickUp activity that can be
piped directly into send_to_slack.py.  The script expects one or more glob
patterns pointing to JSON exports like:
  * *_commit_raw_data.json              – commits coming from PRs
  * *_repo_commits_raw_data.json        – full repo commit history
  * *_pr_raw_data.json                  – pull-request list
  * *_tasks_raw_data.json               – ClickUp tasks list

NEW: AI Summary Feature
-----------------------
The script can now generate an AI-powered executive summary using Google's 
Gemini 2.5 Flash model. To use this feature:
  1. Install: pip install google-generativeai python-dotenv
  2. Set GOOGLE_API_KEY environment variable, use --gemini-api-key flag, or create a .env file
  3. The AI summary will appear at the top of the markdown output

.env file example:
GOOGLE_API_KEY=your_api_key_here

Example
-------
python summarize_activity.py output/*72h*.json \
        --user-map user_map.yml \
        --output 72h-summary.md \
        --gemini-api-key YOUR_API_KEY

The emitted markdown already follows the header patterns recognised by
send_to_slack.py, so GitHub/user mentions and links become proper Slack
mentions when that helper script posts the message.
"""

from __future__ import annotations

import argparse
import glob
import json
import os
import re
import sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Set

import yaml  # type: ignore – requires PyYAML

try:
    import google.generativeai as genai  # type: ignore
    GEMINI_AVAILABLE = True
except ImportError:
    GEMINI_AVAILABLE = False

try:
    from dotenv import load_dotenv  # type: ignore
    load_dotenv()  # Load environment variables from .env file
    DOTENV_AVAILABLE = True
except ImportError:
    DOTENV_AVAILABLE = False

# ---------------------------------------------------------------------------
# Constants & helpers
# ---------------------------------------------------------------------------
DEFAULT_CLICKUP_TEAM_ID = os.environ.get("CLICKUP_TEAM_ID", "9009140026")

Event = Dict[str, Any]
UserBucket = Dict[str, Any]


def load_user_map(path: str) -> Dict[str, Dict[str, str]]:
    """Return two plain dicts: github_login->slack_id, email->slack_id."""
    if not Path(path).exists():
        return {"github": {}, "email": {}}

    with open(path, "r", encoding="utf-8") as f:
        raw = yaml.safe_load(f) or {}

    return {
        "github": raw.get("github_user_to_slack_id", {}) or {},
        "email": raw.get("clickup_user_to_slack_id", {}) or {},
    }


def resolve_slack_id(alias: str, maps: Dict[str, Dict[str, str]]) -> Optional[str]:
    """Return Slack ID for a github login or email; None if not mapped."""
    return maps["github"].get(alias) or maps["email"].get(alias)


# ---------------------------------------------------------------------------
# JSON discovery & parsing helpers
# ---------------------------------------------------------------------------

def discover_files(glob_patterns: List[str]) -> List[Path]:
    files: List[Path] = []
    for pattern in glob_patterns:
        for path in glob.glob(pattern):
            p = Path(path)
            if p.suffix.lower() == ".json":
                files.append(p)
    return files


def derive_timespan_string(glob_patterns: List[str], discovered_files: Optional[List[Path]] = None) -> Optional[str]:
    """Derive a descriptive timespan string from glob patterns or discovered filenames."""
    # First, check the glob patterns themselves
    for pattern_str in glob_patterns:
        # Check for Xh, X H, Xhour, X hour (e.g., 72h, 24H)
        match_hour = re.search(r"(\d+)\s?(?:h|H|hour|Hour)", pattern_str, re.IGNORECASE)
        if match_hour:
            return f"{match_hour.group(1)}-Hour"

        # Check for keywords like daily, weekly, etc. in glob patterns
        keywords_map = {
            "daily": "Daily",
            "weekly": "Weekly",
            "monthly": "Monthly",
            "quarterly": "Quarterly",
        }
        for keyword, display_text in keywords_map.items():
            if keyword in pattern_str.lower():
                return display_text

    # If no timespan found in globs, check the discovered filenames as a fallback
    if discovered_files:
        for file_path in discovered_files:
            filename_lower = file_path.name.lower()
            # Check for Xh, X H, Xhour, X hour (e.g., 72h, 24H) in filenames
            match_hour_fn = re.search(r"(\d+)\s?(?:h|H|hour|Hour)", filename_lower, re.IGNORECASE)
            if match_hour_fn:
                return f"{match_hour_fn.group(1)}-Hour"

            # Check for keywords like daily, weekly, etc. in filenames
            for keyword, display_text in keywords_map.items():
                if keyword in filename_lower:
                    return display_text

    return None # No specific timespan keyword found


# ---------------------------------------------------------------------------
# Per-type extractors – all return List[Event]
# ---------------------------------------------------------------------------

def extract_commits(data: Any, user_maps: Dict[str, Dict[str, str]]) -> List[Event]:
    events: List[Event] = []
    if not isinstance(data, list):
        return events

    for obj in data:
        commit = obj.get("commit", {})
        author_login = obj.get("author", {}) or {}
        login = author_login.get("login") or commit.get("author", {}).get("name") or "unknown"
        slack_id = resolve_slack_id(login, user_maps)

        events.append(
            {
                "slack_id": slack_id,
                "alias": login,
                "event": "commit",
                "sha": obj.get("sha", "")[:7],
                "message": (commit.get("message") or "").split("\n")[0],
                "url": obj.get("html_url"),
            }
        )
    return events


def extract_prs(data: Any, user_maps: Dict[str, Dict[str, str]]) -> List[Event]:
    events: List[Event] = []
    if not isinstance(data, list):
        return events

    for pr in data:
        user = pr.get("user", {}) or {}
        login = user.get("login", "unknown")
        slack_id = resolve_slack_id(login, user_maps)
        events.append(
            {
                "slack_id": slack_id,
                "alias": login,
                "event": "pr",
                "number": pr.get("number"),
                "title": pr.get("title"),
                "state": pr.get("state"),
                "url": pr.get("html_url"),
            }
        )
    return events


def _derive_name_from_email(email: str) -> str:
    local = email.split("@", 1)[0]
    parts = local.split(".")
    return " ".join(p.capitalize() for p in parts)


def extract_tasks(data: Any, user_maps: Dict[str, Dict[str, str]], clickup_team_id: str) -> List[Event]:
    events: List[Event] = []
    if not isinstance(data, list):
        return events

    for task in data:
        creator = task.get("creator", {})
        email = creator.get("email", "")
        slack_id = resolve_slack_id(email, user_maps)
        alias = email or str(creator.get("id", "unknown"))

        status_obj = task.get("status") or {}
        status = status_obj.get("status", "").lower()
        done = any(s in status for s in ("done", "complete", "closed")) or bool(task.get("date_done"))

        event_type = "task_completed" if done else "task_in_progress"
        task_identifier = task.get("custom_id") or task.get("id")
        url = f"https://app.clickup.com/t/{clickup_team_id}/{task_identifier}"

        events.append(
            {
                "slack_id": slack_id,
                "alias": alias,
                "event": event_type,
                "task_id": task_identifier,
                "title": task.get("name"),
                "url": url,
            }
        )
    return events


# ---------------------------------------------------------------------------
# File router
# ---------------------------------------------------------------------------

def parse_file(path: Path, user_maps: Dict[str, Dict[str, str]], clickup_team_id: str) -> List[Event]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:
        print(f"WARNING: failed to parse {path}: {exc}", file=sys.stderr)
        return []

    name = path.name.lower()
    if "commit_raw_data" in name:
        return extract_commits(data, user_maps)
    if "repo_commits_raw_data" in name:
        return extract_commits(data, user_maps)
    if "pr_raw_data" in name:
        return extract_prs(data, user_maps)
    if "tasks_raw_data" in name:
        return extract_tasks(data, user_maps, clickup_team_id)

    # Unknown type
    return []


# ---------------------------------------------------------------------------
# Aggregation
# ---------------------------------------------------------------------------

def aggregate(events: List[Event]) -> Dict[str, UserBucket]:
    buckets: Dict[str, UserBucket] = defaultdict(lambda: {
        "aliases": set(),
        "commits": [],
        "prs": [],
        "tasks_completed": [],
        "tasks_in_progress": [],
    })

    for ev in events:
        key = ev.get("slack_id") or f"unmapped::{ev['alias']}"
        bucket = buckets[key]
        bucket["aliases"].add(ev["alias"])
        etype = ev["event"]
        if etype == "commit":
            bucket["commits"].append(ev)
        elif etype == "pr":
            bucket["prs"].append(ev)
        elif etype == "task_completed":
            bucket["tasks_completed"].append(ev)
        elif etype == "task_in_progress":
            bucket["tasks_in_progress"].append(ev)
    return buckets


# ---------------------------------------------------------------------------
# Rendering helpers
# ---------------------------------------------------------------------------

def generate_ai_summary(buckets: Dict[str, UserBucket], user_maps: Dict[str, Dict[str, str]], api_key: Optional[str] = None) -> Optional[str]:
    """Generate a concise AI summary of team activity using LLM."""
    if not GEMINI_AVAILABLE:
        print("WARNING: google-generativeai not available, skipping AI summary", file=sys.stderr)
        return None
    
    if not api_key:
        api_key = os.environ.get("GOOGLE_API_KEY")
        if not api_key:
            print("WARNING: No Gemini API key provided, skipping AI summary", file=sys.stderr)
            return None
    
    try:
        genai.configure(api_key=api_key)
        model = genai.GenerativeModel('gemini-2.5-pro-preview-05-06')
        
        # Prepare data for the AI
        activity_data = []
        key_items = []
        
        for slack_id, bucket in buckets.items():
            header_alias = _choose_header_alias(bucket, user_maps)
            user_summary = {"user": header_alias, "activities": []}
            
            # Collect commits
            for commit in bucket["commits"]:
                user_summary["activities"].append(f"Commit: {commit['message']} ({commit['url']})")
                key_items.append({"type": "commit", "url": commit["url"], "description": commit["message"]})
            
            # Collect PRs
            for pr in bucket["prs"]:
                user_summary["activities"].append(f"PR #{pr['number']}: {pr['title']} ({pr['state']}) ({pr['url']})")
                key_items.append({"type": "pr", "url": pr["url"], "description": f"PR #{pr['number']}: {pr['title']}"})
            
            # Collect completed tasks
            for task in bucket["tasks_completed"]:
                user_summary["activities"].append(f"Completed Task: {task['title']} ({task['url']})")
                key_items.append({"type": "task", "url": task["url"], "description": f"Completed: {task['title']}"})
            
            # Collect in-progress tasks
            for task in bucket["tasks_in_progress"]:
                user_summary["activities"].append(f"In-Progress Task: {task['title']} ({task['url']})")
        
            if user_summary["activities"]:
                activity_data.append(user_summary)
        
        if not activity_data:
            return None
        
        # Create prompt for Gemini
        prompt = f"""
Analyze this team's development activity and create a concise executive summary in 3-5 sentences.
Focus on the most impactful work and include markdown links to the most important items. All statements in your summary must be directly verifiable from the 'Team Activity Data' provided below.

Team Activity Data:
{json.dumps(activity_data, indent=2)}

Requirements:
- The summary must be between 3 and 7 sentences long.
- You MUST NOT invent any information, links, or facts not explicitly present in the 'Team Activity Data'.
- If specific details for a requested aspect (e.g., categorization by product features) are not clearly available in the data, explicitly state that the information is not found or cannot be determined from the provided context.
- Include markdown links to the most important PRs (GitHub URLs), commits (GitHub URLs), or ClickUp tasks (identified by their URLs or task IDs like NFI-<number>) mentioned in your summary.
- Summarize key achievements and overall progress based *only* on the activities listed in the data. Categorize these by themes like product features, bug fixes, or other relevant groupings *only if* such categories are clearly supported by the task titles and commit messages in the data.
- Use a professional tone suitable for team updates.
- Make it engaging and highlight the team's productivity, grounded in the provided data.

Format the response as plain markdown text without any code blocks or formatting marks.
"""
        
        generation_config = genai.types.GenerationConfig(temperature=0.1)
        response = model.generate_content(
            prompt,
            generation_config=generation_config
        )
        
        if response and response.text:
            return response.text.strip()
        else:
            print("WARNING: Empty response from Gemini API", file=sys.stderr)
            return None
            
    except Exception as exc:
        print(f"WARNING: Failed to generate AI summary: {exc}", file=sys.stderr)
        return None

def _choose_header_alias(bucket: UserBucket, user_maps: Dict[str, Dict[str, str]]) -> str:
    # prefer github alias then email
    github_aliases = [a for a in bucket["aliases"] if a in user_maps["github"]]
    if github_aliases:
        login = github_aliases[0]
        return f"@[{login}](https://github.com/{login})"

    email_aliases = [a for a in bucket["aliases"] if a in user_maps["email"]]
    if email_aliases:
        email = email_aliases[0]
        display = _derive_name_from_email(email)
        return f"@[{display}](mailto:{email})"

    # Fallback plain alias (will not mention)
    return f"{next(iter(bucket['aliases']))}"


def render_markdown(
    buckets: Dict[str, UserBucket],
    user_maps: Dict[str, Dict[str, str]],
    ai_summary: Optional[str] = None,
    timespan_description: Optional[str] = None
) -> str:
    lines: List[str] = []
    
    base_title = "Activity Summary"
    if timespan_description:
        full_title = f"{timespan_description} {base_title}"
    else:
        full_title = base_title
    lines.append(f"# {full_title}  ({datetime.utcnow():%Y-%m-%d %H:%M UTC})\n")

    # Add AI summary if available
    if ai_summary:
        lines.append("## 🤖 AI Team Summary")
        lines.append("")
        lines.append(ai_summary)
        lines.append("")
        lines.append("---")
        lines.append("")
        lines.append("## 📊 Detailed Activity")
        lines.append("")

    for slack_id, bucket in buckets.items():
        header_alias = _choose_header_alias(bucket, user_maps)
        lines.append(f"### {header_alias}")

        # Commits
        if bucket["commits"]:
            lines.append("- *Commits*  ")
            for ev in bucket["commits"]:
                lines.append(f"  - [{ev['sha']}]({ev['url']}) {ev['message']}")

        # PRs
        if bucket["prs"]:
            lines.append("- *Pull Requests*  ")
            for ev in bucket["prs"]:
                num = ev["number"]
                title = ev["title"]
                state = ev["state"]
                lines.append(f"  - [PR #{num}]({ev['url']}) {title} ({state})")

        # Tasks completed
        if bucket["tasks_completed"]:
            lines.append("- *Tasks Completed*  ")
            for ev in bucket["tasks_completed"]:
                task_id = ev["task_id"]
                title = ev["title"]
                lines.append(f"  - [{task_id}]({ev['url']}) {title}")

        # Tasks in progress
        if bucket["tasks_in_progress"]:
            lines.append("- *Tasks In-Progress*  ")
            for ev in bucket["tasks_in_progress"]:
                task_id = ev["task_id"]
                title = ev["title"]
                lines.append(f"  - [{task_id}]({ev['url']}) {title}")

        lines.append("")  # blank line between users

    return "\n".join(lines).rstrip() + "\n"


# ---------------------------------------------------------------------------
# Main CLI
# ---------------------------------------------------------------------------

def main() -> None:
    ap = argparse.ArgumentParser(description="Summarise GitHub & ClickUp activity into Slack-ready markdown.")
    ap.add_argument("glob", nargs="+", help="Glob pattern(s) for JSON files, e.g. output/*72h*.json")
    ap.add_argument("--user-map", default="user_map.yml", help="Path to user_map.yml")
    ap.add_argument("--clickup-team-id", default=DEFAULT_CLICKUP_TEAM_ID, help="ClickUp team/workspace ID")
    ap.add_argument("--output", help="Write markdown to this path instead of stdout")
    ap.add_argument("--gemini-api-key", help="Gemini API key for AI summary (or set GOOGLE_API_KEY env var)")
    ap.add_argument("--no-ai-summary", action="store_true", help="Skip AI summary generation")
    args = ap.parse_args()

    user_maps = load_user_map(args.user_map)
    files = discover_files(args.glob)
    if not files:
        print("No JSON files matched the provided globs", file=sys.stderr)
        sys.exit(1)

    all_events: List[Event] = []
    for path in files:
        all_events.extend(parse_file(path, user_maps, args.clickup_team_id))

    buckets = aggregate(all_events)
    
    # Derive timespan string from glob patterns or discovered filenames
    timespan_description = derive_timespan_string(args.glob, files)

    # Generate AI summary unless disabled
    ai_summary = None
    if not args.no_ai_summary:
        ai_summary = generate_ai_summary(buckets, user_maps, args.gemini_api_key)
    
    markdown = render_markdown(buckets, user_maps, ai_summary, timespan_description)

    if args.output:
        Path(args.output).write_text(markdown, encoding="utf-8")
    else:
        print(markdown)


if __name__ == "__main__":
    main()