Python自动化脚本实战Sora2 API角色一致性短剧批量生成方案当AI视频生成技术遇上角色一致性需求内容创作者便拥有了打造系列短剧的全新生产力工具。本文将深入探讨如何利用Python脚本与Sora2 API构建一套完整的自动化工作流实现从角色创建到批量视频生成的全流程技术方案。1. 环境准备与基础配置在开始编写自动化脚本前我们需要确保开发环境已正确配置。以下是基础环境要求Python 3.8requests库用于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_KEYyour_api_key_here BASE_API_URLhttps://api.wuyinkeji.com/api/sora22. 角色创建与管理模块角色一致性是系列短剧的核心我们需要先实现角色创建功能。以下是封装后的角色创建类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, nameNone): 创建新角色并返回角色ID url f{self.base_url}/createCharacter payload { description: description, name: name or auto_generated_character } try: response requests.post(url, jsonpayload, headersself.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职场精英小李, description30岁男性短发戴黑框眼镜常穿深色西装表情严肃但偶尔会微笑 ) 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_okTrue) def generate_single(self, prompt, duration10, resolution1080p): 生成单个视频并返回任务ID url f{self.base_url}/generateVideo payload { characterId: self.character_id, prompt: prompt, duration: duration, resolution: resolution } try: response requests.post(url, jsonpayload, headersself.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, interval2): 批量生成视频 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 / fbatch_{int(time.time())}.json with open(output_file, w) as f: json.dump(results, f, indent2) return results4. 高级功能实现与优化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, jsonpayload, headersself.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], concurrency5): 并发批量生成 connector aiohttp.TCPConnector(limitconcurrency) timeout aiohttp.ClientTimeout(total60) async with aiohttp.ClientSession(connectorconnector, timeouttimeout) 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_nameNone): 生成场景提示词 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( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), reraiseTrue ) 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, paramsparams, headersself.headers) response.raise_for_status() return response.json() except Exception as e: logger.error(f状态检查失败: {str(e)}) return None5. 完整工作流与实战案例5.1 端到端自动化流程结合上述模块我们可以构建完整的工作流角色创建阶段定义角色特征调用API创建角色存储角色ID内容准备阶段设计场景模板生成多样化提示词准备元数据时长、分辨率等批量生成阶段并发调用生成API记录任务ID实施错误重试后期处理阶段检查任务状态下载生成视频组织输出文件5.2 成本控制策略大规模生成时成本控制至关重要策略实施方法预期效果采样测试先生成少量样本验证质量避免大规模失败智能提示词使用模板生成有效提示减少无效生成错峰生成在低峰期执行批量任务可能获得更好稳定性结果缓存存储成功生成的提示词组合复用已验证内容5.3 实战示例职场成长系列短剧以下是一个完整的系列短剧生成示例# 初始化组件 creator CharacterCreator() prompt_manager PromptManager() # 创建主角 protagonist_id creator.create_character( name林小夏, description25岁职场新人齐肩短发常穿衬衫和西装外套表情从青涩逐渐变得自信 ) if not protagonist_id: raise RuntimeError(角色创建失败终止流程) # 准备场景序列 scenes [ (office, 第一天入职紧张地认识新同事), (office, 第一次独立完成任务获得上司认可), (office, 带领小组完成重要项目), (office, 获得晋升发表感言) ] # 生成提示词 prompts [ prompt_manager.generate_prompt( scene_typescene[0], contextscene[1] ) for scene in scenes ] # 批量生成 generator RobustVideoGenerator(protagonist_id) results generator.batch_generate(prompts, interval1.5) print(f批量生成完成共{len(results)}个任务已提交)这套方案在实际项目中可以帮助内容团队每周生成数百条角色一致的系列短剧大幅降低创作成本。
用Python脚本批量生成AI短剧：速创Sora2 API角色创建接口实战（附完整代码）
Python自动化脚本实战Sora2 API角色一致性短剧批量生成方案当AI视频生成技术遇上角色一致性需求内容创作者便拥有了打造系列短剧的全新生产力工具。本文将深入探讨如何利用Python脚本与Sora2 API构建一套完整的自动化工作流实现从角色创建到批量视频生成的全流程技术方案。1. 环境准备与基础配置在开始编写自动化脚本前我们需要确保开发环境已正确配置。以下是基础环境要求Python 3.8requests库用于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_KEYyour_api_key_here BASE_API_URLhttps://api.wuyinkeji.com/api/sora22. 角色创建与管理模块角色一致性是系列短剧的核心我们需要先实现角色创建功能。以下是封装后的角色创建类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, nameNone): 创建新角色并返回角色ID url f{self.base_url}/createCharacter payload { description: description, name: name or auto_generated_character } try: response requests.post(url, jsonpayload, headersself.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职场精英小李, description30岁男性短发戴黑框眼镜常穿深色西装表情严肃但偶尔会微笑 ) 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_okTrue) def generate_single(self, prompt, duration10, resolution1080p): 生成单个视频并返回任务ID url f{self.base_url}/generateVideo payload { characterId: self.character_id, prompt: prompt, duration: duration, resolution: resolution } try: response requests.post(url, jsonpayload, headersself.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, interval2): 批量生成视频 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 / fbatch_{int(time.time())}.json with open(output_file, w) as f: json.dump(results, f, indent2) return results4. 高级功能实现与优化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, jsonpayload, headersself.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], concurrency5): 并发批量生成 connector aiohttp.TCPConnector(limitconcurrency) timeout aiohttp.ClientTimeout(total60) async with aiohttp.ClientSession(connectorconnector, timeouttimeout) 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_nameNone): 生成场景提示词 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( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), reraiseTrue ) 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, paramsparams, headersself.headers) response.raise_for_status() return response.json() except Exception as e: logger.error(f状态检查失败: {str(e)}) return None5. 完整工作流与实战案例5.1 端到端自动化流程结合上述模块我们可以构建完整的工作流角色创建阶段定义角色特征调用API创建角色存储角色ID内容准备阶段设计场景模板生成多样化提示词准备元数据时长、分辨率等批量生成阶段并发调用生成API记录任务ID实施错误重试后期处理阶段检查任务状态下载生成视频组织输出文件5.2 成本控制策略大规模生成时成本控制至关重要策略实施方法预期效果采样测试先生成少量样本验证质量避免大规模失败智能提示词使用模板生成有效提示减少无效生成错峰生成在低峰期执行批量任务可能获得更好稳定性结果缓存存储成功生成的提示词组合复用已验证内容5.3 实战示例职场成长系列短剧以下是一个完整的系列短剧生成示例# 初始化组件 creator CharacterCreator() prompt_manager PromptManager() # 创建主角 protagonist_id creator.create_character( name林小夏, description25岁职场新人齐肩短发常穿衬衫和西装外套表情从青涩逐渐变得自信 ) if not protagonist_id: raise RuntimeError(角色创建失败终止流程) # 准备场景序列 scenes [ (office, 第一天入职紧张地认识新同事), (office, 第一次独立完成任务获得上司认可), (office, 带领小组完成重要项目), (office, 获得晋升发表感言) ] # 生成提示词 prompts [ prompt_manager.generate_prompt( scene_typescene[0], contextscene[1] ) for scene in scenes ] # 批量生成 generator RobustVideoGenerator(protagonist_id) results generator.batch_generate(prompts, interval1.5) print(f批量生成完成共{len(results)}个任务已提交)这套方案在实际项目中可以帮助内容团队每周生成数百条角色一致的系列短剧大幅降低创作成本。
