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 ( 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, "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 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 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 = "抖音", schema_config_snapshot: Any | None = None, ) -> 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, 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) 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"}, } 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, schema_config_snapshot) 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}"