diff --git a/app/core/config.py b/app/core/config.py index 4b63ec6..dc68617 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -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", "") # 全局实例 diff --git a/app/core/proxy_pool.py b/app/core/proxy_pool.py new file mode 100644 index 0000000..eaea5f5 --- /dev/null +++ b/app/core/proxy_pool.py @@ -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() diff --git a/app/services/grok/cache.py b/app/services/grok/cache.py index 3380d87..51f6a58 100644 --- a/app/services/grok/cache.py +++ b/app/services/grok/cache.py @@ -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]: """获取已缓存的文件""" diff --git a/app/services/grok/client.py b/app/services/grok/client.py index 435a722..888bbea 100644 --- a/app/services/grok/client.py +++ b/app/services/grok/client.py @@ -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]: diff --git a/app/services/grok/create.py b/app/services/grok/create.py index 72be355..2210165 100644 --- a/app/services/grok/create.py +++ b/app/services/grok/create.py @@ -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 diff --git a/app/services/grok/token.py b/app/services/grok/token.py index 08ab037..7911e34 100644 --- a/app/services/grok/token.py +++ b/app/services/grok/token.py @@ -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}") diff --git a/app/services/grok/upload.py b/app/services/grok/upload.py index 9bea85a..4252b35 100644 --- a/app/services/grok/upload.py +++ b/app/services/grok/upload.py @@ -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: diff --git a/app/template/admin.html b/app/template/admin.html index b85bb26..0e47cc9 100644 --- a/app/template/admin.html +++ b/app/template/admin.html @@ -1,639 +1,887 @@ + - - - 管理控制台 - Grok2API - - - - + + + 管理控制台 - Grok2API + + + + + - -
-
-
- Grok2API - by @Chenyme -
-
-
-
- - - - - - - FILE - -
- 加载中... -
-
- + +
+
+
+ Grok2API + by @Chenyme +
+
+
+
+ + + + + + + FILE + +
+ 加载中...
+
+
-
+
+
-
- -
- +
+ +
+ +
+ + +
+ +
+
+

Token 总数

+

-

- -
- -
-
-

Token 总数

-

-

-
- -
-

Token 正常

-

-

-
- -
-

Token 未使用

-

-

-
- -
-

Token 限流中

-

-

-
- -
-

Token 失效

-

-

-
- -
-

Chat 总剩余

-

-

-
- -
-

Image 总剩余

-

-

-
- -
-

Video 总剩余

-

无法统计

-
+
+

Token 正常

+

-

- -
- -
-
-
- - - -
-
+
+

Token 未使用

+

-

+
-
- - - -
+
+

Token 限流中

+

-

+
+ +
+

Token 失效

+

-

+
+ +
+

Chat 总剩余

+

-

+
+ +
+

Image 总剩余

+

-

+
+ +
+

Video 总剩余

+

无法统计

+
+
+ + +
+ +
+
+
+ + +
- - -
- - - - - - - - - - - - - - - - - - -
- - Token类型状态普通高级标签备注创建时间操作
-
- - + +
+ + -
+ +
- -
- - - -
- + + \ No newline at end of file diff --git a/main.py b/main.py index a042695..cd76815 100644 --- a/main.py +++ b/main.py @@ -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数据加载完成") diff --git a/requirements.txt b/requirements.txt index 85145fa..9e51d00 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 \ No newline at end of file +orjson==3.11.4 +aiohttp==3.13.2 \ No newline at end of file