From 8fb9d04ff68c170db2d950b66827f780af60f2fd Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Fri, 19 Dec 2025 11:33:25 +0800 Subject: [PATCH] feat: compatible with Windows system --- app/services/grok/client.py | 166 ++++++++++++++++++++---------------- app/services/grok/token.py | 8 +- main.py | 37 ++++++-- requirements.txt | 3 +- 4 files changed, 126 insertions(+), 88 deletions(-) diff --git a/app/services/grok/client.py b/app/services/grok/client.py index 0403de7..0982919 100644 --- a/app/services/grok/client.py +++ b/app/services/grok/client.py @@ -196,86 +196,102 @@ class GrokClient: if not token: raise GrokApiException("认证令牌缺失", "NO_AUTH_TOKEN") - # 403重试配置 - max_retries = 5 - retry_count = 0 + # 外层重试：可配置状态码（401/429等） + retry_codes = setting.grok_config.get("retry_status_codes", [401, 429]) + MAX_OUTER_RETRY = 3 - 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 and proxy_pool._enabled: - retry_count += 1 + for outer_retry in range(MAX_OUTER_RETRY + 1): # +1确保实际重试3次 + # 内层重试：403代理池重试 + max_403_retries = 5 + retry_403_count = 0 + + while retry_403_count <= max_403_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}" - if retry_count <= max_retries: - logger.warning(f"[Client] 遇到403错误，正在重试 ({retry_count}/{max_retries})...") - await asyncio.sleep(0.5) - continue + # 异步获取代理 + from app.core.proxy_pool import proxy_pool - # 内层重试全部失败 - logger.error(f"[Client] 403错误，已重试{retry_count-1}次，放弃") - - # 检查响应状态 - 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 + # 如果是403重试且使用代理池，强制刷新代理 + if retry_403_count > 0 and proxy_pool._enabled: + logger.info(f"[Client] 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 + + # 执行请求 + 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 and proxy_pool._enabled: + retry_403_count += 1 + + if retry_403_count <= max_403_retries: + logger.warning(f"[Client] 遇到403错误，正在重试 ({retry_403_count}/{max_403_retries})...") + await asyncio.sleep(0.5) + continue + + # 内层重试全部失败 + logger.error(f"[Client] 403错误，已重试{retry_403_count-1}次，放弃") + + # 检查可配置状态码错误 - 外层重试 + if response.status_code in retry_codes: + if outer_retry < MAX_OUTER_RETRY: + delay = (outer_retry + 1) * 0.1 # 渐进延迟：0.1s, 0.2s, 0.3s + logger.warning(f"[Client] 遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})，等待{delay}s...") + await asyncio.sleep(delay) + break # 跳出内层循环，进入外层重试 + else: + logger.error(f"[Client] {response.status_code}错误，已重试{outer_retry}次，放弃") + GrokClient._handle_error(response, token) + + # 检查响应状态 + if response.status_code != 200: + GrokClient._handle_error(response, token) + + # 成功 - 重置失败计数 + asyncio.create_task(token_manager.reset_failure(token)) + + # 如果是重试成功，记录日志 + if outer_retry > 0 or retry_403_count > 0: + logger.info(f"[Client] 重试成功！") + + # 处理响应 + 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") + raise GrokApiException("请求失败：已达到最大重试次数", "MAX_RETRIES_EXCEEDED") @staticmethod def _build_headers(token: str) -> Dict[str, str]: diff --git a/app/services/grok/token.py b/app/services/grok/token.py index 80f2e98..782a35f 100644 --- a/app/services/grok/token.py +++ b/app/services/grok/token.py @@ -109,7 +109,7 @@ class GrokTokenManager: from app.core.config import setting interval = setting.global_config.get("batch_save_interval", 1.0) - logger.info(f"[Token] 批量保存任务已启动，间隔: {interval}s") + logger.info(f"[Token] 存储任务已启动，间隔: {interval}s") while not self._shutdown: await asyncio.sleep(interval) @@ -118,15 +118,15 @@ class GrokTokenManager: try: await self._save_data() self._save_pending = False - logger.debug("[Token] 批量保存完成") + logger.debug("[Token] 存储完成") except Exception as e: - logger.error(f"[Token] 批量保存失败: {e}") + logger.error(f"[Token] 存储失败: {e}") async def start_batch_save(self) -> None: """启动批量保存任务""" if self._save_task is None: self._save_task = asyncio.create_task(self._batch_save_worker()) - logger.info("[Token] 批量保存任务已创建") + logger.info("[Token] 存储任务已创建") async def shutdown(self) -> None: """关闭并刷新所有待保存数据""" diff --git a/main.py b/main.py index cd76815..709b440 100644 --- a/main.py +++ b/main.py @@ -1,7 +1,12 @@ -"""FastAPI应用主入口""" +"""Grok2API""" +import os +import sys from contextlib import asynccontextmanager +from pathlib import Path + from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from app.core.logger import logger from app.core.exception import register_exception_handlers @@ -12,11 +17,20 @@ from app.api.v1.chat import router as chat_router from app.api.v1.models import router as models_router from app.api.v1.images import router as images_router from app.api.admin.manage import router as admin_router - -# 导入MCP服务器（认证配置在server.py中完成） from app.services.mcp import mcp -# 创建MCP的FastAPI应用实例 +# 0. 兼容性检测 +try: + if sys.platform != 'win32': + import uvloop + uvloop.install() + logger.info("[Grok2API] 启用uvloop高性能事件循环") + else: + logger.info("[Grok2API] Windows系统，使用默认asyncio事件循环") +except ImportError: + logger.info("[Grok2API] uvloop未安装，使用默认asyncio事件循环") + +# 1. 创建MCP的FastAPI应用实例 # 使用流式HTTP传输，支持高效的双向流式通信 mcp_app = mcp.http_app(stateless_http=True, transport="streamable-http") @@ -133,7 +147,7 @@ if __name__ == "__main__": import uvicorn import os - # 从环境变量读取 worker 数量，默认为 1 + # 读取 worker 数量，默认为 1 workers = int(os.getenv("WORKERS", "1")) # 提示多进程模式 @@ -143,10 +157,19 @@ if __name__ == "__main__": f"建议使用 Redis/MySQL 存储以获得最佳性能。" ) + # 确定事件循环类型 + loop_type = "auto" + if workers == 1 and sys.platform != 'win32': + try: + import uvloop + loop_type = "uvloop" + except ImportError: + pass + uvicorn.run( - "main:app", # 使用字符串以支持多worker + "main:app", host="0.0.0.0", port=8001, workers=workers, - loop="uvloop" if workers == 1 else "auto" # 单进程用uvloop，多进程用auto + loop=loop_type ) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 9e51d00..48d7d3d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,7 @@ - toml==0.10.2 fastapi==0.119.0 uvicorn==0.37.0 -uvloop==0.21.0 +uvloop==0.21.0; sys_platform != 'win32' python-dotenv==1.1.1 curl_cffi==0.13.0 requests==2.32.5