从零搭一个 AI 短视频生成流水线：一句话出一条成片，中途接入 4sapi

一个选题
   ↓
① 文案脚本   选题 → 分镜脚本（每段：口播文案 + 画面描述）
   ↓
② 画面生成   每段画面描述 → 关键帧图片（文生图）
   ↓
③ 视频片段   关键帧 → 带动效的视频片段（图生视频）
   ↓
④ 配音       口播文案 → 配音音频（TTS）
   ↓
⑤ 合成成片   片段 + 配音 + 字幕 → 一条完整短视频

                   ┌──────────────────────────┐
   输入选题  ──→    │   Pipeline 流水线总控      │
                   └─────────────┬────────────┘
                                 │ 顺序流转
   ┌──────────┬──────────┬───────┼───────┬──────────┐
   ↓          ↓          ↓       ↓       ↓          ↓
[脚本]    [文生图]    [图生视频] [配音]  [字幕]   [合成]
   └──────────┴──────────┼───────┴───────┴──────────┘
                         ↓
            ┌──────────────────────────┐
            │   统一 API 网关 (4sapi)    │  ← 文本/图像/视频/语音全走这
            └──────────────────────────┘

https://4sapi.com/v1

# .env
FOURSAPI_BASE_URL=https://4sapi.com/v1
FOURSAPI_KEY=sk-你的令牌

# gateway.py —— 一个网关，吃下文本/图像/视频/语音
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["FOURSAPI_KEY"],
    base_url=os.environ["FOURSAPI_BASE_URL"],   # 指向中转网关
)

def chat(model, messages, **kw):
    return client.chat.completions.create(
        model=model, messages=messages, **kw).choices[0].message.content

def gen_image(prompt, size="1080x1920"):       # 竖屏短视频用 9:16
    return client.images.generate(
        model="flux-dev", prompt=prompt, size=size).data[0].url

def img_to_video(image_url, motion, seconds=4):
    return client.post("/videos/generations", body={
        "model": "kling-v1", "image": image_url,
        "motion": motion, "duration": seconds,
    })["video_url"]

def tts(text, voice="zh-male-1"):
    return client.audio.speech.create(
        model="tts-1", voice=voice, input=text).content

if __name__ == "__main__":
    print(chat("gpt-4o-mini",
        [{"role": "user", "content": "回复两个字：通了"}]))

import json

def script_agent(topic):
    prompt = f"""你是短视频编导。把选题扩写成 5-8 段分镜脚本，输出 JSON 数组。
每段含字段：narration(口播文案,口语化), visual(画面英文描述,用于文生图),
motion(运镜:推/拉/摇/移)。选题：{topic}"""
    raw = chat("gpt-4o",
        [{"role": "user", "content": prompt}],
        response_format={"type": "json_object"})
    return json.loads(raw)["scenes"]

def visual_agent(scene):
    # 统一画风后缀，让整条片子视觉连贯
    prompt = f"{scene['visual']}, cinematic, vertical 9:16, vivid color"
    frame = gen_image(prompt)                       # 文生图
    clip = img_to_video(frame, scene["motion"])     # 图生视频
    return clip

def audio_agent(scene):
    voice_audio = tts(scene["narration"])     # 配音
    subtitle = scene["narration"]             # 文案即字幕
    return voice_audio, subtitle

def compose_agent(clips, audios, subtitles):
    # 用 ffmpeg/moviepy 把每段视频 + 配音对齐，叠加字幕，顺序拼接
    # 这里给出思路，实际用 moviepy 的 concatenate + CompositeVideo
    final = video_toolkit.merge(
        video_clips=clips,
        audio_tracks=audios,
        subtitles=subtitles,
        bgm="bgm/light.mp3",      # 可选：垫一段背景音乐
    )
    return final     # 输出 mp4

def make_short_video(topic):
    scenes = script_agent(topic)            # ① 脚本

    clips, audios, subs = [], [], []
    for sc in scenes:
        clips.append(visual_agent(sc))      # ②③ 画面+视频
        a, s = audio_agent(sc)              # ④ 配音+字幕
        audios.append(a); subs.append(s)

    return compose_agent(clips, audios, subs)   # ⑤ 合成

function ShortVideoMaker() {
  const [topic, setTopic] = useState("");
  const [scenes, setScenes] = useState([]);   // 每段的实时状态

  async function run() {
    const res = await fetch("/api/video", {
      method: "POST", body: JSON.stringify({ topic }),
    });
    // SSE 回报每段进度：脚本✓ → 画面✓ → 视频✓ → 配音✓ → 合成✓
  }

  return (
    <div>
      <input value={topic} onChange={e => setTopic(e.target.value)}
             placeholder="输入选题，比如：3 个被低估的通勤好物" />
      <button onClick={run}>一键生成短视频</button>
      <SceneTimeline scenes={scenes} />   {/* 分段进度可视化 */}
    </div>
  );
}