File size: 8,513 Bytes
dc4e6da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 | """
Helper script to get Google Drive OAuth token for testing.
This script implements the OAuth flow to get access and refresh tokens
from Google Drive API for testing purposes.
Prerequisites:
1. Google Cloud Project with Drive API enabled
2. OAuth 2.0 Client ID credentials
3. Add http://localhost:8080 as authorized redirect URI
Usage:
python test_get_google_token.py --client-id YOUR_CLIENT_ID --client-secret YOUR_CLIENT_SECRET
"""
import argparse
import webbrowser
from urllib.parse import urlencode, parse_qs
from http.server import HTTPServer, BaseHTTPRequestHandler
import requests
# Global variable to store authorization code
auth_code = None
class OAuthCallbackHandler(BaseHTTPRequestHandler):
"""HTTP server to handle OAuth callback"""
def do_GET(self):
global auth_code
# Parse query parameters
query = self.path.split('?', 1)[-1]
params = parse_qs(query)
if 'code' in params:
auth_code = params['code'][0]
# Send success response
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
html = """
<html>
<head><title>Authorization Successful</title></head>
<body style="font-family: Arial; text-align: center; padding: 50px;">
<h1 style="color: green;">✓ Authorization Successful!</h1>
<p>You can close this window and return to the terminal.</p>
</body>
</html>
"""
self.wfile.write(html.encode())
else:
# Error response
self.send_response(400)
self.send_header('Content-type', 'text/html')
self.end_headers()
error = params.get('error', ['Unknown error'])[0]
html = f"""
<html>
<head><title>Authorization Failed</title></head>
<body style="font-family: Arial; text-align: center; padding: 50px;">
<h1 style="color: red;">✗ Authorization Failed</h1>
<p>Error: {error}</p>
<p>Please try again.</p>
</body>
</html>
"""
self.wfile.write(html.encode())
def log_message(self, format, *args):
"""Suppress default logging"""
pass
def get_google_drive_token(client_id: str, client_secret: str, redirect_uri: str = "http://localhost:8080"):
"""
Get Google Drive OAuth tokens through OAuth flow.
Args:
client_id: Google OAuth client ID
client_secret: Google OAuth client secret
redirect_uri: OAuth redirect URI (must match Google Cloud Console)
Returns:
dict with 'access_token' and 'refresh_token'
"""
global auth_code
print("=" * 80)
print(" " * 20 + "GOOGLE DRIVE OAUTH TOKEN GENERATOR")
print("=" * 80)
print()
# Step 1: Generate authorization URL
auth_params = {
'client_id': client_id,
'redirect_uri': redirect_uri,
'response_type': 'code',
'scope': 'https://www.googleapis.com/auth/drive.file',
'access_type': 'offline', # Get refresh token
'prompt': 'consent' # Force consent to get refresh token
}
auth_url = f"https://accounts.google.com/o/oauth2/v2/auth?{urlencode(auth_params)}"
print("Step 1: Authorize with Google")
print("-" * 80)
print("\nOpening authorization URL in your browser...")
print("If it doesn't open automatically, copy this URL:\n")
print(auth_url)
print()
# Open browser
webbrowser.open(auth_url)
# Step 2: Start local server to receive callback
print("Step 2: Waiting for authorization...")
print("-" * 80)
print(f"Local server listening on {redirect_uri}")
print("Complete the authorization in your browser.")
print()
server = HTTPServer(('localhost', 8080), OAuthCallbackHandler)
# Wait for one request (the callback)
while auth_code is None:
server.handle_request()
server.server_close()
if not auth_code:
print("✗ Failed to get authorization code")
return None
print("✓ Authorization code received!")
print()
# Step 3: Exchange code for tokens
print("Step 3: Exchanging code for tokens...")
print("-" * 80)
token_url = "https://oauth2.googleapis.com/token"
token_data = {
'code': auth_code,
'client_id': client_id,
'client_secret': client_secret,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code'
}
try:
response = requests.post(token_url, data=token_data)
response.raise_for_status()
tokens = response.json()
print("✓ Tokens received!")
print()
print("=" * 80)
print(" " * 30 + "TOKENS")
print("=" * 80)
print()
print("Access Token:")
print(tokens['access_token'])
print()
if 'refresh_token' in tokens:
print("Refresh Token:")
print(tokens['refresh_token'])
print()
else:
print("⚠ No refresh token received (user may have authorized before)")
print(" To get a refresh token:")
print(" 1. Go to: https://myaccount.google.com/permissions")
print(" 2. Remove your app's access")
print(" 3. Run this script again")
print()
print("Expires In: {} seconds".format(tokens.get('expires_in', 'N/A')))
print()
# Show usage instructions
print("=" * 80)
print(" " * 25 + "USAGE INSTRUCTIONS")
print("=" * 80)
print()
print("Option 1: Use with test script directly")
print("-" * 80)
print("python test_async_api.py \\")
print(f" --google-token {tokens['access_token']}")
if 'refresh_token' in tokens:
print(f" --google-refresh-token {tokens['refresh_token']}")
print()
print("Option 2: Set environment variable")
print("-" * 80)
print(f"export GOOGLE_DRIVE_TOKEN=\"{tokens['access_token']}\"")
if 'refresh_token' in tokens:
print(f"export GOOGLE_DRIVE_REFRESH_TOKEN=\"{tokens['refresh_token']}\"")
print("python test_async_api.py")
print()
print("Option 3: Use in your frontend")
print("-" * 80)
print("Store these tokens in your frontend application and include them")
print("in API requests to /generate/async endpoint.")
print()
print("=" * 80)
return tokens
except Exception as e:
print(f"✗ Failed to exchange code for tokens: {e}")
if hasattr(e, 'response') and e.response:
print(f"Response: {e.response.text}")
return None
def main():
parser = argparse.ArgumentParser(
description="Get Google Drive OAuth token for testing"
)
parser.add_argument(
"--client-id",
type=str,
required=True,
help="Google OAuth Client ID"
)
parser.add_argument(
"--client-secret",
type=str,
required=True,
help="Google OAuth Client Secret"
)
parser.add_argument(
"--redirect-uri",
type=str,
default="http://localhost:8080",
help="OAuth redirect URI (default: http://localhost:8080)"
)
args = parser.parse_args()
print()
print("Prerequisites Check:")
print("-" * 80)
print(f"✓ Client ID: {args.client_id[:20]}...")
print(f"✓ Client Secret: {args.client_secret[:10]}...")
print(f"✓ Redirect URI: {args.redirect_uri}")
print()
print("Make sure you've added this redirect URI to your Google Cloud Console:")
print(" https://console.cloud.google.com/apis/credentials")
print()
input("Press Enter to continue...")
print()
tokens = get_google_drive_token(
client_id=args.client_id,
client_secret=args.client_secret,
redirect_uri=args.redirect_uri
)
if tokens:
print("✓ SUCCESS! Use the tokens above to test the async API.")
else:
print("✗ FAILED to get tokens. Please check your credentials and try again.")
if __name__ == "__main__":
main()
|