daa4de8832
Add comprehensive morning report skill with collectors for calendar, email, tasks, infrastructure status, news, stocks, and weather. Add stock lookup skill for quote queries.
143 lines
4.8 KiB
Python
Executable File
143 lines
4.8 KiB
Python
Executable File
#!/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.13/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"])
|