AI 漫剧变现工程化：从分镜生成到商业闭环的专业实践

import json, os, time, requests

SHOT_SCHEMA = {
    "shot_id": "s001",
    "scene": "教室黄昏",
    "character": "女主",
    "emotion": "惊讶",
    "camera": "特写",
    "dialogue": "你怎么会在这里？",
    "duration": 3,
    "style": "日系漫画，赛璐璐上色，柔和光影",
}

def build_image_prompt(shot: dict) -> str:
    return (
        f"{shot['style']}，{shot['scene']}，{shot['character']}，"
        f"表情：{shot['emotion']}，镜头：{shot['camera']}，"
        "构图干净，无文字水印，无多余肢体"
    )

API_BASE = os.environ["AIGC_API_BASE"]
HEADERS = {
    "Authorization": f"Bearer {os.environ['AIGC_API_KEY']}",
    "Content-Type": "application/json",
}

def submit(prompt: str, ratio="9:16") -> str:
    r = requests.post(
        f"{API_BASE}/v1/images/tasks",
        json={"prompt": prompt, "ratio": ratio},
        headers=HEADERS, timeout=60,
    )
    r.raise_for_status()
    return r.json()["task_id"]

def poll(task_id: str, timeout=300):
    start = time.time()
    while time.time() - start < timeout:
        r = requests.get(
            f"{API_BASE}/v1/images/tasks/{task_id}",
            headers=HEADERS, timeout=30,
        )
        data = r.json()
        if data["status"] == "succeeded":
            return data["image_url"]
        if data["status"] == "failed":
            raise RuntimeError(data.get("error"))
        time.sleep(4)
    raise TimeoutError(task_id)

def batch_generate(shots, max_retry=2):
    results = []
    for shot in shots:
        prompt = build_image_prompt(shot)
        for attempt in range(max_retry + 1):
            try:
                url = poll(submit(prompt))
                results.append({**shot, "image_url": url})
                break
            except Exception as e:
                if attempt == max_retry:
                    results.append({**shot, "error": str(e)})
                time.sleep(2 ** attempt)
    return results

def make_tts(text: str, voice="female_01") -> str:
    r = requests.post(
        f"{API_BASE}/v1/tts",
        json={"text": text, "voice": voice},
        headers=HEADERS, timeout=60,
    )
    r.raise_for_status()
    return r.json()["audio_url"]

# 图片 + 音频合成单镜头
ffmpeg -loop 1 -i shot.png -i voice.mp3 -shortest \
  -vf "scale=1080:1920" -c:v libx264 -c:a aac shot.mp4

# 拼接所有镜头
ffmpeg -f concat -safe 0 -i list.txt -c copy final.mp4

def metrics(views, likes, comments, shares, cost):
    engagement = (likes + comments + shares) / max(views, 1)
    roi = (views * 0.001 - cost) / max(cost, 1)  # 示意，非真实单价
    return {"engagement": round(engagement, 4), "roi": round(roi, 4)}