| |
| """ |
| Google Drive CLI — direkte REST API ohne MCP-Server. |
| |
| Token-Lifecycle: |
| 1. Token aus credentials-Datei lesen |
| 2. Prüfen ob expiry erreicht → wenn ja, refresh via oauth2.googleapis.com/token |
| 3. Drive API via curl aufrufen |
| |
| Fallback-Pfade: |
| - taylorwilsdon: /home/smlflg/Projekte/Connector/google-workspace/state/credentials/smlflg@gmail.com.json |
| - gdrive MCP: /home/smlflg/.hermes/gdrive-server-credentials.json |
| """ |
| import argparse |
| import json |
| import os |
| import sys |
| import subprocess |
| from datetime import datetime, timedelta, timezone |
| from urllib.request import Request, urlopen |
| from urllib.parse import urlencode |
| from urllib.error import HTTPError |
|
|
| |
| TOKEN_PATHS = [ |
| "/home/smlflg/Projekte/Connector/google-workspace/state/credentials/smlflg@gmail.com.json", |
| "/home/smlflg/.hermes/gdrive-server-credentials.json", |
| ] |
| OAUTH_ENDPOINT = "https://oauth2.googleapis.com/token" |
| DRIVE_API = "https://www.googleapis.com/drive/v3" |
| REFRESH_MARGIN_MINUTES = 15 |
|
|
|
|
| def load_credentials(): |
| """Load credentials from first available token file.""" |
| for path in TOKEN_PATHS: |
| if os.path.exists(path): |
| with open(path) as f: |
| return json.load(f), path |
| return None, None |
|
|
|
|
| def parse_expiry(expiry_str): |
| """Parse expiry string as naive local datetime (matching upstream format).""" |
| if not expiry_str: |
| return None |
| try: |
| dt = datetime.fromisoformat(expiry_str) |
| |
| if dt.tzinfo is not None: |
| dt = dt.replace(tzinfo=None) |
| return dt |
| except (ValueError, TypeError): |
| return None |
|
|
|
|
| def is_token_valid(creds): |
| """Check if token is valid or will expire within margin.""" |
| expiry = parse_expiry(creds.get("expiry")) |
| if expiry is None: |
| return False |
| now = datetime.now() |
| margin = timedelta(minutes=REFRESH_MARGIN_MINUTES) |
| return now < (expiry - margin) |
|
|
|
|
| def refresh_access_token(creds): |
| """Refresh the access token using refresh_token.""" |
| refresh_tok = creds.get("refresh_token") |
| if not refresh_tok: |
| return None |
|
|
| |
| env = {} |
| env_file = "/home/smlflg/Projekte/Connector/google-workspace/config/local.env" |
| if os.path.exists(env_file): |
| with open(env_file) as f: |
| for line in f: |
| line = line.strip() |
| if "=" in line: |
| k, _, v = line.partition("=") |
| env[k.strip()] = v.strip().strip('"').strip("'") |
|
|
| client_id = env.get("GOOGLE_OAUTH_CLIENT_ID") or creds.get("client_id") |
| client_secret = env.get("GOOGLE_OAUTH_CLIENT_SECRET") or creds.get("client_secret") |
|
|
| if not client_id or not client_secret: |
| print("ERROR: Missing client_id or client_secret", file=sys.stderr) |
| return None |
|
|
| data = urlencode({ |
| "client_id": client_id, |
| "client_secret": client_secret, |
| "refresh_token": refresh_tok, |
| "grant_type": "refresh_token", |
| }).encode() |
|
|
| req = Request(OAUTH_ENDPOINT, data=data, method="POST") |
| req.add_header("Content-Type", "application/x-www-form-urlencoded") |
|
|
| try: |
| with urlopen(req, timeout=30) as resp: |
| result = json.loads(resp.read().decode()) |
| except HTTPError as e: |
| body = e.read().decode("utf-8", errors="replace") |
| print(f"ERROR: Token refresh failed (HTTP {e.code}): {body[:200]}", file=sys.stderr) |
| return None |
|
|
| if "access_token" not in result: |
| print("ERROR: No access_token in refresh response", file=sys.stderr) |
| return None |
|
|
| return result |
|
|
|
|
| def get_valid_token(creds_path=None): |
| """Get a valid access token, refreshing if necessary. Returns (token, creds_dict, path).""" |
| if creds_path is None: |
| creds, path = load_credentials() |
| else: |
| if os.path.exists(creds_path): |
| with open(creds_path) as f: |
| creds = json.load(f) |
| path = creds_path |
| else: |
| creds, path = load_credentials() |
|
|
| if not creds: |
| print("ERROR: No credentials found", file=sys.stderr) |
| return None, None, None |
|
|
| |
| if not is_token_valid(creds): |
| result = refresh_access_token(creds) |
| if not result: |
| return None, None, None |
|
|
| |
| creds["token"] = result["access_token"] |
| new_expiry = datetime.now() + timedelta(seconds=result.get("expires_in", 3600)) |
| |
| creds["expiry"] = new_expiry.replace(tzinfo=None).isoformat() |
|
|
| if path: |
| with open(path, "w") as f: |
| json.dump(creds, f, indent=2) |
|
|
| return creds.get("token"), creds, path |
|
|
|
|
| def curl_drive(endpoint, token, output=None): |
| """Make a curl request to Drive API. Returns JSON dict or None.""" |
| url = f"{DRIVE_API}/{endpoint}" |
| env = os.environ.copy() |
| env["TOKEN"] = token |
|
|
| |
| cmd = ["curl", "-s", "-H", f"Authorization: Bearer {token}", url] |
| result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) |
|
|
| if result.returncode != 0: |
| print(f"ERROR: curl failed: {result.stderr}", file=sys.stderr) |
| return None |
|
|
| try: |
| return json.loads(result.stdout) |
| except json.JSONDecodeError: |
| return result.stdout |
|
|
|
|
| def cmd_status(args): |
| """Show token status.""" |
| creds, path = load_credentials() |
| if not creds: |
| print("No credentials found") |
| return 1 |
|
|
| expiry = parse_expiry(creds.get("expiry")) |
| now = datetime.now() |
|
|
| print(f"Credentials file: {path}") |
| print(f"Has access_token: {'Yes' if creds.get('token') else 'No'}") |
| print(f"Has refresh_token: {'Yes' if creds.get('refresh_token') else 'No'}") |
|
|
| if expiry: |
| valid = now < expiry |
| remaining = expiry - now |
| mins = int(remaining.total_seconds() / 60) |
| print(f"Expiry: {expiry.strftime('%Y-%m-%d %H:%M:%S')}") |
| print(f"Valid: {'Yes' if valid else 'No (expired)'}") |
| if valid: |
| print(f"Remaining: ~{mins} minutes") |
| else: |
| print("Expiry: Unknown") |
|
|
| return 0 |
|
|
|
|
| def cmd_list(args): |
| """List files in a folder.""" |
| token, _, path = get_valid_token() |
| if not token: |
| return 1 |
|
|
| folder_id = args.folder_id or "1Lii9ILFarnqX8_RMxjn2qwoOeGup6VfT" |
| fields = "files(id,name,mimeType,size,modifiedTime)" |
| query = urlencode({ |
| "q": f"'{folder_id}' in parents and trashed=false", |
| "fields": fields, |
| "pageSize": 100, |
| }) |
|
|
| result = curl_drive(f"files?{query}", token) |
| if result is None or isinstance(result, str): |
| print("ERROR: Failed to list files", file=sys.stderr) |
| return 1 |
|
|
| files = result.get("files", []) |
| if args.json: |
| print(json.dumps(files, indent=2)) |
| else: |
| print(f"{len(files)} files in folder {folder_id}:") |
| for f in files: |
| size = f.get("size", "?") |
| modified = f.get("modifiedTime", "?")[:10] |
| mime = f.get("mimeType", "?")[:40] |
| print(f" {f['name'][:50]:<50} {size:>8} bytes {modified} {mime}") |
|
|
| return 0 |
|
|
|
|
| def cmd_read(args): |
| """Read file content.""" |
| token, _, path = get_valid_token() |
| if not token: |
| return 1 |
|
|
| file_id = args.file_id |
| if not file_id: |
| print("ERROR: --file-id required", file=sys.stderr) |
| return 1 |
|
|
| url = f"{DRIVE_API}/files/{file_id}?alt=media" |
| cmd = ["curl", "-s", "-H", f"Authorization: Bearer {token}", url] |
| result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) |
|
|
| if result.returncode != 0: |
| print(f"ERROR: curl failed: {result.stderr}", file=sys.stderr) |
| return 1 |
|
|
| if args.output: |
| with open(args.output, "w") as f: |
| f.write(result.stdout) |
| print(f"Saved to {args.output}") |
| else: |
| print(result.stdout) |
|
|
| return 0 |
|
|
|
|
| def cmd_meta(args): |
| """Get file metadata.""" |
| token, _, path = get_valid_token() |
| if not token: |
| return 1 |
|
|
| file_id = args.file_id |
| if not file_id: |
| print("ERROR: --file-id required", file=sys.stderr) |
| return 1 |
|
|
| fields = "id,name,mimeType,size,modifiedTime,webViewLink" |
| result = curl_drive(f"files/{file_id}?fields={fields}", token) |
| if result is None: |
| print("ERROR: Failed to get metadata", file=sys.stderr) |
| return 1 |
|
|
| if args.json: |
| print(json.dumps(result, indent=2)) |
| else: |
| if isinstance(result, dict): |
| for k, v in result.items(): |
| print(f"{k}: {v}") |
| else: |
| print(result) |
|
|
| return 0 |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser( |
| description="Google Drive CLI — direkte REST API ohne MCP-Server", |
| formatter_class=argparse.RawDescriptionHelpFormatter, |
| epilog=""" |
| Commands: |
| status Show token status |
| list List files in folder |
| read Read file content |
| meta Get file metadata |
| |
| Examples: |
| drive-cli.py status |
| drive-cli.py list --folder-id 1Lii9ILFarnqX8_RMxjn2qwoOeGup6VfT |
| drive-cli.py read --file-id <id> --output /tmp/file.txt |
| drive-cli.py meta --file-id <id> |
| """ |
| ) |
| subparsers = parser.add_subparsers(dest="command", required=True) |
|
|
| |
| subparsers.add_parser("status", help="Show token status") |
|
|
| |
| p_list = subparsers.add_parser("list", help="List files in folder") |
| p_list.add_argument("--folder-id", help="Drive folder ID") |
| p_list.add_argument("--json", action="store_true", help="JSON output") |
|
|
| |
| p_read = subparsers.add_parser("read", help="Read file content") |
| p_read.add_argument("--file-id", required=True, help="Drive file ID") |
| p_read.add_argument("--output", help="Output file (default: stdout)") |
|
|
| |
| p_meta = subparsers.add_parser("meta", help="Get file metadata") |
| p_meta.add_argument("--file-id", required=True, help="Drive file ID") |
| p_meta.add_argument("--json", action="store_true", help="JSON output") |
|
|
| args = parser.parse_args() |
|
|
| if args.command == "status": |
| return cmd_status(args) |
| elif args.command == "list": |
| return cmd_list(args) |
| elif args.command == "read": |
| return cmd_read(args) |
| elif args.command == "meta": |
| return cmd_meta(args) |
| else: |
| parser.print_help() |
| return 1 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
