| import uuid | |
| import requests | |
| import http.server | |
| import webbrowser | |
| import socketserver | |
| import os | |
| import random | |
| def get_refresh_tokens(): | |
| tokens = os.environ.get('REFRESH_TOKENS', '') | |
| return [token.strip() for token in tokens.split(',') if token.strip()] | |
| def refresh_access_token(refresh_token): | |
| url = 'https://token.oaifree.com/api/auth/refresh' | |
| headers = { | |
| 'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8' | |
| } | |
| data = { | |
| 'refresh_token': refresh_token | |
| } | |
| response = requests.post(url, headers=headers, data=data) | |
| if response.status_code == 200: | |
| data = response.json() | |
| return data['access_token'] | |
| return None | |
| def get_livekit_url(access_token): | |
| headers = { | |
| "content-type": "application/json", | |
| "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", | |
| "authorization": f"Bearer {access_token}", | |
| } | |
| res = requests.post("https://chatgpt.com/voice/get_token", headers=headers, cookies={'__cf_bm': ''}, json={ | |
| "voice": "cove", | |
| "voice_mode": "standard", | |
| "parent_message_id": str(uuid.uuid4()), | |
| "model_slug": "auto", | |
| "voice_training_allowed": False, | |
| "enable_message_streaming": False, | |
| "language": "zh", | |
| "video_training_allowed": False, | |
| "voice_session_id": str(uuid.uuid4()) | |
| }).json() | |
| livekit_url = "https://meet.livekit.io" | |
| url = f"{livekit_url}/custom?liveKitUrl={res['url']}&token={res['token']}#{res['e2ee_key']}" | |
| return url | |
| class RedirectHandler(http.server.SimpleHTTPRequestHandler): | |
| def do_GET(self): | |
| refresh_tokens = get_refresh_tokens() | |
| if not refresh_tokens: | |
| self.send_error(500, "未设置 REFRESH_TOKENS 环境变量") | |
| return | |
| refresh_token = random.choice(refresh_tokens) | |
| access_token = refresh_access_token(refresh_token) | |
| if access_token: | |
| target_url = get_livekit_url(access_token) | |
| self.send_response(302) | |
| self.send_header('Location', target_url) | |
| self.end_headers() | |
| else: | |
| self.send_error(500, "无法获取访问令牌") | |
| def run_server(port=8000): | |
| with socketserver.TCPServer(("", port), RedirectHandler) as httpd: | |
| print(f"服务器运行在 http://localhost:{port}") | |
| webbrowser.open(f"http://localhost:{port}") | |
| httpd.serve_forever() | |
| if __name__ == "__main__": | |
| refresh_tokens = get_refresh_tokens() | |
| if not refresh_tokens: | |
| print("请设置 REFRESH_TOKENS 环境变量，多个 token 以逗号分隔") | |
| else: | |
| print(f"成功获取 {len(refresh_tokens)} 个 refresh token") | |
| run_server() | |