File size: 3,408 Bytes
f78578c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | #!/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)
|