okooooooo/grok2api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: more retry & proxy pool

Browse Source
This commit is contained in:
Chenyme committed 2025-12-17 00:29:51 +08:00
1 parent 6ae39351a0
commit 22d8fb9af9
10 files changed
+1454 -784
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/core/config.py
+33 -7
View File
@@ -9,16 +9,21 @@ from typing import Dict, Any, Optional, Literal
DEFAULT_GROK = {
"api_key": "",
"proxy_url": "",
"proxy_pool_url": "",
"proxy_pool_interval": 300,
"cache_proxy_url": "",
"cf_clearance": "",
"x_statsig_id": "",
"dynamic_statsig": True,
"filtered_tags": "xaiartifact,xai:tool_usage_card,grok:render",
"filtered_tags": "xaiartifact,xai:tool_usage_card",
"show_thinking": True,
"temporary": False,
"max_upload_concurrency": 20,
"max_request_concurrency": 100,
"stream_first_response_timeout": 30,
"stream_chunk_timeout": 120,
"stream_total_timeout": 600,
"stream_first_response_timeout": 30,
"temporary": True,
"show_thinking": True
"retry_status_codes": [401, 429], # 可重试的HTTP状态码
}
DEFAULT_GLOBAL = {
@@ -147,20 +152,41 @@ class ConfigManager:
await self.reload()
def get_proxy(self, proxy_type: Literal["service", "cache"] = "service") -> str:
"""获取代理URL
async def get_proxy_async(self, proxy_type: Literal["service", "cache"] = "service") -> str:
"""异步获取代理URL（支持代理池）
Args:
proxy_type: 代理类型
- service: 服务代理（client/upload）
- cache: 缓存代理（cache）
"""
from app.core.proxy_pool import proxy_pool
if proxy_type == "cache":
cache_proxy = self.grok_config.get("cache_proxy_url", "")
if cache_proxy:
return cache_proxy
return self.grok_config.get("proxy_url", "")
# 从代理池获取
return await proxy_pool.get_proxy() or ""
def get_proxy(self, proxy_type: Literal["service", "cache"] = "service") -> str:
"""获取代理URL（同步方法，用于向后兼容）
Args:
proxy_type: 代理类型
- service: 服务代理（client/upload）
- cache: 缓存代理（cache）
"""
from app.core.proxy_pool import proxy_pool
if proxy_type == "cache":
cache_proxy = self.grok_config.get("cache_proxy_url", "")
if cache_proxy:
return cache_proxy
# 返回当前代理（如果是代理池，返回最后一次获取的）
return proxy_pool.get_current_proxy() or self.grok_config.get("proxy_url", "")
# 全局实例
app/core/proxy_pool.py
+143
View File
@@ -0,0 +1,143 @@
"""代理池管理器 - 从URL动态获取代理IP"""
import asyncio
import aiohttp
import time
from typing import Optional, List
from app.core.logger import logger
class ProxyPool:
"""代理池管理器"""
def __init__(self):
self._pool_url: Optional[str] = None
self._static_proxy: Optional[str] = None
self._current_proxy: Optional[str] = None
self._last_fetch_time: float = 0
self._fetch_interval: int = 300 # 5分钟刷新一次
self._enabled: bool = False
self._lock = asyncio.Lock()
def configure(self, proxy_url: str, proxy_pool_url: str = "", proxy_pool_interval: int = 300):
"""配置代理池
Args:
proxy_url: 静态代理URL（socks5h://xxx 或 http://xxx）
proxy_pool_url: 代理池API URL，返回单个代理地址
proxy_pool_interval: 代理池刷新间隔（秒）
"""
self._static_proxy = proxy_url if proxy_url else None
self._pool_url = proxy_pool_url if proxy_pool_url else None
self._fetch_interval = proxy_pool_interval
self._enabled = bool(self._pool_url)
if self._enabled:
logger.info(f"[ProxyPool] 代理池已启用: {self._pool_url}, 刷新间隔: {self._fetch_interval}s")
elif self._static_proxy:
logger.info(f"[ProxyPool] 使用静态代理: {self._static_proxy}")
self._current_proxy = self._static_proxy
else:
logger.info("[ProxyPool] 未配置代理")
async def get_proxy(self) -> Optional[str]:
"""获取代理地址
Returns:
代理URL或None
"""
# 如果未启用代理池，返回静态代理
if not self._enabled:
return self._static_proxy
# 检查是否需要刷新
now = time.time()
if not self._current_proxy or (now - self._last_fetch_time) >= self._fetch_interval:
async with self._lock:
# 双重检查
if not self._current_proxy or (now - self._last_fetch_time) >= self._fetch_interval:
await self._fetch_proxy()
return self._current_proxy
async def force_refresh(self) -> Optional[str]:
"""强制刷新代理（用于403错误重试）
Returns:
新的代理URL或None
"""
if not self._enabled:
return self._static_proxy
async with self._lock:
await self._fetch_proxy()
return self._current_proxy
async def _fetch_proxy(self):
"""从代理池URL获取新的代理"""
try:
logger.debug(f"[ProxyPool] 正在从代理池获取新代理: {self._pool_url}")
timeout = aiohttp.ClientTimeout(total=10)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(self._pool_url) as response:
if response.status == 200:
proxy_text = await response.text()
proxy = proxy_text.strip()
# 验证代理格式
if self._validate_proxy(proxy):
self._current_proxy = proxy
self._last_fetch_time = time.time()
logger.info(f"[ProxyPool] 成功获取新代理: {proxy}")
else:
logger.error(f"[ProxyPool] 代理格式无效: {proxy}")
# 降级到静态代理
if not self._current_proxy:
self._current_proxy = self._static_proxy
else:
logger.error(f"[ProxyPool] 获取代理失败: HTTP {response.status}")
# 降级到静态代理
if not self._current_proxy:
self._current_proxy = self._static_proxy
except asyncio.TimeoutError:
logger.error("[ProxyPool] 获取代理超时")
if not self._current_proxy:
self._current_proxy = self._static_proxy
except Exception as e:
logger.error(f"[ProxyPool] 获取代理异常: {e}")
# 降级到静态代理
if not self._current_proxy:
self._current_proxy = self._static_proxy
def _validate_proxy(self, proxy: str) -> bool:
"""验证代理格式
Args:
proxy: 代理URL
Returns:
是否有效
"""
if not proxy:
return False
# 支持的协议
valid_protocols = ['http://', 'https://', 'socks5://', 'socks5h://']
return any(proxy.startswith(proto) for proto in valid_protocols)
def get_current_proxy(self) -> Optional[str]:
"""获取当前使用的代理（同步方法）
Returns:
当前代理URL或None
"""
return self._current_proxy or self._static_proxy
# 全局代理池实例
proxy_pool = ProxyPool()
app/services/grok/cache.py
+74 -29
View File
@@ -60,37 +60,82 @@ class CacheService:
self._log("debug", "文件已缓存")
return cache_path
try:
proxy = setting.get_proxy("cache")
proxies = {"http": proxy, "https": proxy} if proxy else {}
if proxy:
self._log("debug", f"使用代理: {proxy.split('@')[-1] if '@' in proxy else proxy}")
# 外层重试：可配置状态码（401/429等）
retry_codes = setting.grok_config.get("retry_status_codes", [401, 429])
MAX_OUTER_RETRY = 3
for outer_retry in range(MAX_OUTER_RETRY):
try:
# 内层重试：403代理池重试（cache使用缓存代理，不支持代理池，所以403只重试一次）
max_403_retries = 5
retry_403_count = 0
while retry_403_count <= max_403_retries:
proxy = setting.get_proxy("cache")
proxies = {"http": proxy, "https": proxy} if proxy else {}
if proxy and outer_retry == 0 and retry_403_count == 0:
self._log("debug", f"使用代理: {proxy.split('@')[-1] if '@' in proxy else proxy}")
async with AsyncSession() as session:
url = f"{ASSETS_URL}{file_path}"
self._log("debug", f"下载: {url}")
async with AsyncSession() as session:
url = f"{ASSETS_URL}{file_path}"
if outer_retry == 0 and retry_403_count == 0:
self._log("debug", f"下载: {url}")
response = await session.get(
url,
headers=self._build_headers(file_path, auth_token),
proxies=proxies,
timeout=timeout or self.timeout,
allow_redirects=True,
impersonate="chrome133a"
)
# 检查403错误 - 内层重试(cache不使用代理池，所以直接失败)
if response.status_code == 403:
retry_403_count += 1
if retry_403_count <= max_403_retries:
self._log("warning", f"遇到403错误，正在重试 ({retry_403_count}/{max_403_retries})...")
await asyncio.sleep(0.5)
continue
self._log("error", f"403错误，已重试{retry_403_count-1}次，放弃")
return None
# 检查可配置状态码错误 - 外层重试
if response.status_code in retry_codes:
if outer_retry < MAX_OUTER_RETRY - 1:
self._log("warning", f"遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})...")
await asyncio.sleep(0.5)
break # 跳出内层循环，进入外层重试
else:
self._log("error", f"{response.status_code}错误，已重试{outer_retry}次，放弃")
return None
response.raise_for_status()
cache_path.write_bytes(response.content)
if outer_retry > 0 or retry_403_count > 0:
self._log("info", f"重试成功！")
else:
self._log("debug", "缓存成功")
# 异步清理（带错误处理）
asyncio.create_task(self._safe_cleanup())
return cache_path
except Exception as e:
if outer_retry < MAX_OUTER_RETRY - 1:
self._log("warning", f"下载异常: {e}，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})...")
await asyncio.sleep(0.5)
continue
response = await session.get(
url,
headers=self._build_headers(file_path, auth_token),
proxies=proxies,
timeout=timeout or self.timeout,
allow_redirects=True,
impersonate="chrome133a"
)
response.raise_for_status()
cache_path.write_bytes(response.content)
self._log("debug", "缓存成功")
# 异步清理（带错误处理）
asyncio.create_task(self._safe_cleanup())
return cache_path
except Exception as e:
self._log("error", f"下载失败: {e}")
return None
self._log("error", f"下载失败: {e}（已重试{outer_retry}次）")
return None
return None
def get_cached(self, file_path: str) -> Optional[Path]:
"""获取已缓存的文件"""
app/services/grok/client.py
+85 -45
View File
@@ -78,12 +78,14 @@ class GrokClient:
except GrokApiException as e:
last_err = e
# 仅401/429可重试
# 检查是否可重试
if e.error_code not in ["HTTP_ERROR", "NO_AVAILABLE_TOKEN"]:
raise
status = e.context.get("status") if e.context else None
if status not in [401, 429]:
retry_codes = setting.grok_config.get("retry_status_codes", [401, 429])
if status not in retry_codes:
raise
if i < MAX_RETRY - 1:
@@ -194,49 +196,87 @@ class GrokClient:
if not token:
raise GrokApiException("认证令牌缺失", "NO_AUTH_TOKEN")
try:
# 构建请求
headers = GrokClient._build_headers(token)
if model == "grok-imagine-0.9":
file_attachments = payload.get("fileAttachments", [])
ref_id = post_id or (file_attachments[0] if file_attachments else "")
if ref_id:
headers["Referer"] = f"https://grok.com/imagine/{ref_id}"
proxy = setting.get_proxy("service")
proxies = {"http": proxy, "https": proxy} if proxy else None
# 执行请求
response = await asyncio.to_thread(
curl_requests.post,
API_ENDPOINT,
headers=headers,
data=orjson.dumps(payload),
impersonate=BROWSER,
timeout=TIMEOUT,
stream=True,
proxies=proxies
)
if response.status_code != 200:
GrokClient._handle_error(response, token)
# 成功 - 重置失败计数
asyncio.create_task(token_manager.reset_failure(token))
# 处理响应
result = (GrokResponseProcessor.process_stream(response, token) if stream
else await GrokResponseProcessor.process_normal(response, token, model))
asyncio.create_task(GrokClient._update_limits(token, model))
return result
except curl_requests.RequestsError as e:
logger.error(f"[Client] 网络错误: {e}")
raise GrokApiException(f"网络错误: {e}", "NETWORK_ERROR") from e
except Exception as e:
logger.error(f"[Client] 请求错误: {e}")
raise GrokApiException(f"请求错误: {e}", "REQUEST_ERROR") from e
# 403重试配置
max_retries = 5
retry_count = 0
while retry_count <= max_retries:
try:
# 构建请求
headers = GrokClient._build_headers(token)
if model == "grok-imagine-0.9":
file_attachments = payload.get("fileAttachments", [])
ref_id = post_id or (file_attachments[0] if file_attachments else "")
if ref_id:
headers["Referer"] = f"https://grok.com/imagine/{ref_id}"
# 异步获取代理
from app.core.proxy_pool import proxy_pool
# 如果是重试且使用代理池，强制刷新代理
if retry_count > 0 and proxy_pool._enabled:
logger.info(f"[Client] 403重试 {retry_count}/{max_retries}，刷新代理...")
proxy = await proxy_pool.force_refresh()
else:
proxy = await setting.get_proxy_async("service")
proxies = {"http": proxy, "https": proxy} if proxy else None
# 执行请求
response = await asyncio.to_thread(
curl_requests.post,
API_ENDPOINT,
headers=headers,
data=orjson.dumps(payload),
impersonate=BROWSER,
timeout=TIMEOUT,
stream=True,
proxies=proxies
)
# 检查是否是403错误
if response.status_code == 403:
retry_count += 1
# 如果使用代理池且未达到最大重试次数，继续重试
if proxy_pool._enabled and retry_count <= max_retries:
logger.warning(f"[Client] 遇到403错误，正在重试 ({retry_count}/{max_retries})...")
await asyncio.sleep(0.5) # 短暂延迟
continue
# 达到最大重试次数或未使用代理池，抛出错误
logger.error(f"[Client] 403错误，已重试{retry_count-1}次，放弃")
GrokClient._handle_error(response, token)
if response.status_code != 200:
GrokClient._handle_error(response, token)
# 成功 - 重置失败计数
asyncio.create_task(token_manager.reset_failure(token))
# 如果是重试成功，记录日志
if retry_count > 0:
logger.info(f"[Client] 重试成功！第{retry_count}次重试后请求成功")
# 处理响应
result = (GrokResponseProcessor.process_stream(response, token) if stream
else await GrokResponseProcessor.process_normal(response, token, model))
asyncio.create_task(GrokClient._update_limits(token, model))
return result
except curl_requests.RequestsError as e:
logger.error(f"[Client] 网络错误: {e}")
raise GrokApiException(f"网络错误: {e}", "NETWORK_ERROR") from e
except GrokApiException:
# 重新抛出GrokApiException（包括403错误）
raise
except Exception as e:
logger.error(f"[Client] 请求错误: {e}")
raise GrokApiException(f"请求错误: {e}", "REQUEST_ERROR") from e
# 理论上不应该到这里，但以防万一
raise GrokApiException("403错误：已达到最大重试次数", "MAX_RETRIES_EXCEEDED")
@staticmethod
def _build_headers(token: str) -> Dict[str, str]:
app/services/grok/create.py
+80 -34
View File
@@ -1,5 +1,6 @@
"""Post创建管理器 - 用于视频生成前的会话创建"""
import asyncio
import orjson
from typing import Dict, Any, Optional
from curl_cffi.requests import AsyncSession
@@ -50,41 +51,86 @@ class PostCreateManager:
"Cookie": f"{auth_token};{cf}" if cf else auth_token
}
proxy = setting.grok_config.get("proxy_url", "")
proxies = {"http": proxy, "https": proxy} if proxy else None
# 发送请求
async with AsyncSession() as session:
response = await session.post(
ENDPOINT,
headers=headers,
json=data,
impersonate=BROWSER,
timeout=TIMEOUT,
proxies=proxies
)
if response.status_code == 200:
result = response.json()
post_id = result.get("post", {}).get("id", "")
logger.debug(f"[PostCreate] 成功，会话ID: {post_id}")
return {
"post_id": post_id,
"file_id": file_id,
"file_uri": file_uri,
"success": True,
"data": result
}
# 外层重试：可配置状态码（401/429等）
retry_codes = setting.grok_config.get("retry_status_codes", [401, 429])
MAX_OUTER_RETRY = 3
for outer_retry in range(MAX_OUTER_RETRY):
# 内层重试：403代理池重试
max_403_retries = 5
retry_403_count = 0
# 错误处理
try:
error = response.json()
msg = f"状态码: {response.status_code}, 详情: {error}"
except:
msg = f"状态码: {response.status_code}, 详情: {response.text[:200]}"
logger.error(f"[PostCreate] 失败: {msg}")
raise GrokApiException(f"创建失败: {msg}", "CREATE_ERROR")
while retry_403_count <= max_403_retries:
# 异步获取代理（支持代理池）
from app.core.proxy_pool import proxy_pool
# 如果是403重试且使用代理池，强制刷新代理
if retry_403_count > 0 and proxy_pool._enabled:
logger.info(f"[PostCreate] 403重试 {retry_403_count}/{max_403_retries}，刷新代理...")
proxy = await proxy_pool.force_refresh()
else:
proxy = await setting.get_proxy_async("service")
proxies = {"http": proxy, "https": proxy} if proxy else None
# 发送请求
async with AsyncSession() as session:
response = await session.post(
ENDPOINT,
headers=headers,
json=data,
impersonate=BROWSER,
timeout=TIMEOUT,
proxies=proxies
)
# 检查403错误 - 内层重试
if response.status_code == 403:
retry_403_count += 1
if proxy_pool._enabled and retry_403_count <= max_403_retries:
logger.warning(f"[PostCreate] 遇到403错误，正在重试 ({retry_403_count}/{max_403_retries})...")
await asyncio.sleep(0.5)
continue
logger.error(f"[PostCreate] 403错误，已重试{retry_403_count-1}次，放弃")
raise GrokApiException(f"创建失败: 403错误", "CREATE_ERROR")
# 检查可配置状态码错误 - 外层重试
if response.status_code in retry_codes:
if outer_retry < MAX_OUTER_RETRY - 1:
logger.warning(f"[PostCreate] 遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})...")
await asyncio.sleep(0.5)
break # 跳出内层循环，进入外层重试
else:
logger.error(f"[PostCreate] {response.status_code}错误，已重试{outer_retry}次，放弃")
raise GrokApiException(f"创建失败: {response.status_code}错误", "CREATE_ERROR")
if response.status_code == 200:
result = response.json()
post_id = result.get("post", {}).get("id", "")
if outer_retry > 0 or retry_403_count > 0:
logger.info(f"[PostCreate] 重试成功！")
logger.debug(f"[PostCreate] 成功，会话ID: {post_id}")
return {
"post_id": post_id,
"file_id": file_id,
"file_uri": file_uri,
"success": True,
"data": result
}
# 其他错误处理
try:
error = response.json()
msg = f"状态码: {response.status_code}, 详情: {error}"
except:
msg = f"状态码: {response.status_code}, 详情: {response.text[:200]}"
logger.error(f"[PostCreate] 失败: {msg}")
raise GrokApiException(f"创建失败: {msg}", "CREATE_ERROR")
except GrokApiException:
raise
app/services/grok/token.py
+83 -34
View File
@@ -319,43 +319,92 @@ class GrokTokenManager:
headers = get_dynamic_headers("/rest/rate-limits")
headers["Cookie"] = f"{auth_token};{cf}" if cf else auth_token
proxy = setting.grok_config.get("proxy_url", "")
proxies = {"http": proxy, "https": proxy} if proxy else None
# 外层重试：可配置状态码（401/429等）
retry_codes = setting.grok_config.get("retry_status_codes", [401, 429])
MAX_OUTER_RETRY = 3
async with AsyncSession() as session:
response = await session.post(
RATE_LIMIT_API,
headers=headers,
json=payload,
impersonate=BROWSER,
timeout=TIMEOUT,
proxies=proxies
)
for outer_retry in range(MAX_OUTER_RETRY):
# 内层重试：403代理池重试
max_403_retries = 5
retry_403_count = 0
while retry_403_count <= max_403_retries:
# 异步获取代理（支持代理池）
from app.core.proxy_pool import proxy_pool
# 如果是403重试且使用代理池，强制刷新代理
if retry_403_count > 0 and proxy_pool._enabled:
logger.info(f"[Token] 403重试 {retry_403_count}/{max_403_retries}，刷新代理...")
proxy = await proxy_pool.force_refresh()
else:
proxy = await setting.get_proxy_async("service")
proxies = {"http": proxy, "https": proxy} if proxy else None
async with AsyncSession() as session:
response = await session.post(
RATE_LIMIT_API,
headers=headers,
json=payload,
impersonate=BROWSER,
timeout=TIMEOUT,
proxies=proxies
)
if response.status_code == 200:
data = response.json()
sso = self._extract_sso(auth_token)
if sso:
if model == "grok-4-heavy":
await self.update_limits(sso, normal=None, heavy=data.get("remainingQueries", -1))
logger.info(f"[Token] 更新限制: {sso[:10]}..., heavy={data.get('remainingQueries', -1)}")
# 检查403错误 - 内层重试
if response.status_code == 403:
retry_403_count += 1
if proxy_pool._enabled and retry_403_count <= max_403_retries:
logger.warning(f"[Token] 遇到403错误，正在重试 ({retry_403_count}/{max_403_retries})...")
await asyncio.sleep(0.5)
continue
logger.error(f"[Token] 403错误，已重试{retry_403_count-1}次，放弃")
sso = self._extract_sso(auth_token)
if sso:
await self.record_failure(auth_token, 403, "服务器被Block")
return None
# 检查可配置状态码错误 - 外层重试
if response.status_code in retry_codes:
if outer_retry < MAX_OUTER_RETRY - 1:
logger.warning(f"[Token] 遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})...")
await asyncio.sleep(0.5)
break # 跳出内层循环，进入外层重试
else:
logger.error(f"[Token] {response.status_code}错误，已重试{outer_retry}次，放弃")
sso = self._extract_sso(auth_token)
if sso:
if response.status_code == 401:
await self.record_failure(auth_token, 401, "Token失效")
else:
await self.record_failure(auth_token, response.status_code, f"错误: {response.status_code}")
return None
if response.status_code == 200:
data = response.json()
sso = self._extract_sso(auth_token)
if outer_retry > 0 or retry_403_count > 0:
logger.info(f"[Token] 重试成功！")
if sso:
if model == "grok-4-heavy":
await self.update_limits(sso, normal=None, heavy=data.get("remainingQueries", -1))
logger.info(f"[Token] 更新限制: {sso[:10]}..., heavy={data.get('remainingQueries', -1)}")
else:
await self.update_limits(sso, normal=data.get("remainingTokens", -1), heavy=None)
logger.info(f"[Token] 更新限制: {sso[:10]}..., basic={data.get('remainingTokens', -1)}")
return data
else:
await self.update_limits(sso, normal=data.get("remainingTokens", -1), heavy=None)
logger.info(f"[Token] 更新限制: {sso[:10]}..., basic={data.get('remainingTokens', -1)}")
return data
else:
logger.warning(f"[Token] 获取限制失败: {response.status_code}")
sso = self._extract_sso(auth_token)
if sso:
if response.status_code == 401:
await self.record_failure(auth_token, 401, "Token失效")
elif response.status_code == 403:
await self.record_failure(auth_token, 403, "服务器被Block")
else:
await self.record_failure(auth_token, response.status_code, f"错误: {response.status_code}")
return None
# 其他错误
logger.warning(f"[Token] 获取限制失败: {response.status_code}")
sso = self._extract_sso(auth_token)
if sso:
await self.record_failure(auth_token, response.status_code, f"错误: {response.status_code}")
return None
except Exception as e:
logger.error(f"[Token] 检查限制错误: {e}")
app/services/grok/upload.py
+89 -25
View File
@@ -1,5 +1,6 @@
"""图片上传管理器 - 支持Base64和URL图片上传"""
import asyncio
import base64
import re
from typing import Tuple, Optional
@@ -52,37 +53,100 @@ class ImageUploadManager:
"content": buffer,
}
if not auth_token:
raise GrokApiException("认证令牌缺失", "NO_AUTH_TOKEN")
# 请求配置
cf = setting.grok_config.get("cf_clearance", "")
headers = {
**get_dynamic_headers("/rest/app-chat/upload-file"),
"Cookie": f"{auth_token};{cf}" if cf else auth_token,
}
# 外层重试：可配置状态码（401/429等）
retry_codes = setting.grok_config.get("retry_status_codes", [401, 429])
MAX_OUTER_RETRY = 3
proxy = setting.grok_config.get("proxy_url", "")
proxies = {"http": proxy, "https": proxy} if proxy else None
for outer_retry in range(MAX_OUTER_RETRY):
try:
# 内层重试：403代理池重试
max_403_retries = 5
retry_403_count = 0
while retry_403_count <= max_403_retries:
# 请求配置
cf = setting.grok_config.get("cf_clearance", "")
headers = {
**get_dynamic_headers("/rest/app-chat/upload-file"),
"Cookie": f"{auth_token};{cf}" if cf else auth_token,
}
# 异步获取代理（支持代理池）
from app.core.proxy_pool import proxy_pool
# 如果是403重试且使用代理池，强制刷新代理
if retry_403_count > 0 and proxy_pool._enabled:
logger.info(f"[Upload] 403重试 {retry_403_count}/{max_403_retries}，刷新代理...")
proxy = await proxy_pool.force_refresh()
else:
proxy = await setting.get_proxy_async("service")
proxies = {"http": proxy, "https": proxy} if proxy else None
# 上传
async with AsyncSession() as session:
response = await session.post(
UPLOAD_API,
headers=headers,
json=data,
impersonate=BROWSER,
timeout=TIMEOUT,
proxies=proxies,
)
if response.status_code == 200:
result = response.json()
file_id = result.get("fileMetadataId", "")
file_uri = result.get("fileUri", "")
logger.debug(f"[Upload] 成功，ID: {file_id}")
return file_id, file_uri
# 上传
async with AsyncSession() as session:
response = await session.post(
UPLOAD_API,
headers=headers,
json=data,
impersonate=BROWSER,
timeout=TIMEOUT,
proxies=proxies,
)
# 检查403错误 - 内层重试
if response.status_code == 403:
retry_403_count += 1
if proxy_pool._enabled and retry_403_count <= max_403_retries:
logger.warning(f"[Upload] 遇到403错误，正在重试 ({retry_403_count}/{max_403_retries})...")
await asyncio.sleep(0.5)
continue
logger.error(f"[Upload] 403错误，已重试{retry_403_count-1}次，放弃")
return "", ""
# 检查可配置状态码错误 - 外层重试
if response.status_code in retry_codes:
if outer_retry < MAX_OUTER_RETRY - 1:
logger.warning(f"[Upload] 遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})...")
await asyncio.sleep(0.5)
break # 跳出内层循环，进入外层重试
else:
logger.error(f"[Upload] {response.status_code}错误，已重试{outer_retry}次，放弃")
return "", ""
if response.status_code == 200:
result = response.json()
file_id = result.get("fileMetadataId", "")
file_uri = result.get("fileUri", "")
if outer_retry > 0 or retry_403_count > 0:
logger.info(f"[Upload] 重试成功！")
logger.debug(f"[Upload] 成功，ID: {file_id}")
return file_id, file_uri
# 其他错误直接返回
logger.error(f"[Upload] 失败，状态码: {response._status_code}")
return "", ""
# 内层循环正常结束（非break），说明403重试全部失败
return "", ""
except Exception as e:
if outer_retry < MAX_OUTER_RETRY - 1:
logger.warning(f"[Upload] 异常: {e}，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})...")
await asyncio.sleep(0.5)
continue
logger.warning(f"[Upload] 失败: {e}")
return "", ""
return "", ""
except Exception as e:
app/template/admin.html
+857 -609
View File
File diff suppressed because it is too large. Load diff
main.py
+7
View File
@@ -48,6 +48,13 @@ async def lifespan(app: FastAPI):
await setting.reload()
logger.info("[Grok2API] 核心服务初始化完成")
# 2.5. 初始化代理池
from app.core.proxy_pool import proxy_pool
proxy_url = setting.grok_config.get("proxy_url", "")
proxy_pool_url = setting.grok_config.get("proxy_pool_url", "")
proxy_pool_interval = setting.grok_config.get("proxy_pool_interval", 300)
proxy_pool.configure(proxy_url, proxy_pool_url, proxy_pool_interval)
# 3. 异步加载 token 数据
await token_manager._load_data()
logger.info("[Grok2API] Token数据加载完成")
requirements.txt
+3 -1
View File
@@ -1,3 +1,4 @@
toml==0.10.2
fastapi==0.119.0
uvicorn==0.37.0
@@ -13,4 +14,5 @@ aiomysql==0.2.0
redis==6.4.0
fastmcp==2.12.4
cryptography==46.0.3
orjson==3.11.4
orjson==3.11.4
aiohttp==3.13.2