AI短视频生成的全链路工程实践——从脚本到成品的自动化流水线

import json
import requests

def evaluate_topic(topic: str, api_key: str) -> dict:
    prompt = f"""
    你是一个视频数据分析专家。请评估以下选题的爆款潜力，按0-10分打分：
    选题：{topic}
    输出JSON格式：{{"conflict":0-10, "curiosity":0-10, "practical":0-10, "timeliness":0-10, "total":0-10}}
    """
    response = requests.post(
        "https://api.anthropic.com/v1/messages",
        headers={"x-api-key": api_key},
        json={
            "model": "claude-3-5-sonnet-20241022",
            "max_tokens": 200,
            "messages": [{"role": "user", "content": prompt}]
        }
    )
    return json.loads(response.json()["content"][0]["text"])

角色：短视频脚本架构师
任务：为选题“{topic}”生成分镜骨架，要求每个分镜包含：时间戳(秒)、核心信息点、情绪标签。
输出格式：JSON列表，每项{"id":int, "time":"start-end", "key":"...", "emotion":"..."}
约束：总时长60秒，共12-15个分镜，每镜4-5秒。

def fill_script_for_shot(shot: dict, api_key: str) -> str:
    prompt = f"为以下分镜撰写中文口播文案（约20字）：\n{shot['key']}\n要求口语化、带情绪。"
    # 调用API返回文案
    return response

class SSMLBuilder:
    def __init__(self, voice="zh-CN-XiaoxiaoNeural"):
        self.voice = voice

    def wrap(self, sentences: list, emphasis_keywords: list = None) -> str:
        ssml = f'<speak version="1.0" xml:lang="zh-CN"><voice name="{self.voice}">'
        for sent in sentences:
            if emphasis_keywords:
                for kw in emphasis_keywords:
                    if kw in sent:
                        sent = sent.replace(kw, f'{kw}<break time="300ms"/>')
            ssml += f'<prosody rate="0%" pitch="0%">{sent}</prosody>'
            ssml += '<break time="200ms"/>'
        ssml += '</voice></speak>'
        return ssml

def generate_prompts(subject: str, style="cinematic") -> list:
    base = f"{subject}, {style}, 4k, high quality"
    return [f"{base}, wide shot", f"{base}, close-up", f"{base}, dynamic angle"]

import requests
def search_pexels(keyword: str, api_key: str) -> list:
    url = f"https://api.pexels.com/videos/search?query={keyword}&per_page=5"
    resp = requests.get(url, headers={"Authorization": api_key}).json()
    return [v['url'] for v in resp.get('videos', [])]

import subprocess
def concat_clips(clip_paths: list, output: str):
    with open("concat.txt", "w") as f:
        for p in clip_paths:
            f.write(f"file '{p}'\n")
    subprocess.run(["ffmpeg", "-f", "concat", "-safe", "0", "-i", "concat.txt", "-c", "copy", output])

ffmpeg -i video.mp4 -i bgm.mp3 -filter_complex "[0:a]volume=1[a0];[1:a]volume=0.15[a1];[a0][a1]amix=inputs=2:duration=longest" -c:v copy output.mp4