feat(脚本): 按流程文档重写脚本阶段(自动提取人物/场景 + 标签胶囊改写 + 基础资产按标签 seed)

后端: - 出稿后新增轻量提取调用,从脚本抽取人物/场景标签及每个标签的建议生图提示词, 写入 project.metadata(cast/scenes/cast_prompts/scene_prompts) - 提取调用接入计费闭环(reserve→charge/release,记为 script_optimization); best-effort:无模型/预扣失败/调用失败/解析失败都吞掉返回空,绝不阻断出稿 - generate_base_asset 增加 label,落 group.metadata,把生成结果归回脚本标签卡前端: - 设定卡风格补齐到 5 个(真实测评/痛点种草/小红书种草/开箱测评/对比展示) - 标签改为待确认胶囊流:增删不立即落库,排进队列在输入框上方显示可撤销胶囊, 点发送才整体重写;待删划线、待加虚线 - 基础资产人物/场景按提取标签逐个 seed 卡,显示 AI 提示词、可改后生成测试:新增提取+计费+release+label 共 4 个用例,全套 24 个通过(sqlite test 设置) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-16 17:59:41 +08:00
parent 76b3ce94f9
commit 366e077c05
8 changed files with 370 additions and 46 deletions
@@ -1,3 +1,4 @@
+import json
 import re
 import subprocess
 import tempfile
@@ -112,6 +113,102 @@ def parse_segment_fields(block: str) -> tuple[str, str]:
    return narration or visual, visual or narration


+def build_cast_scene_extract_prompt(content: str) -> list[dict[str, str]]:
+    """轻量抽取提示词:从镜头脚本里提炼人物 / 场景标签,并给每个标签一句可直接生图的画面提示词。"""
+    system = (
+        "你是短视频脚本分析助手。请从给定的镜头脚本中提取出现的『人物』和『场景』,"
+        "并为每个人物 / 场景写一句可直接用于文生图的画面提示词(中文,30 字内,描述外形 / 着装 / 环境 / 光线,9:16 竖屏)。"
+        "人物指出镜的角色(例:女主、同事、闺蜜);场景指画面发生的地点或环境(例:卫生间、地铁、办公室)。"
+        "去重,人物与场景各最多 6 个。只输出一个 JSON 对象,不要 markdown 代码块,不要任何额外文字,格式如下：\n"
+        '{"cast":[{"name":"女主","prompt":"26岁都市女性,自然妆容,米色针织衫,柔和室内光,9:16竖屏"}],'
+        '"scenes":[{"name":"卫生间","prompt":"现代简约浴室,暖色灯光,干净台面,9:16竖屏"}]}'
+    )
+    return [{"role": "system", "content": system}, {"role": "user", "content": f"镜头脚本如下：\n{content}".strip()}]
+
+
+def _coerce_tag_entries(items: object, limit: int = 6) -> tuple[list[str], dict[str, str]]:
+    """把模型回的 [{"name","prompt"}] 列表整理成 (标签列表, {标签: 提示词}),去重保序、容错。"""
+    names: list[str] = []
+    prompts: dict[str, str] = {}
+    if not isinstance(items, list):
+        return names, prompts
+    for item in items:
+        if isinstance(item, dict):
+            name = str(item.get("name") or "").strip()
+            prompt = str(item.get("prompt") or "").strip()
+        else:
+            name, prompt = str(item or "").strip(), ""
+        if not name or name in ("无", "暂无", "未提及") or name in names:
+            continue
+        names.append(name)
+        if prompt:
+            prompts[name] = prompt
+        if len(names) >= limit:
+            break
+    return names, prompts
+
+
+def extract_cast_and_scenes(*, project, user, content: str) -> dict:
+    """轻量调一次文本模型,从脚本里抽取人物 / 场景标签及每个标签的建议生图提示词。
+
+    出稿后的增益步骤(对齐流程文档「自动从脚本提取信息」),走完整的 AITask + 计费闭环
+    (reserve→charge/release,任务类型记为 script_optimization)。但全程 best-effort:
+    无可用模型 / 预扣失败 / 调用失败 / 解析失败都吞掉返回空,绝不阻断脚本生成主流程。
+    返回 {cast, scenes, cast_prompts, scene_prompts}。
+    """
+    empty = {"cast": [], "scenes": [], "cast_prompts": {}, "scene_prompts": {}}
+    model_config = get_default_model(ModelConfig.Capability.TEXT)
+    if model_config is None or not (content or "").strip():
+        return empty
+
+    messages = build_cast_scene_extract_prompt(content)
+    try:
+        task = create_ai_task(
+            project=project,
+            user=user,
+            task_type=AITask.Type.SCRIPT_OPTIMIZATION,
+            model_config=model_config,
+            request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "messages": messages},
+        )
+    except Exception:
+        return empty  # 余额不足等预扣失败:跳过提取,不挡出稿
+    reservation = task.credit_reservation
+
+    try:
+        task.status = AITask.Status.SUBMITTED
+        task.submitted_at = timezone.now()
+        task.save(update_fields=["status", "submitted_at", "updated_at"])
+
+        provider = VolcanoArkProvider(base_url=model_config.provider.base_url or None)
+        response = provider.chat_completion(model=model_config.name, endpoint=model_config.endpoint, messages=messages)
+        text = provider.extract_text(response)
+
+        # LLM 调用已真实消耗 token → 按成功计费(无论后续能否解析出标签)
+        with transaction.atomic():
+            task.status = AITask.Status.SUCCEEDED
+            task.response_payload = response
+            task.actual_cost = task.estimated_cost
+            task.completed_at = timezone.now()
+            task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
+            charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
+
+        match = re.search(r"\{.*\}", text, re.DOTALL)  # 容忍模型多裹了 markdown / 解释文字
+        if not match:
+            return empty
+        data = json.loads(match.group(0))
+        cast, cast_prompts = _coerce_tag_entries(data.get("cast"))
+        scenes, scene_prompts = _coerce_tag_entries(data.get("scenes"))
+        return {"cast": cast, "scenes": scenes, "cast_prompts": cast_prompts, "scene_prompts": scene_prompts}
+    except Exception as exc:
+        with transaction.atomic():
+            task.status = AITask.Status.FAILED
+            task.error_message = str(exc)
+            task.completed_at = timezone.now()
+            task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
+            release_credit(reservation=reservation, reason=str(exc))
+        return empty
+
+
 def split_script_into_segments(content: str, count: int = 4) -> list[str]:
    """把一段脚本稳健地拆成 `count` 个分镜文本,保证每镜都非空、且所有内容都被分配到某一镜。

@@ -196,6 +293,9 @@ def generate_project_script(*, project, user, user_prompt: str, selling_point_id
        response = provider.chat_completion(model=model_config.name, endpoint=model_config.endpoint, messages=messages)
        content = provider.extract_text(response)

+        # 出稿后自动提取人物 / 场景(best-effort,失败返回空,不挡主流程),供脚本页标签与基础资产 seed
+        extracted = extract_cast_and_scenes(project=project, user=user, content=content)
+
        with transaction.atomic():
            task.status = AITask.Status.SUCCEEDED
            task.response_payload = response
@@ -222,6 +322,20 @@ def generate_project_script(*, project, user, user_prompt: str, selling_point_id
                    visual_prompt=visual,
                )

+            # 把提取到的人物 / 场景(含每个标签的建议生图提示词)回填进 project.metadata:
+            # 脚本页标签自动读 cast/scenes,基础资产据 cast_prompts/scene_prompts seed 每张卡。
+            # 仅在提取到内容时覆盖,空结果不清掉用户已有标签。
+            if extracted["cast"] or extracted["scenes"]:
+                metadata = dict(project.metadata or {})
+                if extracted["cast"]:
+                    metadata["cast"] = extracted["cast"]
+                    metadata["cast_prompts"] = extracted["cast_prompts"]
+                if extracted["scenes"]:
+                    metadata["scenes"] = extracted["scenes"]
+                    metadata["scene_prompts"] = extracted["scene_prompts"]
+                project.metadata = metadata
+                project.save(update_fields=["metadata", "updated_at"])
+
            stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.SCRIPT)
            stage.status = ProjectStage.Status.NEEDS_REVIEW
            stage.save(update_fields=["status", "updated_at"])
@@ -412,7 +526,7 @@ def _store_generated_media(*, team, user, project, task, media: str, name: str,
    return asset


-def generate_base_asset(*, project, user, kind: str, prompt: str) -> BaseAssetGroup:
+def generate_base_asset(*, project, user, kind: str, prompt: str, label: str = "") -> BaseAssetGroup:
    model_config = get_default_model(ModelConfig.Capability.IMAGE)
    if model_config is None:
        raise ValueError("no active image model configured")
@@ -455,7 +569,9 @@ def generate_base_asset(*, project, user, kind: str, prompt: str) -> BaseAssetGr
                category=category,
                asset_type=Asset.Type.IMAGE,
            )
-            group = BaseAssetGroup.objects.create(project=project, kind=kind, task=task, prompt=prompt)
+            # label = 该资产对应的脚本提取标签(人物/场景名),用于把生成结果归回对应的标签卡
+            group_meta = {"label": label.strip()} if label and label.strip() else {}
+            group = BaseAssetGroup.objects.create(project=project, kind=kind, task=task, prompt=prompt, metadata=group_meta)
            group.candidate_assets.add(asset)
            group.adopted_asset = asset
            group.save(update_fields=["adopted_asset", "updated_at"])