✨ feat(zai): implement dynamic FE version management

- 添加动态获取 X-FE-Version 的功能
- 新增 fe_version.py 模块用于版本管理
- 实现版本缓存机制，减少网络请求
- 更新 ZAIProvider 使用动态版本号
- 移除硬编码的版本号
- 优化日志输出，移除冗余信息

Author: ZyphrZero

app/providers/zai_provider.py

@@ -20,6 +20,7 @@
 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.fe_version import get_latest_fe_version
 from app.utils.signature import generate_signature
 from app.providers.base import BaseProvider, ProviderConfig
 from app.models.schemas import OpenAIRequest, Message
@@ -42,6 +43,7 @@ def get_zai_dynamic_headers(chat_id: str = "") -> Dict[str, str]:
     browser_choices = ["chrome", "chrome", "chrome", "edge", "edge", "firefox", "safari"]
     browser_type = random.choice(browser_choices)
     user_agent = get_random_user_agent(browser_type)
+    fe_version = get_latest_fe_version()
 
     chrome_version = "139"
     edge_version = "139"
@@ -70,7 +72,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.106",
+        "X-FE-Version": fe_version,
         "Origin": "https://chat.z.ai",
     }
 
@@ -636,6 +638,7 @@ async def transform_request(self, request: OpenAIRequest) -> Dict[str, Any]:
         user_id = _extract_user_id_from_token(token)
         timestamp_ms = int(time.time() * 1000)
         request_id = generate_uuid()
+        fe_version = get_latest_fe_version()
         try:
             signing_metadata = f"requestId,{request_id},timestamp,{timestamp_ms},user_id,{user_id}"
             prompt_for_signature = last_user_text or ""
@@ -650,11 +653,11 @@ async def transform_request(self, request: OpenAIRequest) -> Dict[str, Any]:
             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-FE-Version": fe_version,
             "X-Signature": signature,
         }
 

app/utils/fe_version.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+Utility helpers for resolving the latest X-FE-Version value from chat.z.ai.
+
+The upstream service embeds the current front-end release identifier inside
+its landing page static asset URLs (e.g. `prod-fe-1.0.107`). The helpers in
+this module fetch the landing page, extract the version string, and cache it
+with a configurable TTL so the expensive network fetch only happens when
+necessary.
+"""
+
+from __future__ import annotations
+
+import re
+import time
+from typing import Optional
+
+import httpx
+
+from app.utils.logger import get_logger
+from app.utils.user_agent import get_random_user_agent
+
+# Base URL to probe for the version string.
+FE_VERSION_SOURCE_URL = "https://chat.z.ai"
+
+# Cache TTL in seconds (default: 30 minutes).
+CACHE_TTL_SECONDS = 1800
+
+_logger = get_logger()
+_version_pattern = re.compile(r"prod-fe-\d+\.\d+\.\d+")
+
+_cached_version: str = ""
+_cached_at: float = 0.0
+
+
+def _extract_version(page_content: str) -> Optional[str]:
+    """Extract the version string from the page content."""
+    if not page_content:
+        return None
+
+    matches = _version_pattern.findall(page_content)
+    if not matches:
+        return None
+
+    # Choose the highest lexical value to guard against mixed versions.
+    return max(matches)
+
+
+
+
+def _should_use_cache(force_refresh: bool) -> bool:
+    """Determine whether the cached value can be reused."""
+    if force_refresh:
+        return False
+    if not _cached_version:
+        return False
+    if _cached_at <= 0:
+        return False
+    return (time.time() - _cached_at) < CACHE_TTL_SECONDS
+
+
+def get_latest_fe_version(force_refresh: bool = False) -> str:
+    """
+    Resolve the latest X-FE-Version value from chat.z.ai.
+
+    The lookup order is:
+        1. Cached value within TTL.
+        2. Remote fetch from chat.z.ai.
+    
+    Raises:
+        Exception: If unable to fetch the version from the remote source.
+    """
+    global _cached_version, _cached_at
+
+    if _should_use_cache(force_refresh):
+        return _cached_version
+
+    try:
+        headers = {"User-Agent": get_random_user_agent("chrome")}
+    except Exception:
+        headers = {
+            "User-Agent": (
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/120.0.0.0 Safari/537.36"
+            )
+        }
+
+    try:
+        with httpx.Client(timeout=10.0, follow_redirects=True) as client:
+            response = client.get(FE_VERSION_SOURCE_URL, headers=headers)
+            response.raise_for_status()
+            version = _extract_version(response.text)
+            if version:
+                if version != _cached_version:
+                    _logger.info(f"[Z.AI] Detected X-FE-Version update: {version}")
+                _cached_version = version
+                _cached_at = time.time()
+                return version
+
+            _logger.error("[Z.AI] Unable to locate X-FE-Version in landing page")
+            raise Exception("Unable to locate X-FE-Version in landing page")
+    except Exception as exc:
+        _logger.error(f"[Z.AI] Failed to fetch X-FE-Version from {FE_VERSION_SOURCE_URL}: {exc}")
+        raise Exception(f"Failed to fetch X-FE-Version: {exc}")
+
+
+def refresh_fe_version() -> str:
+    """Force refresh the cached version by bypassing the TTL."""
+    return get_latest_fe_version(force_refresh=True)

app/utils/logger.py

@@ -56,7 +56,6 @@ def setup_logger(log_dir, log_retention_days=7, log_rotation="1 day", debug_mode
                 enqueue=True,
                 catch=True,
             )
-            logger.info(f"✅ 日志文件输出已启用: {log_dir}")
         except (PermissionError, OSError) as e:
             # 如果无法创建日志目录或文件，降级为仅控制台输出
             logger.warning(f"⚠️ 无法创建日志文件 ({e})，将仅使用控制台输出")