Compare commits

..

1 Commits

Author SHA1 Message Date
xiaoxia 866d71a431 feat(viral-video): #2169 接入即梦视觉API,真人参考图方舟50411自动切即梦兜底 (#2169)
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 2s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 2s
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 3s
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Check push changed paths (push) Successful in 24s
CI/CD Pipeline / Validate - Style (pull_request) Has been skipped
CI/CD Pipeline / Validate - Security (pull_request) Has been skipped
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 40s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 1m12s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m15s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 3m27s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 3m39s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 4m13s
CI/CD Pipeline / Integration Tests (push) Successful in 4m20s
CI/CD Pipeline / Validate - Style (push) Successful in 5m3s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 4m38s
CI/CD Pipeline / Build Staging API Image (push) Successful in 4m37s
CI/CD Pipeline / CI Gate (pull_request) Successful in 2s
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 5m34s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m8s
AI Code Review / AI Code Review (pull_request) Successful in 6m45s
CI/CD Pipeline / Validate - Security (push) Successful in 8m18s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 2m37s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 2m48s
CI/CD Pipeline / Unit Tests (push) Successful in 11m45s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Failing after 35m40s
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
2026-10-04 01:48:59 +08:00
8 changed files with 1194 additions and 284 deletions
+14
View File
@@ -220,6 +220,20 @@ DOUBAO_MAX_RETRIES=2
DOUBAO_VISION_MODEL=doubao-1-5-vision-pro-250328
DOUBAO_VISION_LITE_MODEL=doubao-1-5-vision-lite-250315
DOUBAO_VISION_USE_LITE=true
# Embedding 向量化模型(原 large-text-240915 已下线,用多模态 embedding)
DOUBAO_EMBEDDING_MODEL=doubao-embedding-vision-251215
# ==================== 即梦(Jimeng)视觉 API —— 真人参考图兜底通道 (#2169) ====
# 方舟 Seedance 走 B 端审核,真人参考图会被 50411 拦截;即梦走 C 端审核,普通真人照片可过审。
# 需要在火山控制台开通即梦 cvtob 服务,使用 AK/SK(Region=cn-north-1, Service=cv)
# 留空则真人拦截后直接返回错误提示,不会走即梦兜底。
JIMENG_AK=
JIMENG_SK=
JIMENG_BASE_URL=https://visual.volcengineapi.com
# 视频3.0 720P 首帧图生视频(1张图,稳定版;1080P标注下线中)
JIMENG_REQ_KEY=jimeng_i2v_first_v30
JIMENG_VIDEO_TIMEOUT=600
JIMENG_VIDEO_POLL_INTERVAL=5
# ==================== 积分/会员系统 (#1895) ====================
# 积分系统总开关:默认 false(暂停积分系统)。
+27 -40
View File
@@ -647,7 +647,7 @@ def _build_products_summary(image_analysis: dict) -> str:
# 优先 VLM 生成的 summary 段(自然语言,给编导模型看效果最好)
summary = (p.get("summary") or "").strip()
if summary and len(summary) >= 30:
lines.append(f"- 图{i+1} {name}:{summary}")
lines.append(f"- 图{i + 1} {name}:{summary}")
continue
# 结构化字段兜底
brand = p.get("brand") or ""
@@ -669,7 +669,7 @@ def _build_products_summary(image_analysis: dict) -> str:
feats = p.get("key_features") or p.get("features") or []
sellings = p.get("selling_points") or []
scenes = p.get("suitable_scenes") or []
parts = [f"图{i+1} {name}"]
parts = [f"图{i + 1} {name}"]
if brand and brand not in ("未知", "无法判断"):
parts.append(f"品牌={brand}")
if cat and cat not in ("无法判断", "非产品图"):
@@ -802,7 +802,7 @@ def _validate_and_normalize_script(raw, job: ViralVideoJob) -> dict:
continue
shots.append(
{
"time_range": str(s.get("time_range") or f"{i*3}-{(i+1)*3}秒"),
"time_range": str(s.get("time_range") or f"{i * 3}-{(i + 1) * 3}秒"),
"shot_type_angle_movement": str(s.get("shot_type_angle_movement") or "中景平视,固定镜头"),
"scene_and_dialogue": str(s.get("scene_and_dialogue") or ""),
"action_details": str(s.get("action_details") or ""),
@@ -878,8 +878,8 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
style_hint = "无"
if isinstance(job.style_guide, dict):
style_hint = (
f"节奏{job.style_guide.get('cut_speed','')}、转场{job.style_guide.get('transition','')}、"
f"色调{job.style_guide.get('color_grade','')}、能量{job.style_guide.get('energy','')}"
f"节奏{job.style_guide.get('cut_speed', '')}、转场{job.style_guide.get('transition', '')}、"
f"色调{job.style_guide.get('color_grade', '')}、能量{job.style_guide.get('energy', '')}"
)
dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
@@ -1103,14 +1103,14 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str:
ab = s.get("audio_bgm", "")
t = s.get("transition", "")
ref = s.get("reference_image_index")
lines.append(f"- 镜头{i+1}({tr}):")
lines.append(f"- 镜头{i + 1}({tr}):")
lines.append(f" 景别/运镜:{cam}")
lines.append(f" 画面与对白:{sd}")
lines.append(f" 动作细节:{act}")
lines.append(f" 音效/BGM:{ab}")
lines.append(f" 转场:{t}")
if ref is not None and isinstance(ref, int):
lines.append(f" 参考图片:第{ref+1}张产品图")
lines.append(f" 参考图片:第{ref + 1}张产品图")
lines.append("")
lines.append("【硬性约束】")
for c in hc:
@@ -1177,8 +1177,9 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
reference_videos=ref_videos,
)
# 肖像/内容审核拦截降级重试:去掉参考图/参考视频走纯文字 t2v 再试一次
# (不影响音频——去掉首帧和多参考图通常就能绕过真人照片审核,纯文字描述真人是允许的)
# #2169: 真人/肖像拦截已由 ai_client 内部自动切即梦(jimeng-3.0)通道重试——
# 保留首帧图、不走"去掉参考图纯 t2v 降级"(用户明确要求按参考照片生成)。
# 即梦也失败或非拦截类错误时,直接抛错给上层展示用户友好提示。
def _check_and_reraise(result):
if result and isinstance(result, dict):
return result
@@ -1189,11 +1190,11 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
detail = err.get("detail") or ""
err_code = err.get("error_code") or "unknown"
status_code = err.get("status_code", 0)
provider = _mcfg.get("provider", "doubao")
err_msg = user_msg or f"视频生成失败({provider} status={status_code} code={err_code})"
err_provider = err.get("provider") or _mcfg.get("provider", "doubao")
err_msg = user_msg or f"视频生成失败({err_provider} status={status_code} code={err_code})"
logger.error(
"[爆款视频] 视频生成失败: provider=%s model=%s code=%s status=%s user_msg=%s detail=%s",
provider,
err_provider,
model or "default",
err_code,
status_code,
@@ -1203,38 +1204,24 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
raise RuntimeError(err_msg)
if not result or not isinstance(result, dict):
from packages.shared.ai_service import get_last_video_error
err = get_last_video_error() or {}
err_code = err.get("error_code") or "unknown"
is_portrait = err_code == "portrait_intercept" or any(
k in (err.get("detail", "") or "").lower()
for k in ("real person", "privacy", "portrait", "sensitivecontent", "face", "人脸", "肖像", "真人")
)
if is_portrait and (first_image or rest_images or ref_videos):
logger.warning("[爆款视频] 参考素材触发真人/肖像审核拦截(error_code=%s),降级为纯文字t2v重试一次", err_code)
# 去掉所有参考图/参考视频,纯 t2v 泛化生成;音频仍保留(不影响肖像审核)
result = call_video_generation(
prompt=prompt,
image_url=None,
duration=dur,
ratio=ratio,
resolution=resolution,
output_dir=str(tmpdir),
model=model,
generate_audio=gen_audio,
reference_images=[],
reference_audios=ref_audios,
reference_videos=[],
)
_check_and_reraise(result)
video_path = result.get("video_path") or ""
usage = result.get("usage")
if not video_path or not Path(video_path).exists() or Path(video_path).stat().st_size == 0:
raise RuntimeError("Seedance 视频生成失败:返回空文件或路径不存在")
logger.info(
"[爆款视频] Seedance 单次生成完成: %s size=%d usage=%s", video_path, Path(video_path).stat().st_size, usage
)
raise RuntimeError("视频生成失败:返回空文件或路径不存在")
# #2169: 如果实际走了即梦兜底(真人拦截→jimeng),更新 job.video_model 让积分结算用 jimeng-3.0 价格
if isinstance(usage, dict):
actual_provider = usage.get("provider")
actual_model_key = usage.get("model_key")
if actual_provider == "jimeng" and actual_model_key:
logger.info(
"[爆款视频] 实际通过即梦通道生成(原model=%s),更新video_model=%s 用于积分结算",
job.video_model,
actual_model_key,
)
job.video_model = actual_model_key
size = Path(video_path).stat().st_size
logger.info("[爆款视频] 单次生成完成: path=%s size=%d usage=%s", video_path, size, usage)
return str(video_path), (usage if isinstance(usage, dict) else None)
+22 -5
View File
@@ -90,15 +90,21 @@ class SharedSettings(BaseSettings):
# ── 豆包大模型(火山引擎方舟) ────────────────────────────────────────
doubao_api_key: str = ""
doubao_model: str = "doubao-seed-1-6-250615" # 推理模型(通用兜底)
doubao_fast_model: str = "doubao-1-5-pro-32k-250115" # 快速结构化输出模型(编导脚本/意图解析/审核)
doubao_model: str = "doubao-seed-2-1-pro-260915" # 推理模型(Seed 2.1 Pro,深度思考+多模态;原 seed-1-6 已下线)
doubao_fast_model: str = (
"doubao-seed-2-1-lite-260915" # 快速模型(Seed 2.1 Lite,高 RPM,编导/审核/VLM lite;原 1-5-pro-32k 已 Retiring)
)
doubao_base_url: str = "https://ark.cn-beijing.volces.com/api/v3"
doubao_timeout: int = 30
doubao_max_retries: int = 2
doubao_vision_model: str = "doubao-1-5-vision-pro-250328" # 高精度视觉(备用)
doubao_vision_lite_model: str = "doubao-1-5-vision-lite-250315" # 快速视觉(商品识别默认,速度优先)
doubao_vision_model: str = (
"doubao-seed-2-1-pro-260915" # 高精度视觉(Seed 2.1 Pro 原生多模态;原 vision-pro-250328 已下线)
)
doubao_vision_lite_model: str = (
"doubao-seed-2-1-lite-260915" # 快速视觉(Seed 2.1 Lite 原生多模态;原 vision-lite-250315 不可用)
)
doubao_vision_use_lite: bool = True # viral-video 图片分析默认用 lite 提速
doubao_embedding_model: str = "doubao-embedding-large-text-240915"
doubao_embedding_model: str = "doubao-embedding-vision-251215" # 多模态向量化(原 large-text-240915 已 Retiring)
doubao_video_model: str = "doubao-seedance-2-5-260628"
doubao_video_timeout: int = 600 # 视频生成轮询总超时(秒)
doubao_video_poll_interval: int = 10 # 轮询间隔(秒)
@@ -109,6 +115,17 @@ class SharedSettings(BaseSettings):
dashscope_video_timeout: int = 900 # Wan 视频任务轮询总超时(秒)
dashscope_video_poll_interval: int = 10
# ── 即梦(Jimeng)视觉 API —— 火山引擎 cvtob ──────────────────────────
# #2169: 真人参考图在方舟 Seedance 走 B 端审核会被 50411 拦截,
# 即梦走 C 端审核链路,普通真人照片可过审,作为参考图场景兜底通道。
# 鉴权:AK/SK V4 签名(Region=cn-north-1, Service=cv)
jimeng_ak: str = ""
jimeng_sk: str = ""
jimeng_base_url: str = "https://visual.volcengineapi.com"
jimeng_req_key: str = "jimeng_i2v_first_v30" # 视频3.0 720P 首帧图生视频(1张图,稳定版;1080P 标注下线中)
jimeng_video_timeout: int = 600 # 即梦轮询总超时(秒)
jimeng_video_poll_interval: int = 5 # 轮询间隔(秒)
# ── MediaKit (火山引擎 AI 媒体工具) ──────────────────────────────────
mediakit_api_key: str = ""
mediakit_base_url: str = "https://mediakit.cn-beijing.volces.com/api/v1"
+19
View File
@@ -31,6 +31,8 @@ VIRAL_VIDEO_MODEL_PRICES: dict[tuple[str, str, bool], float] = {
("wan-3.0", "480p", False): 0.3,
("wan-3.0", "720p", False): 0.6,
("wan-3.0", "1080p", False): 1.2,
# #2169: 即梦(Jimeng)视频3.0 720P 首帧图生视频,0.28 元/秒(C 端审核,真人可过)
("jimeng-3.0", "720p", False): 0.28,
}
# 固定成本(元):VLM 分析 + LLM 文案 + TTS + OSS + 服务器
@@ -151,6 +153,20 @@ VIRAL_VIDEO_MODEL_CONFIG: dict[str, dict] = {
"billing_mode": "per_second",
"is_default": False,
},
# #2169: 即梦视频3.0(内部兜底通道,方舟 Seedance 返回真人拦截 50411 时自动切到即梦重试,
# 不暴露给前端让用户直接选择,但需支持计费结算)
"jimeng-3.0": {
"key": "jimeng-3.0",
"display_name": "即梦3.0 — 真人图生视频(兜底)",
"model_id": "jimeng_i2v_first_v30",
"provider": "jimeng",
"supports_audio": False, # 即梦返回无声视频,音频由后续 ffmpeg 合成 TTS
"supported_resolutions": ["720p"],
"max_duration": 10, # 即梦 i2v 首帧最长 10s(frames=241)
"billing_mode": "per_second",
"is_default": False,
"_internal_fallback_only": True, # 标记:不对外暴露到模型选择列表
},
}
@@ -173,6 +189,9 @@ def list_viral_video_models(
continue
if cfg.get("provider") == "dashscope" and not dashscope_available:
continue
# #2169: 即梦是内部兜底通道,不在前端模型列表展示
if cfg.get("_internal_fallback_only"):
continue
out.append(
{
"key": cfg["key"],
+203 -163
View File
@@ -34,20 +34,22 @@ _HTTP_STATUS_ERROR = httpx.HTTPStatusError if hasattr(httpx, "HTTPStatusError")
logger = logging.getLogger(__name__)
# 视频模型 ID 解析逻辑(#2159 多模型支持)。
# 内部使用简短别名(seedance-2.5 / seedance-2.0 / wan-3.0 等)做 PRICING key 和前端选择值;
# 视频模型 ID 解析逻辑(#2159 多模型支持,#2169 接入即梦)。
# 内部使用简短别名(seedance-2.5 / seedance-2.0 / wan-3.0 / jimeng-3.0 等)做 PRICING key;
# 实际调用时按 VIRAL_VIDEO_MODEL_CONFIG(domain/points_rules.py)的 provider/model_id 分发。
# - provider=doubao → 火山方舟
# - provider=doubao → 火山方舟 Seedance
# - provider=dashscope → 阿里云 DashScope(Wan 系列)
# - provider=jimeng → 火山引擎即梦 cvtob(jimeng_i2v_first_v30,真人参考图走 C 端审核)
def _resolve_video_provider_and_id(model: str | None) -> tuple[str, str, dict]:
"""把内部 model key 解析成 (provider, model_id, cfg)。
- provider: "doubao" | "dashscope"
- provider: "doubao" | "dashscope" | "jimeng"
- model_id: 对应 API 的真实模型 ID
- cfg: VIRAL_VIDEO_MODEL_CONFIG 条目
未识别或空值回落到默认 seedance-2.5。已经是 doubao-/ep- 开头的完整 ID 视为 doubao provider。
未识别或空值回落到默认 seedance-2.5。已经是 doubao-/ep- 开头的完整 ID 视为 doubao provider;
"jimeng" 开头视为 jimeng provider(内部兜底,不暴露给前端)。
"""
from packages.domain.points_rules import get_viral_video_model_config
@@ -60,6 +62,10 @@ def _resolve_video_provider_and_id(model: str | None) -> tuple[str, str, dict]:
# 已经是 doubao-/ep- 开头:直接透传,默认视为 doubao provider
if m.startswith("doubao-") or m.startswith("ep-"):
return "doubao", m, {"provider": "doubao", "model_id": m, "supports_audio": True}
# 显式 jimeng 关键字:路由到即梦(内部兜底通道使用)
if m.startswith("jimeng"):
cfg = get_viral_video_model_config("jimeng-3.0")
return "jimeng", cfg.get("model_id", "jimeng_i2v_first_v30"), cfg
# 别名 → 从 domain config 查
cfg = get_viral_video_model_config(m)
provider = cfg.get("provider", "doubao")
@@ -183,8 +189,6 @@ class DoubaoClient:
self.embedding_model: str = settings.doubao_embedding_model
# 最近一次视频生成的详细错误(error_code + user_message + raw detail),供上层读取后展示给用户
self.last_video_error: dict = {}
# chat fallback 信号位:_do_chat_request 遇 404/model_not_exist 时置 True
self._last_chat_model_not_found: bool = False
def embed_text(self, text: str, timeout: int | None = None) -> list[float] | None:
"""调用豆包文本 Embedding API,返回浮点向量;失败返回 None。"""
@@ -197,7 +201,7 @@ class DoubaoClient:
"Content-Type": "application/json",
}
payload: dict[str, Any] = {
"model": getattr(self, "embedding_model", None) or "doubao-embedding-large-text-240915",
"model": self.embedding_model,
"input": text.strip(),
"encoding_format": "float",
}
@@ -232,88 +236,6 @@ class DoubaoClient:
"""是否可用(配置了 API Key)."""
return bool(self.api_key)
@staticmethod
def _is_model_not_found_err(e: Exception) -> bool:
"""判断 HTTP 异常是否是模型/endpoint 不存在(404 或 body 含 not found 关键词)。"""
try:
resp = getattr(e, "response", None)
if resp is not None:
sc = getattr(resp, "status_code", 0)
if sc == 404:
return True
try:
body = resp.text or ""
except Exception:
body = ""
bl = body.lower()
if sc == 400 and any(
k in bl
for k in (
"modelnotfound",
"model_not_exist",
"endpointnotfound",
"model not found",
"endpoint not found",
"不存在",
)
):
return True
except Exception:
pass
return False
def _do_chat_request(self, url: str, headers: dict, payload: dict, model_label: str) -> Optional[str]:
"""单次 chat 请求,带 5xx/网络重试;4xx 直接返回 None,同时置位 self._last_chat_model_not_found。"""
self._last_chat_model_not_found = False
last_err: 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()
return (data["choices"][0]["message"]["content"] or "").strip()
except Exception as e:
last_err = e
sc = getattr(getattr(e, "response", None), "status_code", 0) or 0
if 400 <= sc < 500:
# 判断是否模型/endpoint 不存在
is_mnf = sc == 404
if not is_mnf:
try:
body = getattr(e.response, "text", "") or ""
except Exception:
body = ""
bl = body.lower()
is_mnf = any(
k in bl
for k in (
"modelnotfound",
"model_not_exist",
"endpointnotfound",
"model not found",
"endpoint not found",
)
)
if is_mnf:
self._last_chat_model_not_found = True
logger.warning(
"[豆包chat] HTTP %d model=%s mnf=%s err=%s 不重试", sc, model_label, is_mnf, str(e)[:200]
)
return None
if attempt < self.max_retries:
wait = 0.5 * (2**attempt)
logger.warning(
"[豆包chat] 调用失败 %.1fs后重试(%d/%d) model=%s err=%s",
wait,
attempt + 1,
self.max_retries + 1,
model_label,
e,
)
time.sleep(wait)
logger.error("[豆包chat] 调用最终失败 model=%s err=%s", model_label, last_err)
return None
def chat_completion(
self,
messages: list[dict[str, str]],
@@ -323,8 +245,13 @@ class DoubaoClient:
) -> Optional[str]:
"""调用 Chat Completion 接口.
- 若指定 model 返回 404/model not found,自动 fallback 到默认推理模型 self.model
- 5xx/网络错误按 max_retries 重试;4xx(鉴权/配额/参数/模型不存在)不重试
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
@@ -334,24 +261,40 @@ class DoubaoClient:
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
req_model = model or self.model
payload: dict[str, Any] = {
"model": req_model,
"model": model or self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
content = self._do_chat_request(url, headers, payload, req_model)
if content is not None:
return content
# fallback:仅当请求的模型是"模型不存在/endpoint 不存在"时才降级到默认模型
# (鉴权/配额/限流等全局错误 fallback 也会失败,别浪费请求)
if req_model != self.model and getattr(self, "_last_chat_model_not_found", False):
logger.warning("[豆包chat] 模型 %s 不存在,fallback 到默认模型 %s", req_model, self.model)
payload["model"] = self.model
self._last_chat_model_not_found = False
return self._do_chat_request(url, headers, payload, self.model)
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(
@@ -411,72 +354,41 @@ class DoubaoClient:
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
req_model = model or self.vision_model
payload: dict[str, Any] = {
"model": req_model,
"model": model or self.vision_model,
"messages": vision_messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
vision_mnf = {"mnf": False}
def _do_vision(curr_model: str) -> Optional[str]:
payload["model"] = curr_model
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()
return (data["choices"][0]["message"]["content"] or "").strip()
except Exception as e:
sc = getattr(getattr(e, "response", None), "status_code", 0) or 0
if 400 <= sc < 500:
is_mnf = sc == 404
if not is_mnf:
try:
body = getattr(e.response, "text", "") or ""
except Exception:
body = ""
bl = body.lower()
is_mnf = any(
k in bl
for k in (
"modelnotfound",
"model_not_exist",
"endpointnotfound",
"model not found",
"endpoint not found",
)
)
if is_mnf:
vision_mnf["mnf"] = True
logger.warning(
"[豆包vision] HTTP %d model=%s mnf=%s err=%s 不重试", sc, curr_model, is_mnf, str(e)[:200]
)
return None
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)
return None
req_timeout = timeout or self.timeout
content = _do_vision(req_model)
if content is not None:
return content
# 仅 404/模型不存在时才 fallback 到默认 chat 模型
if req_model != self.model and vision_mnf["mnf"]:
logger.warning("[豆包vision] 模型 %s 不存在,fallback 到默认模型 %s", req_model, self.model)
content = _do_vision(self.model)
if content is not None:
return content
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,异步任务)────────────────────────────
@@ -575,6 +487,18 @@ class DoubaoClient:
}
return None
if provider == "jimeng":
# #2169: 即梦 cvtob(jimeng_i2v_first_v30)— 真人参考图兜底通道
return self._call_jimeng_video_generation(
prompt=prompt,
image_url=image_url,
duration=duration,
ratio=ratio,
resolution=resolution,
output_dir=output_dir,
generate_audio=generate_audio,
)
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)]
@@ -745,6 +669,28 @@ class DoubaoClient:
last_err,
(body or "")[:500],
)
# #2169: 方舟返回 portrait_intercept 且有参考图 → 自动切即梦重试一次(保留首帧图)
if err_code == "portrait_intercept" and image_url:
logger.warning(
"[viral-video] Seedance 真人拦截(code=%s),自动切即梦通道重试(首帧图) img=%s",
err_code,
bool(image_url),
)
jm_result = self._call_jimeng_video_generation(
prompt=prompt,
image_url=image_url,
duration=duration,
ratio=ratio,
resolution=resolution,
output_dir=output_dir,
generate_audio=False, # 即梦 i2v 不带音频,音频由后续 ffmpeg 合成
_portrait_fallback=True,
)
if jm_result is not None:
return jm_result
# 即梦也失败了,保留即梦的 last_video_error(已经由 _call_jimeng 设置)
logger.error("[viral-video] 即梦通道重试也失败: %s", self.last_video_error)
return None
return None
logger.info("Seedance 任务已创建: task_id=%s ratio=%s", task_id, create_payload["ratio"])
@@ -888,6 +834,100 @@ class DoubaoClient:
}
return None
def _call_jimeng_video_generation(
self,
*,
prompt: str,
image_url: str | None,
duration: int,
ratio: str | None,
resolution: str,
output_dir: str | None,
generate_audio: bool = False,
_portrait_fallback: bool = False,
) -> dict | None:
"""#2169: 调用即梦 cvtob 客户端做图生视频(真人参考图兜底通道)。
- 即梦 i2v 首帧接口只接受 1 张图、无原生音频(返回无声视频,音频由 ffmpeg 后合)。
- 成功返回 {"video_path": str, "usage": {...}};失败写 self.last_video_error 并返回 None。
- _portrait_fallback=True 时在日志里标注是从方舟拦截切过来的。
"""
from packages.shared.jimeng_client import get_jimeng_client
jm = get_jimeng_client()
if jm is None:
detail = "即梦 client 不可用(JIMENG_AK/SK 未配置)"
if _portrait_fallback:
# 从真人拦截切过来但即梦没配,仍把错误归到 portrait_intercept,让上层提示用户
self.last_video_error = {
"error_code": "portrait_intercept",
"user_message": "参考素材包含真人照片被安全策略拦截,即梦兜底通道未启用,请联系管理员配置 JIMENG_AK/SK。",
"status_code": 0,
"detail": detail,
"provider": "jimeng",
}
else:
self.last_video_error = {
"error_code": "auth_error",
"user_message": "即梦视频通道未配置(JIMENG_AK/SK 缺失),请联系管理员。",
"status_code": 0,
"detail": detail,
"provider": "jimeng",
}
logger.error("[jimeng] %s, portrait_fallback=%s", detail, _portrait_fallback)
return None
if not image_url:
self.last_video_error = {
"error_code": "invalid_param",
"user_message": "即梦图生视频必须提供参考图片。",
"status_code": 0,
"detail": "empty image_url for jimeng i2v",
"provider": "jimeng",
}
return None
# 即梦 i2v 无声视频,generate_audio 强制 False
jm.last_video_error = {}
tag = "[portrait-fallback→jimeng]" if _portrait_fallback else "[jimeng-direct]"
logger.info("%s 调用即梦: dur=%s ratio=%s res=%s img=%s", tag, duration, ratio, resolution, bool(image_url))
try:
result = jm.video_generation(
prompt=prompt,
image_url=image_url,
duration=duration,
ratio=ratio,
resolution=resolution,
output_dir=output_dir,
generate_audio=False,
)
except Exception as je:
logger.error("%s 即梦 video_generation 异常: %s", tag, je, exc_info=True)
self.last_video_error = {
"error_code": "unknown",
"user_message": f"即梦视频生成异常:{je!s}"[:200],
"status_code": 0,
"detail": str(je),
"provider": "jimeng",
}
return None
if not result and jm.last_video_error:
# 透传即梦错误;如果即梦也返回 portrait_intercept,说明图片真的有问题,直接给用户
jm_err = dict(jm.last_video_error)
jm_err["provider"] = "jimeng"
if _portrait_fallback and jm_err.get("error_code") == "portrait_intercept":
jm_err["user_message"] = (
"参考素材真人肖像审核未通过(方舟+即梦双通道均被拦截),请更换非真人或授权清晰的照片后重试。"
)
self.last_video_error = jm_err
return None
if result:
# 补充 usage 里的 provider 标记
u = result.get("usage") or {}
u.setdefault("provider", "jimeng")
u.setdefault("model_key", "jimeng-3.0")
result["usage"] = u
logger.info("%s 即梦生成成功: %s", tag, result.get("video_path"))
return result
def get_last_video_error(self) -> dict:
"""返回最近一次 video_generation 失败的详细错误。空 dict 表示上次成功或未调用。"""
return dict(self.last_video_error or {})
+526
View File
@@ -0,0 +1,526 @@
"""即梦(Jimeng)视觉 API 客户端 —— 火山引擎 cvtob。
#2169: 真人参考图在方舟 Seedance 走 B 端审核会被 50411 拦截,
即梦走 C 端审核链路,普通真人照片可过审。接入即梦 i2v 作为参考图场景兜底通道。
接口协议(jimeng_i2v_first_v30 —— 视频3.0 720P 首帧图生视频):
- 接口地址:https://visual.volcengineapi.com
- 鉴权:火山 V4 签名(Region=cn-north-1, Service=cv),使用 AK/SK
- 提交任务:POST ?Action=CVSync2AsyncSubmitTask&Version=2022-08-31
body: {"req_key": "jimeng_i2v_first_v30", "image_urls": ["<url>"], "prompt": "...", "seed": -1, "frames": 121}
-> {"code": 10000, "data": {"task_id": "..."}}
- 查询任务:POST ?Action=CVSync2AsyncGetResult&Version=2022-08-31
body: {"req_key": "jimeng_i2v_first_v30", "task_id": "..."}
-> {"code": 10000, "data": {"status": "in_queue|generating|done", "video_url": "..."}}
- 视频 URL 有效期 1 小时,必须立即下载到本地。
"""
from __future__ import annotations
import hashlib
import hmac
import json
import logging
import os
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote, urlparse
import httpx
from packages.shared.config import get_shared_settings
# 网络/超时类异常父类集合
_HTTP_NETWORK_ERRORS = ()
try:
_HTTP_NETWORK_ERRORS = (httpx.TimeoutException, httpx.NetworkError)
except Exception:
_HTTP_NETWORK_ERRORS = (Exception,)
logger = logging.getLogger(__name__)
_JIMENG_CLIENT_SINGLETON: "JimengClient | None" = None
# ── V4 签名常量 ────────────────────────────────────────────────────────
_JIMENG_REGION = "cn-north-1"
_JIMENG_SERVICE = "cv"
_JIMENG_VERSION = "2022-08-31"
_ACTION_SUBMIT = "CVSync2AsyncSubmitTask"
_ACTION_POLL = "CVSync2AsyncGetResult"
_CONTENT_TYPE = "application/json"
_SIGNED_HEADERS_LIST = ["content-type", "host", "x-content-sha256", "x-date"]
_SIGNED_HEADERS_STR = ";".join(_SIGNED_HEADERS_LIST)
def _norm_query(params: dict[str, str]) -> str:
"""构造规范查询串:按 key 排序,URL 编码(safe=-_.~),空格->%20。"""
parts = []
for k in sorted(params.keys()):
v = params[k]
ek = quote(str(k), safe="-_.~")
ev = quote(str(v), safe="-_.~").replace("+", "%20")
parts.append(f"{ek}={ev}")
return "&".join(parts)
def _hmac_sha256(key: bytes, msg: str) -> bytes:
return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
def _sha256_hex(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _sign_v4(
ak: str,
sk: str,
method: str,
host: str,
query: dict[str, str],
body_bytes: bytes,
x_date: str,
) -> dict[str, str]:
"""火山 V4 签名,返回需要附加到请求的 headers 字典。
x_date 形如 "20260101T120000Z"(UTC)。
short_date = x_date[:8](YYYYMMDD)。
"""
short_date = x_date[:8]
payload_hash = _sha256_hex(body_bytes)
canon_uri = "/"
canon_query = _norm_query(query)
canon_headers = f"content-type:{_CONTENT_TYPE}\nhost:{host}\nx-content-sha256:{payload_hash}\nx-date:{x_date}\n"
canon_request = f"{method}\n{canon_uri}\n{canon_query}\n{canon_headers}\n{_SIGNED_HEADERS_STR}\n{payload_hash}"
credential_scope = f"{short_date}/{_JIMENG_REGION}/{_JIMENG_SERVICE}/request"
string_to_sign = f"HMAC-SHA256\n{x_date}\n{credential_scope}\n{_sha256_hex(canon_request.encode('utf-8'))}"
k_date = _hmac_sha256(sk.encode("utf-8"), short_date)
k_region = _hmac_sha256(k_date, _JIMENG_REGION)
k_service = _hmac_sha256(k_region, _JIMENG_SERVICE)
k_signing = _hmac_sha256(k_service, "request")
signature = hmac.new(k_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
authorization = (
f"HMAC-SHA256 Credential={ak}/{credential_scope}, SignedHeaders={_SIGNED_HEADERS_STR}, Signature={signature}"
)
return {
"Content-Type": _CONTENT_TYPE,
"Host": host,
"X-Content-Sha256": payload_hash,
"X-Date": x_date,
"Authorization": authorization,
}
# ── 错误分类 ──────────────────────────────────────────────────────────
# 即梦业务码 -> 是否可重试映射
_JIMENG_RETRYABLE_CODES = {50511, 50516, 50429, 50430, 50500, 50501}
_JIMENG_NON_RETRYABLE_CODES = {50411, 50412, 50413, 50512, 50513, 50514}
def _classify_jimeng_error(status_code: int, body: str, biz_code: int | None = None) -> tuple[str, str, bool]:
"""即梦错误分类,返回 (error_code, user_message, is_retryable)。"""
code = biz_code if biz_code is not None else 0
body_lower = (body or "").lower()
# 业务码优先
if code == 50411:
return (
"portrait_intercept",
"即梦通道:参考图片前审核未通过(Pre Img Risk Not Pass),请更换参考图后重试。",
False,
)
if code == 50511:
return "task_failed", "即梦通道:输出图片后审核未通过,可稍后重试。", True
if code in (50412, 50413, 50512):
return "invalid_param", "即梦通道:提示词或文本审核不通过,请调整文案后重试。", False
if code == 50516:
return "task_failed", "即梦通道:输出视频后审核未通过,可稍后重试。", True
if code in (50429, 50430):
return "rate_limit", "即梦通道:QPS/并发超限,请稍等 1-2 分钟后重试。", True
if code in (50500, 50501):
return "network_error", "即梦通道:服务内部错误,可稍后重试。", True
# HTTP 层兜底
if status_code in (401, 403):
return "auth_error", "即梦通道:AK/SK 鉴权失败,请联系管理员检查 JIMENG_AK/SK 配置。", False
if status_code == 429:
return "rate_limit", "即梦通道:服务限流,请稍后重试。", True
if status_code == 404:
return "model_not_found", "即梦通道:接口不存在(req_key 或 Action 错误),请联系管理员。", False
if status_code in (402, 400) and any(kw in body_lower for kw in ("quota", "billing", "insufficient", "余额")):
return "quota_exceeded", "即梦通道:账户余额/配额不足,请联系管理员充值。", False
if status_code == 400:
msg = ""
try:
msg = str(json.loads(body or "{}").get("message", "") or "")
except Exception:
pass
return "invalid_param", f"即梦通道:参数错误:{msg or body[:200]}", False
if status_code == 0:
return "network_error", "即梦通道:网络连接失败,请稍后重试。", True
# 任务内失败
if code and code != 10000:
return "unknown", f"即梦通道:视频生成失败(错误码 {code}),请稍后重试。", code in _JIMENG_RETRYABLE_CODES
detail = body[:200]
return "unknown", f"即梦通道:视频生成失败(HTTP {status_code}):{detail}", False
# ── 即梦客户端 ────────────────────────────────────────────────────────
class JimengClient:
"""火山引擎即梦视觉 API(cvtob)异步客户端,支持图生视频首帧(jimeng_i2v_first_v30)。"""
def __init__(self) -> None:
settings = get_shared_settings()
self.ak: str = getattr(settings, "jimeng_ak", "") or os.getenv("JIMENG_AK", "")
self.sk: str = getattr(settings, "jimeng_sk", "") or os.getenv("JIMENG_SK", "")
self.base_url: str = (getattr(settings, "jimeng_base_url", "") or "https://visual.volcengineapi.com").rstrip(
"/"
)
self.req_key: str = getattr(settings, "jimeng_req_key", "") or "jimeng_i2v_first_v30"
self.poll_interval: int = int(getattr(settings, "jimeng_video_poll_interval", 5) or 5)
self.total_timeout: int = int(getattr(settings, "jimeng_video_timeout", 600) or 600)
self.max_retries: int = 2
self.last_video_error: dict = {}
# 解析 base_url 里的 host(用于签名 Host 头)
parsed = urlparse(self.base_url)
self.host: str = parsed.netloc or "visual.volcengineapi.com"
@property
def is_available(self) -> bool:
return bool(self.ak and self.sk)
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 "",
"provider": "jimeng",
**extra,
}
# ── 内部 HTTP:签名 + 请求 ──────────────────────────────────────
def _signed_request(
self,
method: str,
action: str,
body_obj: dict[str, Any],
timeout: float = 60.0,
) -> tuple[int, str, dict]:
"""发送一次带 V4 签名的请求,返回 (status_code, body_text, parsed_json)。"""
body_bytes = json.dumps(body_obj, ensure_ascii=False).encode("utf-8")
query = {"Action": action, "Version": _JIMENG_VERSION}
x_date = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
headers = _sign_v4(self.ak, self.sk, method, self.host, query, body_bytes, x_date)
url = f"{self.base_url}/?{_norm_query(query)}"
resp = httpx.request(
method,
url,
headers=headers,
content=body_bytes,
timeout=timeout,
)
sc = int(getattr(resp, "status_code", 0) or 0)
text = getattr(resp, "text", "") or ""
try:
data = resp.json()
except Exception:
data = {}
return sc, text, data
# ── 提交任务 ────────────────────────────────────────────────────
def _submit_task(
self,
prompt: str,
image_url: str,
frames: int = 121,
seed: int = -1,
) -> str | None:
"""提交图生视频任务,成功返回 task_id;失败写 last_video_error 并返回 None。"""
body: dict[str, Any] = {
"req_key": self.req_key,
"prompt": prompt.strip()[:800],
"image_urls": [image_url],
"seed": int(seed) if seed and seed > 0 else -1,
"frames": int(frames),
}
last_sc = 0
last_body = ""
for attempt in range(self.max_retries + 1):
try:
sc, text, data = self._signed_request("POST", _ACTION_SUBMIT, body, timeout=60.0)
last_sc, last_body = sc, text
if sc >= 400:
logger.error("[jimeng] 提交 HTTP %d: %s", sc, text[:500])
if sc >= 500 and attempt < self.max_retries:
time.sleep(0.8 * (2**attempt))
continue
biz_code = data.get("code") if isinstance(data, dict) else None
err_code, user_msg, _ = _classify_jimeng_error(sc, text, biz_code)
self._set_error(err_code, user_msg, sc, text, req_key=self.req_key)
return None
code = data.get("code") if isinstance(data, dict) else None
if code == 10000:
d = data.get("data") or {}
tid = d.get("task_id")
if tid:
return str(tid)
err_code, user_msg, retry = _classify_jimeng_error(sc, text, code)
logger.error(
"[jimeng] 提交业务错误 code=%s msg=%s",
code,
(data.get("message") if isinstance(data, dict) else ""),
)
if retry and attempt < self.max_retries:
time.sleep(0.8 * (2**attempt))
continue
self._set_error(err_code, user_msg, sc, text, req_key=self.req_key, biz_code=code)
return None
except _HTTP_NETWORK_ERRORS as ne:
last_sc, last_body = 0, f"network error: {ne}"
logger.warning(
"[jimeng] 提交网络异常 %s,重试 %d/%d", type(ne).__name__, attempt + 1, self.max_retries + 1
)
if attempt < self.max_retries:
time.sleep(0.8 * (2**attempt))
continue
self._set_error("network_error", "即梦通道:提交任务网络异常,请稍后重试。", 0, str(ne))
return None
except Exception as e:
last_sc, last_body = 0, f"exception: {e}"
logger.error("[jimeng] 提交异常: %s", e, exc_info=True)
if attempt < self.max_retries:
time.sleep(0.8 * (2**attempt))
continue
self._set_error("unknown", f"即梦通道:提交任务异常:{e!s}"[:200], 0, str(e))
return None
if not self.last_video_error:
err_code, user_msg, _ = _classify_jimeng_error(last_sc, last_body)
self._set_error(err_code, user_msg, last_sc, last_body)
return None
# ── 轮询结果 ────────────────────────────────────────────────────
def _poll_result(self, task_id: str) -> str | None:
"""轮询任务直到 done/failed/expired/timeout,成功返回 video_url。"""
deadline = time.time() + self.total_timeout
poll_count = 0
last_status = ""
poll_body = {"req_key": self.req_key, "task_id": task_id}
while time.time() < deadline:
poll_count += 1
try:
sc, text, data = self._signed_request("POST", _ACTION_POLL, poll_body, timeout=30.0)
if sc >= 400:
logger.warning("[jimeng] 轮询 HTTP %d: %s", sc, text[:300])
if poll_count < 3:
time.sleep(self.poll_interval)
continue
err_code, user_msg, _ = _classify_jimeng_error(sc, text)
self._set_error(err_code, user_msg, sc, text, task_id=task_id)
return None
code = data.get("code") if isinstance(data, dict) else None
d = data.get("data") if isinstance(data, dict) else None
if code != 10000 or not isinstance(d, dict):
err_code, user_msg, retry = _classify_jimeng_error(sc, text, code)
logger.error(
"[jimeng] 轮询业务错误 task=%s code=%s msg=%s",
task_id,
code,
(data.get("message") if isinstance(data, dict) else ""),
)
if retry and poll_count < 3:
time.sleep(self.poll_interval)
continue
self._set_error(err_code, user_msg, sc, text, task_id=task_id, biz_code=code)
return None
status = d.get("status", "") or ""
last_status = status
if status == "done":
video_url = d.get("video_url") or ""
if video_url:
logger.info("[jimeng] 任务 %s 完成 polls=%d", task_id, poll_count)
return str(video_url)
logger.error("[jimeng] 任务 %s done 但无 video_url: %s", task_id, str(d)[:500])
self._set_error(
"unknown",
"即梦通道:任务成功但未返回视频URL,请联系管理员。",
200,
str(d)[:500],
task_id=task_id,
)
return None
if status in ("not_found", "expired"):
logger.error("[jimeng] 任务 %s 状态 %s", task_id, status)
self._set_error(
"network_error" if status == "expired" else "unknown",
f"即梦通道:任务{'已过期' if status == 'expired' else '未找到'},请重新提交。",
200,
f"task {status}",
task_id=task_id,
)
return None
if poll_count % 6 == 0:
logger.info("[jimeng] 轮询中 task=%s status=%s polls=%d", task_id, status, poll_count)
except _HTTP_NETWORK_ERRORS as ne:
logger.warning("[jimeng] 轮询网络异常 %s", ne)
except Exception as e:
logger.debug("[jimeng] 轮询异常: %s", e)
time.sleep(self.poll_interval)
logger.error(
"[jimeng] 任务 %s 轮询超时(%ds)polls=%d last_status=%s",
task_id,
self.total_timeout,
poll_count,
last_status,
)
self._set_error(
"network_error",
f"即梦通道:视频生成超时(>{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
# ── 下载视频 ────────────────────────────────────────────────────
def _download_video(self, video_url: str, output_dir: str, task_id: str) -> str | None:
os.makedirs(output_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(output_dir, f"jimeng_{safe_tid}_{uuid.uuid4().hex[:8]}{suffix}")
try:
with httpx.stream("GET", video_url, timeout=180, follow_redirects=True) as resp:
dsc = int(getattr(resp, "status_code", 0) or 0)
if dsc >= 400:
logger.error("[jimeng] 下载 HTTP %d", dsc)
self._set_error("network_error", "即梦通道:视频下载失败(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("[jimeng] 下载视频失败: %s", e, exc_info=True)
self._set_error("network_error", f"即梦通道:视频下载失败:{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("[jimeng] 下载文件过小: %d bytes", size)
self._set_error("unknown", "即梦通道:视频下载文件过小,请稍后重试。", 0, f"downloaded only {size} bytes")
try:
os.remove(out_path)
except Exception:
pass
return None
logger.info("[jimeng] 视频已下载: %s (%d bytes)", out_path, size)
return out_path
# ── 对外主入口 ──────────────────────────────────────────────────
def video_generation(
self,
prompt: str,
*,
image_url: str,
duration: int = 5,
ratio: str | None = "9:16",
resolution: str = "720p",
output_dir: str | None = None,
generate_audio: bool = False,
) -> dict | None:
"""即梦图生视频主入口。
成功返回 {"video_path": str, "usage": {"provider","duration_seconds","frames","req_key","billing_mode"}};
失败返回 None,详情在 self.last_video_error。
注意:jimeng_i2v_first_v30 不支持原生音频(generate_audio 被忽略,返回无声视频),
音频由后续 ffmpeg 合成阶段叠加 TTS。
支持时长:5s(frames=121)/10s(frames=241),>10s 截断并打 warning。
分辨率固定 720P;ratio 对首帧 i2v 无效(自动按图片比例)。
"""
self.last_video_error = {}
if not self.is_available:
self._set_error(
"auth_error",
"即梦通道未配置(JIMENG_AK/SK 缺失),请联系管理员。",
detail="jimeng ak/sk empty",
)
logger.error("[jimeng] AK/SK 未配置,无法调用")
return None
if not prompt or not prompt.strip():
self._set_error("invalid_param", "视频生成提示词不能为空。", detail="empty prompt")
return None
if not image_url or not image_url.strip():
self._set_error("invalid_param", "即梦图生视频必须提供参考图片。", detail="empty image_url")
return None
dur = int(duration or 5)
if dur <= 5:
frames = 121
real_dur = 5
elif dur <= 10:
frames = 241
real_dur = 10
else:
logger.warning("[jimeng] 请求时长 %ds 超出即梦 i2v 上限 10s,截断到 10s(frames=241)", dur)
frames = 241
real_dur = 10
out_dir = output_dir or "/tmp"
logger.info(
"[jimeng] 提交任务: req_key=%s dur=%ds(frames=%d) ratio=%s res=%s gen_audio=%s img=%s",
self.req_key,
real_dur,
frames,
ratio,
resolution,
generate_audio,
bool(image_url),
)
task_id = self._submit_task(prompt=prompt, image_url=image_url, frames=frames, seed=-1)
if not task_id:
return None
logger.info("[jimeng] 任务已提交: task_id=%s", task_id)
video_url = self._poll_result(task_id)
if not video_url:
return None
local_path = self._download_video(video_url, out_dir, task_id)
if not local_path:
return None
usage = {
"provider": "jimeng",
"duration_seconds": real_dur,
"frames": frames,
"req_key": self.req_key,
"billing_mode": "per_second",
}
return {"video_path": local_path, "usage": usage}
def get_jimeng_client() -> "JimengClient | None":
"""返回即梦客户端单例;未配置 AK/SK 时返回 None。"""
global _JIMENG_CLIENT_SINGLETON
if _JIMENG_CLIENT_SINGLETON is None:
_JIMENG_CLIENT_SINGLETON = JimengClient()
if not _JIMENG_CLIENT_SINGLETON.is_available:
return None
return _JIMENG_CLIENT_SINGLETON
+7 -76
View File
@@ -563,7 +563,7 @@ class TestResolveVideoModelId:
class TestVideoGenerationLastError:
def test_create_400_portrait_returns_user_message(self, tmp_path):
"""HTTP 400 + 真人拦截关键词 → portrait_intercept 错误码,用户提示友好。"""
"""#2169: HTTP 400 + 真人拦截关键词 → 自动尝试即梦兜底;即梦未配时返回 portrait_intercept。"""
client = _make_client(max_retries=0)
create_resp = MagicMock()
create_resp.status_code = 400
@@ -576,6 +576,8 @@ class TestVideoGenerationLastError:
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
patch("packages.shared.ai_client.time.sleep", return_value=None),
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
# jimeng 未配置,fallback 后仍返回 portrait_intercept(提示用户需要配置即梦)
patch("packages.shared.jimeng_client.get_jimeng_client", return_value=None),
):
mock_s.return_value = MagicMock(
doubao_video_poll_interval=0, doubao_video_timeout=1, doubao_video_model="seedance"
@@ -584,8 +586,10 @@ class TestVideoGenerationLastError:
assert result is None
err = client.get_last_video_error()
assert err["error_code"] == "portrait_intercept"
assert "真人" in err["user_message"]
assert err["status_code"] == 400
# 即梦兜底未启用时提示包含"真人照片"/"即梦"等关键字
assert "真人" in err["user_message"] or "肖像" in err["user_message"] or "即梦" in err["user_message"]
# 方舟本身 status_code=400(最后一个错误可能被即梦兜底覆盖,但 error_code 不变)
assert err["status_code"] in (0, 400)
def test_create_401_returns_auth_error(self, tmp_path):
client = _make_client(max_retries=0)
@@ -656,76 +660,3 @@ class TestAiServiceLastVideoError:
err = ai_service.get_last_video_error()
assert err["error_code"] == "unknown"
assert "user_message" in err
class TestChatCompletion404Fallback:
"""chat_completion / vision_completion 404 自动 fallback 到默认模型。"""
def test_chat_404_fallback_to_default_model(self, monkeypatch):
"""指定模型 404 时自动用 self.model 重试一次。"""
from packages.shared import ai_client as ac_mod
calls = []
class FakeResp:
def __init__(self, sc, body):
self.status_code = sc
self._body = body
def raise_for_status(self):
import httpx
if self.status_code >= 400:
raise httpx.HTTPStatusError("x", request=object(), response=self)
def json(self):
return self._body
@property
def text(self):
import json as _json
return _json.dumps(self._body)
def fake_post(url, headers=None, json=None, timeout=None):
calls.append(json.get("model"))
if json.get("model") == "doubao-1-5-pro-32k-250115":
return FakeResp(404, {"error": {"message": "model not found"}})
return FakeResp(200, {"choices": [{"message": {"content": "ok-from-default"}}]})
monkeypatch.setattr(ac_mod.httpx, "post", fake_post)
c = ac_mod.DoubaoClient()
c.api_key = "test"
c.model = "doubao-seed-1-6-250615"
c.max_retries = 0
out = c.chat_completion([{"role": "user", "content": "hi"}], model="doubao-1-5-pro-32k-250115")
assert out == "ok-from-default"
assert calls == ["doubao-1-5-pro-32k-250115", "doubao-seed-1-6-250615"]
def test_chat_401_does_not_retry(self, monkeypatch):
"""401(鉴权失败)不重试也不 fallback,直接返回 None。"""
from packages.shared import ai_client as ac_mod
calls = []
class FakeResp:
def __init__(self, sc):
self.status_code = sc
def raise_for_status(self):
import httpx
raise httpx.HTTPStatusError("x", request=object(), response=self)
def fake_post(url, headers=None, json=None, timeout=None):
calls.append(json.get("model"))
return FakeResp(401)
monkeypatch.setattr(ac_mod.httpx, "post", fake_post)
c = ac_mod.DoubaoClient()
c.api_key = "test"
c.model = "doubao-seed-1-6-250615"
c.max_retries = 0
out = c.chat_completion([{"role": "user", "content": "hi"}], model="doubao-1-5-pro-32k-250115")
assert out is None
assert calls == ["doubao-1-5-pro-32k-250115"] # 不重试不 fallback(401 是全局鉴权错,fallback 也没用)
+376
View File
@@ -0,0 +1,376 @@
"""tests for packages/shared/jimeng_client.py (#2169 即梦 i2v 客户端)."""
from __future__ import annotations
import json
import os
from unittest.mock import MagicMock, mock_open, patch
import pytest
@pytest.fixture(autouse=True)
def reset_singleton():
import packages.shared.jimeng_client as j
j._JIMENG_CLIENT_SINGLETON = None
yield
j._JIMENG_CLIENT_SINGLETON = None
def _make_settings(ak="test-ak", sk="test-sk", req_key="jimeng_i2v_first_v30", timeout=60, poll_interval=2):
return MagicMock(
jimeng_ak=ak,
jimeng_sk=sk,
jimeng_base_url="https://visual.volcengineapi.com",
jimeng_req_key=req_key,
jimeng_video_timeout=timeout,
jimeng_video_poll_interval=poll_interval,
)
# ── V4 签名单元测试 ──────────────────────────────────────────────────
class TestV4Signature:
def test_sign_returns_required_headers(self):
from packages.shared.jimeng_client import _sign_v4
headers = _sign_v4(
ak="AK_TEST",
sk="SK_TEST",
method="POST",
host="visual.volcengineapi.com",
query={"Action": "CVSync2AsyncSubmitTask", "Version": "2022-08-31"},
body_bytes=b'{"req_key":"jimeng_i2v_first_v30"}',
x_date="20260101T120000Z",
)
assert headers["Content-Type"] == "application/json"
assert headers["Host"] == "visual.volcengineapi.com"
assert headers["X-Date"] == "20260101T120000Z"
assert "X-Content-Sha256" in headers
assert headers["Authorization"].startswith("HMAC-SHA256 Credential=AK_TEST/20260101/cn-north-1/cv/request")
assert "SignedHeaders=content-type;host;x-content-sha256;x-date" in headers["Authorization"]
assert "Signature=" in headers["Authorization"]
# 签名是 64 字符 hex
sig = headers["Authorization"].split("Signature=")[-1]
assert len(sig) == 64
assert all(c in "0123456789abcdef" for c in sig)
def test_sign_deterministic(self):
"""相同输入必须产生相同签名(幂等)。"""
from packages.shared.jimeng_client import _sign_v4
kwargs = dict(
ak="AK",
sk="SK",
method="POST",
host="h",
query={"A": "1", "B": "2"},
body_bytes=b"{}",
x_date="20260101T000000Z",
)
h1 = _sign_v4(**kwargs)
h2 = _sign_v4(**kwargs)
assert h1["Authorization"] == h2["Authorization"]
assert h1["X-Content-Sha256"] == h2["X-Content-Sha256"]
def test_sign_different_body_different_sig(self):
from packages.shared.jimeng_client import _sign_v4
base = dict(ak="AK", sk="SK", method="POST", host="h", query={}, x_date="20260101T000000Z")
h1 = _sign_v4(body_bytes=b"a", **base)
h2 = _sign_v4(body_bytes=b"b", **base)
assert h1["Authorization"] != h2["Authorization"]
def test_payload_sha256_matches(self):
import hashlib
from packages.shared.jimeng_client import _sign_v4
body = b'{"prompt":"hello"}'
h = _sign_v4("ak", "sk", "POST", "h", {}, body, "20260101T000000Z")
expected = hashlib.sha256(body).hexdigest()
assert h["X-Content-Sha256"] == expected
def test_norm_query_sorted_and_encoded(self):
from packages.shared.jimeng_client import _norm_query
q = _norm_query({"B": "2", "A": "1", "C": "a b"})
# key 排序 + 空格→%20
assert q.startswith("A=1")
assert "B=2" in q
assert "C=a%20b" in q
# ── 可用性 / 单例 ─────────────────────────────────────────────────────
class TestAvailability:
def test_unavailable_without_ak_sk(self):
from packages.shared.jimeng_client import JimengClient, get_jimeng_client
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings(ak="", sk="")
# 重置单例
import packages.shared.jimeng_client as j
j._JIMENG_CLIENT_SINGLETON = None
assert get_jimeng_client() is None
def test_available_with_ak_sk(self):
from packages.shared.jimeng_client import get_jimeng_client
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings()
import packages.shared.jimeng_client as j
j._JIMENG_CLIENT_SINGLETON = None
c = get_jimeng_client()
assert c is not None
assert c.is_available is True
assert c.req_key == "jimeng_i2v_first_v30"
# ── 错误分类 ────────────────────────────────────────────────────────
class TestClassifyError:
def test_50411_is_portrait_intercept_non_retryable(self):
from packages.shared.jimeng_client import _classify_jimeng_error
code, msg, retry = _classify_jimeng_error(200, '{"code":50411,"message":"Pre Img Risk"}', 50411)
assert code == "portrait_intercept"
assert retry is False
def test_50429_is_rate_limit_retryable(self):
from packages.shared.jimeng_client import _classify_jimeng_error
code, msg, retry = _classify_jimeng_error(200, "", 50429)
assert code == "rate_limit"
assert retry is True
def test_50430_is_rate_limit(self):
from packages.shared.jimeng_client import _classify_jimeng_error
code, _, _ = _classify_jimeng_error(200, "", 50430)
assert code == "rate_limit"
def test_50500_is_network_error_retryable(self):
from packages.shared.jimeng_client import _classify_jimeng_error
code, _, retry = _classify_jimeng_error(200, "", 50500)
assert code == "network_error"
assert retry is True
def test_50412_is_invalid_param_non_retryable(self):
from packages.shared.jimeng_client import _classify_jimeng_error
code, _, retry = _classify_jimeng_error(200, "", 50412)
assert code == "invalid_param"
assert retry is False
def test_401_auth(self):
from packages.shared.jimeng_client import _classify_jimeng_error
code, msg, retry = _classify_jimeng_error(401, "auth fail", None)
assert code == "auth_error"
assert retry is False
def test_400_text_audit(self):
from packages.shared.jimeng_client import _classify_jimeng_error
code, _, _ = _classify_jimeng_error(400, "text error", None)
assert code == "invalid_param"
# ── video_generation 主流程 ──────────────────────────────────────────
class TestVideoGenerationHappyPath:
def test_missing_ak_returns_none(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings(ak="", sk="")
c = JimengClient()
assert c.video_generation("hi", image_url="http://x/y.jpg") is None
err = c.last_video_error
assert err["error_code"] == "auth_error"
def test_empty_prompt_returns_none(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings()
c = JimengClient()
assert c.video_generation(" ", image_url="http://x/y.jpg") is None
assert c.last_video_error["error_code"] == "invalid_param"
def test_empty_image_url_returns_none(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings()
c = JimengClient()
assert c.video_generation("prompt", image_url="") is None
assert c.last_video_error["error_code"] == "invalid_param"
def test_duration_5s_frames_121(self):
"""5s → frames=121,10s→frames=241,>10s 截断到10s。"""
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings(timeout=1, poll_interval=0)
c = JimengClient()
captured = {}
def fake_submit(prompt, image_url, frames, seed=-1):
captured["frames"] = frames
return "task-xyz"
def fake_poll(tid):
captured["tid"] = tid
return "http://example.com/v.mp4"
def fake_download(url, out_dir, tid):
captured["url"] = url
return "/tmp/fake.mp4"
# 构造一个假文件
os.makedirs("/tmp", exist_ok=True)
with open("/tmp/fake.mp4", "wb") as f:
f.write(b"x" * 2048)
with (
patch.object(c, "_submit_task", side_effect=fake_submit),
patch.object(c, "_poll_result", side_effect=fake_poll),
patch.object(c, "_download_video", side_effect=fake_download),
):
r = c.video_generation("test", image_url="http://x/y.jpg", duration=5, output_dir="/tmp")
assert r is not None
assert captured["frames"] == 121
assert r["usage"]["duration_seconds"] == 5
assert r["usage"]["billing_mode"] == "per_second"
assert r["usage"]["req_key"] == "jimeng_i2v_first_v30"
def test_duration_10s_frames_241(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings(timeout=1, poll_interval=0)
c = JimengClient()
captured = {}
def fake_submit(prompt, image_url, frames, seed=-1):
captured["frames"] = frames
return "tid"
def fake_poll(tid):
return "http://x/v.mp4"
def fake_download(url, out_dir, tid):
with open("/tmp/fake2.mp4", "wb") as f:
f.write(b"x" * 2048)
return "/tmp/fake2.mp4"
with (
patch.object(c, "_submit_task", side_effect=fake_submit),
patch.object(c, "_poll_result", side_effect=fake_poll),
patch.object(c, "_download_video", side_effect=fake_download),
):
r = c.video_generation("hi", image_url="http://x/y.jpg", duration=10, output_dir="/tmp")
assert captured["frames"] == 241
assert r["usage"]["duration_seconds"] == 10
def test_duration_over_10s_truncates_to_10s(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings(timeout=1, poll_interval=0)
c = JimengClient()
captured = {}
def fake_submit(prompt, image_url, frames, seed=-1):
captured["frames"] = frames
return "tid"
def fake_poll(tid):
return "http://x/v.mp4"
def fake_download(url, out_dir, tid):
with open("/tmp/fake3.mp4", "wb") as f:
f.write(b"x" * 2048)
return "/tmp/fake3.mp4"
with (
patch.object(c, "_submit_task", side_effect=fake_submit),
patch.object(c, "_poll_result", side_effect=fake_poll),
patch.object(c, "_download_video", side_effect=fake_download),
):
r = c.video_generation("hi", image_url="http://x/y.jpg", duration=30, output_dir="/tmp")
assert captured["frames"] == 241
assert r["usage"]["duration_seconds"] == 10
class TestSubmitTaskErrors:
def test_submit_50411_writes_portrait_intercept(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings()
c = JimengClient()
fake_resp = MagicMock(status_code=200, text='{"code":50411,"message":"Pre Img Risk Not Pass"}')
fake_resp.json.return_value = {"code": 50411, "message": "Pre Img Risk Not Pass"}
with patch("packages.shared.jimeng_client.httpx.request", return_value=fake_resp):
tid = c._submit_task("p", "http://x/y.jpg", frames=121)
assert tid is None
assert c.last_video_error["error_code"] == "portrait_intercept"
def test_submit_returns_task_id(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings()
c = JimengClient()
fake_resp = MagicMock(status_code=200, text='{"code":10000,"data":{"task_id":"abc"}}')
fake_resp.json.return_value = {"code": 10000, "data": {"task_id": "abc"}}
with patch("packages.shared.jimeng_client.httpx.request", return_value=fake_resp):
tid = c._submit_task("p", "http://x/y.jpg", frames=121)
assert tid == "abc"
class TestPollResult:
def test_poll_done_returns_video_url(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings(timeout=10, poll_interval=0)
c = JimengClient()
done_resp = MagicMock(status_code=200)
done_resp.json.return_value = {"code": 10000, "data": {"status": "done", "video_url": "http://x/v.mp4"}}
with (
patch("packages.shared.jimeng_client.httpx.request", return_value=done_resp),
patch("packages.shared.jimeng_client.time.sleep"),
):
url = c._poll_result("abc")
assert url == "http://x/v.mp4"
def test_poll_timeout_returns_none(self):
from packages.shared.jimeng_client import JimengClient
with patch("packages.shared.jimeng_client.get_shared_settings") as ms:
ms.return_value = _make_settings(timeout=1, poll_interval=0)
c = JimengClient()
queue_resp = MagicMock(status_code=200)
queue_resp.json.return_value = {"code": 10000, "data": {"status": "in_queue"}}
# time.time 会被调用,模拟超时
with (
patch("packages.shared.jimeng_client.httpx.request", return_value=queue_resp),
patch("packages.shared.jimeng_client.time.sleep"),
):
url = c._poll_result("abc")
assert url is None
assert c.last_video_error["error_code"] == "network_error"
assert "超时" in c.last_video_error["user_message"]