claude-code/skills/morning-report/scripts/collectors/gmail.py

#!/usr/bin/env python3
"""Gmail collector using existing gmail skill."""

import os
import subprocess
import sys
from collections import defaultdict
from pathlib import Path


def fetch_unread_emails(days: int = 7, max_results: int = 15) -> list:
    """Fetch unread emails directly using gmail_mcp library."""
    # Set credentials path
    os.environ.setdefault(
        "GMAIL_CREDENTIALS_PATH", os.path.expanduser("~/.gmail-mcp/credentials.json")
    )

    try:
        # Add gmail venv to path
        venv_site = Path.home() / ".claude/mcp/gmail/venv/lib/python3.14/site-packages"
        if str(venv_site) not in sys.path:
            sys.path.insert(0, str(venv_site))

        from gmail_mcp.utils.GCP.gmail_auth import get_gmail_service

        service = get_gmail_service()
        results = (
            service.users()
            .messages()
            .list(
                userId="me", q=f"is:unread newer_than:{days}d", maxResults=max_results
            )
            .execute()
        )

        emails = []
        for msg in results.get("messages", []):
            detail = (
                service.users()
                .messages()
                .get(
                    userId="me",
                    id=msg["id"],
                    format="metadata",
                    metadataHeaders=["From", "Subject"],
                )
                .execute()
            )
            headers = {h["name"]: h["value"] for h in detail["payload"]["headers"]}
            emails.append(
                {
                    "from": headers.get("From", "Unknown"),
                    "subject": headers.get("Subject", "(no subject)"),
                    "id": msg["id"],
                }
            )

        return emails

    except Exception as e:
        return [{"error": str(e)}]


def triage_with_sonnet(emails: list) -> str:
    """Use Sonnet to triage and summarize emails."""
    if not emails or (len(emails) == 1 and "error" in emails[0]):
        error = emails[0].get("error", "Unknown error") if emails else "No data"
        return f"⚠️ Could not fetch emails: {error}"

    # Build email summary for Sonnet
    email_text = []
    for i, e in enumerate(emails[:10], 1):
        sender = e.get("from", "Unknown").split("<")[0].strip().strip('"')
        subject = e.get("subject", "(no subject)")[:80]
        email_text.append(f"{i}. From: {sender}\n   Subject: {subject}")

    email_context = "\n\n".join(email_text)

    prompt = f"""You are triaging emails for a morning report. Given these unread emails, provide a brief summary.

Format:
- First line: count and any urgent items (e.g., "5 unread, 1 urgent")
- Then list top emails with [!] for urgent, or plain bullet
- Keep each email to one line: sender - subject snippet (max 50 chars)
- Maximum 5 emails shown

Emails:
{email_context}

Output the formatted email section, nothing else."""

    try:
        result = subprocess.run(
            [
                "/home/will/.local/bin/claude",
                "--print",
                "--model",
                "sonnet",
                "-p",
                prompt,
            ],
            capture_output=True,
            text=True,
            timeout=60,
        )

        if result.returncode == 0 and result.stdout.strip():
            return result.stdout.strip()
    except Exception:
        pass

    # Fallback to basic format
    lines = [f"{len(emails)} unread"]
    for e in emails[:5]:
        sender = e.get("from", "Unknown").split("<")[0].strip().strip('"')[:20]
        subject = e.get("subject", "(no subject)")[:40]
        lines.append(f"  • {sender} - {subject}")

    return "\n".join(lines)


def collect(config: dict) -> dict:
    """Main collector entry point."""
    email_config = config.get("email", {})
    max_display = email_config.get("max_display", 5)
    use_triage = email_config.get("triage", True)

    emails = fetch_unread_emails(days=7, max_results=max_display + 10)

    if use_triage and emails and "error" not in emails[0]:
        formatted = triage_with_sonnet(emails)
    else:
        # Basic format or error
        if emails and "error" not in emails[0]:
            lines = [f"{len(emails)} unread"]
            for e in emails[:max_display]:
                sender = e.get("from", "Unknown").split("<")[0].strip().strip('"')[:20]
                subject = e.get("subject", "(no subject)")[:40]
                lines.append(f"  • {sender} - {subject}")
            formatted = "\n".join(lines)
        else:
            error = emails[0].get("error", "Unknown") if emails else "No data"
            formatted = f"⚠️ Could not fetch emails: {error}"

    has_error = emails and len(emails) == 1 and "error" in emails[0]

    return {
        "section": "Email",
        "icon": "📧",
        "content": formatted,
        "raw": emails if not has_error else None,
        "count": len(emails) if not has_error else 0,
        "error": emails[0].get("error") if has_error else None,
    }


if __name__ == "__main__":
    config = {"email": {"max_display": 5, "triage": True}}
    result = collect(config)
    print(f"## {result['icon']} {result['section']}")
    print(result["content"])