159 lines
5.6 KiB
Python
159 lines
5.6 KiB
Python
"""从已落库脚本本地拆出角色/场景 —— 不调模型、不扣积分。
|
|
|
|
脚本生成时已要求结构化 entities;这里负责:
|
|
1. 读 ScriptVersion.metadata / content 里的 entities
|
|
2. 只保留 character / scene(商品不抽)
|
|
3. 至少保证 1 角色 + 1 场景(缺则按分镜补默认)
|
|
4. 回填 project.metadata + 每镜 entity_refs + entities_extracted
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
|
|
def _as_list(value: Any) -> list:
|
|
return value if isinstance(value, list) else []
|
|
|
|
|
|
def _load_draft_entities(script) -> list[dict]:
|
|
meta = script.metadata if isinstance(script.metadata, dict) else {}
|
|
ents = _as_list(meta.get("entities"))
|
|
if ents:
|
|
return [e for e in ents if isinstance(e, dict)]
|
|
raw = (script.content or "").strip()
|
|
if not raw:
|
|
return []
|
|
try:
|
|
data = json.loads(raw)
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
return []
|
|
if not isinstance(data, dict):
|
|
return []
|
|
return [e for e in _as_list(data.get("entities")) if isinstance(e, dict)]
|
|
|
|
|
|
def _normalize_entities(raw: list[dict]) -> list[dict]:
|
|
out: list[dict] = []
|
|
seen: set[str] = set()
|
|
for i, ent in enumerate(raw):
|
|
etype = str(ent.get("type") or "").strip().lower()
|
|
if etype in {"person", "cast", "role"}:
|
|
etype = "character"
|
|
if etype in {"location", "bg", "background"}:
|
|
etype = "scene"
|
|
if etype not in {"character", "scene"}:
|
|
continue
|
|
eid = str(ent.get("id") or f"{'c' if etype == 'character' else 's'}{i + 1}").strip()
|
|
while eid in seen:
|
|
eid = f"{eid}_{i}"
|
|
seen.add(eid)
|
|
name = str(ent.get("name") or eid).strip() or eid
|
|
visual = str(ent.get("visual_prompt") or ent.get("prompt") or "").strip()
|
|
if not visual:
|
|
visual = (
|
|
f"{name},全身出镜,自然光,9:16竖屏"
|
|
if etype == "character"
|
|
else f"{name},环境空镜,干净构图,16:9横屏"
|
|
)
|
|
out.append(
|
|
{
|
|
"id": eid,
|
|
"type": etype,
|
|
"name": name,
|
|
"visual_prompt": visual,
|
|
"ref_index": len(out) + 1,
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def _default_character(segments) -> dict:
|
|
# 优先用第一镜画面里像「人」的描述,否则通用主角
|
|
hint = ""
|
|
for seg in segments:
|
|
text = f"{getattr(seg, 'visual_prompt', '') or ''} {getattr(seg, 'narration', '') or ''}"
|
|
if re.search(r"女|男|人|主播|模特|客户|宝妈|姐姐|哥哥", text):
|
|
hint = (getattr(seg, "visual_prompt", None) or text).strip()
|
|
break
|
|
visual = hint[:80] if hint else "电商短视频出镜主角,全身,自然妆容,干净背景,9:16竖屏"
|
|
return {
|
|
"id": "c1",
|
|
"type": "character",
|
|
"name": "主角",
|
|
"visual_prompt": visual,
|
|
"ref_index": 1,
|
|
}
|
|
|
|
|
|
def _default_scene(segments) -> dict:
|
|
hint = ""
|
|
for seg in segments:
|
|
text = (getattr(seg, "visual_prompt", None) or "").strip()
|
|
if text:
|
|
hint = text
|
|
break
|
|
visual = hint[:80] if hint else "干净室内场景,柔和光线,适合电商口播,16:9横屏"
|
|
return {
|
|
"id": "s1",
|
|
"type": "scene",
|
|
"name": "主场景",
|
|
"visual_prompt": visual,
|
|
"ref_index": 1,
|
|
}
|
|
|
|
|
|
def _ensure_min_entities(entities: list[dict], segments) -> list[dict]:
|
|
has_char = any(e["type"] == "character" for e in entities)
|
|
has_scene = any(e["type"] == "scene" for e in entities)
|
|
if not has_char:
|
|
entities.append(_default_character(segments))
|
|
if not has_scene:
|
|
entities.append(_default_scene(segments))
|
|
# 重排 ref_index
|
|
for i, e in enumerate(entities):
|
|
e["ref_index"] = i + 1
|
|
return entities
|
|
|
|
|
|
def _backfill_segment_refs(segments, entities: list[dict]) -> None:
|
|
valid = {e["id"] for e in entities}
|
|
char_ids = [e["id"] for e in entities if e["type"] == "character"]
|
|
scene_ids = [e["id"] for e in entities if e["type"] == "scene"]
|
|
default_refs = (char_ids[:1] + scene_ids[:1]) or list(valid)[:2]
|
|
for seg in segments:
|
|
refs = [r for r in (seg.entity_refs or []) if r in valid]
|
|
speaker = (seg.speaker or "").strip()
|
|
if speaker and speaker in valid and speaker not in refs:
|
|
refs.append(speaker)
|
|
if not refs:
|
|
refs = list(default_refs)
|
|
if refs != (seg.entity_refs or []):
|
|
seg.entity_refs = refs
|
|
seg.save(update_fields=["entity_refs", "updated_at"])
|
|
|
|
|
|
def materialize_script_entities(*, project, script) -> list[dict]:
|
|
"""本地拆角色/场景并落库。返回规范化后的 entities 列表。"""
|
|
from django.db import transaction
|
|
|
|
from apps.ai.script_agent import _map_entities_to_project_metadata
|
|
|
|
segments = list(script.segments.order_by("sort_order"))
|
|
entities = _ensure_min_entities(_normalize_entities(_load_draft_entities(script)), segments)
|
|
with transaction.atomic():
|
|
_map_entities_to_project_metadata(project, entities)
|
|
md = dict(project.metadata or {})
|
|
md["entities_extracted"] = True
|
|
md["entities_extract_mode"] = "local"
|
|
project.metadata = md
|
|
project.save(update_fields=["metadata", "updated_at"])
|
|
_backfill_segment_refs(segments, entities)
|
|
# 同步写回脚本 metadata.entities,方便下次读
|
|
sm = dict(script.metadata or {})
|
|
sm["entities"] = entities
|
|
script.metadata = sm
|
|
script.save(update_fields=["metadata", "updated_at"])
|
|
return entities
|