xiaoxia-saas/packages/shared/ai_client.py

"""豆包大模型 API 客户端（共享层）.

API 和 Worker 两边共用。基于火山引擎方舟平台的 OpenAI 兼容接口。

使用方式：
    from packages.shared.ai_client import get_doubao_client

    client = get_doubao_client()
    if client.is_available:
        result = client.chat_completion(messages=[...])
"""

from __future__ import annotations

import logging
import os
import time
import uuid
from typing import Any, Optional

import httpx

from packages.shared.config import get_shared_settings

# 网络/超时类异常父类集合：覆盖 Timeout/Connect/Network/ReadTimeout/WriteTimeout/PoolTimeout
_HTTP_NETWORK_ERRORS = ()
try:
    _HTTP_NETWORK_ERRORS = (httpx.TimeoutException, httpx.NetworkError)
except Exception:
    _HTTP_NETWORK_ERRORS = (Exception,)

_HTTP_STATUS_ERROR = httpx.HTTPStatusError if hasattr(httpx, "HTTPStatusError") else Exception

logger = logging.getLogger(__name__)


# 视频模型 ID 解析逻辑（#2159 多模型支持，#2170 方舟信任链统一走方舟）。
# 内部使用简短别名（seedance-2.5 / seedance-2.0 / wan-3.0 等）做 PRICING key；
# 实际调用时按 VIRAL_VIDEO_MODEL_CONFIG（domain/points_rules.py）的 provider/model_id 分发。
# - provider=doubao → 火山方舟 Seedance（含信任链真人 AI 化）
# - provider=dashscope → 阿里云 DashScope（Wan 系列，可选）


def _resolve_video_provider_and_id(model: str | None) -> tuple[str, str, dict]:
    """把内部 model key 解析成 (provider, model_id, cfg)。

    - provider: "doubao" | "dashscope"
    - model_id: 对应 API 的真实模型 ID
    - cfg: VIRAL_VIDEO_MODEL_CONFIG 条目
    未识别或空值回落到默认 seedance-2.5。已经是 doubao-/ep- 开头的完整 ID 视为 doubao provider。
    """
    from packages.domain.points_rules import get_viral_video_model_config

    settings = get_shared_settings()
    default_id = getattr(settings, "doubao_video_model", None) or "doubao-seedance-2-5-260628"
    m = (model or "").strip()
    if not m:
        cfg = get_viral_video_model_config("seedance-2.5")
        return "doubao", default_id, cfg
    # 已经是 doubao-/ep- 开头：直接透传，默认视为 doubao provider
    if m.startswith("doubao-") or m.startswith("ep-"):
        return "doubao", m, {"provider": "doubao", "model_id": m, "supports_audio": True}
    # 别名 → 从 domain config 查
    cfg = get_viral_video_model_config(m)
    provider = cfg.get("provider", "doubao")
    resolved_id = cfg.get("model_id", "")
    if not resolved_id:
        logger.warning("[ai_client] model %r 无 model_id，回落到默认 %s", m, default_id)
        return "doubao", default_id, cfg
    return provider, resolved_id, cfg


def _resolve_video_model_id(model: str | None) -> str:
    """兼容旧调用：只返回 doubao model_id。wan/dashscope 调用方应直接用 _resolve_video_provider_and_id。"""
    _provider, mid, _cfg = _resolve_video_provider_and_id(model)
    return mid


# ── 视频错误分类（给前端/用户展示友好提示）────────────────────────────


def _classify_video_error(status_code: int, body: str, err: Exception | None) -> tuple[str, str]:
    """根据 HTTP 状态码和响应 body 判断错误类型。

    返回 (error_code, user_message)：
    - error_code: 机器可读的错误码（"portrait_intercept" / "quota_exceeded" / "model_not_found"
      / "invalid_param" / "auth_error" / "rate_limit" / "network_error" / "task_failed" / "unknown"）
    - user_message: 给用户看的中文提示
    """
    body_lower = (body or "").lower()
    code_in_body = ""
    msg_in_body = ""
    try:
        import json as _json

        parsed = _json.loads(body or "{}")
        if isinstance(parsed, dict):
            err_obj = parsed.get("error") or {}
            if isinstance(err_obj, dict):
                code_in_body = str(err_obj.get("code", "") or "")
                msg_in_body = str(err_obj.get("message", "") or err_obj.get("msg", "") or "")
            else:
                msg_in_body = str(parsed.get("message", "") or "")
    except Exception:
        pass

    # 真人肖像/内容安全拦截
    if (
        status_code == 400
        and any(
            kw in body_lower
            for kw in ("portrait", "real_face", "human_face", "真人", "肖像", "人脸", "privacy", "real person", "face")
        )
    ) or (
        "content" in body_lower
        and ("risk" in body_lower or "block" in body_lower or "reject" in body_lower)
        and status_code == 400
    ):
        return (
            "portrait_intercept",
            "参考素材包含真人照片被安全策略拦截，AI视频模型暂不支持上传真人照片作为参考图，请移除真人图片后重试。",
        )

    # 配额/计费问题
    if status_code in (402, 429) or any(
        kw in body_lower for kw in ("quota", "billing", "insufficient", "欠费", "余额", "限流", "rate limit")
    ):
        if "rate" in body_lower or status_code == 429:
            return "rate_limit", "视频生成服务当前繁忙（限流），请稍等1-2分钟后重试。"
        return "quota_exceeded", "视频生成服务配额不足，请联系管理员充值或稍后重试。"

    # 模型/Endpoint 不存在
    if status_code == 404 or any(
        kw in body_lower for kw in ("model not found", "endpoint not found", "不存在", "not found", "model_not_exist")
    ):
        return "model_not_found", f"视频模型未开通或模型ID无效（{code_in_body or ''}），请联系管理员。"

    # 鉴权失败
    if status_code in (401, 403):
        return "auth_error", "视频生成服务鉴权失败（API Key无效或过期），请联系管理员。"

    # 任务本身失败（轮询阶段拿到 status=failed）
    if err and "task failed" in str(err).lower():
        detail = msg_in_body or str(err)[:200]
        # 失败原因里再细分真人拦截
        if any(kw in detail.lower() for kw in ("portrait", "真人", "肖像", "人脸", "content_risk")):
            return (
                "portrait_intercept",
                "视频内容被安全策略拦截（疑似包含真人肖像），请更换参考图或调整文案后重试。",
            )
        return "task_failed", f"视频生成失败：{detail}"

    # 参数错误
    if status_code == 400:
        return "invalid_param", f"视频生成参数错误：{msg_in_body or body[:200]}"

    # 网络/连接问题
    if status_code == 0:
        return "network_error", "视频生成服务连接失败（网络超时），请稍后重试。"

    # 默认
    detail = msg_in_body or (str(err) if err else "") or body[:200]
    return "unknown", f"视频生成失败（HTTP {status_code}）：{detail}"


class DoubaoClient:
    """豆包大模型 API 客户端.

    封装 OpenAI 兼容的 Chat Completion 接口，支持自动重试。
    未配置 API Key 时 is_available 为 False，调用方应降级处理。
    """

    def __init__(self) -> None:
        settings = get_shared_settings()
        self.api_key: str = settings.doubao_api_key
        self.model: str = settings.doubao_model
        self.base_url: str = settings.doubao_base_url.rstrip("/")
        self.timeout: int = settings.doubao_timeout
        self.max_retries: int = settings.doubao_max_retries
        self.vision_model: str = settings.doubao_vision_model
        self.vision_lite_model: str = settings.doubao_vision_lite_model
        self.fast_model: str = settings.doubao_fast_model
        self.embedding_model: str = settings.doubao_embedding_model
        self.image_model: str = settings.doubao_image_model
        self.image_timeout: int = getattr(settings, "doubao_image_timeout", 120) or 120
        # 最近一次视频生成的详细错误（error_code + user_message + raw detail），供上层读取后展示给用户
        self.last_video_error: dict = {}
        # 最近一次图片生成的详细错误，供上层读取
        self.last_image_error: dict = {}

    def embed_text(self, text: str, timeout: int | None = None) -> list[float] | None:
        """调用豆包文本 Embedding API，返回浮点向量；失败返回 None。"""
        if not self.is_available or not text or not text.strip():
            return None

        url = f"{self.base_url}/embeddings"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload: dict[str, Any] = {
            "model": self.embedding_model,
            "input": text.strip(),
            "encoding_format": "float",
        }

        req_timeout = timeout or self.timeout
        last_error: Exception | None = None
        for attempt in range(self.max_retries + 1):
            try:
                resp = httpx.post(url, headers=headers, json=payload, timeout=req_timeout)
                resp.raise_for_status()
                data = resp.json()
                emb_list = data.get("data") or []
                if emb_list and isinstance(emb_list, list):
                    vec = emb_list[0].get("embedding")
                    if isinstance(vec, list) and vec:
                        return [float(x) for x in vec]
                logger.warning("embedding 返回结构异常: %s", str(data)[:200])
                return None
            except Exception as e:
                last_error = e
                if attempt < self.max_retries:
                    wait = 0.5 * (2**attempt)
                    logger.warning(
                        "豆包 Embedding 调用失败，%.1fs 后重试 (%d/%d): %s", wait, attempt + 1, self.max_retries + 1, e
                    )
                    time.sleep(wait)
        logger.error("豆包 Embedding 调用最终失败: %s", last_error)
        return None

    @property
    def is_available(self) -> bool:
        """是否可用（配置了 API Key）."""
        return bool(self.api_key)

    def chat_completion(
        self,
        messages: list[dict[str, str]],
        temperature: float = 0.7,
        max_tokens: int = 1024,
        model: str | None = None,
    ) -> Optional[str]:
        """调用 Chat Completion 接口.

        Args:
            messages: 对话消息列表，[{"role": "user"/"system"/"assistant", "content": "..."}]
            temperature: 采样温度，0-2，默认0.7
            max_tokens: 最大生成token数，默认1024

        Returns:
            模型返回的文本内容，失败返回 None
        """
        if not self.is_available:
            return None

        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload: dict[str, Any] = {
            "model": model or self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }

        last_error: Optional[Exception] = None
        for attempt in range(self.max_retries + 1):
            try:
                response = httpx.post(
                    url,
                    headers=headers,
                    json=payload,
                    timeout=self.timeout,
                )
                response.raise_for_status()
                data = response.json()
                content = data["choices"][0]["message"]["content"]
                return content.strip()
            except Exception as e:
                last_error = e
                if attempt < self.max_retries:
                    wait = 0.5 * (2**attempt)
                    logger.warning(
                        "豆包API调用失败，%.1fs后重试 (第%d/%d次): %s",
                        wait,
                        attempt + 1,
                        self.max_retries + 1,
                        e,
                    )
                    time.sleep(wait)

        logger.error("豆包API调用最终失败: %s", last_error)
        return None

    def vision_completion(
        self,
        messages: list[dict],
        images: list[str] | None = None,
        max_tokens: int = 2048,
        temperature: float = 0.3,
        timeout: int | None = None,
        model: str | None = None,
    ) -> Optional[str]:
        """调用豆包视觉理解 API（OpenAI 兼容多模态格式）.

        将 images 附加到最后一条 user message 的 content 中，
        使用 vision_model（默认 doubao-1-5-vision-pro-250328）。

        Args:
            messages: 对话消息列表。最后一条 user message 会被注入图片内容。
            images: 图片列表，支持 base64 data URI 或 HTTP(S) URL。
            max_tokens: 最大生成 token 数，默认 2048。
            temperature: 采样温度，默认 0.3（视觉任务偏低更稳定）。
            timeout: 单次请求超时秒数，不传则使用默认 self.timeout。

        Returns:
            模型返回的文本内容，失败返回 None。
        """
        if not self.is_available:
            return None

        # 构造多模态 content：先追加文本，再追加图片
        vision_messages = []
        for msg in messages:
            vision_messages.append(dict(msg))

        # 将图片注入最后一条 user message
        if images and vision_messages:
            # 找到最后一条 user message
            for i in range(len(vision_messages) - 1, -1, -1):
                if vision_messages[i].get("role") == "user":
                    text_content = vision_messages[i].get("content", "")
                    multi_content: list[dict[str, Any]] = []
                    if text_content:
                        multi_content.append({"type": "text", "text": text_content})
                    for img in images:
                        if img.startswith("data:") or img.startswith("http://") or img.startswith("https://"):
                            multi_content.append({"type": "image_url", "image_url": {"url": img}})
                        else:
                            # 当作 base64 编码
                            multi_content.append(
                                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img}"}}
                            )
                    vision_messages[i]["content"] = multi_content
                    break

        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload: dict[str, Any] = {
            "model": model or self.vision_model,
            "messages": vision_messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }

        req_timeout = timeout or self.timeout
        last_error: Optional[Exception] = None
        for attempt in range(self.max_retries + 1):
            try:
                response = httpx.post(
                    url,
                    headers=headers,
                    json=payload,
                    timeout=req_timeout,
                )
                response.raise_for_status()
                data = response.json()
                content = data["choices"][0]["message"]["content"]
                return content.strip()
            except Exception as e:
                last_error = e
                if attempt < self.max_retries:
                    wait = 0.5 * (2**attempt)
                    logger.warning(
                        "豆包视觉API调用失败，%.1fs后重试 (第%d/%d次): %s",
                        wait,
                        attempt + 1,
                        self.max_retries + 1,
                        e,
                    )
                    time.sleep(wait)

        logger.error("豆包视觉API调用最终失败: %s", last_error)
        return None

    # ── 视频生成（Seedance 2.5，异步任务）────────────────────────────

    def video_generation(
        self,
        prompt: str,
        *,
        image_url: str | None = None,
        duration: int = 5,
        ratio: str | None = "9:16",
        resolution: str = "720p",
        generate_audio: bool = True,
        watermark: bool = False,
        output_dir: str | None = None,
        model: str | None = None,
        reference_images: list[str] | None = None,
        reference_audios: list[str] | None = None,
        reference_videos: list[str] | None = None,
    ) -> dict | None:
        """调用 Seedance 2.5 生视频（异步任务→轮询→下载）。

        成功返回 {"video_path": str, "usage": dict | None}，失败返回 None。
        失败时把详细错误信息（HTTP状态码、响应 body、分类后的用户提示）写入 self.last_video_error，
        上层可通过 get_last_video_error() 读取并展示给用户，不再笼统显示"返回为空"。

        【v1.6.1 修复】严格按官方 content 数组协议构造请求：
        - 所有参考（图/音/视）必须放进 content 数组并带 role 字段，不能放顶层 reference_audios/reference_videos（非官方字段，会被忽略或导致异常）。
        - 首帧图（first_frame 模式）Seedance 2.5 强制 ratio=adaptive；走 omni_reference（参考生视频）模式时才能指定 9:16/1:1 等具体比例。
          判定：传了参考音频/视频或 ≥1 张多参考图时，走 omni_reference（首张图 role=reference_image）；纯首帧无参考时走 first_frame（ratio 强制 adaptive）。
        - 创建任务若因 ratio 报错（HTTP 400），自动回退到 ratio=adaptive 重试一次。
        """
        # 每次调用前清空上次错误
        self.last_video_error = {}

        if not self.is_available:
            self.last_video_error = {
                "error_code": "auth_error",
                "user_message": "视频生成服务未配置（API Key 缺失），请联系管理员。",
                "status_code": 0,
                "detail": "DoubaoClient not available (api_key empty)",
            }
            return None
        if not prompt or not prompt.strip():
            self.last_video_error = {
                "error_code": "invalid_param",
                "user_message": "视频生成提示词不能为空。",
                "status_code": 0,
                "detail": "empty prompt",
            }
            return None

        settings = get_shared_settings()
        poll_interval = getattr(settings, "doubao_video_poll_interval", 10) or 10
        # 收紧总超时：轮询 8min + 下载 2min = 最长 ~10min，防止出现 20min 卡死
        total_timeout = getattr(settings, "doubao_video_timeout", 480) or 480
        # 内部 key → (provider, 实际模型 ID, cfg)，按 provider 分发
        provider, video_model, model_cfg = _resolve_video_provider_and_id(model)
        if provider == "dashscope":
            from packages.shared.dashscope_client import get_dashscope_client

            ds = get_dashscope_client()
            if ds is None:
                err_msg = "DashScope client 不可用（未配置 DASHSCOPE_API_KEY）"
                logger.error("%s, video_model=%s", err_msg, model)
                self.last_video_error = {
                    "error_code": "auth_error",
                    "user_message": "Wan 3.0 视频模型未配置 API Key，请联系管理员。",
                    "status_code": 0,
                    "detail": err_msg,
                }
                return None
            try:
                # DashScope 客户端也设置 last_video_error 语义（如果它支持）
                if hasattr(ds, "last_video_error"):
                    ds.last_video_error = {}
                result = ds.video_generation(
                    prompt=prompt,
                    image_url=image_url,
                    duration=duration,
                    ratio=ratio,
                    resolution=resolution,
                    output_dir=output_dir,
                    model=video_model,
                )
                if not result and hasattr(ds, "last_video_error") and ds.last_video_error:
                    self.last_video_error = dict(ds.last_video_error)
                return result
            except Exception as de:
                logger.error("DashScope video_generation 异常: %s", de, exc_info=True)
                self.last_video_error = {
                    "error_code": "unknown",
                    "user_message": f"Wan 3.0 视频生成异常：{de!s}"[:200],
                    "status_code": 0,
                    "detail": str(de),
                }
                return None

        ref_audios = [u for u in (reference_audios or [])[:10] if u and isinstance(u, str)]
        ref_videos = [u for u in (reference_videos or [])[:3] if u and isinstance(u, str)]
        ref_imgs = [u for u in (reference_images or [])[:9] if u and isinstance(u, str)]

        # ── #2170 方舟信任链（Trust Chain）────────────────────────────────────
        # 真人照片直接传给 Seedance 会触发 50411 肖像审核拦截。
        # 解决：先通过同账号的 Seedream 5.0 Pro 图生图 AI 化（保持五官特征），
        # 得到的 AI 产物图属于"模型信任产物"，再作为 reference_image 传给 Seedance 即可通过审核。
        # 信任链只作用于 doubao provider；DashScope(Wan) 保持原行为。
        trust_chain_applied = False
        if provider == "doubao":
            seedream_prompt = (
                "保持此人五官特征、发型、肤色、面部轮廓、年龄感，生成一张高清写实人像照片，"
                "人物外貌特征与参考图完全一致，皮肤自然，光线柔和，高清细节，不要过度美化。"
            )
            raw_portrait_urls: list[str] = []
            if image_url:
                raw_portrait_urls.append(image_url)
            for u in ref_imgs:
                if u not in raw_portrait_urls:
                    raw_portrait_urls.append(u)
            if raw_portrait_urls:
                trusted_urls: list[str] = []
                for idx, raw_url in enumerate(raw_portrait_urls):
                    sd_prompt = (
                        seedream_prompt if len(raw_portrait_urls) == 1 else f"{seedream_prompt}（这是参考图{idx + 1}）"
                    )
                    sd_result = self.image_generation(
                        prompt=sd_prompt,
                        reference_images=[raw_url],
                        size="2K",
                        timeout=120,
                    )
                    if not sd_result:
                        logger.warning(
                            "[trust-chain] Seedream 第 %d/%d 张失败: %s，回退直传原图",
                            idx + 1,
                            len(raw_portrait_urls),
                            self.last_image_error,
                        )
                        break
                    trusted_urls.append(sd_result["url"])
                if trusted_urls and len(trusted_urls) == len(raw_portrait_urls):
                    trust_chain_applied = True
                    # 替换：原 image_url 用第一张 AI 图，ref_imgs 用剩余
                    if image_url and trusted_urls:
                        image_url = trusted_urls[0]
                        ref_imgs = trusted_urls[1:] if len(trusted_urls) > 1 else []
                    else:
                        ref_imgs = trusted_urls
                    logger.info(
                        "[trust-chain] Seedream AI 化完成 %d 张，替换为 reference_image 模式",
                        len(trusted_urls),
                    )
                else:
                    # Seedream 部分失败 → 回退原图直传（仍可能被 50411 拦截，但保留降级路径）
                    logger.warning(
                        "[trust-chain] Seedream AI 化不完整(%d/%d)，回退原图直传",
                        len(trusted_urls),
                        len(raw_portrait_urls),
                    )
        # ─────────────────────────────────────────────────────────────────

        # 判断任务模式：
        # - 信任链强制走 reference_image（不是 first_frame；产品语义是人物参考，不是从图开始动）
        # - 有参考音/视/多图 → omni_reference（支持指定 ratio）
        # - 纯首帧无其他参考 → first_frame（ratio=adaptive）
        has_extra_refs = bool(ref_audios or ref_videos or ref_imgs)
        is_first_frame_mode = bool(image_url) and not has_extra_refs and not trust_chain_applied
        # 最终 ratio：first_frame 模式强制 adaptive，否则按用户传值（默认 9:16）
        final_ratio = "adaptive" if is_first_frame_mode else (ratio or "9:16")

        # 构造 content 数组：text + 图 + 音 + 视
        content: list[dict[str, Any]] = [{"type": "text", "text": prompt.strip()}]
        if image_url:
            if has_extra_refs or trust_chain_applied:
                # omni_reference 或信任链模式：首张图作为 reference_image，允许指定 ratio
                content.append(
                    {
                        "type": "image_url",
                        "image_url": {"url": image_url},
                        "role": "reference_image",
                    }
                )
            else:
                # 纯首帧：显式 role=first_frame
                content.append(
                    {
                        "type": "image_url",
                        "image_url": {"url": image_url},
                        "role": "first_frame",
                    }
                )
        for u in ref_imgs:
            content.append({"type": "image_url", "image_url": {"url": u}, "role": "reference_image"})
        for u in ref_audios:
            content.append({"type": "audio_url", "audio_url": {"url": u}, "role": "reference_audio"})
        for u in ref_videos:
            content.append({"type": "video_url", "video_url": {"url": u}, "role": "reference_video"})

        create_payload: dict[str, Any] = {
            "model": video_model,
            "content": content,
            "generate_audio": bool(generate_audio),
            "duration": int(duration),
            "resolution": resolution,
            "watermark": bool(watermark),
            "ratio": final_ratio,
        }

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        create_url = f"{self.base_url}/contents/generations/tasks"
        logger.info(
            "Seedance 创建任务: model=%s dur=%ds ratio=%s mode=%s gen_audio=%s img=%d aud=%d vid=%d",
            video_model,
            duration,
            final_ratio,
            "first_frame" if is_first_frame_mode else ("omni_ref+trust_chain" if trust_chain_applied else "omni_ref"),
            generate_audio,
            (1 if image_url else 0) + len(ref_imgs),
            len(ref_audios),
            len(ref_videos),
        )
        # 打印完整 payload 便于排查（截断 prompt）
        debug_payload = dict(create_payload)
        if "content" in debug_payload:
            dbg_content = []
            for item in debug_payload["content"]:
                item_copy = dict(item)
                if item_copy.get("type") == "text" and isinstance(item_copy.get("text"), str):
                    item_copy["text"] = item_copy["text"][:200] + ("..." if len(item_copy["text"]) > 200 else "")
                dbg_content.append(item_copy)
            debug_payload["content"] = dbg_content
        logger.info("Seedance 创建任务 payload: %s", json_safe_dumps(debug_payload))

        def _do_create(payload: dict) -> tuple[str | None, Exception | None, int, str]:
            """返回 (task_id, last_err, status_code, body_text)。"""
            last_err: Exception | None = None
            last_sc = 0
            last_body = ""
            for attempt in range(self.max_retries + 1):
                try:
                    resp = httpx.post(create_url, headers=headers, json=payload, timeout=self.timeout)
                    sc = int(getattr(resp, "status_code", 0) or 0)
                    body = (getattr(resp, "text", "") or "")[:2000]
                    last_sc = sc
                    last_body = body
                    if sc >= 400:
                        logger.error("Seedance 创建任务 HTTP %d: body=%s", sc, body)
                        try:
                            resp.raise_for_status()
                        except Exception as ee:
                            last_err = ee
                            if attempt < self.max_retries and sc >= 500:
                                # 仅 5xx 重试，4xx 不重试（参数/鉴权/配额错误重试无意义）
                                time.sleep(0.5 * (2**attempt))
                                continue
                            return None, last_err, sc, body
                    data = resp.json()
                    tid = data.get("id")
                    if tid:
                        return tid, None, sc, body
                    last_err = RuntimeError(f"create ok but no id: {str(data)[:300]}")
                except _HTTP_NETWORK_ERRORS as ne:
                    last_err = ne
                    last_sc = 0
                    last_body = f"network error: {ne}"
                    logger.warning(
                        "Seedance 创建网络异常（%s），重试 %d/%d", type(ne).__name__, attempt + 1, self.max_retries + 1
                    )
                    if attempt < self.max_retries:
                        time.sleep(0.5 * (2**attempt))
                        continue
                except Exception as e:
                    last_err = e
                    if attempt < self.max_retries and not isinstance(e, _HTTP_STATUS_ERROR):
                        wait = 0.5 * (2**attempt)
                        logger.warning(
                            "Seedance 创建任务失败，%.1fs 后重试 (%d/%d): %s",
                            wait,
                            attempt + 1,
                            self.max_retries + 1,
                            e,
                        )
                        time.sleep(wait)
            return None, last_err, last_sc, last_body

        # 第一次尝试
        task_id, last_err, sc, body = _do_create(create_payload)

        # ratio 兜底：HTTP 400 且 body 提到 ratio / adaptive → 回退 adaptive 再试一次
        if (
            not task_id
            and sc == 400
            and final_ratio != "adaptive"
            and (
                "ratio" in (body or "").lower()
                or "aspect" in (body or "").lower()
                or "adaptive" in (body or "").lower()
            )
        ):
            logger.warning("Seedance 创建因 ratio 失败，回退 ratio=adaptive 重试")
            create_payload["ratio"] = "adaptive"
            task_id, last_err, sc2, body2 = _do_create(create_payload)
            if task_id:
                sc, body = sc2, body2
            else:
                # 保留第二次的错误信息
                sc, body = sc2, body2

        if not task_id:
            err_code, user_msg = _classify_video_error(sc, body, last_err)
            self.last_video_error = {
                "error_code": err_code,
                "user_message": user_msg,
                "status_code": sc,
                "detail": (body or "")[:500] or (str(last_err) if last_err else ""),
                "model": video_model,
                "base_url": self.base_url,
            }
            logger.error(
                "Seedance 创建任务最终失败: model=%s base_url=%s status=%d code=%s err=%s body=%s",
                video_model,
                self.base_url,
                sc,
                err_code,
                last_err,
                (body or "")[:500],
            )
            return None

        logger.info("Seedance 任务已创建: task_id=%s ratio=%s", task_id, create_payload["ratio"])

        # 2) 轮询状态
        poll_url = f"{create_url}/{task_id}"
        deadline = time.time() + total_timeout
        video_url: str | None = None
        usage: dict | None = None
        last_status: str = "queued"
        poll_count = 0
        last_poll_body: str = ""
        last_poll_sc: int = 0
        while time.time() < deadline:
            poll_count += 1
            try:
                resp = httpx.get(poll_url, headers=headers, timeout=self.timeout)
                last_poll_sc = int(getattr(resp, "status_code", 200) or 200)
                last_poll_body = (getattr(resp, "text", "") or "")[:1500]
                if last_poll_sc >= 400:
                    logger.warning("Seedance 轮询 HTTP %d: %s", last_poll_sc, last_poll_body[:300])
                    if poll_count < 3:
                        time.sleep(poll_interval)
                        continue
                    last_err = RuntimeError(f"poll HTTP {last_poll_sc}: {last_poll_body[:200]}")
                    break
                data = resp.json()
                status = data.get("status", "")
                last_status = status
                if status == "succeeded":
                    content_obj = data.get("content") or {}
                    video_url = content_obj.get("video_url")
                    usage = data.get("usage") or content_obj.get("usage") or None
                    if video_url:
                        logger.info("Seedance 任务成功: task_id=%s polls=%d usage=%s", task_id, poll_count, usage)
                        break
                    # 成功但没 video_url：记录完整响应便于排查
                    logger.error(
                        "Seedance succeeded 但无 video_url: task_id=%s full_response=%s",
                        task_id,
                        str(data)[:1000],
                    )
                    last_err = RuntimeError("task succeeded but no video_url in response")
                    last_poll_body = str(data)[:1000]
                    break
                if status == "failed":
                    err = data.get("error") or {}
                    err_code = str(err.get("code", "") or "")
                    err_msg = str(err.get("message", "") or err.get("msg", "") or "")
                    last_err = RuntimeError(f"task failed: code={err_code} msg={err_msg}")
                    logger.error("Seedance 任务失败 task_id=%s code=%s msg=%s", task_id, err_code, err_msg)
                    last_poll_body = str(data)[:1000]
                    break
                if status in ("expired", "cancelled"):
                    last_err = RuntimeError(f"task {status}")
                    logger.error("Seedance 任务 %s: task_id=%s", status, task_id)
                    break
                # 每 5 次轮询打一次 info 日志，便于观察进度
                if poll_count % 5 == 0:
                    logger.info("Seedance 轮询中: task_id=%s status=%s polls=%d", task_id, status, poll_count)
            except httpx.HTTPStatusError as e:
                last_err = e
                last_poll_sc = e.response.status_code
                last_poll_body = (e.response.text or "")[:500]
                logger.warning("Seedance 轮询 HTTP %d: %s", e.response.status_code, last_poll_body[:300])
            except Exception as e:
                last_err = e
                logger.debug("Seedance 轮询异常: %s", e)
            time.sleep(poll_interval)

        if not video_url:
            # 区分轮询超时 vs 任务失败
            if last_status in ("queued", "running", "pending") and poll_count > 0 and time.time() >= deadline:
                err_code, user_msg = (
                    "network_error",
                    f"视频生成超时（>{total_timeout}s），任务仍在排队，请稍后重试或联系管理员。",
                )
                detail = f"timeout after {total_timeout}s, polls={poll_count}, last_status={last_status}"
            else:
                err_code, user_msg = _classify_video_error(last_poll_sc, last_poll_body, last_err)
                detail = (last_poll_body or "")[:500] or (str(last_err) if last_err else f"last_status={last_status}")
            self.last_video_error = {
                "error_code": err_code,
                "user_message": user_msg,
                "status_code": last_poll_sc,
                "detail": detail,
                "task_id": task_id,
                "last_status": last_status,
            }
            logger.error(
                "Seedance 任务未成功: task_id=%s last_status=%s polls=%d code=%s err=%s (总等待 %.0fs)",
                task_id,
                last_status,
                poll_count,
                err_code,
                last_err,
                total_timeout,
            )
            return None

        # 3) 下载到本地（下载超时收紧到 120s）
        try:
            out_dir = output_dir or "/tmp"
            os.makedirs(out_dir, exist_ok=True)
            local_path = f"{out_dir}/seedance_{task_id}_{uuid.uuid4().hex[:8]}.mp4"
            download_timeout = 120.0
            logger.info(
                "Seedance 开始下载: task_id=%s url=%s timeout=%.0fs", task_id, video_url[:120], download_timeout
            )
            with httpx.stream("GET", video_url, timeout=download_timeout) as r:
                r.raise_for_status()
                downloaded = 0
                with open(local_path, "wb") as f:
                    for chunk in r.iter_bytes(chunk_size=1024 * 256):
                        if chunk:
                            f.write(chunk)
                            downloaded += len(chunk)
            size = os.path.getsize(local_path)
            logger.info("Seedance 视频下载完成: %s size=%d bytes", local_path, size)
            if size == 0:
                logger.error("Seedance 下载文件大小为 0")
                try:
                    os.remove(local_path)
                except Exception:
                    pass
                self.last_video_error = {
                    "error_code": "unknown",
                    "user_message": "视频生成成功但下载文件为空，请稍后重试。",
                    "status_code": 0,
                    "detail": f"downloaded 0 bytes from {video_url[:120]}",
                }
                return None
            return {"video_path": local_path, "usage": usage}
        except Exception as e:
            logger.error("Seedance 视频下载失败: %s", e, exc_info=True)
            self.last_video_error = {
                "error_code": "network_error",
                "user_message": f"视频下载失败：{e!s}"[:200],
                "status_code": 0,
                "detail": str(e),
            }
            return None

    def image_generation(
        self,
        prompt: str,
        *,
        reference_images: list[str] | None = None,
        size: str = "2K",
        model: str | None = None,
        watermark: bool = False,
        output_format: str = "png",
        timeout: int | None = None,
    ) -> dict | None:
        """#2170: 调用方舟 Seedream 图片生成（文生图/图生图）。

        - reference_images: 0~10 张参考图 URL；0 张 = 纯文生图；1 张 string/URL 直传；多张 list[str]。
        - 成功返回 {"url": str, "usage": dict | None}；失败返回 None，错误写入 self.last_image_error。
        - 返回的 url 有时效性（通常 24h），应立即使用，不持久化存储。
        """
        self.last_image_error = {}
        if not self.is_available:
            self.last_image_error = {
                "error_code": "auth_error",
                "user_message": "图片生成服务未配置（API Key 缺失），请联系管理员。",
                "detail": "DoubaoClient not available (api_key empty)",
            }
            return None
        if not prompt or not prompt.strip():
            self.last_image_error = {
                "error_code": "invalid_param",
                "user_message": "图片生成提示词不能为空。",
                "detail": "empty prompt",
            }
            return None

        img_model = model or self.image_model
        url = f"{self.base_url}/images/generations"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload: dict[str, Any] = {
            "model": img_model,
            "prompt": prompt.strip(),
            "size": size,
            "response_format": "url",
            "output_format": output_format,
            "watermark": bool(watermark),
        }
        ref_imgs_local = [u for u in (reference_images or []) if u and isinstance(u, str)]
        if ref_imgs_local:
            if len(ref_imgs_local) == 1:
                payload["image"] = ref_imgs_local[0]
            else:
                payload["image"] = ref_imgs_local[:10]

        req_timeout = timeout or self.image_timeout
        last_err: Exception | None = None
        last_sc = 0
        last_body = ""
        for attempt in range(self.max_retries + 1):
            try:
                resp = httpx.post(url, headers=headers, json=payload, timeout=req_timeout)
                last_sc = int(getattr(resp, "status_code", 0) or 0)
                last_body = (getattr(resp, "text", "") or "")[:2000]
                if last_sc >= 400:
                    logger.error("Seedream 图片生成 HTTP %d: %s", last_sc, last_body[:500])
                    try:
                        resp.raise_for_status()
                    except Exception as ee:
                        last_err = ee
                        if attempt < self.max_retries and last_sc >= 500:
                            time.sleep(0.5 * (2**attempt))
                            continue
                        break
                data = resp.json()
                data_list = data.get("data") or []
                if data_list and isinstance(data_list, list):
                    item = data_list[0]
                    img_url = item.get("url")
                    if img_url:
                        logger.info(
                            "Seedream 图片生成成功 model=%s ref_imgs=%d size=%s",
                            img_model,
                            len(ref_imgs_local),
                            size,
                        )
                        return {"url": img_url, "usage": data.get("usage")}
                last_err = RuntimeError(f"Seedream 返回结构异常: {str(data)[:300]}")
                break
            except _HTTP_NETWORK_ERRORS as ne:
                last_err = ne
                last_sc = 0
                last_body = f"network error: {ne}"
                logger.warning(
                    "Seedream 网络异常 (%s)，重试 %d/%d", type(ne).__name__, attempt + 1, self.max_retries + 1
                )
                if attempt < self.max_retries:
                    time.sleep(0.5 * (2**attempt))
                    continue
                break
            except Exception as e:
                last_err = e
                if attempt < self.max_retries and not isinstance(e, _HTTP_STATUS_ERROR):
                    wait = 0.5 * (2**attempt)
                    logger.warning(
                        "Seedream 图片生成失败，%.1fs 后重试 (%d/%d): %s", wait, attempt + 1, self.max_retries + 1, e
                    )
                    time.sleep(wait)
                    continue
                break

        # 分类错误
        err_code = "unknown"
        user_msg = "图片生成失败，请稍后重试。"
        body_lower = (last_body or "").lower()
        if last_sc == 401 or last_sc == 403:
            err_code, user_msg = "auth_error", "图片生成服务鉴权失败，请联系管理员。"
        elif last_sc == 400:
            if any(k in body_lower for k in ("quota", "billing", "insufficient", "balance")):
                err_code, user_msg = "quota_exceeded", "图片生成配额不足或账号欠费，请联系管理员。"
            elif any(k in body_lower for k in ("rate", "throughput", "too many", "frequency")):
                err_code, user_msg = "rate_limit", "图片生成请求过于频繁，请稍后重试。"
            elif any(k in body_lower for k in ("sensitive", "porn", "terror", "risk", "audit", "content", "violat")):
                err_code, user_msg = "portrait_intercept", "参考素材未通过内容安全审核，请更换照片后重试。"
            else:
                err_code, user_msg = "invalid_param", f"图片生成参数错误：{last_body[:200]}"
        elif last_sc == 404:
            err_code, user_msg = "model_not_found", f"图片模型 {img_model} 不存在，请联系管理员。"
        elif last_sc >= 500:
            err_code, user_msg = "network_error", "图片生成服务暂时不可用，请稍后重试。"
        elif last_sc == 0:
            err_code, user_msg = "network_error", f"图片生成网络错误：{last_err!s}"[:200]
        self.last_image_error = {
            "error_code": err_code,
            "user_message": user_msg,
            "status_code": last_sc,
            "detail": (last_body or "")[:500] or (str(last_err) if last_err else ""),
            "model": img_model,
        }
        logger.error(
            "Seedream 图片生成最终失败: model=%s status=%d code=%s err=%s", img_model, last_sc, err_code, last_err
        )
        return None

    def get_last_image_error(self) -> dict:
        """返回最近一次 image_generation 失败的详细错误。空 dict 表示上次成功或未调用。"""
        return dict(self.last_image_error or {})

    def get_last_video_error(self) -> dict:
        """返回最近一次 video_generation 失败的详细错误。空 dict 表示上次成功或未调用。"""
        return dict(self.last_video_error or {})


def json_safe_dumps(obj: Any, max_len: int = 2000) -> str:
    """安全 json 序列化，失败则 fallback 到 repr，超长截断。"""
    try:
        import json as _json

        s = _json.dumps(obj, ensure_ascii=False, default=str)
    except Exception:
        s = repr(obj)
    if len(s) > max_len:
        s = s[:max_len] + f"...(truncated, total {len(s)})"
    return s


# ── 单例 ─────────────────────────────────────────────────────────────────────


_client: Optional[DoubaoClient] = None


def get_doubao_client() -> DoubaoClient:
    """获取豆包客户端单例."""
    global _client
    if _client is None:
        _client = DoubaoClient()
    return _client