From 90fc01680765a87b65a0db4f26f1e80ca896da93 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Wed, 11 Feb 2026 11:32:54 +0800 Subject: [PATCH] merge: remove unused header, retry, and Statsig utility files --- app/services/grok/utils/headers.py | 31 ---- app/services/grok/utils/retry.py | 264 ----------------------------- app/services/grok/utils/statsig.py | 51 ------ app/services/reverse/app_chat.py | 15 +- 4 files changed, 13 insertions(+), 348 deletions(-) delete mode 100644 app/services/grok/utils/headers.py delete mode 100644 app/services/grok/utils/retry.py delete mode 100644 app/services/grok/utils/statsig.py diff --git a/app/services/grok/utils/headers.py b/app/services/grok/utils/headers.py deleted file mode 100644 index 7a5e1c2..0000000 --- a/app/services/grok/utils/headers.py +++ /dev/null @@ -1,31 +0,0 @@ -""" -Common header helpers for Grok services. -""" - -from __future__ import annotations - -import uuid -from typing import Dict - -from app.core.config import get_config -from app.services.grok.utils.statsig import StatsigService - - -def _normalize_token(token: str) -> str: - return token[4:] if token.startswith("sso=") else token - - -def build_sso_cookie(token: str, include_rw: bool = False) -> str: - token = _normalize_token(token) - cf = get_config("security.cf_clearance") - cookie = f"sso={token}" - if include_rw: - cookie = f"{cookie}; sso-rw={token}" - if cf: - cookie = f"{cookie};cf_clearance={cf}" - return cookie - - -def apply_statsig(headers: Dict[str, str]) -> None: - headers["x-statsig-id"] = StatsigService.gen_id() - headers["x-xai-request-id"] = str(uuid.uuid4()) diff --git a/app/services/grok/utils/retry.py b/app/services/grok/utils/retry.py deleted file mode 100644 index 162c4f8..0000000 --- a/app/services/grok/utils/retry.py +++ /dev/null @@ -1,264 +0,0 @@ -""" -Grok API 重试工具 - -提供可配置的重试机制，支持: -- 指数退避 + decorrelated jitter -- Retry-After header 支持 -- 429 专用退避策略 -- 重试预算控制 -""" - -import asyncio -import random -from typing import Callable, Any, Optional -from functools import wraps - -from app.core.logger import logger -from app.core.config import get_config -from app.core.exceptions import UpstreamException - - -class RetryContext: - """重试上下文""" - - def __init__(self): - self.attempt = 0 - self.max_retry = int(get_config("retry.max_retry")) - self.retry_codes = get_config("retry.retry_status_codes") - self.last_error = None - self.last_status = None - self.total_delay = 0.0 - self.retry_budget = float(get_config("retry.retry_budget")) - - # 退避参数 - self.backoff_base = float(get_config("retry.retry_backoff_base")) - self.backoff_factor = float(get_config("retry.retry_backoff_factor")) - self.backoff_max = float(get_config("retry.retry_backoff_max")) - - # decorrelated jitter 状态 - self._last_delay = self.backoff_base - - def should_retry(self, status_code: int) -> bool: - """判断是否重试""" - if self.attempt >= self.max_retry: - return False - if status_code not in self.retry_codes: - return False - if self.total_delay >= self.retry_budget: - return False - return True - - def record_error(self, status_code: int, error: Exception): - """记录错误信息""" - self.last_status = status_code - self.last_error = error - self.attempt += 1 - - def calculate_delay( - self, status_code: int, retry_after: Optional[float] = None - ) -> float: - """ - 计算退避延迟时间 - - Args: - status_code: HTTP 状态码 - retry_after: Retry-After header 值(秒) - - Returns: - 延迟时间(秒) - """ - # 优先使用 Retry-After - if retry_after is not None and retry_after > 0: - delay = min(retry_after, self.backoff_max) - self._last_delay = delay - return delay - - # 429 使用 decorrelated jitter - if status_code == 429: - # decorrelated jitter: delay = random(base, last_delay * 3) - delay = random.uniform(self.backoff_base, self._last_delay * 3) - delay = min(delay, self.backoff_max) - self._last_delay = delay - return delay - - # 其他状态码使用指数退避 + full jitter - exp_delay = self.backoff_base * (self.backoff_factor**self.attempt) - delay = random.uniform(0, min(exp_delay, self.backoff_max)) - return delay - - def record_delay(self, delay: float): - """记录延迟时间""" - self.total_delay += delay - - -def extract_retry_after(error: Exception) -> Optional[float]: - """ - 从异常中提取 Retry-After 值 - - Args: - error: 异常对象 - - Returns: - Retry-After 秒数，或 None - """ - if not isinstance(error, UpstreamException): - return None - - details = error.details or {} - - # 尝试从 details 中获取 - retry_after = details.get("retry_after") - if retry_after is not None: - try: - return float(retry_after) - except (ValueError, TypeError): - pass - - # 尝试从 headers 中获取 - headers = details.get("headers", {}) - if isinstance(headers, dict): - retry_after = headers.get("Retry-After") or headers.get("retry-after") - if retry_after is not None: - try: - return float(retry_after) - except (ValueError, TypeError): - pass - - return None - - -async def retry_on_status( - func: Callable, - *args, - extract_status: Callable[[Exception], Optional[int]] = None, - on_retry: Callable[[int, int, Exception, float], None] = None, - **kwargs, -) -> Any: - """ - 通用重试函数 - - Args: - func: 重试的异步函数 - *args: 函数参数 - extract_status: 异常提取状态码的函数 - on_retry: 重试时的回调函数 (attempt, status_code, error, delay) - **kwargs: 函数关键字参数 - - Returns: - 函数执行结果 - - Raises: - 最后一次失败的异常 - """ - ctx = RetryContext() - - # 状态码提取器 - if extract_status is None: - - def extract_status(e: Exception) -> Optional[int]: - if isinstance(e, UpstreamException): - # 优先从 details 获取，回退到 status_code 属性 - if e.details and "status" in e.details: - return e.details["status"] - return getattr(e, "status_code", None) - return None - - while ctx.attempt <= ctx.max_retry: - try: - result = await func(*args, **kwargs) - - # 记录日志 - if ctx.attempt > 0: - logger.info( - f"Retry succeeded after {ctx.attempt} attempts, " - f"total delay: {ctx.total_delay:.2f}s" - ) - - return result - - except Exception as e: - # 提取状态码 - status_code = extract_status(e) - - if status_code is None: - # 错误无法识别 - logger.error(f"Non-retryable error: {e}") - raise - - # 记录错误 - ctx.record_error(status_code, e) - - # 判断是否重试 - if ctx.should_retry(status_code): - # 提取 Retry-After - retry_after = extract_retry_after(e) - - # 计算延迟 - delay = ctx.calculate_delay(status_code, retry_after) - - # 检查是否超出预算 - if ctx.total_delay + delay > ctx.retry_budget: - logger.warning( - f"Retry budget exhausted: {ctx.total_delay:.2f}s + {delay:.2f}s > {ctx.retry_budget}s" - ) - raise - - ctx.record_delay(delay) - - logger.warning( - f"Retry {ctx.attempt}/{ctx.max_retry} for status {status_code}, " - f"waiting {delay:.2f}s (total: {ctx.total_delay:.2f}s)" - + (f", Retry-After: {retry_after}s" if retry_after else "") - ) - - # 回调 - if on_retry: - on_retry(ctx.attempt, status_code, e, delay) - - await asyncio.sleep(delay) - continue - else: - # 不可重试或重试次数耗尽 - if status_code in ctx.retry_codes: - logger.error( - f"Retry exhausted after {ctx.attempt} attempts, " - f"last status: {status_code}, total delay: {ctx.total_delay:.2f}s" - ) - else: - logger.error(f"Non-retryable status code: {status_code}") - - # 抛出最后一次的错误 - raise - - -def with_retry( - extract_status: Callable[[Exception], Optional[int]] = None, - on_retry: Callable[[int, int, Exception, float], None] = None, -): - """ - 重试装饰器 - - Usage: - @with_retry() - async def my_api_call(): - ... - """ - - def decorator(func: Callable): - @wraps(func) - async def wrapper(*args, **kwargs): - return await retry_on_status( - func, *args, extract_status=extract_status, on_retry=on_retry, **kwargs - ) - - return wrapper - - return decorator - - -__all__ = [ - "RetryContext", - "retry_on_status", - "with_retry", - "extract_retry_after", -] diff --git a/app/services/grok/utils/statsig.py b/app/services/grok/utils/statsig.py deleted file mode 100644 index c2cd15f..0000000 --- a/app/services/grok/utils/statsig.py +++ /dev/null @@ -1,51 +0,0 @@ -""" -Statsig ID 生成服务 -""" - -import base64 -import random -import string - -from app.core.config import get_config - - -class StatsigService: - """Statsig ID 生成服务""" - - @staticmethod - def _rand(length: int, alphanumeric: bool = False) -> str: - """生成随机字符串""" - chars = ( - string.ascii_lowercase + string.digits - if alphanumeric - else string.ascii_lowercase - ) - return "".join(random.choices(chars, k=length)) - - @staticmethod - def gen_id() -> str: - """ - 生成 Statsig ID - - Returns: - Base64 编码的 ID - """ - dynamic = get_config("chat.dynamic_statsig") - - if not dynamic: - return "ZTpUeXBlRXJyb3I6IENhbm5vdCByZWFkIHByb3BlcnRpZXMgb2YgdW5kZWZpbmVkIChyZWFkaW5nICdjaGlsZE5vZGVzJyk=" - - # 随机格式 - if random.choice([True, False]): - rand = StatsigService._rand(5, alphanumeric=True) - message = f"e:TypeError: Cannot read properties of null (reading 'children['{rand}']')" - else: - rand = StatsigService._rand(10) - message = ( - f"e:TypeError: Cannot read properties of undefined (reading '{rand}')" - ) - - return base64.b64encode(message.encode()).decode() - - -__all__ = ["StatsigService"] diff --git a/app/services/reverse/app_chat.py b/app/services/reverse/app_chat.py index 9d3c28b..77ce685 100644 --- a/app/services/reverse/app_chat.py +++ b/app/services/reverse/app_chat.py @@ -3,7 +3,7 @@ Reverse interface: app chat conversations. """ import orjson -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional from curl_cffi.requests import AsyncSession from app.core.logger import logger @@ -165,7 +165,18 @@ class AppChatReverse: return response - response = await retry_on_status(_do_request) + def extract_status(e: Exception) -> Optional[int]: + if isinstance(e, UpstreamException): + if e.details and "status" in e.details: + status = e.details["status"] + else: + status = getattr(e, "status_code", None) + if status == 429: + return None + return status + return None + + response = await retry_on_status(_do_request, extract_status=extract_status) # Stream response async def stream_response():