"""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