1769 lines
80 KiB
Python
1769 lines
80 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.common import LogEventStatusEnum, LogSourceEnum
|
||
from app.enums.hot_opening_replicate import HotOpeningLogEventEnum, HotOpeningRemoteActionEnum, ModuleCodeEnum as HotModuleCodeEnum
|
||
from app.enums.shot_replicate import ModuleCodeEnum as ShotModuleCodeEnum, ShotReplicateLogEventEnum, ShotReplicateRemoteActionEnum
|
||
from app.services.operation_log_service import log_ai_model_event
|
||
from app.enums.common import (
|
||
VIDEO_SCHEMA_CONFIG_DATABASE_SOURCE,
|
||
VIDEO_SCHEMA_CONFIG_DEFAULT_SOURCE,
|
||
VIDEO_SCHEMA_CONFIG_VERSION,
|
||
VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN,
|
||
VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN,
|
||
VIDEO_SCHEMA_TIME_DESC_MAX_LEN,
|
||
VIDEO_SCHEMA_TIME_STAGE_MAX_LEN,
|
||
)
|
||
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
|
||
|
||
# 后台配置可以控制业务字段是否启用/可编辑，但视频规格字段属于接口参数，
|
||
# 必须强制保留并锁定，避免管理端误删/禁用后导致第 4 步和第 5 步参数不一致。
|
||
VIDEO_SPEC_SECTION_KEY = "画面属性"
|
||
VIDEO_SPEC_LOCKED_FIELDS = {"视频时长", "视频比例", "清晰度", "帧率", "推荐分辨率"}
|
||
ALWAYS_ALLOWED_TOP_LEVEL_KEYS = {"schema_version", "schema_usage", "动态时间规划", "输出规格限制", VIDEO_SPEC_SECTION_KEY}
|
||
TIME_PLAN_PLACEHOLDER_TEXTS = {"", "无", "阶段说明", "说明", "null", "None", "none", "未提及", "不适用"}
|
||
|
||
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 _default_time_plan_rules() -> list[dict[str, Any]]:
|
||
return [
|
||
{
|
||
"min_duration": 1,
|
||
"max_duration": 5,
|
||
"ratios": [0.2, 0.4, 0.4],
|
||
"segments": [
|
||
{"stage": "开场吸引", "description": "快速建立主体、产品和画面风格"},
|
||
{"stage": "核心展示", "description": "展示主体动作、核心卖点或主要视觉内容"},
|
||
{"stage": "行动引导", "description": "强化记忆点并给出转化引导"},
|
||
],
|
||
},
|
||
{
|
||
"min_duration": 6,
|
||
"max_duration": 8,
|
||
"ratios": [0.15, 0.3, 0.35, 0.2],
|
||
"segments": [
|
||
{"stage": "开场吸引", "description": "快速吸引注意力"},
|
||
{"stage": "主体展示", "description": "展示主体和产品关系"},
|
||
{"stage": "核心卖点", "description": "突出新项目核心内容点"},
|
||
{"stage": "收尾引导", "description": "给出行动引导并稳定落版"},
|
||
],
|
||
},
|
||
{
|
||
"min_duration": 9,
|
||
"max_duration": 15,
|
||
"ratios": [0.13, 0.2, 0.27, 0.25, 0.15],
|
||
"segments": [
|
||
{"stage": "爆款开头", "description": "复刻参考素材开头节奏和视觉吸引点"},
|
||
{"stage": "主体建立", "description": "明确新项目主体和产品信息"},
|
||
{"stage": "卖点放大", "description": "围绕核心内容点展开动作和镜头"},
|
||
{"stage": "情绪推进", "description": "用动作、字幕或镜头变化强化记忆"},
|
||
{"stage": "转化收尾", "description": "给出清晰行动引导"},
|
||
],
|
||
},
|
||
]
|
||
|
||
|
||
def _normalize_bool(value: Any, default: bool = True) -> bool:
|
||
if isinstance(value, bool):
|
||
return value
|
||
if value is None:
|
||
return default
|
||
return str(value).strip().lower() not in {"0", "false", "no", "off", "否", "禁用"}
|
||
|
||
|
||
def _normalize_int(value: Any, default: int, *, min_value: int = 1, max_value: int = 100000) -> int:
|
||
try:
|
||
number = int(value)
|
||
except Exception:
|
||
number = default
|
||
return max(min_value, min(max_value, number))
|
||
|
||
|
||
def _truncate_for_config(value: Any, limit: int) -> str:
|
||
text = str(value or "").strip()
|
||
return text[:limit]
|
||
|
||
|
||
def _schema_value_type(value: Any) -> str:
|
||
if isinstance(value, dict):
|
||
return "object"
|
||
if isinstance(value, list):
|
||
return "array"
|
||
if isinstance(value, bool):
|
||
return "boolean"
|
||
if isinstance(value, (int, float)):
|
||
return "number"
|
||
return "string"
|
||
|
||
|
||
def _default_node(key: str, value: Any, *, editable: bool = True) -> dict[str, Any]:
|
||
node: dict[str, Any] = {
|
||
"key": str(key),
|
||
"label": str(key),
|
||
"type": _schema_value_type(value),
|
||
"enabled": True,
|
||
"editable": editable,
|
||
"max_length": VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN,
|
||
}
|
||
if isinstance(value, dict):
|
||
node["children"] = [_default_node(child_key, child_value, editable=editable) for child_key, child_value in value.items()]
|
||
else:
|
||
node["value"] = copy.deepcopy(value)
|
||
return node
|
||
|
||
|
||
def default_video_prompt_schema_config() -> dict[str, Any]:
|
||
sections: list[dict[str, Any]] = []
|
||
for key, value in CLIENT_SCHEMA_V1.items():
|
||
if key in {"schema_version", "schema_usage"}:
|
||
continue
|
||
if key == "动作流程":
|
||
sections.append(
|
||
{
|
||
"key": "动作流程",
|
||
"label": "动作流程",
|
||
"type": "flow",
|
||
"enabled": True,
|
||
"editable": True,
|
||
"content_key": "动作内容",
|
||
"content_aliases": list(ACTION_FLOW_CONTENT_KEYS),
|
||
"max_length": VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN,
|
||
"item_fields": [
|
||
{"key": "动作内容", "label": "动作内容", "enabled": True, "editable": True, "max_length": VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN, "value": ""}
|
||
],
|
||
}
|
||
)
|
||
elif key == "镜头流程":
|
||
sections.append(
|
||
{
|
||
"key": "镜头流程",
|
||
"label": "镜头流程",
|
||
"type": "flow",
|
||
"enabled": True,
|
||
"editable": True,
|
||
"content_key": "镜头内容",
|
||
"content_aliases": list(CAMERA_FLOW_CONTENT_KEYS),
|
||
"max_length": VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN,
|
||
"item_fields": [
|
||
{"key": "镜头内容", "label": "镜头内容", "enabled": True, "editable": True, "max_length": VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN, "value": ""}
|
||
],
|
||
}
|
||
)
|
||
else:
|
||
sections.append(_default_node(key, value, editable=True))
|
||
return {
|
||
"version": VIDEO_SCHEMA_CONFIG_VERSION,
|
||
"enabled": True,
|
||
"sections": sections,
|
||
"time_plan_rules": _default_time_plan_rules(),
|
||
}
|
||
|
||
|
||
def _extract_config_data(schema_config_snapshot: Any | None) -> dict[str, Any] | None:
|
||
if not isinstance(schema_config_snapshot, dict):
|
||
return None
|
||
data = schema_config_snapshot.get("data")
|
||
if isinstance(data, dict):
|
||
return data
|
||
if schema_config_snapshot.get("sections") or schema_config_snapshot.get("time_plan_rules"):
|
||
return schema_config_snapshot
|
||
return None
|
||
|
||
|
||
def normalize_video_prompt_schema_config(config: Any | None) -> dict[str, Any]:
|
||
source = config if isinstance(config, dict) else {}
|
||
default_config = default_video_prompt_schema_config()
|
||
if not source:
|
||
return default_config
|
||
|
||
normalized: dict[str, Any] = {
|
||
"version": str(source.get("version") or VIDEO_SCHEMA_CONFIG_VERSION),
|
||
"enabled": _normalize_bool(source.get("enabled"), True),
|
||
"sections": [],
|
||
"time_plan_rules": [],
|
||
}
|
||
|
||
raw_sections = source.get("sections") if isinstance(source.get("sections"), list) else default_config["sections"]
|
||
for section in raw_sections:
|
||
if not isinstance(section, dict):
|
||
continue
|
||
key = _truncate_for_config(section.get("key") or section.get("label"), 64)
|
||
if not key:
|
||
continue
|
||
section_type = str(section.get("type") or "object")
|
||
if key == VIDEO_SPEC_SECTION_KEY:
|
||
section_type = "object"
|
||
|
||
item: dict[str, Any] = {
|
||
"key": key,
|
||
"label": _truncate_for_config(section.get("label") or key, 64),
|
||
"type": section_type,
|
||
"enabled": True if key == VIDEO_SPEC_SECTION_KEY else _normalize_bool(section.get("enabled"), True),
|
||
"editable": _normalize_bool(section.get("editable"), True),
|
||
"max_length": _normalize_int(section.get("max_length"), VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN, min_value=1, max_value=20000),
|
||
}
|
||
if section_type == "object":
|
||
children = section.get("children") if isinstance(section.get("children"), list) else []
|
||
item["children"] = []
|
||
for child in children:
|
||
if not isinstance(child, dict):
|
||
continue
|
||
child_key = _truncate_for_config(child.get("key") or child.get("label"), 64)
|
||
if not child_key:
|
||
continue
|
||
is_locked_video_spec = key == VIDEO_SPEC_SECTION_KEY and child_key in VIDEO_SPEC_LOCKED_FIELDS
|
||
item["children"].append(
|
||
{
|
||
"key": child_key,
|
||
"label": _truncate_for_config(child.get("label") or child_key, 64),
|
||
"type": str(child.get("type") or _schema_value_type(child.get("value"))),
|
||
"enabled": True if is_locked_video_spec else _normalize_bool(child.get("enabled"), True),
|
||
"editable": False if is_locked_video_spec else _normalize_bool(child.get("editable"), True),
|
||
"max_length": _normalize_int(child.get("max_length") or child.get("maxLength"), VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN, min_value=1, max_value=20000),
|
||
"value": copy.deepcopy(child.get("value", "无")),
|
||
}
|
||
)
|
||
if key == VIDEO_SPEC_SECTION_KEY:
|
||
existing_child_keys = {str(child.get("key") or "") for child in item["children"] if isinstance(child, dict)}
|
||
for locked_key in ("视频时长", "视频比例", "清晰度", "帧率", "推荐分辨率"):
|
||
if locked_key not in existing_child_keys:
|
||
item["children"].append(
|
||
{
|
||
"key": locked_key,
|
||
"label": locked_key,
|
||
"type": "string",
|
||
"enabled": True,
|
||
"editable": False,
|
||
"max_length": VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN,
|
||
"value": "无",
|
||
}
|
||
)
|
||
elif section_type == "flow":
|
||
content_key = _truncate_for_config(section.get("content_key") or section.get("contentKey") or ("镜头内容" if key == "镜头流程" else "动作内容"), 64)
|
||
aliases = section.get("content_aliases") or section.get("contentAliases")
|
||
if not isinstance(aliases, list) or not aliases:
|
||
aliases = list(CAMERA_FLOW_CONTENT_KEYS if key == "镜头流程" else ACTION_FLOW_CONTENT_KEYS)
|
||
item["content_key"] = content_key
|
||
item["content_aliases"] = [_truncate_for_config(alias, 64) for alias in aliases if _truncate_for_config(alias, 64)]
|
||
raw_fields = section.get("item_fields") or section.get("itemFields")
|
||
if not isinstance(raw_fields, list) or not raw_fields:
|
||
raw_fields = [{"key": content_key, "label": content_key, "enabled": True, "editable": True, "max_length": item["max_length"], "value": ""}]
|
||
item["item_fields"] = []
|
||
for field in raw_fields:
|
||
if not isinstance(field, dict):
|
||
continue
|
||
field_key = _truncate_for_config(field.get("key") or field.get("label"), 64)
|
||
if not field_key or field_key == "时间段":
|
||
continue
|
||
item["item_fields"].append(
|
||
{
|
||
"key": field_key,
|
||
"label": _truncate_for_config(field.get("label") or field_key, 64),
|
||
"enabled": _normalize_bool(field.get("enabled"), True),
|
||
"editable": _normalize_bool(field.get("editable"), True),
|
||
"max_length": _normalize_int(field.get("max_length"), item["max_length"], min_value=1, max_value=20000),
|
||
"value": copy.deepcopy(field.get("value", "")),
|
||
}
|
||
)
|
||
if not any(field["key"] == content_key for field in item["item_fields"]):
|
||
item["item_fields"].insert(0, {"key": content_key, "label": content_key, "enabled": True, "editable": True, "max_length": item["max_length"], "value": ""})
|
||
else:
|
||
item["value"] = copy.deepcopy(section.get("value", [] if section_type == "array" else "无"))
|
||
normalized["sections"].append(item)
|
||
|
||
raw_rules = source.get("time_plan_rules") or source.get("timePlanRules")
|
||
if not isinstance(raw_rules, list) or not raw_rules:
|
||
raw_rules = default_config["time_plan_rules"]
|
||
for rule in raw_rules:
|
||
if not isinstance(rule, dict):
|
||
continue
|
||
segments = rule.get("segments") if isinstance(rule.get("segments"), list) else []
|
||
ratios = rule.get("ratios") if isinstance(rule.get("ratios"), list) else []
|
||
normalized_segments: list[dict[str, str]] = []
|
||
for segment in segments:
|
||
if not isinstance(segment, dict):
|
||
continue
|
||
normalized_segments.append(
|
||
{
|
||
"stage": _truncate_for_config(segment.get("stage") or "无", VIDEO_SCHEMA_TIME_STAGE_MAX_LEN),
|
||
"description": _truncate_for_config(segment.get("description") or segment.get("desc") or "无", VIDEO_SCHEMA_TIME_DESC_MAX_LEN),
|
||
}
|
||
)
|
||
if not normalized_segments:
|
||
continue
|
||
try:
|
||
normalized_ratios = [float(item) for item in ratios]
|
||
except Exception:
|
||
normalized_ratios = []
|
||
if len(normalized_ratios) != len(normalized_segments):
|
||
normalized_ratios = [1 / len(normalized_segments)] * len(normalized_segments)
|
||
ratio_total = sum(item for item in normalized_ratios if item > 0)
|
||
if ratio_total <= 0:
|
||
normalized_ratios = [1 / len(normalized_segments)] * len(normalized_segments)
|
||
else:
|
||
normalized_ratios = [max(0.01, item) / ratio_total for item in normalized_ratios]
|
||
normalized["time_plan_rules"].append(
|
||
{
|
||
"min_duration": _normalize_int(rule.get("min_duration") or rule.get("minDuration"), 1, min_value=1, max_value=3600),
|
||
"max_duration": _normalize_int(rule.get("max_duration") or rule.get("maxDuration"), 15, min_value=1, max_value=3600),
|
||
"ratios": normalized_ratios,
|
||
"segments": normalized_segments,
|
||
}
|
||
)
|
||
if not normalized["time_plan_rules"]:
|
||
normalized["time_plan_rules"] = default_config["time_plan_rules"]
|
||
return normalized
|
||
|
||
|
||
def build_client_schema_from_config(schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
data = normalize_video_prompt_schema_config(_extract_config_data(schema_config_snapshot))
|
||
if not _normalize_bool(data.get("enabled"), True):
|
||
data = default_video_prompt_schema_config()
|
||
schema: dict[str, Any] = {
|
||
"schema_version": PromptSchemaVersionEnum.CLIENT_V1.value,
|
||
"schema_usage": VideoPromptSchemaUsageEnum.CLIENT_DISPLAY.value,
|
||
}
|
||
for section in data.get("sections", []):
|
||
if not isinstance(section, dict) or not _normalize_bool(section.get("enabled"), True):
|
||
continue
|
||
key = str(section.get("key") or "").strip()
|
||
if not key:
|
||
continue
|
||
section_type = str(section.get("type") or "object")
|
||
if section_type == "object":
|
||
obj: dict[str, Any] = {}
|
||
for child in section.get("children") or []:
|
||
if not isinstance(child, dict) or not _normalize_bool(child.get("enabled"), True):
|
||
continue
|
||
child_key = str(child.get("key") or "").strip()
|
||
if not child_key:
|
||
continue
|
||
obj[child_key] = copy.deepcopy(child.get("value", "无"))
|
||
schema[key] = obj
|
||
elif section_type == "flow":
|
||
schema[key] = []
|
||
else:
|
||
schema[key] = copy.deepcopy(section.get("value", [] if section_type == "array" else "无"))
|
||
return schema
|
||
|
||
|
||
def build_schema_config_snapshot(schema_config: Any | None = None, *, source: str = VIDEO_SCHEMA_CONFIG_DEFAULT_SOURCE) -> dict[str, Any]:
|
||
data = normalize_video_prompt_schema_config(_extract_config_data(schema_config) or schema_config)
|
||
if not data:
|
||
data = default_video_prompt_schema_config()
|
||
return {"version": VIDEO_SCHEMA_CONFIG_VERSION, "source": source, "data": data}
|
||
|
||
|
||
def _select_time_plan_rule(duration: int, schema_config_snapshot: Any | None = None) -> dict[str, Any] | None:
|
||
data = normalize_video_prompt_schema_config(_extract_config_data(schema_config_snapshot))
|
||
rules = data.get("time_plan_rules") if isinstance(data.get("time_plan_rules"), list) else []
|
||
for rule in rules:
|
||
if not isinstance(rule, dict):
|
||
continue
|
||
min_duration = _normalize_int(rule.get("min_duration"), 1, min_value=1, max_value=3600)
|
||
max_duration = _normalize_int(rule.get("max_duration"), min_duration, min_value=1, max_value=3600)
|
||
if min_duration <= duration <= max_duration:
|
||
return rule
|
||
return None
|
||
|
||
|
||
def build_time_plan(duration: int, schema_config_snapshot: Any | None = None) -> list[dict[str, str]]:
|
||
duration = max(1, int(duration))
|
||
rule = _select_time_plan_rule(duration, schema_config_snapshot)
|
||
if rule:
|
||
segments = rule.get("segments") if isinstance(rule.get("segments"), list) else []
|
||
ratios = rule.get("ratios") if isinstance(rule.get("ratios"), list) else []
|
||
stages: list[tuple[str, str]] = []
|
||
for segment in segments:
|
||
if isinstance(segment, dict):
|
||
stages.append((str(segment.get("stage") or "无"), str(segment.get("description") or "无")))
|
||
if stages:
|
||
try:
|
||
ratio_values = [float(item) for item in ratios]
|
||
except Exception:
|
||
ratio_values = []
|
||
if len(ratio_values) != len(stages):
|
||
ratio_values = [1 / len(stages)] * len(stages)
|
||
ratio_total = sum(item for item in ratio_values if item > 0)
|
||
if ratio_total <= 0:
|
||
ratio_values = [1 / len(stages)] * len(stages)
|
||
else:
|
||
ratio_values = [max(0.01, item) / ratio_total for item in ratio_values]
|
||
return _bounds_to_plan(_build_scaled_bounds(duration, ratio_values), stages)
|
||
# fallback 保留旧规则，避免配置为空或异常时影响原流程。
|
||
return _bounds_to_plan(
|
||
_build_scaled_bounds(duration, [0.2, 0.4, 0.4]),
|
||
[
|
||
("开场吸引", "快速建立主体、产品和画面风格"),
|
||
("核心展示", "展示主体动作、核心卖点或主要视觉内容"),
|
||
("行动引导", "强化记忆点并给出转化引导"),
|
||
],
|
||
) if duration <= 5 else (
|
||
_bounds_to_plan(
|
||
_build_scaled_bounds(duration, [0.15, 0.3, 0.35, 0.2]),
|
||
[
|
||
("开场吸引", "快速吸引注意力"),
|
||
("主体展示", "展示主体和产品关系"),
|
||
("核心卖点", "突出新项目核心内容点"),
|
||
("收尾引导", "给出行动引导并稳定落版"),
|
||
],
|
||
) if duration <= 8 else _bounds_to_plan(
|
||
_build_scaled_bounds(duration, [0.13, 0.2, 0.27, 0.25, 0.15]),
|
||
[
|
||
("爆款开头", "复刻参考素材开头节奏和视觉吸引点"),
|
||
("主体建立", "明确新项目主体和产品信息"),
|
||
("卖点放大", "围绕核心内容点展开动作和镜头"),
|
||
("情绪推进", "用动作、字幕或镜头变化强化记忆"),
|
||
("转化收尾", "给出清晰行动引导"),
|
||
],
|
||
)
|
||
)
|
||
|
||
|
||
def _time_plan_schema_for_ai_input(plan: list[dict[str, str]]) -> list[dict[str, str]]:
|
||
"""
|
||
AI 入参 schema 只固定时间段和字段结构，不把后台配置的阶段/说明当成最终分析结果。
|
||
阶段/说明应由 AI 结合参考素材和新项目重新填写，后续归一化仅在 AI 未填写时兜底。
|
||
"""
|
||
result: list[dict[str, str]] = []
|
||
for item in plan:
|
||
result.append(
|
||
{
|
||
"时间段": item.get("时间段") or "无",
|
||
"阶段": "",
|
||
"说明": "",
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def _time_plan_time_ranges_for_ai(plan: list[dict[str, str]]) -> list[dict[str, str]]:
|
||
return [{"时间段": item.get("时间段") or "无"} for item in plan]
|
||
|
||
|
||
def build_dynamic_schema(video_config: dict[str, Any], schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
schema = build_client_schema_from_config(schema_config_snapshot)
|
||
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)
|
||
|
||
frame = schema.setdefault("画面属性", {})
|
||
if not isinstance(frame, dict):
|
||
frame = {}
|
||
schema["画面属性"] = frame
|
||
frame.update(
|
||
{
|
||
"视频时长": f"{duration}秒",
|
||
"视频比例": video_ratio,
|
||
"清晰度": resolution,
|
||
"推荐分辨率": recommended_resolution,
|
||
"帧率": frame_rate,
|
||
}
|
||
)
|
||
time_plan = build_time_plan(duration, schema_config_snapshot)
|
||
schema["动态时间规划"] = _time_plan_schema_for_ai_input(time_plan)
|
||
|
||
# 预览/AI 入参阶段只锁定时间段和字段结构，不把后台阶段/说明/流程说明写死到 schema 值里。
|
||
# 阶段、说明、动作内容、动作详解、镜头内容、镜头详解都应由 AI 基于素材分析填写；
|
||
# 只有 AI 返回缺失/空/无/阶段说明等占位内容时，归一化阶段才使用后台规则兜底。
|
||
for flow_key in ("动作流程", "镜头流程"):
|
||
if flow_key not in schema:
|
||
continue
|
||
content_key = _flow_content_key(schema_config_snapshot, flow_key)
|
||
schema[flow_key] = _align_flow_time_ranges(
|
||
schema.get(flow_key),
|
||
time_plan,
|
||
content_key,
|
||
content_keys=_flow_content_aliases(schema_config_snapshot, flow_key),
|
||
item_fields=_flow_item_field_rules(schema_config_snapshot, flow_key),
|
||
fallback_to_plan=False,
|
||
)
|
||
|
||
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],
|
||
schema_config_snapshot: Any | None = None,
|
||
) -> 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),
|
||
"动态时间规划时间段必须严格等于": _time_plan_time_ranges_for_ai(build_time_plan(duration, schema_config_snapshot)),
|
||
"动态时间规划兜底参考": build_time_plan(duration, schema_config_snapshot),
|
||
"动态时间规划填写要求": "时间段必须和给定时间段一致；阶段、说明必须结合参考素材、新项目、核心内容点、动作和镜头重新分析填写；只有无法判断时才允许使用兜底参考；不能原样复制后台配置里的阶段说明。",
|
||
"必须填充动作流程": "动作流程时间段必须覆盖完整视频时长，每一段都必须填写动作流程对象中的全部字段。",
|
||
"动作流程字段要求": [field.get("key") for field in _flow_item_field_rules(schema_config_snapshot, "动作流程")],
|
||
"必须填充镜头流程": "镜头流程时间段必须覆盖完整视频时长，每一段都必须填写镜头流程对象中的全部字段。",
|
||
"镜头流程字段要求": [field.get("key") for field in _flow_item_field_rules(schema_config_snapshot, "镜头流程")],
|
||
"字段启用规则": "只输出 schema 中存在的启用字段；不要输出已禁用字段；不要新增 schema 外字段。",
|
||
"最终提示词限制": "最终提示词下所有字段都不能写入视频时长、秒数、视频比例、清晰度、分辨率、帧率、推荐像素、竖屏、横屏等视频规格参数，这些规格只能写在画面属性/动态时间规划/输出规格限制。",
|
||
"禁止": ["输出 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}})
|
||
log_content_parts.append({"type": "video_url", "video_url": {"url": ref_url}})
|
||
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
|
||
|
||
|
||
TOP_LEVEL_SCHEMA_KEYS = tuple(CLIENT_SCHEMA_V1.keys()) + ("动态时间规划", "输出规格限制")
|
||
|
||
OBJECT_FIELD_WHITELISTS: dict[str, set[str]] = {
|
||
key: set(value.keys())
|
||
for key, value in CLIENT_SCHEMA_V1.items()
|
||
if isinstance(value, dict)
|
||
}
|
||
OBJECT_FIELD_WHITELISTS.setdefault("画面属性", set()).add("推荐分辨率")
|
||
OBJECT_FIELD_WHITELISTS["输出规格限制"] = {"支持时长", "支持比例", "支持分辨率", "当前推荐分辨率"}
|
||
|
||
ACTION_FLOW_CONTENT_KEYS = ("动作内容", "动作", "动作说明", "内容", "说明", "主体动作", "动作变化")
|
||
CAMERA_FLOW_CONTENT_KEYS = ("镜头内容", "镜头", "镜头说明", "运镜", "运镜说明", "内容", "说明")
|
||
TIME_PLAN_ALLOWED_KEYS = ("时间段", "阶段", "说明")
|
||
PLACEHOLDER_FLOW_TEXTS = {
|
||
"展示主体动作、核心卖点或主要视觉内容",
|
||
"展示主要动作、核心卖点或主要视觉内容",
|
||
"展示核心卖点或主要视觉内容",
|
||
"展示主体动作",
|
||
"无",
|
||
}
|
||
EMPTY_VALUE_TEXTS = {"", "无", "null", "None", "none", "未提及", "不适用"}
|
||
|
||
|
||
def _clean_schema_text(value: Any) -> str:
|
||
if value is None:
|
||
return ""
|
||
if isinstance(value, (dict, list)):
|
||
try:
|
||
return json.dumps(value, ensure_ascii=False)
|
||
except Exception:
|
||
return str(value)
|
||
return str(value).strip()
|
||
|
||
|
||
def _is_empty_schema_value(value: Any) -> bool:
|
||
return _clean_schema_text(value) in EMPTY_VALUE_TEXTS
|
||
|
||
|
||
def _schema_default_for_normalize(schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
schema = build_client_schema_from_config(schema_config_snapshot)
|
||
schema.setdefault("schema_version", PromptSchemaVersionEnum.CLIENT_V1.value)
|
||
schema.setdefault("schema_usage", VideoPromptSchemaUsageEnum.CLIENT_DISPLAY.value)
|
||
return schema
|
||
|
||
|
||
def _object_field_whitelists(schema_config_snapshot: Any | None = None) -> dict[str, set[str]]:
|
||
default_schema = _schema_default_for_normalize(schema_config_snapshot)
|
||
whitelists: dict[str, set[str]] = {
|
||
key: set(value.keys())
|
||
for key, value in default_schema.items()
|
||
if isinstance(value, dict)
|
||
}
|
||
whitelists.setdefault("画面属性", set()).update({"视频时长", "视频比例", "清晰度", "帧率", "推荐分辨率"})
|
||
whitelists["输出规格限制"] = {"支持时长", "支持比例", "支持分辨率", "当前推荐分辨率"}
|
||
return whitelists
|
||
|
||
|
||
def _normalize_schema_object(section_key: str, value: Any, schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
default_schema = _schema_default_for_normalize(schema_config_snapshot)
|
||
default_value = default_schema.get(section_key)
|
||
if not isinstance(default_value, dict):
|
||
default_value = {}
|
||
source = value if isinstance(value, dict) else {}
|
||
merged = copy.deepcopy(default_value)
|
||
allowed_keys = _object_field_whitelists(schema_config_snapshot).get(section_key, set(default_value.keys()))
|
||
for field_key in allowed_keys:
|
||
if field_key in source:
|
||
merged[field_key] = fill_none_with_wu(source.get(field_key))
|
||
return merged
|
||
|
||
|
||
def normalize_top_level_schema_fields(result: dict[str, Any], schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
"""按客户端视频提词 schema 白名单清洗顶层和普通对象字段。"""
|
||
source = result if isinstance(result, dict) else {}
|
||
default_schema = _schema_default_for_normalize(schema_config_snapshot)
|
||
whitelists = _object_field_whitelists(schema_config_snapshot)
|
||
top_level_schema_keys = tuple(default_schema.keys()) + ("动态时间规划", "输出规格限制")
|
||
normalized: dict[str, Any] = {}
|
||
for key in top_level_schema_keys:
|
||
if key in whitelists:
|
||
normalized[key] = _normalize_schema_object(key, source.get(key), schema_config_snapshot)
|
||
elif key in source:
|
||
normalized[key] = fill_none_with_wu(source.get(key))
|
||
elif key in default_schema:
|
||
normalized[key] = copy.deepcopy(default_schema[key])
|
||
for key, default_value in default_schema.items():
|
||
if key not in normalized:
|
||
normalized[key] = copy.deepcopy(default_value)
|
||
return normalized
|
||
|
||
|
||
def ensure_top_keys(result: dict[str, Any], schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
return normalize_top_level_schema_fields(result, schema_config_snapshot)
|
||
|
||
|
||
def _flow_section_config(schema_config_snapshot: Any | None, flow_key: str) -> dict[str, Any]:
|
||
data = normalize_video_prompt_schema_config(_extract_config_data(schema_config_snapshot))
|
||
for section in data.get("sections", []):
|
||
if isinstance(section, dict) and section.get("key") == flow_key and section.get("type") == "flow" and _normalize_bool(section.get("enabled"), True):
|
||
return section
|
||
default_key = "镜头内容" if flow_key == "镜头流程" else "动作内容"
|
||
aliases = CAMERA_FLOW_CONTENT_KEYS if flow_key == "镜头流程" else ACTION_FLOW_CONTENT_KEYS
|
||
return {
|
||
"key": flow_key,
|
||
"type": "flow",
|
||
"enabled": True,
|
||
"editable": True,
|
||
"content_key": default_key,
|
||
"content_aliases": list(aliases),
|
||
"max_length": VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN,
|
||
"item_fields": [{"key": default_key, "enabled": True, "editable": True, "max_length": VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN, "value": ""}],
|
||
}
|
||
|
||
|
||
def _flow_content_key(schema_config_snapshot: Any | None, flow_key: str) -> str:
|
||
section = _flow_section_config(schema_config_snapshot, flow_key)
|
||
return str(section.get("content_key") or ("镜头内容" if flow_key == "镜头流程" else "动作内容"))
|
||
|
||
|
||
def _flow_content_aliases(schema_config_snapshot: Any | None, flow_key: str) -> tuple[str, ...]:
|
||
section = _flow_section_config(schema_config_snapshot, flow_key)
|
||
aliases = section.get("content_aliases") if isinstance(section.get("content_aliases"), list) else []
|
||
values = [str(item) for item in aliases if str(item).strip()]
|
||
content_key = _flow_content_key(schema_config_snapshot, flow_key)
|
||
if content_key not in values:
|
||
values.insert(0, content_key)
|
||
return tuple(values)
|
||
|
||
|
||
def _flow_item_field_rules(schema_config_snapshot: Any | None, flow_key: str) -> list[dict[str, Any]]:
|
||
section = _flow_section_config(schema_config_snapshot, flow_key)
|
||
fields = section.get("item_fields") if isinstance(section.get("item_fields"), list) else []
|
||
result: list[dict[str, Any]] = []
|
||
for field in fields:
|
||
if not isinstance(field, dict) or not _normalize_bool(field.get("enabled"), True):
|
||
continue
|
||
key = str(field.get("key") or "").strip()
|
||
if not key or key == "时间段":
|
||
continue
|
||
result.append(
|
||
{
|
||
"key": key,
|
||
"editable": _normalize_bool(field.get("editable"), True),
|
||
"max_length": _normalize_int(field.get("max_length"), VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN, min_value=1, max_value=20000),
|
||
"value": field.get("value", ""),
|
||
}
|
||
)
|
||
content_key = _flow_content_key(schema_config_snapshot, flow_key)
|
||
if not any(item["key"] == content_key for item in result):
|
||
result.insert(0, {"key": content_key, "editable": True, "max_length": VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN, "value": ""})
|
||
return result
|
||
|
||
|
||
def _pick_flow_content(item: dict[str, Any], content_keys: tuple[str, ...]) -> str:
|
||
for key in content_keys:
|
||
if key in item and not _is_empty_schema_value(item.get(key)):
|
||
return _clean_schema_text(item.get(key))
|
||
return ""
|
||
|
||
|
||
def _collect_extra_flow_texts(
|
||
item: dict[str, Any],
|
||
*,
|
||
content_keys: tuple[str, ...],
|
||
allowed_keys: set[str],
|
||
) -> list[str]:
|
||
extras: list[str] = []
|
||
for key, value in item.items():
|
||
if key in allowed_keys or key in content_keys:
|
||
continue
|
||
key_text = _clean_schema_text(key)
|
||
value_text = _clean_schema_text(value)
|
||
if key_text and key_text not in EMPTY_VALUE_TEXTS:
|
||
extras.append(key_text)
|
||
if value_text and value_text not in EMPTY_VALUE_TEXTS and value_text != key_text:
|
||
extras.append(value_text)
|
||
# 去重但保留顺序，避免 AI 重复写入同一句。
|
||
deduped: list[str] = []
|
||
for text in extras:
|
||
if text not in deduped:
|
||
deduped.append(text)
|
||
return deduped
|
||
|
||
|
||
def _merge_flow_content(base_content: str, extra_texts: list[str], fallback: str) -> str:
|
||
base_content = _clean_schema_text(base_content)
|
||
if extra_texts and (not base_content or base_content in PLACEHOLDER_FLOW_TEXTS or base_content == fallback):
|
||
return "；".join(extra_texts)
|
||
parts = [base_content] if base_content and base_content not in EMPTY_VALUE_TEXTS else []
|
||
for text in extra_texts:
|
||
if text and text not in parts:
|
||
parts.append(text)
|
||
return "；".join(parts) if parts else fallback
|
||
|
||
|
||
def _normalize_flow_item(
|
||
raw_item: Any,
|
||
*,
|
||
plan_item: dict[str, str],
|
||
content_key: str,
|
||
content_keys: tuple[str, ...],
|
||
item_fields: list[dict[str, Any]] | None = None,
|
||
fallback_to_plan: bool = True,
|
||
) -> dict[str, str]:
|
||
item = raw_item if isinstance(raw_item, dict) else {}
|
||
field_rules = item_fields or [{"key": content_key, "value": ""}]
|
||
allowed_keys = {"时间段", *(str(field.get("key")) for field in field_rules if field.get("key"))}
|
||
base_content = _pick_flow_content(item, content_keys)
|
||
extra_texts = _collect_extra_flow_texts(item, content_keys=content_keys, allowed_keys=allowed_keys)
|
||
fallback = plan_item.get("说明") or "无"
|
||
empty_fallback = "无" if fallback_to_plan else ""
|
||
result: dict[str, str] = {"时间段": plan_item.get("时间段") or _clean_schema_text(item.get("时间段")) or "无"}
|
||
for field in field_rules:
|
||
field_key = str(field.get("key") or "").strip()
|
||
if not field_key:
|
||
continue
|
||
if field_key == content_key:
|
||
if fallback_to_plan:
|
||
result[field_key] = _merge_flow_content(base_content, extra_texts, fallback)
|
||
else:
|
||
result[field_key] = _merge_flow_content(base_content, extra_texts, "")
|
||
else:
|
||
# AI 入参预览阶段不能把后台默认值写死到流程字段中；只输出字段结构。
|
||
# AI 返回归一化阶段才允许在缺失/无时用字段默认值或通用占位兜底。
|
||
raw_value = _clean_schema_text(item.get(field_key))
|
||
if raw_value and raw_value not in EMPTY_VALUE_TEXTS and raw_value not in TIME_PLAN_PLACEHOLDER_TEXTS:
|
||
result[field_key] = raw_value
|
||
elif fallback_to_plan:
|
||
result[field_key] = _clean_schema_text(field.get("value")) or "无"
|
||
else:
|
||
result[field_key] = empty_fallback
|
||
return result
|
||
|
||
|
||
def _normalize_time_plan(plan: list[dict[str, str]], value: Any) -> list[dict[str, str]]:
|
||
source = value if isinstance(value, list) else []
|
||
normalized: list[dict[str, str]] = []
|
||
for index, plan_item in enumerate(plan):
|
||
raw_item = source[index] if index < len(source) and isinstance(source[index], dict) else {}
|
||
raw_stage = _clean_schema_text(raw_item.get("阶段"))
|
||
raw_desc = _clean_schema_text(raw_item.get("说明"))
|
||
if raw_stage in TIME_PLAN_PLACEHOLDER_TEXTS:
|
||
raw_stage = ""
|
||
if raw_desc in TIME_PLAN_PLACEHOLDER_TEXTS:
|
||
raw_desc = ""
|
||
# 只有时间段由服务端锁定；阶段、说明优先保留 AI 结合素材分析后的结果。
|
||
# 仅当 AI 未填写、填写“无/阶段说明/说明”等占位内容时，才使用后台时间切片规则兜底。
|
||
item: dict[str, str] = {
|
||
"时间段": plan_item.get("时间段") or _clean_schema_text(raw_item.get("时间段")) or "无",
|
||
"阶段": raw_stage or plan_item.get("阶段") or "无",
|
||
"说明": raw_desc or plan_item.get("说明") or "无",
|
||
}
|
||
normalized.append(item)
|
||
return normalized
|
||
|
||
def ensure_flow_matches_time_plan(result: dict[str, Any], duration: int, schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
base_plan = build_time_plan(duration, schema_config_snapshot)
|
||
plan = _normalize_time_plan(base_plan, result.get("动态时间规划"))
|
||
action_key = _flow_content_key(schema_config_snapshot, "动作流程")
|
||
camera_key = _flow_content_key(schema_config_snapshot, "镜头流程")
|
||
result["动作流程"] = _align_flow_time_ranges(
|
||
result.get("动作流程"),
|
||
plan,
|
||
action_key,
|
||
content_keys=_flow_content_aliases(schema_config_snapshot, "动作流程"),
|
||
item_fields=_flow_item_field_rules(schema_config_snapshot, "动作流程"),
|
||
)
|
||
result["镜头流程"] = _align_flow_time_ranges(
|
||
result.get("镜头流程"),
|
||
plan,
|
||
camera_key,
|
||
content_keys=_flow_content_aliases(schema_config_snapshot, "镜头流程"),
|
||
item_fields=_flow_item_field_rules(schema_config_snapshot, "镜头流程"),
|
||
)
|
||
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,
|
||
*,
|
||
content_keys: tuple[str, ...] | None = None,
|
||
item_fields: list[dict[str, Any]] | None = None,
|
||
fallback_to_plan: bool = True,
|
||
) -> list[dict[str, str]]:
|
||
source = flow if isinstance(flow, list) else []
|
||
aliases = content_keys or (ACTION_FLOW_CONTENT_KEYS if default_content_key == "动作内容" else CAMERA_FLOW_CONTENT_KEYS)
|
||
aligned: list[dict[str, str]] = []
|
||
for index, plan_item in enumerate(plan):
|
||
raw_item = source[index] if index < len(source) else {}
|
||
aligned.append(
|
||
_normalize_flow_item(
|
||
raw_item,
|
||
plan_item=plan_item,
|
||
content_key=default_content_key,
|
||
content_keys=aliases,
|
||
item_fields=item_fields,
|
||
fallback_to_plan=fallback_to_plan,
|
||
)
|
||
)
|
||
return aligned
|
||
|
||
|
||
def _config_editable_field_rules(schema_config_snapshot: Any | None = None) -> dict[tuple[str, str], dict[str, Any]]:
|
||
data = normalize_video_prompt_schema_config(_extract_config_data(schema_config_snapshot))
|
||
rules: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for section in data.get("sections", []):
|
||
if not isinstance(section, dict) or not _normalize_bool(section.get("enabled"), True):
|
||
continue
|
||
if str(section.get("type") or "object") != "object":
|
||
continue
|
||
section_key = str(section.get("key") or "").strip()
|
||
for child in section.get("children") or []:
|
||
if not isinstance(child, dict) or not _normalize_bool(child.get("enabled"), True):
|
||
continue
|
||
field_key = str(child.get("key") or "").strip()
|
||
if not section_key or not field_key:
|
||
continue
|
||
rules[(section_key, field_key)] = {
|
||
"editable": _normalize_bool(child.get("editable"), True),
|
||
"max_length": _normalize_int(child.get("max_length"), VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN, min_value=1, max_value=20000),
|
||
}
|
||
return rules
|
||
|
||
|
||
def _validate_editable_value_length(path: str, value: Any, max_length: int) -> None:
|
||
if isinstance(value, list):
|
||
for index, item in enumerate(value):
|
||
_validate_editable_value_length(f"{path}[{index}]", item, max_length)
|
||
return
|
||
if isinstance(value, dict):
|
||
for key, item in value.items():
|
||
_validate_editable_value_length(f"{path}.{key}", item, max_length)
|
||
return
|
||
text = str(value or "")
|
||
if len(text) > max_length:
|
||
raise ValueError(f"字段【{path}】长度不能超过 {max_length} 个字符")
|
||
|
||
|
||
def _is_editable_config_field(section_key: str, field_key: str, schema_config_snapshot: Any | None = None) -> tuple[bool, int]:
|
||
# 视频规格、动态规划、输出限制、合规和质量控制继续按原逻辑锁定。
|
||
if section_key in {"合规控制", "质量控制", "输出规格限制"}:
|
||
return False, VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN
|
||
if section_key == "画面属性" and field_key in {"视频时长", "视频比例", "清晰度", "帧率", "推荐分辨率"}:
|
||
return False, VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN
|
||
rules = _config_editable_field_rules(schema_config_snapshot)
|
||
rule = rules.get((section_key, field_key))
|
||
if rule is None:
|
||
return False, VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN
|
||
return bool(rule.get("editable", True)), int(rule.get("max_length") or VIDEO_SCHEMA_EDITABLE_TEXT_MAX_LEN)
|
||
|
||
|
||
def _merge_editable_dict_fields(
|
||
section_key: str,
|
||
base: dict[str, Any],
|
||
patch: dict[str, Any],
|
||
allowed_keys: set[str],
|
||
schema_config_snapshot: Any | None = None,
|
||
) -> None:
|
||
for key in allowed_keys:
|
||
if key not in patch:
|
||
continue
|
||
editable, max_length = _is_editable_config_field(section_key, key, schema_config_snapshot)
|
||
if not editable:
|
||
continue
|
||
_validate_editable_value_length(f"{section_key}.{key}", patch.get(key), max_length)
|
||
base[key] = fill_none_with_wu(patch.get(key))
|
||
|
||
|
||
def _merge_flow_patch(base_flow: Any, patch_flow: Any, flow_key: str, schema_config_snapshot: Any | None = None) -> 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]] = []
|
||
field_rules = {item["key"]: item for item in _flow_item_field_rules(schema_config_snapshot, flow_key)}
|
||
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 == "时间段" or key not in field_rules:
|
||
continue
|
||
rule = field_rules[key]
|
||
if not _normalize_bool(rule.get("editable"), True):
|
||
continue
|
||
max_length = _normalize_int(rule.get("max_length"), VIDEO_SCHEMA_FLOW_CONTENT_MAX_LEN, min_value=1, max_value=20000)
|
||
_validate_editable_value_length(f"{flow_key}[{index}].{key}", value, max_length)
|
||
merged[key] = fill_none_with_wu(value)
|
||
merged["时间段"] = original_time_range
|
||
result.append(merged)
|
||
return result
|
||
|
||
|
||
def _prune_schema_by_runtime_config(schema: dict[str, Any], schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
default_schema = build_client_schema_from_config(schema_config_snapshot)
|
||
allowed_top_keys = set(default_schema.keys()) | ALWAYS_ALLOWED_TOP_LEVEL_KEYS
|
||
pruned: dict[str, Any] = {}
|
||
for key, value in schema.items():
|
||
if key not in allowed_top_keys:
|
||
continue
|
||
if isinstance(value, dict) and isinstance(default_schema.get(key), dict):
|
||
allowed_fields = set(default_schema[key].keys())
|
||
if key == VIDEO_SPEC_SECTION_KEY:
|
||
allowed_fields |= VIDEO_SPEC_LOCKED_FIELDS
|
||
pruned[key] = {field_key: field_value for field_key, field_value in value.items() if field_key in allowed_fields}
|
||
else:
|
||
pruned[key] = value
|
||
|
||
if VIDEO_SPEC_SECTION_KEY not in pruned or not isinstance(pruned.get(VIDEO_SPEC_SECTION_KEY), dict):
|
||
pruned[VIDEO_SPEC_SECTION_KEY] = {}
|
||
return pruned
|
||
|
||
|
||
def apply_locked_video_schema_fields(schema: dict[str, Any], video_config: dict[str, Any], schema_config_snapshot: Any | None = None) -> 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, schema_config_snapshot)
|
||
plan = build_time_plan(duration, schema_config_snapshot)
|
||
|
||
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["动态时间规划"] = _normalize_time_plan(plan, schema.get("动态时间规划"))
|
||
schema["输出规格限制"] = dynamic_schema.get("输出规格限制", {})
|
||
|
||
default_schema = build_client_schema_from_config(schema_config_snapshot)
|
||
if "动作流程" in default_schema:
|
||
schema["动作流程"] = _align_flow_time_ranges(
|
||
schema.get("动作流程"),
|
||
schema["动态时间规划"],
|
||
_flow_content_key(schema_config_snapshot, "动作流程"),
|
||
content_keys=_flow_content_aliases(schema_config_snapshot, "动作流程"),
|
||
item_fields=_flow_item_field_rules(schema_config_snapshot, "动作流程"),
|
||
)
|
||
else:
|
||
schema.pop("动作流程", None)
|
||
|
||
if "镜头流程" in default_schema:
|
||
schema["镜头流程"] = _align_flow_time_ranges(
|
||
schema.get("镜头流程"),
|
||
schema["动态时间规划"],
|
||
_flow_content_key(schema_config_snapshot, "镜头流程"),
|
||
content_keys=_flow_content_aliases(schema_config_snapshot, "镜头流程"),
|
||
item_fields=_flow_item_field_rules(schema_config_snapshot, "镜头流程"),
|
||
)
|
||
else:
|
||
schema.pop("镜头流程", None)
|
||
|
||
# 合规和质量控制不能被前端降低；仅当配置启用对应分组时才补齐，禁用分组必须从最终 schema 中移除。
|
||
if not isinstance(schema.get("合规控制"), dict) and isinstance(default_schema.get("合规控制"), dict):
|
||
schema["合规控制"] = copy.deepcopy(default_schema["合规控制"])
|
||
if not isinstance(schema.get("质量控制"), dict) and isinstance(default_schema.get("质量控制"), dict):
|
||
schema["质量控制"] = copy.deepcopy(default_schema["质量控制"])
|
||
|
||
return _prune_schema_by_runtime_config(clean_final_prompt_specs(schema, video_config), schema_config_snapshot)
|
||
|
||
|
||
def normalize_video_prompt_schema_from_ai(result: dict[str, Any], video_config: dict[str, Any], schema_config_snapshot: Any | None = None) -> dict[str, Any]:
|
||
duration = int(video_config["duration"])
|
||
normalized = ensure_top_keys(fill_none_with_wu(result if isinstance(result, dict) else {}), schema_config_snapshot)
|
||
normalized = ensure_flow_matches_time_plan(normalized, duration, schema_config_snapshot)
|
||
normalized = ensure_negative_prompt(normalized)
|
||
return apply_locked_video_schema_fields(normalized, video_config, schema_config_snapshot)
|
||
|
||
|
||
def patch_video_prompt_schema_from_client(
|
||
*,
|
||
server_schema: dict[str, Any],
|
||
client_schema: dict[str, Any],
|
||
video_config: dict[str, Any],
|
||
schema_config_snapshot: Any | None = None,
|
||
) -> dict[str, Any]:
|
||
"""以前端 JSON 作为 patch，回填到服务端已有 schema。
|
||
|
||
禁止整包覆盖：数组长度、时间段、视频规格、输出规格、质量控制、合规控制、schema 协议字段均以服务端为准。
|
||
仅对 schema_config_snapshot 中 enabled=true 且 editable=true 的用户可编辑字段做长度校验，不自动裁剪。
|
||
"""
|
||
base = ensure_top_keys(fill_none_with_wu(copy.deepcopy(server_schema if isinstance(server_schema, dict) else {})), schema_config_snapshot)
|
||
patch = client_schema if isinstance(client_schema, dict) else {}
|
||
whitelists = _object_field_whitelists(schema_config_snapshot)
|
||
|
||
for section_key, allowed_keys in whitelists.items():
|
||
if section_key in {"画面属性", "输出规格限制", "合规控制", "质量控制"}:
|
||
continue
|
||
if isinstance(base.get(section_key), dict) and isinstance(patch.get(section_key), dict):
|
||
_merge_editable_dict_fields(section_key, base[section_key], patch[section_key], allowed_keys, schema_config_snapshot)
|
||
|
||
if isinstance(base.get("画面属性"), dict) and isinstance(patch.get("画面属性"), dict):
|
||
_merge_editable_dict_fields(
|
||
"画面属性",
|
||
base["画面属性"],
|
||
patch["画面属性"],
|
||
{"主体描述", "主体数量", "主体位置", "主体占比", "场景描述", "构图方式", "画面风格", "光影色彩"},
|
||
schema_config_snapshot,
|
||
)
|
||
|
||
if isinstance(patch.get("动作流程"), list):
|
||
base["动作流程"] = _merge_flow_patch(base.get("动作流程"), patch.get("动作流程"), "动作流程", schema_config_snapshot)
|
||
if isinstance(patch.get("镜头流程"), list):
|
||
base["镜头流程"] = _merge_flow_patch(base.get("镜头流程"), patch.get("镜头流程"), "镜头流程", schema_config_snapshot)
|
||
|
||
# 动态时间规划保持服务端数组长度和时间段，只允许保留原值；不接受客户端 patch。
|
||
|
||
if isinstance(base.get("最终提示词"), dict) and isinstance(patch.get("最终提示词"), dict):
|
||
for key in VIDEO_SPEC_PROMPT_KEYS:
|
||
if key not in patch["最终提示词"]:
|
||
continue
|
||
editable, max_length = _is_editable_config_field("最终提示词", key, schema_config_snapshot)
|
||
if not editable:
|
||
continue
|
||
_validate_editable_value_length(f"最终提示词.{key}", patch["最终提示词"].get(key), max_length)
|
||
base["最终提示词"][key] = fill_none_with_wu(patch["最终提示词"].get(key))
|
||
|
||
return apply_locked_video_schema_fields(base, video_config, schema_config_snapshot)
|
||
|
||
|
||
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
|
||
|
||
|
||
|
||
|
||
def _safe_response_json(response: httpx.Response | None) -> Any:
|
||
if response is None:
|
||
return None
|
||
try:
|
||
return response.json()
|
||
except Exception:
|
||
return {"raw_text": response.text}
|
||
|
||
|
||
def _extract_remote_request_id(data: Any) -> str | None:
|
||
if isinstance(data, dict):
|
||
error = data.get("error") if isinstance(data.get("error"), dict) else {}
|
||
request_id = data.get("request_id") or error.get("request_id")
|
||
message = error.get("message") or data.get("message")
|
||
if not request_id and isinstance(message, str):
|
||
import re
|
||
match = re.search(r"Request id:\s*([a-zA-Z0-9_.:-]+)", message, flags=re.IGNORECASE)
|
||
if match:
|
||
request_id = match.group(1)
|
||
return str(request_id) if request_id else None
|
||
return None
|
||
|
||
|
||
def _video_prompt_remote_event(module: str, *, started: bool = False, success: bool = False, parse_failed: bool = False, empty: bool = False) -> tuple[str, str]:
|
||
if module == ShotModuleCodeEnum.SHOT_REPLICATE.value:
|
||
action = ShotReplicateRemoteActionEnum.VIDEO_PROMPT_OPTIMIZE.value
|
||
if started:
|
||
return ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_STARTED.value, action
|
||
if success:
|
||
return ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_SUCCESS.value, action
|
||
if parse_failed:
|
||
return ShotReplicateLogEventEnum.ANALYSIS_RESPONSE_PARSE_FAILED.value, action
|
||
if empty:
|
||
return ShotReplicateLogEventEnum.ANALYSIS_RESPONSE_EMPTY.value, action
|
||
return ShotReplicateLogEventEnum.ANALYSIS_REMOTE_API_FAILED.value, action
|
||
action = HotOpeningRemoteActionEnum.VIDEO_PROMPT_OPTIMIZE.value
|
||
if started:
|
||
return HotOpeningLogEventEnum.VIDEO_PROMPT_REMOTE_API_STARTED.value, action
|
||
if success:
|
||
return HotOpeningLogEventEnum.VIDEO_PROMPT_REMOTE_API_SUCCESS.value, action
|
||
if parse_failed:
|
||
return HotOpeningLogEventEnum.VIDEO_PROMPT_RESPONSE_PARSE_FAILED.value, action
|
||
if empty:
|
||
return HotOpeningLogEventEnum.VIDEO_PROMPT_RESPONSE_EMPTY.value, action
|
||
return HotOpeningLogEventEnum.VIDEO_PROMPT_REMOTE_API_FAILED.value, action
|
||
|
||
|
||
def _log_video_prompt_ai_event(
|
||
*,
|
||
module: str,
|
||
event_type: str,
|
||
action: str,
|
||
event_status: str,
|
||
config: ModelConfig,
|
||
trace_id: str | None,
|
||
user_id: str | None,
|
||
project_id: str | None,
|
||
step_id: str | None,
|
||
request_data: dict[str, Any] | None = None,
|
||
response_data: Any = None,
|
||
token_usage: dict[str, Any] | None = None,
|
||
http_status: int | None = None,
|
||
remote_request_id: str | None = None,
|
||
message: str | None = None,
|
||
error: str | None = None,
|
||
detail: dict[str, Any] | None = None,
|
||
) -> None:
|
||
log_ai_model_event(
|
||
event_type=event_type,
|
||
event_status=event_status,
|
||
source=LogSourceEnum.REMOTE_API.value,
|
||
module=module,
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
remote_action=action,
|
||
remote_request_id=remote_request_id,
|
||
model_config_id=str(config.id),
|
||
model_config_name=config.name,
|
||
model_name=config.model_name,
|
||
provider=config.provider,
|
||
api_base=config.api_base,
|
||
http_status=http_status,
|
||
request=request_data,
|
||
response=response_data,
|
||
token_usage=token_usage,
|
||
message=message,
|
||
detail=detail,
|
||
error=error,
|
||
)
|
||
|
||
async def _select_model_config(db: AsyncSession) -> ModelConfig | None:
|
||
result = await db.execute(select(ModelConfig).where(ModelConfig.is_active == True, ModelConfig.deleted_at.is_(None)).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 | None,
|
||
video_config: dict[str, Any],
|
||
target_platform: str = "抖音",
|
||
schema_config_snapshot: Any | None = None,
|
||
module: str = HotModuleCodeEnum.HOT_OPENING_REPLICATE.value,
|
||
project_id: str | None = None,
|
||
step_id: str | None = None,
|
||
trace_id: str | None = None,
|
||
) -> tuple[dict[str, Any], str, dict[str, Any]]:
|
||
duration = int(video_config["duration"])
|
||
from app.utils.media import media_to_base64, get_llm_media_as_base64
|
||
use_base64 = await get_llm_media_as_base64(db)
|
||
if use_base64:
|
||
video_url_final = await media_to_base64(material_video_url, "video/mp4")
|
||
else:
|
||
video_url_final = _build_file_url_or_data_uri(material_video_url)
|
||
references = [{"type": "video", "url": video_url_final}]
|
||
if generated_image_url:
|
||
image_url_final = (
|
||
await media_to_base64(generated_image_url, "image/png")
|
||
if use_base64
|
||
else _build_file_url_or_data_uri(generated_image_url)
|
||
)
|
||
references.append({"type": "image", "url": image_url_final})
|
||
client_schema = build_dynamic_schema(video_config, schema_config_snapshot)
|
||
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)
|
||
# All module/project claims are committed by the caller. Release this
|
||
# configuration read transaction before the remote model request.
|
||
await db.commit()
|
||
if not config:
|
||
result = normalize_video_prompt_schema_from_ai(_mock_result(video_config, target_platform), video_config, schema_config_snapshot)
|
||
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,
|
||
schema_config_snapshot=schema_config_snapshot,
|
||
)
|
||
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"},
|
||
}
|
||
api_url = f"{config.api_base.rstrip('/')}/chat/completions"
|
||
log_request_data = {
|
||
**request_data,
|
||
"messages": [{"role": "system", "content": build_system_prompt()}, log_user_message],
|
||
"model_config_id": config.id,
|
||
"model_config_name": config.name,
|
||
"provider": config.provider,
|
||
"module": module,
|
||
"project_id": project_id,
|
||
"step_id": step_id,
|
||
"material_video_url": material_video_url,
|
||
"generated_image_url": generated_image_url,
|
||
"target_platform": target_platform,
|
||
"api_url": api_url,
|
||
}
|
||
started_event, remote_action = _video_prompt_remote_event(module, started=True)
|
||
_log_video_prompt_ai_event(
|
||
module=module,
|
||
event_type=started_event,
|
||
action=remote_action,
|
||
event_status=LogEventStatusEnum.STARTED.value,
|
||
config=config,
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
request_data=log_request_data,
|
||
message="视频提词模型请求开始",
|
||
)
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=int(settings.CHATAPI_REQUEST_TIMEOUT_SECONDS or 180)) as client:
|
||
response = await client.post(
|
||
api_url,
|
||
headers={"Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json"},
|
||
json=request_data,
|
||
)
|
||
except Exception as exc:
|
||
failed_event, remote_action = _video_prompt_remote_event(module)
|
||
_log_video_prompt_ai_event(
|
||
module=module,
|
||
event_type=failed_event,
|
||
action=remote_action,
|
||
event_status=LogEventStatusEnum.FAILED.value,
|
||
config=config,
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
request_data=log_request_data,
|
||
message="视频提词模型请求异常",
|
||
error=str(exc),
|
||
detail={"exception_type": type(exc).__name__},
|
||
)
|
||
raise
|
||
response_data = _safe_response_json(response)
|
||
remote_request_id = _extract_remote_request_id(response_data)
|
||
if response.status_code >= 400:
|
||
failed_event, remote_action = _video_prompt_remote_event(module)
|
||
_log_video_prompt_ai_event(
|
||
module=module,
|
||
event_type=failed_event,
|
||
action=remote_action,
|
||
event_status=LogEventStatusEnum.FAILED.value,
|
||
config=config,
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
request_data=log_request_data,
|
||
response_data=response_data,
|
||
http_status=response.status_code,
|
||
remote_request_id=remote_request_id,
|
||
message="视频提词模型请求失败",
|
||
error=f"HTTP {response.status_code}: {response.text}",
|
||
)
|
||
raise RuntimeError(f"视频提词优化失败 HTTP {response.status_code}: {response.text}")
|
||
|
||
try:
|
||
data = response.json()
|
||
content = data["choices"][0]["message"]["content"].strip()
|
||
if not content:
|
||
raise RuntimeError("视频提词模型响应 content 为空")
|
||
except Exception as exc:
|
||
parse_event, remote_action = _video_prompt_remote_event(module, empty="content 为空" in str(exc), parse_failed="content 为空" not in str(exc))
|
||
_log_video_prompt_ai_event(
|
||
module=module,
|
||
event_type=parse_event,
|
||
action=remote_action,
|
||
event_status=LogEventStatusEnum.FAILED.value,
|
||
config=config,
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
request_data=log_request_data,
|
||
response_data=response_data,
|
||
http_status=response.status_code,
|
||
remote_request_id=remote_request_id,
|
||
message="视频提词模型响应解析失败",
|
||
error=str(exc),
|
||
)
|
||
raise
|
||
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,
|
||
}
|
||
token_usage_id = generate_id()
|
||
db.add(
|
||
TokenUsage(
|
||
id=token_usage_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()
|
||
token_usage.update({
|
||
"token_usage_id": token_usage_id,
|
||
"model_config_id": config.id,
|
||
"model_config_name": config.name,
|
||
"model_provider": config.provider,
|
||
"model_name": config.model_name,
|
||
})
|
||
|
||
try:
|
||
result = parse_model_json(content)
|
||
result = normalize_video_prompt_schema_from_ai(result, video_config, schema_config_snapshot)
|
||
except Exception as exc:
|
||
parse_event, remote_action = _video_prompt_remote_event(module, parse_failed=True)
|
||
_log_video_prompt_ai_event(
|
||
module=module,
|
||
event_type=parse_event,
|
||
action=remote_action,
|
||
event_status=LogEventStatusEnum.FAILED.value,
|
||
config=config,
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
request_data=log_request_data,
|
||
response_data=data,
|
||
http_status=response.status_code,
|
||
remote_request_id=remote_request_id,
|
||
message="视频提词业务 JSON 解析失败",
|
||
error=str(exc),
|
||
)
|
||
raise
|
||
success_event, remote_action = _video_prompt_remote_event(module, success=True)
|
||
_log_video_prompt_ai_event(
|
||
module=module,
|
||
event_type=success_event,
|
||
action=remote_action,
|
||
event_status=LogEventStatusEnum.SUCCESS.value,
|
||
config=config,
|
||
trace_id=trace_id,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
step_id=step_id,
|
||
request_data=log_request_data,
|
||
response_data=data,
|
||
token_usage=token_usage,
|
||
http_status=response.status_code,
|
||
remote_request_id=remote_request_id,
|
||
message="视频提词模型请求成功",
|
||
)
|
||
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}" |