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.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 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 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]), [ ("开场吸引", "快速建立主体、产品和画面风格"), ("核心展示", "展示主体动作、核心卖点或主要视觉内容"), ("行动引导", "强化记忆点并给出转化引导"), ], ) if duration <= 8: return _bounds_to_plan( _build_scaled_bounds(duration, [0.15, 0.3, 0.35, 0.2]), [ ("开场吸引", "快速吸引注意力"), ("主体展示", "展示主体和产品关系"), ("核心卖点", "突出新项目核心内容点"), ("收尾引导", "给出行动引导并稳定落版"), ], ) 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) 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( { "视频时长": f"{duration}秒", "视频比例": video_ratio, "清晰度": resolution, "推荐分辨率": recommended_resolution, "帧率": frame_rate, } ) schema["动态时间规划"] = build_time_plan(duration) 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], ) -> 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), "必须填充动态时间规划": build_time_plan(duration), "必须填充动作流程": "动作流程时间段必须覆盖完整视频时长", "必须填充镜头流程": "镜头流程时间段必须覆盖完整视频时长", "最终提示词限制": "最终提示词下所有字段都不能写入视频时长、秒数、视频比例、清晰度、分辨率、帧率、推荐像素、竖屏、横屏等视频规格参数，这些规格只能写在画面属性/动态时间规划/输出规格限制。", "禁止": ["输出 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 def ensure_top_keys(result: dict[str, Any]) -> dict[str, Any]: schema = copy.deepcopy(CLIENT_SCHEMA_V1) for key, default_value in schema.items(): if key not in result: result[key] = default_value elif isinstance(default_value, dict) and isinstance(result.get(key), dict): merged = copy.deepcopy(default_value) merged.update(result[key]) result[key] = merged return result def ensure_flow_matches_time_plan(result: dict[str, Any], duration: int) -> dict[str, Any]: plan = build_time_plan(duration) if not isinstance(result.get("动作流程"), list) or not result["动作流程"]: result["动作流程"] = [ {"时间段": item["时间段"], "动作内容": item["说明"]} for item in plan ] if not isinstance(result.get("镜头流程"), list) or not result["镜头流程"]: result["镜头流程"] = [ {"时间段": item["时间段"], "镜头内容": item["说明"]} for item in plan ] result["动作流程"] = _align_flow_time_ranges(result["动作流程"], plan, "动作内容") result["镜头流程"] = _align_flow_time_ranges(result["镜头流程"], plan, "镜头内容") 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) -> list[dict[str, Any]]: source = flow if isinstance(flow, list) else [] aligned: list[dict[str, Any]] = [] for index, plan_item in enumerate(plan): old_item = source[index] if index < len(source) and isinstance(source[index], dict) else {} item = dict(old_item) item["时间段"] = plan_item["时间段"] if not any(k in item and str(item.get(k)).strip() for k in (default_content_key, "动作", "镜头", "说明", "内容")): item[default_content_key] = plan_item["说明"] aligned.append(item) return aligned def _merge_editable_dict_fields(base: dict[str, Any], patch: dict[str, Any], allowed_keys: set[str]) -> None: for key in allowed_keys: if key in patch: base[key] = fill_none_with_wu(patch.get(key)) def _merge_flow_patch(base_flow: Any, patch_flow: Any) -> 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]] = [] 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 == "时间段": continue 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]: 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) 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["动态时间规划"] = plan schema["输出规格限制"] = dynamic_schema.get("输出规格限制", {}) schema["动作流程"] = _align_flow_time_ranges(schema.get("动作流程"), plan, "动作内容") schema["镜头流程"] = _align_flow_time_ranges(schema.get("镜头流程"), plan, "镜头内容") # 合规和质量控制不能被前端降低；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["质量控制"] 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]: 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_negative_prompt(normalized) return apply_locked_video_schema_fields(normalized, video_config) def patch_video_prompt_schema_from_client( *, server_schema: dict[str, Any], client_schema: dict[str, Any], video_config: dict[str, Any], ) -> dict[str, Any]: """以前端 JSON 作为 patch，回填到服务端已有 schema。 禁止整包覆盖：数组长度、时间段、视频规格、输出规格、质量控制、合规控制、schema 协议字段均以服务端为准。 """ base = ensure_top_keys(fill_none_with_wu(copy.deepcopy(server_schema if isinstance(server_schema, dict) else {}))) patch = client_schema if isinstance(client_schema, dict) else {} for key in ("基础分类", "素材理解", "业务属性", "字幕与口播", "音频与节奏"): if isinstance(base.get(key), dict) and isinstance(patch.get(key), dict): base[key].update(fill_none_with_wu(patch[key])) if isinstance(base.get("画面属性"), dict) and isinstance(patch.get("画面属性"), dict): _merge_editable_dict_fields( base["画面属性"], patch["画面属性"], {"主体描述", "主体数量", "主体位置", "主体占比", "场景描述", "构图方式", "画面风格", "光影色彩"}, ) if isinstance(patch.get("动作流程"), list): base["动作流程"] = _merge_flow_patch(base.get("动作流程"), patch.get("动作流程")) if isinstance(patch.get("镜头流程"), list): base["镜头流程"] = _merge_flow_patch(base.get("镜头流程"), patch.get("镜头流程")) # 动态时间规划保持服务端数组长度和时间段，只允许保留原值；不接受客户端 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)) return apply_locked_video_schema_fields(base, video_config) 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 = "抖音", ) -> tuple[dict[str, Any], str, dict[str, Any]]: duration = int(video_config["duration"]) references = [ {"type": "video", "url": material_video_url}, {"type": "image", "url": _build_file_url_or_data_uri(generated_image_url)}, ] client_schema = build_dynamic_schema(video_config) 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) 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, ) user_message, log_user_message = build_user_message(user_text, references, reference_video_fps) request_data = { "model": config.model_name, "messages": [{"role": "system", "content": build_system_prompt()}, user_message], "max_tokens": 6000, "temperature": 0.15, "response_format": {"type": "json_object"}, } async with httpx.AsyncClient(timeout=int(settings.CHATAPI_REQUEST_TIMEOUT_SECONDS or 180)) as client: response = await client.post( f"{config.api_base.rstrip('/')}/chat/completions", headers={"Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json"}, json=request_data, ) if response.status_code >= 400: raise RuntimeError(f"视频提词优化失败 HTTP {response.status_code}: {response.text}") data = response.json() content = data["choices"][0]["message"]["content"].strip() usage = data.get("usage", {}) or {} token_usage = { "input_tokens": int(usage.get("prompt_tokens") or 0), "output_tokens": int(usage.get("completion_tokens") or 0), "total_tokens": int(usage.get("total_tokens") or 0), "log_user_message": log_user_message, } db.add( TokenUsage( id=generate_id(), model_config_id=config.id, user_id=user_id, input_tokens=token_usage["input_tokens"], output_tokens=token_usage["output_tokens"], total_tokens=token_usage["total_tokens"], ) ) await db.flush() result = parse_model_json(content) result = normalize_video_prompt_schema_from_ai(result, video_config) 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}"