Compare commits

..

1 Commits

Author SHA1 Message Date
xiaoxia-agent e5cdb93c78 fix(viral-video) P0: 修复GET /models响应包装导致页面白屏,数组操作加防御
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
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 / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker 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 Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build API 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 / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m25s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 1m32s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m43s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m30s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 3m9s
AI Code Review / AI Code Review (pull_request) Successful in 7m10s
CI/CD Pipeline / Validate - Style (pull_request) Successful in 11m56s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 10m42s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 14m20s
CI/CD Pipeline / Validate - Security (pull_request) Successful in 33m4s
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 / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 1s
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 2m22s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Successful in 5m21s
根因:后端 GET /api/v1/viral-video/models 返回 {"models": [...]} 包装对象,
前端 getViralVideoModels() 声明返回 ViralVideoModel[] 并在 .then 直接对 r.data 调 .some(),
实际 r.data 是对象而非数组,导致 TypeError: s.some is not a function 白屏。

修复:
- 新增 ViralVideoModelsResponse 类型
- getViralVideoModels() 改为泛型 ViralVideoModelsResponse,解析时兼容数组/包装两种返回
- 模型加载 useEffect:list/supported_resolutions 加 Array.isArray 防御
- 模型切换 onChange:m.supported_resolutions 加防御兜底 [480p,720p,1080p]
- 清晰度 Select 过滤池:curM?.supported_resolutions 加 Array.isArray 守卫
- options 渲染:filter 掉非对象空项,label fallback 到 key
- max_duration 加 typeof === 'number' 防御
2026-10-03 19:35:15 +08:00
21 changed files with 139 additions and 2149 deletions
-14
View File
@@ -220,20 +220,6 @@ 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(暂停积分系统)。
+12 -15
View File
@@ -145,22 +145,19 @@ def get_rules(
def get_packages(
current_user: AuthenticatedUser = Depends(get_current_user),
):
"""查询可购买的积分包列表(读管理后台 credit_packages 表真实数据)。
仅返回 is_active=true;后台改价/启停后最多 30 秒生效。
"""
from packages.application.catalog.admin_catalog import get_points_packages
packages = [
PointsPackageItem(
code=row["code"],
name=row["name"],
points=row["points"],
price_cents=row["price_cents"],
unit_price=row["unit_price"],
"""查询可购买的积分包列表。"""
packages = []
for code, pkg in POINTS_PACKAGES.items():
unit_price = f"¥{pkg['price_cents'] / 100 / pkg['points']:.3f}/积分"
packages.append(
PointsPackageItem(
code=code,
name=pkg["name"],
points=pkg["points"],
price_cents=pkg["price_cents"],
unit_price=unit_price,
)
)
for row in get_points_packages()
]
mt = _member_type(current_user)
discount = MEMBER_DISCOUNT.get(mt) if mt else None
return PointsPackagesResponse(packages=packages, user_discount=discount)
+24 -4
View File
@@ -86,13 +86,33 @@ async def get_current_subscription(
def list_membership_plans(
current_user: AuthenticatedUser = Depends(get_current_user),
) -> dict[str, list[dict[str, Any]]]:
"""查询可购买的会员套餐(读管理后台 plans 表真实数据)。
"""查询所有会员档位(供前端会员购买页展示)。
仅返回 is_enabled=true 的套餐;后台启停/改价后最多 30 秒生效。
返回 points 积分体系下的会员档位(月卡/季卡/年卡),含价格、时长、积分折扣等信息。
"""
from packages.application.catalog.admin_catalog import get_membership_plans
from packages.domain.points_rules import MEMBER_DISCOUNT, MEMBERSHIP_PRICES
return {"plans": get_membership_plans()}
plans: list[dict[str, Any]] = []
for plan_id, info in MEMBERSHIP_PRICES.items():
days = info["duration_days"]
monthly_cents = round(info["price_cents"] * 30 / days)
features: dict[str, Any] = {"max_resolution": "1080p"}
if plan_id == MembershipType.MONTHLY:
features.update({"free_clips_daily": 2})
elif plan_id == MembershipType.QUARTERLY:
features.update({"free_clips_daily": 5})
elif plan_id == MembershipType.YEARLY:
features.update({"free_clips_daily": "unlimited"})
plans.append({
"plan_id": plan_id,
"name": info["name"],
"price_cents": info["price_cents"],
"monthly_price_cents": monthly_cents,
"duration_days": days,
"points_discount": MEMBER_DISCOUNT.get(plan_id, 1.0),
"features": features,
})
return {"plans": plans}
@router.get("/billing-records", response_model=list[BillingRecord])
+2 -22
View File
@@ -80,8 +80,6 @@ const AiAvatarPage: React.FC = () => {
const [finalizeLoading, setFinalizeLoading] = useState(false)
/* ── 对口型轮询 ── */
/** 对口型轮询总时长上限(10分钟):超过后停止轮询并提示去历史记录查看 */
const LIPSYNC_POLL_MAX_MS = 10 * 60 * 1000
const lipsyncTimerRef = useRef<ReturnType<typeof setInterval> | null>(null)
/* ── 渲染进度轮询 ── */
const renderTimerRef = useRef<ReturnType<typeof setInterval> | null>(null)
@@ -272,24 +270,7 @@ const AiAvatarPage: React.FC = () => {
// 如果是预合成模式,后端会同步把状态置为 submitted(甚至可能已返回 running),
// 但仍需轮询等 completed
if (lipsyncTimerRef.current) clearInterval(lipsyncTimerRef.current)
// 轮询间隔 5 秒;单请求超时 5 分钟(见 api/aiAvatar.ts);总轮询上限 10 分钟
// 单次请求失败/超时不中断轮询,继续下一轮;超过总上限后停止并提示用户去历史记录查看
lipsyncTimerRef.current = setInterval(async () => {
// 总时长保护:超过 10 分钟停止轮询
if (Date.now() - lipsyncStartAtRef.current > LIPSYNC_POLL_MAX_MS) {
if (lipsyncTimerRef.current) {
clearInterval(lipsyncTimerRef.current)
lipsyncTimerRef.current = null
}
if (lipsyncTickRef.current) {
clearInterval(lipsyncTickRef.current)
lipsyncTickRef.current = null
}
setLipsyncStatus("failed")
setLipsyncErrorMessage("渲染时间较长,请稍后在历史记录中查看")
message.warning("对口型渲染时间较长,已停止自动刷新,请稍后在历史记录中查看")
return
}
try {
const updated = await getLipsyncJob(job.id)
state.setLipsyncJob(updated)
@@ -315,10 +296,9 @@ const AiAvatarPage: React.FC = () => {
setLipsyncErrorMessage(updated.error_message || "对口型生成失败")
}
} catch (err) {
// 单次轮询失败(含 timeout):不中断轮询,打印日志后等下一轮
console.warn("[对口型] 轮询请求失败,将继续下一轮:", err)
console.error("[对口型] 轮询错误:", err)
}
}, 5000)
}, 3000)
} catch (err) {
console.error("[对口型] 创建失败:", {
status: (err as { response?: { status?: number } })?.response?.status,
+2 -6
View File
@@ -72,8 +72,7 @@ export const previewTts = async (data: {
}
export const getLipsyncJob = async (id: string): Promise<LipsyncJob> => {
// MuseTalk 渲染 8s 视频约 54s + 排队时间,给足 5 分钟超时避免单次轮询 AxiosError 中断
const response = await apiClient.get<LipsyncJob>(`/lipsync/jobs/${id}`, { timeout: 300_000 })
const response = await apiClient.get<LipsyncJob>(`/lipsync/jobs/${id}`, { timeout: 60000 })
return response.data
}
@@ -92,10 +91,7 @@ export const submitRender = async (data: {
}
export const getRenderJob = async (jobId: string): Promise<RenderJob> => {
// 渲染链路(对口型+B-roll+标题+合成+上传)耗时较长,给足 5 分钟超时
const response = await apiClient.get<RenderJob>(`/ai-avatar/render/${jobId}`, {
timeout: 300_000,
})
const response = await apiClient.get<RenderJob>(`/ai-avatar/render/${jobId}`, { timeout: 60000 })
return response.data
}
+12 -50
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:
@@ -1162,7 +1162,6 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
)
logger.info("[爆款视频] Seedance prompt (前300字): %s", prompt[:300])
# 第一次调用:带参考图/首帧/音频/参考视频
result = call_video_generation(
prompt=prompt,
image_url=first_image,
@@ -1176,52 +1175,15 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
reference_audios=ref_audios,
reference_videos=ref_videos,
)
# #2169: 真人/肖像拦截已由 ai_client 内部自动切即梦(jimeng-3.0)通道重试——
# 保留首帧图、不走"去掉参考图纯 t2v 降级"(用户明确要求按参考照片生成)。
# 即梦也失败或非拦截类错误时,直接抛错给上层展示用户友好提示。
def _check_and_reraise(result):
if result and isinstance(result, dict):
return result
from packages.shared.ai_service import get_last_video_error
err = get_last_video_error() or {}
user_msg = err.get("user_message") or ""
detail = err.get("detail") or ""
err_code = err.get("error_code") or "unknown"
status_code = err.get("status_code", 0)
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",
err_provider,
model or "default",
err_code,
status_code,
user_msg,
(detail or "")[:500],
)
raise RuntimeError(err_msg)
if not result or not isinstance(result, dict):
_check_and_reraise(result)
raise RuntimeError("Seedance 视频生成失败:返回为空")
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("视频生成失败:返回空文件或路径不存在")
# #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)
raise RuntimeError("Seedance 视频生成失败:返回空文件或路径不存在")
logger.info(
"[爆款视频] Seedance 单次生成完成: %s size=%d usage=%s", video_path, Path(video_path).stat().st_size, usage
)
return str(video_path), (usage if isinstance(usage, dict) else None)
@@ -39,9 +39,6 @@ class SQLAlchemyUserRepository(UserRepository):
model.phone_verified = user.phone_verified
model.binding_completed_at = user.binding_completed_at
model.profile_completed = user.profile_completed
model.is_member = user.is_member
model.member_type = user.member_type
model.member_expires_at = user.member_expires_at
model.created_at = user.created_at
self.session.commit()
@@ -118,8 +115,5 @@ class SQLAlchemyUserRepository(UserRepository):
phone_verified=model.phone_verified or False,
binding_completed_at=model.binding_completed_at,
profile_completed=model.profile_completed if model.profile_completed is not None else True,
is_member=model.is_member if model.is_member is not None else False,
member_type=model.member_type,
member_expires_at=model.member_expires_at,
created_at=model.created_at,
)
-1
View File
@@ -1 +0,0 @@
"""应用层:对外展示目录(套餐/积分包)。"""
@@ -1,152 +0,0 @@
"""读取管理后台配置的会员套餐 / 积分充值包(共享库真实数据)。
替代旧的硬编码 MEMBERSHIP_PRICES / POINTS_PACKAGES。
短 TTL 缓存(30 秒),后台改价/启停后用户端最多 30 秒可见。
"""
from __future__ import annotations
import threading
import time
from typing import Any
_CACHE_TTL = 30.0
_lock = threading.Lock()
_cache: dict[str, tuple[float, Any]] = {}
_QUOTA_LABELS = {
"4k": "4K 超清分辨率",
"batch_render": "批量渲染",
"priority_queue": "优先处理队列",
"ai_matting": "AI 智能抠像",
"remove_watermark": "去水印",
}
def _cached(key: str, loader):
now = time.time()
hit = _cache.get(key)
if hit and now - hit[0] < _CACHE_TTL:
return hit[1]
with _lock:
hit = _cache.get(key)
if hit and time.time() - hit[0] < _CACHE_TTL:
return hit[1]
value = loader()
_cache[key] = (time.time(), value)
return value
def _quota_features(quotas: dict[str, Any] | None) -> dict[str, Any]:
quotas = quotas or {}
features: dict[str, Any] = {}
for k, v in quotas.items():
if k == "credits_per_month":
features["credits_per_month"] = v
elif k in _QUOTA_LABELS:
features[_QUOTA_LABELS[k]] = v
else:
features[k] = v
return features
def get_membership_plans() -> list[dict[str, Any]]:
"""读取 is_enabled=true 的套餐,按年/月周期展开为用户端档位。"""
def _load() -> list[dict[str, Any]]:
from sqlalchemy import text
from packages.adapters.sqlalchemy_impl.session import SessionLocal
if SessionLocal is None:
return []
session = SessionLocal()
try:
rows = session.execute(text("""
SELECT plan_key, name, description, monthly_price, yearly_price,
quotas, display_order
FROM plans
WHERE is_enabled = TRUE
ORDER BY display_order NULLS LAST, created_at
""")).fetchall()
finally:
session.close()
plans: list[dict[str, Any]] = []
for r in rows:
base_features = _quota_features(r.quotas if isinstance(r.quotas, dict) else None)
if r.yearly_price and float(r.yearly_price) > 0:
plans.append(
{
"plan_id": r.plan_key,
"billing_cycle": "yearly",
"name": r.name,
"description": r.description,
"price_cents": int(round(float(r.yearly_price) * 100)),
"monthly_price_cents": int(round(float(r.yearly_price) * 100 / 12)),
"duration_days": 365,
"features": dict(base_features),
}
)
if r.monthly_price and float(r.monthly_price) > 0:
plans.append(
{
"plan_id": r.plan_key,
"billing_cycle": "monthly",
"name": r.name,
"description": r.description,
"price_cents": int(round(float(r.monthly_price) * 100)),
"monthly_price_cents": int(round(float(r.monthly_price) * 100)),
"duration_days": 30,
"features": dict(base_features),
}
)
return plans
return _cached("membership_plans", _load)
def get_points_packages() -> list[dict[str, Any]]:
"""读取 is_active=true 的积分充值包。"""
def _load() -> list[dict[str, Any]]:
from sqlalchemy import text
from packages.adapters.sqlalchemy_impl.session import SessionLocal
if SessionLocal is None:
return []
session = SessionLocal()
try:
rows = session.execute(text("""
SELECT package_key, name, price, credits, bonus_credits,
is_recommended, description, sort_order
FROM credit_packages
WHERE is_active = TRUE
ORDER BY sort_order NULLS LAST, price
""")).fetchall()
finally:
session.close()
packages: list[dict[str, Any]] = []
for r in rows:
total_points = int(r.credits or 0) + int(r.bonus_credits or 0)
price_cents = int(round(float(r.price) * 100))
unit = (price_cents / 100 / total_points) if total_points else 0
packages.append(
{
"code": r.package_key,
"name": r.name,
"points": total_points,
"bonus_credits": int(r.bonus_credits or 0),
"price_cents": price_cents,
"unit_price": f"¥{unit:.3f}/积分",
"is_recommended": bool(r.is_recommended),
"description": r.description,
}
)
return packages
return _cached("points_packages", _load)
+5 -22
View File
@@ -90,21 +90,15 @@ class SharedSettings(BaseSettings):
# ── 豆包大模型(火山引擎方舟) ────────────────────────────────────────
doubao_api_key: str = ""
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_model: str = "doubao-seed-1-6-250615" # 推理模型(通用兜底)
doubao_fast_model: str = "doubao-1-5-pro-32k-250115" # 快速结构化输出模型(编导脚本/意图解析/审核)
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-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_model: str = "doubao-1-5-vision-pro-250328" # 高精度视觉(备用)
doubao_vision_lite_model: str = "doubao-1-5-vision-lite-250315" # 快速视觉(商品识别默认,速度优先)
doubao_vision_use_lite: bool = True # viral-video 图片分析默认用 lite 提速
doubao_embedding_model: str = "doubao-embedding-vision-251215" # 多模态向量化(原 large-text-240915 已 Retiring)
doubao_embedding_model: str = "doubao-embedding-large-text-240915"
doubao_video_model: str = "doubao-seedance-2-5-260628"
doubao_video_timeout: int = 600 # 视频生成轮询总超时(秒)
doubao_video_poll_interval: int = 10 # 轮询间隔(秒)
@@ -115,17 +109,6 @@ 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"
-5
View File
@@ -60,11 +60,6 @@ class User:
# 资料是否已完善(微信新用户首次设置昵称后置 True;邮箱注册默认 True)
profile_completed: bool = True
# 会员字段 (#1895):与 users 表列对应
is_member: bool = False
member_type: str | None = None
member_expires_at: datetime | None = None
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
-19
View File
@@ -31,8 +31,6 @@ 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 + 服务器
@@ -153,20 +151,6 @@ 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, # 标记:不对外暴露到模型选择列表
},
}
@@ -189,9 +173,6 @@ 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"],
+29 -401
View File
@@ -22,34 +22,23 @@ 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 多模型支持,#2169 接入即梦)。
# 内部使用简短别名(seedance-2.5 / seedance-2.0 / wan-3.0 / jimeng-3.0 等)做 PRICING key;
# 视频模型 ID 解析逻辑(#2159 多模型支持)。
# 内部使用简短别名(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
# - provider=doubao → 火山方舟
# - 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" | "jimeng"
- provider: "doubao" | "dashscope"
- model_id: 对应 API 的真实模型 ID
- cfg: VIRAL_VIDEO_MODEL_CONFIG 条目
未识别或空值回落到默认 seedance-2.5。已经是 doubao-/ep- 开头的完整 ID 视为 doubao provider;
"jimeng" 开头视为 jimeng provider(内部兜底,不暴露给前端)。
未识别或空值回落到默认 seedance-2.5。已经是 doubao-/ep- 开头的完整 ID 视为 doubao provider。
"""
from packages.domain.points_rules import get_viral_video_model_config
@@ -62,10 +51,6 @@ 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")
@@ -82,93 +67,6 @@ def _resolve_video_model_id(model: str | None) -> str:
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 客户端.
@@ -186,9 +84,6 @@ class DoubaoClient:
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
# 最近一次视频生成的详细错误(error_code + user_message + raw detail),供上层读取后展示给用户
self.last_video_error: dict = {}
def embed_text(self, text: str, timeout: int | None = None) -> list[float] | None:
"""调用豆包文本 Embedding API,返回浮点向量;失败返回 None。"""
@@ -201,7 +96,7 @@ class DoubaoClient:
"Content-Type": "application/json",
}
payload: dict[str, Any] = {
"model": self.embedding_model,
"model": getattr(self, "embedding_model", None) or "doubao-embedding-large-text-240915",
"input": text.strip(),
"encoding_format": "float",
}
@@ -412,8 +307,7 @@ class DoubaoClient:
"""调用 Seedance 2.5 生视频(异步任务→轮询→下载)。
成功返回 {"video_path": str, "usage": dict | None},失败返回 None。
失败时把详细错误信息(HTTP状态码、响应 body、分类后的用户提示)写入 self.last_video_error,
上层可通过 get_last_video_error() 读取并展示给用户,不再笼统显示"返回为空"。
usage 是 Seedance 返回的计费信息(含 completion_tokens)。
【v1.6.1 修复】严格按官方 content 数组协议构造请求:
- 所有参考(图/音/视)必须放进 content 数组并带 role 字段,不能放顶层 reference_audios/reference_videos(非官方字段,会被忽略或导致异常)。
@@ -421,24 +315,9 @@ class DoubaoClient:
判定:传了参考音频/视频或 ≥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()
@@ -452,20 +331,10 @@ class DoubaoClient:
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,
}
logger.error("DashScope client 不可用(未配置 DASHSCOPE_API_KEY),video_model=%s", model)
return None
try:
# DashScope 客户端也设置 last_video_error 语义(如果它支持)
if hasattr(ds, "last_video_error"):
ds.last_video_error = {}
result = ds.video_generation(
return ds.video_generation(
prompt=prompt,
image_url=image_url,
duration=duration,
@@ -474,31 +343,10 @@ class DoubaoClient:
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
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)]
@@ -522,7 +370,7 @@ class DoubaoClient:
}
)
else:
# 纯首帧:显式 role=first_frame
# 纯首帧:不带 role,服务端识别为 first_frame(或显式 role=first_frame)
content.append(
{
"type": "image_url",
@@ -563,38 +411,22 @@ class DoubaoClient:
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
body = (getattr(resp, "text", "") or "")[:1500]
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 不重试(参数/鉴权/配额错误重试无意义)
if attempt < self.max_retries:
time.sleep(0.5 * (2**attempt))
continue
return None, last_err, sc, body
@@ -603,19 +435,9 @@ class DoubaoClient:
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):
if attempt < self.max_retries:
wait = 0.5 * (2**attempt)
logger.warning(
"Seedance 创建任务失败,%.1fs 后重试 (%d/%d): %s",
@@ -625,7 +447,7 @@ class DoubaoClient:
e,
)
time.sleep(wait)
return None, last_err, last_sc, last_body
return None, last_err, 0, ""
# 第一次尝试
task_id, last_err, sc, body = _do_create(create_payload)
@@ -644,53 +466,18 @@ class DoubaoClient:
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",
"Seedance 创建任务最终失败: model=%s base_url=%s err=%s body=%s 【排查】"
"1) 方舟控制台已开通 %s;2) API Key 有该模型权限;"
"3) DOUBAO_BASE_URL=https://ark.cn-beijing.volces.com/api/v3;4) 参考素材 URL 公网可访问。",
video_model,
self.base_url,
sc,
err_code,
last_err,
(body or "")[:500],
video_model,
)
# #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"])
@@ -702,21 +489,15 @@ class DoubaoClient:
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
try:
if int(getattr(resp, "status_code", 200)) >= 400:
resp.raise_for_status()
except (TypeError, ValueError):
pass
data = resp.json()
status = data.get("status", "")
last_status = status
@@ -727,22 +508,13 @@ class DoubaoClient:
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]
last_err = RuntimeError(f"task succeeded but no video_url: {str(data)[:300]}")
logger.error("Seedance succeeded 但无 video_url: %s", last_err)
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]
last_err = RuntimeError(f"task failed: code={err.get('code','')} msg={err.get('message','')}")
logger.error("Seedance 任务失败 task_id=%s: %s", task_id, last_err)
break
if status in ("expired", "cancelled"):
last_err = RuntimeError(f"task {status}")
@@ -753,39 +525,18 @@ class DoubaoClient:
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])
logger.warning("Seedance 轮询 HTTP %d: %s", e.response.status_code, (e.response.text or "")[: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)",
"Seedance 任务未成功: task_id=%s last_status=%s polls=%d err=%s (总等待 %.0fs)",
task_id,
last_status,
poll_count,
err_code,
last_err,
total_timeout,
)
@@ -816,135 +567,12 @@ class DoubaoClient:
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 _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 {})
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
# ── 单例 ─────────────────────────────────────────────────────────────────────
+9 -29
View File
@@ -618,23 +618,20 @@ def call_video_generation(
reference_audios: list[str] | None = None,
reference_videos: list[str] | None = None,
) -> dict | None:
"""调用 Seedance / Wan 视频生成(v1.6.2 多模型版)。
"""调用 Seedance 2.5 生成视频(v1.6.1 单次出片版)。
成功返回 {"video_path": str, "usage": dict | None}(usage 含 completion_tokens),失败返回 None。
失败时错误详情会写入 client.last_video_error,可通过 get_last_video_error() 读取:
{"error_code": str, "user_message": str, "status_code": int, "detail": str, ...}
v1.6.1 关键约束(避免 20min 卡死):
- 参考音频/视频/多图全部放进 content 数组并带 role=reference_audio/reference_video/reference_image;
- 纯首帧无参考时(first_frame 模式),Seedance 2.5 强制 ratio=adaptive;
传了参考音/视/多图时走 omni_reference 模式,ratio 可指定为 9:16(客户端内部自动判断)。
- ratio 默认 9:16(竖屏),客户端会根据是否有参考自动在 first_frame/adaptive 与 omni/9:16 间切换;
若创建任务因 ratio 报错(HTTP 400),客户端会自动回退到 adaptive 再试一次。
"""
client = get_doubao_client()
if not client.is_available:
msg = "豆包客户端未配置(DOUBAO_API_KEY 缺失),跳过视频生成"
logger.warning("[ai_service] %s", msg)
# 写入 last_video_error 供上层读取
client.last_video_error = {
"error_code": "auth_error",
"user_message": "视频生成服务未配置,请联系管理员。",
"status_code": 0,
"detail": msg,
}
logger.warning("[ai_service] 豆包客户端未配置,跳过视频生成")
return None
effective_ratio = ratio or "9:16"
try:
@@ -656,21 +653,4 @@ def call_video_generation(
return client.video_generation(**kwargs)
except Exception as e:
logger.error("[ai_service] call_video_generation 异常: %s", e, exc_info=True)
client.last_video_error = {
"error_code": "unknown",
"user_message": f"视频生成异常:{e!s}"[:200],
"status_code": 0,
"detail": str(e),
}
return None
def get_last_video_error() -> dict:
"""读取最近一次视频生成失败的详细错误(含 error_code/user_message/status_code/detail)。
成功或未调用过返回空 dict。
"""
try:
client = get_doubao_client()
return client.get_last_video_error() if hasattr(client, "get_last_video_error") else {}
except Exception:
return {}
+19 -142
View File
@@ -21,59 +21,11 @@ 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 等视频生成)。"""
@@ -86,24 +38,11 @@ class DashScopeClient:
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,
@@ -118,15 +57,12 @@ class DashScopeClient:
) -> dict | None:
"""调用 DashScope 异步视频合成接口,轮询完成后下载到本地。
返回 {"video_path": str, "usage": dict | None};失败返回 None,错误详情写入 self.last_video_error。
返回 {"video_path": str, "usage": dict | None};失败返回 None。
"""
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)
@@ -170,27 +106,18 @@ class DashScopeClient:
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 = ""
last_err: Exception | None = None
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
body_text = (getattr(resp, "text", "") or "")[:1500]
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
resp.raise_for_status()
data = resp.json()
tid = (data.get("output") or {}).get("task_id")
if tid:
@@ -199,34 +126,17 @@ class DashScopeClient:
# 部分情况下 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
last_err = RuntimeError(f"dashscope create failed: {body_text[:300]}")
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
)
last_err = RuntimeError(f"create ok but no task_id: {str(data)[:300]}")
except Exception as e:
last_err = e
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))
logger.error("[dashscope] 创建任务最终失败: %s", last_err)
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
# 轮询任务
@@ -234,26 +144,16 @@ class DashScopeClient:
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
if int(getattr(r, "status_code", 0) or 0) >= 400:
logger.warning("[dashscope] 轮询 HTTP %d", r.status_code)
time.sleep(self.poll_interval)
continue
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")
@@ -265,41 +165,22 @@ class DashScopeClient:
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,
)
logger.error("[dashscope] 任务 %s SUCCEEDED 但无 video_url", 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"
msg = out.get("message") or d.get("message") 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)
logger.debug("[dashscope] 任务 %s 状态 %s,继续轮询", task_id, task_status)
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
# 下载视频
@@ -312,23 +193,19 @@ class DashScopeClient:
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)
if int(getattr(resp, "status_code", 0) or 0) >= 400:
logger.error("[dashscope] 下载 HTTP %d", resp.status_code)
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))
logger.error("[dashscope] 下载视频失败: %s", 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}
-526
View File
@@ -1,526 +0,0 @@
"""即梦(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
-104
View File
@@ -556,107 +556,3 @@ class TestResolveVideoModelId:
# 未知 model key 会通过 get_viral_video_model_config 回落到 seedance-2.5
with caplog.at_level(logging.WARNING, logger="shared.ai_client"):
assert fn("some-random-model") == "doubao-seedance-2-5-260628"
# ── #2165 详细错误信息和 last_video_error ─────────────────────────
class TestVideoGenerationLastError:
def test_create_400_portrait_returns_user_message(self, tmp_path):
"""#2169: HTTP 400 + 真人拦截关键词 → 自动尝试即梦兜底;即梦未配时返回 portrait_intercept。"""
client = _make_client(max_retries=0)
create_resp = MagicMock()
create_resp.status_code = 400
create_resp.text = '{"error":{"code":"ContentRisk","message":"Real person face detected in reference image, portrait blocked"}}'
create_resp.json.return_value = {"error": {"code": "ContentRisk", "message": "..."}}
create_resp.raise_for_status.side_effect = httpx.HTTPStatusError(
"bad", request=MagicMock(), response=create_resp
)
with (
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"
)
result = client.video_generation("p", output_dir=str(tmp_path), image_url="https://img/x.jpg")
assert result is None
err = client.get_last_video_error()
assert err["error_code"] == "portrait_intercept"
# 即梦兜底未启用时提示包含"真人照片"/"即梦"等关键字
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)
create_resp = MagicMock()
create_resp.status_code = 401
create_resp.text = '{"error":{"message":"Unauthorized"}}'
create_resp.json.return_value = {"error": {"message": "Unauthorized"}}
create_resp.raise_for_status.side_effect = httpx.HTTPStatusError(
"auth", request=MagicMock(), response=create_resp
)
with (
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,
):
mock_s.return_value = MagicMock(
doubao_video_poll_interval=0, doubao_video_timeout=1, doubao_video_model="seedance"
)
result = client.video_generation("p", output_dir=str(tmp_path))
assert result is None
err = client.get_last_video_error()
assert err["error_code"] == "auth_error"
assert err["status_code"] == 401
def test_poll_failed_returns_task_failed_error(self, tmp_path):
"""轮询 status=failed 时应记录 task_failed 错误并含 detail。"""
client = _make_client(max_retries=0)
create_resp = MagicMock()
create_resp.status_code = 200
create_resp.json.return_value = {"id": "t-fail"}
create_resp.raise_for_status = MagicMock()
poll_resp = MagicMock()
poll_resp.status_code = 200
poll_resp.json.return_value = {
"status": "failed",
"error": {"code": "InvalidParam", "message": "resolution invalid"},
}
poll_resp.raise_for_status = MagicMock()
with (
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
patch("packages.shared.ai_client.httpx.get", return_value=poll_resp),
patch("packages.shared.ai_client.time.sleep", return_value=None),
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory()),
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
):
mock_s.return_value = MagicMock(
doubao_video_poll_interval=0, doubao_video_timeout=10, doubao_video_model="seedance"
)
result = client.video_generation("p", output_dir=str(tmp_path))
assert result is None
err = client.get_last_video_error()
assert err["error_code"] == "task_failed"
assert "InvalidParam" in err.get("detail", "") or err["status_code"] == 200
class TestAiServiceLastVideoError:
def test_call_video_generation_returns_none_sets_error(self):
"""失败后 get_last_video_error 应返回结构化错误信息。"""
from packages.shared import ai_service
mock_client = MagicMock()
mock_client.is_available = True
mock_client.last_video_error = {"error_code": "unknown", "user_message": "test"}
mock_client.get_last_video_error.return_value = {"error_code": "unknown", "user_message": "test"}
mock_client.video_generation.return_value = None
with patch("packages.shared.ai_service.get_doubao_client", return_value=mock_client):
assert ai_service.call_video_generation("p") is None
err = ai_service.get_last_video_error()
assert err["error_code"] == "unknown"
assert "user_message" in err
-198
View File
@@ -1,198 +0,0 @@
"""catalog 应用服务单测:会员套餐 / 积分包从共享库读取与字段映射。"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture(autouse=True)
def _clear_cache():
from packages.application.catalog import admin_catalog
admin_catalog._cache.clear()
yield
admin_catalog._cache.clear()
def _row(**kw):
row = MagicMock()
for k, v in kw.items():
setattr(row, k, v)
return row
class TestMembershipPlans:
def test_yearly_plan_mapping(self):
from packages.application.catalog import admin_catalog
row = _row(
plan_key="premium_yearly",
name="高级会员年卡",
description="年度订阅",
monthly_price=0,
yearly_price=399,
quotas={"4k": True, "batch_render": True, "credits_per_month": 500},
display_order=1,
)
session = MagicMock()
session.execute.return_value.fetchall.return_value = [row]
sl = MagicMock(return_value=session)
with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
plans = admin_catalog.get_membership_plans()
assert len(plans) == 1
p = plans[0]
assert p["plan_id"] == "premium_yearly"
assert p["billing_cycle"] == "yearly"
assert p["price_cents"] == 39900
assert p["monthly_price_cents"] == 3325
assert p["duration_days"] == 365
assert p["features"]["4K 超清分辨率"] is True
assert p["features"]["credits_per_month"] == 500
session.close.assert_called_once()
def test_monthly_plan_mapping(self):
from packages.application.catalog import admin_catalog
row = _row(
plan_key="premium_monthly",
name="高级会员月卡",
description=None,
monthly_price=39,
yearly_price=0,
quotas=None,
display_order=2,
)
session = MagicMock()
session.execute.return_value.fetchall.return_value = [row]
sl = MagicMock(return_value=session)
with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
plans = admin_catalog.get_membership_plans()
assert len(plans) == 1
p = plans[0]
assert p["billing_cycle"] == "monthly"
assert p["price_cents"] == 3900
assert p["monthly_price_cents"] == 3900
assert p["duration_days"] == 30
assert p["features"] == {}
def test_both_cycles_expanded(self):
from packages.application.catalog import admin_catalog
row = _row(
plan_key="premium",
name="高级会员",
description=None,
monthly_price=39,
yearly_price=399,
quotas={},
display_order=1,
)
session = MagicMock()
session.execute.return_value.fetchall.return_value = [row]
sl = MagicMock(return_value=session)
with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
plans = admin_catalog.get_membership_plans()
cycles = {p["billing_cycle"] for p in plans}
assert cycles == {"yearly", "monthly"}
def test_no_session_returns_empty(self):
from packages.application.catalog import admin_catalog
with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", None, create=True):
assert admin_catalog.get_membership_plans() == []
class TestPointsPackages:
def test_package_mapping_with_bonus(self):
from packages.application.catalog import admin_catalog
row = _row(
package_key="pkg_100",
name="100元充值包",
price=100,
credits=1000,
bonus_credits=100,
is_recommended=True,
description="推荐",
sort_order=4,
)
session = MagicMock()
session.execute.return_value.fetchall.return_value = [row]
sl = MagicMock(return_value=session)
with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
packages = admin_catalog.get_points_packages()
assert len(packages) == 1
pkg = packages[0]
assert pkg["code"] == "pkg_100"
assert pkg["points"] == 1100
assert pkg["price_cents"] == 10000
assert pkg["is_recommended"] is True
assert pkg["unit_price"] == "¥0.091/积分"
def test_zero_credits_unit_price_safe(self):
from packages.application.catalog import admin_catalog
row = _row(
package_key="pkg_0",
name="空包",
price=0,
credits=0,
bonus_credits=0,
is_recommended=False,
description=None,
sort_order=0,
)
session = MagicMock()
session.execute.return_value.fetchall.return_value = [row]
sl = MagicMock(return_value=session)
with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
packages = admin_catalog.get_points_packages()
assert packages[0]["points"] == 0
assert packages[0]["price_cents"] == 0
assert packages[0]["unit_price"] == "¥0.000/积分"
def test_no_session_returns_empty(self):
from packages.application.catalog import admin_catalog
with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", None, create=True):
assert admin_catalog.get_points_packages() == []
class TestPackagesRoute:
def test_get_packages_route_returns_items(self):
from app.api.routes.points import get_packages
cu = MagicMock()
cu.user.member_type = None
rows = [
{
"code": "pkg_10",
"name": "10元充值包",
"points": 100,
"price_cents": 1000,
"unit_price": "¥0.100/积分",
}
]
with patch(
"packages.application.catalog.admin_catalog.get_points_packages",
return_value=rows,
):
resp = get_packages(current_user=cu)
assert len(resp.packages) == 1
item = resp.packages[0]
assert item.code == "pkg_10"
assert item.points == 100
assert item.price_cents == 1000
-19
View File
@@ -4,7 +4,6 @@ from __future__ import annotations
from unittest.mock import MagicMock, mock_open, patch
import httpx
import pytest
_SINGLETON = "_DASHSCOPE_CLIENT_SINGLETON"
@@ -158,21 +157,3 @@ class TestDashScopeVideoGeneration:
c.video_generation(prompt=" ", duration=5, ratio="9:16", resolution="720p", output_dir="/tmp/videos")
is None
)
def test_create_400_sets_last_video_error(self, tmp_path):
"""创建任务 HTTP 400 时应写 last_video_error。"""
from packages.shared import dashscope_client as dc
dc._DASHSCOPE_CLIENT_SINGLETON = None
with patch.dict("os.environ", {"DASHSCOPE_API_KEY": "test-key"}):
c = dc.DashScopeClient()
r = MagicMock()
r.status_code = 401
r.text = '{"code":"InvalidApiKey","message":"bad key"}'
r.raise_for_status.side_effect = httpx.HTTPStatusError("auth", request=MagicMock(), response=r)
with patch.object(dc.httpx, "post", return_value=r), patch.object(dc, "time"):
out = c.video_generation("p", output_dir=str(tmp_path))
assert out is None
err = c.get_last_video_error()
assert err["error_code"] == "auth_error"
assert c.last_video_error is not None
-376
View File
@@ -1,376 +0,0 @@
"""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"]
+25 -38
View File
@@ -173,46 +173,33 @@ class TestSubscriptionPlans:
_spec.loader.exec_module(_mod)
return _mod.list_membership_plans
def test_plans_endpoint_reads_admin_table(self):
"""/subscription/plans 改读管理后台 plans 表:返回 catalog 服务提供的真实档位。"""
list_membership_plans = self._import_plans_fn()
real_plan = {
"plan_id": "premium_yearly",
"billing_cycle": "yearly",
"name": "高级会员年卡",
"description": "高级会员年度订阅,享受全部功能",
"price_cents": 39900,
"monthly_price_cents": 3325,
"duration_days": 365,
"features": {
"4K 超清分辨率": True,
"批量渲染": True,
"优先处理队列": True,
"credits_per_month": 500,
},
}
with patch(
"packages.application.catalog.admin_catalog.get_membership_plans",
return_value=[real_plan],
):
resp = list_membership_plans(current_user=_make_cu())
plans = resp["plans"]
assert len(plans) == 1
p0 = plans[0]
assert p0["plan_id"] == "premium_yearly"
assert p0["price_cents"] == 39900
assert p0["duration_days"] == 365
assert p0["features"]["4K 超清分辨率"] is True
def test_plans_endpoint_returns_three_tiers(self):
import os # noqa: F401 (used by _import_plans_fn)
def test_plans_endpoint_empty_when_all_disabled(self):
"""后台停用全部套餐时,用户端返回空列表。"""
list_membership_plans = self._import_plans_fn()
with patch(
"packages.application.catalog.admin_catalog.get_membership_plans",
return_value=[],
):
resp = list_membership_plans(current_user=_make_cu())
assert resp["plans"] == []
resp = list_membership_plans(current_user=_make_cu())
plans = resp["plans"]
plan_ids = {p["plan_id"] for p in plans}
assert plan_ids == {"monthly", "quarterly", "yearly"}
for p in plans:
assert p["price_cents"] > 0
assert p["duration_days"] in (30, 90, 365)
assert 0 < p["points_discount"] <= 1.0
assert "max_resolution" in p["features"]
def test_longer_plans_cheaper_per_month(self):
import os # noqa: F401
list_membership_plans = self._import_plans_fn()
resp = list_membership_plans(current_user=_make_cu())
plans = resp["plans"]
monthly = next(p for p in plans if p["plan_id"] == "monthly")
quarterly = next(p for p in plans if p["plan_id"] == "quarterly")
yearly = next(p for p in plans if p["plan_id"] == "yearly")
assert monthly["monthly_price_cents"] == 1990
assert quarterly["monthly_price_cents"] < monthly["monthly_price_cents"]
assert yearly["monthly_price_cents"] < quarterly["monthly_price_cents"]
# ── P1-7: multiplier consistency ──────────────────────────────────────