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,
    }
    token_usage_id = generate_id()
    db.add(
        TokenUsage(
            id=token_usage_id,
            model_config_id=config.id,
            user_id=user_id,
            input_tokens=token_usage["input_tokens"],
            output_tokens=token_usage["output_tokens"],
            total_tokens=token_usage["total_tokens"],
        )
    )
    await db.flush()
    token_usage.update({
        "token_usage_id": token_usage_id,
        "model_config_id": config.id,
        "model_config_name": config.name,
        "model_provider": config.provider,
        "model_name": config.model_name,
    })

    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}"