707 lines
31 KiB
Python
707 lines
31 KiB
Python
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import re
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import settings
|
||
from app.enums.video_prompt_schema import PromptSchemaVersionEnum, VideoPromptSchemaUsageEnum
|
||
from app.models.model_config import ModelConfig
|
||
from app.models.token_usage import TokenUsage
|
||
from app.utils.id_gen import generate_id
|
||
from app.services.resource_signed_url_service import build_resource_signed_url
|
||
|
||
DEFAULT_FRAME_RATE = "30fps"
|
||
DEFAULT_REFERENCE_VIDEO_FPS = 1
|
||
|
||
CLIENT_SCHEMA_V1: dict[str, Any] = {
|
||
"schema_version": PromptSchemaVersionEnum.CLIENT_V1.value,
|
||
"schema_usage": VideoPromptSchemaUsageEnum.CLIENT_DISPLAY.value,
|
||
"基础分类": {
|
||
"生成类型": "文生视频/图生视频/视频生视频/数字人视频/无",
|
||
"视频大类": "产品广告视频/电商带货视频/口播讲解视频/剧情视频/教程视频/风景旅行视频/美食视频/宠物视频/动漫卡通视频/游戏视频/企业宣传视频/新闻资讯视频/直播切片视频/图文快闪视频/音乐舞蹈视频/运动健身视频/无",
|
||
"视频子类": "产品推广短视频/口播讲解/电商带货/剧情演绎/操作教程/旅行风景/美食展示/宠物互动/二次元动画/游戏宣传/企业介绍/资讯播报/直播高光/图文快闪/无",
|
||
"视频用途": "广告投放/社媒发布/产品展示/课程教学/品牌宣传/娱乐内容/信息科普/无",
|
||
"目标平台": "抖音/快手/小红书/微信视频号/B站/TikTok/YouTube Shorts/Instagram Reels/无",
|
||
},
|
||
"素材理解": {
|
||
"是否有参考图片": "是/否",
|
||
"是否有参考视频": "是/否",
|
||
"参考视频用途": "动作参考/镜头参考/风格参考/运镜参考/节奏参考/无",
|
||
"需要保留": [],
|
||
"允许改动": [],
|
||
"禁止改动": [],
|
||
},
|
||
"业务属性": {
|
||
"产品类型": "APP/实物商品/食品/服饰/美妆/电子产品/汽车/房产/课程/服务/无",
|
||
"产品名称": "无",
|
||
"品牌名称": "无",
|
||
"核心卖点": [],
|
||
"目标受众": "无",
|
||
"核心表达目标": "无",
|
||
"内容风格": "无",
|
||
"行动引导": "立即体验/立即下载/立即购买/点击了解/预约咨询/无",
|
||
},
|
||
"画面属性": {
|
||
"视频时长": "无",
|
||
"视频比例": "无",
|
||
"清晰度": "无",
|
||
"帧率": "无",
|
||
"主体描述": "无",
|
||
"主体数量": "无",
|
||
"主体位置": "无",
|
||
"主体占比": "无",
|
||
"场景描述": "无",
|
||
"构图方式": "无",
|
||
"画面风格": "无",
|
||
"光影色彩": "无",
|
||
},
|
||
"动作流程": [],
|
||
"镜头流程": [],
|
||
"字幕与口播": {
|
||
"是否需要字幕": "是/否",
|
||
"字幕内容": [],
|
||
"字幕位置": "无",
|
||
"字幕样式": "无",
|
||
"是否口播": "是/否",
|
||
"口播内容": "无",
|
||
"口播语气": "无",
|
||
"口播语速": "无",
|
||
"是否需要口型同步": "是/否/无",
|
||
},
|
||
"音频与节奏": {
|
||
"背景音乐": "无",
|
||
"音乐风格": "无",
|
||
"音乐节奏": "无",
|
||
"环境音": "无",
|
||
"动作音效": "无",
|
||
"整体节奏": "慢节奏/中等节奏/快节奏/卡点节奏/无",
|
||
},
|
||
"合规控制": {
|
||
"是否广告": "是/否",
|
||
"风险等级": "低/中/高",
|
||
"安全表达": "无",
|
||
"禁用词": [],
|
||
"合规说明": "无",
|
||
},
|
||
"质量控制": {
|
||
"主体一致性": "低/中/高/无",
|
||
"产品一致性": "低/中/高/无",
|
||
"动作自然度": "低/中/高/无",
|
||
"镜头稳定性": "低/中/高/无",
|
||
"字幕准确性": "低/中/高/无",
|
||
},
|
||
"最终提示词": {
|
||
"主提示词": "无",
|
||
"动作提示词": "无",
|
||
"镜头提示词": "无",
|
||
"字幕提示词": "无",
|
||
"音频提示词": "无",
|
||
"风格提示词": "无",
|
||
"负面提示词": "无",
|
||
},
|
||
}
|
||
|
||
|
||
def _safe_list(value: Any) -> list[Any]:
|
||
return value if isinstance(value, list) else []
|
||
|
||
|
||
def _format_options(options: list[Any], fallback: str = "无") -> str:
|
||
values = [str(item) for item in _safe_list(options) if str(item).strip()]
|
||
return "/".join(values) if values else fallback
|
||
|
||
|
||
def get_recommended_resolution(video_ratio: str, resolution: str) -> str:
|
||
"""按比例和清晰度粗略计算推荐像素，不在服务内维护固定比例/分辨率白名单。"""
|
||
try:
|
||
width_ratio, height_ratio = [float(x) for x in str(video_ratio).split(":", 1)]
|
||
short_edge = int(str(resolution).lower().replace("p", ""))
|
||
if width_ratio >= height_ratio:
|
||
height = short_edge
|
||
width = round(short_edge * width_ratio / height_ratio)
|
||
else:
|
||
width = short_edge
|
||
height = round(short_edge * height_ratio / width_ratio)
|
||
return f"{width}x{height}"
|
||
except Exception:
|
||
return "无"
|
||
|
||
|
||
def _build_scaled_bounds(duration: int, ratios: list[float]) -> list[int]:
|
||
duration = max(1, int(duration))
|
||
raw = [0]
|
||
acc = 0.0
|
||
for ratio in ratios[:-1]:
|
||
acc += ratio
|
||
raw.append(max(raw[-1] + 1, min(duration - 1, round(duration * acc))))
|
||
raw.append(duration)
|
||
for index in range(1, len(raw)):
|
||
if raw[index] <= raw[index - 1]:
|
||
raw[index] = min(duration, raw[index - 1] + 1)
|
||
raw[-1] = duration
|
||
return raw
|
||
|
||
|
||
def _bounds_to_plan(bounds: list[int], stages: list[tuple[str, str]]) -> list[dict[str, str]]:
|
||
plan: list[dict[str, str]] = []
|
||
for idx, (stage, desc) in enumerate(stages):
|
||
start = bounds[idx]
|
||
end = bounds[idx + 1]
|
||
plan.append({"时间段": f"{start}-{end}秒", "阶段": stage, "说明": desc})
|
||
return plan
|
||
|
||
|
||
def build_time_plan(duration: int) -> list[dict[str, str]]:
|
||
duration = max(1, int(duration))
|
||
if duration <= 5:
|
||
return _bounds_to_plan(
|
||
_build_scaled_bounds(duration, [0.2, 0.4, 0.4]),
|
||
[
|
||
("开场吸引", "快速建立主体、产品和画面风格"),
|
||
("核心展示", "展示主体动作、核心卖点或主要视觉内容"),
|
||
("行动引导", "强化记忆点并给出转化引导"),
|
||
],
|
||
)
|
||
if duration <= 8:
|
||
return _bounds_to_plan(
|
||
_build_scaled_bounds(duration, [0.15, 0.3, 0.35, 0.2]),
|
||
[
|
||
("开场吸引", "快速吸引注意力"),
|
||
("主体展示", "展示主体和产品关系"),
|
||
("核心卖点", "突出新项目核心内容点"),
|
||
("收尾引导", "给出行动引导并稳定落版"),
|
||
],
|
||
)
|
||
return _bounds_to_plan(
|
||
_build_scaled_bounds(duration, [0.13, 0.2, 0.27, 0.25, 0.15]),
|
||
[
|
||
("爆款开头", "复刻参考素材开头节奏和视觉吸引点"),
|
||
("主体建立", "明确新项目主体和产品信息"),
|
||
("卖点放大", "围绕核心内容点展开动作和镜头"),
|
||
("情绪推进", "用动作、字幕或镜头变化强化记忆"),
|
||
("转化收尾", "给出清晰行动引导"),
|
||
],
|
||
)
|
||
|
||
|
||
def build_dynamic_schema(video_config: dict[str, Any]) -> dict[str, Any]:
|
||
schema = copy.deepcopy(CLIENT_SCHEMA_V1)
|
||
duration = int(video_config["duration"])
|
||
video_ratio = str(video_config["aspect_ratio"])
|
||
resolution = str(video_config["resolution"])
|
||
frame_rate = str(video_config.get("frame_rate") or DEFAULT_FRAME_RATE)
|
||
recommended_resolution = get_recommended_resolution(video_ratio, resolution)
|
||
|
||
schema["画面属性"].update(
|
||
{
|
||
"视频时长": f"{duration}秒",
|
||
"视频比例": video_ratio,
|
||
"清晰度": resolution,
|
||
"推荐分辨率": recommended_resolution,
|
||
"帧率": frame_rate,
|
||
}
|
||
)
|
||
schema["动态时间规划"] = build_time_plan(duration)
|
||
schema["输出规格限制"] = {
|
||
"支持时长": _safe_list(video_config.get("supported_durations")),
|
||
"支持比例": _safe_list(video_config.get("supported_ratios")),
|
||
"支持分辨率": _safe_list(video_config.get("supported_resolutions")),
|
||
"当前推荐分辨率": recommended_resolution,
|
||
}
|
||
return schema
|
||
|
||
|
||
def infer_generation_type(references: list[dict[str, str]] | None) -> str:
|
||
has_image = any(item.get("type") == "image" for item in references or [])
|
||
has_video = any(item.get("type") == "video" for item in references or [])
|
||
if has_image and has_video:
|
||
return "图生视频/视频生视频"
|
||
if has_image:
|
||
return "图生视频"
|
||
if has_video:
|
||
return "视频生视频"
|
||
return "文生视频"
|
||
|
||
|
||
def infer_video_category(text: str) -> tuple[str, str, list[str]]:
|
||
text = text or ""
|
||
if any(key in text for key in ["APP", "应用", "下载", "社交", "脱单", "附近"]):
|
||
return "产品广告视频", "产品推广短视频", ["产品推广", "用户转化", "核心卖点展示"]
|
||
return "产品广告视频", "产品推广短视频", ["产品展示", "视觉吸引", "行动引导"]
|
||
|
||
|
||
def build_system_prompt() -> str:
|
||
return (
|
||
"你是专业短视频广告导演和AI视频提示词工程师。"
|
||
"你必须只输出一个合法 JSON 对象，不能输出 Markdown。"
|
||
"输出必须严格遵循用户提供的 schema 顶层结构。"
|
||
"所有未知、无法判断或不适用的字段填写'无'，数组字段可填写 []。"
|
||
"必须根据参考视频复刻爆款开头的节奏、构图、动作和镜头语言，但不能照抄品牌、水印、字幕或侵权元素。"
|
||
)
|
||
|
||
|
||
def build_user_text(
|
||
*,
|
||
source_project_name: str,
|
||
target_project_name: str,
|
||
core_content_point: str,
|
||
target_platform: str,
|
||
references: list[dict[str, str]],
|
||
video_config: dict[str, Any],
|
||
client_schema: dict[str, Any],
|
||
) -> str:
|
||
duration = int(video_config["duration"])
|
||
video_ratio = str(video_config["aspect_ratio"])
|
||
resolution = str(video_config["resolution"])
|
||
frame_rate = str(video_config.get("frame_rate") or DEFAULT_FRAME_RATE)
|
||
category, sub_category, default_points = infer_video_category(" ".join([source_project_name, target_project_name, core_content_point]))
|
||
return json.dumps(
|
||
{
|
||
"任务": "基于参考素材视频和新项目图片，生成可用于AI视频生成的中文结构化提示词JSON",
|
||
"业务输入": {
|
||
"视频素材内容项目名称": source_project_name,
|
||
"生成项目名称": target_project_name,
|
||
"生成项目核心内容点": core_content_point,
|
||
"目标平台": target_platform,
|
||
"默认视频大类": category,
|
||
"默认视频子类": sub_category,
|
||
"建议核心卖点": default_points,
|
||
},
|
||
"视频规格": {
|
||
"视频时长": f"{duration}秒",
|
||
"视频比例": video_ratio,
|
||
"清晰度": resolution,
|
||
"帧率": frame_rate,
|
||
"支持时长": _safe_list(video_config.get("supported_durations")),
|
||
"支持比例": _safe_list(video_config.get("supported_ratios")),
|
||
"支持分辨率": _safe_list(video_config.get("supported_resolutions")),
|
||
"推荐分辨率": get_recommended_resolution(video_ratio, resolution),
|
||
},
|
||
"参考素材": references,
|
||
"输出要求": {
|
||
"生成类型": infer_generation_type(references),
|
||
"必须填充动态时间规划": build_time_plan(duration),
|
||
"必须填充动作流程": "动作流程时间段必须覆盖完整视频时长",
|
||
"必须填充镜头流程": "镜头流程时间段必须覆盖完整视频时长",
|
||
"最终提示词限制": "最终提示词下所有字段都不能写入视频时长、秒数、视频比例、清晰度、分辨率、帧率、推荐像素、竖屏、横屏等视频规格参数，这些规格只能写在画面属性/动态时间规划/输出规格限制。",
|
||
"禁止": ["输出 Markdown", "输出 schema 之外的解释文字", "照抄参考素材品牌水印", "生成违法违规内容", "在最终提示词中写入秒数/比例/分辨率/帧率"],
|
||
},
|
||
"必须按此schema输出": client_schema,
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
|
||
def build_user_message(user_content: str, references: list[dict[str, str]], reference_video_fps: int) -> tuple[dict[str, Any], dict[str, Any]]:
|
||
content_parts: list[dict[str, Any]] = [{"type": "text", "text": user_content}]
|
||
log_content_parts: list[dict[str, Any]] = [{"type": "text", "text": user_content}]
|
||
for ref in references:
|
||
ref_type = ref.get("type")
|
||
ref_url = ref.get("url")
|
||
if not ref_url:
|
||
continue
|
||
if ref_type == "image":
|
||
content_parts.append({"type": "image_url", "image_url": {"url": ref_url}})
|
||
log_content_parts.append({"type": "image_url", "image_url": {"url": ref_url}})
|
||
elif ref_type == "video":
|
||
content_parts.append({"type": "video_url", "video_url": {"url": ref_url, "fps": reference_video_fps}})
|
||
log_content_parts.append({"type": "video_url", "video_url": {"url": ref_url, "fps": reference_video_fps}})
|
||
return {"role": "user", "content": content_parts}, {"role": "user", "content": log_content_parts}
|
||
|
||
|
||
def strip_json_code_fence(text: str) -> str:
|
||
text = (text or "").strip()
|
||
if text.startswith("```"):
|
||
text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.I)
|
||
text = re.sub(r"\s*```$", "", text)
|
||
return text.strip()
|
||
|
||
|
||
def parse_model_json(content: str) -> dict[str, Any]:
|
||
content = strip_json_code_fence(content)
|
||
data = json.loads(content)
|
||
if not isinstance(data, dict):
|
||
raise ValueError("视频提词优化返回值不是 JSON 对象")
|
||
return data
|
||
|
||
|
||
def fill_none_with_wu(value: Any) -> Any:
|
||
if value is None or value == "":
|
||
return "无"
|
||
if isinstance(value, dict):
|
||
return {k: fill_none_with_wu(v) for k, v in value.items()}
|
||
if isinstance(value, list):
|
||
return [fill_none_with_wu(v) for v in value]
|
||
return value
|
||
|
||
|
||
def ensure_top_keys(result: dict[str, Any]) -> dict[str, Any]:
|
||
schema = copy.deepcopy(CLIENT_SCHEMA_V1)
|
||
for key, default_value in schema.items():
|
||
if key not in result:
|
||
result[key] = default_value
|
||
elif isinstance(default_value, dict) and isinstance(result.get(key), dict):
|
||
merged = copy.deepcopy(default_value)
|
||
merged.update(result[key])
|
||
result[key] = merged
|
||
return result
|
||
|
||
|
||
def ensure_flow_matches_time_plan(result: dict[str, Any], duration: int) -> dict[str, Any]:
|
||
plan = build_time_plan(duration)
|
||
if not isinstance(result.get("动作流程"), list) or not result["动作流程"]:
|
||
result["动作流程"] = [
|
||
{"时间段": item["时间段"], "动作内容": item["说明"]}
|
||
for item in plan
|
||
]
|
||
if not isinstance(result.get("镜头流程"), list) or not result["镜头流程"]:
|
||
result["镜头流程"] = [
|
||
{"时间段": item["时间段"], "镜头内容": item["说明"]}
|
||
for item in plan
|
||
]
|
||
result["动作流程"] = _align_flow_time_ranges(result["动作流程"], plan, "动作内容")
|
||
result["镜头流程"] = _align_flow_time_ranges(result["镜头流程"], plan, "镜头内容")
|
||
result["动态时间规划"] = plan
|
||
return result
|
||
|
||
|
||
def ensure_negative_prompt(result: dict[str, Any]) -> dict[str, Any]:
|
||
final = result.setdefault("最终提示词", {})
|
||
if not isinstance(final, dict):
|
||
final = {}
|
||
result["最终提示词"] = final
|
||
if not final.get("负面提示词") or final.get("负面提示词") == "无":
|
||
final["负面提示词"] = "画面模糊、主体畸变、手指畸形、脸部崩坏、字幕乱码、产品变形、镜头抖动、画面闪烁"
|
||
return result
|
||
|
||
|
||
def build_final_video_prompt(result: dict[str, Any]) -> str:
|
||
final = result.get("最终提示词", {}) if isinstance(result.get("最终提示词"), dict) else {}
|
||
parts = [
|
||
final.get("主提示词"),
|
||
final.get("动作提示词"),
|
||
final.get("镜头提示词"),
|
||
final.get("字幕提示词"),
|
||
final.get("音频提示词"),
|
||
final.get("风格提示词"),
|
||
]
|
||
return "\n".join(str(item).strip() for item in parts if item and str(item).strip() != "无")
|
||
|
||
|
||
VIDEO_SPEC_PROMPT_KEYS = ("主提示词", "动作提示词", "镜头提示词", "字幕提示词", "音频提示词", "风格提示词", "负面提示词")
|
||
|
||
|
||
def _normalize_prompt_text(text: str) -> str:
|
||
text = re.sub(r"[，、,；;：:]\s*([，、,；;：:])", r"\1", text)
|
||
text = re.sub(r"\s{2,}", " ", text)
|
||
text = re.sub(r"^[，、,；;：:\s]+", "", text)
|
||
text = re.sub(r"[，、,；;：:\s]+$", "", text)
|
||
return text.strip() or "无"
|
||
|
||
|
||
def clean_video_spec_from_prompt_text(text: Any, video_config: dict[str, Any] | None = None) -> str:
|
||
"""清洗最终提示词里的视频规格参数。
|
||
|
||
视频时长、比例、清晰度、分辨率、帧率属于接口参数和锁定字段，
|
||
不能混入最终提示词，避免用户绕过扣费参数或与第5步视频生成参数冲突。
|
||
"""
|
||
if text is None:
|
||
return "无"
|
||
value = str(text).strip()
|
||
if not value or value == "无":
|
||
return "无"
|
||
|
||
cfg = video_config or {}
|
||
exact_values = {
|
||
str(cfg.get("aspect_ratio") or "").strip(),
|
||
str(cfg.get("resolution") or "").strip(),
|
||
str(cfg.get("frame_rate") or "").strip(),
|
||
}
|
||
try:
|
||
if cfg.get("duration") is not None:
|
||
exact_values.add(f"{int(cfg.get('duration'))}秒")
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if cfg.get("aspect_ratio") and cfg.get("resolution"):
|
||
exact_values.add(get_recommended_resolution(str(cfg.get("aspect_ratio")), str(cfg.get("resolution"))))
|
||
except Exception:
|
||
pass
|
||
|
||
for item in sorted((v for v in exact_values if v and v != "无"), key=len, reverse=True):
|
||
value = value.replace(item, "")
|
||
|
||
patterns = [
|
||
r"\d+\s*秒",
|
||
r"\b\d+\s*[sS]\b",
|
||
r"\d+\s*[:：]\s*\d+",
|
||
r"\d{3,4}\s*[pP]",
|
||
r"\d{2,4}\s*[xX×]\s*\d{2,4}",
|
||
r"\d+\s*(?:fps|FPS|帧)",
|
||
r"(?:竖屏|横屏|方屏|超清|高清|标清|蓝光|4K|8K)",
|
||
r"(?:视频时长|时长|视频比例|画面比例|比例|分辨率|清晰度|帧率|推荐分辨率)\s*[：:]?\s*",
|
||
]
|
||
for pattern in patterns:
|
||
value = re.sub(pattern, "", value, flags=re.I)
|
||
return _normalize_prompt_text(value)
|
||
|
||
|
||
def clean_final_prompt_specs(schema: dict[str, Any], video_config: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
final = schema.setdefault("最终提示词", {})
|
||
if not isinstance(final, dict):
|
||
final = {}
|
||
schema["最终提示词"] = final
|
||
for key in VIDEO_SPEC_PROMPT_KEYS:
|
||
final[key] = clean_video_spec_from_prompt_text(final.get(key), video_config)
|
||
return schema
|
||
|
||
|
||
def _align_flow_time_ranges(flow: Any, plan: list[dict[str, str]], default_content_key: str) -> list[dict[str, Any]]:
|
||
source = flow if isinstance(flow, list) else []
|
||
aligned: list[dict[str, Any]] = []
|
||
for index, plan_item in enumerate(plan):
|
||
old_item = source[index] if index < len(source) and isinstance(source[index], dict) else {}
|
||
item = dict(old_item)
|
||
item["时间段"] = plan_item["时间段"]
|
||
if not any(k in item and str(item.get(k)).strip() for k in (default_content_key, "动作", "镜头", "说明", "内容")):
|
||
item[default_content_key] = plan_item["说明"]
|
||
aligned.append(item)
|
||
return aligned
|
||
|
||
|
||
def _merge_editable_dict_fields(base: dict[str, Any], patch: dict[str, Any], allowed_keys: set[str]) -> None:
|
||
for key in allowed_keys:
|
||
if key in patch:
|
||
base[key] = fill_none_with_wu(patch.get(key))
|
||
|
||
|
||
def _merge_flow_patch(base_flow: Any, patch_flow: Any) -> list[dict[str, Any]]:
|
||
base = [dict(item) for item in base_flow] if isinstance(base_flow, list) else []
|
||
patch = patch_flow if isinstance(patch_flow, list) else []
|
||
result: list[dict[str, Any]] = []
|
||
for index, base_item in enumerate(base):
|
||
merged = dict(base_item)
|
||
patch_item = patch[index] if index < len(patch) and isinstance(patch[index], dict) else {}
|
||
original_time_range = merged.get("时间段")
|
||
for key, value in patch_item.items():
|
||
if key == "时间段":
|
||
continue
|
||
merged[key] = fill_none_with_wu(value)
|
||
merged["时间段"] = original_time_range
|
||
result.append(merged)
|
||
return result
|
||
|
||
|
||
def apply_locked_video_schema_fields(schema: dict[str, Any], video_config: dict[str, Any]) -> dict[str, Any]:
|
||
duration = int(video_config["duration"])
|
||
video_ratio = str(video_config["aspect_ratio"])
|
||
resolution = str(video_config["resolution"])
|
||
frame_rate = str(video_config.get("frame_rate") or DEFAULT_FRAME_RATE)
|
||
recommended_resolution = get_recommended_resolution(video_ratio, resolution)
|
||
dynamic_schema = build_dynamic_schema(video_config)
|
||
plan = build_time_plan(duration)
|
||
|
||
schema["schema_version"] = PromptSchemaVersionEnum.CLIENT_V1.value
|
||
schema["schema_usage"] = VideoPromptSchemaUsageEnum.CLIENT_DISPLAY.value
|
||
|
||
frame = schema.setdefault("画面属性", {})
|
||
if not isinstance(frame, dict):
|
||
frame = {}
|
||
schema["画面属性"] = frame
|
||
frame.update(
|
||
{
|
||
"视频时长": f"{duration}秒",
|
||
"视频比例": video_ratio,
|
||
"清晰度": resolution,
|
||
"帧率": frame_rate,
|
||
"推荐分辨率": recommended_resolution,
|
||
}
|
||
)
|
||
|
||
schema["动态时间规划"] = plan
|
||
schema["输出规格限制"] = dynamic_schema.get("输出规格限制", {})
|
||
|
||
schema["动作流程"] = _align_flow_time_ranges(schema.get("动作流程"), plan, "动作内容")
|
||
schema["镜头流程"] = _align_flow_time_ranges(schema.get("镜头流程"), plan, "镜头内容")
|
||
|
||
# 合规和质量控制不能被前端降低；AI 返回缺失时使用服务端默认结构补齐。
|
||
default_schema = copy.deepcopy(CLIENT_SCHEMA_V1)
|
||
if not isinstance(schema.get("合规控制"), dict):
|
||
schema["合规控制"] = default_schema["合规控制"]
|
||
if not isinstance(schema.get("质量控制"), dict):
|
||
schema["质量控制"] = default_schema["质量控制"]
|
||
|
||
return clean_final_prompt_specs(schema, video_config)
|
||
|
||
|
||
def normalize_video_prompt_schema_from_ai(result: dict[str, Any], video_config: dict[str, Any]) -> dict[str, Any]:
|
||
duration = int(video_config["duration"])
|
||
normalized = ensure_top_keys(fill_none_with_wu(result if isinstance(result, dict) else {}))
|
||
normalized = ensure_flow_matches_time_plan(normalized, duration)
|
||
normalized = ensure_negative_prompt(normalized)
|
||
return apply_locked_video_schema_fields(normalized, video_config)
|
||
|
||
|
||
def patch_video_prompt_schema_from_client(
|
||
*,
|
||
server_schema: dict[str, Any],
|
||
client_schema: dict[str, Any],
|
||
video_config: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
"""以前端 JSON 作为 patch，回填到服务端已有 schema。
|
||
|
||
禁止整包覆盖：数组长度、时间段、视频规格、输出规格、质量控制、合规控制、schema 协议字段均以服务端为准。
|
||
"""
|
||
base = ensure_top_keys(fill_none_with_wu(copy.deepcopy(server_schema if isinstance(server_schema, dict) else {})))
|
||
patch = client_schema if isinstance(client_schema, dict) else {}
|
||
|
||
for key in ("基础分类", "素材理解", "业务属性", "字幕与口播", "音频与节奏"):
|
||
if isinstance(base.get(key), dict) and isinstance(patch.get(key), dict):
|
||
base[key].update(fill_none_with_wu(patch[key]))
|
||
|
||
if isinstance(base.get("画面属性"), dict) and isinstance(patch.get("画面属性"), dict):
|
||
_merge_editable_dict_fields(
|
||
base["画面属性"],
|
||
patch["画面属性"],
|
||
{"主体描述", "主体数量", "主体位置", "主体占比", "场景描述", "构图方式", "画面风格", "光影色彩"},
|
||
)
|
||
|
||
if isinstance(patch.get("动作流程"), list):
|
||
base["动作流程"] = _merge_flow_patch(base.get("动作流程"), patch.get("动作流程"))
|
||
if isinstance(patch.get("镜头流程"), list):
|
||
base["镜头流程"] = _merge_flow_patch(base.get("镜头流程"), patch.get("镜头流程"))
|
||
|
||
# 动态时间规划保持服务端数组长度和时间段，只允许保留原值；不接受客户端 patch。
|
||
|
||
if isinstance(base.get("最终提示词"), dict) and isinstance(patch.get("最终提示词"), dict):
|
||
for key in VIDEO_SPEC_PROMPT_KEYS:
|
||
if key in patch["最终提示词"]:
|
||
base["最终提示词"][key] = fill_none_with_wu(patch["最终提示词"].get(key))
|
||
|
||
return apply_locked_video_schema_fields(base, video_config)
|
||
|
||
|
||
def _mock_result(video_config: dict[str, Any], target_platform: str) -> dict[str, Any]:
|
||
duration = int(video_config["duration"])
|
||
video_ratio = str(video_config["aspect_ratio"])
|
||
resolution = str(video_config["resolution"])
|
||
schema = build_dynamic_schema(video_config)
|
||
schema["基础分类"].update({"生成类型": "图生视频/视频生视频", "视频大类": "产品广告视频", "视频子类": "产品推广短视频", "视频用途": "社媒发布", "目标平台": target_platform})
|
||
schema["素材理解"].update({"是否有参考图片": "是", "是否有参考视频": "是", "参考视频用途": "动作参考/镜头参考/风格参考/节奏参考"})
|
||
schema["业务属性"].update({"产品类型": "APP", "核心表达目标": "突出新项目核心内容点", "内容风格": "轻快、活泼、广告感适中", "行动引导": "立即体验"})
|
||
schema["动作流程"] = [{"时间段": item["时间段"], "动作": item["阶段"], "说明": item["说明"]} for item in build_time_plan(duration)]
|
||
schema["镜头流程"] = [{"时间段": item["时间段"], "镜头": item["阶段"], "说明": item["说明"]} for item in build_time_plan(duration)]
|
||
schema["最终提示词"] = {
|
||
"主提示词": f"生成一段{duration}秒、{video_ratio}、{resolution}的产品推广短视频，参考素材视频的爆款开头节奏，结合新项目图片进行自然展示。",
|
||
"动作提示词": "主体动作自然，产品展示稳定，节奏轻快。",
|
||
"镜头提示词": "镜头稳定，开头快速吸引注意，后续平滑推进。",
|
||
"字幕提示词": "字幕简洁清晰，突出核心内容点。",
|
||
"音频提示词": "轻快背景音乐，节奏自然。",
|
||
"风格提示词": "年轻化、明亮、真实、广告感适中。",
|
||
"负面提示词": "画面模糊、主体畸变、手指畸形、脸部崩坏、字幕乱码、产品变形、镜头抖动、画面闪烁",
|
||
}
|
||
return schema
|
||
|
||
|
||
async def _select_model_config(db: AsyncSession) -> ModelConfig | None:
|
||
result = await db.execute(select(ModelConfig).where(ModelConfig.is_active == True).order_by(ModelConfig.priority.desc()).limit(1))
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
async def optimize_hot_opening_video_prompt(
|
||
db: AsyncSession,
|
||
*,
|
||
user_id: str,
|
||
source_project_name: str,
|
||
target_project_name: str,
|
||
core_content_point: str,
|
||
material_video_url: str,
|
||
generated_image_url: str,
|
||
video_config: dict[str, Any],
|
||
target_platform: str = "抖音",
|
||
) -> tuple[dict[str, Any], str, dict[str, Any]]:
|
||
duration = int(video_config["duration"])
|
||
references = [
|
||
{"type": "video", "url": _build_file_url_or_data_uri(material_video_url)},
|
||
{"type": "image", "url": _build_file_url_or_data_uri(generated_image_url)},
|
||
]
|
||
client_schema = build_dynamic_schema(video_config)
|
||
reference_video_fps = int(video_config.get("reference_video_fps") or DEFAULT_REFERENCE_VIDEO_FPS)
|
||
|
||
# if settings.LLM_MOCK:
|
||
# result = _mock_result(video_config, target_platform)
|
||
# result = ensure_negative_prompt(ensure_flow_matches_time_plan(ensure_top_keys(fill_none_with_wu(result)), duration))
|
||
# return result, build_final_video_prompt(result), {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
|
||
|
||
config = await _select_model_config(db)
|
||
if not config:
|
||
result = normalize_video_prompt_schema_from_ai(_mock_result(video_config, target_platform), video_config)
|
||
return result, build_final_video_prompt(result), {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
|
||
|
||
user_text = build_user_text(
|
||
source_project_name=source_project_name,
|
||
target_project_name=target_project_name,
|
||
core_content_point=core_content_point,
|
||
target_platform=target_platform,
|
||
references=references,
|
||
video_config=video_config,
|
||
client_schema=client_schema,
|
||
)
|
||
user_message, log_user_message = build_user_message(user_text, references, reference_video_fps)
|
||
request_data = {
|
||
"model": config.model_name,
|
||
"messages": [{"role": "system", "content": build_system_prompt()}, user_message],
|
||
"max_tokens": 6000,
|
||
"temperature": 0.15,
|
||
"response_format": {"type": "json_object"},
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=int(settings.CHATAPI_REQUEST_TIMEOUT_SECONDS or 180)) as client:
|
||
response = await client.post(
|
||
f"{config.api_base.rstrip('/')}/chat/completions",
|
||
headers={"Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json"},
|
||
json=request_data,
|
||
)
|
||
if response.status_code >= 400:
|
||
raise RuntimeError(f"视频提词优化失败 HTTP {response.status_code}: {response.text}")
|
||
|
||
data = response.json()
|
||
content = data["choices"][0]["message"]["content"].strip()
|
||
usage = data.get("usage", {}) or {}
|
||
token_usage = {
|
||
"input_tokens": int(usage.get("prompt_tokens") or 0),
|
||
"output_tokens": int(usage.get("completion_tokens") or 0),
|
||
"total_tokens": int(usage.get("total_tokens") or 0),
|
||
"log_user_message": log_user_message,
|
||
}
|
||
db.add(
|
||
TokenUsage(
|
||
id=generate_id(),
|
||
model_config_id=config.id,
|
||
user_id=user_id,
|
||
input_tokens=token_usage["input_tokens"],
|
||
output_tokens=token_usage["output_tokens"],
|
||
total_tokens=token_usage["total_tokens"],
|
||
)
|
||
)
|
||
await db.flush()
|
||
|
||
result = parse_model_json(content)
|
||
result = normalize_video_prompt_schema_from_ai(result, video_config)
|
||
return result, build_final_video_prompt(result), token_usage
|
||
|
||
def _build_file_url_or_data_uri(file_url: str) -> str:
|
||
"""
|
||
Convert local upload path to base64 data URI.
|
||
Keep remote http/https/data URLs as-is.
|
||
"""
|
||
if file_url.startswith(("http://", "https://", "data:")):
|
||
return file_url
|
||
file_url_sign = build_resource_signed_url(resource_url=file_url, expire_seconds=86400)
|
||
return f"{settings.BASE_URL}{file_url_sign}" |