用Python脚本批量生成AI短剧：速创Sora2 API角色创建接口实战（附完整代码）

Python自动化脚本实战：Sora2 API角色一致性短剧批量生成方案

当AI视频生成技术遇上角色一致性需求，内容创作者便拥有了打造系列短剧的全新生产力工具。本文将深入探讨如何利用Python脚本与Sora2 API构建一套完整的自动化工作流，实现从角色创建到批量视频生成的全流程技术方案。

1. 环境准备与基础配置

在开始编写自动化脚本前，我们需要确保开发环境已正确配置。以下是基础环境要求：

• Python 3.8+
• requests库（用于API调用）
• aiohttp库（可选，用于异步请求）
• dotenv库（推荐，用于管理敏感信息）

首先创建项目目录结构：

sora2-automation/ ├── config/ │ └── .env ├── scripts/ │ ├── character_creator.py │ ├── video_generator.py │ └── batch_processor.py ├── data/ │ ├── prompts/ │ └── outputs/ └── utils/ ├── logger.py └── retry.py

安装必要的Python包：

pip install requests python-dotenv aiohttp

在.env文件中配置API密钥：

SORA2_API_KEY=your_api_key_here BASE_API_URL=https://api.wuyinkeji.com/api/sora2

2. 角色创建与管理模块

角色一致性是系列短剧的核心，我们需要先实现角色创建功能。以下是封装后的角色创建类：

import os import requests from dotenv import load_dotenv load_dotenv() class CharacterCreator: def __init__(self): self.api_key = os.getenv('SORA2_API_KEY') self.base_url = os.getenv('BASE_API_URL') self.headers = { 'Authorization': self.api_key, 'Content-Type': 'application/json' } def create_character(self, description, name=None): """创建新角色并返回角色ID""" url = f"{self.base_url}/createCharacter" payload = { "description": description, "name": name or "auto_generated_character" } try: response = requests.post(url, json=payload, headers=self.headers) response.raise_for_status() data = response.json() if data.get('code') == 0: return data['data']['characterId'] raise ValueError(data.get('msg', 'Unknown error')) except Exception as e: print(f"角色创建失败: {str(e)}") return None

使用示例：

creator = CharacterCreator() character_id = creator.create_character( name="职场精英小李", description="30岁男性，短发，戴黑框眼镜，常穿深色西装，表情严肃但偶尔会微笑" ) if character_id: print(f"角色创建成功，ID: {character_id}") # 建议将角色ID保存到数据库或文件

3. 批量视频生成引擎

实现角色创建后，我们需要构建批量视频生成系统。以下是核心生成器的实现：

import json import time from pathlib import Path from utils.logger import setup_logger logger = setup_logger('video_generator') class VideoGenerator: def __init__(self, character_id): self.character_id = character_id self.api_key = os.getenv('SORA2_API_KEY') self.base_url = os.getenv('BASE_API_URL') self.headers = { 'Authorization': self.api_key, 'Content-Type': 'application/json' } self.output_dir = Path('data/outputs') self.output_dir.mkdir(exist_ok=True) def generate_single(self, prompt, duration=10, resolution='1080p'): """生成单个视频并返回任务ID""" url = f"{self.base_url}/generateVideo" payload = { "characterId": self.character_id, "prompt": prompt, "duration": duration, "resolution": resolution } try: response = requests.post(url, json=payload, headers=self.headers) response.raise_for_status() data = response.json() if data.get('code') == 0: task_id = data['data']['taskId'] logger.info(f"视频生成任务已创建: {task_id}") return task_id raise ValueError(data.get('msg', 'Unknown error')) except Exception as e: logger.error(f"视频生成失败: {str(e)}") return None def batch_generate(self, prompts, interval=2): """批量生成视频""" results = [] for idx, prompt in enumerate(prompts, 1): logger.info(f"正在处理第 {idx}/{len(prompts)} 个提示词") task_id = self.generate_single(prompt) if task_id: results.append({ 'prompt': prompt, 'task_id': task_id, 'timestamp': time.time() }) time.sleep(interval) # 避免API限流 # 保存任务记录 output_file = self.output_dir / f"batch_{int(time.time())}.json" with open(output_file, 'w') as f: json.dump(results, f, indent=2) return results

4. 高级功能实现与优化

4.1 异步处理提升效率

对于大规模批量生成，同步请求效率较低。我们可以使用aiohttp实现异步请求：

import aiohttp import asyncio from typing import List class AsyncVideoGenerator(VideoGenerator): async def async_generate(self, session, prompt): """异步生成单个视频""" url = f"{self.base_url}/generateVideo" payload = { "characterId": self.character_id, "prompt": prompt, "duration": 10, "resolution": "1080p" } try: async with session.post(url, json=payload, headers=self.headers) as response: data = await response.json() if data.get('code') == 0: return data['data']['taskId'] raise ValueError(data.get('msg', 'Unknown error')) except Exception as e: logger.error(f"异步生成失败: {str(e)}") return None async def async_batch(self, prompts: List[str], concurrency=5): """并发批量生成""" connector = aiohttp.TCPConnector(limit=concurrency) timeout = aiohttp.ClientTimeout(total=60) async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session: tasks = [self.async_generate(session, prompt) for prompt in prompts] return await asyncio.gather(*tasks)

4.2 智能提示词池管理

为了实现多样化的短剧内容，我们需要管理提示词池：

import random from dataclasses import dataclass @dataclass class SceneTemplate: location: str actions: list emotions: list class PromptManager: def __init__(self): self.templates = { 'office': SceneTemplate( location="现代化办公室", actions=["主持会议", "撰写报告", "与同事讨论", "接听重要电话"], emotions=["专注", "疲惫", "兴奋", "困惑"] ), 'outdoor': SceneTemplate( location="城市公园", actions=["慢跑", "遛狗", "阅读", "野餐"], emotions=["放松", "快乐", "沉思", "期待"] ) } def generate_prompt(self, scene_type, character_name=None): """生成场景提示词""" template = self.templates.get(scene_type) if not template: raise ValueError(f"未知场景类型: {scene_type}") action = random.choice(template.actions) emotion = random.choice(template.emotions) character_ref = f"{character_name} " if character_name else "角色 " return ( f"{character_ref}在{template.location}，正在{action}，" f"表情看起来{emotion}" )

4.3 错误处理与重试机制

健壮的错误处理是自动化脚本的关键：

from tenacity import retry, stop_after_attempt, wait_exponential class RobustVideoGenerator(VideoGenerator): @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True ) def generate_with_retry(self, prompt): """带重试机制的生成方法""" return self.generate_single(prompt) def check_status(self, task_id): """检查任务状态""" url = f"{self.base_url}/taskStatus" params = {'taskId': task_id} try: response = requests.get(url, params=params, headers=self.headers) response.raise_for_status() return response.json() except Exception as e: logger.error(f"状态检查失败: {str(e)}") return None

5. 完整工作流与实战案例

5.1 端到端自动化流程

结合上述模块，我们可以构建完整的工作流：

1. 角色创建阶段

   • 定义角色特征
   • 调用API创建角色
   • 存储角色ID

2. 内容准备阶段

   • 设计场景模板
   • 生成多样化提示词
   • 准备元数据（时长、分辨率等）

3. 批量生成阶段

   • 并发调用生成API
   • 记录任务ID
   • 实施错误重试

4. 后期处理阶段

   • 检查任务状态
   • 下载生成视频
   • 组织输出文件

5.2 成本控制策略

大规模生成时，成本控制至关重要：

5.3 实战示例：职场成长系列短剧

以下是一个完整的系列短剧生成示例：

# 初始化组件 creator = CharacterCreator() prompt_manager = PromptManager() # 创建主角 protagonist_id = creator.create_character( name="林小夏", description="25岁职场新人，齐肩短发，常穿衬衫和西装外套，表情从青涩逐渐变得自信" ) if not protagonist_id: raise RuntimeError("角色创建失败，终止流程") # 准备场景序列 scenes = [ ("office", "第一天入职，紧张地认识新同事"), ("office", "第一次独立完成任务，获得上司认可"), ("office", "带领小组完成重要项目"), ("office", "获得晋升，发表感言") ] # 生成提示词 prompts = [ prompt_manager.generate_prompt( scene_type=scene[0], context=scene[1] ) for scene in scenes ] # 批量生成 generator = RobustVideoGenerator(protagonist_id) results = generator.batch_generate(prompts, interval=1.5) print(f"批量生成完成，共{len(results)}个任务已提交")

这套方案在实际项目中，可以帮助内容团队每周生成数百条角色一致的系列短剧，大幅降低创作成本。