"""
Google OAuth2 授权脚本 - 在有浏览器的电脑上运行
运行后会打开浏览器让你登录Google账号授权
授权完成后会输出 refresh_token，把它填到 .env 文件里
"""
import http.server
import urllib.parse
import urllib.request
import json
import webbrowser
import threading
CLIENT_ID = "248230709364-fl5fha3poilomghbcf3n037cqq8vqpkv.apps.googleusercontent.com"
CLIENT_SECRET = "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"
REDIRECT_URI = "http://localhost:8888"
SCOPES = [
"https://www.googleapis.com/auth/gmail.modify",
"https://www.googleapis.com/auth/gmail.send",
"https://www.googleapis.com/auth/calendar",
"https://www.googleapis.com/auth/tasks",
"https://www.googleapis.com/auth/documents",
"https://www.googleapis.com/auth/drive",
"https://www.googleapis.com/auth/contacts.readonly",
]
class AuthHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
query = urllib.parse.urlparse(self.path).query
params = urllib.parse.parse_qs(query)
if "code" not in params:
self.send_response(400)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
error = params.get("error", ["unknown"])[0]
self.wfile.write(f"授权失败: {error}
".encode())
return
code = params["code"][0]
# Exchange code for tokens
token_data = urllib.parse.urlencode({
"code": code,
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"redirect_uri": REDIRECT_URI,
"grant_type": "authorization_code",
}).encode()
req = urllib.request.Request(
"https://oauth2.googleapis.com/token",
data=token_data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
try:
with urllib.request.urlopen(req) as resp:
tokens = json.loads(resp.read().decode())
refresh_token = tokens.get("refresh_token", "")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(f"""
✅ 授权成功！
请把下面这行添加到服务器的 .env 文件中：
GOOGLE_REFRESH_TOKEN={refresh_token}
可以关闭此页面了。
""".encode())
print("\n" + "="*60)
print("授权成功！refresh_token:")
print(refresh_token)
print("="*60)
print("\n请把这行加到 .env:")
print(f"GOOGLE_REFRESH_TOKEN={refresh_token}")
print()
except Exception as e:
self.send_response(500)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(f"换取token失败: {e}
".encode())
threading.Thread(target=self.server.shutdown).start()
def log_message(self, format, *args):
pass
if __name__ == "__main__":
auth_params = urllib.parse.urlencode({
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"response_type": "code",
"scope": " ".join(SCOPES),
"access_type": "offline",
"prompt": "consent",
})
auth_url = f"https://accounts.google.com/o/oauth2/v2/auth?{auth_params}"
print("正在打开浏览器进行授权...")
print(f"如果浏览器没有自动打开，请手动访问:\n{auth_url}\n")
webbrowser.open(auth_url)
server = http.server.HTTPServer(("localhost", 8888), AuthHandler)
print("等待授权回调...")
server.serve_forever()