mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

爆款开头API开发完成

Browse Source
This commit is contained in:
GinHa
2026-06-10 15:17:51 +08:00
parent 949bb794cc
commit 2923b1d6e1
16 changed files with 4034 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/services/hot_opening_video_prompt_service.py
+707
View File
@@ -0,0 +1,707 @@
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": 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}"