saas-backend
75ec9db439
- ai_client/dashscope_client 新增 last_video_error 字段，记录每次失败的 error_code/user_message/status_code/detail/model 等结构化信息 - 新增 _classify_video_error 错误分类器，自动识别： - 真人肖像拦截 (portrait_intercept) → 提示用户移除真人图片 - 鉴权失败 (auth_error) / 配额不足 (quota_exceeded) / 限流 (rate_limit) - 模型/Endpoint 不存在 (model_not_found) / 参数错误 (invalid_param) - 网络超时 (network_error) / 任务失败 (task_failed) / 未知 (unknown) - call_video_generation 失败时通过 get_last_video_error() 暴露详细错误 - worker _step_render 不再笼统抛"返回为空"，改为读取错误详情并展示具体用户提示 - 增强日志：完整 payload 打印（截断 prompt）、轮询 HTTP 错误记录完整 body、 下载/超时时记录具体失败原因 - 修复重试策略：4xx 错误不再重试（参数/鉴权错误重试无意义），仅 5xx/网络错误重试 - DashScope 客户端同样增加 last_video_error 和错误分类，保持语义一致 - 新增单测覆盖：400真人拦截、401鉴权失败、轮询failed、ai_service错误透传、dashscope 401
345 lines
16 KiB
Python
345 lines
16 KiB
Python
"""DashScope 客户端（阿里云百炼 Wan 3.0 等非方舟模型）。
|
||
|
||
#2159: 新增 Wan 3.0 视频生成支持。DashScope 异步协议：
|
||
- POST {base_url}/services/aigc/video-generation/video-synthesis (X-DashScope-Async: enable)
|
||
→ 返回 output.task_id
|
||
- GET {base_url}/tasks/{task_id} 轮询状态
|
||
→ SUCCEEDED 时 output.video_url 可下载
|
||
认证：Authorization: Bearer {DASHSCOPE_API_KEY}
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.parse import urlparse
|
||
|
||
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__)
|
||
|
||
_DASHSCOPE_CLIENT_SINGLETON: "DashScopeClient | None" = None
|
||
|
||
|
||
def _classify_dashscope_error(status_code: int, body: str, task_msg: str = "") -> tuple[str, str]:
|
||
"""DashScope 错误分类，返回 (error_code, user_message)。"""
|
||
body_lower = (body or "").lower()
|
||
msg_in_body = task_msg or ""
|
||
try:
|
||
import json as _json
|
||
|
||
parsed = _json.loads(body or "{}")
|
||
if isinstance(parsed, dict):
|
||
msg_in_body = msg_in_body or str(parsed.get("message", "") or "")
|
||
except Exception:
|
||
pass
|
||
|
||
if status_code in (401, 403):
|
||
return "auth_error", "Wan 3.0 服务鉴权失败（DASHSCOPE_API_KEY 无效或过期），请联系管理员。"
|
||
if status_code == 429 or "rate" in body_lower or "throttl" in body_lower:
|
||
return "rate_limit", "Wan 3.0 服务繁忙（限流），请稍等1-2分钟后重试。"
|
||
if status_code == 400 and any(
|
||
kw in body_lower for kw in ("portrait", "真人", "人脸", "肖像", "content_violation", "risk", "blocked")
|
||
):
|
||
return (
|
||
"portrait_intercept",
|
||
"参考素材包含真人照片或违规内容被安全策略拦截，请移除真人图片或调整文案后重试。",
|
||
)
|
||
if status_code == 404 or ("not found" in body_lower) or ("model" in body_lower and "not exist" in body_lower):
|
||
return "model_not_found", "Wan 3.0 模型未开通或模型ID无效，请联系管理员。"
|
||
if status_code in (402, 400) and ("quota" in body_lower or "billing" in body_lower or "insufficient" in body_lower):
|
||
return "quota_exceeded", "Wan 3.0 服务配额不足，请联系管理员充值或稍后重试。"
|
||
if status_code == 400:
|
||
return "invalid_param", f"Wan 3.0 参数错误：{msg_in_body or body[:200]}"
|
||
if status_code == 0:
|
||
return "network_error", "Wan 3.0 服务连接失败（网络超时），请稍后重试。"
|
||
# 任务内失败
|
||
if task_msg and any(kw in task_msg.lower() for kw in ("portrait", "真人", "人脸", "violation", "blocked")):
|
||
return "portrait_intercept", "Wan 3.0 视频内容被安全策略拦截，请调整文案或参考图后重试。"
|
||
detail = msg_in_body or body[:200]
|
||
return "unknown", f"Wan 3.0 视频生成失败（HTTP {status_code}）：{detail}"
|
||
|
||
|
||
class DashScopeClient:
|
||
"""阿里云 DashScope 异步 API 客户端（Wan 3.0 等视频生成）。"""
|
||
|
||
def __init__(self) -> None:
|
||
settings = get_shared_settings()
|
||
self.api_key: str = getattr(settings, "dashscope_api_key", "") or os.getenv("DASHSCOPE_API_KEY", "")
|
||
self.base_url: str = (
|
||
getattr(settings, "dashscope_base_url", "") or "https://dashscope.aliyuncs.com/api/v1"
|
||
).rstrip("/")
|
||
self.poll_interval: int = int(getattr(settings, "dashscope_video_poll_interval", 10) or 10)
|
||
self.total_timeout: int = int(getattr(settings, "dashscope_video_timeout", 900) or 900)
|
||
self.max_retries: int = 2
|
||
self.last_video_error: dict = {}
|
||
|
||
@property
|
||
def is_available(self) -> bool:
|
||
return bool(self.api_key)
|
||
|
||
def get_last_video_error(self) -> dict:
|
||
return dict(self.last_video_error or {})
|
||
|
||
def _set_error(self, error_code: str, user_message: str, status_code: int = 0, detail: str = "", **extra) -> None:
|
||
self.last_video_error = {
|
||
"error_code": error_code,
|
||
"user_message": user_message,
|
||
"status_code": status_code,
|
||
"detail": detail[:500] if detail else "",
|
||
**extra,
|
||
}
|
||
|
||
def video_generation(
|
||
self,
|
||
prompt: str,
|
||
*,
|
||
image_url: str | None = None,
|
||
duration: int = 5,
|
||
ratio: str | None = "9:16",
|
||
resolution: str = "720p",
|
||
watermark: bool = False,
|
||
output_dir: str | None = None,
|
||
model: str = "wan3.0-video",
|
||
) -> dict | None:
|
||
"""调用 DashScope 异步视频合成接口，轮询完成后下载到本地。
|
||
|
||
返回 {"video_path": str, "usage": dict | None}；失败返回 None，错误详情写入 self.last_video_error。
|
||
"""
|
||
self.last_video_error = {}
|
||
if not self.is_available:
|
||
self._set_error("auth_error", "Wan 3.0 API key 未配置，请联系管理员。", detail="dashscope api_key empty")
|
||
logger.error("[dashscope] API key 未配置，无法调用视频生成")
|
||
return None
|
||
if not prompt or not prompt.strip():
|
||
self._set_error("invalid_param", "视频生成提示词不能为空。", detail="empty prompt")
|
||
return None
|
||
|
||
# DashScope 分辨率参数：720P / 1080P / 480P（大写 P）
|
||
res_upper = (resolution or "720p").upper().replace("P", "P")
|
||
if res_upper == "480P":
|
||
ds_res = "480P"
|
||
elif res_upper == "1080P":
|
||
ds_res = "1080P"
|
||
else:
|
||
ds_res = "720P"
|
||
|
||
# 构造 input+parameters
|
||
input_obj: dict[str, Any] = {"prompt": prompt.strip()}
|
||
if image_url:
|
||
input_obj["img_url"] = image_url
|
||
params: dict[str, Any] = {
|
||
"resolution": ds_res,
|
||
"duration": str(float(duration)),
|
||
"watermark": bool(watermark),
|
||
}
|
||
# 比例透传：Wan 支持 "9:16" / "16:9" / "1:1" 等
|
||
if ratio and ratio != "adaptive":
|
||
params["aspect_ratio"] = ratio
|
||
|
||
payload: dict[str, Any] = {
|
||
"model": model,
|
||
"input": input_obj,
|
||
"parameters": params,
|
||
}
|
||
headers = {
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json",
|
||
"X-DashScope-Async": "enable",
|
||
}
|
||
create_url = f"{self.base_url}/services/aigc/video-generation/video-synthesis"
|
||
logger.info(
|
||
"[dashscope] 创建任务: model=%s dur=%ds ratio=%s res=%s img=%s",
|
||
model,
|
||
duration,
|
||
ratio,
|
||
ds_res,
|
||
bool(image_url),
|
||
)
|
||
logger.info("[dashscope] 创建任务 payload: model=%s params=%s", model, params)
|
||
|
||
# 创建任务
|
||
task_id: str | 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=60)
|
||
sc = int(getattr(resp, "status_code", 0) or 0)
|
||
body_text = (getattr(resp, "text", "") or "")[:2000]
|
||
last_sc = sc
|
||
last_body = body_text
|
||
if sc >= 400:
|
||
logger.error("[dashscope] 创建任务 HTTP %d: %s", sc, body_text)
|
||
if sc >= 500 and attempt < self.max_retries:
|
||
time.sleep(0.5 * (2**attempt))
|
||
continue
|
||
err_code, user_msg = _classify_dashscope_error(sc, body_text)
|
||
self._set_error(err_code, user_msg, sc, body_text, model=model)
|
||
return None
|
||
data = resp.json()
|
||
tid = (data.get("output") or {}).get("task_id")
|
||
if tid:
|
||
task_id = tid
|
||
break
|
||
# 部分情况下 code != 错误
|
||
code = data.get("code")
|
||
if code and code != "":
|
||
err_code, user_msg = _classify_dashscope_error(400, body_text, str(code))
|
||
self._set_error(err_code, user_msg, sc, body_text, model=model)
|
||
return None
|
||
else:
|
||
self._set_error("unknown", "Wan 3.0 响应格式异常，未返回任务ID", sc, str(data)[:500], model=model)
|
||
return None
|
||
except _HTTP_NETWORK_ERRORS as ne:
|
||
last_sc = 0
|
||
last_body = f"network error: {ne}"
|
||
logger.warning(
|
||
"[dashscope] 网络异常 %s，重试 %d/%d", type(ne).__name__, attempt + 1, self.max_retries + 1
|
||
)
|
||
if attempt < self.max_retries:
|
||
time.sleep(0.5 * (2**attempt))
|
||
continue
|
||
self._set_error("network_error", "Wan 3.0 服务连接失败（网络超时），请稍后重试。", 0, str(ne))
|
||
return None
|
||
except Exception as _e:
|
||
if attempt < self.max_retries:
|
||
time.sleep(0.5 * (2**attempt))
|
||
continue
|
||
logger.error("[dashscope] 创建任务最终失败: %s", _e)
|
||
self._set_error("unknown", f"Wan 3.0 创建任务异常：{_e!s}"[:200], 0, str(_e))
|
||
return None
|
||
if not task_id:
|
||
if not self.last_video_error:
|
||
err_code, user_msg = _classify_dashscope_error(last_sc, last_body)
|
||
self._set_error(err_code, user_msg, last_sc, last_body, model=model)
|
||
return None
|
||
|
||
# 轮询任务
|
||
poll_url = f"{self.base_url}/tasks/{task_id}"
|
||
deadline = time.time() + self.total_timeout
|
||
video_url: str | None = None
|
||
usage: dict | None = None
|
||
poll_count = 0
|
||
last_status = ""
|
||
while time.time() < deadline:
|
||
poll_count += 1
|
||
try:
|
||
r = httpx.get(poll_url, headers=headers, timeout=30)
|
||
psc = int(getattr(r, "status_code", 0) or 0)
|
||
pbody = (getattr(r, "text", "") or "")[:1500]
|
||
if psc >= 400:
|
||
logger.warning("[dashscope] 轮询 HTTP %d: %s", psc, pbody[:300])
|
||
if poll_count < 3:
|
||
time.sleep(self.poll_interval)
|
||
continue
|
||
err_code, user_msg = _classify_dashscope_error(psc, pbody)
|
||
self._set_error(err_code, user_msg, psc, pbody, task_id=task_id)
|
||
return None
|
||
d = r.json()
|
||
out = d.get("output") or {}
|
||
task_status = out.get("task_status") or d.get("task_status") or ""
|
||
last_status = task_status
|
||
if task_status == "SUCCEEDED":
|
||
video_url = out.get("video_url") or ""
|
||
usage = d.get("usage")
|
||
if not video_url:
|
||
# 结果在 results 数组
|
||
results = out.get("results") or []
|
||
if results and isinstance(results, list):
|
||
video_url = results[0].get("url") or results[0].get("video_url")
|
||
if video_url:
|
||
logger.info("[dashscope] 任务 %s 完成: %s", task_id, video_url[:120])
|
||
break
|
||
logger.error("[dashscope] 任务 %s SUCCEEDED 但无 video_url: %s", task_id, str(d)[:500])
|
||
self._set_error(
|
||
"unknown",
|
||
"Wan 3.0 任务成功但未返回视频URL，请联系管理员。",
|
||
200,
|
||
str(d)[:500],
|
||
task_id=task_id,
|
||
)
|
||
return None
|
||
if task_status in ("FAILED", "FAILED_WITH_ERROR", "ERROR"):
|
||
msg = out.get("message") or d.get("message") or out.get("error_msg") or "unknown error"
|
||
logger.error("[dashscope] 任务 %s 失败: %s", task_id, msg)
|
||
err_code, user_msg = _classify_dashscope_error(200, "", msg)
|
||
self._set_error(err_code, user_msg, 200, msg, task_id=task_id, last_status=task_status)
|
||
return None
|
||
if task_status in ("CANCELED", "CANCELLED"):
|
||
logger.warning("[dashscope] 任务 %s 被取消", task_id)
|
||
self._set_error("unknown", "Wan 3.0 任务被取消。", 200, "task cancelled", task_id=task_id)
|
||
return None
|
||
# PENDING / RUNNING / SUSPENDED → 继续轮询
|
||
if poll_count % 5 == 0:
|
||
logger.info("[dashscope] 轮询中 task=%s status=%s polls=%d", task_id, task_status, poll_count)
|
||
except Exception as e:
|
||
logger.warning("[dashscope] 轮询异常: %s", e)
|
||
time.sleep(self.poll_interval)
|
||
if not video_url:
|
||
logger.error("[dashscope] 任务 %s 轮询超时（%ds）", task_id, self.total_timeout)
|
||
self._set_error(
|
||
"network_error",
|
||
f"Wan 3.0 视频生成超时（>{self.total_timeout}s），任务仍在排队，请稍后重试。",
|
||
0,
|
||
f"timeout after {self.total_timeout}s, polls={poll_count}, last_status={last_status}",
|
||
task_id=task_id,
|
||
last_status=last_status,
|
||
)
|
||
return None
|
||
|
||
# 下载视频
|
||
out_dir = output_dir or os.path.join(os.getcwd(), "seedance_outputs")
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
suffix = Path(urlparse(video_url).path).suffix or ".mp4"
|
||
if suffix.lower() not in (".mp4", ".mov", ".webm"):
|
||
suffix = ".mp4"
|
||
safe_tid = "".join(c if c.isalnum() or c in "-_" else "_" for c in task_id)[:40]
|
||
out_path = os.path.join(out_dir, f"wan_{safe_tid}{suffix}")
|
||
try:
|
||
with httpx.stream("GET", video_url, timeout=300, follow_redirects=True) as resp:
|
||
dsc = int(getattr(resp, "status_code", 0) or 0)
|
||
if dsc >= 400:
|
||
logger.error("[dashscope] 下载 HTTP %d", dsc)
|
||
self._set_error("network_error", "Wan 3.0 视频下载失败（HTTP错误），请稍后重试。", dsc)
|
||
return None
|
||
with open(out_path, "wb") as f:
|
||
for chunk in resp.iter_bytes(chunk_size=1024 * 256):
|
||
if chunk:
|
||
f.write(chunk)
|
||
except Exception as e:
|
||
logger.error("[dashscope] 下载视频失败: %s", e, exc_info=True)
|
||
self._set_error("network_error", f"Wan 3.0 视频下载失败：{e!s}"[:200], 0, str(e))
|
||
return None
|
||
size = os.path.getsize(out_path) if os.path.exists(out_path) else 0
|
||
if size < 1024:
|
||
logger.error("[dashscope] 下载文件过小: %d bytes", size)
|
||
self._set_error("unknown", "Wan 3.0 视频下载文件过小，请稍后重试。", 0, f"downloaded only {size} bytes")
|
||
return None
|
||
logger.info("[dashscope] 视频已下载: %s (%d bytes)", out_path, size)
|
||
return {"video_path": out_path, "usage": usage}
|
||
|
||
|
||
def get_dashscope_client() -> DashScopeClient | None:
|
||
"""返回 DashScope 客户端单例；未配置 API key 时返回 None。"""
|
||
global _DASHSCOPE_CLIENT_SINGLETON
|
||
if _DASHSCOPE_CLIENT_SINGLETON is None:
|
||
_DASHSCOPE_CLIENT_SINGLETON = DashScopeClient()
|
||
if not _DASHSCOPE_CLIENT_SINGLETON.is_available:
|
||
return None
|
||
return _DASHSCOPE_CLIENT_SINGLETON
|