短剧工作室如何用 API 批量生成剧本？从任务拆分到审核交付

import json
import os
import urllib.request
from pathlib import Path

character_notes = Path("characters.txt").read_text(encoding="utf-8")
outline = Path("episode-01-outline.txt").read_text(encoding="utf-8")
prompt = f"""根据人物档案与大纲写第一集，不要增加角色。
只输出JSON对象，不加代码围栏或解释：
{{"episode":1,"scenes":[{{"location":"地点","action":"可拍动作",
"dialogue":[{{"speaker":"角色名","line":"台词"}}]}}],
"continuity_notes":["需要人工确认的状态变化"]}}
人物档案：{character_notes}
本集大纲：{outline}"""
payload = {
    "model": os.environ["TEAMOROUTER_MODEL"],
    "messages": [{"role": "user", "content": prompt}],
    "stream": False,
}
request = urllib.request.Request(
    "https://api.teamorouter.com/v1/chat/completions",
    data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
    headers={
        "Authorization": "Bearer " + os.environ["TEAMOROUTER_API_KEY"],
        "Content-Type": "application/json",
    },
    method="POST",
)
with urllib.request.urlopen(request, timeout=180) as response:
    result = json.load(response)
choice = result["choices"][0]
if choice.get("finish_reason") != "stop":
    raise ValueError("未正常结束，需人工检查，不能按完成稿入库")
script = json.loads(choice["message"]["content"])
if not isinstance(script, dict) or script.get("episode") != 1:
    raise ValueError("集数或顶层结构不符合约定")
scenes = script.get("scenes")
if not isinstance(scenes, list) or not scenes:
    raise ValueError("缺少有效场次")
for scene in scenes:
    if not isinstance(scene, dict):
        raise ValueError("场次必须为对象")
    if not all(isinstance(scene.get(k), str) and scene[k].strip()
               for k in ("location", "action")):
        raise ValueError("场景或动作缺失")
    dialogue = scene.get("dialogue")
    if not isinstance(dialogue, list):
        raise ValueError("对白字段必须为列表")
    for line in dialogue:
        if not isinstance(line, dict) or not all(
            isinstance(line.get(k), str) and line[k].strip()
            for k in ("speaker", "line")
        ):
            raise ValueError("对白结构不符合约定")
notes = script.get("continuity_notes")
if not isinstance(notes, list) or not all(isinstance(n, str) for n in notes):
    raise ValueError("连续性备注结构不符合约定")
print(json.dumps({"script": script, "usage": result.get("usage")},
                 ensure_ascii=False, indent=2))