movie-create/examples/refer_video_create/mcps/script_create.py

import os
import asyncio
from typing import Optional, Dict
from taskflow import FileIOHandler
from api_modules.ark_client_async import AsyncArkClient
from api_modules.ark_client import ArkMessage, APIError
from taskflow import get_logger

io_handler = FileIOHandler()
logger = get_logger("examples.refer_video_create.mcps.script_create")

system_prompt_develop_script = \
"""
## 角色
- 你兼具专业的视频分析能力与视频脚本创作能力
- 你坚持“先过拟合，再泛化”的创作理念，确保脚本内容与输入视频内容完全一致

## 任务
- 仔细拆解分析输入视频的完整内容，完全对标视频内容进行视频脚本创作输出，实现对输入视频的完全复刻。

## 输出格式
- 必须按照以下JSON格式进行输出（必须严格遵守JSON格式，不得有任何其他内容）：
```json
{
    "basic_info": {
        "script_theme": // str;脚本主题，明确整体场景基调,
        "total_lenses": // int;总镜头数，方便后期拼接核对,
        "total_duration": // int;总时长:所有lens_details的lens_duration之和,
        "unified_style": // str;全片统一风格/氛围，保证AI生成一致性
    },
    "lens_details": [
        {
            "lens_id": // int;镜头编号
            "lens_params": // str;核心镜头参数：景别（远景/全景/中景/近景/特写等景别）+ 视角（俯拍/仰拍/过肩/平视/微距/航拍/顶拍/斜拍等视角） + 运镜（推/拉/摇/移/跟随/环绕/跟焦/滑轨/固定镜头等运镜）,
            "core_vision": // str;核心视觉画面：人物信息（如人物神情/姿态/动作，但是避免对人物的样貌和穿着进行描写；如：‘一个女生正在散步，非常开心的神态’，而非‘**黄头发的**美女正在散步，**穿着白色连衣裙**，非常开心的神态’）+ 场景细节（A-Roll）；或景物描写（B-Roll）,
            "lens_type": // str;镜头类型：A-Roll或B-Roll；A-Roll为人物镜头，B-Roll为场景镜头,
            "lines_narration": // str;台词/旁白，可为None,
            "bgm_style": // str;BGM风格，如：轻柔的小提琴独奏，节奏缓慢，旋律偏温暖，可为None,
            "sound_effects": // str;音效，如：树叶飘落的轻微沙沙声，可为None
            "lens_duration": // int;镜头时长，取值范围3-12
        },
        ... // 更多镜头信息
    ]
}
```

## 要求
- 需要对输入视频中的每个镜头片段进行拆解分析，不能遗漏任何一个镜头片段。（输入视频中每一个无转场的镜头片段，都应被视为一个独立的镜头片段，也即对应JSON中的一个lens_details元素）
- 请重读一遍任务和所有约束条件后，开始进行脚本创作。
"""

async def create_script_refer_video(
    client: AsyncArkClient,
    video_url: str,
    user_prompt: str = "请开始执行你的任务"
) -> str:
    user_message = ArkMessage(role="user")
    user_message.add_text(user_prompt)
    user_message.add_video(video_url)

    try:
        response = await client.chat(
            model="doubao-seed-1-6-251015",
            messages=[user_message],
            system_prompt=system_prompt_develop_script,
        )
        logger.info(f"开发脚本成功")
        return client.get_response_text(response)
    except APIError as e:
        logger.error(f"API错误: {e}")
        raise e

if __name__ == "__main__":
    async def main():
        async with AsyncArkClient() as client:
            script = await create_script_refer_video(
                client=client,
                video_url="video.mp4" 
            )
            io_handler.write_json(script, "script.json")
    asyncio.run(main())