Python程序化访问Google认证的REST服务指南

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time

# 配置无头Chrome浏览器
options = webdriver.ChromeOptions()
options.add_argument('--headless') # 启用无头模式
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
# 可以添加用户代理等其他选项
# options.add_argument("user-agent=Mozilla/5.0...")

driver = webdriver.Chrome(options=options)

try:
    # 导航到需要Google登录的网站
    target_url = "https://your-target-website.com/login" # 替换为您的目标网站登录页
    driver.get(target_url)

    # 等待页面重定向到Google登录页面
    # 实际情况可能需要更复杂的等待条件，例如检查URL是否包含"accounts.google.com"
    WebDriverWait(driver, 20).until(
        EC.url_contains("accounts.google.com")
    )
    print("已重定向到Google登录页面")

    # 模拟输入Google邮箱
    email_input = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.ID, "identifierId"))
    )
    email_input.send_keys("your_google_email@gmail.com") # 替换为您的Google邮箱
    driver.find_element(By.ID, "identifierNext").click()

    # 等待并模拟输入密码
    password_input = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.NAME, "password"))
    )
    password_input.send_keys("your_google_password") # 替换为您的Google密码
    driver.find_element(By.ID, "passwordNext").click()

    # 等待登录完成，重定向回目标网站
    # 同样，实际情况可能需要更复杂的等待条件
    WebDriverWait(driver, 20).until(
        EC.url_contains("your-target-website.com") # 替换为您的目标网站域名
    )
    print("Google登录成功，已重定向回目标网站")

    # 现在您可以获取页面的内容或执行其他操作
    print("当前页面标题:", driver.title)
    # 获取所有Cookies，可以用于后续requests请求
    cookies = driver.get_cookies()
    print("获取到的Cookies:", cookies)

    # 如果需要使用requests库继续访问，可以将Selenium获取的Cookies转换
    # import requests
    # s = requests.Session()
    # for cookie in cookies:
    #     s.cookies.set(cookie['name'], cookie['value'])
    # response = s.get("https://your-target-website.com/protected-data")
    # print(response.text)

except Exception as e:
    print(f"发生错误: {e}")
finally:
    driver.quit() # 关闭浏览器

import requests
import json
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import os
import pickle

# 定义OAuth2范围（scopes），根据您需要访问的API功能选择
# 例如，访问用户信息：'https://www.googleapis.com/auth/userinfo.email'
# 访问Google Drive：'https://www.googleapis.com/auth/drive.readonly'
SCOPES = ['https://www.googleapis.com/auth/userinfo.email', 'openid'] # 示例范围

# OAuth2客户端配置路径
# 假设您已从Google Cloud Console下载了credentials.json文件
CLIENT_SECRET_FILE = 'credentials.json' # 替换为您的凭据文件路径
TOKEN_PICKLE_FILE = 'token.pickle' # 用于存储和加载令牌的文件

def get_google_credentials():
    """获取Google OAuth2凭据，如果需要则进行用户授权"""
    creds = None
    # 尝试从本地文件加载已保存的凭据
    if os.path.exists(TOKEN_PICKLE_FILE):
        with open(TOKEN_PICKLE_FILE, 'rb') as token:
            creds = pickle.load(token)

    # 如果没有有效凭据或凭据已过期，则进行刷新或重新授权
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            print("刷新访问令牌...")
            creds.refresh(Request())
        else:
            print("需要用户授权，请在浏览器中完成授权...")
            flow = InstalledAppFlow.from_client_secrets_file(
                CLIENT_SECRET_FILE, SCOPES)
            # 在本地启动一个Web服务器来处理授权回调
            creds = flow.run_local_oauthserver(port=8080) # 确保此端口未被占用

        # 保存凭据以备将来使用
        with open(TOKEN_PICKLE_FILE, 'wb') as token:
            pickle.dump(creds, token)
            print("凭据已保存到", TOKEN_PICKLE_FILE)
    return creds

def call_protected_api(api_url, credentials, payload=None):
    """使用Access Token调用受保护的API"""
    if not credentials or not credentials.valid:
        raise ValueError("无效的Google凭据。")

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

    try:
        response = requests.post(api_url, headers=headers, json=payload)
        response.raise_for_status() # 对4xx/5xx错误抛出HTTPError
        return response.json()
    except requests.exceptions.HTTPError as err:
        print(f"HTTP错误发生: {err}")
        print(f"响应内容: {response.text}")
        raise
    except Exception as err:
        print(f"发生其他错误: {err}")
        raise

if __name__ == "__main__":
    try:
        # 1. 获取Google凭据 (可能需要浏览器交互)
        google_creds = get_google_credentials()
        print("成功获取Google凭据。")

        # 2. 定义您要访问的REST API端点
        # 这是一个示例，您需要替换为实际的API URL
        # 注意：此处假设您的目标API也使用Google OAuth2认证，
        # 并且您的Access Token具有访问该API的权限。
        target_api_url = "https://your-target-api.com/data" # 替换为您的目标API URL

        # 示例payload
        api_payload = {
            "query": "example",
            "limit": 10
        }

        # 3. 使用凭据调用受保护的API
        print(f"尝试调用API: {target_api_url}")
        api_response = call_protected_api(target_api_url, google_creds, payload=api_payload)
        print("API调用成功，响应如下:")
        print(json.dumps(api_response, indent=2, ensure_ascii=False))

    except Exception as e:
        print(f"程序执行失败: {e}")

{
  "web": {
    "client_id": "YOUR_CLIENT_ID.apps.googleusercontent.com",
    "project_id": "your-project-id",
    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
    "token_uri": "https://oauth2.googleapis.com/token",
    "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
    "client_secret": "YOUR_CLIENT_SECRET",
    "redirect_uris": ["http://localhost:8080"]
  }
}