获取与刷新 Google 用户令牌

# Python 3 标准库；保存为 google_oauth.py，在自己电脑运行。
import json, os, secrets, time, webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlencode, urlparse, parse_qs
from urllib.request import Request, urlopen
from urllib.error import HTTPError

client_id = os.environ["GOOGLE_CLIENT_ID"]
client_secret = os.environ["GOOGLE_CLIENT_SECRET"]
scopes = os.environ.get("GOOGLE_SCOPES", "https://www.googleapis.com/auth/drive.metadata.readonly")
redirect_uri = "http://localhost:8765/oauth2callback"
state = secrets.token_urlsafe(32)
result = {}

class Callback(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass  # 不把回调里的授权码写入访问日志
    def do_GET(self):
        parsed = urlparse(self.path)
        if parsed.path != "/oauth2callback":
            self.send_response(404); self.end_headers(); return
        query = parse_qs(parsed.query)
        if query.get("state", [""])[0] != state:
            self.send_response(400); self.end_headers(); self.wfile.write(b"Invalid state"); return
        if "error" in query:
            result["error"] = query["error"][0]
        elif "code" in query:
            result["code"] = query["code"][0]
        else:
            result["error"] = "missing_authorization_code"
        self.send_response(200); self.end_headers()
        self.wfile.write(b"Authorization received. Return to your terminal.")

with HTTPServer(("127.0.0.1", 8765), Callback) as server:
    server.timeout = 1
    url = "https://accounts.google.com/o/oauth2/v2/auth?" + urlencode({
        "client_id": client_id, "redirect_uri": redirect_uri,
        "response_type": "code", "scope": scopes, "access_type": "offline",
        "include_granted_scopes": "true", "prompt": "consent", "state": state
    })
    print("在浏览器确认自己申请的权限：\n" + url)
    webbrowser.open(url)
    deadline = time.monotonic() + 300
    while not result and time.monotonic() < deadline:
        server.handle_request()
if not result:
    raise SystemExit("等待授权超时，请重新运行。")
if "error" in result:
    raise SystemExit("授权失败：" + result["error"])
body = urlencode({"client_id": client_id, "client_secret": client_secret,
                  "code": result["code"], "grant_type": "authorization_code",
                  "redirect_uri": redirect_uri}).encode()
request = Request("https://oauth2.googleapis.com/token", data=body,
                  headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST")
try:
    with urlopen(request, timeout=30) as response:
        tokens = json.load(response)
except HTTPError as exc:
    raise SystemExit(f"令牌交换失败 HTTP {exc.code}: {exc.read().decode()}")
if not tokens.get("access_token"):
    raise SystemExit("响应缺少 access_token")
# 令牌只保存在本机；不要上传此文件或提交 Git。
filename = "google-oauth-token.json"
fd = os.open(filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w") as output:
    json.dump(tokens, output, ensure_ascii=False, indent=2)
print("令牌已保存到本机 google-oauth-token.json；将 access_token 放入 GOOGLE_ACCESS_TOKEN。")
print("授权范围：", tokens.get("scope"), "有效秒数：", tokens.get("expires_in"))

# Python 3 标准库；先在自己终端设置下列环境变量。
import json, os
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import HTTPError
body = urlencode({"client_id": os.environ["GOOGLE_CLIENT_ID"],
                  "client_secret": os.environ["GOOGLE_CLIENT_SECRET"],
                  "refresh_token": os.environ["GOOGLE_REFRESH_TOKEN"],
                  "grant_type": "refresh_token"}).encode()
request = Request("https://oauth2.googleapis.com/token", data=body,
                  headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST")
try:
    with urlopen(request, timeout=30) as response:
        data = json.load(response)
except HTTPError as exc:
    raise SystemExit(f"刷新失败 HTTP {exc.code}: {exc.read().decode()}")
if not data.get("access_token"):
    raise SystemExit("响应缺少 access_token")
# 仅在自己的私密终端显示；生产系统应保存到安全的令牌存储。
print(json.dumps(data, ensure_ascii=False, indent=2))