Google Slides

keyring>=24.0.0

#!/usr/bin/env python3
"""
OAuth token management for Google Slides API.
Standalone authentication - does not require the MCP server.
Cross-platform support using keyring library for secure token storage.
"""

import http.server
import json
import secrets
import socket
import sys
import time
import webbrowser
from base64 import b64encode
from dataclasses import dataclass
from typing import Optional
from urllib.parse import urlencode, parse_qs, urlparse
import urllib.request
import urllib.error

import keyring

# OAuth Configuration - uses same cloud function as MCP server
CLIENT_ID = "338689075775-o75k922vn5fdl18qergr96rp8g63e4d7.apps.googleusercontent.com"
CLOUD_FUNCTION_URL = "https://google-workspace-extension.geminicli.com"
REFRESH_ENDPOINT = f"{CLOUD_FUNCTION_URL}/refreshToken"

# Keyring configuration
KEYCHAIN_SERVICE = "google-slides-skill-oauth"
KEYCHAIN_ACCOUNT = "main-account"

# Google Slides requires these scopes (read/write)
SCOPES = [
    "https://www.googleapis.com/auth/presentations",
    "https://www.googleapis.com/auth/drive",
    "https://www.googleapis.com/auth/userinfo.profile",
]

TOKEN_EXPIRY_BUFFER_MS = 5 * 60 * 1000  # 5 minutes


@dataclass
class TokenInfo:
    access_token: str
    refresh_token: Optional[str]
    expires_at: Optional[int]
    scope: Optional[str]


def get_available_port() -> int:
    """Find an available port for the OAuth callback server."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(('localhost', 0))
        return s.getsockname()[1]


def get_tokens_from_keychain() -> Optional[TokenInfo]:
    """Read OAuth tokens from keyring secure storage."""
    try:
        data_str = keyring.get_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT)
        if not data_str:
            return None
        data = json.loads(data_str)
        token = data.get("token", {})
        return TokenInfo(
            access_token=token.get("accessToken", ""),
            refresh_token=token.get("refreshToken"),
            expires_at=token.get("expiresAt"),
            scope=token.get("scope")
        )
    except (json.JSONDecodeError, keyring.errors.KeyringError):
        return None


def save_tokens_to_keychain(token_info: TokenInfo) -> bool:
    """Save OAuth tokens to keyring secure storage."""
    data = {
        "serverName": KEYCHAIN_ACCOUNT,
        "token": {
            "accessToken": token_info.access_token,
            "refreshToken": token_info.refresh_token,
            "tokenType": "Bearer",
            "scope": token_info.scope,
            "expiresAt": token_info.expires_at
        },
        "updatedAt": int(time.time() * 1000)
    }
    try:
        keyring.set_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT, json.dumps(data))
        return True
    except keyring.errors.KeyringError:
        return False


def clear_tokens_from_keychain() -> bool:
    """Clear OAuth tokens from keyring secure storage."""
    try:
        keyring.delete_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT)
        return True
    except (keyring.errors.PasswordDeleteError, keyring.errors.KeyringError):
        return False


def is_token_expired(token_info: TokenInfo) -> bool:
    """Check if the access token is expired or expiring soon."""
    if not token_info.expires_at:
        return False
    current_time_ms = int(time.time() * 1000)
    return token_info.expires_at < (current_time_ms + TOKEN_EXPIRY_BUFFER_MS)


def refresh_access_token(refresh_token: str) -> Optional[dict]:
    """Refresh access token using the cloud function."""
    try:
        data = json.dumps({"refresh_token": refresh_token}).encode('utf-8')
        req = urllib.request.Request(
            REFRESH_ENDPOINT,
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST"
        )

        with urllib.request.urlopen(req, timeout=30) as response:
            return json.loads(response.read().decode('utf-8'))
    except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError) as e:
        print(f"Error refreshing token: {e}", file=sys.stderr)
        return None


class OAuthCallbackHandler(http.server.BaseHTTPRequestHandler):
    """HTTP request handler for OAuth callback."""

    token_info: Optional[TokenInfo] = None
    csrf_token: str = ""
    error: Optional[str] = None

    def log_message(self, format: str, *args) -> None:  # noqa: A002
        """Suppress default logging."""
        _ = format, args  # Unused

    def do_GET(self):
        """Handle the OAuth callback."""
        parsed = urlparse(self.path)

        if not parsed.path.startswith("/oauth2callback"):
            self.send_error(404)
            return

        params = parse_qs(parsed.query)

        # Validate CSRF token
        state = params.get("state", [""])[0]
        if state != OAuthCallbackHandler.csrf_token:
            OAuthCallbackHandler.error = "State mismatch - possible CSRF attack"
            self.send_response(400)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"<html><body><h1>Error: State mismatch</h1></body></html>")
            return

        # Check for errors
        if "error" in params:
            OAuthCallbackHandler.error = params.get("error_description", params["error"])[0]
            self.send_response(400)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(f"<html><body><h1>Error: {OAuthCallbackHandler.error}</h1></body></html>".encode())
            return

        # Extract tokens
        access_token = params.get("access_token", [""])[0]
        refresh_token = params.get("refresh_token", [""])[0]
        expiry_date = params.get("expiry_date", [""])[0]
        scope = params.get("scope", [""])[0]

        if access_token:
            OAuthCallbackHandler.token_info = TokenInfo(
                access_token=access_token,
                refresh_token=refresh_token or None,
                expires_at=int(expiry_date) if expiry_date else None,
                scope=scope or None
            )

            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"""<html><body>
                <h1>Authentication successful!</h1>
                <p>You can close this window and return to the terminal.</p>
                <script>window.close();</script>
            </body></html>""")
        else:
            OAuthCallbackHandler.error = "No access token received"
            self.send_response(400)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"<html><body><h1>Error: No access token received</h1></body></html>")


def perform_oauth_flow() -> Optional[TokenInfo]:
    """Perform the full OAuth flow with browser-based authentication."""
    port = get_available_port()
    redirect_uri = f"http://localhost:{port}/oauth2callback"
    csrf_token = secrets.token_hex(32)

    # Build state payload for cloud function
    state_payload = {
        "uri": redirect_uri,
        "manual": False,
        "csrf": csrf_token
    }
    state = b64encode(json.dumps(state_payload).encode()).decode()

    # Build auth URL
    auth_params = {
        "client_id": CLIENT_ID,
        "redirect_uri": CLOUD_FUNCTION_URL,  # Cloud function handles the secret
        "response_type": "code",
        "scope": " ".join(SCOPES),
        "access_type": "offline",
        "prompt": "consent",
        "state": state
    }
    auth_url = f"https://accounts.google.com/o/oauth2/v2/auth?{urlencode(auth_params)}"

    # Set up callback handler
    OAuthCallbackHandler.csrf_token = csrf_token
    OAuthCallbackHandler.token_info = None
    OAuthCallbackHandler.error = None

    # Start local server
    server = http.server.HTTPServer(('localhost', port), OAuthCallbackHandler)
    server.timeout = 300  # 5 minute timeout

    print(f"Opening browser for authentication...", file=sys.stderr)
    print(f"If browser doesn't open, visit: {auth_url}", file=sys.stderr)

    webbrowser.open(auth_url)

    # Wait for callback
    print("Waiting for authentication...", file=sys.stderr)
    while OAuthCallbackHandler.token_info is None and OAuthCallbackHandler.error is None:
        server.handle_request()

    server.server_close()

    if OAuthCallbackHandler.error:
        print(f"Authentication failed: {OAuthCallbackHandler.error}", file=sys.stderr)
        return None

    return OAuthCallbackHandler.token_info


def get_valid_access_token(interactive: bool = True) -> Optional[str]:
    """
    Get a valid access token.

    Args:
        interactive: If True, will prompt for OAuth flow if no tokens exist.
                    If False, will return None if no valid tokens.
    """
    token_info = get_tokens_from_keychain()

    # No tokens - need to authenticate
    if not token_info or not token_info.access_token:
        if not interactive:
            return None

        print("No OAuth tokens found. Starting authentication...", file=sys.stderr)
        token_info = perform_oauth_flow()

        if not token_info:
            return None

        save_tokens_to_keychain(token_info)
        print("Authentication successful!", file=sys.stderr)
        return token_info.access_token

    # Check if token needs refresh
    if is_token_expired(token_info) and token_info.refresh_token:
        print("Access token expired, refreshing...", file=sys.stderr)
        new_tokens = refresh_access_token(token_info.refresh_token)

        if new_tokens:
            token_info.access_token = new_tokens.get("access_token", token_info.access_token)
            token_info.expires_at = new_tokens.get("expiry_date", token_info.expires_at)
            save_tokens_to_keychain(token_info)
            print("Token refreshed successfully.", file=sys.stderr)
        else:
            # Refresh failed - try re-authenticating if interactive
            if interactive:
                print("Token refresh failed. Re-authenticating...", file=sys.stderr)
                token_info = perform_oauth_flow()
                if token_info:
                    save_tokens_to_keychain(token_info)
                else:
                    return None
            else:
                print("Warning: Failed to refresh token.", file=sys.stderr)

    return token_info.access_token if token_info else None


def logout() -> bool:
    """Clear stored OAuth tokens."""
    return clear_tokens_from_keychain()


def main():
    """CLI for auth operations."""
    import argparse

    parser = argparse.ArgumentParser(description="Google Slides OAuth management")
    subparsers = parser.add_subparsers(dest="command")

    subparsers.add_parser("login", help="Authenticate with Google")
    subparsers.add_parser("logout", help="Clear stored tokens")
    subparsers.add_parser("token", help="Print current access token")
    subparsers.add_parser("status", help="Check authentication status")

    args = parser.parse_args()

    if args.command == "login":
        token_info = perform_oauth_flow()
        if token_info:
            save_tokens_to_keychain(token_info)
            print("Login successful!")
        else:
            print("Login failed.")
            sys.exit(1)

    elif args.command == "logout":
        if logout():
            print("Logged out successfully.")
        else:
            print("No tokens to clear.")

    elif args.command == "token":
        token = get_valid_access_token(interactive=False)
        if token:
            print(token)
        else:
            print("Not authenticated. Run: python auth.py login", file=sys.stderr)
            sys.exit(1)

    elif args.command == "status":
        token_info = get_tokens_from_keychain()
        if token_info and token_info.access_token:
            expired = is_token_expired(token_info)
            print(f"Status: Authenticated")
            print(f"Token expired: {expired}")
            if token_info.expires_at:
                expires_in = (token_info.expires_at - int(time.time() * 1000)) / 1000 / 60
                print(f"Expires in: {expires_in:.1f} minutes")
        else:
            print("Status: Not authenticated")
            print("Run: python auth.py login")
            sys.exit(1)

    else:
        parser.print_help()


if __name__ == "__main__":
    main()

keyring>=24.0.0

#!/usr/bin/env python3
"""
Google Slides API operations.
Lightweight alternative to the full Google Workspace MCP server.
"""

import argparse
import json
import re
import sys
import urllib.request
import urllib.error
import urllib.parse
from typing import Optional

from auth import get_valid_access_token

SLIDES_API_BASE = "https://slides.googleapis.com/v1"
DRIVE_API_BASE = "https://www.googleapis.com/drive/v3"


def extract_presentation_id(presentation_id_or_url: str) -> str:
    """Extract presentation ID from URL or return as-is if already an ID."""
    # Match Google Slides URL patterns
    patterns = [
        r'docs\.google\.com/presentation/d/([a-zA-Z0-9_-]+)',
        r'drive\.google\.com/.*?/d/([a-zA-Z0-9_-]+)',
    ]
    for pattern in patterns:
        match = re.search(pattern, presentation_id_or_url)
        if match:
            return match.group(1)
    return presentation_id_or_url


def api_request(method: str, url: str, data: Optional[dict] = None, params: Optional[dict] = None) -> dict:
    """Make an authenticated request to a Google API."""
    token = get_valid_access_token()
    if not token:
        return {"error": "Failed to get access token"}

    if params:
        url += "?" + urllib.parse.urlencode(params)

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    body = json.dumps(data).encode('utf-8') if data else None

    try:
        req = urllib.request.Request(url, data=body, headers=headers, method=method)
        with urllib.request.urlopen(req, timeout=30) as response:
            return json.loads(response.read().decode('utf-8'))
    except urllib.error.HTTPError as e:
        error_body = e.read().decode('utf-8') if e.fp else str(e)
        return {"error": f"HTTP {e.code}: {error_body}"}
    except urllib.error.URLError as e:
        return {"error": f"Request failed: {e.reason}"}
    except json.JSONDecodeError:
        return {"error": "Invalid JSON response"}


def extract_text_from_text_content(text_content: dict) -> str:
    """Extract plain text from a Slides text content object."""
    text = ''
    text_elements = text_content.get('textElements', [])
    for element in text_elements:
        if 'textRun' in element and element['textRun'].get('content'):
            text += element['textRun']['content']
        elif 'paragraphMarker' in element:
            text += '\n'
    return text


def get_text(presentation_id: str) -> dict:
    """Get all text content from a presentation."""
    pid = extract_presentation_id(presentation_id)

    params = {
        "fields": "title,slides(pageElements(shape(text,shapeProperties),table(tableRows(tableCells(text)))))"
    }

    result = api_request("GET", f"{SLIDES_API_BASE}/presentations/{pid}", params=params)

    if "error" in result:
        return result

    content = ''

    # Add presentation title
    if result.get('title'):
        content += f"Presentation Title: {result['title']}\n\n"

    # Process each slide
    slides = result.get('slides', [])
    for slide_index, slide in enumerate(slides):
        content += f"\n--- Slide {slide_index + 1} ---\n"

        page_elements = slide.get('pageElements', [])
        for element in page_elements:
            # Extract text from shapes
            if 'shape' in element and 'text' in element['shape']:
                shape_text = extract_text_from_text_content(element['shape']['text'])
                if shape_text.strip():
                    content += shape_text + '\n'

            # Extract text from tables
            if 'table' in element and 'tableRows' in element['table']:
                content += '\n--- Table Data ---\n'
                for row in element['table']['tableRows']:
                    row_text = []
                    for cell in row.get('tableCells', []):
                        if 'text' in cell:
                            cell_text = extract_text_from_text_content(cell['text'])
                            row_text.append(cell_text.strip())
                        else:
                            row_text.append('')
                    content += ' | '.join(row_text) + '\n'
                content += '--- End Table Data ---\n'

        content += '\n'

    return {"text": content.strip()}


def find_presentations(query: str, page_size: int = 10, page_token: Optional[str] = None) -> dict:
    """Find presentations by search query using Drive API."""
    # Build Drive search query for presentations
    mime_type = "application/vnd.google-apps.presentation"
    drive_query = f"mimeType='{mime_type}' and trashed=false"

    if query:
        # Escape single quotes in query
        escaped_query = query.replace("'", "\\'")
        drive_query += f" and (name contains '{escaped_query}' or fullText contains '{escaped_query}')"

    params = {
        "pageSize": page_size,
        "fields": "nextPageToken,files(id,name,modifiedTime,owners)",
        "q": drive_query
    }

    if page_token:
        params["pageToken"] = page_token

    result = api_request("GET", f"{DRIVE_API_BASE}/files", params=params)

    if "error" in result:
        return result

    return {
        "presentations": result.get("files", []),
        "nextPageToken": result.get("nextPageToken")
    }


def get_metadata(presentation_id: str) -> dict:
    """Get presentation metadata."""
    pid = extract_presentation_id(presentation_id)

    params = {
        "fields": "presentationId,title,slides(objectId),pageSize,notesMaster,masters,layouts"
    }

    result = api_request("GET", f"{SLIDES_API_BASE}/presentations/{pid}", params=params)

    if "error" in result:
        return result

    metadata = {
        "presentationId": result.get("presentationId"),
        "title": result.get("title"),
        "slideCount": len(result.get("slides", [])),
        "pageSize": result.get("pageSize"),
        "hasMasters": bool(result.get("masters")),
        "hasLayouts": bool(result.get("layouts")),
        "hasNotesMaster": bool(result.get("notesMaster"))
    }

    return metadata


def create_presentation(title: str) -> dict:
    """Create a new empty presentation."""
    data = {"title": title}
    return api_request("POST", f"{SLIDES_API_BASE}/presentations", data=data)


def add_slide(presentation_id: str, layout: str = "BLANK", insert_at: Optional[int] = None) -> dict:
    """
    Add a new slide to a presentation.

    Args:
        presentation_id: Presentation ID or URL
        layout: Predefined layout - BLANK, TITLE, TITLE_AND_BODY, TITLE_AND_TWO_COLUMNS,
                TITLE_ONLY, SECTION_HEADER, SECTION_TITLE_AND_DESCRIPTION, ONE_COLUMN_TEXT,
                MAIN_POINT, BIG_NUMBER
        insert_at: 0-based index to insert slide at (appends if not specified)
    """
    pid = extract_presentation_id(presentation_id)

    request = {
        "createSlide": {
            "slideLayoutReference": {
                "predefinedLayout": layout
            }
        }
    }

    if insert_at is not None:
        request["createSlide"]["insertionIndex"] = insert_at

    return api_request("POST", f"{SLIDES_API_BASE}/presentations/{pid}:batchUpdate",
                       data={"requests": [request]})


def replace_text(presentation_id: str, find_text: str, replace_with: str, match_case: bool = False) -> dict:
    """
    Find and replace text across all slides.

    Args:
        presentation_id: Presentation ID or URL
        find_text: Text to search for
        replace_with: Replacement text
        match_case: Whether the search is case-sensitive
    """
    pid = extract_presentation_id(presentation_id)

    request = {
        "replaceAllText": {
            "containsText": {
                "text": find_text,
                "matchCase": match_case
            },
            "replaceText": replace_with
        }
    }

    return api_request("POST", f"{SLIDES_API_BASE}/presentations/{pid}:batchUpdate",
                       data={"requests": [request]})


def delete_slide(presentation_id: str, slide_object_id: str) -> dict:
    """
    Delete a slide by its object ID.

    Use get-metadata to find slide object IDs.
    """
    pid = extract_presentation_id(presentation_id)

    request = {
        "deleteObject": {
            "objectId": slide_object_id
        }
    }

    return api_request("POST", f"{SLIDES_API_BASE}/presentations/{pid}:batchUpdate",
                       data={"requests": [request]})


def batch_update(presentation_id: str, requests: list) -> dict:
    """
    Execute batch update requests for advanced operations.

    Args:
        presentation_id: Presentation ID or URL
        requests: List of request objects (see Google Slides API batchUpdate docs)
    """
    pid = extract_presentation_id(presentation_id)
    return api_request("POST", f"{SLIDES_API_BASE}/presentations/{pid}:batchUpdate",
                       data={"requests": requests})


def main():
    parser = argparse.ArgumentParser(description="Google Slides API operations")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # get-text
    get_text_parser = subparsers.add_parser("get-text", help="Get text content from a presentation")
    get_text_parser.add_argument("presentation", help="Presentation ID or URL")

    # find
    find_parser = subparsers.add_parser("find", help="Find presentations by search query")
    find_parser.add_argument("query", help="Search query")
    find_parser.add_argument("--limit", type=int, default=10, help="Max results to return")
    find_parser.add_argument("--page-token", help="Pagination token")

    # get-metadata
    get_metadata_parser = subparsers.add_parser("get-metadata", help="Get presentation metadata")
    get_metadata_parser.add_argument("presentation", help="Presentation ID or URL")

    # create
    create_parser = subparsers.add_parser("create", help="Create a new presentation")
    create_parser.add_argument("title", help="Presentation title")

    # add-slide
    add_slide_parser = subparsers.add_parser("add-slide", help="Add a slide to a presentation")
    add_slide_parser.add_argument("presentation", help="Presentation ID or URL")
    add_slide_parser.add_argument("--layout", default="BLANK",
                                  help="Slide layout (BLANK, TITLE, TITLE_AND_BODY, TITLE_ONLY, SECTION_HEADER, etc.)")
    add_slide_parser.add_argument("--at", type=int, default=None, help="Insert position (0-based index)")

    # replace-text
    replace_text_parser = subparsers.add_parser("replace-text", help="Find and replace text across all slides")
    replace_text_parser.add_argument("presentation", help="Presentation ID or URL")
    replace_text_parser.add_argument("find", help="Text to find")
    replace_text_parser.add_argument("replace", help="Replacement text")
    replace_text_parser.add_argument("--match-case", action="store_true", help="Case-sensitive search")

    # delete-slide
    delete_slide_parser = subparsers.add_parser("delete-slide", help="Delete a slide by object ID")
    delete_slide_parser.add_argument("presentation", help="Presentation ID or URL")
    delete_slide_parser.add_argument("slide_id", help="Slide object ID (from get-metadata)")

    # batch-update
    batch_parser = subparsers.add_parser("batch-update", help="Execute batch update requests")
    batch_parser.add_argument("presentation", help="Presentation ID or URL")
    batch_parser.add_argument("requests", help="JSON array of batch update request objects")

    args = parser.parse_args()

    if args.command == "get-text":
        result = get_text(args.presentation)
    elif args.command == "find":
        result = find_presentations(args.query, args.limit, args.page_token)
    elif args.command == "get-metadata":
        result = get_metadata(args.presentation)
    elif args.command == "create":
        result = create_presentation(args.title)
    elif args.command == "add-slide":
        result = add_slide(args.presentation, args.layout, args.at)
    elif args.command == "replace-text":
        result = replace_text(args.presentation, args.find, args.replace, args.match_case)
    elif args.command == "delete-slide":
        result = delete_slide(args.presentation, args.slide_id)
    elif args.command == "batch-update":
        requests_data = json.loads(args.requests)
        result = batch_update(args.presentation, requests_data)
    else:
        result = {"error": f"Unknown command: {args.command}"}

    print(json.dumps(result, indent=2))

    if isinstance(result, dict) and "error" in result:
        sys.exit(1)


if __name__ == "__main__":
    main()