Spaces:
Runtime error
Runtime error
| """ | |
| 一次性脚本：通过浏览器 OAuth 授权获取 Bing Ads Refresh Token。 | |
| 运行方法: | |
| python _get_bing_token.py | |
| 运行前确保 .env 中已配置: | |
| BING_CLIENT_ID=你的Azure应用Client ID | |
| BING_CLIENT_SECRET=你的Azure应用Client Secret | |
| 运行后会自动打开浏览器进行微软登录授权，授权完成后自动获取 Refresh Token。 | |
| 将输出的 Refresh Token 填入 .env 的 BING_REFRESH_TOKEN 即可。 | |
| """ | |
| import os | |
| import sys | |
| import webbrowser | |
| from http.server import HTTPServer, BaseHTTPRequestHandler | |
| from urllib.parse import urlparse, parse_qs | |
| import requests | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| CLIENT_ID = os.getenv("BING_CLIENT_ID", "") | |
| CLIENT_SECRET = os.getenv("BING_CLIENT_SECRET", "") | |
| REDIRECT_URI = "http://localhost:8400/callback" | |
| SCOPES = "https://ads.microsoft.com/msads.manage offline_access" | |
| TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token" | |
| AUTH_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/authorize" | |
| if not CLIENT_ID or not CLIENT_SECRET: | |
| print("请先在 .env 中配置 BING_CLIENT_ID 和 BING_CLIENT_SECRET") | |
| sys.exit(1) | |
| class CallbackHandler(BaseHTTPRequestHandler): | |
| def do_GET(self): | |
| qs = parse_qs(urlparse(self.path).query) | |
| code = qs.get("code", [None])[0] | |
| if code: | |
| self.server.auth_code = code | |
| self.send_response(200) | |
| self.send_header("Content-Type", "text/html; charset=utf-8") | |
| self.end_headers() | |
| self.wfile.write("授权成功！可以关闭此页面。".encode("utf-8")) | |
| else: | |
| error = qs.get("error_description", ["未知错误"])[0] | |
| self.server.auth_code = None | |
| self.send_response(400) | |
| self.send_header("Content-Type", "text/html; charset=utf-8") | |
| self.end_headers() | |
| self.wfile.write(f"授权失败: {error}".encode("utf-8")) | |
| def log_message(self, format, *args): | |
| pass | |
| def main(): | |
| auth_params = ( | |
| f"client_id={CLIENT_ID}" | |
| f"&response_type=code" | |
| f"&redirect_uri={REDIRECT_URI}" | |
| f"&response_mode=query" | |
| f"&scope={SCOPES.replace(' ', '%20')}" | |
| ) | |
| auth_full_url = f"{AUTH_URL}?{auth_params}" | |
| print(f"\n即将打开浏览器进行微软账户授权...") | |
| print(f"如果浏览器未自动打开，请手动访问:\n{auth_full_url}\n") | |
| webbrowser.open(auth_full_url) | |
| server = HTTPServer(("localhost", 8400), CallbackHandler) | |
| server.auth_code = None | |
| print("等待浏览器回调...") | |
| server.handle_request() | |
| code = server.auth_code | |
| if not code: | |
| print("未获取到授权码，请重试。") | |
| sys.exit(1) | |
| print("已获取授权码，正在换取 Refresh Token...") | |
| resp = requests.post(TOKEN_URL, data={ | |
| "client_id": CLIENT_ID, | |
| "client_secret": CLIENT_SECRET, | |
| "code": code, | |
| "redirect_uri": REDIRECT_URI, | |
| "grant_type": "authorization_code", | |
| "scope": SCOPES, | |
| }) | |
| if resp.status_code != 200: | |
| print(f"Token 获取失败 ({resp.status_code}): {resp.text}") | |
| sys.exit(1) | |
| data = resp.json() | |
| refresh_token = data.get("refresh_token") | |
| if not refresh_token: | |
| print(f"响应中没有 refresh_token: {data}") | |
| sys.exit(1) | |
| print("\n" + "=" * 60) | |
| print("Refresh Token 获取成功！") | |
| print("=" * 60) | |
| print(f"\nBING_REFRESH_TOKEN={refresh_token}") | |
| print("\n请将上面的值复制到 .env 文件中。") | |
| if __name__ == "__main__": | |
| main() | |