重定向后如何在 Python 脚本中获取 OAuth 授权？

Question

我需要编写一个 Python 脚本来运行 OAuth 2.0 客户端流程。在流程中，我需要让用户运行 Python 脚本，打开浏览器以登录并授予访问权限，然后获取授权以换取访问令牌并继续流程。我一直在搜索，但没有找到关于如何执行此操作的好答案，因为大多数示例不涉及实际的 OAuth 登录或需要复制/粘贴。有没有一种干净的方法可以将流程无缝集成到我的 Python 代码中以供本地使用？

我曾短暂尝试过线程和http来创建本地主机服务器来获取请求，尝试过flask，并尝试查看创建协议处理程序是否有帮助。我在所有这些方面都取得了一些成功，并且能够完成流程，但似乎仍然无法从重定向 uri 中提取授权代码。该脚本将在内部运行并且不会分发，但至少需要足够干净，以便不必进行复制/粘贴。

Answer 1

以下内容将创建一个临时本地主机，并允许从 URL 中抓取授权授予以及检索和返回访问令牌。

import requests
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import webbrowser

access_token = None

# Simple HTTP request handler to capture the authorization code
class AuthorizationHandler(BaseHTTPRequestHandler):
    authorization_code = None

    def do_GET(self):
        global access_token
        if self.path.startswith("/oauth-callback?"):
            # Extract the authorization code from the query parameters
            authorization_code = self.path.split("&code=")[1]

            # Display the authorization code
            print("Authorization Code:", self.authorization_code)

            # Send a response to the browser
            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"<h1>Authorization Code Received</h1>")

            # Obtain OAuth Client Access Token
            headers = {'accept':'*/*'}
            parameters = {'client_id':'{{Client_ID}}','client_secret':'{{Client_Secret}}','access_code':authorization_code}
            response = requests.get("{{tokenURL}}",headers=headers,params=parameters)
            print(response.text)
            token = response.text

            # Log in using Access Token from OAuth
            headers = {'accept':'*/*','Content-Type':'application/json'}
            body = {'token':token}
            response = requests.post("{{loginURL}}",headers=headers,json=body)
            print(response.text)
            json_response = response.json()
            access_token= json_response["token"]

            # Exit the function after obtaining the token
            raise SystemExit

# Start a temporary HTTP server in a separate thread
def start_temp_server():
    server = HTTPServer(("localhost", 8000), AuthorizationHandler)
    server.serve_forever()

# Example usage
authorization_url = "{{AuthorizationURL}}"

# Start the temporary server in a separate thread
server_thread = threading.Thread(target=start_temp_server)
server_thread.start()

# Open the authorization URL in the default web browser
webbrowser.open(authorization_url)

server_thread.join()