update sth at 2025-10-23 17:38:54

.env

@@ -1,54 +1,16 @@
-# 代理服务配置文件示例
-# 复制此文件为 .env 并根据需要修改配置值
+# 代理服务配置文件
+# 匿名模式配置
 
-# ========== API 基础配置 ==========
-# 客户端认证密钥（您自定义的 API 密钥，用于客户端访问本服务）
-AUTH_TOKEN=sk-your-api-key
-
-
-# 跳过客户端认证（仅开发环境使用）
-SKIP_AUTH_TOKEN=false
-
-# ========== Z.ai Token池配置 ==========
-# Token失败阈值（失败多少次后标记为不可用）
-TOKEN_FAILURE_THRESHOLD=3
-
-# Token恢复超时时间（秒，失败token在此时间后重新尝试）
-TOKEN_RECOVERY_TIMEOUT=1800
+# 跳过客户端认证（启用匿名访问）
+SKIP_AUTH_TOKEN=true
 
 # Z.AI 匿名用户模式
 # false: 使用认证 Token 令牌，失败时自动降级为匿名请求
 # true: 自动从 Z.ai 获取临时访问令牌，避免对话历史共享
 ANONYMOUS_MODE=true
 
-# ========== LongCat 配置 ==========
-# LongCat token（单个token）
-# LONGCAT_TOKEN=your_passport_token_here
-
-# ========== 服务器配置 ==========
 # 服务监听端口
 LISTEN_PORT=7860
 
-# 服务名称
-SERVICE_NAME=z-ai2api-server
-
 # 调试日志
-DEBUG_LOGGING=false
-
-# Nginx 反向代理路径前缀（可选，用于在子路径下部署）
-# 例如：ROOT_PATH=/ai2api 则服务部署在 http://domain.com/ai2api
-# 留空表示部署在根路径
-ROOT_PATH=/api
-
-# Function Call 功能开关
-TOOL_SUPPORT=true
-
-# 工具调用扫描限制（字符数）
-SCAN_LIMIT=200000
-
-# ========== 管理后台认证 ==========
-# 管理后台登录密码（建议修改为复杂密码）
-ADMIN_PASSWORD=admin123
-
-# Session 密钥（用于加密会话，建议生成随机字符串）
-SESSION_SECRET_KEY=your-secret-key-change-in-production
+DEBUG_LOGGING=true

.env.example

@@ -1,53 +1,35 @@
-# 代理服务配置文件示例
-# 复制此文件为 .env 并根据需要修改配置值
+# ==============================================
+# Z.AI API Server - Docker 环境变量配置示例
+# ==============================================
 
-# ========== API 基础配置 ==========
-# 客户端认证密钥（您自定义的 API 密钥，用于客户端访问本服务）
-AUTH_TOKEN=sk-your-api-key
-
-# 跳过客户端认证（仅开发环境使用）
-SKIP_AUTH_TOKEN=false
-
-# ========== Z.ai Token池配置 ==========
-# Token失败阈值（失败多少次后标记为不可用）
-TOKEN_FAILURE_THRESHOLD=3
-
-# Token恢复超时时间（秒，失败token在此时间后重新尝试）
-TOKEN_RECOVERY_TIMEOUT=1800
-
-# Z.AI 匿名用户模式
-# false: 使用认证 Token 令牌，失败时自动降级为匿名请求
-# true: 自动从 Z.ai 获取临时访问令牌，避免对话历史共享
-ANONYMOUS_MODE=true
-
-# ========== LongCat 配置 ==========
-# LongCat token（单个token）
-# LONGCAT_TOKEN=your_passport_token_here
+# 管理后台密码
+ADMIN_PASSWORD=admin123
 
-# ========== 服务器配置 ==========
-# 服务监听端口
-LISTEN_PORT=7860
+# API 认证密钥 (用于验证客户端请求)
+AUTH_TOKEN=sk-your-api-key-here
 
-# 服务名称
-SERVICE_NAME=z-ai2api-server
+# 是否跳过 API Key 验证 (开发环境可设为 true)
+SKIP_AUTH_TOKEN=false
 
-# 调试日志
-DEBUG_LOGGING=false
+# 调试日志 (生产环境建议设为 false)
+DEBUG_LOGGING=true
 
-# Nginx 反向代理路径前缀（可选，用于在子路径下部署）
-# 例如：ROOT_PATH=/ai2api 则服务部署在 http://domain.com/ai2api
-# 留空表示部署在根路径
-ROOT_PATH=
+# 匿名模式 (允许无 token 访问,需要配合 SKIP_AUTH_TOKEN=true)
+ANONYMOUS_MODE=false
 
-# Function Call 功能开关
+# Function Call 功能开关 (是否支持工具调用)
 TOOL_SUPPORT=true
 
-# 工具调用扫描限制（字符数）
+# 工具调用扫描限制 (字符数)
 SCAN_LIMIT=200000
 
-# ========== 管理后台认证 ==========
-# 管理后台登录密码（建议修改为复杂密码）
-ADMIN_PASSWORD=admin123
+# 数据库路径 (Docker 环境使用持久化卷)
+DB_PATH=/app/data/tokens.db
 
-# Session 密钥（用于加密会话，建议生成随机字符串）
-SESSION_SECRET_KEY=your-secret-key-change-in-production
+# Token 池配置
+TOKEN_FAILURE_THRESHOLD=3
+TOKEN_RECOVERY_TIMEOUT=300
+
+# 服务配置
+SERVICE_NAME=Z.AI_API_Server
+LISTEN_PORT=7860

Dockerfile

@@ -10,20 +10,22 @@ RUN mkdir -p /app/data /app/logs && \
 # Install dependencies
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
-RUN  pip install --upgrade pip
+
 # Copy application code
 COPY . .
 
 # Set environment variable for database path
 ENV DB_PATH=/app/data/tokens.db
-ENV AUTH_TOKEN=sk-your-key
-ENV ROOT_PATH=/api
+ENV SKIP_AUTH_TOKEN=true
 ENV ANONYMOUS_MODE=true
+# 服务监听端口
 ENV LISTEN_PORT=7860
-ENV TOOL_SUPPORT=true
+# 调试日志
+ENV DEBUG_LOGGING=true
 
 # Expose port
 EXPOSE 7860
 
+
 # Run the application
 CMD ["python", "main.py"]

deploy/NGINX_SETUP.md → NGINX_SETUP.md

@@ -37,7 +37,7 @@ server {
 
     location /ai2api {
         # 代理到后端服务
-        proxy_pass http://127.0.0.1:8080;
+        proxy_pass http://127.0.0.1:7860;
 
         # 传递原始请求信息
         proxy_set_header Host $host;
@@ -257,9 +257,9 @@ server {
 
 ```nginx
 upstream ai2api_backend {
-     server 127.0.0.1:7860;
-     server 127.0.0.1:7861;
-     server 127.0.0.1:7862;
+    server 127.0.0.1:7860;
+    server 127.0.0.1:8081;
+    server 127.0.0.1:8082;
 }
 
 server {

deploy/README_DOCKER.md → README_DOCKER.md

@@ -225,7 +225,7 @@ docker compose config
 如端口 7860 被占用,修改 `docker-compose.yml`:
 ```yaml
 ports:
-  - "7861:7860"  # 映射到宿主机 7861 端口
+  - "8081:7860"  # 映射到宿主机 8081 端口
 ```
 
 ### 健康检查失败

app/core/config.py

@@ -70,23 +70,15 @@ class Settings(BaseSettings):
     # Provider Configuration
     DEFAULT_PROVIDER: str = os.getenv("DEFAULT_PROVIDER", "zai")  # 默认提供商：zai/k2think/longcat
 
+    # Proxy Configuration
+    HTTP_PROXY: Optional[str] = os.getenv("HTTP_PROXY")  # HTTP代理,格式: http://user:pass@host:port 或 http://host:port
+    HTTPS_PROXY: Optional[str] = os.getenv("HTTPS_PROXY")  # HTTPS代理,格式同上
+    SOCKS5_PROXY: Optional[str] = os.getenv("SOCKS5_PROXY")  # SOCKS5代理,格式: socks5://user:pass@host:port
+
     # Admin Panel Authentication
     ADMIN_PASSWORD: str = os.getenv("ADMIN_PASSWORD", "admin123")  # 管理后台密码
     SESSION_SECRET_KEY: str = os.getenv("SESSION_SECRET_KEY", "your-secret-key-change-in-production")  # Session 密钥
 
-    # Browser Headers
-    CLIENT_HEADERS: Dict[str, str] = {
-        "Content-Type": "application/json",
-        "Accept": "application/json, text/event-stream",
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0",
-        "Accept-Language": "zh-CN",
-        "sec-ch-ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"',
-        "sec-ch-ua-mobile": "?0",
-        "sec-ch-ua-platform": '"Windows"',
-        "X-FE-Version": "prod-fe-1.0.98",
-        "Origin": "https://chat.z.ai",
-    }
-
     class Config:
         env_file = ".env"
         extra = "ignore"  # 忽略额外字段，防止环境变量中的未知字段导致验证错误

app/core/openai.py

@@ -93,6 +93,8 @@ async def handle_non_stream_response(stream_response, request: OpenAIRequest) ->
 
 
 @router.get("/v1/models")
+@router.get("/hf/v1/models")
+@router.get("/api/v1/models")
 async def list_models():
     """List available models from all providers"""
     try:
@@ -115,6 +117,8 @@ async def list_models():
 
 
 @router.post("/v1/chat/completions")
+@router.post("/hf/v1/chat/completions")
+@router.post("/api/v1/chat/completions")
 async def chat_completions(request: OpenAIRequest, authorization: str = Header(...)):
     """Handle chat completion requests with multi-provider architecture"""
     role = request.messages[0].role if request.messages else "unknown"

app/providers/zai_provider.py

@@ -12,6 +12,7 @@ import httpx
 import hmac
 import hashlib
 import base64
+import asyncio
 from urllib.parse import urlencode
 import os
 import uuid
@@ -19,6 +20,7 @@ import random
 from datetime import datetime
 from typing import Dict, List, Any, Optional, AsyncGenerator, Union
 from app.utils.user_agent import get_random_user_agent
+from app.utils.signature import generate_signature
 from app.providers.base import BaseProvider, ProviderConfig
 from app.models.schemas import OpenAIRequest, Message
 from app.core.config import settings
@@ -68,7 +70,7 @@ def get_zai_dynamic_headers(chat_id: str = "") -> Dict[str, str]:
         "Cache-Control": "no-cache",
         "User-Agent": user_agent,
         "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
-        "X-FE-Version": "prod-fe-1.0.98",
+        "X-FE-Version": "prod-fe-1.0.106",
         "Origin": "https://chat.z.ai",
     }
 
@@ -116,26 +118,6 @@ def _extract_user_id_from_token(token: str) -> str:
     return "guest"
 
 
-def generate_signature(message_text: str, request_id: str, timestamp_ms: int, user_id: str, secret: str = "junjie") -> str:
-    """Dual-layer HMAC-SHA256 signature.
-
-    Layer1: derived key = HMAC(secret, window_index)
-    Layer2: signature = HMAC(derived_key, canonical_string)
-    canonical_string = "requestId,<id>,timestamp,<ts>,user_id,<uid>|<msg>|<ts>"
-    """
-    r = str(timestamp_ms)
-    e = f"requestId,{request_id},timestamp,{timestamp_ms},user_id,{user_id}"
-    t = message_text or ""
-    # Add content_base64 processing for new signature algorithm
-    content_base64 = base64.b64encode(t.encode('utf-8')).decode('ascii')
-    i = f"{e}|{content_base64}|{r}"
-
-    window_index = timestamp_ms // (5 * 60 * 1000)
-    root_key = (secret or "junjie").encode("utf-8")
-    derived_hex = hmac.new(root_key, str(window_index).encode("utf-8"), hashlib.sha256).hexdigest()
-    signature = hmac.new(derived_hex.encode("utf-8"), i.encode("utf-8"), hashlib.sha256).hexdigest()
-    return signature
-
 
 class ZAIProvider(BaseProvider):
     """Z.AI 提供商"""
@@ -179,30 +161,94 @@ class ZAIProvider(BaseProvider):
             settings.GLM46_SEARCH_MODEL,
             settings.GLM46_ADVANCED_SEARCH_MODEL,
         ]
-    
+
+    def _get_proxy_config(self) -> Optional[str]:
+        """Get proxy configuration from settings"""
+        # In httpx 0.28.1, proxy parameter expects a single URL string
+        # Support HTTP_PROXY, HTTPS_PROXY and SOCKS5_PROXY
+        
+        if settings.HTTPS_PROXY:
+            self.logger.info(f"🔄 使用HTTPS代理: {settings.HTTPS_PROXY}")
+            return settings.HTTPS_PROXY
+            
+        if settings.HTTP_PROXY:
+            self.logger.info(f"🔄 使用HTTP代理: {settings.HTTP_PROXY}")
+            return settings.HTTP_PROXY
+            
+        if settings.SOCKS5_PROXY:
+            self.logger.info(f"🔄 使用SOCKS5代理: {settings.SOCKS5_PROXY}")
+            return settings.SOCKS5_PROXY
+
+        return None
+
     async def get_token(self) -> str:
         """获取认证令牌"""
         # 如果启用匿名模式，只尝试获取访客令牌
         if settings.ANONYMOUS_MODE:
-            try:
-                headers = get_zai_dynamic_headers()
-                async with httpx.AsyncClient() as client:
-                    response = await client.get(self.auth_url, headers=headers, timeout=10.0)
-                    if response.status_code == 200:
-                        data = response.json()
-                        token = data.get("token", "")
-                        if token:
-                            # 判断令牌类型（通过检查邮箱或user_id）
-                            email = data.get("email", "")
-                            is_guest = "@guest.com" in email or "Guest-" in email
-                            token_type = "匿名用户" if is_guest else "认证用户"
-                            self.logger.debug(f"获取令牌成功 ({token_type}): {token[:20]}...")
-                            return token
-            except Exception as e:
-                self.logger.warning(f"异步获取访客令牌失败: {e}")
+            max_retries = 3
+            retry_count = 0
+            
+            while retry_count < max_retries:
+                try:
+                    headers = get_zai_dynamic_headers()
+                    self.logger.debug(f"尝试获取访客令牌 (第{retry_count + 1}次): {self.auth_url}")
+                    self.logger.debug(f"请求头: {headers}")
+
+                    # Get proxy configuration
+                    proxies = self._get_proxy_config()
+
+                    async with httpx.AsyncClient(timeout=30.0, follow_redirects=True, proxy=proxies) as client:
+                        response = await client.get(self.auth_url, headers=headers)
+                        
+                        self.logger.debug(f"响应状态码: {response.status_code}")
+                        self.logger.debug(f"响应头: {dict(response.headers)}")
+                        
+                        if response.status_code == 200:
+                            data = response.json()
+                            self.logger.debug(f"响应数据: {data}")
+                            
+                            token = data.get("token", "")
+                            if token:
+                                # 判断令牌类型（通过检查邮箱或user_id）
+                                email = data.get("email", "")
+                                is_guest = "@guest.com" in email or "Guest-" in email
+                                token_type = "匿名用户" if is_guest else "认证用户"
+                                self.logger.info(f"✅ 获取令牌成功 ({token_type}): {token[:20]}...")
+                                return token
+                            else:
+                                self.logger.warning(f"响应中未找到token字段: {data}")
+                        elif response.status_code == 405:
+                            # WAF拦截
+                            self.logger.error(f"🚫 请求被WAF拦截 (状态码405),请求头可能被识别为异常,请稍后重试...")
+                            break
+                        else:
+                            self.logger.warning(f"HTTP请求失败,状态码: {response.status_code}")
+                            try:
+                                error_data = response.json()
+                                self.logger.warning(f"错误响应: {error_data}")
+                            except:
+                                self.logger.warning(f"错误响应文本: {response.text}")
+                                
+                except httpx.TimeoutException as e:
+                    self.logger.warning(f"请求超时 (第{retry_count + 1}次): {e}")
+                except httpx.ConnectError as e:
+                    self.logger.warning(f"连接错误 (第{retry_count + 1}次): {e}")
+                except httpx.HTTPStatusError as e:
+                    self.logger.warning(f"HTTP状态错误 (第{retry_count + 1}次): {e}")
+                except json.JSONDecodeError as e:
+                    self.logger.warning(f"JSON解析错误 (第{retry_count + 1}次): {e}")
+                except Exception as e:
+                    self.logger.warning(f"异步获取访客令牌失败 (第{retry_count + 1}次): {e}")
+                    import traceback
+                    self.logger.debug(f"错误堆栈: {traceback.format_exc()}")
+                
+                retry_count += 1
+                if retry_count < max_retries:
+                    self.logger.info(f"等待2秒后重试...")
+                    await asyncio.sleep(2)
 
             # 匿名模式下，如果获取访客令牌失败，直接返回空
-            self.logger.error("❌ 匿名模式下获取访客令牌失败")
+            self.logger.error("❌ 匿名模式下获取访客令牌失败，已重试3次")
             return ""
 
         # 非匿名模式：首先使用token池获取备份令牌
@@ -253,7 +299,7 @@ class ZAIProvider(BaseProvider):
 
             self.logger.debug(f"📤 上传图片: {filename}, 大小: {len(image_data)} bytes")
 
-            # 构建上传请求 - 使用简化的请求头配置
+            # 构建上传请求
             upload_url = f"{self.base_url}/api/v1/files/"
             headers = {
                 "Accept": "*/*",
@@ -273,8 +319,11 @@ class ZAIProvider(BaseProvider):
                 "Authorization": f"Bearer {token}",
             }
 
+            # Get proxy configuration
+            proxies = self._get_proxy_config()
+
             # 使用 httpx 上传文件
-            async with httpx.AsyncClient(timeout=30.0) as client:
+            async with httpx.AsyncClient(timeout=30.0, proxy=proxies) as client:
                 files = {
                     "file": (filename, image_data, mime_type)
                 }
@@ -501,12 +550,8 @@ class ZAIProvider(BaseProvider):
         if is_advanced_search:
             mcp_servers.append("advanced-search")
             self.logger.info("🔍 检测到高级搜索模型，添加 advanced-search MCP 服务器")
-        elif is_search and "-4.5" in requested_model:
-            mcp_servers.append("deep-web-search")
-            self.logger.info("🔍 检测到搜索模型，添加 deep-web-search MCP 服务器")
-
-        # 构建上游请求体（chat_id 已在前面生成）
 
+        # 构建上游请求体
         body = {
             "stream": True,  # 总是使用流式
             "model": upstream_model_id,
@@ -587,34 +632,48 @@ class ZAIProvider(BaseProvider):
         if request.max_tokens is not None:
             body["params"]["max_tokens"] = request.max_tokens
         
-        # 构建请求头
-        headers = get_zai_dynamic_headers(chat_id)
-        if token:
-            headers["Authorization"] = f"Bearer {token}"
-
         # Dual-layer HMAC signing metadata and header
         user_id = _extract_user_id_from_token(token)
         timestamp_ms = int(time.time() * 1000)
         request_id = generate_uuid()
-        secret = os.getenv("ZAI_SIGNING_SECRET", "junjie") or "junjie"
-        signature = generate_signature(
-            message_text=last_user_text,
-            request_id=request_id,
-            timestamp_ms=timestamp_ms,
-            user_id=user_id,
-            secret=secret,
-        )
+        try:
+            signing_metadata = f"requestId,{request_id},timestamp,{timestamp_ms},user_id,{user_id}"
+            prompt_for_signature = last_user_text or ""
+            signature_result = generate_signature(
+                e=signing_metadata,
+                t=prompt_for_signature,
+                s=timestamp_ms,
+            )
+            signature = signature_result["signature"]
+            logger.debug(f"[Z.AI] 生成签名成功: {signature[:16]}... (user_id={user_id}, request_id={request_id})")
+        except Exception as e:
+            logger.error(f"[Z.AI] 签名生成失败: {e}")
+            signature = ""
+
+        # 构建请求头 (匹配 X-FE-Version 和 X-Signature)
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "X-FE-Version": "prod-fe-1.0.106",
+            "X-Signature": signature,
+        }
+
         query_params = {
-            "timestamp": timestamp_ms,
+            "timestamp": str(timestamp_ms),
             "requestId": request_id,
             "user_id": user_id,
-            "token": token or "",
+            "token": token,
+            "version": "0.0.1",
+            "platform": "web",
             "current_url": f"https://chat.z.ai/c/{chat_id}",
             "pathname": f"/c/{chat_id}",
-            "signature_timestamp": timestamp_ms,
+            "signature_timestamp": str(timestamp_ms),
         }
         signed_url = f"{self.config.api_endpoint}?{urlencode(query_params)}"
-        headers["X-Signature"] = signature
+
+        # 记录请求详情用于调试
+        logger.debug(f"[Z.AI] 请求头: Authorization=Bearer *****, X-Signature={signature[:16] if signature else '(空)'}...")
+        logger.debug(f"[Z.AI] URL 参数: timestamp={timestamp_ms}, requestId={request_id}, user_id={user_id}")
         
         # 存储当前token用于错误处理
         self._current_token = token
@@ -645,8 +704,11 @@ class ZAIProvider(BaseProvider):
                 # 流式响应
                 return self._create_stream_response(request, transformed)
             else:
+                # Get proxy configuration
+                proxies = self._get_proxy_config()
+
                 # 非流式响应
-                async with httpx.AsyncClient(timeout=30.0) as client:
+                async with httpx.AsyncClient(timeout=30.0, proxy=proxies) as client:
                     response = await client.post(
                         transformed["url"],
                         headers=transformed["headers"],
@@ -673,9 +735,13 @@ class ZAIProvider(BaseProvider):
 
         current_token = transformed.get("token", "")
         try:
+            # Get proxy configuration
+            proxies = self._get_proxy_config()
+
             async with httpx.AsyncClient(
                 timeout=60.0,
                 http2=True,
+                proxy=proxies,
             ) as client:
                 self.logger.info(f"🎯 发送请求到 Z.AI: {transformed['url']}")
                 # self.logger.info(f"📦 请求体 model: {transformed['body']['model']}")
@@ -692,13 +758,25 @@ class ZAIProvider(BaseProvider):
                         error_msg = error_text.decode('utf-8', errors='ignore')
                         if error_msg:
                             self.logger.error(f"❌ 错误详情: {error_msg}")
-                        error_response = {
-                            "error": {
-                                "message": f"Upstream error: {response.status_code}",
-                                "type": "upstream_error",
-                                "code": response.status_code
+
+                        # 特殊处理 405 状态码(WAF拦截)
+                        if response.status_code == 405:
+                            self.logger.error(f"🚫 请求被上游WAF拦截,可能是请求头或签名异常,请稍后重试...")
+                            error_response = {
+                                "error": {
+                                    "message": "请求被上游WAF拦截(405 Method Not Allowed),可能是请求头或签名异常,请稍后重试...",
+                                    "type": "waf_blocked",
+                                    "code": 405
+                                }
+                            }
+                        else:
+                            error_response = {
+                                "error": {
+                                    "message": f"Upstream error: {response.status_code}",
+                                    "type": "upstream_error",
+                                    "code": response.status_code
+                                }
                             }
-                        }
                         yield f"data: {json.dumps(error_response)}\n\n"
                         yield "data: [DONE]\n\n"
                         return
@@ -859,10 +937,9 @@ class ZAIProvider(BaseProvider):
 
                                         # 尝试从缓冲区提取 tool_calls
                                         tool_calls = None
-                                        cleaned_content = buffered_content
 
                                         if has_tools:
-                                            tool_calls, cleaned_content = parse_and_extract_tool_calls(buffered_content)
+                                            tool_calls, _ = parse_and_extract_tool_calls(buffered_content)
 
                                         if tool_calls:
                                             # 发现工具调用
@@ -909,28 +986,8 @@ class ZAIProvider(BaseProvider):
                                             yield "data: [DONE]\n\n"
 
                                         else:
-                                            # 没有工具调用,正常返回内容
-                                            # 处理思考结束和答案开始
-                                            if edit_content and "</details>\n" in edit_content:
-                                                if has_thinking:
-                                                    # 发送思考签名
-                                                    thinking_signature = str(int(time.time() * 1000))
-                                                    sig_chunk = self.create_openai_chunk(
-                                                        chat_id,
-                                                        model,
-                                                        {
-                                                            "role": "assistant",
-                                                            "thinking": {
-                                                                "content": "",
-                                                                "signature": thinking_signature,
-                                                            }
-                                                        }
-                                                    )
-                                                    yield await self.format_sse_chunk(sig_chunk)
-
-                                                # 提取答案内容
-                                                cleaned_content = edit_content.split("</details>\n")[-1]
-
+                                            # 没有工具调用,流式内容已经在上面的增量输出中发送过了
+                                            # 这里只需要发送 finish 块即可,不要再次发送内容
                                             if not has_sent_role and not has_thinking:
                                                 role_chunk = self.create_openai_chunk(
                                                     chat_id,
@@ -940,17 +997,6 @@ class ZAIProvider(BaseProvider):
                                                 yield await self.format_sse_chunk(role_chunk)
                                                 has_sent_role = True
 
-                                            if cleaned_content:
-                                                content_chunk = self.create_openai_chunk(
-                                                    chat_id,
-                                                    model,
-                                                    {
-                                                        "role": "assistant",
-                                                        "content": cleaned_content
-                                                    }
-                                                )
-                                                yield await self.format_sse_chunk(content_chunk)
-
                                             finish_chunk = self.create_openai_chunk(
                                                 chat_id,
                                                 model,

app/utils/__init__.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
-from app.utils import sse_tool_handler, reload_config, logger
+from app.utils import reload_config, logger
 
-__all__ = ["sse_tool_handler", "reload_config", "logger"]
+__all__ = ["reload_config", "logger"]

app/utils/signature.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+Z.AI 签名工具模块
+"""
+
+import hmac
+import hashlib
+import base64
+from typing import Dict
+
+
+def generate_signature(e: str, t: str, s: int) -> dict:
+    """Generate signature matching JavaScript zs function.
+
+    Args:
+        e: canonical metadata string, e.g. "requestId,<uuid>,timestamp,<ms>,user_id,<id>"
+        t: latest user message text that feeds into the signature prompt (may be empty)
+        s: timestamp in milliseconds
+    
+    Returns:
+        Dictionary with signature and timestamp
+    """
+    # r = Number(s) - convert to number (already a number in Python)
+    r = s
+    # i = s - timestamp as string
+    i = str(s)
+    
+    # n = new TextEncoder
+    # a = n.encode(t)
+    a = t.encode('utf-8')
+    
+    # w = btoa(String.fromCharCode(...a))
+    # This is equivalent to base64 encoding the UTF-8 bytes
+    w = base64.b64encode(a).decode('ascii')
+    
+    # c = `${e}|${w}|${i}`
+    c = f"{e}|{w}|{i}"
+    
+    # E = Math.floor(r / (5 * 60 * 1e3))
+    E = r // (5 * 60 * 1000)
+    
+    # A = CryptoJS.HmacSHA256(`${E}`, "key-@@@@)))()((9))-xxxx&&&%%%%%")
+    secret = "key-@@@@)))()((9))-xxxx&&&%%%%%"
+    A = hmac.new(secret.encode('utf-8'), str(E).encode('utf-8'), hashlib.sha256).hexdigest()
+    
+    # k = CryptoJS.HmacSHA256(c, A).toString()
+    k = hmac.new(A.encode('utf-8'), c.encode('utf-8'), hashlib.sha256).hexdigest()
+    
+    # return n.encode(c), { signature: k, timestamp: i }
+    # Note: n.encode(c) is not used in the return value, so we ignore it
+    return {
+        "signature": k,
+        "timestamp": i
+    }

app/utils/sse_tool_handler.py

@@ -1,612 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-"""
-SSE Tool Handler
-
-处理 Z.AI SSE 流数据并转换为 OpenAI 兼容格式的工具调用处理器。
-
-主要功能：
-- 解析 glm_block 格式的工具调用
-- 从 metadata.arguments 提取完整参数
-- 支持多阶段处理：thinking → tool_call → other → answer
-- 输出符合 OpenAI API 规范的流式响应
-"""
-
-import json
-import time
-from typing import Dict, Any, Generator
-from enum import Enum
-
-from app.utils.logger import get_logger
-
-logger = get_logger()
-
-
-class SSEPhase(Enum):
-    """SSE 处理阶段枚举"""
-    THINKING = "thinking"
-    TOOL_CALL = "tool_call"
-    OTHER = "other"
-    ANSWER = "answer"
-    DONE = "done"
-
-
-class SSEToolHandler:
-    """SSE 工具调用处理器"""
-
-    def __init__(self, model: str, stream: bool = True):
-        self.model = model
-        self.stream = stream
-
-        # 状态管理
-        self.current_phase = None
-        self.has_tool_call = False
-
-        # 工具调用状态
-        self.tool_id = ""
-        self.tool_name = ""
-        self.tool_args = ""
-        self.tool_call_usage = {}
-        self.content_index = 0  # 工具调用索引
-
-        # 性能优化：内容缓冲
-        self.content_buffer = ""
-        self.buffer_size = 0
-        self.last_flush_time = time.time()
-        self.flush_interval = 0.05  # 50ms 刷新间隔
-        self.max_buffer_size = 100  # 最大缓冲字符数
-
-        logger.debug(f"🔧 初始化工具处理器: model={model}, stream={stream}")
-
-    def process_sse_chunk(self, chunk_data: Dict[str, Any]) -> Generator[str, None, None]:
-        """
-        处理 SSE 数据块，返回 OpenAI 格式的流式响应
-
-        Args:
-            chunk_data: Z.AI SSE 数据块
-
-        Yields:
-            str: OpenAI 格式的 SSE 响应行
-        """
-        try:
-            phase = chunk_data.get("phase")
-            edit_content = chunk_data.get("edit_content", "")
-            delta_content = chunk_data.get("delta_content", "")
-            edit_index = chunk_data.get("edit_index")
-            usage = chunk_data.get("usage", {})
-
-            # 数据验证
-            if not phase:
-                logger.warning("⚠️ 收到无效的 SSE 块：缺少 phase 字段")
-                return
-
-            # 阶段变化检测和日志
-            if phase != self.current_phase:
-                # 阶段变化时强制刷新缓冲区
-                if hasattr(self, 'content_buffer') and self.content_buffer:
-                    yield from self._flush_content_buffer()
-
-                logger.info(f"📈 SSE 阶段变化: {self.current_phase} → {phase}")
-                content_preview = edit_content or delta_content
-                if content_preview:
-                    logger.debug(f"   📝 内容预览: {content_preview[:1000]}{'...' if len(content_preview) > 1000 else ''}")
-                if edit_index is not None:
-                    logger.debug(f"   📍 edit_index: {edit_index}")
-                self.current_phase = phase
-
-            # 根据阶段处理
-            if phase == SSEPhase.THINKING.value:
-                yield from self._process_thinking_phase(delta_content)
-
-            elif phase == SSEPhase.TOOL_CALL.value:
-                yield from self._process_tool_call_phase(edit_content)
-
-            elif phase == SSEPhase.OTHER.value:
-                yield from self._process_other_phase(usage, edit_content)
-
-            elif phase == SSEPhase.ANSWER.value:
-                yield from self._process_answer_phase(delta_content)
-
-            elif phase == SSEPhase.DONE.value:
-                yield from self._process_done_phase(chunk_data)
-            else:
-                logger.warning(f"⚠️ 未知的 SSE 阶段: {phase}")
-
-        except Exception as e:
-            logger.error(f"❌ 处理 SSE 块时发生错误: {e}")
-            logger.debug(f"   📦 错误块数据: {chunk_data}")
-            # 不中断流，继续处理后续块
-
-    def _process_thinking_phase(self, delta_content: str) -> Generator[str, None, None]:
-        """处理思考阶段"""
-        if not delta_content:
-            return
-
-        logger.debug(f"🤔 思考内容: +{len(delta_content)} 字符")
-
-        # 在流模式下输出思考内容
-        if self.stream:
-            chunk = self._create_content_chunk(delta_content)
-            yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
-
-    def _process_tool_call_phase(self, edit_content: str) -> Generator[str, None, None]:
-        """处理工具调用阶段"""
-        if not edit_content:
-            return
-
-        logger.debug(f"🔧 进入工具调用阶段，内容长度: {len(edit_content)}")
-
-        # 检测 glm_block 标记
-        if "<glm_block " in edit_content:
-            yield from self._handle_glm_blocks(edit_content)
-        else:
-            # 没有 glm_block 标记，可能是参数补充
-            if self.has_tool_call:
-                # 只累积���数部分，找到第一个 ", "result"" 之前的内容
-                result_pos = edit_content.find('", "result"')
-                if result_pos > 0:
-                    param_fragment = edit_content[:result_pos]
-                    self.tool_args += param_fragment
-                    logger.debug(f"📦 累积参数片段: {param_fragment}")
-                else:
-                    # 如果没有找到结束标记，累积整个内容（可能是中间片段）
-                    self.tool_args += edit_content
-                    logger.debug(f"📦 累积参数片段: {edit_content[:100]}...")
-
-    def _handle_glm_blocks(self, edit_content: str) -> Generator[str, None, None]:
-        """处理 glm_block 标记的内容"""
-        blocks = edit_content.split('<glm_block ')
-        logger.debug(f"📦 分割得到 {len(blocks)} 个块")
-
-        for index, block in enumerate(blocks):
-            if not block.strip():
-                continue
-
-            if index == 0:
-                # 第一个块：提取参数片段
-                if self.has_tool_call:
-                    logger.debug(f"📦 从第一个块提取参数片段")
-                    # 找到 "result" 的位置，提取之前的参数片段
-                    result_pos = edit_content.find('"result"')
-                    if result_pos > 0:
-                        # 往前退3个字符去掉 ", "
-                        param_fragment = edit_content[:result_pos - 3]
-                        self.tool_args += param_fragment
-                        logger.debug(f"📦 累积参数片段: {param_fragment}")
-                else:
-                    # 没有活跃工具调用，跳过第一个块
-                    continue
-            else:
-                # 后续块：处理新工具调用
-                if "</glm_block>" not in block:
-                    continue
-
-                # 如果有活跃的工具调用，先完成它
-                if self.has_tool_call:
-                    # 补全参数并完成工具调用
-                    self.tool_args += '"'  # 补全最后的引号
-                    yield from self._finish_current_tool()
-
-                # 处理新工具调用
-                yield from self._process_metadata_block(block)
-
-    def _process_metadata_block(self, block: str) -> Generator[str, None, None]:
-        """处理包含工具元数据的块"""
-        try:
-            # 提取 JSON 内容
-            start_pos = block.find('>')
-            end_pos = block.rfind('</glm_block>')
-
-            if start_pos == -1 or end_pos == -1:
-                logger.warning(f"❌ 无法找到 JSON 内容边界: {block[:1000]}...")
-                return
-
-            json_content = block[start_pos + 1:end_pos]
-            logger.debug(f"📦 提取的 JSON 内容: {json_content[:1000]}...")
-
-            # 解析工具元数据
-            metadata_obj = json.loads(json_content)
-
-            if "data" in metadata_obj and "metadata" in metadata_obj["data"]:
-                metadata = metadata_obj["data"]["metadata"]
-
-                # 开始新的工具调用
-                self.tool_id = metadata.get("id", f"call_{int(time.time() * 1000000)}")
-                self.tool_name = metadata.get("name", "unknown")
-                self.has_tool_call = True
-
-                # 只有在这是第二个及以后的工具调用时才递增 index
-                # 第一个工具调用应该使用 index 0
-
-                # 从 metadata.arguments 获取参数起始部分
-                if "arguments" in metadata:
-                    arguments_str = metadata["arguments"]
-                    # 去掉最后一个字符
-                    self.tool_args = arguments_str[:-1] if arguments_str.endswith('"') else arguments_str
-                    logger.debug(f"🎯 新工具调用: {self.tool_name}(id={self.tool_id}), 初始参数: {self.tool_args}")
-                else:
-                    self.tool_args = "{}"
-                    logger.debug(f"🎯 新工具调用: {self.tool_name}(id={self.tool_id}), 空参数")
-
-        except (json.JSONDecodeError, KeyError, AttributeError) as e:
-            logger.error(f"❌ 解析工具元数据失败: {e}, 块内容: {block[:1000]}...")
-
-        # 确保返回生成器（即使为空）
-        if False:  # 永远不会执行，但确保函数是生成器
-            yield
-
-    def _process_other_phase(self, usage: Dict[str, Any], edit_content: str = "") -> Generator[str, None, None]:
-        """处理其他阶段"""
-        # 保存使用统计信息
-        if usage:
-            self.tool_call_usage = usage
-            logger.debug(f"📊 保存使用统计: {usage}")
-
-        # 工具调用完成判断：检测到 "null," 开头的 edit_content
-        if self.has_tool_call and edit_content and edit_content.startswith("null,"):
-            logger.info(f"🏁 检测到工具调用结束标记")
-
-            # 完成当前工具调用
-            yield from self._finish_current_tool()
-
-            # 发��流结束标记
-            if self.stream:
-                yield "data: [DONE]\n\n"
-
-            # 重置状态
-            self._reset_all_state()
-
-    def _process_answer_phase(self, delta_content: str) -> Generator[str, None, None]:
-        """处理回答阶段（优化版本）"""
-        if not delta_content:
-            return
-
-        logger.info(f"📝 工具处理器收到答案内容: {delta_content[:50]}...")
-
-        # 添加到缓冲区
-        self.content_buffer += delta_content
-        self.buffer_size += len(delta_content)
-
-        current_time = time.time()
-        time_since_last_flush = current_time - self.last_flush_time
-
-        # 检查是否需要刷新缓冲区
-        should_flush = (
-            self.buffer_size >= self.max_buffer_size or  # 缓冲区满了
-            time_since_last_flush >= self.flush_interval or  # 时间间隔到了
-            '\n' in delta_content or  # 包含换行符
-            '。' in delta_content or '！' in delta_content or '？' in delta_content  # 包含句子结束符
-        )
-
-        if should_flush and self.content_buffer:
-            yield from self._flush_content_buffer()
-
-    def _flush_content_buffer(self) -> Generator[str, None, None]:
-        """刷新内容缓冲区"""
-        if not self.content_buffer:
-            return
-
-        logger.info(f"💬 工具处理器刷新缓冲区: {self.buffer_size} 字符 - {self.content_buffer[:50]}...")
-
-        if self.stream:
-            chunk = self._create_content_chunk(self.content_buffer)
-            output_data = f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
-            logger.info(f"➡️ 工具处理器输出: {output_data[:100]}...")
-            yield output_data
-
-        # 清空缓冲区
-        self.content_buffer = ""
-        self.buffer_size = 0
-        self.last_flush_time = time.time()
-
-    def _process_done_phase(self, chunk_data: Dict[str, Any]) -> Generator[str, None, None]:
-        """处理完成阶段"""
-        logger.info("🏁 对话完成")
-
-        # 先刷新任何剩余的缓冲内容
-        if self.content_buffer:
-            yield from self._flush_content_buffer()
-
-        # 完成任何未完成的工具调用
-        if self.has_tool_call:
-            yield from self._finish_current_tool()
-
-        # 发送流结束标记
-        if self.stream:
-            # 创建最终的完成块
-            final_chunk = {
-                "id": f"chatcmpl-{int(time.time())}",
-                "object": "chat.completion.chunk",
-                "created": int(time.time()),
-                "model": self.model,
-                "choices": [{
-                    "index": 0,
-                    "delta": {},
-                    "finish_reason": "stop"
-                }]
-            }
-
-            # 如果有 usage 信息，添加到最终块中
-            if "usage" in chunk_data:
-                final_chunk["usage"] = chunk_data["usage"]
-
-            yield f"data: {json.dumps(final_chunk, ensure_ascii=False)}\n\n"
-            yield "data: [DONE]\n\n"
-
-        # 重置所有状态
-        self._reset_all_state()
-
-    def _finish_current_tool(self) -> Generator[str, None, None]:
-        """完成当前工具调用"""
-        if not self.has_tool_call:
-            return
-
-        # 修复参数格式
-        fixed_args = self._fix_tool_arguments(self.tool_args)
-        logger.debug(f"✅ 完成工具调用: {self.tool_name}, 参数: {fixed_args}")
-
-        # 输出工具调用（开始 + 参数 + 完成）
-        if self.stream:
-            # 发送工具开始块
-            start_chunk = self._create_tool_start_chunk()
-            yield f"data: {json.dumps(start_chunk, ensure_ascii=False)}\n\n"
-
-            # 发送参数块
-            args_chunk = self._create_tool_arguments_chunk(fixed_args)
-            yield f"data: {json.dumps(args_chunk, ensure_ascii=False)}\n\n"
-
-            # 发送完成块
-            finish_chunk = self._create_tool_finish_chunk()
-            yield f"data: {json.dumps(finish_chunk, ensure_ascii=False)}\n\n"
-
-        # 重置工具状态
-        self._reset_tool_state()
-
-    def _fix_tool_arguments(self, raw_args: str) -> str:
-        """使用 json-repair 库修复工具参数格式"""
-        if not raw_args or raw_args == "{}":
-            return "{}"
-
-        logger.debug(f"🔧 开始修复参数: {raw_args[:1000]}{'...' if len(raw_args) > 1000 else ''}")
-
-        # 统一的修复流程：预处理 -> json-repair -> 后处理
-        try:
-            # 1. 预处理：只处理 json-repair 无法处理的问题
-            processed_args = self._preprocess_json_string(raw_args.strip())
-
-            # 2. 使用 json-repair 进行主要修复
-            from json_repair import repair_json
-            repaired_json = repair_json(processed_args)
-            logger.debug(f"🔧 json-repair 修复结果: {repaired_json}")
-
-            # 3. 解析并后处理
-            args_obj = json.loads(repaired_json)
-            args_obj = self._post_process_args(args_obj)
-
-            # 4. 生成最终结果
-            fixed_result = json.dumps(args_obj, ensure_ascii=False)
-
-            return fixed_result
-
-        except Exception as e:
-            logger.error(f"❌ JSON 修复失败: {e}, 原始参数: {raw_args[:1000]}..., 使用空参数")
-            return "{}"
-
-    def _post_process_args(self, args_obj: Dict[str, Any]) -> Dict[str, Any]:
-        """统一的后处理方法"""
-        # 修复路径中的过度转义
-        args_obj = self._fix_path_escaping_in_args(args_obj)
-
-        # 修复命令中的多余引号
-        args_obj = self._fix_command_quotes(args_obj)
-
-        return args_obj
-
-    def _preprocess_json_string(self, text: str) -> str:
-        """预处理 JSON 字符串，只处理 json-repair 无法处理的问题"""
-        import re
-
-        # 只保留 json-repair 无法处理的预处理步骤
-
-        # 1. 修复缺少开始括号的情况（json-repair 无法处理）
-        if not text.startswith('{') and text.endswith('}'):
-            text = '{' + text
-            logger.debug(f"🔧 补全开始括号")
-
-        # 2. 修复末尾多余的反斜杠和引号（json-repair 可能处理不当）
-        # 匹配模式：字符串值末尾的 \" 后面跟着 } 或 ,
-        # 例如：{"url":"https://www.bilibili.com\"} -> {"url":"https://www.bilibili.com"}
-        # 例如：{"url":"https://www.bilibili.com\",} -> {"url":"https://www.bilibili.com",}
-        pattern = r'([^\\])\\"([}\s,])'
-        if re.search(pattern, text):
-            text = re.sub(pattern, r'\1"\2', text)
-            logger.debug(f"🔧 修复末尾多余的反斜杠")
-
-        return text
-
-    def _fix_path_escaping_in_args(self, args_obj: Dict[str, Any]) -> Dict[str, Any]:
-        """修复参数对象中路径的过度转义问题"""
-        import re
-
-        # 需要检查的路径字段
-        path_fields = ['file_path', 'path', 'directory', 'folder']
-
-        for field in path_fields:
-            if field in args_obj and isinstance(args_obj[field], str):
-                path_value = args_obj[field]
-
-                # 检查是否是Windows路径且包含过度转义
-                if path_value.startswith('C:') and '\\\\' in path_value:
-                    logger.debug(f"🔍 检查路径字段 {field}: {repr(path_value)}")
-
-                    # 分析路径结构：正常路径应该是 C:\Users\...
-                    # 但过度转义的路径可能是 C:\Users\\Documents（多了一个反斜杠）
-                    # 我们需要找到不正常的双反斜杠模式并修复
-
-                    # 先检查是否有不正常的双反斜杠（不在路径开头）
-                    # 正常：C:\Users\Documents
-                    # 异常：C:\Users\\Documents 或 C:\Users\\\\Documents
-
-                    # 使用更精确的模式：匹配路径分隔符后的额外反斜杠
-                    # 但要保留正常的路径分隔符
-                    fixed_path = path_value
-
-                    # 检查是否有连续的多个反斜杠（超过正常的路径分隔符）
-                    if '\\\\' in path_value:
-                        # 计算反斜杠的数量，如果超过正常数量就修复
-                        parts = path_value.split('\\')
-                        # 重新组装路径，去除空的部分（由多余的反斜杠造成）
-                        clean_parts = [part for part in parts if part]
-                        if len(clean_parts) > 1:
-                            fixed_path = '\\'.join(clean_parts)
-
-                    logger.debug(f"🔍 修复后路径: {repr(fixed_path)}")
-
-                    if fixed_path != path_value:
-                        args_obj[field] = fixed_path
-                        logger.debug(f"🔧 修复字段 {field} 的路径转义: {path_value} -> {fixed_path}")
-                    else:
-                        logger.debug(f"🔍 路径无需修复: {path_value}")
-
-        return args_obj
-
-    def _fix_command_quotes(self, args_obj: Dict[str, Any]) -> Dict[str, Any]:
-        """修复命令中的多余引号问题"""
-        import re
-
-        # 检查命令字段
-        if 'command' in args_obj and isinstance(args_obj['command'], str):
-            command = args_obj['command']
-
-            # 检查是否以双引号结尾（多余的引号）
-            if command.endswith('""'):
-                logger.debug(f"🔧 发现命令末尾多余引号: {command}")
-                # 移除最后一个多余的引号
-                fixed_command = command[:-1]
-                args_obj['command'] = fixed_command
-                logger.debug(f"🔧 修复命令引号: {command} -> {fixed_command}")
-
-            # 检查其他可能的引号问题
-            # 例如：路径末尾的 \"" 模式
-            elif re.search(r'\\""+$', command):
-                logger.debug(f"🔧 发现命令末尾引号模式问题: {command}")
-                # 修复路径末尾的引号问题
-                fixed_command = re.sub(r'\\""+$', '\\"', command)
-                args_obj['command'] = fixed_command
-                logger.debug(f"🔧 修复命令引号模式: {command} -> {fixed_command}")
-
-        return args_obj
-
-    def _create_content_chunk(self, content: str) -> Dict[str, Any]:
-        """创建内容块"""
-        return {
-            "id": f"chatcmpl-{int(time.time())}",
-            "object": "chat.completion.chunk",
-            "created": int(time.time()),
-            "model": self.model,
-            "choices": [{
-                "index": 0,
-                "delta": {
-                    "role": "assistant",
-                    "content": content
-                },
-                "finish_reason": None
-            }]
-        }
-
-    def _create_tool_start_chunk(self) -> Dict[str, Any]:
-        """创建工具开始块"""
-        return {
-            "id": f"chatcmpl-{int(time.time())}",
-            "object": "chat.completion.chunk", 
-            "created": int(time.time()),
-            "model": self.model,
-            "choices": [{
-                "index": 0,
-                "delta": {
-                    "role": "assistant",
-                    "tool_calls": [{
-                        "index": self.content_index,
-                        "id": self.tool_id,
-                        "type": "function",
-                        "function": {
-                            "name": self.tool_name,
-                            "arguments": ""
-                        }
-                    }]
-                },
-                "finish_reason": None
-            }]
-        }
-
-    def _create_tool_arguments_chunk(self, arguments: str) -> Dict[str, Any]:
-        """创建工具参数块"""
-        return {
-            "id": f"chatcmpl-{int(time.time())}",
-            "object": "chat.completion.chunk",
-            "created": int(time.time()),
-            "model": self.model,
-            "choices": [{
-                "index": 0,
-                "delta": {
-                    "tool_calls": [{
-                        "index": self.content_index,
-                        "id": self.tool_id,
-                        "function": {
-                            "arguments": arguments
-                        }
-                    }]
-                },
-                "finish_reason": None
-            }]
-        }
-
-    def _create_tool_finish_chunk(self) -> Dict[str, Any]:
-        """创建工具完成块"""
-        chunk = {
-            "id": f"chatcmpl-{int(time.time())}",
-            "object": "chat.completion.chunk",
-            "created": int(time.time()),
-            "model": self.model,
-            "choices": [{
-                "index": 0,
-                "delta": {
-                    "tool_calls": []
-                },
-                "finish_reason": "tool_calls"
-            }]
-        }
-        
-        # 添加使用统计（如果有）
-        if self.tool_call_usage:
-            chunk["usage"] = self.tool_call_usage
-            
-        return chunk
-
-    def _reset_tool_state(self):
-        """重置工具状态"""
-        self.tool_id = ""
-        self.tool_name = ""
-        self.tool_args = ""
-        self.has_tool_call = False
-        # content_index 在单次对话中应该保持不变，只有在新的工具调用开始时才递增
-
-    def _reset_all_state(self):
-        """重置所有状态"""
-        # 先刷新任何剩余的缓冲内容
-        if hasattr(self, 'content_buffer') and self.content_buffer:
-            list(self._flush_content_buffer())  # 消费生成器
-
-        self._reset_tool_state()
-        self.current_phase = None
-        self.tool_call_usage = {}
-
-        # 重置缓冲区
-        self.content_buffer = ""
-        self.buffer_size = 0
-        self.last_flush_time = time.time()
-
-        # content_index 重置为 0，为下一轮对话做准备
-        self.content_index = 0
-        logger.debug("🔄 重置所有处理器状态")

deploy/.env.example

@@ -1,35 +0,0 @@
-# ==============================================
-# Z.AI API Server - Docker 环境变量配置示例
-# ==============================================
-
-# 管理后台密码
-ADMIN_PASSWORD=admin123
-
-# API 认证密钥 (用于验证客户端请求)
-AUTH_TOKEN=sk-your-api-key-here
-
-# 是否跳过 API Key 验证 (开发环境可设为 true)
-SKIP_AUTH_TOKEN=false
-
-# 调试日志 (生产环境建议设为 false)
-DEBUG_LOGGING=true
-
-# 匿名模式 (允许无 token 访问,需要配合 SKIP_AUTH_TOKEN=true)
-ANONYMOUS_MODE=false
-
-# Function Call 功能开关 (是否支持工具调用)
-TOOL_SUPPORT=true
-
-# 工具调用扫描限制 (字符数)
-SCAN_LIMIT=200000
-
-# 数据库路径 (Docker 环境使用持久化卷)
-DB_PATH=/app/data/tokens.db
-
-# Token 池配置
-TOKEN_FAILURE_THRESHOLD=3
-TOKEN_RECOVERY_TIMEOUT=300
-
-# 服务配置
-SERVICE_NAME=Z.AI_API_Server
-LISTEN_PORT=7860

deploy/Dockerfile

@@ -1,24 +0,0 @@
-FROM python:3.12-slim
-
-# Set working directory
-WORKDIR /app
-
-# Create data and logs directories with proper permissions
-RUN mkdir -p /app/data /app/logs && \
-    chmod 755 /app/data /app/logs
-
-# Install dependencies
-COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
-
-# Copy application code
-COPY . .
-
-# Set environment variable for database path
-ENV DB_PATH=/app/data/tokens.db
-
-# Expose port
-EXPOSE 7860
-
-# Run the application
-CMD ["python", "main.py"]

tests/test_signature.py

@@ -0,0 +1,19 @@
+import time
+import sys
+import os
+
+# 添加项目根目录到 Python 路径
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.utils.signature import generate_signature
+
+
+if __name__ == "__main__":
+    # 示例用法
+    e_value = "requestId,eef12d6c-6dc9-47a0-aae8-b9f3454f98c5,timestamp,1761038714733,user_id,21ea9ec3-e492-4dbb-b522-fc0eaf64f0f6"
+    t_value = "hi"
+    # r_value = int(time.time() * 1000)
+    r_value = 1761038714733
+    result = generate_signature(e_value, t_value, r_value)
+    print(f"生成的签名: {result['signature']}")
+    print(f"时间戳: {result['timestamp']}")

tests/test_simple_signature.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+简单测试签名工具
+"""
+
+import sys
+import os
+
+# 添加项目根目录到 Python 路径
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+# 直接导入签名模块，避免导入整个应用
+import importlib.util
+spec = importlib.util.spec_from_file_location("signature", os.path.join(os.path.dirname(os.path.dirname(__file__)), "app/utils/signature.py"))
+signature_module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(signature_module)
+
+generate_signature = signature_module.generate_signature
+
+
+if __name__ == "__main__":
+    # 示例用法
+    e_value = "requestId,eef12d6c-6dc9-47a0-aae8-b9f3454f98c5,timestamp,1761038714733,user_id,21ea9ec3-e492-4dbb-b522-fc0eaf64f0f6"
+    t_value = "hi"
+    r_value = 1761038714733
+    result = generate_signature(e_value, t_value, r_value)
+    print(f"生成的签名: {result['signature']}")
+    print(f"时间戳: {result['timestamp']}")
+    
+    # 验证函数是否正常工作
+    assert "signature" in result
+    assert "timestamp" in result
+    assert result["timestamp"] == str(r_value)
+    print("签名函数测试通过！")