Google Sheets

keyring>=24.0.0

#!/usr/bin/env python3
"""
OAuth token management for Google Sheets API.
Standalone authentication - does not require the MCP server.
Cross-platform support using keyring library for secure credential 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"

# Keychain configuration
KEYCHAIN_SERVICE = "google-sheets-skill-oauth"
KEYCHAIN_ACCOUNT = "main-account"

# Google Sheets requires these scopes (read/write)
SCOPES = [
    "https://www.googleapis.com/auth/spreadsheets",
    "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 secure storage using keyring."""
    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 secure storage using keyring."""
    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 secure storage using keyring."""
    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 Sheets 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 Sheets 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

SHEETS_API_BASE = "https://sheets.googleapis.com/v4"
DRIVE_API_BASE = "https://www.googleapis.com/drive/v3"

# MIME type for Google Sheets
SHEETS_MIME_TYPE = "application/vnd.google-apps.spreadsheet"


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

    url = f"{base_url}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)

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

    try:
        body = json.dumps(data).encode('utf-8')
        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_spreadsheet_id(spreadsheet_id_or_url: str) -> str:
    """
    Extract spreadsheet ID from a URL or return the ID as-is.

    Handles URLs like:
    - https://docs.google.com/spreadsheets/d/SPREADSHEET_ID/edit
    - https://docs.google.com/spreadsheets/d/SPREADSHEET_ID
    """
    # Pattern to match Google Sheets URLs
    url_pattern = r'docs\.google\.com/spreadsheets/d/([a-zA-Z0-9_-]+)'
    match = re.search(url_pattern, spreadsheet_id_or_url)
    if match:
        return match.group(1)
    return spreadsheet_id_or_url


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

    url = f"{base_url}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)

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

    try:
        req = urllib.request.Request(url, headers=headers, method="GET")
        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 get_text(spreadsheet_id: str, output_format: str = "text") -> dict:
    """
    Get spreadsheet content in various formats.

    Args:
        spreadsheet_id: Spreadsheet ID or URL
        output_format: 'text', 'csv', or 'json'
    """
    sheet_id = extract_spreadsheet_id(spreadsheet_id)

    # First, get spreadsheet metadata
    metadata = api_request(SHEETS_API_BASE, f"spreadsheets/{sheet_id}", {
        "includeGridData": "false"
    })

    if "error" in metadata:
        return metadata

    title = metadata.get("properties", {}).get("title", "Untitled")
    sheets = metadata.get("sheets", [])

    content = ""
    json_data = {}

    if output_format != "json":
        content += f"Spreadsheet Title: {title}\n\n"

    # Get data from each sheet
    for sheet in sheets:
        sheet_name = sheet.get("properties", {}).get("title", "")
        if not sheet_name:
            continue

        # Get values for this sheet
        values_result = api_request(SHEETS_API_BASE, f"spreadsheets/{sheet_id}/values/{urllib.parse.quote(sheet_name)}")

        if "error" in values_result:
            if output_format == "json":
                continue
            content += f"Sheet Name: {sheet_name}\n(Error reading sheet)\n\n"
            continue

        values = values_result.get("values", [])

        if output_format == "json":
            json_data[sheet_name] = values
        else:
            content += f"Sheet Name: {sheet_name}\n"

            if not values:
                content += "(Empty sheet)\n"
            else:
                for row in values:
                    if output_format == "csv":
                        # Convert to CSV format
                        csv_row = []
                        for cell in row:
                            cell_str = str(cell) if cell else ""
                            if "," in cell_str or '"' in cell_str or "\n" in cell_str:
                                csv_row.append(f'"{cell_str.replace(chr(34), chr(34)+chr(34))}"')
                            else:
                                csv_row.append(cell_str)
                        content += ",".join(csv_row) + "\n"
                    else:
                        # Plain text format with pipe separators
                        content += " | ".join(str(cell) if cell else "" for cell in row) + "\n"

            content += "\n"

    if output_format == "json":
        return {"data": json_data}

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


def get_range(spreadsheet_id: str, range_notation: str) -> dict:
    """
    Get values from a specific range.

    Args:
        spreadsheet_id: Spreadsheet ID or URL
        range_notation: A1 notation (e.g., 'Sheet1!A1:B10', 'A1:C5')
    """
    sheet_id = extract_spreadsheet_id(spreadsheet_id)

    result = api_request(SHEETS_API_BASE, f"spreadsheets/{sheet_id}/values/{urllib.parse.quote(range_notation)}")

    if "error" in result:
        return result

    return {
        "range": result.get("range"),
        "values": result.get("values", [])
    }


def find_spreadsheets(query: str, page_size: int = 10, page_token: Optional[str] = None) -> dict:
    """
    Find spreadsheets by search query.

    Args:
        query: Search query (searches name and content)
        page_size: Number of results to return
        page_token: Pagination token
    """
    # Build Drive API search query
    # Search for spreadsheets matching the query in name or fullText
    search_query = f"mimeType='{SHEETS_MIME_TYPE}' and (name contains '{query}' or fullText contains '{query}')"

    params = {
        "q": search_query,
        "pageSize": str(page_size),
        "fields": "nextPageToken, files(id, name, createdTime, modifiedTime, webViewLink)"
    }

    if page_token:
        params["pageToken"] = page_token

    result = api_request(DRIVE_API_BASE, "files", params)

    if "error" in result:
        return result

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


def get_metadata(spreadsheet_id: str) -> dict:
    """
    Get spreadsheet metadata.

    Args:
        spreadsheet_id: Spreadsheet ID or URL
    """
    sheet_id = extract_spreadsheet_id(spreadsheet_id)

    result = api_request(SHEETS_API_BASE, f"spreadsheets/{sheet_id}", {
        "includeGridData": "false"
    })

    if "error" in result:
        return result

    metadata = {
        "spreadsheetId": result.get("spreadsheetId"),
        "title": result.get("properties", {}).get("title"),
        "locale": result.get("properties", {}).get("locale"),
        "timeZone": result.get("properties", {}).get("timeZone"),
        "sheets": []
    }

    for sheet in result.get("sheets", []):
        props = sheet.get("properties", {})
        grid_props = props.get("gridProperties", {})
        metadata["sheets"].append({
            "sheetId": props.get("sheetId"),
            "title": props.get("title"),
            "index": props.get("index"),
            "rowCount": grid_props.get("rowCount"),
            "columnCount": grid_props.get("columnCount")
        })

    return metadata


def quote_range(range_notation: str) -> str:
    """URL-encode a range notation, matching the read API encoding."""
    return urllib.parse.quote(range_notation)


def update_range(spreadsheet_id: str, range_notation: str, values: list, value_input_option: str = "USER_ENTERED") -> dict:
    """
    Update a range of cells with new values.

    Args:
        spreadsheet_id: Spreadsheet ID or URL
        range_notation: A1 notation (e.g., 'Sheet1!A1:B10')
        values: 2D list of values, e.g. [["A1","B1"],["A2","B2"]]
        value_input_option: USER_ENTERED (parses formulas/numbers) or RAW
    """
    sheet_id = extract_spreadsheet_id(spreadsheet_id)
    data = {
        "range": range_notation,
        "majorDimension": "ROWS",
        "values": values
    }
    return api_write_request(
        SHEETS_API_BASE,
        f"spreadsheets/{sheet_id}/values/{quote_range(range_notation)}",
        data,
        method="PUT",
        params={"valueInputOption": value_input_option}
    )


def append_rows(spreadsheet_id: str, range_notation: str, values: list, value_input_option: str = "USER_ENTERED") -> dict:
    """
    Append rows after the last row with data in a sheet.

    Args:
        spreadsheet_id: Spreadsheet ID or URL
        range_notation: A1 notation for the target sheet/range (e.g., 'Sheet1!A:Z')
        values: 2D list of row values to append
        value_input_option: USER_ENTERED or RAW
    """
    sheet_id = extract_spreadsheet_id(spreadsheet_id)
    data = {
        "range": range_notation,
        "majorDimension": "ROWS",
        "values": values
    }
    return api_write_request(
        SHEETS_API_BASE,
        f"spreadsheets/{sheet_id}/values/{quote_range(range_notation)}:append",
        data,
        method="POST",
        params={
            "valueInputOption": value_input_option,
            "insertDataOption": "INSERT_ROWS"
        }
    )


def clear_range(spreadsheet_id: str, range_notation: str) -> dict:
    """
    Clear values from a range of cells.

    Args:
        spreadsheet_id: Spreadsheet ID or URL
        range_notation: A1 notation (e.g., 'Sheet1!A1:B10')
    """
    sheet_id = extract_spreadsheet_id(spreadsheet_id)
    return api_write_request(
        SHEETS_API_BASE,
        f"spreadsheets/{sheet_id}/values/{quote_range(range_notation)}:clear",
        {},
        method="POST"
    )


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

    Args:
        spreadsheet_id: Spreadsheet ID or URL
        requests: List of request objects (see Google Sheets API batchUpdate docs)
    """
    sheet_id = extract_spreadsheet_id(spreadsheet_id)
    return api_write_request(
        SHEETS_API_BASE,
        f"spreadsheets/{sheet_id}:batchUpdate",
        {"requests": requests},
        method="POST"
    )


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

    # get-text
    get_text_parser = subparsers.add_parser("get-text", help="Get spreadsheet content")
    get_text_parser.add_argument("spreadsheet_id", help="Spreadsheet ID or URL")
    get_text_parser.add_argument("--format", choices=["text", "csv", "json"], default="text",
                                  help="Output format (default: text)")

    # get-range
    get_range_parser = subparsers.add_parser("get-range", help="Get values from a specific range")
    get_range_parser.add_argument("spreadsheet_id", help="Spreadsheet ID or URL")
    get_range_parser.add_argument("range", help="A1 notation range (e.g., 'Sheet1!A1:B10')")

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

    # get-metadata
    get_metadata_parser = subparsers.add_parser("get-metadata", help="Get spreadsheet metadata")
    get_metadata_parser.add_argument("spreadsheet_id", help="Spreadsheet ID or URL")

    # update-range
    update_range_parser = subparsers.add_parser("update-range", help="Update a range of cells")
    update_range_parser.add_argument("spreadsheet_id", help="Spreadsheet ID or URL")
    update_range_parser.add_argument("range", help="A1 notation range (e.g., 'Sheet1!A1:B2')")
    update_range_parser.add_argument("values", help="JSON 2D array of values (e.g., '[[\"A1\",\"B1\"],[\"A2\",\"B2\"]]')")
    update_range_parser.add_argument("--raw", action="store_true", help="Use RAW input (no formula parsing)")

    # append-rows
    append_parser = subparsers.add_parser("append-rows", help="Append rows to a sheet")
    append_parser.add_argument("spreadsheet_id", help="Spreadsheet ID or URL")
    append_parser.add_argument("range", help="Target sheet range (e.g., 'Sheet1!A:Z')")
    append_parser.add_argument("values", help="JSON 2D array of row values")
    append_parser.add_argument("--raw", action="store_true", help="Use RAW input (no formula parsing)")

    # clear-range
    clear_parser = subparsers.add_parser("clear-range", help="Clear values from a range")
    clear_parser.add_argument("spreadsheet_id", help="Spreadsheet ID or URL")
    clear_parser.add_argument("range", help="A1 notation range to clear")

    # batch-update
    batch_parser = subparsers.add_parser("batch-update", help="Execute batch update requests")
    batch_parser.add_argument("spreadsheet_id", help="Spreadsheet 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.spreadsheet_id, args.format)
    elif args.command == "get-range":
        result = get_range(args.spreadsheet_id, args.range)
    elif args.command == "find":
        result = find_spreadsheets(args.query, args.limit, args.page_token)
    elif args.command == "get-metadata":
        result = get_metadata(args.spreadsheet_id)
    elif args.command == "update-range":
        values = json.loads(args.values)
        input_option = "RAW" if args.raw else "USER_ENTERED"
        result = update_range(args.spreadsheet_id, args.range, values, input_option)
    elif args.command == "append-rows":
        values = json.loads(args.values)
        input_option = "RAW" if args.raw else "USER_ENTERED"
        result = append_rows(args.spreadsheet_id, args.range, values, input_option)
    elif args.command == "clear-range":
        result = clear_range(args.spreadsheet_id, args.range)
    elif args.command == "batch-update":
        requests_data = json.loads(args.requests)
        result = batch_update(args.spreadsheet_id, 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()