川
2ef126e445
- 新增品类专属背面/侧面描述(BACK_VIEW_HINTS/SIDE_VIEW_HINTS) - 强化一致性前缀策略，按视角定制相机位置描述 - 更新视角映射提示词为纯摄影术语 - 修复前端下载逻辑：改用fetch直接下载当前视角图片 - HTTPS改HTTP修复外网URL访问 - 新增多视角一致性与3D视频生成技术文档
281 lines
9.5 KiB
Python
281 lines
9.5 KiB
Python
"""
|
||
AI 视频生成服务 - 可灵（Kling）多图参考生视频
|
||
使用可灵 AI 的多图参考生视频 API，原生支持传入 1-4 张参考图片
|
||
AI 会理解为同一物体的多角度参考，生成单品旋转展示视频
|
||
|
||
API 文档: https://app.klingai.com/cn/dev/document-api/apiReference/model/multiImageToVideo
|
||
认证方式: JWT (Access Key + Secret Key)
|
||
API 端点: https://api.klingai.com
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Optional, List
|
||
|
||
import httpx
|
||
import jwt
|
||
|
||
from .config_service import get_config_value
|
||
|
||
# 视频本地存储目录
|
||
VIDEO_UPLOAD_DIR = Path(__file__).resolve().parent.parent.parent / "uploads" / "videos"
|
||
VIDEO_UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 可灵 API 配置（中国区域名）
|
||
KLING_API_BASE = "https://api-beijing.klingai.com"
|
||
|
||
# 超时与轮询配置
|
||
SUBMIT_TIMEOUT = 30
|
||
POLL_TIMEOUT = 15
|
||
MAX_POLL_ATTEMPTS = 120 # 约 10 分钟
|
||
POLL_INTERVAL = 5
|
||
|
||
|
||
# ============================================================
|
||
# JWT 认证
|
||
# ============================================================
|
||
|
||
def _generate_jwt_token(access_key: str, secret_key: str) -> str:
|
||
"""
|
||
使用 Access Key 和 Secret Key 生成 JWT Token
|
||
可灵 API 使用 JWT 认证，token 有效期 30 分钟
|
||
"""
|
||
now = int(time.time())
|
||
headers = {
|
||
"alg": "HS256",
|
||
"typ": "JWT"
|
||
}
|
||
payload = {
|
||
"iss": access_key,
|
||
"exp": now + 1800, # 30 分钟过期
|
||
"nbf": now - 5, # 允许 5 秒时钟偏差
|
||
"iat": now, # 签发时间
|
||
}
|
||
token = jwt.encode(payload, secret_key, algorithm="HS256", headers=headers)
|
||
return token
|
||
|
||
|
||
def _get_kling_keys() -> tuple:
|
||
"""获取可灵 Access Key 和 Secret Key"""
|
||
access_key = get_config_value("KLING_ACCESS_KEY", "")
|
||
secret_key = get_config_value("KLING_SECRET_KEY", "")
|
||
if not access_key or not secret_key:
|
||
raise RuntimeError(
|
||
"未配置 KLING_ACCESS_KEY 或 KLING_SECRET_KEY，无法使用可灵视频生成。"
|
||
"请在管理后台 系统配置 中添加可灵 AI 的 Access Key 和 Secret Key。"
|
||
)
|
||
return access_key, secret_key
|
||
|
||
|
||
def _build_headers(access_key: str, secret_key: str) -> dict:
|
||
"""构建带 JWT 认证的请求头"""
|
||
token = _generate_jwt_token(access_key, secret_key)
|
||
return {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {token}",
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# 视频生成核心逻辑
|
||
# ============================================================
|
||
|
||
async def generate_video(
|
||
image_urls: List[str],
|
||
prompt: str = "",
|
||
duration_seconds: int = 5,
|
||
) -> str:
|
||
"""
|
||
调用可灵多图参考生视频 API，生成 360 度旋转展示视频
|
||
|
||
核心优势：原生支持传入多张参考图（1-4张），
|
||
AI 理解为同一物体的多角度参考，生成单品视频。
|
||
|
||
Args:
|
||
image_urls: 多视角图片 URL 列表（最多4张）
|
||
prompt: 视频生成提示词
|
||
duration_seconds: 视频时长（5 或 10 秒）
|
||
|
||
Returns:
|
||
生成的视频本地 URL
|
||
"""
|
||
access_key, secret_key = _get_kling_keys()
|
||
|
||
logger.info(f"可灵视频生成，传入图片数量: {len(image_urls)}")
|
||
|
||
# 将本地路径转换为外网可访问URL（可灵API需要完整URL）
|
||
from .ai_3d_generator import _to_public_url
|
||
image_urls = [_to_public_url(u) for u in image_urls]
|
||
|
||
# 可灵最多支持 4 张参考图
|
||
if len(image_urls) > 4:
|
||
image_urls = image_urls[:4]
|
||
logger.info("图片数量超过4张，截取前4张")
|
||
|
||
# 构建提示词
|
||
if not prompt:
|
||
prompt = get_config_value("VIDEO_PROMPT", "")
|
||
if not prompt:
|
||
prompt = (
|
||
"精美玉雕工艺品在专业珠宝摄影棚内展示，"
|
||
"纯白色背景，柔和的珠宝摄影灯光，"
|
||
"玉石作品放在旋转展台上缓慢平稳地旋转360度，"
|
||
"展示正面、侧面、背面全貌，"
|
||
"展现玉石温润的质感、细腻的雕刻纹理和通透的光泽，"
|
||
"电影级画质，微距细节感，平稳流畅的转台旋转"
|
||
)
|
||
|
||
# 视频时长，仅支持 5 或 10
|
||
duration = str(duration_seconds) if duration_seconds in (5, 10) else "5"
|
||
|
||
# Step 1: 提交任务
|
||
task_id = await _submit_video_task(
|
||
access_key, secret_key, image_urls, prompt, duration
|
||
)
|
||
logger.info(f"可灵视频生成任务已提交: task_id={task_id}")
|
||
|
||
# Step 2: 轮询等待结果
|
||
remote_video_url = await _poll_video_result(access_key, secret_key, task_id)
|
||
logger.info(f"可灵视频生成完成: {remote_video_url[:80]}...")
|
||
|
||
# Step 3: 下载视频到本地存储
|
||
local_path = await _download_video_to_local(remote_video_url)
|
||
logger.info(f"视频已保存到本地: {local_path}")
|
||
|
||
return local_path
|
||
|
||
|
||
async def _submit_video_task(
|
||
access_key: str,
|
||
secret_key: str,
|
||
image_urls: List[str],
|
||
prompt: str,
|
||
duration: str = "5",
|
||
) -> str:
|
||
"""提交多图参考生视频任务到可灵 API"""
|
||
url = f"{KLING_API_BASE}/v1/videos/multi-image2video"
|
||
|
||
# 构建 image_list（每个元素是 {"image": "url"}）
|
||
image_list = [{"image": img_url} for img_url in image_urls]
|
||
|
||
payload = {
|
||
"model_name": "kling-v1-6",
|
||
"image_list": image_list,
|
||
"prompt": prompt,
|
||
"mode": "pro", # 高品质模式
|
||
"duration": duration, # 视频时长
|
||
"aspect_ratio": "1:1", # 1:1 正方形
|
||
}
|
||
|
||
headers = _build_headers(access_key, secret_key)
|
||
body = json.dumps(payload, ensure_ascii=False)
|
||
|
||
logger.info(f"提交可灵视频任务: {len(image_urls)}张参考图, 时长={duration}s, 模式=pro")
|
||
|
||
async with httpx.AsyncClient(timeout=SUBMIT_TIMEOUT) as client:
|
||
resp = await client.post(url, content=body, headers=headers)
|
||
if resp.status_code not in (200, 201):
|
||
error_body = resp.text[:1000]
|
||
logger.error(f"可灵视频任务提交失败: status={resp.status_code}, body={error_body}")
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
|
||
# 检查响应
|
||
code = data.get("code", -1)
|
||
if code != 0:
|
||
msg = data.get("message", "未知错误")
|
||
raise RuntimeError(f"可灵视频任务提交失败 (code={code}): {msg}")
|
||
|
||
task_id = data.get("data", {}).get("task_id")
|
||
if not task_id:
|
||
raise RuntimeError(f"可灵响应中未找到 task_id: {data}")
|
||
|
||
return task_id
|
||
|
||
|
||
async def _poll_video_result(
|
||
access_key: str,
|
||
secret_key: str,
|
||
task_id: str,
|
||
) -> str:
|
||
"""轮询可灵视频生成结果"""
|
||
url = f"{KLING_API_BASE}/v1/videos/multi-image2video/{task_id}"
|
||
|
||
for attempt in range(1, MAX_POLL_ATTEMPTS + 1):
|
||
await asyncio.sleep(POLL_INTERVAL)
|
||
|
||
# 每次轮询重新生成 JWT（避免过期）
|
||
headers = _build_headers(access_key, secret_key)
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=POLL_TIMEOUT) as client:
|
||
resp = await client.get(url, headers=headers)
|
||
if resp.status_code != 200:
|
||
logger.warning(
|
||
f"轮询可灵视频结果失败 (attempt={attempt}): "
|
||
f"status={resp.status_code}, body={resp.text[:300]}"
|
||
)
|
||
continue
|
||
data = resp.json()
|
||
except Exception as e:
|
||
logger.warning(f"轮询可灵视频异常 (attempt={attempt}): {e}")
|
||
continue
|
||
|
||
code = data.get("code", -1)
|
||
if code != 0:
|
||
msg = data.get("message", "未知错误")
|
||
logger.warning(f"轮询可灵视频返回错误 (attempt={attempt}): code={code}, msg={msg}")
|
||
continue
|
||
|
||
task_data = data.get("data", {})
|
||
task_status = task_data.get("task_status", "")
|
||
|
||
if task_status == "succeed":
|
||
# 从 task_result.videos 中提取视频 URL
|
||
task_result = task_data.get("task_result", {})
|
||
videos = task_result.get("videos", [])
|
||
if videos and videos[0].get("url"):
|
||
return videos[0]["url"]
|
||
raise RuntimeError(f"可灵视频生成完成但未找到视频URL: {data}")
|
||
|
||
elif task_status == "failed":
|
||
fail_msg = task_data.get("task_status_msg", "未知原因")
|
||
raise RuntimeError(f"可灵视频生成失败: {fail_msg}")
|
||
|
||
else:
|
||
# submitted / processing
|
||
if attempt % 6 == 0:
|
||
logger.info(
|
||
f"可灵视频生成中... (attempt={attempt}, status={task_status})"
|
||
)
|
||
|
||
raise RuntimeError(f"可灵视频生成超时: 轮询 {MAX_POLL_ATTEMPTS} 次后仍未完成")
|
||
|
||
|
||
async def _download_video_to_local(remote_url: str) -> str:
|
||
"""
|
||
下载远程视频到本地 uploads/videos/ 目录
|
||
|
||
Returns:
|
||
本地视频的 URL 路径，如 /uploads/videos/xxx.mp4
|
||
"""
|
||
filename = f"{uuid.uuid4().hex}.mp4"
|
||
local_file = VIDEO_UPLOAD_DIR / filename
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=120, follow_redirects=True) as client:
|
||
resp = await client.get(remote_url)
|
||
resp.raise_for_status()
|
||
local_file.write_bytes(resp.content)
|
||
logger.info(f"视频下载完成: {len(resp.content)} 字节 -> {local_file}")
|
||
except Exception as e:
|
||
logger.error(f"视频下载失败: {e}")
|
||
raise RuntimeError(f"视频下载失败: {e}")
|
||
|
||
return f"/uploads/videos/{filename}"
|