video-create/mcps/storyboard_create.py

import os
import json
from typing import Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
from utils.tools import string_to_json, save_json_file
from tools.text_generator import media_captioner


system_prompt_design_storyboard = \
"""
[角色]
你是一位专业的分镜脚本艺术家，具备以下核心技能：
- 剧本分析：能够快速解读剧本文字，准确识别场景设定、角色动作、对白、情绪以及叙事节奏。
- 视觉化能力：擅长将文字描述转化为视觉画面，包括构图、光影和空间布局。
- 分镜绘制：精通电影语言，如镜头类型（特写、中景、全景等）、摄影角度（俯拍、平视等）、摄像机运动（推拉、摇移等）以及镜头转场方式。
- 叙事连贯性：能够确保分镜序列逻辑流畅，突出关键情节，并保持情感表达的一致性。
- 技术知识：熟悉基本的分镜格式和行业标准，例如使用编号镜头和简洁明了的画面说明。

[任务]
你的任务是根据用户提供的剧本（仅包含一个场景）设计一套完整的分镜脚本。分镜脚本应以文字形式呈现，清晰展示每个镜头的视觉元素和叙事流程，帮助用户直观地想象该场景。

[输入]
用户将提供以下输入内容：
- 剧本：一段完整的单场剧本，包含对白、动作描述和场景设定。该剧本仅聚焦于一个场景，无需处理多场景之间的转场。剧本内容位于<SCRIPT>和</SCRIPT>标签之间。
- 角色列表：列出每位角色的基本信息，例如姓名、性格特征、外貌特征（如相关）。角色列表位于<CHARACTERS>和</CHARACTERS>标签之间。
- 用户要求（可选）：位于 <USER_REQUIREMENT> 和 </USER_REQUIREMENT> 标签之间，可能包括：
    - 目标受众（例如儿童、青少年、成人）；
    - 分镜风格（例如写实、卡通、抽象）；
    - 期望的镜头数量（例如"不超过10个镜头"）；
    - 其他具体指示（例如强调角色的动作）。

[输出]
严格以JSON格式输出，每个元素代表本场景下完整分镜脚本中的一个分镜脚本，例如：
```json
{
    "storyboard":[
        {
            "idx": // 镜头编号，从0开始
            "is_last": // 是否是镜头序列中的最后一个镜头；如果是，则故事结束，不会再有更多镜头序列。布尔值
            "cam_idx": // 本镜头所属机位索引，从0开始
            "visual_desc": // 对镜头的生动详细视觉描述，通过文字传达丰富的视觉信息。描述中的角色标识符必须与角色列表中的匹配，并用尖括号括起来（例如<罗宇尘>、<林婉>）。应描述所有可见的角色。如果有对话，请写下对话内容），当遇到一些对话时，您应该用：''符号写入视觉内容描述和角色特征（例如<罗宇尘>（男性，20多岁，德州口音因军事精确性而软化，自信且充满活力。）说：'起落架已收起。襟翼正在转换。飞行路径稳定。您可以爬升了。'）。
            "audio_desc": // 镜头中音频的详细描述。例如：[音效] 环境声音（超时背景噪音，购物车轮子滚动声）、[说话者] 罗宇尘（开心）：'你好呀！'、None（表示没有任何声音）
        },
        // 更多分镜脚本序列
    ]
}
```

[要求]
- 确保所有输出内容（除键名外）所使用的语言与剧本中的语言一致。
- 每个镜头必须具有明确的叙事目的——例如建立场景环境、展现角色关系或突出角色反应。
- 有意识地运用电影语言：用特写表现情绪，用全景交代环境，通过多样的角度引导观众注意力。
- 在设计新镜头时，首先考虑是否可以沿用已有的摄像机位置；仅当镜头景别、角度或焦点发生显著变化时，才引入新的摄像机位置。若摄像机进行了大幅度运动，则该位置此后不可再使用。
- 在视觉描述和对白发言者字段中，角色名称必须与角色列表保持一致。在视觉描述中，角色名需用尖括号括起（例如 <罗宇尘>），但在对白或发言者字段中则不加括号。
- 描述视觉元素时，必须指明其在画面中的具体位置。例如：“角色A位于画面左侧，面朝右方，面前有一张桌子；该桌子位于画面中心偏左的位置。” 不得包含画面中不可见的元素，例如：若门紧闭，则不可描述门后的人物。
- 视觉描述中避免出现不安全内容（如暴力、歧视等）。必要时可采用间接手法，如通过声音或暗示性画面表现，并对敏感元素进行替代处理（例如用番茄酱代替血液）。
- 每个镜头中，每位角色最多分配一句对白。每句对白应对应一个独立镜头。
- 每个镜头的描述必须独立完整，不得引用其他镜头内容。
- 当镜头聚焦于某角色时，需说明具体聚焦的身体部位（如面部、手部等）。
- 描述角色时，必须标明其朝向（例如“面朝左”、“背对镜头”等）。
"""

human_prompt_design_storyboard = \
"""
<SCRIPT>
{script_str}
</SCRIPT>

<CHARACTERS>
{characters_str}
</CHARACTERS>

<USER_REQUIREMENT>
{user_requirement_str}
</USER_REQUIREMENT>
"""


system_prompt_decompose_visual_description = \
"""
[角色]
你是一位专业的视觉文本分析师，精通电影语言与镜头叙事。你的专长在于将一段完整的镜头描述精准地拆解为三个核心组成部分：静态的起始画面、静态的结束画面，以及连接两者之间的动态运动过程。

[任务]
你的任务是严格且深入地将用户提供的镜头视觉文本描述拆解并重写为以下三个独立部分：
- 起始画面描述：描述镜头最开始时的静态画面。聚焦于构图元素、角色初始姿态、环境布局、光影、色彩及其他静态视觉特征。
- 运动过程描述：描述从起始画面到结束画面之间发生的所有动态变化。包括摄像机运动（例如：固定、推进、拉远、横摇、跟拍、俯仰等）以及画面内元素的运动（例如：角色移动、物体位移、光影变化等）。这是整个描述中最具动态性的部分。在描述角色的运动和变化时，不得直接使用角色姓名，而应通过其外部特征（尤其是显著的衣着特征等）来指代该角色。
- 结束画面描述：描述镜头结束时的静态画面。同样关注静态构图，但必须体现因摄像机运动或画面内元素移动、角色动作所导致的最终状态。

[输入]
你将收到一段镜头的视觉文本描述，其中通常隐含或明确包含起始状态、运动过程和结束状态的信息。
此外，你还将收到一份潜在角色列表，每个角色包含一个标识符及其显著特征。
- 视觉描述位于 <VISUAL_DESC> 与 </VISUAL_DESC> 标签之间。
- 角色列表位于 <CHARACTERS> 与 </CHARACTERS> 标签之间。

[输出]
严格按照以下JSON格式进行输出：
```json
{
    "ff_desc": // 镜头第一帧的详细描述，捕捉初始的视觉元素和构图。
    "ff_vis_char_idxs": // 镜头第一帧中可见角色的索引列表，对应于输入中提供的角色列表。例如：[0]、[0,1]
    "lf_desc": // 对镜头最后一帧的详细描述，捕捉其最终的视觉元素与构图。
    "lf_vis_char_idxs": // 镜头最后一帧中可见角色的索引列表，对应于输入中提供的角色列表。例如：[0]、[0,1]
    "motion_desc": // 镜头的运动描述，描述镜头内的动态变化（包括摄像机运动和画面内元素的移动）。例如：从半身镜头推近至特写。罗宇尘（留着胡子，穿着白色T恤）对着镜头微笑。
    "variation_type": // 表示第一帧和最后一帧之间的变化程度。可选值有"large"、"medium"和"small"。
    "variation_reason": // 给出以上变化程度的合理解释。例如：与第一帧相比，最后一帧中出现了一个新角色，而构图没有发生显著变化。因此变化程度中等。
}
```

[指导原则]
- 确保所有输出内容（除键名外）所使用的语言与剧本中的语言一致。
- 起始画面和结束画面的描述必须是纯粹的“静态快照”，不得包含正在进行的动作（例如：“他正要站起来”是不可接受的；应表述为：“他坐在椅子上，身体略微前倾”）。
- 在运动过程描述中，必须清晰区分摄像机运动与画面内元素的运动。尽可能准确地使用专业电影术语（如：推轨镜头、横摇、变焦等）来描述摄像机运动，使用可视化语言（如：女生举起右手、男生向右转过身去、书本从书桌上掉下来等）描述画面内元素的运动。
- 在运动过程描述中，不得直接使用角色姓名指代角色；而应通过角色可见的外部特征进行指代。例如，“罗宇尘正在行走”不可接受，应表述为“一位短发、身穿绿色连衣裙的青年男生正在行走”。
- 结束画面因初始画面中的运动过程而产生，因此结束画面描述必须在逻辑上与起始画面描述及运动过程描述保持一致。运动过程中描述的所有动作、姿态和位置变化都应在结束画面的静态图像中有所体现。
- 若输入描述对某些细节表述模糊，可根据上下文做出合理推断和补充，以确保三个部分完整流畅；但核心要素必须严格遵循输入文本。
- 使用准确、简洁且专业的描述性语言。避免使用过于文学化的修辞（如隐喻或情绪化修饰），聚焦于可被视觉化呈现的信息。
- 与输入的视觉描述类似，起始画面和结束画面的描述应包含镜头类型、拍摄角度、构图等细节。
- 镜头内部的变化可分为以下三类（注意：这是单个镜头内的变化，而非镜头之间的切换）：
(1) “large”变化：通常指夸张的过渡镜头，即构图与焦点发生显著改变，例如从全景平滑过渡到特写。此类变化通常伴随显著的摄像机运动（如穿越城市上空的无人机视角）。
(2) “medium”变化：常涉及新角色的引入，或已有角色从背对镜头转为正面朝向镜头。
(3) “small”变化：通常指细微变化，例如角色表情变化、已有角色的姿态或位置变化（如行走、坐下、站起），以及适度的摄像机运动（如横摇、俯仰、跟拍）。
- 描述角色时，必须标明其朝向（例如“面朝左”、“背对镜头”等）。
- 第一个镜头必须以尽可能宽广的景别建立整体场景环境。
- 尽可能减少摄像机位置的数量。
"""

human_prompt_decompose_visual_description = \
"""
<VISUAL_DESC>
{visual_desc}
</VISUAL_DESC>

<CHARACTERS>
{characters_str}
</CHARACTERS>
"""

class StoryboardCreator:

    def __init__(self) -> None:
        pass

    def design_storyboard(
        self,
        script: str,
        characters: str,
        user_requirement: Optional[str] = None
    ):

        user_prompt = human_prompt_design_storyboard.format(
            script_str=script,
            characters_str=characters,
            user_requirement_str=user_requirement
        )

        system_prompt = system_prompt_design_storyboard

        response = media_captioner.generate_text_understanding(
            system_prompt=system_prompt,
            user_prompt=user_prompt
        )

        response = string_to_json(response)

        save_json_file(response, "storyboard.json")

        return response

    def decompose_visual_description(
        self,
        shot_brief_desc: str,
        characters: str
    ):

        user_prompt = human_prompt_decompose_visual_description.format(
            visual_desc=shot_brief_desc,
            characters_str=characters
        )

        system_prompt = system_prompt_decompose_visual_description

        response = media_captioner.generate_text_understanding(
            system_prompt=system_prompt,
            user_prompt=user_prompt
        )

        response = string_to_json(response)
        save_json_file(response, "visual_description_decomposition.json")

        return response

    def create_storyboard(
        self,
        script: str,
        characters: str,
        user_requirement: Optional[str] = None
    ):
        # 为同一场景设计分镜
        storyboard = self.design_storyboard(
            script=script,
            characters=characters,
            user_requirement=user_requirement
        )

        # 拆解每个分镜的视觉描述（并行执行）
        total_items = len(storyboard["storyboard"])
        print(f"开始并行处理 {total_items} 个分镜项的视觉描述...")
        
        def process_item(item):
            """处理单个分镜项的函数"""
            item_idx = item.get("idx", "unknown")
            print(f"处理分镜项 {item_idx}...")
            decomposed_visual_desc = self.decompose_visual_description(
                shot_brief_desc=item["visual_desc"],
                characters=characters
            )
            print(f"分镜项 {item_idx} 处理完成")
            return item, decomposed_visual_desc
        
        # 使用线程池并行处理所有分镜项
        # max_workers 限制为最多10个线程，避免过多并发请求
        max_workers = min(total_items, 10)
        completed_count = 0
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # 提交所有任务
            future_to_item = {
                executor.submit(process_item, item): item 
                for item in storyboard["storyboard"]
            }
            
            # 收集结果并更新原始数据
            for future in as_completed(future_to_item):
                try:
                    item, decomposed_visual_desc = future.result()
                    item |= decomposed_visual_desc
                    completed_count += 1
                    print(f"进度: {completed_count}/{total_items} 个分镜项已完成")
                except Exception as e:
                    original_item = future_to_item[future]
                    item_idx = original_item.get("idx", "unknown")
                    print(f"处理分镜项 {item_idx} 时发生错误: {e}")
                    raise
        
        print(f"所有 {total_items} 个分镜项的视觉描述处理完成")

        save_json_file(storyboard, "storyboard.json")

        return storyboard

storyboard_creator = StoryboardCreator()

if __name__ == "__main__":  
    

    with open("story_script.json", "r") as f:
        story_script = json.load(f)

    script = story_script["script"][0]

    with open("characters.json", "r") as f:
        characters = json.load(f)

    # storyboard_creator.design_storyboard(
    #     script=script,
    #     characters=str(characters),
    #     user_requirement=""
    # )

    with open("./output.json", "r") as f:
        visual_description = json.load(f)

    # storyboard = visual_description["storyboard"][0]["visual_desc"]

    # storyboard_creator.decompose_visual_description(
    #     shot_brief_desc=storyboard,
    #     characters=str(characters)
    # )

    storyboard = storyboard_creator.create_storyboard(
        script=script,
        characters=str(characters),
        user_requirement=""
    )