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

视频提词优化管理后台配置BUG修复

Browse Source
This commit is contained in:
GinHa
2026-06-22 15:56:09 +08:00
parent ada0243919
commit 03e5b049b4
5 changed files with 597 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/services/hot_opening_video_prompt_service.py
+139 -37
View File
@@ -28,6 +28,13 @@ 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,
@@ -332,11 +339,14 @@ def normalize_video_prompt_schema_config(config: Any | None) -> dict[str, Any]:
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": _normalize_bool(section.get("enabled"), True),
"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),
}
@@ -349,17 +359,33 @@ def normalize_video_prompt_schema_config(config: Any | None) -> dict[str, Any]:
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": _normalize_bool(child.get("enabled"), True),
"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),
"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")
@@ -542,6 +568,27 @@ def build_time_plan(duration: int, schema_config_snapshot: Any | None = None) ->
)
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"])
@@ -564,11 +611,11 @@ def build_dynamic_schema(video_config: dict[str, Any], schema_config_snapshot: A
}
)
time_plan = build_time_plan(duration, schema_config_snapshot)
schema["动态时间规划"] = time_plan
schema["动态时间规划"] = _time_plan_schema_for_ai_input(time_plan)
# 预览/AI 入参阶段也要把流程数组按秒数切片规则初始化出来。
# 这样管理后台配置了 4 段/5 段规则后，动作流程、镜头流程会和动态时间规划保持相同长度，
# AI 生成时也能明确知道每个时间段需要填哪个流程字段。
# 预览/AI 入参阶段只锁定时间段和字段结构，不把后台阶段/说明/流程说明写死到 schema 值里。
# 阶段、说明、动作内容、动作详解、镜头内容、镜头详解都应由 AI 基于素材分析填写；
# 只有 AI 返回缺失/空/无/阶段说明等占位内容时，归一化阶段才使用后台规则兜底。
for flow_key in ("动作流程", "镜头流程"):
if flow_key not in schema:
continue
@@ -579,6 +626,7 @@ def build_dynamic_schema(video_config: dict[str, Any], schema_config_snapshot: A
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["输出规格限制"] = {
@@ -660,9 +708,14 @@ def build_user_text(
"参考素材": references,
"输出要求": {
"生成类型": infer_generation_type(references),
"必须填充动态时间规划": build_time_plan(duration, schema_config_snapshot),
"必须填充动作流程": "动作流程时间段必须覆盖完整视频时长",
"必须填充镜头流程": "镜头流程时间段必须覆盖完整视频时长",
"动态时间规划时间段必须严格等于": _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 之外的解释文字", "照抄参考素材品牌水印", "生成违法违规内容", "在最终提示词中写入秒数/比例/分辨率/帧率"],
},
@@ -917,6 +970,7 @@ def _normalize_flow_item(
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": ""}]
@@ -924,15 +978,27 @@ def _normalize_flow_item(
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:
result[field_key] = _merge_flow_content(base_content, extra_texts, fallback)
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:
result[field_key] = _clean_schema_text(item.get(field_key)) or _clean_schema_text(field.get("value")) or "无"
# 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
@@ -941,18 +1007,25 @@ def _normalize_time_plan(plan: list[dict[str, str]], value: Any) -> list[dict[st
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 "无",
"阶段": _clean_schema_text(raw_item.get("阶段")) or plan_item.get("阶段") or "无",
"说明": _clean_schema_text(raw_item.get("说明")) or plan_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]:
plan = build_time_plan(duration, schema_config_snapshot)
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(
@@ -969,7 +1042,7 @@ def ensure_flow_matches_time_plan(result: dict[str, Any], duration: int, schema_
content_keys=_flow_content_aliases(schema_config_snapshot, "镜头流程"),
item_fields=_flow_item_field_rules(schema_config_snapshot, "镜头流程"),
)
result["动态时间规划"] = _normalize_time_plan(plan, result.get("动态时间规划"))
result["动态时间规划"] = plan
return result
@@ -1071,6 +1144,7 @@ def _align_flow_time_ranges(
*,
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)
@@ -1084,6 +1158,7 @@ def _align_flow_time_ranges(
content_key=default_content_key,
content_keys=aliases,
item_fields=item_fields,
fallback_to_plan=fallback_to_plan,
)
)
return aligned
@@ -1178,6 +1253,26 @@ def _merge_flow_patch(base_flow: Any, patch_flow: Any, flow_key: str, schema_con
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"])
@@ -1204,32 +1299,39 @@ def apply_locked_video_schema_fields(schema: dict[str, Any], video_config: dict[
}
)
schema["动态时间规划"] = plan
schema["动态时间规划"] = _normalize_time_plan(plan, schema.get("动态时间规划"))
schema["输出规格限制"] = dynamic_schema.get("输出规格限制", {})
schema["动作流程"] = _align_flow_time_ranges(
schema.get("动作流程"),
plan,
_flow_content_key(schema_config_snapshot, "动作流程"),
content_keys=_flow_content_aliases(schema_config_snapshot, "动作流程"),
item_fields=_flow_item_field_rules(schema_config_snapshot, "动作流程"),
)
schema["镜头流程"] = _align_flow_time_ranges(
schema.get("镜头流程"),
plan,
_flow_content_key(schema_config_snapshot, "镜头流程"),
content_keys=_flow_content_aliases(schema_config_snapshot, "镜头流程"),
item_fields=_flow_item_field_rules(schema_config_snapshot, "镜头流程"),
)
# 合规和质量控制不能被前端降低；AI 返回缺失时使用服务端默认结构补齐。
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 clean_final_prompt_specs(schema, video_config)
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]: