拆镜复刻、爆款开头复刻管理后台完成
This commit is contained in:
@@ -341,33 +341,162 @@ def fill_none_with_wu(value: Any) -> Any:
|
||||
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 _normalize_schema_object(section_key: str, value: Any) -> dict[str, Any]:
|
||||
default_value = CLIENT_SCHEMA_V1.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()))
|
||||
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 中单独处理。
|
||||
"""
|
||||
source = result if isinstance(result, dict) else {}
|
||||
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))
|
||||
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():
|
||||
if key not in normalized:
|
||||
normalized[key] = copy.deepcopy(default_value)
|
||||
return normalized
|
||||
|
||||
|
||||
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
|
||||
return normalize_top_level_schema_fields(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, ...],
|
||||
) -> dict[str, str]:
|
||||
item = raw_item if isinstance(raw_item, dict) else {}
|
||||
allowed_keys = {"时间段", content_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),
|
||||
}
|
||||
|
||||
|
||||
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 {}
|
||||
item: dict[str, str] = {
|
||||
"时间段": plan_item.get("时间段") or _clean_schema_text(raw_item.get("时间段")) or "无",
|
||||
"阶段": _clean_schema_text(raw_item.get("阶段")) or plan_item.get("阶段") or "无",
|
||||
"说明": _clean_schema_text(raw_item.get("说明")) or plan_item.get("说明") or "无",
|
||||
}
|
||||
# 动态时间规划只保留 时间段/阶段/说明，多余字段不入库。
|
||||
normalized.append(item)
|
||||
return normalized
|
||||
|
||||
|
||||
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
|
||||
result["动作流程"] = _align_flow_time_ranges(result.get("动作流程"), plan, "动作内容")
|
||||
result["镜头流程"] = _align_flow_time_ranges(result.get("镜头流程"), plan, "镜头内容")
|
||||
result["动态时间规划"] = _normalize_time_plan(plan, result.get("动态时间规划"))
|
||||
return result
|
||||
|
||||
|
||||
@@ -462,16 +591,20 @@ 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, Any]]:
|
||||
def _align_flow_time_ranges(flow: Any, plan: list[dict[str, str]], default_content_key: str) -> list[dict[str, str]]:
|
||||
source = flow if isinstance(flow, list) else []
|
||||
aligned: list[dict[str, Any]] = []
|
||||
content_keys = 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):
|
||||
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)
|
||||
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=content_keys,
|
||||
)
|
||||
)
|
||||
return aligned
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user