Upload 11 files

Dockerfile

@@ -8,6 +8,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     && apt-get autoremove -y \
     && rm -rf /var/lib/apt/lists/*
 COPY main.py .
+# 复制 uptime_tracker 模块
+COPY uptime_tracker.py .
 # 复制 core 模块
 COPY core ./core
 # 复制 util 目录

main.py

@@ -6,6 +6,7 @@ import logging
 from dotenv import load_dotenv
 
 import httpx
+import aiofiles
 from fastapi import FastAPI, HTTPException, Header, Request, Body
 from fastapi.responses import StreamingResponse, HTMLResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
@@ -18,6 +19,9 @@ from functools import wraps
 # 导入认证装饰器
 from core.auth import require_path_prefix, require_admin_auth, require_path_and_admin
 
+# 导入 Uptime 追踪器
+import uptime_tracker
+
 # ---------- 日志配置 ----------
 
 # 内存日志缓冲区 (保留最近 3000 条日志，重启后清空)
@@ -26,14 +30,15 @@ log_lock = Lock()
 
 # 统计数据持久化
 STATS_FILE = "stats.json"
-stats_lock = Lock()
+stats_lock = asyncio.Lock()  # 改为异步锁
 
-def load_stats():
-    """加载统计数据"""
+async def load_stats():
+    """加载统计数据（异步）"""
     try:
         if os.path.exists(STATS_FILE):
-            with open(STATS_FILE, 'r', encoding='utf-8') as f:
-                return json.load(f)
+            async with aiofiles.open(STATS_FILE, 'r', encoding='utf-8') as f:
+                content = await f.read()
+                return json.loads(content)
     except Exception:
         pass
     return {
@@ -44,16 +49,22 @@ def load_stats():
         "account_conversations": {}  # {account_id: conversation_count} 账户对话次数
     }
 
-def save_stats(stats):
-    """保存统计数据"""
+async def save_stats(stats):
+    """保存统计数据（异步，避免阻塞事件循环）"""
     try:
-        with open(STATS_FILE, 'w', encoding='utf-8') as f:
-            json.dump(stats, f, ensure_ascii=False, indent=2)
+        async with aiofiles.open(STATS_FILE, 'w', encoding='utf-8') as f:
+            await f.write(json.dumps(stats, ensure_ascii=False, indent=2))
     except Exception as e:
         logger.error(f"[STATS] 保存统计数据失败: {str(e)[:50]}")
 
-# 初始化统计数据
-global_stats = load_stats()
+# 初始化统计数据（需要在启动时异步加载）
+global_stats = {
+    "total_visitors": 0,
+    "total_requests": 0,
+    "request_timestamps": [],
+    "visitor_ips": {},
+    "account_conversations": {}
+}
 
 class MemoryLogHandler(logging.Handler):
     """自定义日志处理器，将日志写入内存缓冲区"""
@@ -127,7 +138,10 @@ http_client = httpx.AsyncClient(
     verify=False,
     http2=False,
     timeout=httpx.Timeout(TIMEOUT_SECONDS, connect=60.0),
-    limits=httpx.Limits(max_keepalive_connections=20, max_connections=50)
+    limits=httpx.Limits(
+        max_keepalive_connections=100,  # 增加5倍：20 -> 100
+        max_connections=200              # 增加4倍：50 -> 200
+    )
 )
 
 # ---------- 工具函数 ----------
@@ -340,11 +354,16 @@ class MultiAccountManager:
         self.accounts: Dict[str, AccountManager] = {}
         self.account_list: List[str] = []  # 账户ID列表 (用于轮询)
         self.current_index = 0
-        self._lock = asyncio.Lock()
+        self._cache_lock = asyncio.Lock()  # 缓存操作专用锁
+        self._index_lock = asyncio.Lock()  # 索引更新专用锁
         # 全局会话缓存：{conv_key: {"account_id": str, "session_id": str, "updated_at": float}}
         self.global_session_cache: Dict[str, dict] = {}
         self.cache_max_size = 1000  # 最大缓存条目数
         self.cache_ttl = SESSION_CACHE_TTL_SECONDS  # 缓存过期时间（秒）
+        # Session级别锁：防止同一对话的并发请求冲突
+        self._session_locks: Dict[str, asyncio.Lock] = {}
+        self._session_locks_lock = asyncio.Lock()  # 保护锁字典的锁
+        self._session_locks_max_size = 2000  # 最大锁数量
 
     def _clean_expired_cache(self):
         """清理过期的缓存条目"""
@@ -376,7 +395,7 @@ class MultiAccountManager:
         try:
             while True:
                 await asyncio.sleep(300)  # 5分钟
-                async with self._lock:
+                async with self._cache_lock:
                     self._clean_expired_cache()
                     self._ensure_cache_size()
         except asyncio.CancelledError:
@@ -386,7 +405,7 @@ class MultiAccountManager:
 
     async def set_session_cache(self, conv_key: str, account_id: str, session_id: str):
         """线程安全地设置会话缓存"""
-        async with self._lock:
+        async with self._cache_lock:
             self.global_session_cache[conv_key] = {
                 "account_id": account_id,
                 "session_id": session_id,
@@ -397,10 +416,25 @@ class MultiAccountManager:
 
     async def update_session_time(self, conv_key: str):
         """线程安全地更新会话时间戳"""
-        async with self._lock:
+        async with self._cache_lock:
             if conv_key in self.global_session_cache:
                 self.global_session_cache[conv_key]["updated_at"] = time.time()
 
+    async def acquire_session_lock(self, conv_key: str) -> asyncio.Lock:
+        """获取指定对话的锁（用于防止同一对话的并发请求冲突）"""
+        async with self._session_locks_lock:
+            # 清理过多的锁（LRU策略：删除不在缓存中的锁）
+            if len(self._session_locks) > self._session_locks_max_size:
+                # 只保留当前缓存中存在的锁
+                valid_keys = set(self.global_session_cache.keys())
+                keys_to_remove = [k for k in self._session_locks if k not in valid_keys]
+                for k in keys_to_remove[:len(keys_to_remove)//2]:  # 删除一半无效锁
+                    del self._session_locks[k]
+
+            if conv_key not in self._session_locks:
+                self._session_locks[conv_key] = asyncio.Lock()
+            return self._session_locks[conv_key]
+
     def add_account(self, config: AccountConfig):
         """添加账户"""
         manager = AccountManager(config)
@@ -412,43 +446,40 @@ class MultiAccountManager:
         logger.info(f"[MULTI] [ACCOUNT] 添加账户: {config.account_id}")
 
     async def get_account(self, account_id: Optional[str] = None, request_id: str = "") -> AccountManager:
-        """获取账户 (轮询或指定)"""
-        async with self._lock:
-            # 定期清理过期缓存（每次获取账户时检查）
-            self._clean_expired_cache()
-
-            req_tag = f"[req_{request_id}] " if request_id else ""
-
-            # 如果指定了账户ID
-            if account_id:
-                if account_id not in self.accounts:
-                    raise HTTPException(404, f"Account {account_id} not found")
-                account = self.accounts[account_id]
-                if not account.should_retry():
-                    raise HTTPException(503, f"Account {account_id} temporarily unavailable")
-                return account
-
-            # 轮询选择可用账户（排除过期账户和手动禁用账户）
-            available_accounts = [
-                acc_id for acc_id in self.account_list
-                if self.accounts[acc_id].should_retry()
-                and not self.accounts[acc_id].config.is_expired()
-                and not self.accounts[acc_id].config.disabled
-            ]
+        """获取账户 (轮询或指定) - 优化锁粒度，减少竞争"""
+        req_tag = f"[req_{request_id}] " if request_id else ""
 
-            if not available_accounts:
-                raise HTTPException(503, "No available accounts")
+        # 如果指定了账户ID（无需锁）
+        if account_id:
+            if account_id not in self.accounts:
+                raise HTTPException(404, f"Account {account_id} not found")
+            account = self.accounts[account_id]
+            if not account.should_retry():
+                raise HTTPException(503, f"Account {account_id} temporarily unavailable")
+            return account
 
-            # Round-robin（修复：基于可用账户列表的索引）
+        # 轮询选择可用账户（无锁读取账户列表）
+        available_accounts = [
+            acc_id for acc_id in self.account_list
+            if self.accounts[acc_id].should_retry()
+            and not self.accounts[acc_id].config.is_expired()
+            and not self.accounts[acc_id].config.disabled
+        ]
+
+        if not available_accounts:
+            raise HTTPException(503, "No available accounts")
+
+        # 只在更新索引时加锁（最小化锁持有时间）
+        async with self._index_lock:
             if not hasattr(self, '_available_index'):
                 self._available_index = 0
 
             account_id = available_accounts[self._available_index % len(available_accounts)]
             self._available_index = (self._available_index + 1) % len(available_accounts)
 
-            account = self.accounts[account_id]
-            logger.info(f"[MULTI] [ACCOUNT] {req_tag}选择账户: {account_id}")
-            return account
+        account = self.accounts[account_id]
+        logger.info(f"[MULTI] [ACCOUNT] {req_tag}选择账户: {account_id}")
+        return account
 
 # ---------- 配置文件管理 ----------
 ACCOUNTS_FILE = "accounts.json"
@@ -708,41 +739,51 @@ async def upload_context_file(session_name: str, mime_type: str, base64_content:
 
 # ---------- 消息处理逻辑 ----------
 def get_conversation_key(messages: List[dict]) -> str:
-    """使用第一条user消息生成对话指纹"""
+    """
+    生成对话指纹（使用前3条消息，平衡唯一性和Session复用）
+
+    策略：
+    1. 使用前3条消息生成指纹（而非仅第1条）
+    2. 大幅降低不同用户共享Session的概率
+    3. 保持Session复用能力（后续消息仍能找到同一Session）
+    """
     if not messages:
         return "empty"
 
-    # 只使用第一条user消息生成指纹（对话起点不变）
-    user_messages = [msg for msg in messages if msg.get("role") == "user"]
-    if not user_messages:
-        return "no_user_msg"
+    # 提取前3条消息的关键信息（角色+内容）
+    message_fingerprints = []
+    for msg in messages[:3]:  # 只取前3条
+        role = msg.get("role", "")
+        content = msg.get("content", "")
 
-    # 只取第一条user消息
-    first_user_msg = user_messages[0]
-    content = first_user_msg.get("content", "")
+        # 统一处理内容格式（字符串或数组）
+        if isinstance(content, list):
+            # 多模态消息：只提取文本部分
+            text = "".join([x.get("text", "") for x in content if x.get("type") == "text"])
+        else:
+            text = str(content)
 
-    # 统一处理内容格式（字符串或数组）
-    if isinstance(content, list):
-        text = "".join([x.get("text", "") for x in content if x.get("type") == "text"])
-    else:
-        text = str(content)
+        # 标准化：去除首尾空白，转小写
+        text = text.strip().lower()
 
-    # 标准化：去除首尾空白，转小写（避免因空格/大小写导致指纹不同）
-    text = text.strip().lower()
+        # 组合角色和内容
+        message_fingerprints.append(f"{role}:{text}")
 
-    # 生成指纹
-    return hashlib.md5(text.encode()).hexdigest()
+    # 使用前3条消息生成指纹
+    conversation_prefix = "|".join(message_fingerprints)
+    return hashlib.md5(conversation_prefix.encode()).hexdigest()
 
-def parse_last_message(messages: List['Message']):
-    """解析最后一条消息，分离文本和图片"""
+async def parse_last_message(messages: List['Message'], request_id: str = ""):
+    """解析最后一条消息，分离文本和图片（支持 base64 和 URL）"""
     if not messages:
         return "", []
-    
+
     last_msg = messages[-1]
     content = last_msg.content
-    
+
     text_content = ""
     images = [] # List of {"mime": str, "data": str_base64}
+    image_urls = []  # 需要下载的 URL
 
     if isinstance(content, str):
         text_content = content
@@ -756,8 +797,29 @@ def parse_last_message(messages: List['Message']):
                 match = re.match(r"data:(image/[^;]+);base64,(.+)", url)
                 if match:
                     images.append({"mime": match.group(1), "data": match.group(2)})
+                elif url.startswith(("http://", "https://")):
+                    image_urls.append(url)
                 else:
-                    logger.warning(f"[FILE] 不支持的图片格式: {url[:30]}...")
+                    logger.warning(f"[FILE] [req_{request_id}] 不支持的图片格式: {url[:30]}...")
+
+    # 并行下载所有 URL 图片
+    if image_urls:
+        async def download_url(url: str):
+            try:
+                resp = await http_client.get(url, timeout=30, follow_redirects=True)
+                resp.raise_for_status()
+                content_type = resp.headers.get("content-type", "image/jpeg").split(";")[0]
+                if not content_type.startswith("image/"):
+                    content_type = "image/jpeg"
+                b64 = base64.b64encode(resp.content).decode()
+                logger.info(f"[FILE] [req_{request_id}] URL图片下载成功: {url[:50]}... ({len(resp.content)} bytes)")
+                return {"mime": content_type, "data": b64}
+            except Exception as e:
+                logger.warning(f"[FILE] [req_{request_id}] URL图片下载失败: {url[:50]}... - {e}")
+                return None
+
+        results = await asyncio.gather(*[download_url(u) for u in image_urls])
+        images.extend([r for r in results if r])
 
     return text_content, images
 
@@ -782,6 +844,43 @@ def build_full_context_text(messages: List['Message']) -> str:
 # ---------- OpenAI 兼容接口 ----------
 app = FastAPI(title="Gemini-Business OpenAI Gateway")
 
+# ---------- Uptime 追踪中间件 ----------
+@app.middleware("http")
+async def track_uptime_middleware(request: Request, call_next):
+    """追踪每个请求的成功/失败状态，用于 Uptime 监控"""
+    # 只追踪 API 请求（排除静态文件、管理端点等）
+    path = request.url.path
+    if path.startswith("/images/") or path.startswith("/public/") or path.startswith("/favicon"):
+        return await call_next(request)
+
+    start_time = time.time()
+    success = False
+    model = None
+
+    try:
+        response = await call_next(request)
+        success = response.status_code < 400
+
+        # 尝试从请求中提取模型信息
+        if hasattr(request.state, "model"):
+            model = request.state.model
+
+        # 记录 API 主服务状态
+        uptime_tracker.record_request("api_service", success)
+
+        # 如果有模型信息，记录模型状态
+        if model and model in uptime_tracker.SUPPORTED_MODELS:
+            uptime_tracker.record_request(model, success)
+
+        return response
+
+    except Exception as e:
+        # 请求失败
+        uptime_tracker.record_request("api_service", False)
+        if model and model in uptime_tracker.SUPPORTED_MODELS:
+            uptime_tracker.record_request(model, False)
+        raise
+
 # ---------- 图片静态服务初始化 ----------
 os.makedirs(IMAGE_DIR, exist_ok=True)
 app.mount("/images", StaticFiles(directory=IMAGE_DIR), name="images")
@@ -794,10 +893,20 @@ else:
 @app.on_event("startup")
 async def startup_event():
     """应用启动时初始化后台任务"""
+    global global_stats
+
+    # 加载统计数据
+    global_stats = await load_stats()
+    logger.info(f"[SYSTEM] 统计数据已加载: {global_stats['total_requests']} 次请求, {global_stats['total_visitors']} 位访客")
+
     # 启动缓存清理任务
     asyncio.create_task(multi_account_mgr.start_background_cleanup())
     logger.info("[SYSTEM] 后台缓存清理任务已启动（间隔: 5分钟）")
 
+    # 启动 Uptime 数据聚合任务
+    asyncio.create_task(uptime_tracker.uptime_aggregation_task())
+    logger.info("[SYSTEM] Uptime 数据聚合任务已启动（间隔: 240秒）")
+
 # ---------- 导入模板模块 ----------
 # 注意：必须在所有全局变量初始化之后导入，避免循环依赖
 from core import templates
@@ -1335,10 +1444,10 @@ async def chat(
     request_id = str(uuid.uuid4())[:6]
 
     # 记录请求统计
-    with stats_lock:
+    async with stats_lock:
         global_stats["total_requests"] += 1
         global_stats["request_timestamps"].append(time.time())
-        save_stats(global_stats)
+        await save_stats(global_stats)
 
     # 2. 模型校验
     if req.model not in MODEL_MAPPING:
@@ -1348,45 +1457,56 @@ async def chat(
             detail=f"Model '{req.model}' not found. Available models: {list(MODEL_MAPPING.keys())}"
         )
 
-    # 3. 生成会话指纹，检查是否已有绑定的账户
-    conv_key = get_conversation_key([m.dict() for m in req.messages])
-    cached_session = multi_account_mgr.global_session_cache.get(conv_key)
-
-    if cached_session:
-        # 使用已绑定的账户
-        account_id = cached_session["account_id"]
-        account_manager = await multi_account_mgr.get_account(account_id, request_id)
-        google_session = cached_session["session_id"]
-        is_new_conversation = False
-        logger.info(f"[CHAT] [{account_id}] [req_{request_id}] 继续会话: {google_session[-12:]}")
-    else:
-        # 新对话：轮询选择可用账户，失败时尝试其他账户
-        max_account_tries = min(MAX_NEW_SESSION_TRIES, len(multi_account_mgr.accounts))
-        last_error = None
+    # 保存模型信息到 request.state（用于 Uptime 追踪）
+    request.state.model = req.model
 
-        for attempt in range(max_account_tries):
-            try:
-                account_manager = await multi_account_mgr.get_account(None, request_id)
-                google_session = await create_google_session(account_manager, request_id)
-                # 线程安全地绑定账户到此对话
-                await multi_account_mgr.set_session_cache(
-                    conv_key,
-                    account_manager.config.account_id,
-                    google_session
-                )
-                is_new_conversation = True
-                logger.info(f"[CHAT] [{account_manager.config.account_id}] [req_{request_id}] 新会话创建并绑定账户")
-                break
-            except Exception as e:
-                last_error = e
-                error_type = type(e).__name__
-                # 安全获取账户ID
-                account_id = account_manager.config.account_id if 'account_manager' in locals() and account_manager else 'unknown'
-                logger.error(f"[CHAT] [req_{request_id}] 账户 {account_id} 创建会话失败 (尝试 {attempt + 1}/{max_account_tries}) - {error_type}: {str(e)}")
-                if attempt == max_account_tries - 1:
-                    logger.error(f"[CHAT] [req_{request_id}] 所有账户均不可用")
-                    raise HTTPException(503, f"All accounts unavailable: {str(last_error)[:100]}")
-                # 继续尝试下一个账户
+    # 3. 生成会话指纹，获取Session锁（防止同一对话的并发请求冲突）
+    conv_key = get_conversation_key([m.dict() for m in req.messages])
+    session_lock = await multi_account_mgr.acquire_session_lock(conv_key)
+
+    # 4. 在锁的保护下检查缓存和处理Session（保证同一对话的请求串行化）
+    async with session_lock:
+        cached_session = multi_account_mgr.global_session_cache.get(conv_key)
+
+        if cached_session:
+            # 使用已绑定的账户
+            account_id = cached_session["account_id"]
+            account_manager = await multi_account_mgr.get_account(account_id, request_id)
+            google_session = cached_session["session_id"]
+            is_new_conversation = False
+            logger.info(f"[CHAT] [{account_id}] [req_{request_id}] 继续会话: {google_session[-12:]}")
+        else:
+            # 新对话：轮询选择可用账户，失败时尝试其他账户
+            max_account_tries = min(MAX_NEW_SESSION_TRIES, len(multi_account_mgr.accounts))
+            last_error = None
+
+            for attempt in range(max_account_tries):
+                try:
+                    account_manager = await multi_account_mgr.get_account(None, request_id)
+                    google_session = await create_google_session(account_manager, request_id)
+                    # 线程安全地绑定账户到此对话
+                    await multi_account_mgr.set_session_cache(
+                        conv_key,
+                        account_manager.config.account_id,
+                        google_session
+                    )
+                    is_new_conversation = True
+                    logger.info(f"[CHAT] [{account_manager.config.account_id}] [req_{request_id}] 新会话创建并绑定账户")
+                    # 记录服务状态（账户可用）
+                    uptime_tracker.record_request("service_status", True)
+                    break
+                except Exception as e:
+                    last_error = e
+                    error_type = type(e).__name__
+                    # 安全获取账户ID
+                    account_id = account_manager.config.account_id if 'account_manager' in locals() and account_manager else 'unknown'
+                    logger.error(f"[CHAT] [req_{request_id}] 账户 {account_id} 创建会话失败 (尝试 {attempt + 1}/{max_account_tries}) - {error_type}: {str(e)}")
+                    if attempt == max_account_tries - 1:
+                        logger.error(f"[CHAT] [req_{request_id}] 所有账户均不可用")
+                        # 记录服务状态（账户不可用）
+                        uptime_tracker.record_request("service_status", False)
+                        raise HTTPException(503, f"All accounts unavailable: {str(last_error)[:100]}")
+                    # 继续尝试下一个账户
 
     # 提取用户消息内容用于日志
     if req.messages:
@@ -1409,7 +1529,7 @@ async def chat(
     logger.info(f"[CHAT] [{account_manager.config.account_id}] [req_{request_id}] 用户消息: {preview}")
 
     # 3. 解析请求内容
-    last_text, current_images = parse_last_message(req.messages)
+    last_text, current_images = await parse_last_message(req.messages, request_id)
 
     # 4. 准备文本内容
     if is_new_conversation:
@@ -1493,11 +1613,11 @@ async def chat(
                 account_manager.conversation_count += 1  # 增加对话次数
 
                 # 保存对话次数到统计数据
-                with stats_lock:
+                async with stats_lock:
                     if "account_conversations" not in global_stats:
                         global_stats["account_conversations"] = {}
                     global_stats["account_conversations"][account_manager.config.account_id] = account_manager.conversation_count
-                    save_stats(global_stats)
+                    await save_stats(global_stats)
 
                 break
 
@@ -1715,25 +1835,66 @@ def build_image_download_url(session_name: str, file_id: str) -> str:
     return f"https://biz-discoveryengine.googleapis.com/v1alpha/{session_name}:downloadFile?fileId={file_id}&alt=media"
 
 
-async def download_image_with_jwt(account_mgr: AccountManager, session_name: str, file_id: str, request_id: str = "") -> bytes:
-    """使用JWT认证下载图片"""
+async def download_image_with_jwt(account_mgr: AccountManager, session_name: str, file_id: str, request_id: str = "", max_retries: int = 3) -> bytes:
+    """
+    使用JWT认证下载图片（带超时和重试机制）
+
+    Args:
+        account_mgr: 账户管理器
+        session_name: Session名称
+        file_id: 文件ID
+        request_id: 请求ID
+        max_retries: 最大重试次数（默认3次）
+
+    Returns:
+        图片字节数据
+
+    Raises:
+        HTTPException: 下载失败
+        asyncio.TimeoutError: 超时
+    """
     url = build_image_download_url(session_name, file_id)
-    logger.info(f"[IMAGE] [DEBUG] 下载URL: {url}")
-    logger.info(f"[IMAGE] [DEBUG] Session完整路径: {session_name}")
-    jwt = await account_mgr.get_jwt(request_id)
-    headers = get_common_headers(jwt)
+    logger.info(f"[IMAGE] [{account_mgr.config.account_id}] [req_{request_id}] 开始下载图片: {file_id[:8]}...")
 
-    # 复用全局http_client
-    resp = await http_client.get(url, headers=headers, follow_redirects=True)
+    for attempt in range(max_retries):
+        try:
+            # 3分钟超时（180秒）
+            async with asyncio.timeout(180):
+                jwt = await account_mgr.get_jwt(request_id)
+                headers = get_common_headers(jwt)
+
+                # 复用全局http_client
+                resp = await http_client.get(url, headers=headers, follow_redirects=True)
+
+                if resp.status_code == 401:
+                    # JWT过期，刷新后重试
+                    jwt = await account_mgr.get_jwt(request_id)
+                    headers = get_common_headers(jwt)
+                    resp = await http_client.get(url, headers=headers, follow_redirects=True)
+
+                resp.raise_for_status()
+                logger.info(f"[IMAGE] [{account_mgr.config.account_id}] [req_{request_id}] 图片下载成功: {file_id[:8]}... ({len(resp.content)} bytes)")
+                return resp.content
+
+        except asyncio.TimeoutError:
+            logger.warning(f"[IMAGE] [{account_mgr.config.account_id}] [req_{request_id}] 图片下载超时 (尝试 {attempt + 1}/{max_retries}): {file_id[:8]}...")
+            if attempt == max_retries - 1:
+                raise HTTPException(504, f"Image download timeout after {max_retries} attempts")
+            await asyncio.sleep(2 ** attempt)  # 指数退避：2s, 4s, 8s
+
+        except httpx.HTTPError as e:
+            logger.warning(f"[IMAGE] [{account_mgr.config.account_id}] [req_{request_id}] 图片下载失败 (尝试 {attempt + 1}/{max_retries}): {type(e).__name__}")
+            if attempt == max_retries - 1:
+                raise HTTPException(500, f"Image download failed: {str(e)[:100]}")
+            await asyncio.sleep(2 ** attempt)  # 指数退避
 
-    if resp.status_code == 401:
-        # JWT过期，刷新后重试
-        jwt = await account_mgr.get_jwt(request_id)
-        headers = get_common_headers(jwt)
-        resp = await http_client.get(url, headers=headers, follow_redirects=True)
+        except Exception as e:
+            logger.error(f"[IMAGE] [{account_mgr.config.account_id}] [req_{request_id}] 图片下载异常: {type(e).__name__}: {str(e)[:100]}")
+            raise
+
+    # 不应该到达这里
+    raise HTTPException(500, "Image download failed unexpectedly")
 
-    resp.raise_for_status()
-    return resp.content
 
 
 def save_image_to_hf(image_data: bytes, chat_id: str, file_id: str, mime_type: str, base_url: str) -> str:
@@ -1845,28 +2006,44 @@ async def stream_chat_generator(session: str, text_content: str, file_ids: List[
 
                         # 获取文件元数据，找到正确的session路径
                         file_metadata = await get_session_file_metadata(account_manager, session_name, request_id)
-                        logger.info(f"[IMAGE] [DEBUG] 获取到{len(file_metadata)}个文件元数据")
+                        logger.info(f"[IMAGE] [{account_manager.config.account_id}] [req_{request_id}] 获取到{len(file_metadata)}个文件元数据")
 
-                        for idx, file_info in enumerate(file_ids, 1):
-                            try:
-                                fid = file_info["fileId"]
-                                mime = file_info["mimeType"]
+                        # 并行下载所有图片（提升多图响应速度）
+                        download_tasks = []
+                        for file_info in file_ids:
+                            fid = file_info["fileId"]
+                            mime = file_info["mimeType"]
+
+                            # 从元数据中获取正确的session路径
+                            meta = file_metadata.get(fid, {})
+                            correct_session = meta.get("session") or session_name
 
-                                # 从元数据中获取正确的session路径
-                                meta = file_metadata.get(fid, {})
-                                correct_session = meta.get("session") or session_name
-                                logger.info(f"[IMAGE] [DEBUG] 文件{fid}使用session: {correct_session}")
+                            # 创建下载任务
+                            task = download_image_with_jwt(account_manager, correct_session, fid, request_id)
+                            download_tasks.append((fid, mime, task))
 
-                                image_data = await download_image_with_jwt(account_manager, correct_session, fid, request_id)
+                        # 并行执行所有下载任务
+                        results = await asyncio.gather(*[task for _, _, task in download_tasks], return_exceptions=True)
+
+                        # 处理下载结果
+                        for idx, ((fid, mime, _), result) in enumerate(zip(download_tasks, results), 1):
+                            try:
+                                if isinstance(result, Exception):
+                                    logger.error(f"[IMAGE] [{account_manager.config.account_id}] [req_{request_id}] 图片{idx}下载失败: {type(result).__name__}: {str(result)[:100]}")
+                                    continue
+
+                                # 保存图片
+                                image_data = result
                                 image_url = save_image_to_hf(image_data, chat_id, fid, mime, base_url)
-                                logger.info(f"[IMAGE] [{account_manager.config.account_id}] [req_{request_id}] 图片已保存: {image_url}")
+                                logger.info(f"[IMAGE] [{account_manager.config.account_id}] [req_{request_id}] 图片{idx}已保存: {image_url}")
 
                                 # 返回Markdown格式图片
                                 markdown = f"\n\n![生成的图片]({image_url})\n\n"
                                 chunk = create_chunk(chat_id, created_time, model_name, {"content": markdown}, None)
                                 yield f"data: {chunk}\n\n"
                             except Exception as e:
-                                logger.error(f"[IMAGE] [{account_manager.config.account_id}] [req_{request_id}] 单张图片处理失败: {str(e)}")
+                                logger.error(f"[IMAGE] [{account_manager.config.account_id}] [req_{request_id}] 图片{idx}处理失败: {str(e)}")
+
 
                     except Exception as e:
                         logger.error(f"[IMAGE] [{account_manager.config.account_id}] [req_{request_id}] 图片处理失败: {str(e)}")
@@ -1887,10 +2064,22 @@ async def stream_chat_generator(session: str, text_content: str, file_ids: List[
         yield "data: [DONE]\n\n"
 
 # ---------- 公开端点（无需认证） ----------
+@app.get("/public/uptime")
+async def get_public_uptime(days: int = 90):
+    """获取 Uptime 监控数据（JSON格式）"""
+    if days < 1 or days > 90:
+        days = 90
+    return await uptime_tracker.get_uptime_summary(days)
+
+@app.get("/public/uptime/html")
+async def get_public_uptime_html():
+    """Uptime 监控页面（类似 status.openai.com）"""
+    return await templates.get_uptime_html()
+
 @app.get("/public/stats")
 async def get_public_stats():
     """获取公开统计信息"""
-    with stats_lock:
+    async with stats_lock:
         # 清理1小时前的请求时间戳
         current_time = time.time()
         global_stats["request_timestamps"] = [
@@ -1927,41 +2116,45 @@ async def get_public_stats():
 @app.get("/public/log")
 async def get_public_logs(request: Request, limit: int = 100):
     """获取脱敏后的日志（JSON格式）"""
-    # 基于IP的访问统计（24小时内去重）
-    # 优先从 X-Forwarded-For 获取真实IP（处理代理情况）
-    client_ip = request.headers.get("x-forwarded-for")
-    if client_ip:
-        # X-Forwarded-For 可能包含多个IP，取第一个
-        client_ip = client_ip.split(",")[0].strip()
-    else:
-        # 没有代理时使用直连IP
-        client_ip = request.client.host if request.client else "unknown"
+    try:
+        # 基于IP的访问统计（24小时内去重）
+        # 优先从 X-Forwarded-For 获取真实IP（处理代理情况）
+        client_ip = request.headers.get("x-forwarded-for")
+        if client_ip:
+            # X-Forwarded-For 可能包含多个IP，取第一个
+            client_ip = client_ip.split(",")[0].strip()
+        else:
+            # 没有代理时使用直连IP
+            client_ip = request.client.host if request.client else "unknown"
 
-    current_time = time.time()
+        current_time = time.time()
 
-    with stats_lock:
-        # 清理24小时前的IP记录
-        if "visitor_ips" not in global_stats:
-            global_stats["visitor_ips"] = {}
+        async with stats_lock:
+            # 清理24小时前的IP记录
+            if "visitor_ips" not in global_stats:
+                global_stats["visitor_ips"] = {}
 
-        expired_ips = [
-            ip for ip, timestamp in global_stats["visitor_ips"].items()
-            if current_time - timestamp > 86400  # 24小时
-        ]
-        for ip in expired_ips:
-            del global_stats["visitor_ips"][ip]
+            expired_ips = [
+                ip for ip, timestamp in global_stats["visitor_ips"].items()
+                if current_time - timestamp > 86400  # 24小时
+            ]
+            for ip in expired_ips:
+                del global_stats["visitor_ips"][ip]
 
-        # 记录新访问（24小时内同一IP只计数一次）
-        if client_ip not in global_stats["visitor_ips"]:
-            global_stats["visitor_ips"][client_ip] = current_time
-            global_stats["total_visitors"] = len(global_stats["visitor_ips"])
-            save_stats(global_stats)
+            # 记录新访问（24小时内同一IP只计数一次）
+            if client_ip not in global_stats["visitor_ips"]:
+                global_stats["visitor_ips"][client_ip] = current_time
+                global_stats["total_visitors"] = len(global_stats["visitor_ips"])
+                await save_stats(global_stats)
 
-    sanitized_logs = get_sanitized_logs(limit=min(limit, 1000))
-    return {
-        "total": len(sanitized_logs),
-        "logs": sanitized_logs
-    }
+        sanitized_logs = get_sanitized_logs(limit=min(limit, 1000))
+        return {
+            "total": len(sanitized_logs),
+            "logs": sanitized_logs
+        }
+    except Exception as e:
+        logger.error(f"[LOG] 获取公开日志失败: {e}")
+        return {"total": 0, "logs": [], "error": str(e)}
 
 @app.get("/public/log/html")
 async def get_public_logs_html():

requirements.txt

@@ -1,4 +1,6 @@
 fastapi==0.110.0
 uvicorn[standard]==0.29.0
 httpx==0.27.0
-pydantic==2.7.0
+pydantic==2.7.0
+aiofiles==24.1.0
+python-dotenv==1.0.1

uptime_tracker.py

@@ -0,0 +1,78 @@
+"""
+Uptime 实时监控追踪器
+类似 Uptime Kuma 的心跳监控，显示最近请求状态
+"""
+
+from collections import deque
+from datetime import datetime, timezone, timedelta
+from typing import Dict, List
+
+# 北京时区 UTC+8
+BEIJING_TZ = timezone(timedelta(hours=8))
+
+# 每个服务保留最近 60 条心跳记录
+MAX_HEARTBEATS = 60
+
+# 服务配置
+SERVICES = {
+    "api_service": {"name": "API 服务", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "service_status": {"name": "服务资源", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "gemini-2.5-flash": {"name": "Gemini 2.5 Flash", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "gemini-2.5-pro": {"name": "Gemini 2.5 Pro", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "gemini-3-flash-preview": {"name": "Gemini 3 Flash Preview", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "gemini-3-pro-preview": {"name": "Gemini 3 Pro Preview", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+}
+
+SUPPORTED_MODELS = ["gemini-2.5-flash", "gemini-2.5-pro", "gemini-3-flash-preview", "gemini-3-pro-preview"]
+
+
+def record_request(service: str, success: bool):
+    """记录请求心跳"""
+    if service not in SERVICES:
+        return
+
+    SERVICES[service]["heartbeats"].append({
+        "time": datetime.now(BEIJING_TZ).strftime("%H:%M:%S"),
+        "success": success
+    })
+
+
+def get_realtime_status() -> Dict:
+    """获取实时状态数据"""
+    result = {"services": {}}
+
+    for service_id, service_data in SERVICES.items():
+        heartbeats = list(service_data["heartbeats"])
+        total = len(heartbeats)
+        success = sum(1 for h in heartbeats if h["success"])
+
+        # 计算可用率
+        uptime = (success / total * 100) if total > 0 else 100.0
+
+        # 最近状态
+        last_status = "unknown"
+        if heartbeats:
+            last_status = "up" if heartbeats[-1]["success"] else "down"
+
+        result["services"][service_id] = {
+            "name": service_data["name"],
+            "status": last_status,
+            "uptime": round(uptime, 1),
+            "total": total,
+            "success": success,
+            "heartbeats": heartbeats[-MAX_HEARTBEATS:]  # 最近的心跳
+        }
+
+    result["updated_at"] = datetime.now(BEIJING_TZ).strftime("%Y-%m-%d %H:%M:%S")
+    return result
+
+
+# 兼容旧接口
+async def get_uptime_summary(days: int = 90) -> Dict:
+    """兼容旧接口，返回实时数据"""
+    return get_realtime_status()
+
+
+async def uptime_aggregation_task():
+    """后台任务（保留兼容性，实际不需要聚合）"""
+    pass