backend/utils/text_client.py

"""Text API 客户端封装"""
import time
import random
import base64
import requests
from functools import wraps
from typing import List, Optional, Union
from .image_compressor import compress_image


def retry_on_429(max_retries=3, base_delay=2):
    """429 错误自动重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    error_str = str(e)
                    if "429" in error_str or "rate" in error_str.lower():
                        if attempt < max_retries - 1:
                            wait_time = (base_delay ** attempt) + random.uniform(0, 1)
                            print(f"[重试] 遇到限流，{wait_time:.1f}秒后重试 (尝试 {attempt + 2}/{max_retries})")
                            time.sleep(wait_time)
                            continue
                    raise
            raise Exception(
                f"Text API 重试 {max_retries} 次后仍失败。\n"
                "可能原因：\n"
                "1. API持续限流或配额不足\n"
                "2. 网络连接持续不稳定\n"
                "3. API服务暂时不可用\n"
                "建议：稍后再试，或联系API服务提供商"
            )
        return wrapper
    return decorator


class TextChatClient:
    """Text API 客户端封装类"""

    def __init__(self, api_key: str = None, base_url: str = None, endpoint_type: str = None):
        self.api_key = api_key
        if not self.api_key:
            raise ValueError(
                "Text API Key 未配置。\n"
                "解决方案：在系统设置页面编辑文本生成服务商，填写 API Key"
            )

        self.base_url = (base_url or "https://api.openai.com").rstrip('/').rstrip('/v1')

        # 支持自定义端点路径
        endpoint = endpoint_type or '/v1/chat/completions'
        # 确保端点以 / 开头
        if not endpoint.startswith('/'):
            endpoint = '/' + endpoint
        self.chat_endpoint = f"{self.base_url}{endpoint}"

    def _encode_image_to_base64(self, image_data: bytes) -> str:
        """将图片数据编码为 base64"""
        return base64.b64encode(image_data).decode('utf-8')

    def _build_content_with_images(
        self,
        text: str,
        images: List[Union[bytes, str]] = None
    ) -> Union[str, List[dict]]:
        """
        构建包含图片的 content

        Args:
            text: 文本内容
            images: 图片列表，可以是 bytes（图片数据）或 str（URL）

        Returns:
            如果没有图片，返回纯文本；有图片则返回多模态内容列表
        """
        if not images:
            return text

        content = [{"type": "text", "text": text}]

        for img in images:
            if isinstance(img, bytes):
                # 压缩图片到 200KB 以内
                compressed_img = compress_image(img, max_size_kb=200)
                # 图片数据，转为 base64 data URL
                base64_data = self._encode_image_to_base64(compressed_img)
                image_url = f"data:image/png;base64,{base64_data}"
            else:
                # 已经是 URL
                image_url = img

            content.append({
                "type": "image_url",
                "image_url": {"url": image_url}
            })

        return content

    @retry_on_429(max_retries=3, base_delay=2)
    def generate_text(
        self,
        prompt: str,
        model: str = "gemini-3-pro-preview",
        temperature: float = 1.0,
        max_output_tokens: int = 8000,
        images: List[Union[bytes, str]] = None,
        system_prompt: str = None,
        **kwargs
    ) -> str:
        """
        生成文本（支持图片输入）

        Args:
            prompt: 提示词
            model: 模型名称
            temperature: 温度
            max_output_tokens: 最大输出 token
            images: 图片列表（可选）
            system_prompt: 系统提示词（可选）

        Returns:
            生成的文本
        """
        messages = []

        # 添加系统提示词
        if system_prompt:
            messages.append({
                "role": "system",
                "content": system_prompt
            })

        # 构建用户消息内容
        content = self._build_content_with_images(prompt, images)
        messages.append({
            "role": "user",
            "content": content
        })

        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_output_tokens,
            "stream": False
        }

        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }

        response = requests.post(
            self.chat_endpoint,
            json=payload,
            headers=headers,
            timeout=300  # 5分钟超时
        )

        if response.status_code != 200:
            error_detail = response.text[:500]
            status_code = response.status_code

            # 根据状态码给出更详细的错误信息
            if status_code == 401:
                raise Exception(
                    "❌ API Key 认证失败\n\n"
                    "【可能原因】\n"
                    "1. API Key 无效或已过期\n"
                    "2. API Key 格式错误（复制时可能包含空格）\n"
                    "3. API Key 被禁用或删除\n\n"
                    "【解决方案】\n"
                    "1. 在系统设置页面检查 API Key 是否正确\n"
                    "2. 重新获取 API Key\n"
                    f"\n【请求地址】{self.chat_endpoint}"
                )
            elif status_code == 403:
                raise Exception(
                    "❌ 权限被拒绝\n\n"
                    "【可能原因】\n"
                    "1. API Key 没有访问该模型的权限\n"
                    "2. 账户配额已用尽\n"
                    "3. 区域限制\n\n"
                    "【解决方案】\n"
                    "1. 检查 API 权限配置\n"
                    "2. 尝试使用其他模型\n"
                    f"\n【原始错误】{error_detail[:200]}"
                )
            elif status_code == 404:
                raise Exception(
                    "❌ 模型不存在或 API 端点错误\n\n"
                    "【可能原因】\n"
                    f"1. 模型 '{model}' 不存在或已下线\n"
                    "2. Base URL 配置错误\n\n"
                    "【解决方案】\n"
                    "1. 检查模型名称是否正确\n"
                    "2. 检查 Base URL 配置\n"
                    f"\n【请求地址】{self.chat_endpoint}"
                )
            elif status_code == 429:
                raise Exception(
                    "⏳ API 配额或速率限制\n\n"
                    "【说明】\n"
                    "请求频率过高或配额已用尽。\n\n"
                    "【解决方案】\n"
                    "1. 稍后再试（等待 1-2 分钟）\n"
                    "2. 检查 API 配额使用情况\n"
                    "3. 考虑升级计划获取更多配额"
                )
            elif status_code >= 500:
                raise Exception(
                    f"⚠️ API 服务器错误 ({status_code})\n\n"
                    "【说明】\n"
                    "这是服务端的临时故障，与您的配置无关。\n\n"
                    "【解决方案】\n"
                    "1. 稍等几分钟后重试\n"
                    "2. 如果持续出现，检查服务商状态页"
                )
            else:
                raise Exception(
                    f"❌ API 请求失败 (状态码: {status_code})\n\n"
                    f"【原始错误】\n{error_detail}\n\n"
                    f"【请求地址】{self.chat_endpoint}\n"
                    f"【模型】{model}\n\n"
                    "【通用解决方案】\n"
                    "1. 检查 API Key 是否正确\n"
                    "2. 检查 Base URL 配置\n"
                    "3. 检查模型名称是否正确"
                )

        result = response.json()

        # 提取生成的文本
        if "choices" in result and len(result["choices"]) > 0:
            return result["choices"][0]["message"]["content"]
        else:
            raise Exception(
                f"Text API 响应格式异常：未找到生成的文本。\n"
                f"响应数据: {str(result)[:500]}\n"
                "可能原因：\n"
                "1. API返回格式与OpenAI标准不一致\n"
                "2. 请求被拒绝或过滤\n"
                "3. 模型输出为空\n"
                "建议：检查API文档确认响应格式"
            )


def get_text_chat_client(provider_config: dict):
    """
    获取 Text Chat 客户端实例（根据 type 返回对应客户端）

    Args:
        provider_config: 服务商配置字典
            - type: 'google_gemini' 或 'openai_compatible'
            - api_key: API密钥
            - base_url: API基础URL（可选）
            - endpoint_type: 自定义端点路径（可选）

    Returns:
        GenAIClient 或 TextChatClient
    """
    provider_type = provider_config.get('type', 'openai_compatible')
    api_key = provider_config.get('api_key')
    base_url = provider_config.get('base_url')
    endpoint_type = provider_config.get('endpoint_type')

    if provider_type == 'google_gemini':
        from .genai_client import GenAIClient
        return GenAIClient(api_key=api_key, base_url=base_url)
    else:
        return TextChatClient(api_key=api_key, base_url=base_url, endpoint_type=endpoint_type)