From 6798426c7b62886b082cda5a89a1e5fdffbf9874 Mon Sep 17 00:00:00 2001 From: Debug Assistant Date: Fri, 20 Feb 2026 00:05:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E4=BB=A3=E7=90=86?= =?UTF-8?q?=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在 .env 中通过 PROXY_URL 配置代理 - 支持 http/https/socks4/socks5 代理 - 所有网络请求（curl_cffi 和 requests）都走代理 - 服务类构造函数新增 proxies 参数传递代理配置 --- .env.example | 7 +++++++ g/email_service.py | 6 +++++- g/nsfw_service.py | 4 ++++ g/turnstile_service.py | 11 ++++++----- g/user_agreement_service.py | 2 ++ grok.py | 30 +++++++++++++++++++++++------- 6 files changed, 47 insertions(+), 13 deletions(-) diff --git a/.env.example b/.env.example index 0dbd412..eeaacfd 100644 --- a/.env.example +++ b/.env.example @@ -5,3 +5,10 @@ FREEMAIL_TOKEN=your-jwt-token # Turnstile 验证配置 # 如果不填则使用本地 Turnstile Solver（http://127.0.0.1:5072） YESCAPTCHA_KEY= + +# 代理配置（可选） +# 支持 http、https、socks4、socks5 +# 示例： +# PROXY_URL=http://127.0.0.1:7890 +# PROXY_URL=socks5://127.0.0.1:10808 +PROXY_URL= diff --git a/g/email_service.py b/g/email_service.py index 795b7f4..54005c1 100644 --- a/g/email_service.py +++ b/g/email_service.py @@ -6,7 +6,7 @@ from dotenv import load_dotenv class EmailService: - def __init__(self): + def __init__(self, proxies=None): load_dotenv() self.worker_domain = os.getenv("WORKER_DOMAIN") self.freemail_token = os.getenv("FREEMAIL_TOKEN") @@ -14,6 +14,7 @@ class EmailService: raise ValueError("Missing: WORKER_DOMAIN or FREEMAIL_TOKEN") self.base_url = f"https://{self.worker_domain}" self.headers = {"Authorization": f"Bearer {self.freemail_token}"} + self.proxies = proxies or {} def create_email(self): """创建临时邮箱 GET /api/generate""" @@ -21,6 +22,7 @@ class EmailService: res = requests.get( f"{self.base_url}/api/generate", headers=self.headers, + proxies=self.proxies, timeout=10 ) if res.status_code == 200: @@ -44,6 +46,7 @@ class EmailService: f"{self.base_url}/api/emails", params={"mailbox": email}, headers=self.headers, + proxies=self.proxies, timeout=10 ) if debug: @@ -91,6 +94,7 @@ class EmailService: f"{self.base_url}/api/mailboxes", params={"address": address}, headers=self.headers, + proxies=self.proxies, timeout=10 ) return res.status_code == 200 and res.json().get("success") diff --git a/g/nsfw_service.py b/g/nsfw_service.py index 86eb1eb..202713a 100644 --- a/g/nsfw_service.py +++ b/g/nsfw_service.py @@ -25,6 +25,7 @@ class NsfwSettingsService: user_agent: Optional[str] = None, cf_clearance: Optional[str] = None, timeout: int = 15, + proxies: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: """ 启用 always_show_nsfw_content。 @@ -88,6 +89,7 @@ class NsfwSettingsService: data=data, impersonate=impersonate or "chrome120", timeout=timeout, + proxies=proxies or {}, ) hex_reply = response.content.hex() grpc_status = response.headers.get("grpc-status") @@ -124,6 +126,7 @@ class NsfwSettingsService: impersonate: str = "chrome120", user_agent: Optional[str] = None, timeout: int = 30, + proxies: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: """ 使用帖子方法开启 Unhinged 模式（二次验证）。 @@ -153,6 +156,7 @@ class NsfwSettingsService: data=data, impersonate=impersonate, timeout=timeout, + proxies=proxies or {}, ) return { "ok": response.status_code == 200, diff --git a/g/turnstile_service.py b/g/turnstile_service.py index d03eccc..017ea19 100644 --- a/g/turnstile_service.py +++ b/g/turnstile_service.py @@ -12,13 +12,14 @@ load_dotenv() class TurnstileService: """Turnstile验证服务类""" - def __init__(self, solver_url="http://127.0.0.1:5072"): + def __init__(self, solver_url="http://127.0.0.1:5072", proxies=None): """ 初始化Turnstile服务 """ self.yescaptcha_key = os.getenv('YESCAPTCHA_KEY', '').strip() self.solver_url = solver_url self.yescaptcha_api = "https://api.yescaptcha.com" + self.proxies = proxies or {} def create_task(self, siteurl, sitekey): """ @@ -35,7 +36,7 @@ class TurnstileService: "websiteKey": sitekey } } - response = requests.post(url, json=payload) + response = requests.post(url, json=payload, proxies=self.proxies) response.raise_for_status() data = response.json() if data.get('errorId') != 0: @@ -44,7 +45,7 @@ class TurnstileService: else: # 使用本地 Turnstile Solver url = f"{self.solver_url}/turnstile?url={siteurl}&sitekey={sitekey}" - response = requests.get(url) + response = requests.get(url, proxies=self.proxies) response.raise_for_status() return response.json()['taskId'] @@ -63,7 +64,7 @@ class TurnstileService: "clientKey": self.yescaptcha_key, "taskId": task_id } - response = requests.post(url, json=payload) + response = requests.post(url, json=payload, proxies=self.proxies) response.raise_for_status() data = response.json() @@ -86,7 +87,7 @@ class TurnstileService: else: # 使用本地 Turnstile Solver url = f"{self.solver_url}/result?id={task_id}" - response = requests.get(url) + response = requests.get(url, proxies=self.proxies) response.raise_for_status() data = response.json() captcha = data.get('solution', {}).get('token', None) diff --git a/g/user_agreement_service.py b/g/user_agreement_service.py index 94a1c19..fabd03f 100644 --- a/g/user_agreement_service.py +++ b/g/user_agreement_service.py @@ -25,6 +25,7 @@ class UserAgreementService: user_agent: Optional[str] = None, cf_clearance: Optional[str] = None, timeout: int = 15, + proxies: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: """ 同意 TOS 版本。 @@ -85,6 +86,7 @@ class UserAgreementService: data=data, impersonate=impersonate or "chrome120", timeout=timeout, + proxies=proxies or {}, ) hex_reply = response.content.hex() grpc_status = response.headers.get("grpc-status") diff --git a/grok.py b/grok.py index 8ccf54a..5a05550 100644 --- a/grok.py +++ b/grok.py @@ -6,9 +6,22 @@ import traceback from urllib.parse import urljoin, urlparse from curl_cffi import requests from bs4 import BeautifulSoup +from dotenv import load_dotenv from g import EmailService, TurnstileService, UserAgreementService, NsfwSettingsService +# 加载环境变量 +load_dotenv() + +# 获取代理配置 +PROXY_URL = os.getenv("PROXY_URL", "").strip() +if PROXY_URL: + PROXIES = {"http": PROXY_URL, "https": PROXY_URL} + print(f"[*] 使用代理: {PROXY_URL}") +else: + PROXIES = {} + print("[*] 未配置代理，使用直连") + # 基础配置 # 基础 URL（用于 API 请求和 Solver） base_url = "https://accounts.x.ai" @@ -39,10 +52,7 @@ def get_random_chrome_profile(): f"Chrome/{profile['version']} Safari/537.36" ) return profile["impersonate"], ua -PROXIES = { - # "http": "http://127.0.0.1:10808", - # "https": "http://127.0.0.1:10808" -} + # 动态获取的全局变量 config = { @@ -283,6 +293,7 @@ def register_single_thread(debug_mode=False, single_run=False): sso_rw=sso_rw or "", impersonate=impersonate_fingerprint, user_agent=account_user_agent, + proxies=PROXIES, ) tos_hex = tos_result.get("hex_reply") or "" if debug_mode: @@ -300,6 +311,7 @@ def register_single_thread(debug_mode=False, single_run=False): sso_rw=sso_rw or "", impersonate=impersonate_fingerprint, user_agent=account_user_agent, + proxies=PROXIES, ) nsfw_hex = nsfw_result.get("hex_reply") or "" nsfw_ok = nsfw_result.get("ok", False) @@ -311,7 +323,11 @@ def register_single_thread(debug_mode=False, single_run=False): # 立即进行二次验证 (enable_unhinged) if debug_mode: print(f"[DEBUG] [{thread_id}] 启用 Unhinged...") - unhinged_result = nsfw_service.enable_unhinged(sso, sso_rw or "") + unhinged_result = nsfw_service.enable_unhinged( + sso=sso, + sso_rw=sso_rw or "", + proxies=PROXIES, + ) unhinged_ok = unhinged_result.get("ok", False) if debug_mode: print(f"[DEBUG] [{thread_id}] Unhinged 结果: ok={unhinged_ok}") @@ -385,7 +401,7 @@ def main(): print("[*] 正在初始化...") start_url = site_url print(f"[DEBUG] 请求 URL: {start_url}") - with requests.Session(impersonate=DEFAULT_IMPERSONATE) as s: + with requests.Session(impersonate=DEFAULT_IMPERSONATE, proxies=PROXIES) as s: try: print("[DEBUG] 正在获取页面...") html = s.get(start_url, timeout=30).text @@ -407,7 +423,7 @@ def main(): print(f"[DEBUG] 找到 {len(js_urls)} 个 JS 文件") for js_url in js_urls: print(f"[DEBUG] 正在请求 JS: {js_url}") - js_content = s.get(js_url, timeout=30).text + js_content = s.get(js_url, timeout=30, proxies=PROXIES).text match = re.search(r'7f[a-fA-F0-9]{40}', js_content) if match: config["action_id"] = match.group(0)