From 8255d0b32302742cb1b06b58c367e8f76f5d6d76 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Thu, 16 Oct 2025 13:33:45 +0800 Subject: [PATCH 01/43] feat: use socks5h proxy --- app/core/config.py | 10 +++++++++- app/services/grok/cache.py | 3 +++ app/services/grok/client.py | 20 +++++++++----------- app/services/grok/token.py | 3 +++ app/services/grok/upload.py | 4 ++++ app/template/admin.html | 2 +- 6 files changed, 29 insertions(+), 13 deletions(-) diff --git a/app/core/config.py b/app/core/config.py index 01226f7..f307bb6 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -25,7 +25,15 @@ class ConfigManager: """配置加载器""" try: with open(self.config_path, "r", encoding="utf-8") as f: - return toml.load(f)[section] + config = toml.load(f)[section] + + # 自动将 SOCKS5 转换为 SOCKS5H + if section == "grok" and "proxy_url" in config: + proxy_url = config["proxy_url"] + if proxy_url and proxy_url.startswith("socks5://"): + config["proxy_url"] = proxy_url.replace("socks5://", "socks5h://", 1) + + return config except Exception as e: raise Exception(f"[Setting] 配置加载失败: {e}") diff --git a/app/services/grok/cache.py b/app/services/grok/cache.py index d9d9220..5446ced 100644 --- a/app/services/grok/cache.py +++ b/app/services/grok/cache.py @@ -48,6 +48,9 @@ class CacheService: proxy_url = setting.grok_config.get("proxy_url") proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else {} + + if proxy_url: + logger.debug(f"[{self.cache_type.upper()}Cache] 使用代理: {proxy_url.split('@')[-1] if '@' in proxy_url else proxy_url}") async with AsyncSession() as session: logger.debug(f"[{self.cache_type.upper()}Cache] 开始下载: https://assets.grok.com{file_path}") diff --git a/app/services/grok/client.py b/app/services/grok/client.py index b275308..ef5b4b7 100644 --- a/app/services/grok/client.py +++ b/app/services/grok/client.py @@ -146,9 +146,15 @@ class GrokClient: raise GrokApiException("认证令牌缺失", "NO_AUTH_TOKEN") try: - # 构建请求头和代理 + # 构建请求头 headers = GrokClient._build_headers(auth_token) - proxy_config = GrokClient._get_proxy() + + # 获取代理配置 + proxy_url = setting.grok_config.get("proxy_url", "") + proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None + + if proxy_url: + logger.debug(f"[Client] 使用代理: {proxy_url.split('@')[-1] if '@' in proxy_url else proxy_url}") # 构建请求参数 request_kwargs = { @@ -157,7 +163,7 @@ class GrokClient: "impersonate": IMPERSONATE_BROWSER, "timeout": REQUEST_TIMEOUT, "stream": True, - "proxies": proxy_config if proxy_config else None + "proxies": proxies } # 在线程池中执行同步HTTP请求，避免阻塞事件循环 @@ -200,14 +206,6 @@ class GrokClient: return headers - @staticmethod - def _get_proxy() -> Dict[str, str]: - """获取代理配置""" - proxy_url = setting.grok_config.get("proxy_url", "") - if proxy_url: - return {"http": proxy_url, "https": proxy_url} - return {} - @staticmethod def _handle_error(response, auth_token: str): """处理错误响应""" diff --git a/app/services/grok/token.py b/app/services/grok/token.py index a8678b0..ac62222 100644 --- a/app/services/grok/token.py +++ b/app/services/grok/token.py @@ -258,6 +258,9 @@ class GrokTokenManager: # 获取代理配置 proxy_url = setting.grok_config.get("proxy_url", "") proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None + + if proxy_url: + logger.debug(f"[Token] 使用代理: {proxy_url.split('@')[-1] if '@' in proxy_url else proxy_url}") # 发送异步请求 async with AsyncSession() as session: diff --git a/app/services/grok/upload.py b/app/services/grok/upload.py index f5f92b3..c7ee7de 100644 --- a/app/services/grok/upload.py +++ b/app/services/grok/upload.py @@ -61,7 +61,11 @@ class ImageUploadManager: cf_clearance = setting.grok_config.get("cf_clearance", "") cookie = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token + proxy_url = setting.grok_config.get("proxy_url", "") + if proxy_url: + logger.debug(f"[Upload] 使用代理: {proxy_url.split('@')[-1] if '@' in proxy_url else proxy_url}") + proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None # 发送异步请求 diff --git a/app/template/admin.html b/app/template/admin.html index 41e850d..190505e 100644 --- a/app/template/admin.html +++ b/app/template/admin.html @@ -378,7 +378,7 @@ Proxy Url ? - +
@@ -222,8 +225,9 @@ 状态 普通剩余 高级剩余 - 创建时间 - 操作 + 标签 + 创建时间 + 操作 @@ -481,6 +485,50 @@ + + +
-

未使用

-

-

-
- -
-

限流中

-

-

-
- -
-

失效

-

-

-
- -
-

正常

+

Token 正常

-

-

总剩余

-

-

+

Token 未使用

+

-

-

普通剩余

-

-

+

Token 限流中

+

-

-

高级剩余

-

-

+

Token 失效

+

-

+
+ +
+

Chat 总剩余

+

-

+
+ +
+

Image 总剩余

+

-

+
+ +
+

Video 总剩余

+

无法统计

@@ -217,7 +218,7 @@ 标签 备注 创建时间 - 操作 + 操作 @@ -592,9 +593,9 @@ loadStats=async()=>{try{const r=await apiRequest('/api/stats');if(!r)return;const d=await r.json();if(d.success){const s=d.data;$('statTotal').textContent=s.total||0;['Unused','Limited','Expired','Active'].forEach((k,i)=>$(`stat${k}`).textContent=(s.normal?.[k.toLowerCase()]||0)+(s.super?.[k.toLowerCase()]||0))}}catch(e){console.error('加载统计失败:',e)}}, calcRemaining=()=>{let n=0,h=0;allTokens.forEach(t=>{if(t.remaining_queries>0)n+=t.remaining_queries;if(t.heavy_remaining_queries>0)h+=t.heavy_remaining_queries});return{normal:n,heavy:h,total:n+h}}, loadTokens=async()=>{try{const r=await apiRequest('/api/tokens');if(!r)return;const d=await r.json();d.success&&(allTokens=d.data.map(t=>({...t,tags:t.tags||[],note:t.note||''})),filteredTokens=allTokens,selectedTokens.clear(),renderTokens(),updateRemaining(),await loadAllTags())}catch(e){console.error('加载列表失败:',e)}}, - updateRemaining=()=>{const r=calcRemaining();['Total','Normal','Heavy'].forEach(k=>$(`stat${k}Remaining`).textContent=r[k.toLowerCase()]===0?'-':r[k.toLowerCase()].toLocaleString())} + updateRemaining=()=>{const r=calcRemaining();const chatTotal=r.total;const imageTotal=Math.floor(chatTotal/2);$('statChatRemaining').textContent=chatTotal===0?'-':chatTotal.toLocaleString();$('statImageRemaining').textContent=imageTotal===0?'-':imageTotal.toLocaleString();$('statVideoRemaining').textContent='无法统计'} - const renderTokens=()=>{const tb=$('tokenTableBody'),es=$('emptyState'),ss={'未使用':'bg-muted text-muted-foreground','限流中':'bg-orange-50 text-orange-700 border-orange-200','失效':'bg-destructive/10 text-destructive border-destructive/20','正常':'bg-green-50 text-green-700 border-green-200'},ts={sso:'bg-blue-50 text-blue-700 border-blue-200',ssoSuper:'bg-purple-50 text-purple-700 border-purple-200'},tl={sso:'SSO',ssoSuper:'SuperSSO'};if(!filteredTokens.length){tb.innerHTML='';es.classList.remove('hidden');$('selectAll').checked=false;return updateBatchActions()}es.classList.add('hidden');tb.innerHTML=filteredTokens.map(t=>{const tagsHtml=t.tags&&t.tags.length?t.tags.map(tag=>`${tag}`).join(' '):'-';const noteHtml=t.note&&t.note.length?`${t.note.length>20?t.note.substring(0,20)+'...':t.note}`:'-';return`
${t.token.substring(0,30)}...
${tl[t.token_type]}${t.status}${t.remaining_queries===-1?'-':t.remaining_queries}${t.heavy_remaining_queries===-1?'-':t.heavy_remaining_queries}
${tagsHtml}
${noteHtml}${t.created_time?new Date(t.created_time).toLocaleString('zh-CN',{dateStyle:'short',timeStyle:'short'}):'-'}
`}).join('');updateBatchActions()}, + const renderTokens=()=>{const tb=$('tokenTableBody'),es=$('emptyState'),ss={'未使用':'bg-muted text-muted-foreground','限流中':'bg-orange-50 text-orange-700 border-orange-200','失效':'bg-destructive/10 text-destructive border-destructive/20','正常':'bg-green-50 text-green-700 border-green-200'},ts={sso:'bg-blue-50 text-blue-700 border-blue-200',ssoSuper:'bg-purple-50 text-purple-700 border-purple-200'},tl={sso:'SSO',ssoSuper:'SuperSSO'};if(!filteredTokens.length){tb.innerHTML='';es.classList.remove('hidden');$('selectAll').checked=false;return updateBatchActions()}es.classList.add('hidden');tb.innerHTML=filteredTokens.map(t=>{const tagsHtml=t.tags&&t.tags.length?t.tags.map(tag=>`${tag}`).join(' '):'-';const noteHtml=t.note&&t.note.length?`${t.note.length>20?t.note.substring(0,20)+'...':t.note}`:'-';return`
${t.token.substring(0,30)}...
${tl[t.token_type]}${t.status}${t.remaining_queries===-1?'-':t.remaining_queries}${t.heavy_remaining_queries===-1?'-':t.heavy_remaining_queries}
${tagsHtml}
${noteHtml}${t.created_time?new Date(t.created_time).toLocaleString('zh-CN',{dateStyle:'short',timeStyle:'short'}):'-'}
`}).join('');updateBatchActions()}, toggleToken=t=>selectedTokens[selectedTokens.has(t)?'delete':'add'](t)||updateBatchActions(), toggleSelectAll=()=>{const sa=$('selectAll');sa.checked?filteredTokens.forEach(t=>selectedTokens.add(t.token)):selectedTokens.clear();renderTokens()}, updateBatchActions=()=>{const ba=$('batchActions'),sc=$('selectedCount'),c=selectedTokens.size;ba.classList[c>0?'add':'remove']('flex');ba.classList[c>0?'remove':'add']('hidden');c>0&&(sc.textContent=`已选择 ${c} 项`);$('selectAll').checked=filteredTokens.length>0&&c===filteredTokens.length}, From 685c2b576e4e196b6b9e7a94776119c46bb28f88 Mon Sep 17 00:00:00 2001 From: Zhi Yang <196515526+FakerL@users.noreply.github.com> Date: Fri, 21 Nov 2025 14:58:24 +0000 Subject: [PATCH 33/43] fix: use Redis.from_url() for proper SSL support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace custom _parse_url() with redis.Redis.from_url() which properly handles rediss:// SSL connections and all URL parameters. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- app/core/storage.py | 32 ++++++-------------------------- readme.md | 2 +- 2 files changed, 7 insertions(+), 27 deletions(-) diff --git a/app/core/storage.py b/app/core/storage.py index db96fc3..6ef93ec 100644 --- a/app/core/storage.py +++ b/app/core/storage.py @@ -319,14 +319,13 @@ class RedisStorage(BaseStorage): async def init_db(self) -> None: """初始化Redis""" try: - import redis.asyncio as redis - parsed = self._parse_url(self.redis_url) - logger.info(f"[Storage] Redis: {parsed['host']}:{parsed['port']}/{parsed['db']}") + import redis.asyncio as aioredis + parsed = urlparse(self.redis_url) + db = int(parsed.path.lstrip('/')) if parsed.path and parsed.path != '/' else 0 + logger.info(f"[Storage] Redis: {parsed.hostname}:{parsed.port or 6379}/{db}") - self._redis = redis.Redis( - host=parsed['host'], port=parsed['port'], password=parsed.get('password'), - username=parsed.get('username'), db=parsed.get('db', 0), - encoding="utf-8", decode_responses=True + self._redis = aioredis.Redis.from_url( + self.redis_url, encoding="utf-8", decode_responses=True ) await self._redis.ping() @@ -341,25 +340,6 @@ class RedisStorage(BaseStorage): logger.error(f"[Storage] Redis初始化失败: {e}") raise - def _parse_url(self, url: str) -> Dict[str, Any]: - """解析Redis URL""" - if url.startswith('redis://'): - url = url[8:] - p = urlparse(f'//{url}') - - result = { - 'host': p.hostname or 'localhost', - 'port': p.port or 6379, - 'db': int(p.path.lstrip('/')) if p.path and p.path != '/' else 0, - 'username': unquote(p.username) if p.username else None, - 'password': unquote(p.password) if p.password else None - } - - if result['password'] and not result['username']: - result['username'] = 'default' - - return result - async def _sync_data(self) -> None: """同步数据""" try: diff --git a/readme.md b/readme.md index 6087ec7..2e1e76e 100644 --- a/readme.md +++ b/readme.md @@ -78,7 +78,7 @@ services: # - DATABASE_URL=mysql://user:password@host:3306/grok2api ## MySQL格式: mysql://user:password@host:port/database - ## Redis格式: redis://host:port/db 或 redis://user:password@host:port/db + ## Redis格式: redis://host:port/db 或 redis://user:password@host:port/db (SSL: rediss://) volumes: grok_data: From aa02c32fa1bd3d6f95aa6c69b70c340e264a0246 Mon Sep 17 00:00:00 2001 From: bloodycrownD <104297061+bloodycrownD@users.noreply.github.com> Date: Sat, 22 Nov 2025 00:50:26 +0800 Subject: [PATCH 34/43] Fix index assignment in OpenAIChatCompletionChunkChoice --- app/services/grok/processer.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/services/grok/processer.py b/app/services/grok/processer.py index 4eeb5b9..a2a8fe9 100644 --- a/app/services/grok/processer.py +++ b/app/services/grok/processer.py @@ -154,7 +154,7 @@ class GrokResponseProcessor: created=int(time.time()), model=model or "grok-4-mini-thinking-tahoe", choices=[OpenAIChatCompletionChunkChoice( - index=chunk_index, + index=0, delta=OpenAIChatCompletionChunkMessage( role="assistant", content=content @@ -431,4 +431,4 @@ class GrokResponseProcessor: finish_reason="stop" )], usage=None - ) \ No newline at end of file + ) From 1a16748abd889d669d64a1d0357f0c70ee40b44d Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Sat, 22 Nov 2025 13:57:22 +0800 Subject: [PATCH 35/43] fix: delete the "chunk_index" --- app/services/grok/processer.py | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/app/services/grok/processer.py b/app/services/grok/processer.py index a2a8fe9..d8de920 100644 --- a/app/services/grok/processer.py +++ b/app/services/grok/processer.py @@ -132,7 +132,6 @@ class GrokResponseProcessor: is_image = False is_thinking = False thinking_finished = False - chunk_index = 0 model = None filtered_tags = setting.grok_config.get("filtered_tags", "").split(",") video_progress_started = False @@ -218,14 +217,12 @@ class GrokResponseProcessor: else: content = f"视频已生成{progress}%\\n" yield make_chunk(content) - chunk_index += 1 # 视频URL if v_url: logger.debug("[Processor] 视频生成完成") video_content = await GrokResponseProcessor._build_video_content(v_url, auth_token) yield make_chunk(video_content) - chunk_index += 1 continue @@ -252,22 +249,16 @@ class GrokResponseProcessor: parts = base64_str.split(",", 1) if len(parts) == 2: yield make_chunk(f"![Generated Image](data:{parts[0]},") - chunk_index += 1 # 8KB分块 for i in range(0, len(parts[1]), 8192): yield make_chunk(parts[1][i:i+8192]) - chunk_index += 1 yield make_chunk(")\\n") - chunk_index += 1 else: yield make_chunk(f"![Generated Image]({base64_str})\\n") - chunk_index += 1 else: yield make_chunk(f"![Generated Image]({base64_str})\\n") - chunk_index += 1 else: yield make_chunk(f"![Generated Image](https://assets.grok.com/{img})\\n") - chunk_index += 1 else: # URL模式 await image_cache_service.download_image(f"/{img}", auth_token) @@ -283,7 +274,6 @@ class GrokResponseProcessor: return elif token: yield make_chunk(token) - chunk_index += 1 # 对话处理 else: @@ -341,7 +331,6 @@ class GrokResponseProcessor: if not should_skip: yield make_chunk(content) - chunk_index += 1 is_thinking = current_is_thinking From cd72db567a181a90cfa271511461173ea7ea1ba4 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Sat, 22 Nov 2025 14:05:15 +0800 Subject: [PATCH 36/43] fix: include setting.toml in Docker image to prevent first-run failure --- Dockerfile | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/Dockerfile b/Dockerfile index c7882e1..2a3d598 100644 --- a/Dockerfile +++ b/Dockerfile @@ -31,14 +31,17 @@ RUN rm -rf /usr/share/doc/* \ # 从构建阶段复制已安装的包 COPY --from=builder /install /usr/local -# 创建必要的目录和文件 -RUN mkdir -p /app/logs /app/data/temp/image /app/data/temp/video && \ - echo '{"ssoNormal": {}, "ssoSuper": {}}' > /app/data/token.json +# 创建必要的目录 +RUN mkdir -p /app/logs /app/data/temp/image /app/data/temp/video -# 复制应用代码 +# 复制应用代码和配置文件 COPY app/ ./app/ +COPY data/setting.toml ./data/setting.toml COPY main.py . +# 创建默认的 token.json 文件 +RUN echo '{"ssoNormal": {}, "ssoSuper": {}}' > /app/data/token.json + # 删除 Python 字节码和缓存 ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 From 300b03fdde6138c43b02e9a9c040083b578c5244 Mon Sep 17 00:00:00 2001 From: itismygo <167516635+itismygo@users.noreply.github.com> Date: Fri, 12 Dec 2025 02:26:29 +0000 Subject: [PATCH 37/43] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20grok-imagine-?= =?UTF-8?q?0.9=20=E7=BA=AF=E6=96=87=E6=9C=AC=E7=94=9F=E6=88=90=E5=9B=BE?= =?UTF-8?q?=E7=89=87=E7=9A=84bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 问题描述 当使用 grok-imagine-0.9 模型进行纯文本到图片生成时，会抛出 "list index out of range" 错误。 ## 根本原因 1. **模型配置错误**: grok-imagine-0.9 被错误地标记为视频模型 (`is_video_model: True`) 2. **数组访问bug**: 在没有图片附件时，代码尝试访问空列表的第一个元素 `payload.get("fileAttachments", [""])[0]` ## 修复内容 ### app/models/grok_models.py - 将 `is_video_model` 从 `True` 改为 `False` - 更新 description 为 "Image generation model powered by Grok" ### app/services/grok/client.py - 修复空列表访问bug，使用安全的访问方式： ```python file_attachments = payload.get("fileAttachments", []) ref_id = post_id or (file_attachments[0] if file_attachments else "") ``` ## 测试结果 ✅ 纯文本生成图片：成功 ✅ 文本对话：正常 ✅ 其他模型：不受影响 ## 相关信息 根据 xAI 官方文档，Grok Imagine 主要是图像生成模型（使用 Aurora 引擎），支持 text-to-image 功能。 --- app/models/grok_models.py | 4 ++-- app/services/grok/client.py | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/app/models/grok_models.py b/app/models/grok_models.py index 6fab827..fe15fc7 100644 --- a/app/models/grok_models.py +++ b/app/models/grok_models.py @@ -103,13 +103,13 @@ _MODEL_CONFIG: Dict[str, Dict[str, Any]] = { "cost": {"type": "low_cost", "multiplier": 1, "description": "计1次调用"}, "requires_super": False, "display_name": "Grok Imagine 0.9", - "description": "Video generation model powered by Grok", + "description": "Image generation model powered by Grok", "raw_model_path": "xai/grok-imagine-0.9", "default_temperature": 1.0, "default_max_output_tokens": 8192, "supported_max_output_tokens": 131072, "default_top_p": 0.95, - "is_video_model": True + "is_video_model": False } } diff --git a/app/services/grok/client.py b/app/services/grok/client.py index 456bdfb..277d640 100644 --- a/app/services/grok/client.py +++ b/app/services/grok/client.py @@ -188,7 +188,8 @@ class GrokClient: # 构建请求 headers = GrokClient._build_headers(token) if model == "grok-imagine-0.9": - ref_id = post_id or payload.get("fileAttachments", [""])[0] + 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 eff701c485c484134ce7907f0ecff996f6ff5c2c Mon Sep 17 00:00:00 2001 From: itismygo <167516635+itismygo@users.noreply.github.com> Date: Fri, 12 Dec 2025 04:14:01 +0000 Subject: [PATCH 38/43] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20grok-imagine-?= =?UTF-8?q?0.9=20=E5=90=8C=E6=97=B6=E6=94=AF=E6=8C=81=E6=96=87=E7=94=9F?= =?UTF-8?q?=E5=9B=BE=E5=92=8C=E5=9B=BE=E7=94=9F=E8=A7=86=E9=A2=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 恢复 is_video_model: True 以支持视频生成 - 更新模型描述说明支持两种模式 - 保留空列表安全访问修复 根据 xAI 官方文档，grok-imagine-0.9 支持： 1. 文本 → 图片 (text-to-image) 2. 图片 + 文本 → 视频 (image-to-video) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 --- app/models/grok_models.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/models/grok_models.py b/app/models/grok_models.py index fe15fc7..b067c54 100644 --- a/app/models/grok_models.py +++ b/app/models/grok_models.py @@ -103,13 +103,13 @@ _MODEL_CONFIG: Dict[str, Dict[str, Any]] = { "cost": {"type": "low_cost", "multiplier": 1, "description": "计1次调用"}, "requires_super": False, "display_name": "Grok Imagine 0.9", - "description": "Image generation model powered by Grok", + "description": "Image and video generation model. Supports text-to-image and image-to-video generation.", "raw_model_path": "xai/grok-imagine-0.9", "default_temperature": 1.0, "default_max_output_tokens": 8192, "supported_max_output_tokens": 131072, "default_top_p": 0.95, - "is_video_model": False + "is_video_model": True } } From 3aef4347d74984585b6420197c66e49e540f4aeb Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Tue, 16 Dec 2025 16:40:32 +0800 Subject: [PATCH 39/43] fix: automatically disable token expiration --- app/services/grok/processer.py | 42 +++++++++++++++++----------------- app/services/grok/token.py | 9 ++++++-- 2 files changed, 28 insertions(+), 23 deletions(-) diff --git a/app/services/grok/processer.py b/app/services/grok/processer.py index d8de920..2c51c69 100644 --- a/app/services/grok/processer.py +++ b/app/services/grok/processer.py @@ -210,12 +210,12 @@ class GrokResponseProcessor: last_video_progress = progress if show_thinking: if not video_progress_started: - content = f"视频已生成{progress}%\\n" + content = f"视频已生成{progress}%\n" video_progress_started = True elif progress < 100: - content = f"视频已生成{progress}%\\n" + content = f"视频已生成{progress}%\n" else: - content = f"视频已生成{progress}%\\n" + content = f"视频已生成{progress}%\n" yield make_chunk(content) # 视频URL @@ -252,23 +252,23 @@ class GrokResponseProcessor: # 8KB分块 for i in range(0, len(parts[1]), 8192): yield make_chunk(parts[1][i:i+8192]) - yield make_chunk(")\\n") + yield make_chunk(")\n") else: - yield make_chunk(f"![Generated Image]({base64_str})\\n") + yield make_chunk(f"![Generated Image]({base64_str})\n") else: - yield make_chunk(f"![Generated Image]({base64_str})\\n") + yield make_chunk(f"![Generated Image]({base64_str})\n") else: - yield make_chunk(f"![Generated Image](https://assets.grok.com/{img})\\n") + yield make_chunk(f"![Generated Image](https://assets.grok.com/{img})\n") else: # URL模式 await image_cache_service.download_image(f"/{img}", auth_token) img_path = img.replace('/', '-') base_url = setting.global_config.get("base_url", "") img_url = f"{base_url}/images/{img_path}" if base_url else f"/images/{img_path}" - content += f"![Generated Image]({img_url})\\n" + content += f"![Generated Image]({img_url})\n" except Exception as e: logger.warning(f"[Processor] 处理图片失败: {e}") - content += f"![Generated Image](https://assets.grok.com/{img})\\n" + content += f"![Generated Image](https://assets.grok.com/{img})\n" yield make_chunk(content.strip(), "stop") return @@ -298,9 +298,9 @@ class GrokResponseProcessor: title = result.get("title", "") url = result.get("url", "") preview = result.get("preview", "") - preview_clean = preview.replace("\\n", "") if isinstance(preview, str) else "" - token += f'\\n- [{title}]({url} "{preview_clean}")' - token += "\\n" + preview_clean = preview.replace("\n", "") if isinstance(preview, str) else "" + token += f'\n- [{title}]({url} "{preview_clean}")' + token += "\n" else: continue else: @@ -318,12 +318,12 @@ class GrokResponseProcessor: should_skip = False if not is_thinking and current_is_thinking: if show_thinking: - content = f"\\n{content}" + content = f"\n{content}" else: should_skip = True elif is_thinking and not current_is_thinking: if show_thinking: - content = f"\\n\\n{content}" + content = f"\n\n{content}" thinking_finished = True elif current_is_thinking: if not show_thinking: @@ -369,11 +369,11 @@ class GrokResponseProcessor: video_path = video_url.replace('/', '-') base_url = setting.global_config.get("base_url", "") local_url = f"{base_url}/images/{video_path}" if base_url else f"/images/{video_path}" - return f'\\n' + return f'\n' except Exception as e: logger.warning(f"[Processor] 缓存视频失败: {e}") - return f'\\n' + return f'\n' @staticmethod async def _append_images(content: str, images: list, auth_token: str) -> str: @@ -385,21 +385,21 @@ class GrokResponseProcessor: if image_mode == "base64": base64_str = await image_cache_service.download_base64(f"/{img}", auth_token) if base64_str: - content += f"\\n![Generated Image]({base64_str})" + content += f"\n![Generated Image]({base64_str})" else: - content += f"\\n![Generated Image](https://assets.grok.com/{img})" + content += f"\n![Generated Image](https://assets.grok.com/{img})" else: cache_path = await image_cache_service.download_image(f"/{img}", auth_token) if cache_path: img_path = img.replace('/', '-') base_url = setting.global_config.get("base_url", "") img_url = f"{base_url}/images/{img_path}" if base_url else f"/images/{img_path}" - content += f"\\n![Generated Image]({img_url})" + content += f"\n![Generated Image]({img_url})" else: - content += f"\\n![Generated Image](https://assets.grok.com/{img})" + content += f"\n![Generated Image](https://assets.grok.com/{img})" except Exception as e: logger.warning(f"[Processor] 处理图片失败: {e}") - content += f"\\n![Generated Image](https://assets.grok.com/{img})" + content += f"\n![Generated Image](https://assets.grok.com/{img})" return content diff --git a/app/services/grok/token.py b/app/services/grok/token.py index 1e58008..95aee40 100644 --- a/app/services/grok/token.py +++ b/app/services/grok/token.py @@ -169,8 +169,13 @@ class GrokTokenManager: unused, used = [], [] for key, data in tokens.items(): + # 跳过已失效的token if data.get("status") == "expired": continue + + # 跳过失败次数过多的token（任何错误状态码） + if data.get("failedCount", 0) >= MAX_FAILURES: + continue remaining = int(data.get(field, -1)) if remaining == 0: @@ -312,9 +317,9 @@ class GrokTokenManager: f"次数: {data['failedCount']}/{MAX_FAILURES}, 原因: {msg}" ) - if status == TOKEN_INVALID and data["failedCount"] >= MAX_FAILURES: + if 400 <= status < 500 and data["failedCount"] >= MAX_FAILURES: data["status"] = "expired" - logger.error(f"[Token] 标记失效: {sso[:10]}... (连续401错误{data['failedCount']}次)") + logger.error(f"[Token] 标记失效: {sso[:10]}... (连续{status}错误{data['failedCount']}次)") await self._save_data() From 1ddec29ffb22b3f6b7ae64f584ec7ad90b923ad3 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Tue, 16 Dec 2025 17:38:15 +0800 Subject: [PATCH 40/43] feat: optimize concurrency limit --- Dockerfile | 42 +++--- app/core/config.py | 6 +- app/services/grok/client.py | 16 ++- app/services/grok/token.py | 117 ++++++++++++--- docker-entrypoint.sh | 53 +++++++ main.py | 45 +++++- requirements.txt | 2 + test_concurrency.py | 276 ++++++++++++++++++++++++++++++++++++ test_concurrency.sh | 177 +++++++++++++++++++++++ 9 files changed, 688 insertions(+), 46 deletions(-) create mode 100644 docker-entrypoint.sh create mode 100755 test_concurrency.py create mode 100755 test_concurrency.sh diff --git a/Dockerfile b/Dockerfile index 2a3d598..f625296 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,13 +6,13 @@ WORKDIR /build # 安装依赖到独立目录 COPY requirements.txt . RUN pip install --no-cache-dir --only-binary=:all: --prefix=/install -r requirements.txt && \ - find /install -type d -name "__pycache__" -exec rm -rf {} + 2>/dev/null || true && \ - find /install -type d -name "tests" -exec rm -rf {} + 2>/dev/null || true && \ - find /install -type d -name "test" -exec rm -rf {} + 2>/dev/null || true && \ - find /install -type d -name "*.dist-info" -exec sh -c 'rm -f "$1"/RECORD "$1"/INSTALLER' _ {} \; && \ - find /install -type f -name "*.pyc" -delete && \ - find /install -type f -name "*.pyo" -delete && \ - find /install -name "*.so" -exec strip --strip-unneeded {} \; 2>/dev/null || true + find /install -type d -name "__pycache__" -exec rm -rf {} + 2>/dev/null || true && \ + find /install -type d -name "tests" -exec rm -rf {} + 2>/dev/null || true && \ + find /install -type d -name "test" -exec rm -rf {} + 2>/dev/null || true && \ + find /install -type d -name "*.dist-info" -exec sh -c 'rm -f "$1"/RECORD "$1"/INSTALLER' _ {} \; && \ + find /install -type f -name "*.pyc" -delete && \ + find /install -type f -name "*.pyo" -delete && \ + find /install -name "*.so" -exec strip --strip-unneeded {} \; 2>/dev/null || true # 运行阶段 - 使用最小镜像 FROM python:3.11-slim @@ -21,31 +21,35 @@ WORKDIR /app # 清理基础镜像中的冗余文件 RUN rm -rf /usr/share/doc/* \ - /usr/share/man/* \ - /usr/share/locale/* \ - /var/cache/apt/* \ - /var/lib/apt/lists/* \ - /tmp/* \ - /var/tmp/* + /usr/share/man/* \ + /usr/share/locale/* \ + /var/cache/apt/* \ + /var/lib/apt/lists/* \ + /tmp/* \ + /var/tmp/* # 从构建阶段复制已安装的包 COPY --from=builder /install /usr/local -# 创建必要的目录 +# 创建必要的目录（包括用于挂载的data目录） RUN mkdir -p /app/logs /app/data/temp/image /app/data/temp/video -# 复制应用代码和配置文件 +# 复制应用代码 COPY app/ ./app/ -COPY data/setting.toml ./data/setting.toml COPY main.py . -# 创建默认的 token.json 文件 -RUN echo '{"ssoNormal": {}, "ssoSuper": {}}' > /app/data/token.json +# 复制并设置 entrypoint 脚本 +COPY docker-entrypoint.sh /usr/local/bin/ +RUN chmod +x /usr/local/bin/docker-entrypoint.sh # 删除 Python 字节码和缓存 ENV PYTHONDONTWRITEBYTECODE=1 \ - PYTHONUNBUFFERED=1 + PYTHONUNBUFFERED=1 EXPOSE 8000 +# 使用 entrypoint 脚本初始化配置 +ENTRYPOINT ["docker-entrypoint.sh"] + +# 默认启动命令 CMD ["python", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/app/core/config.py b/app/core/config.py index 8036266..4b63ec6 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -28,7 +28,11 @@ DEFAULT_GLOBAL = { "admin_password": "admin", "admin_username": "admin", "image_cache_max_size_mb": 512, - "video_cache_max_size_mb": 1024 + "video_cache_max_size_mb": 1024, + "max_upload_concurrency": 20, # 最大并发上传数 + "max_request_concurrency": 50, # 最大并发请求数 + "batch_save_interval": 1.0, # 批量保存间隔（秒） + "batch_save_threshold": 10 # 触发批量保存的变更数阈值 } diff --git a/app/services/grok/client.py b/app/services/grok/client.py index 277d640..435a722 100644 --- a/app/services/grok/client.py +++ b/app/services/grok/client.py @@ -21,13 +21,23 @@ API_ENDPOINT = "https://grok.com/rest/app-chat/conversations/new" TIMEOUT = 120 BROWSER = "chrome133a" MAX_RETRY = 3 -MAX_UPLOADS = 5 +MAX_UPLOADS = 20 # 提高并发上传限制以支持更高并发 class GrokClient: """Grok API 客户端""" - _upload_sem = asyncio.Semaphore(MAX_UPLOADS) + _upload_sem = None # 延迟初始化 + + @staticmethod + def _get_upload_semaphore(): + """获取上传信号量（动态配置）""" + if GrokClient._upload_sem is None: + # 从配置读取，如果不可用则使用默认值 + max_concurrency = setting.global_config.get("max_upload_concurrency", MAX_UPLOADS) + GrokClient._upload_sem = asyncio.Semaphore(max_concurrency) + logger.debug(f"[Client] 初始化上传并发限制: {max_concurrency}") + return GrokClient._upload_sem @staticmethod async def openai_to_grok(request: dict): @@ -109,7 +119,7 @@ class GrokClient: return [], [] async def upload_limited(url): - async with GrokClient._upload_sem: + async with GrokClient._get_upload_semaphore(): return await ImageUploadManager.upload(url, token) results = await asyncio.gather(*[upload_limited(u) for u in urls], return_exceptions=True) diff --git a/app/services/grok/token.py b/app/services/grok/token.py index 95aee40..08ab037 100644 --- a/app/services/grok/token.py +++ b/app/services/grok/token.py @@ -4,6 +4,7 @@ import orjson import time import asyncio import aiofiles +import portalocker from pathlib import Path from curl_cffi.requests import AsyncSession from typing import Dict, Any, Optional, Tuple @@ -43,7 +44,13 @@ class GrokTokenManager: self._file_lock = asyncio.Lock() self.token_file.parent.mkdir(parents=True, exist_ok=True) self._storage = None - self._load_data() + self.token_data = None # 延迟加载 + + # 批量保存队列 + self._save_pending = False # 标记是否有待保存的数据 + self._save_task = None # 后台保存任务 + self._shutdown = False # 关闭标志 + self._initialized = True logger.debug(f"[Token] 初始化完成: {self.token_file}") @@ -51,14 +58,21 @@ class GrokTokenManager: """设置存储实例""" self._storage = storage - def _load_data(self) -> None: - """加载Token数据""" + async def _load_data(self) -> None: + """异步加载Token数据（支持多进程）""" default = {TokenType.NORMAL.value: {}, TokenType.SUPER.value: {}} try: if self.token_file.exists(): - with open(self.token_file, "r", encoding="utf-8") as f: - self.token_data = orjson.loads(f.read()) + # 使用进程锁读取文件 + async with self._file_lock: + with open(self.token_file, "r", encoding="utf-8") as f: + portalocker.lock(f, portalocker.LOCK_SH) # 共享锁（读） + try: + content = f.read() + self.token_data = orjson.loads(content) + finally: + portalocker.unlock(f) else: self.token_data = default logger.debug("[Token] 创建新数据文件") @@ -67,18 +81,69 @@ class GrokTokenManager: self.token_data = default async def _save_data(self) -> None: - """保存Token数据""" + """保存Token数据（支持多进程）""" try: if not self._storage: async with self._file_lock: - async with aiofiles.open(self.token_file, "w", encoding="utf-8") as f: - await f.write(orjson.dumps(self.token_data, option=orjson.OPT_INDENT_2).decode()) + # 使用进程锁写入文件 + with open(self.token_file, "w", encoding="utf-8") as f: + portalocker.lock(f, portalocker.LOCK_EX) # 独占锁（写） + try: + content = orjson.dumps(self.token_data, option=orjson.OPT_INDENT_2).decode() + f.write(content) + f.flush() # 确保写入磁盘 + finally: + portalocker.unlock(f) else: await self._storage.save_tokens(self.token_data) except IOError as e: logger.error(f"[Token] 保存失败: {e}") raise GrokApiException(f"保存失败: {e}", "TOKEN_SAVE_ERROR", {"file": str(self.token_file)}) + def _mark_dirty(self) -> None: + """标记有待保存的数据""" + self._save_pending = True + + async def _batch_save_worker(self) -> None: + """批量保存后台任务""" + from app.core.config import setting + + interval = setting.global_config.get("batch_save_interval", 1.0) + logger.info(f"[Token] 批量保存任务已启动，间隔: {interval}s") + + while not self._shutdown: + await asyncio.sleep(interval) + + if self._save_pending and not self._shutdown: + try: + await self._save_data() + self._save_pending = False + logger.debug("[Token] 批量保存完成") + except Exception as 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] 批量保存任务已创建") + + async def shutdown(self) -> None: + """关闭并刷新所有待保存数据""" + self._shutdown = True + + if self._save_task: + self._save_task.cancel() + try: + await self._save_task + except asyncio.CancelledError: + pass + + # 最终刷新 + if self._save_pending: + await self._save_data() + logger.info("[Token] 关闭时刷新完成") + @staticmethod def _extract_sso(auth_token: str) -> Optional[str]: """提取SSO值""" @@ -117,7 +182,7 @@ class GrokTokenManager: } count += 1 - await self._save_data() + self._mark_dirty() # 批量保存 logger.info(f"[Token] 添加 {count} 个 {token_type.value} Token") async def delete_token(self, tokens: list[str], token_type: TokenType) -> None: @@ -131,7 +196,7 @@ class GrokTokenManager: del self.token_data[token_type.value][token] count += 1 - await self._save_data() + self._mark_dirty() # 批量保存 logger.info(f"[Token] 删除 {count} 个 {token_type.value} Token") async def update_token_tags(self, token: str, token_type: TokenType, tags: list[str]) -> None: @@ -141,7 +206,7 @@ class GrokTokenManager: cleaned = [t.strip() for t in tags if t and t.strip()] self.token_data[token_type.value][token]["tags"] = cleaned - await self._save_data() + self._mark_dirty() # 批量保存 logger.info(f"[Token] 更新标签: {token[:10]}... -> {cleaned}") async def update_token_note(self, token: str, token_type: TokenType, note: str) -> None: @@ -150,20 +215,40 @@ class GrokTokenManager: raise GrokApiException("Token不存在", "TOKEN_NOT_FOUND", {"token": token[:10]}) self.token_data[token_type.value][token]["note"] = note.strip() - await self._save_data() + self._mark_dirty() # 批量保存 logger.info(f"[Token] 更新备注: {token[:10]}...") def get_tokens(self) -> Dict[str, Any]: """获取所有Token""" return self.token_data.copy() + def _reload_if_needed(self) -> None: + """在多进程模式下重新加载数据（同步版本，用于select_token）""" + # 只在文件模式且多进程环境下才重新加载 + if self._storage: + return # 数据库模式不需要 + + try: + if self.token_file.exists(): + with open(self.token_file, "r", encoding="utf-8") as f: + portalocker.lock(f, portalocker.LOCK_SH) + try: + content = f.read() + self.token_data = orjson.loads(content) + finally: + portalocker.unlock(f) + except Exception as e: + logger.warning(f"[Token] 重新加载失败: {e}") + def get_token(self, model: str) -> str: """获取Token""" jwt = self.select_token(model) return f"sso-rw={jwt};sso={jwt}" def select_token(self, model: str) -> str: - """选择最优Token""" + """选择最优Token（多进程安全）""" + # 重新加载最新数据（多进程模式） + self._reload_if_needed() def select_best(tokens: Dict[str, Any], field: str) -> Tuple[Optional[str], Optional[int]]: """选择最佳Token""" unused, used = [], [] @@ -285,7 +370,7 @@ class GrokTokenManager: self.token_data[token_type][sso]["remainingQueries"] = normal if heavy is not None: self.token_data[token_type][sso]["heavyremainingQueries"] = heavy - await self._save_data() + self._mark_dirty() # 批量保存 logger.info(f"[Token] 更新限制: {sso[:10]}...") return logger.warning(f"[Token] 未找到: {sso[:10]}...") @@ -321,7 +406,7 @@ class GrokTokenManager: data["status"] = "expired" logger.error(f"[Token] 标记失效: {sso[:10]}... (连续{status}错误{data['failedCount']}次)") - await self._save_data() + self._mark_dirty() # 批量保存 except Exception as e: logger.error(f"[Token] 记录失败错误: {e}") @@ -341,7 +426,7 @@ class GrokTokenManager: data["failedCount"] = 0 data["lastFailureTime"] = None data["lastFailureReason"] = None - await self._save_data() + self._mark_dirty() # 批量保存 logger.info(f"[Token] 重置失败计数: {sso[:10]}...") except Exception as e: diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh new file mode 100644 index 0000000..b98e9f5 --- /dev/null +++ b/docker-entrypoint.sh @@ -0,0 +1,53 @@ +#!/bin/sh +set -e + +# 初始化配置文件（如果不存在） +echo "[Grok2API] 检查配置文件..." + +# 确保数据目录存在 +mkdir -p /app/data/temp/image /app/data/temp/video /app/logs + +# 如果 setting.toml 不存在，创建默认配置 +if [ ! -f /app/data/setting.toml ]; then + echo "[Grok2API] 初始化 setting.toml..." + cat > /app/data/setting.toml << 'EOF' +[global] +base_url = "http://localhost:8000" +log_level = "INFO" +image_mode = "url" +admin_password = "admin" +admin_username = "admin" +image_cache_max_size_mb = 512 +video_cache_max_size_mb = 1024 +max_upload_concurrency = 20 +max_request_concurrency = 50 +batch_save_interval = 1.0 +batch_save_threshold = 10 + +[grok] +api_key = "" +proxy_url = "" +cache_proxy_url = "" +cf_clearance = "" +x_statsig_id = "" +dynamic_statsig = true +filtered_tags = "xaiartifact,xai:tool_usage_card,grok:render" +stream_chunk_timeout = 120 +stream_total_timeout = 600 +stream_first_response_timeout = 30 +temporary = true +show_thinking = true +EOF +fi + +# 如果 token.json 不存在，创建空token文件 +if [ ! -f /app/data/token.json ]; then + echo "[Grok2API] 初始化 token.json..." + echo '{"ssoNormal": {}, "ssoSuper": {}}' > /app/data/token.json +fi + +echo "[Grok2API] 配置文件检查完成" +echo "[Grok2API] 启动应用..." + +# 执行传入的命令 +exec "$@" diff --git a/main.py b/main.py index 2b9c463..a042695 100644 --- a/main.py +++ b/main.py @@ -26,11 +26,14 @@ async def lifespan(app: FastAPI): """ 启动顺序: 1. 初始化核心服务 (storage, settings, token_manager) - 2. 启动MCP服务生命周期 + 2. 异步加载 token 数据 + 3. 启动批量保存任务 + 4. 启动MCP服务生命周期 关闭顺序 (LIFO): 1. 关闭MCP服务生命周期 - 2. 关闭核心服务 + 2. 关闭批量保存任务并刷新数据 + 3. 关闭核心服务 """ # --- 启动过程 --- # 1. 初始化核心服务 @@ -41,12 +44,18 @@ async def lifespan(app: FastAPI): setting.set_storage(storage) token_manager.set_storage(storage) - # 重新加载配置和token数据 + # 2. 重新加载配置 await setting.reload() - token_manager._load_data() logger.info("[Grok2API] 核心服务初始化完成") + + # 3. 异步加载 token 数据 + await token_manager._load_data() + logger.info("[Grok2API] Token数据加载完成") + + # 4. 启动批量保存任务 + await token_manager.start_batch_save() - # 2. 管理MCP服务的生命周期 + # 5. 管理MCP服务的生命周期 mcp_lifespan_context = mcp_app.lifespan(app) await mcp_lifespan_context.__aenter__() logger.info("[MCP] MCP服务初始化完成") @@ -61,7 +70,11 @@ async def lifespan(app: FastAPI): await mcp_lifespan_context.__aexit__(None, None, None) logger.info("[MCP] MCP服务已关闭") - # 2. 关闭核心服务 + # 2. 关闭批量保存任务并刷新数据 + await token_manager.shutdown() + logger.info("[Token] Token管理器已关闭") + + # 3. 关闭核心服务 await storage_manager.close() logger.info("[Grok2API] 应用关闭成功") @@ -111,4 +124,22 @@ app.mount("", mcp_app) if __name__ == "__main__": import uvicorn - uvicorn.run(app, host="0.0.0.0", port=8001) \ No newline at end of file + import os + + # 从环境变量读取 worker 数量，默认为 1 + workers = int(os.getenv("WORKERS", "1")) + + # 提示多进程模式 + if workers > 1: + logger.info( + f"[Grok2API] 多进程模式已启用 (workers={workers})。" + f"建议使用 Redis/MySQL 存储以获得最佳性能。" + ) + + uvicorn.run( + "main:app", # 使用字符串以支持多worker + host="0.0.0.0", + port=8001, + workers=workers, + loop="uvloop" if workers == 1 else "auto" # 单进程用uvloop，多进程用auto + ) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index cab633d..85145fa 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,12 +1,14 @@ toml==0.10.2 fastapi==0.119.0 uvicorn==0.37.0 +uvloop==0.21.0 python-dotenv==1.1.1 curl_cffi==0.13.0 requests==2.32.5 starlette==0.48.0 pydantic==2.12.2 aiofiles==25.1.0 +portalocker==3.0.0 aiomysql==0.2.0 redis==6.4.0 fastmcp==2.12.4 diff --git a/test_concurrency.py b/test_concurrency.py new file mode 100755 index 0000000..c0089b1 --- /dev/null +++ b/test_concurrency.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +""" +Grok2API 并发性能测试脚本 + +测试不同并发级别下的API性能表现 +""" + +import asyncio +import aiohttp +import time +import statistics +import argparse +from datetime import datetime +from typing import List, Dict, Any +import json + + +class ConcurrencyTester: + """并发测试器""" + + def __init__(self, base_url: str, api_key: str = None): + self.base_url = base_url.rstrip('/') + self.api_key = api_key + self.results: List[Dict[str, Any]] = [] + + async def test_request(self, session: aiohttp.ClientSession, request_id: int) -> Dict[str, Any]: + """发送单个测试请求""" + url = f"{self.base_url}/v1/chat/completions" + + headers = { + "Content-Type": "application/json" + } + if self.api_key: + headers["Authorization"] = f"Bearer {self.api_key}" + + payload = { + "model": "grok-3-fast", + "messages": [ + {"role": "user", "content": f"测试请求 #{request_id}，请简短回复OK"} + ], + "stream": False, + "max_tokens": 10 + } + + start_time = time.time() + + try: + async with session.post(url, json=payload, headers=headers, timeout=30) as response: + status = response.status + + if status == 200: + data = await response.json() + elapsed = time.time() - start_time + + return { + "id": request_id, + "status": "success", + "http_status": status, + "elapsed": elapsed, + "response_length": len(json.dumps(data)) + } + else: + elapsed = time.time() - start_time + error_text = await response.text() + + return { + "id": request_id, + "status": "error", + "http_status": status, + "elapsed": elapsed, + "error": error_text[:200] + } + + except asyncio.TimeoutError: + elapsed = time.time() - start_time + return { + "id": request_id, + "status": "timeout", + "elapsed": elapsed, + "error": "Request timeout" + } + + except Exception as e: + elapsed = time.time() - start_time + return { + "id": request_id, + "status": "exception", + "elapsed": elapsed, + "error": str(e) + } + + async def run_concurrent_test(self, concurrency: int, total_requests: int): + """运行并发测试""" + print(f"\n{'='*60}") + print(f"📊 测试配置：并发数 {concurrency}, 总请求数 {total_requests}") + print(f"{'='*60}") + + connector = aiohttp.TCPConnector(limit=concurrency, limit_per_host=concurrency) + timeout = aiohttp.ClientTimeout(total=60) + + async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session: + # 预热 + print("🔥 预热中...") + await self.test_request(session, 0) + + # 开始测试 + print(f"🚀 开始并发测试...") + start_time = time.time() + + # 创建任务 + tasks = [] + for i in range(1, total_requests + 1): + task = asyncio.create_task(self.test_request(session, i)) + tasks.append(task) + + # 控制并发数 + if len(tasks) >= concurrency: + results = await asyncio.gather(*tasks) + self.results.extend(results) + tasks = [] + + # 显示进度 + print(f" 进度: {i}/{total_requests} ({i/total_requests*100:.1f}%)", end='\r') + + # 处理剩余任务 + if tasks: + results = await asyncio.gather(*tasks) + self.results.extend(results) + + total_time = time.time() - start_time + + # 统计和输出 + self.print_statistics(concurrency, total_requests, total_time) + + def print_statistics(self, concurrency: int, total_requests: int, total_time: float): + """打印统计信息""" + success_results = [r for r in self.results if r["status"] == "success"] + error_results = [r for r in self.results if r["status"] != "success"] + + success_count = len(success_results) + error_count = len(error_results) + + if success_results: + latencies = [r["elapsed"] for r in success_results] + avg_latency = statistics.mean(latencies) + min_latency = min(latencies) + max_latency = max(latencies) + p50_latency = statistics.median(latencies) + p95_latency = sorted(latencies)[int(len(latencies) * 0.95)] if len(latencies) > 1 else latencies[0] + p99_latency = sorted(latencies)[int(len(latencies) * 0.99)] if len(latencies) > 1 else latencies[0] + else: + avg_latency = min_latency = max_latency = p50_latency = p95_latency = p99_latency = 0 + + throughput = total_requests / total_time if total_time > 0 else 0 + + print(f"\n\n{'='*60}") + print(f"📈 测试结果统计") + print(f"{'='*60}") + print(f" 测试时间: {total_time:.2f}s") + print(f" 总请求数: {total_requests}") + print(f" 并发数: {concurrency}") + print(f"") + print(f" 成功请求: {success_count} ({success_count/total_requests*100:.1f}%)") + print(f" 失败请求: {error_count} ({error_count/total_requests*100:.1f}%)") + print(f"") + print(f" 吞吐量: {throughput:.2f} req/s") + print(f"") + print(f" 延迟统计:") + print(f" 最小: {min_latency*1000:.0f}ms") + print(f" 平均: {avg_latency*1000:.0f}ms") + print(f" 最大: {max_latency*1000:.0f}ms") + print(f" P50: {p50_latency*1000:.0f}ms") + print(f" P95: {p95_latency*1000:.0f}ms") + print(f" P99: {p99_latency*1000:.0f}ms") + + # 错误详情 + if error_results: + print(f"\n ⚠️ 错误详情:") + error_types = {} + for r in error_results: + error_type = r.get("status", "unknown") + error_types[error_type] = error_types.get(error_type, 0) + 1 + + for error_type, count in error_types.items(): + print(f" {error_type}: {count}") + + print(f"{'='*60}\n") + + # 性能评级 + self.print_performance_rating(throughput, avg_latency) + + def print_performance_rating(self, throughput: float, avg_latency: float): + """打印性能评级""" + print(f"🎯 性能评级:") + + # 吞吐量评级 + if throughput >= 100: + rating = "⭐⭐⭐⭐⭐ 优秀" + elif throughput >= 60: + rating = "⭐⭐⭐⭐ 良好" + elif throughput >= 30: + rating = "⭐⭐⭐ 中等" + elif throughput >= 10: + rating = "⭐⭐ 较低" + else: + rating = "⭐ 需优化" + + print(f" 吞吐量 ({throughput:.1f} req/s): {rating}") + + # 延迟评级 + if avg_latency < 0.5: + rating = "⭐⭐⭐⭐⭐ 优秀" + elif avg_latency < 1.0: + rating = "⭐⭐⭐⭐ 良好" + elif avg_latency < 2.0: + rating = "⭐⭐⭐ 中等" + elif avg_latency < 5.0: + rating = "⭐⭐ 较高" + else: + rating = "⭐ 需优化" + + print(f" 平均延迟 ({avg_latency*1000:.0f}ms): {rating}") + print() + + +async def main(): + """主函数""" + parser = argparse.ArgumentParser(description='Grok2API 并发性能测试') + parser.add_argument('--url', default='http://localhost:8001', help='API 基础URL') + parser.add_argument('--key', default='', help='API Key（可选）') + parser.add_argument('-c', '--concurrency', type=int, default=10, help='并发数') + parser.add_argument('-n', '--requests', type=int, default=50, help='总请求数') + parser.add_argument('--multi-test', action='store_true', help='运行多级并发测试') + + args = parser.parse_args() + + print(f""" +╔══════════════════════════════════════════════════════════╗ +║ Grok2API 并发性能测试工具 ║ +╚══════════════════════════════════════════════════════════╝ + +🔗 测试目标: {args.url} +🔑 API Key: {'已设置' if args.key else '未设置'} +⏰ 开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} +""") + + tester = ConcurrencyTester(args.url, args.key) + + if args.multi_test: + # 多级并发测试 + test_configs = [ + (5, 20), # 5并发，20请求 + (10, 50), # 10并发，50请求 + (20, 100), # 20并发，100请求 + (50, 200), # 50并发，200请求 + ] + + for concurrency, requests in test_configs: + tester.results = [] # 清空结果 + await tester.run_concurrent_test(concurrency, requests) + await asyncio.sleep(2) # 间隔2秒 + else: + # 单次测试 + await tester.run_concurrent_test(args.concurrency, args.requests) + + print(f"\n✅ 测试完成！") + print(f"⏰ 结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + print("\n\n⚠️ 测试被用户中断") + except Exception as e: + print(f"\n\n❌ 测试失败: {e}") diff --git a/test_concurrency.sh b/test_concurrency.sh new file mode 100755 index 0000000..b100e2c --- /dev/null +++ b/test_concurrency.sh @@ -0,0 +1,177 @@ +#!/bin/bash + +# Grok2API 并发测试脚本（Shell版本） +# 使用 curl 和 GNU parallel 进行并发测试 + +set -e + +# 配置 +BASE_URL="${BASE_URL:-http://localhost:8001}" +API_KEY="${API_KEY:-}" +CONCURRENCY="${CONCURRENCY:-10}" +TOTAL_REQUESTS="${TOTAL_REQUESTS:-50}" + +# 颜色 +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +BLUE='\033[0;34m' +NC='\033[0m' # No Color + +echo -e "${BLUE}╔══════════════════════════════════════════════════════════╗${NC}" +echo -e "${BLUE}║ Grok2API 并发性能测试工具 (Shell版) ║${NC}" +echo -e "${BLUE}╚══════════════════════════════════════════════════════════╝${NC}" +echo "" +echo -e "${GREEN}🔗 测试目标:${NC} $BASE_URL" +echo -e "${GREEN}🔑 API Key:${NC} ${API_KEY:-(未设置)}" +echo -e "${GREEN}📊 并发数:${NC} $CONCURRENCY" +echo -e "${GREEN}📈 总请求数:${NC} $TOTAL_REQUESTS" +echo "" + +# 检查依赖 +if ! command -v curl &> /dev/null; then + echo -e "${RED}❌ 错误: 需要安装 curl${NC}" + exit 1 +fi + +# 创建临时目录 +TMP_DIR=$(mktemp -d) +trap "rm -rf $TMP_DIR" EXIT + +# 单个请求函数 +test_request() { + local request_id=$1 + local start_time=$(date +%s.%N) + + # 构建请求 + local headers="Content-Type: application/json" + if [ -n "$API_KEY" ]; then + headers="${headers}\nAuthorization: Bearer ${API_KEY}" + fi + + local response=$(curl -s -w "\n%{http_code}\n%{time_total}" \ + -X POST "${BASE_URL}/v1/chat/completions" \ + -H "Content-Type: application/json" \ + ${API_KEY:+-H "Authorization: Bearer $API_KEY"} \ + -d "{ + \"model\": \"grok-3-fast\", + \"messages\": [{\"role\": \"user\", \"content\": \"测试请求 #${request_id}，请简短回复OK\"}], + \"stream\": false, + \"max_tokens\": 10 + }" 2>&1) + + local http_code=$(echo "$response" | tail -n 2 | head -n 1) + local time_total=$(echo "$response" | tail -n 1) + + # 记录结果 + echo "${request_id},${http_code},${time_total}" >> "$TMP_DIR/results.csv" + + # 显示进度 + echo -ne "\r 进度: ${request_id}/${TOTAL_REQUESTS}" +} + +# 导出函数供 parallel 使用 +export -f test_request +export BASE_URL API_KEY TMP_DIR + +# 清空结果文件 +echo "id,status,time" > "$TMP_DIR/results.csv" + +echo -e "${YELLOW}🚀 开始并发测试...${NC}" +START_TIME=$(date +%s.%N) + +# 使用 GNU parallel（如果可用），否则使用简单循环 +if command -v parallel &> /dev/null; then + seq 1 $TOTAL_REQUESTS | parallel -j $CONCURRENCY test_request {} +else + # 简单的后台任务并发 + for i in $(seq 1 $TOTAL_REQUESTS); do + test_request $i & + + # 控制并发数 + if (( i % CONCURRENCY == 0 )); then + wait + fi + done + wait +fi + +END_TIME=$(date +%s.%N) +TOTAL_TIME=$(echo "$END_TIME - $START_TIME" | bc) + +echo -e "\n" + +# 统计结果 +echo -e "${BLUE}═══════════════════════════════════════════════════════════${NC}" +echo -e "${BLUE}📈 测试结果统计${NC}" +echo -e "${BLUE}═══════════════════════════════════════════════════════════${NC}" + +# 统计成功/失败 +SUCCESS_COUNT=$(awk -F',' '$2 == 200 {count++} END {print count+0}' "$TMP_DIR/results.csv") +ERROR_COUNT=$((TOTAL_REQUESTS - SUCCESS_COUNT)) + +echo -e " 测试时间: ${TOTAL_TIME}s" +echo -e " 总请求数: ${TOTAL_REQUESTS}" +echo -e " 并发数: ${CONCURRENCY}" +echo "" +echo -e " 成功请求: ${GREEN}${SUCCESS_COUNT}${NC} ($(echo "scale=1; $SUCCESS_COUNT * 100 / $TOTAL_REQUESTS" | bc)%)" +echo -e " 失败请求: ${RED}${ERROR_COUNT}${NC} ($(echo "scale=1; $ERROR_COUNT * 100 / $TOTAL_REQUESTS" | bc)%)" +echo "" + +# 计算吞吐量 +THROUGHPUT=$(echo "scale=2; $TOTAL_REQUESTS / $TOTAL_TIME" | bc) +echo -e " 吞吐量: ${GREEN}${THROUGHPUT}${NC} req/s" +echo "" + +# 延迟统计（只统计成功的请求） +if [ $SUCCESS_COUNT -gt 0 ]; then + echo -e " 延迟统计:" + + # 提取成功请求的延迟时间 + awk -F',' '$2 == 200 {print $3}' "$TMP_DIR/results.csv" | sort -n > "$TMP_DIR/latencies.txt" + + MIN=$(head -n 1 "$TMP_DIR/latencies.txt" | awk '{printf "%.0f", $1*1000}') + MAX=$(tail -n 1 "$TMP_DIR/latencies.txt" | awk '{printf "%.0f", $1*1000}') + AVG=$(awk '{sum+=$1; count++} END {printf "%.0f", sum/count*1000}' "$TMP_DIR/latencies.txt") + + # P50 + P50_LINE=$((SUCCESS_COUNT / 2)) + P50=$(sed -n "${P50_LINE}p" "$TMP_DIR/latencies.txt" | awk '{printf "%.0f", $1*1000}') + + # P95 + P95_LINE=$(echo "scale=0; $SUCCESS_COUNT * 0.95 / 1" | bc) + P95=$(sed -n "${P95_LINE}p" "$TMP_DIR/latencies.txt" | awk '{printf "%.0f", $1*1000}') + + # P99 + P99_LINE=$(echo "scale=0; $SUCCESS_COUNT * 0.99 / 1" | bc) + P99=$(sed -n "${P99_LINE}p" "$TMP_DIR/latencies.txt" | awk '{printf "%.0f", $1*1000}') + + echo -e " 最小: ${MIN}ms" + echo -e " 平均: ${AVG}ms" + echo -e " 最大: ${MAX}ms" + echo -e " P50: ${P50}ms" + echo -e " P95: ${P95}ms" + echo -e " P99: ${P99}ms" +fi + +echo -e "${BLUE}═══════════════════════════════════════════════════════════${NC}" + +# 性能评级 +echo -e "${YELLOW}🎯 性能评级:${NC}" + +if (( $(echo "$THROUGHPUT >= 100" | bc -l) )); then + RATING="⭐⭐⭐⭐⭐ 优秀" +elif (( $(echo "$THROUGHPUT >= 60" | bc -l) )); then + RATING="⭐⭐⭐⭐ 良好" +elif (( $(echo "$THROUGHPUT >= 30" | bc -l) )); then + RATING="⭐⭐⭐ 中等" +elif (( $(echo "$THROUGHPUT >= 10" | bc -l) )); then + RATING="⭐⭐ 较低" +else + RATING="⭐ 需优化" +fi + +echo -e " 吞吐量 (${THROUGHPUT} req/s): ${RATING}" + +echo "" +echo -e "${GREEN}✅ 测试完成！${NC}" From 22d8fb9af9ff5987d2ba99b97e51e47b977d0e23 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Wed, 17 Dec 2025 00:29:51 +0800 Subject: [PATCH 41/43] feat: more retry & proxy pool --- app/core/config.py | 40 +- app/core/proxy_pool.py | 143 ++++ app/services/grok/cache.py | 103 ++- app/services/grok/client.py | 130 ++-- app/services/grok/create.py | 114 ++- app/services/grok/token.py | 117 ++- app/services/grok/upload.py | 114 ++- app/template/admin.html | 1466 ++++++++++++++++++++--------------- main.py | 7 + requirements.txt | 4 +- 10 files changed, 1454 insertions(+), 784 deletions(-) create mode 100644 app/core/proxy_pool.py 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 From a2d59860a914d84308d21fa530f1fe74e2ea19a5 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Wed, 17 Dec 2025 00:29:57 +0800 Subject: [PATCH 42/43] Update setting.toml --- data/setting.toml | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/data/setting.toml b/data/setting.toml index ba04374..6c9fc17 100644 --- a/data/setting.toml +++ b/data/setting.toml @@ -10,10 +10,14 @@ stream_total_timeout = 600 stream_first_response_timeout = 30 temporary = true show_thinking = true +dynamic_statsig = true +proxy_pool_url = "" +proxy_pool_interval = 300 +retry_status_codes = [ 401, 429,] [global] base_url = "" -log_level = "DEBUG" +log_level = "INFO" image_mode = "url" admin_password = "admin" admin_username = "admin" From 7ffb1859514279754b3dff346a69a3b5ea091f62 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Wed, 17 Dec 2025 00:43:44 +0800 Subject: [PATCH 43/43] fix: retry error --- app/services/grok/cache.py | 9 +++++---- app/services/grok/client.py | 13 ++++++------- app/services/grok/create.py | 17 +++++++++-------- app/services/grok/token.py | 17 +++++++++-------- app/services/grok/upload.py | 17 +++++++++-------- 5 files changed, 38 insertions(+), 35 deletions(-) diff --git a/app/services/grok/cache.py b/app/services/grok/cache.py index 51f6a58..95e86b1 100644 --- a/app/services/grok/cache.py +++ b/app/services/grok/cache.py @@ -64,7 +64,7 @@ class CacheService: retry_codes = setting.grok_config.get("retry_status_codes", [401, 429]) MAX_OUTER_RETRY = 3 - for outer_retry in range(MAX_OUTER_RETRY): + for outer_retry in range(MAX_OUTER_RETRY + 1): # +1 确保实际重试3次 try: # 内层重试：403代理池重试（cache使用缓存代理，不支持代理池，所以403只重试一次） max_403_retries = 5 @@ -105,9 +105,10 @@ class CacheService: # 检查可配置状态码错误 - 外层重试 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) + if outer_retry < MAX_OUTER_RETRY: + delay = (outer_retry + 1) * 0.1 # 渐进延迟：0.1s, 0.2s, 0.3s + self._log("warning", f"遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})，等待{delay}s...") + await asyncio.sleep(delay) break # 跳出内层循环，进入外层重试 else: self._log("error", f"{response.status_code}错误，已重试{outer_retry}次，放弃") diff --git a/app/services/grok/client.py b/app/services/grok/client.py index 888bbea..0403de7 100644 --- a/app/services/grok/client.py +++ b/app/services/grok/client.py @@ -234,20 +234,19 @@ class GrokClient: proxies=proxies ) - # 检查是否是403错误 - if response.status_code == 403: + # 内层403重试：仅当有代理池时触发 + if response.status_code == 403 and proxy_pool._enabled: retry_count += 1 - # 如果使用代理池且未达到最大重试次数，继续重试 - if proxy_pool._enabled and retry_count <= max_retries: + if retry_count <= max_retries: logger.warning(f"[Client] 遇到403错误，正在重试 ({retry_count}/{max_retries})...") - await asyncio.sleep(0.5) # 短暂延迟 + 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) diff --git a/app/services/grok/create.py b/app/services/grok/create.py index 2210165..3652bc3 100644 --- a/app/services/grok/create.py +++ b/app/services/grok/create.py @@ -55,7 +55,7 @@ class PostCreateManager: retry_codes = setting.grok_config.get("retry_status_codes", [401, 429]) MAX_OUTER_RETRY = 3 - for outer_retry in range(MAX_OUTER_RETRY): + for outer_retry in range(MAX_OUTER_RETRY + 1): # +1 确保实际重试3次 # 内层重试：403代理池重试 max_403_retries = 5 retry_403_count = 0 @@ -84,23 +84,24 @@ class PostCreateManager: proxies=proxies ) - # 检查403错误 - 内层重试 - if response.status_code == 403: + # 内层403重试：仅当有代理池时触发 + if response.status_code == 403 and proxy_pool._enabled: retry_403_count += 1 - if proxy_pool._enabled and retry_403_count <= max_403_retries: + if 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) + if outer_retry < MAX_OUTER_RETRY: + delay = (outer_retry + 1) * 0.1 # 渐进延迟：0.1s, 0.2s, 0.3s + logger.warning(f"[PostCreate] 遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})，等待{delay}s...") + await asyncio.sleep(delay) break # 跳出内层循环，进入外层重试 else: logger.error(f"[PostCreate] {response.status_code}错误，已重试{outer_retry}次，放弃") diff --git a/app/services/grok/token.py b/app/services/grok/token.py index 7911e34..80f2e98 100644 --- a/app/services/grok/token.py +++ b/app/services/grok/token.py @@ -323,7 +323,7 @@ class GrokTokenManager: retry_codes = setting.grok_config.get("retry_status_codes", [401, 429]) MAX_OUTER_RETRY = 3 - for outer_retry in range(MAX_OUTER_RETRY): + for outer_retry in range(MAX_OUTER_RETRY + 1): # +1 确保实际重试3次 # 内层重试：403代理池重试 max_403_retries = 5 retry_403_count = 0 @@ -351,26 +351,27 @@ class GrokTokenManager: proxies=proxies ) - # 检查403错误 - 内层重试 - if response.status_code == 403: + # 内层403重试：仅当有代理池时触发 + if response.status_code == 403 and proxy_pool._enabled: retry_403_count += 1 - if proxy_pool._enabled and retry_403_count <= max_403_retries: + if 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) + if outer_retry < MAX_OUTER_RETRY: + delay = (outer_retry + 1) * 0.1 # 渐进延迟：0.1s, 0.2s, 0.3s + logger.warning(f"[Token] 遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})，等待{delay}s...") + await asyncio.sleep(delay) break # 跳出内层循环，进入外层重试 else: logger.error(f"[Token] {response.status_code}错误，已重试{outer_retry}次，放弃") diff --git a/app/services/grok/upload.py b/app/services/grok/upload.py index 4252b35..4d0c8b4 100644 --- a/app/services/grok/upload.py +++ b/app/services/grok/upload.py @@ -61,7 +61,7 @@ class ImageUploadManager: retry_codes = setting.grok_config.get("retry_status_codes", [401, 429]) MAX_OUTER_RETRY = 3 - for outer_retry in range(MAX_OUTER_RETRY): + for outer_retry in range(MAX_OUTER_RETRY + 1): # +1 确保实际重试3次 try: # 内层重试：403代理池重试 max_403_retries = 5 @@ -98,23 +98,24 @@ class ImageUploadManager: proxies=proxies, ) - # 检查403错误 - 内层重试 - if response.status_code == 403: + # 内层403重试：仅当有代理池时触发 + if response.status_code == 403 and proxy_pool._enabled: retry_403_count += 1 - if proxy_pool._enabled and retry_403_count <= max_403_retries: + if 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) + if outer_retry < MAX_OUTER_RETRY: + delay = (outer_retry + 1) * 0.1 # 渐进延迟：0.1s, 0.2s, 0.3s + logger.warning(f"[Upload] 遇到{response.status_code}错误，外层重试 ({outer_retry+1}/{MAX_OUTER_RETRY})，等待{delay}s...") + await asyncio.sleep(delay) break # 跳出内层循环，进入外层重试 else: logger.error(f"[Upload] {response.status_code}错误，已重试{outer_retry}次，放弃")