microsoft-graph-skill/scripts/graph_api.py

#!/usr/bin/env python3
"""Microsoft Graph API client for Outlook and OneDrive."""

import argparse
import json
import os
import sys
import requests

TOKEN_FILE = os.path.expanduser("~/.config/microsoft-graph-tokens.json")
GRAPH_BASE = "https://graph.microsoft.com/v1.0"

def load_tokens():
    """Load access tokens from file."""
    if not os.path.exists(TOKEN_FILE):
        print(f"❌ Token file not found: {TOKEN_FILE}")
        print("Run: python3 ~/.openclaw/skills/microsoft-graph/scripts/auth.py")
        sys.exit(1)
    with open(TOKEN_FILE, "r") as f:
        return json.load(f)

def get_headers():
    """Get authorization headers."""
    tokens = load_tokens()
    return {"Authorization": f"Bearer {tokens['access_token']}"}

def api_get(endpoint, params=None):
    """Make GET request to Graph API."""
    url = f"{GRAPH_BASE}{endpoint}"
    response = requests.get(url, headers=get_headers(), params=params)
    if response.status_code == 401:
        print("❌ Token expired. Run: python3 ~/.openclaw/skills/microsoft-graph/scripts/graph_api.py refresh")
        sys.exit(1)
    return response.json()

def list_folders(parent_id=None):
    """List mail folders."""
    if parent_id:
        data = api_get(f"/me/mailFolders/{parent_id}/childFolders", {"$top": 100})
    else:
        data = api_get("/me/mailFolders", {"$top": 100})
    
    folders = data.get("value", [])
    for f in folders:
        name = f["displayName"]
        count = f["totalItemCount"]
        children = f["childFolderCount"]
        print(f"📁 {name} ({count} items, {children} subfolders) [id: {f['id'][:20]}...]")
    return folders

def list_emails(folder_name="Inbox", limit=10):
    """List emails in a folder."""
    # Get folder ID
    data = api_get("/me/mailFolders", {"$top": 100})
    folder_id = None
    for f in data.get("value", []):
        if f["displayName"].lower() == folder_name.lower():
            folder_id = f["id"]
            break
    
    if not folder_id:
        print(f"❌ Folder not found: {folder_name}")
        return
    
    # Get messages
    messages = api_get(f"/me/mailFolders/{folder_id}/messages", {
        "$top": limit,
        "$select": "id,subject,from,receivedDateTime,isRead",
        "$orderby": "receivedDateTime desc"
    })
    
    for msg in messages.get("value", []):
        read = "✓" if msg["isRead"] else "○"
        sender = msg.get("from", {}).get("emailAddress", {}).get("address", "?")
        date = msg["receivedDateTime"][:10]
        subject = msg.get("subject", "(no subject)")[:50]
        print(f"{read} [{date}] {sender}: {subject}")
    
    return messages.get("value", [])

def get_email(message_id):
    """Get full email content."""
    msg = api_get(f"/me/messages/{message_id}")
    print(f"Subject: {msg.get('subject')}")
    print(f"From: {msg.get('from', {}).get('emailAddress', {}).get('address')}")
    print(f"Date: {msg.get('receivedDateTime')}")
    print(f"\n{msg.get('body', {}).get('content', '')[:2000]}")
    return msg

def list_onedrive(path="/"):
    """List OneDrive files."""
    if path == "/":
        data = api_get("/me/drive/root/children", {"$top": 50})
    else:
        data = api_get(f"/me/drive/root:/{path}:/children", {"$top": 50})
    
    items = data.get("value", [])
    for item in items:
        icon = "📁" if "folder" in item else "📄"
        size = item.get("size", 0)
        name = item["name"]
        print(f"{icon} {name} ({size} bytes)")
    return items

def refresh_token():
    """Refresh the access token."""
    import msal
    
    tokens = load_tokens()
    client_id = tokens.get("client_id", "14d82eec-204b-4c2f-b7e8-296a70dab67e")
    
    app = msal.PublicClientApplication(client_id, authority="https://login.microsoftonline.com/consumers")
    
    # Try to refresh
    accounts = app.get_accounts()
    if accounts:
        result = app.acquire_token_silent(tokens.get("scopes", []), account=accounts[0])
        if result and "access_token" in result:
            tokens["access_token"] = result["access_token"]
            with open(TOKEN_FILE, "w") as f:
                json.dump(tokens, f, indent=2)
            print("✅ Token refreshed!")
            return
    
    # Need re-auth
    print("❌ Cannot refresh. Please re-authenticate:")
    print("python3 ~/.openclaw/skills/microsoft-graph/scripts/auth.py")

def main():
    parser = argparse.ArgumentParser(description="Microsoft Graph API client")
    subparsers = parser.add_subparsers(dest="command", help="Commands")
    
    # folders
    folders_p = subparsers.add_parser("folders", help="List mail folders")
    folders_p.add_argument("--parent", help="Parent folder ID")
    
    # emails
    emails_p = subparsers.add_parser("emails", help="List emails")
    emails_p.add_argument("--folder", default="Indbakke", help="Folder name")
    emails_p.add_argument("--limit", type=int, default=10, help="Number of emails")
    
    # email
    email_p = subparsers.add_parser("email", help="Get email content")
    email_p.add_argument("--id", required=True, help="Message ID")
    
    # onedrive
    od_p = subparsers.add_parser("onedrive", help="List OneDrive files")
    od_p.add_argument("--path", default="/", help="Path")
    
    # refresh
    subparsers.add_parser("refresh", help="Refresh access token")
    
    args = parser.parse_args()
    
    if args.command == "folders":
        list_folders(args.parent)
    elif args.command == "emails":
        list_emails(args.folder, args.limit)
    elif args.command == "email":
        get_email(args.id)
    elif args.command == "onedrive":
        list_onedrive(args.path)
    elif args.command == "refresh":
        refresh_token()
    else:
        parser.print_help()

if __name__ == "__main__":
    main()