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

2026-06-22 14:39:00 +08:00
parent f62dd8bf67
commit 73567b4143
16 changed files with 1983 additions and 92 deletions
@@ -10,6 +10,15 @@ 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
@@ -157,19 +166,362 @@ def _bounds_to_plan(bounds: list[int], stages: list[tuple[str, str]]) -> list[di
    return plan


-def build_time_plan(duration: int) -> list[dict[str, str]]:
-    duration = max(1, int(duration))
-    if duration <= 5:
-        return _bounds_to_plan(
-            _build_scaled_bounds(duration, [0.2, 0.4, 0.4]),
-            [
-                ("开场吸引", "快速建立主体、产品和画面风格"),
-                ("核心展示", "展示主体动作、核心卖点或主要视觉内容"),
-                ("行动引导", "强化记忆点并给出转化引导"),
+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")
+        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),
+            "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
+                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),
+                        "value": copy.deepcopy(child.get("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 duration <= 8:
-        return _bounds_to_plan(
+    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]),
            [
                ("开场吸引", "快速吸引注意力"),
@@ -177,28 +529,32 @@ def build_time_plan(duration: int) -> list[dict[str, str]]:
                ("核心卖点", "突出新项目核心内容点"),
                ("收尾引导", "给出行动引导并稳定落版"),
            ],
+        ) if duration <= 8 else _bounds_to_plan(
+            _build_scaled_bounds(duration, [0.13, 0.2, 0.27, 0.25, 0.15]),
+            [
+                ("爆款开头", "复刻参考素材开头节奏和视觉吸引点"),
+                ("主体建立", "明确新项目主体和产品信息"),
+                ("卖点放大", "围绕核心内容点展开动作和镜头"),
+                ("情绪推进", "用动作、字幕或镜头变化强化记忆"),
+                ("转化收尾", "给出清晰行动引导"),
+            ],
        )
-    return _bounds_to_plan(
-        _build_scaled_bounds(duration, [0.13, 0.2, 0.27, 0.25, 0.15]),
-        [
-            ("爆款开头", "复刻参考素材开头节奏和视觉吸引点"),
-            ("主体建立", "明确新项目主体和产品信息"),
-            ("卖点放大", "围绕核心内容点展开动作和镜头"),
-            ("情绪推进", "用动作、字幕或镜头变化强化记忆"),
-            ("转化收尾", "给出清晰行动引导"),
-        ],
    )


-def build_dynamic_schema(video_config: dict[str, Any]) -> dict[str, Any]:
-    schema = copy.deepcopy(CLIENT_SCHEMA_V1)
+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)

-    schema["画面属性"].update(
+    frame = schema.setdefault("画面属性", {})
+    if not isinstance(frame, dict):
+        frame = {}
+        schema["画面属性"] = frame
+    frame.update(
        {
            "视频时长": f"{duration}秒",
            "视频比例": video_ratio,
@@ -207,7 +563,24 @@ def build_dynamic_schema(video_config: dict[str, Any]) -> dict[str, Any]:
            "帧率": frame_rate,
        }
    )
-    schema["动态时间规划"] = build_time_plan(duration)
+    time_plan = build_time_plan(duration, schema_config_snapshot)
+    schema["动态时间规划"] = time_plan
+
+    # 预览/AI 入参阶段也要把流程数组按秒数切片规则初始化出来。
+    # 这样管理后台配置了 4 段/5 段规则后，动作流程、镜头流程会和动态时间规划保持相同长度，
+    # 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),
+        )
+
    schema["输出规格限制"] = {
        "支持时长": _safe_list(video_config.get("supported_durations")),
        "支持比例": _safe_list(video_config.get("supported_ratios")),
@@ -255,6 +628,7 @@ def build_user_text(
    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"])
@@ -286,7 +660,7 @@ def build_user_text(
            "参考素材": references,
            "输出要求": {
                "生成类型": infer_generation_type(references),
-                "必须填充动态时间规划": build_time_plan(duration),
+                "必须填充动态时间规划": build_time_plan(duration, schema_config_snapshot),
                "必须填充动作流程": "动作流程时间段必须覆盖完整视频时长",
                "必须填充镜头流程": "镜头流程时间段必须覆盖完整视频时长",
                "最终提示词限制": "最终提示词下所有字段都不能写入视频时长、秒数、视频比例、清晰度、分辨率、帧率、推荐像素、竖屏、横屏等视频规格参数，这些规格只能写在画面属性/动态时间规划/输出规格限制。",
@@ -379,42 +753,119 @@ def _is_empty_schema_value(value: Any) -> bool:
    return _clean_schema_text(value) in EMPTY_VALUE_TEXTS


-def _normalize_schema_object(section_key: str, value: Any) -> dict[str, Any]:
-    default_value = CLIENT_SCHEMA_V1.get(section_key)
+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.get(section_key, set(default_value.keys()))
+    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]) -> dict[str, Any]:
-    """按客户端视频提词 schema 白名单清洗顶层和普通对象字段。
-
-    AI 偶尔会把解释性文本作为 JSON key 输出。普通对象中的非预设字段直接过滤，
-    动作流程/镜头流程这类列表字段在后续 flow normalize 中单独处理。
-    """
+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 OBJECT_FIELD_WHITELISTS:
-            normalized[key] = _normalize_schema_object(key, source.get(key))
+    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 CLIENT_SCHEMA_V1:
-            normalized[key] = copy.deepcopy(CLIENT_SCHEMA_V1[key])
-    for key, default_value in CLIENT_SCHEMA_V1.items():
+        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]) -> dict[str, Any]:
-    return normalize_top_level_schema_fields(result)
+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:
@@ -465,16 +916,24 @@ def _normalize_flow_item(
    plan_item: dict[str, str],
    content_key: str,
    content_keys: tuple[str, ...],
+    item_fields: list[dict[str, Any]] | None = None,
 ) -> dict[str, str]:
    item = raw_item if isinstance(raw_item, dict) else {}
-    allowed_keys = {"时间段", content_key}
+    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 "无"
-    return {
-        "时间段": plan_item.get("时间段") or _clean_schema_text(item.get("时间段")) or "无",
-        content_key: _merge_flow_content(base_content, extra_texts, fallback),
-    }
+    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)
+        else:
+            result[field_key] = _clean_schema_text(item.get(field_key)) or _clean_schema_text(field.get("value")) or "无"
+    return result


 def _normalize_time_plan(plan: list[dict[str, str]], value: Any) -> list[dict[str, str]]:
@@ -492,10 +951,24 @@ def _normalize_time_plan(plan: list[dict[str, str]], value: Any) -> list[dict[st
    return normalized


-def ensure_flow_matches_time_plan(result: dict[str, Any], duration: int) -> dict[str, Any]:
-    plan = build_time_plan(duration)
-    result["动作流程"] = _align_flow_time_ranges(result.get("动作流程"), plan, "动作内容")
-    result["镜头流程"] = _align_flow_time_ranges(result.get("镜头流程"), plan, "镜头内容")
+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)
+    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["动态时间规划"] = _normalize_time_plan(plan, result.get("动态时间规划"))
    return result

@@ -591,9 +1064,16 @@ def clean_final_prompt_specs(schema: dict[str, Any], video_config: dict[str, Any
    return schema


-def _align_flow_time_ranges(flow: Any, plan: list[dict[str, str]], default_content_key: str) -> list[dict[str, str]]:
+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,
+) -> list[dict[str, str]]:
    source = flow if isinstance(flow, list) else []
-    content_keys = ACTION_FLOW_CONTENT_KEYS if default_content_key == "动作内容" else CAMERA_FLOW_CONTENT_KEYS
+    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 {}
@@ -602,43 +1082,110 @@ def _align_flow_time_ranges(flow: Any, plan: list[dict[str, str]], default_conte
                raw_item,
                plan_item=plan_item,
                content_key=default_content_key,
-                content_keys=content_keys,
+                content_keys=aliases,
+                item_fields=item_fields,
            )
        )
    return aligned


-def _merge_editable_dict_fields(base: dict[str, Any], patch: dict[str, Any], allowed_keys: set[str]) -> None:
+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 in patch:
-            base[key] = fill_none_with_wu(patch.get(key))
+        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) -> list[dict[str, Any]]:
+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 == "时间段":
+            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 apply_locked_video_schema_fields(schema: dict[str, Any], video_config: dict[str, Any]) -> dict[str, Any]:
+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)
-    plan = build_time_plan(duration)
+    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
@@ -660,25 +1207,37 @@ def apply_locked_video_schema_fields(schema: dict[str, Any], video_config: dict[
    schema["动态时间规划"] = plan
    schema["输出规格限制"] = dynamic_schema.get("输出规格限制", {})

-    schema["动作流程"] = _align_flow_time_ranges(schema.get("动作流程"), plan, "动作内容")
-    schema["镜头流程"] = _align_flow_time_ranges(schema.get("镜头流程"), plan, "镜头内容")
+    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 = copy.deepcopy(CLIENT_SCHEMA_V1)
-    if not isinstance(schema.get("合规控制"), dict):
-        schema["合规控制"] = default_schema["合规控制"]
-    if not isinstance(schema.get("质量控制"), dict):
-        schema["质量控制"] = default_schema["质量控制"]
+    default_schema = build_client_schema_from_config(schema_config_snapshot)
+    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)


-def normalize_video_prompt_schema_from_ai(result: dict[str, Any], video_config: dict[str, Any]) -> dict[str, Any]:
+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 {}))
-    normalized = ensure_flow_matches_time_plan(normalized, 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)
+    return apply_locked_video_schema_fields(normalized, video_config, schema_config_snapshot)


 def patch_video_prompt_schema_from_client(
@@ -686,38 +1245,50 @@ 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 {})))
+    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 key in ("基础分类", "素材理解", "业务属性", "字幕与口播", "音频与节奏"):
-        if isinstance(base.get(key), dict) and isinstance(patch.get(key), dict):
-            base[key].update(fill_none_with_wu(patch[key]))
+    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("动作流程"))
+        base["动作流程"] = _merge_flow_patch(base.get("动作流程"), patch.get("动作流程"), "动作流程", schema_config_snapshot)
    if isinstance(patch.get("镜头流程"), list):
-        base["镜头流程"] = _merge_flow_patch(base.get("镜头流程"), patch.get("镜头流程"))
+        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 in patch["最终提示词"]:
-                base["最终提示词"][key] = fill_none_with_wu(patch["最终提示词"].get(key))
+            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)
+    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]:
@@ -758,13 +1329,14 @@ async def optimize_hot_opening_video_prompt(
    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)
+    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:
@@ -774,7 +1346,7 @@ async def optimize_hot_opening_video_prompt(

    config = await _select_model_config(db)
    if not config:
-        result = normalize_video_prompt_schema_from_ai(_mock_result(video_config, target_platform), video_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(
@@ -785,6 +1357,7 @@ async def optimize_hot_opening_video_prompt(
        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 = {
@@ -826,7 +1399,7 @@ async def optimize_hot_opening_video_prompt(
    await db.flush()

    result = parse_model_json(content)
-    result = normalize_video_prompt_schema_from_ai(result, video_config)
+    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: