feat: optimizing concurrent performance

app/api/admin/manage.py

@@ -21,8 +21,12 @@ from app.models.grok_models import TokenType
 # 创建路由器
 router = APIRouter(tags=["管理"])
 
-# 静态文件路径
+# 常量定义
 STATIC_DIR = Path(__file__).parents[2] / "template"
+TEMP_DIR = Path(__file__).parents[3] / "data" / "temp"
+SESSION_EXPIRE_HOURS = 24
+BYTES_PER_KB = 1024
+BYTES_PER_MB = 1024 * 1024
 
 # 简单的会话存储
 _sessions: Dict[str, datetime] = {}
@@ -225,8 +229,8 @@ async def admin_login(request: LoginRequest) -> LoginResponse:
         # 生成会话token
         session_token = secrets.token_urlsafe(32)
 
-        # 设置会话过期时间（24小时）
-        expire_time = datetime.now() + timedelta(hours=24)
+        # 设置会话过期时间
+        expire_time = datetime.now() + timedelta(hours=SESSION_EXPIRE_HOURS)
         _sessions[session_token] = expire_time
 
         logger.debug(f"[Admin] 管理员登录成功 - 用户名: {request.username}")
@@ -344,7 +348,7 @@ async def add_tokens(request: AddTokensRequest,
         token_type = validate_token_type(request.token_type)
 
         # 添加Token
-        token_manager.add_token(request.tokens, token_type)
+        await token_manager.add_token(request.tokens, token_type)
 
         logger.debug(f"[Admin] Token添加成功 - 类型: {request.token_type}, 数量: {len(request.tokens)}")
 
@@ -379,7 +383,7 @@ async def delete_tokens(request: DeleteTokensRequest,
         token_type = validate_token_type(request.token_type)
 
         # 删除Token
-        token_manager.delete_token(request.tokens, token_type)
+        await token_manager.delete_token(request.tokens, token_type)
 
         logger.debug(f"[Admin] Token删除成功 - 类型: {request.token_type}, 数量: {len(request.tokens)}")
 
@@ -427,11 +431,13 @@ async def update_settings(request: UpdateSettingsRequest, authenticated: bool =
     """更新全局配置"""
     try:
         import toml
+        import aiofiles
         logger.debug("[Admin] 更新全局配置")
 
-        # 读取现有配置
-        with open(setting.config_path, "r", encoding="utf-8") as f:
-            config = toml.load(f)
+        # 异步读取现有配置
+        async with aiofiles.open(setting.config_path, "r", encoding="utf-8") as f:
+            content = await f.read()
+            config = toml.loads(content)
 
         # 更新配置
         if request.global_config:
@@ -439,9 +445,9 @@ async def update_settings(request: UpdateSettingsRequest, authenticated: bool =
         if request.grok_config:
             config["grok"].update(request.grok_config)
 
-        # 写回配置文件
-        with open(setting.config_path, "w", encoding="utf-8") as f:
-            toml.dump(config, f)
+        # 异步写回配置文件
+        async with aiofiles.open(setting.config_path, "w", encoding="utf-8") as f:
+            await f.write(toml.dumps(config))
 
         # 重新加载配置
         setting.global_config = setting.load("global")
@@ -454,48 +460,43 @@ async def update_settings(request: UpdateSettingsRequest, authenticated: bool =
         raise HTTPException(status_code=500, detail={"error": f"更新配置失败: {str(e)}", "code": "UPDATE_SETTINGS_ERROR"})
 
 
+def _calculate_dir_size(directory: Path) -> int:
+    """计算目录中所有文件的大小（字节）"""
+    total_size = 0
+    for file_path in directory.iterdir():
+        if file_path.is_file():
+            try:
+                total_size += file_path.stat().st_size
+            except Exception as e:
+                logger.warning(f"[Admin] 无法获取文件大小: {file_path.name}, 错误: {str(e)}")
+    return total_size
+
+
+def _format_size(size_bytes: int) -> str:
+    """格式化字节大小为可读字符串"""
+    size_mb = size_bytes / BYTES_PER_MB
+    if size_mb < 1:
+        size_kb = size_bytes / BYTES_PER_KB
+        return f"{size_kb:.1f} KB"
+    return f"{size_mb:.1f} MB"
+
+
 @router.get("/api/cache/size")
 async def get_cache_size(authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
-    """
-    获取缓存大小
-
-    计算/data/temp目录中所有文件的大小。
-    """
+    """获取缓存大小"""
     try:
         logger.debug("[Admin] 开始获取缓存大小")
 
-        # 获取temp目录路径
-        temp_dir = Path(__file__).parents[3] / "data" / "temp"
-
-        if not temp_dir.exists():
-            logger.warning(f"[Admin] 缓存目录不存在: {temp_dir}")
-            return {
-                "success": True,
-                "data": {"size": "0 MB"}
-            }
+        if not TEMP_DIR.exists():
+            logger.warning(f"[Admin] 缓存目录不存在: {TEMP_DIR}")
+            return {"success": True, "data": {"size": "0 MB"}}
 
         # 计算目录大小
-        total_size = 0
-        for file_path in temp_dir.iterdir():
-            if file_path.is_file():
-                try:
-                    total_size += file_path.stat().st_size
-                except Exception as e:
-                    logger.error(f"[Admin] 获取文件大小失败: {file_path.name}, 错误: {str(e)}")
-
-        # 转换为MB并格式化
-        size_mb = total_size / (1024 * 1024)
-        if size_mb < 1:
-            size_kb = total_size / 1024
-            size_str = f"{size_kb:.1f} KB"
-        else:
-            size_str = f"{size_mb:.1f} MB"
+        total_size = _calculate_dir_size(TEMP_DIR)
+        size_str = _format_size(total_size)
 
         logger.debug(f"[Admin] 缓存大小获取完成 - 大小: {size_str}")
-        return {
-            "success": True,
-            "data": {"size": size_str}
-        }
+        return {"success": True, "data": {"size": size_str}}
 
     except Exception as e:
         logger.error(f"[Admin] 获取缓存大小异常 - 错误: {str(e)}")
@@ -507,28 +508,21 @@ async def get_cache_size(authenticated: bool = Depends(verify_admin_session)) ->
 
 @router.post("/api/cache/clear")
 async def clear_cache(authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
-    """
-    清理缓存
-
-    删除/data/temp目录中的所有文件。
-    """
+    """清理缓存 - 删除所有临时文件"""
     try:
         logger.debug("[Admin] 开始清理缓存")
 
-        # 获取temp目录路径
-        temp_dir = Path(__file__).parents[3] / "data" / "temp"
-
-        if not temp_dir.exists():
-            logger.warning(f"[Admin] 缓存目录不存在: {temp_dir}")
+        if not TEMP_DIR.exists():
+            logger.warning(f"[Admin] 缓存目录不存在: {TEMP_DIR}")
             return {
                 "success": True,
                 "message": "缓存目录不存在，无需清理",
                 "data": {"deleted_count": 0}
             }
 
-        # 获取所有文件
+        # 删除所有文件
         deleted_count = 0
-        for file_path in temp_dir.iterdir():
+        for file_path in TEMP_DIR.iterdir():
             if file_path.is_file():
                 try:
                     file_path.unlink()

app/services/grok/client.py

@@ -2,7 +2,6 @@
 
 import asyncio
 import json
-import threading
 from typing import Dict, List, Tuple, Any
 
 from curl_cffi import requests as curl_requests
@@ -42,7 +41,7 @@ class GrokClient:
         model_name, model_mode = Models.to_grok(model)
 
         # 上传图片并获取附件ID列表
-        image_attachments = GrokClient._upload_images(image_urls, auth_token)
+        image_attachments = await GrokClient._upload_images(image_urls, auth_token)
 
         # 构建Grok请求载荷
         payload = GrokClient._build_payload(content, model_name, model_mode, image_attachments)
@@ -75,16 +74,18 @@ class GrokClient:
         return "".join(content_parts), image_urls
 
     @staticmethod
-    def _upload_images(image_urls: List[str], auth_token: str) -> List[str]:
+    async def _upload_images(image_urls: List[str], auth_token: str) -> List[str]:
         """上传图片并返回附件ID列表"""
         image_attachments = []
-        for url in image_urls:
-            try:
-                image_id = ImageUploadManager.upload(url, auth_token)
-                if image_id:
-                    image_attachments.append(image_id)
-            except Exception as e:
-                logger.warning(f"[Client] 图片上传失败: {url}, 错误: {e}")
+        # 并发上传所有图片
+        tasks = [ImageUploadManager.upload(url, auth_token) for url in image_urls]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        for url, result in zip(image_urls, results):
+            if isinstance(result, Exception):
+                logger.warning(f"[Client] 图片上传失败: {url}, 错误: {result}")
+            elif result:
+                image_attachments.append(result)
 
         return image_attachments
 
@@ -150,10 +151,10 @@ class GrokClient:
                 GrokClient._handle_error_response(response, auth_token)
 
             # 请求成功，重置失败计数
-            token_manager.reset_token_failure(auth_token)
+            asyncio.create_task(token_manager.reset_token_failure(auth_token))
 
             # 处理并返回响应
-            return GrokClient._process_response(response, auth_token, model, stream)
+            return await GrokClient._process_response(response, auth_token, model, stream)
 
         except curl_requests.RequestsError as e:
             raise GrokApiException(f"网络错误: {e}", "NETWORK_ERROR") from e
@@ -190,7 +191,7 @@ class GrokClient:
             error_message = error_data[:200] if error_data else "未知错误"
 
         # 记录Token失败
-        token_manager.record_token_failure(auth_token, response.status_code, error_message)
+        asyncio.create_task(token_manager.record_token_failure(auth_token, response.status_code, error_message))
 
         raise GrokApiException(
             f"请求失败: {response.status_code} - {error_message}",
@@ -199,26 +200,23 @@ class GrokClient:
         )
 
     @staticmethod
-    def _process_response(response, auth_token: str, model: str, stream: bool):
+    async def _process_response(response, auth_token: str, model: str, stream: bool):
         """处理API响应"""
         if stream:
-            # 流式响应：异步更新速率限制
+            # 流式响应：返回异步生成器，在后台更新速率限制
             result = GrokResponseProcessor.process_stream(response, auth_token)
-            threading.Thread(
-                target=lambda: GrokClient._update_rate_limits(auth_token, model),
-                daemon=True
-            ).start()
+            asyncio.create_task(GrokClient._update_rate_limits_async(auth_token, model))
         else:
-            # 非流式响应：同步更新速率限制
-            result = GrokResponseProcessor.process_response(response, auth_token)
-            GrokClient._update_rate_limits(auth_token, model)
+            # 非流式响应：等待处理完成后更新速率限制
+            result = await GrokResponseProcessor.process_response(response, auth_token)
+            asyncio.create_task(GrokClient._update_rate_limits_async(auth_token, model))
 
         return result
 
     @staticmethod
-    def _update_rate_limits(auth_token: str, model: str):
-        """更新速率限制信息"""
+    async def _update_rate_limits_async(auth_token: str, model: str):
+        """异步更新速率限制信息"""
         try:
-            token_manager.check_limits(auth_token, model)
+            await token_manager.check_limits(auth_token, model)
         except Exception as e:
             logger.error(f"[Client] 更新速率限制失败: {e}")

app/services/grok/processer.py

@@ -3,7 +3,6 @@
 import json
 import uuid
 import time
-import asyncio
 from typing import Iterator
 from app.core.config import setting
 from app.core.exception import GrokApiException
@@ -19,26 +18,11 @@ from app.models.openai_schema import (
 from app.services.grok.image_cache import image_cache_service
 
 
-def _safe_run_async(coro):
-    """安全地运行异步协程，无论是否有事件循环"""
-    try:
-        loop = asyncio.get_event_loop()
-        if loop.is_running():
-            # 如果循环正在运行，使用 ensure_future 调度任务
-            return asyncio.ensure_future(coro)
-        else:
-            # 循环存在但未运行，直接运行
-            return loop.run_until_complete(coro)
-    except RuntimeError:
-        # 没有事件循环，创建新的
-        return asyncio.run(coro)
-
-
 class GrokResponseProcessor:
     """Grok API 响应处理器"""
 
     @staticmethod
-    def process_response(response, auth_token: str) -> OpenAIChatCompletionResponse:
+    async def process_response(response, auth_token: str) -> OpenAIChatCompletionResponse:
         """处理非流式响应"""
         try:
             for chunk in response.iter_lines():
@@ -76,7 +60,7 @@ class GrokResponseProcessor:
                     for img in images:
                         # 下载并缓存图片
                         try:
-                            cache_path = _safe_run_async(image_cache_service.download_image(f"/{img}", auth_token))
+                            cache_path = await image_cache_service.download_image(f"/{img}", auth_token)
                             if cache_path:
                                 # 使用本地缓存路径
                                 img_path = img.replace('/', '-')
@@ -113,7 +97,7 @@ class GrokResponseProcessor:
             raise GrokApiException(f"JSON解析失败: {e}", "JSON_ERROR") from e
 
     @staticmethod
-    def process_stream(response, auth_token: str) -> Iterator[str]:
+    async def process_stream(response, auth_token: str) -> Iterator[str]:
         """处理流式响应"""
         is_image = False
         is_thinking = False
@@ -182,8 +166,8 @@ class GrokResponseProcessor:
                             content = ""
                             for img in model_resp.get("generatedImageUrls", []):
                                 try:
-                                    # 异步下载并缓存图片（不阻塞）
-                                    _safe_run_async(image_cache_service.download_image(f"/{img}", auth_token))
+                                    # 异步下载并缓存图片
+                                    await image_cache_service.download_image(f"/{img}", auth_token)
                                     # 使用本地缓存路径
                                     img_path = img.replace('/', '-')
                                     base_url = setting.global_config.get("base_url", "")

app/services/grok/token.py

@@ -2,11 +2,12 @@
 
 import json
 import time
-import threading
+import asyncio
 from pathlib import Path
 from typing import Dict, Any, Optional, Tuple
 
-from curl_cffi import requests as curl_requests
+import aiofiles
+from curl_cffi.requests import AsyncSession
 
 from app.models.grok_models import TokenType, Models
 from app.core.exception import GrokApiException
@@ -35,14 +36,12 @@ class GrokTokenManager:
     """
     
     _instance: Optional['GrokTokenManager'] = None
-    _lock = threading.Lock()
+    _lock = asyncio.Lock()
 
     def __new__(cls) -> 'GrokTokenManager':
         """单例模式实现"""
         if cls._instance is None:
-            with cls._lock:
-                if cls._instance is None:
-                    cls._instance = super().__new__(cls)
+            cls._instance = super().__new__(cls)
         return cls._instance
 
     def __init__(self):
@@ -51,16 +50,17 @@ class GrokTokenManager:
             return
 
         self.token_file = Path(__file__).parents[3] / "data" / "token.json"
-        self._file_lock = threading.Lock()
+        self._file_lock = asyncio.Lock()
         self.token_file.parent.mkdir(parents=True, exist_ok=True)
 
-        self._load_token_data()
+        # 同步加载初始数据
+        self._load_token_data_sync()
         self._initialized = True
 
         logger.debug(f"[Token] 管理器初始化完成，文件: {self.token_file}")
 
-    def _load_token_data(self) -> None:
-        """加载Token数据"""
+    def _load_token_data_sync(self) -> None:
+        """同步加载Token数据（仅用于初始化）"""
         default_data = {
             TokenType.NORMAL.value: {},
             TokenType.SUPER.value: {}
@@ -72,18 +72,17 @@ class GrokTokenManager:
                     self.token_data = json.load(f)
             else:
                 self.token_data = default_data
-                self._save_token_data()
                 logger.debug("[Token] 创建新的Token数据文件")
         except (json.JSONDecodeError, IOError) as e:
             logger.error(f"[Token] 加载Token数据失败: {str(e)}")
             self.token_data = default_data
 
-    def _save_token_data(self) -> None:
-        """保存Token数据到文件"""
+    async def _save_token_data(self) -> None:
+        """异步保存Token数据到文件"""
         try:
-            with self._file_lock:
-                with open(self.token_file, "w", encoding="utf-8") as f:
-                    json.dump(self.token_data, f, indent=2, ensure_ascii=False)
+            async with self._file_lock:
+                async with aiofiles.open(self.token_file, "w", encoding="utf-8") as f:
+                    await f.write(json.dumps(self.token_data, indent=2, ensure_ascii=False))
         except IOError as e:
             logger.error(f"[Token] 保存Token数据失败: {str(e)}")
             raise GrokApiException(
@@ -115,10 +114,10 @@ class GrokTokenManager:
             return {"http": proxy_url, "https": proxy_url}
         return {}
 
-    def add_token(self, tokens: list[str], token_type: TokenType) -> None:
+    async def add_token(self, tokens: list[str], token_type: TokenType) -> None:
         """
         添加Token到管理器
-        
+
         Args:
             tokens: Token列表
             token_type: Token类型
@@ -132,7 +131,7 @@ class GrokTokenManager:
             if not token or not token.strip():
                 logger.debug("[Token] 跳过空的Token")
                 continue
-                
+
             self.token_data[token_type.value][token] = {
                 "createdTime": int(time.time() * 1000),
                 "remainingQueries": -1,
@@ -143,14 +142,14 @@ class GrokTokenManager:
                 "lastFailureReason": None
             }
             added_count += 1
-            
-        self._save_token_data()
+
+        await self._save_token_data()
         logger.info(f"[Token] 成功添加 {added_count} 个 {token_type.value} Token")
-    
-    def delete_token(self, tokens: list[str], token_type: TokenType) -> None:
+
+    async def delete_token(self, tokens: list[str], token_type: TokenType) -> None:
         """
         删除指定的Token
-        
+
         Args:
             tokens: 要删除的Token列表
             token_type: Token类型
@@ -167,7 +166,7 @@ class GrokTokenManager:
             else:
                 logger.debug(f"[Token] Token不存在: {token[:10]}...")
 
-        self._save_token_data()
+        await self._save_token_data()
         logger.info(f"[Token] 成功删除 {deleted_count} 个 {token_type.value} Token")
     
     def get_all_token(self) -> Dict[str, Any]:
@@ -263,7 +262,7 @@ class GrokTokenManager:
         logger.debug(f"[Token] 为模型 {model} 选择Token ({status_text})")
         return max_token_key
     
-    def check_limits(self, auth_token: str, model: str) -> Optional[Dict[str, Any]]:
+    async def check_limits(self, auth_token: str, model: str) -> Optional[Dict[str, Any]]:
         """检查并更新模型速率限制"""
         try:
             rate_limit_model_name = Models.to_rate_limit(model)
@@ -277,24 +276,29 @@ class GrokTokenManager:
             headers = get_dynamic_headers("/rest/rate-limits")
             headers["Cookie"] = cookie
 
-            # 发送请求
-            response = curl_requests.post(
-                RATE_LIMIT_ENDPOINT,
-                headers=headers,
-                json=payload,
-                impersonate=IMPERSONATE_BROWSER,
-                timeout=REQUEST_TIMEOUT,
-                **self._get_proxy_config()
-            )
+            # 获取代理配置
+            proxy_url = setting.grok_config.get("proxy_url", "")
+            proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
 
-            if response.status_code == 200:
-                rate_limit_data = response.json()
-                logger.debug(f"[Token] 成功获取速率限制信息")
-                self._save_to_storage(auth_token, model, rate_limit_data)
-                return rate_limit_data
-            else:
-                logger.warning(f"[Token] 获取速率限制失败，状态码: {response.status_code}")
-                return None
+            # 发送异步请求
+            async with AsyncSession() as session:
+                response = await session.post(
+                    RATE_LIMIT_ENDPOINT,
+                    headers=headers,
+                    json=payload,
+                    impersonate=IMPERSONATE_BROWSER,
+                    timeout=REQUEST_TIMEOUT,
+                    proxies=proxies
+                )
+
+                if response.status_code == 200:
+                    rate_limit_data = response.json()
+                    logger.debug(f"[Token] 成功获取速率限制信息")
+                    await self._save_to_storage(auth_token, model, rate_limit_data)
+                    return rate_limit_data
+                else:
+                    logger.warning(f"[Token] 获取速率限制失败，状态码: {response.status_code}")
+                    return None
 
         except Exception as e:
             logger.error(f"[Token] 检查速率限制时发生错误: {str(e)}")
@@ -312,7 +316,7 @@ class GrokTokenManager:
             logger.error(f"[Token] 解析剩余Token数量时发生错误: {str(e)}")
             return 0
     
-    def _save_to_storage(self, auth_token: str, model: str, data: Dict[str, Any]) -> None:
+    async def _save_to_storage(self, auth_token: str, model: str, data: Dict[str, Any]) -> None:
         """存储速率限制信息到token.json"""
         try:
             sso_value = self._extract_sso_value(auth_token)
@@ -327,13 +331,13 @@ class GrokTokenManager:
                 normal_remaining = data.get("remainingTokens", -1)
                 heavy_remaining = -1
 
-            self.update_limits(sso_value, normal_remaining, heavy_remaining)
+            await self.update_limits(sso_value, normal_remaining, heavy_remaining)
             logger.info(f"[Token] 已更新限制: sso={sso_value[:10]}..., 通用={normal_remaining}, heavy={heavy_remaining}")
 
         except Exception as e:
             logger.error(f"[Token] 存储速率限制信息时发生错误: {str(e)}")
-    
-    def update_limits(self, sso_value: str, normal: int = None, heavy: int = None) -> None:
+
+    async def update_limits(self, sso_value: str, normal: int = None, heavy: int = None) -> None:
         """更新Token限制信息"""
         try:
             for token_type in [TokenType.NORMAL.value, TokenType.SUPER.value]:
@@ -342,8 +346,8 @@ class GrokTokenManager:
                         self.token_data[token_type][sso_value]["remainingQueries"] = normal
                     if heavy is not None:
                         self.token_data[token_type][sso_value]["heavyremainingQueries"] = heavy
-                    
-                    self._save_token_data()
+
+                    await self._save_token_data()
                     logger.info(f"[Token] 已更新Token {sso_value[:10]}... 的限制信息")
                     return
 
@@ -352,7 +356,7 @@ class GrokTokenManager:
         except Exception as e:
             logger.error(f"[Token] 更新Token限制时发生错误: {str(e)}")
     
-    def record_token_failure(self, auth_token: str, status_code: int, error_message: str) -> None:
+    async def record_token_failure(self, auth_token: str, status_code: int, error_message: str) -> None:
         """记录Token失败信息
 
         错误码说明：
@@ -398,12 +402,12 @@ class GrokTokenManager:
                     f"(连续401错误{token_data['failedCount']}次)"
                 )
 
-            self._save_token_data()
+            await self._save_token_data()
 
         except Exception as e:
             logger.error(f"[Token] 记录Token失败信息时发生错误: {str(e)}")
-    
-    def reset_token_failure(self, auth_token: str) -> None:
+
+    async def reset_token_failure(self, auth_token: str) -> None:
         """重置Token失败计数
 
         当Token成功完成请求时调用此方法，用于清除失败记录。
@@ -427,7 +431,7 @@ class GrokTokenManager:
                 token_data["lastFailureTime"] = None
                 token_data["lastFailureReason"] = None
 
-                self._save_token_data()
+                await self._save_token_data()
                 logger.info(f"[Token] Token {sso_value[:10]}... 失败计数已重置")
 
         except Exception as e:

app/services/grok/upload.py

@@ -4,7 +4,7 @@ import base64
 import re
 from urllib.parse import urlparse
 
-from curl_cffi import requests as curl_requests
+from curl_cffi.requests import AsyncSession
 
 from app.services.grok.statsig import get_dynamic_headers
 from app.core.exception import GrokApiException
@@ -23,12 +23,12 @@ class ImageUploadManager:
     """
 
     @staticmethod
-    def upload(image_input: str, auth_token: str) -> str:
+    async def upload(image_input: str, auth_token: str) -> str:
         """上传图片到Grok，支持Base64或URL"""
         try:
             if ImageUploadManager._is_url(image_input):
                 # 下载 URL 图片
-                image_buffer, mime_type = ImageUploadManager._download(image_input)
+                image_buffer, mime_type = await ImageUploadManager._download(image_input)
 
                 # 获取图片信息
                 file_name, _ = ImageUploadManager._get_info("", mime_type)
@@ -53,27 +53,29 @@ class ImageUploadManager:
 
             cf_clearance = setting.grok_config.get("cf_clearance", "")
             cookie = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token
-            proxy_config = {"http": setting.grok_config.get("proxy_url", ""), "https": setting.grok_config.get("proxy_url", "")} if setting.grok_config.get("proxy_url") else {}
+            proxy_url = setting.grok_config.get("proxy_url", "")
+            proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
 
-            # 发送请求
-            response = curl_requests.post(
-                "https://grok.com/rest/app-chat/upload-file",
-                headers={
-                    **get_dynamic_headers("/rest/app-chat/upload-file"),
-                    "Cookie": cookie,
-                },
-                json=upload_data,
-                impersonate="chrome133a",
-                timeout=30,
-                **proxy_config,
-            )
+            # 发送异步请求
+            async with AsyncSession() as session:
+                response = await session.post(
+                    "https://grok.com/rest/app-chat/upload-file",
+                    headers={
+                        **get_dynamic_headers("/rest/app-chat/upload-file"),
+                        "Cookie": cookie,
+                    },
+                    json=upload_data,
+                    impersonate="chrome133a",
+                    timeout=30,
+                    proxies=proxies,
+                )
 
-            # 检查响应
-            if response.status_code == 200:
-                result = response.json()
-                file_id = result.get("fileMetadataId", "")
-                logger.debug(f"[Upload] 图片上传成功，文件ID: {file_id}")
-                return file_id
+                # 检查响应
+                if response.status_code == 200:
+                    result = response.json()
+                    file_id = result.get("fileMetadataId", "")
+                    logger.debug(f"[Upload] 图片上传成功，文件ID: {file_id}")
+                    return file_id
 
             return ""
 
@@ -92,20 +94,20 @@ class ImageUploadManager:
             return False
 
     @staticmethod
-    def _download(url: str) -> tuple[str, str]:
-
+    async def _download(url: str) -> tuple[str, str]:
         try:
-            response = curl_requests.get(url, timeout=5)
-            response.raise_for_status()
+            async with AsyncSession() as session:
+                response = await session.get(url, timeout=5)
+                response.raise_for_status()
 
-            # 获取内容类型
-            content_type = response.headers.get('content-type', 'image/jpeg')
-            if not content_type.startswith('image/'):
-                content_type = 'image/jpeg'
+                # 获取内容类型
+                content_type = response.headers.get('content-type', 'image/jpeg')
+                if not content_type.startswith('image/'):
+                    content_type = 'image/jpeg'
 
-            # 转换为 Base64
-            image_base64 = base64.b64encode(response.content).decode('utf-8')
-            return image_base64, content_type
+                # 转换为 Base64
+                image_base64 = base64.b64encode(response.content).decode('utf-8')
+                return image_base64, content_type
         except Exception as e:
             logger.warning(f"[Upload] 下载图片失败: {e}")
             return "", ""

requirements.txt

@@ -5,4 +5,5 @@ python-dotenv==1.1.1
 curl_cffi==0.13.0
 requests==2.32.5
 starlette==0.48.0
-pydantic==2.12.0
+pydantic==2.12.0
+aiofiles==24.1.0