| name | ai-video-platform |
| description | 构建 AI 短视频一键生产发布客户端。整合 MoneyPrinterTurbo(AI视频生成)与 AiToEarn/social-auto-upload(多平台一键发布)。支持5种视频生成方式(拼素材/可灵/即梦/Wan2.1/Runway),AI文案生成,自动配音字幕,一键发布到抖音/快手/B站/小红书/微博/视频号/TikTok/YouTube/Instagram等12+平台。技术栈:Electron + React + Python FastAPI + Playwright + SQLite。触发词:AI视频平台、视频生成发布、短视频客户端、MoneyPrinterTurbo、AiToEarn、一键发布、视频矩阵。 |
AI 短视频一键生产发布平台 SKILL
项目目标
将 MoneyPrinterTurbo(视频生成)和 AiToEarn / social-auto-upload(多平台发布)整合为一个 Windows 桌面客户端。用户输入主题 → AI生成视频 → 一键发布全平台。
核心依赖项目
完整目录结构
ai-video-platform/
├── electron/
│ ├── main.ts # 主进程,窗口管理,启动Python后端
│ ├── preload.ts # 预加载,暴露IPC给渲染进程
│ └── python-manager.ts # Python子进程管理
├── src/ # React前端
│ ├── App.tsx
│ ├── main.tsx
│ ├── pages/
│ │ ├── Create.tsx # 视频创作页(核心)
│ │ ├── Publish.tsx # 发布管理页
│ │ ├── Queue.tsx # 任务队列页
│ │ ├── Accounts.tsx # 账号管理页
│ │ ├── Library.tsx # 素材库页
│ │ └── Settings.tsx # 设置页
│ ├── components/
│ │ ├── VideoPreview.tsx # 视频预览组件
│ │ ├── CopywriterPanel.tsx # 文案生成面板
│ │ ├── PublishPanel.tsx # 发布配置面板
│ │ └── TaskCard.tsx # 任务卡片组件
│ ├── store/
│ │ ├── videoStore.ts # 视频生成状态
│ │ ├── publishStore.ts # 发布状态
│ │ └── settingsStore.ts # 设置状态
│ └── api/
│ └── client.ts # 封装后端API调用
├── backend/ # Python FastAPI后端
│ ├── main.py # FastAPI入口
│ ├── config.py # 配置管理(API Keys等)
│ ├── video/
│ │ ├── generator.py # 视频生成调度器(路由到不同方式)
│ │ ├── stock.py # 拼素材方式(Pexels/Pixabay)
│ │ ├── kling.py # 可灵API
│ │ ├── jimeng.py # 即梦API
│ │ ├── wan21.py # Wan2.1本地模型
│ │ └── runway.py # Runway API
│ ├── llm/
│ │ └── copywriter.py # AI文案生成(支持多模型)
│ ├── tts/
│ │ └── synthesizer.py # 配音合成(Edge/Azure/OpenAI TTS)
│ ├── subtitle/
│ │ └── generator.py # 字幕生成(faster-whisper)
│ ├── compose/
│ │ └── merger.py # 视频合成(moviepy+ffmpeg)
│ ├── publish/
│ │ ├── base.py # 发布基类
│ │ ├── douyin.py # 抖音发布
│ │ ├── bilibili.py # B站发布
│ │ ├── xiaohongshu.py # 小红书发布
│ │ ├── kuaishou.py # 快手发布
│ │ ├── wechat_video.py # 微信视频号发布
│ │ ├── weibo.py # 微博发布
│ │ ├── tiktok.py # TikTok发布
│ │ └── youtube.py # YouTube发布
│ ├── db/
│ │ ├── database.py # SQLite连接
│ │ └── models.py # 数据模型
│ └── vendor/ # 第三方项目(git clone进来)
│ ├── MoneyPrinterTurbo/
│ └── social-auto-upload/
├── accounts/ # 平台账号Cookie存储(加密)
├── output/ # 生成视频输出目录
├── assets/ # 本地素材库
├── package.json
├── vite.config.ts
├── tsconfig.json
├── electron-builder.yml
└── requirements.txt
Step 1: 初始化项目
mkdir ai-video-platform && cd ai-video-platform
npm init -y
npm install electron react react-dom antd @ant-design/icons zustand react-router-dom
npm install -D typescript @types/react @types/react-dom vite @vitejs/plugin-react electron-builder
pip install fastapi uvicorn moviepy edge-tts faster-whisper requests pyyaml openai playwright
pip install --break-system-packages yt-dlp
python -m playwright install chromium
git clone https://github.com/harry0703/MoneyPrinterTurbo backend/vendor/MoneyPrinterTurbo
git clone https://github.com/dreammis/social-auto-upload backend/vendor/social-auto-upload
Step 2: Python 后端
backend/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from video.generator import router as video_router
from llm.copywriter import router as llm_router
from tts.synthesizer import router as tts_router
from publish.douyin import router as douyin_router
from publish.bilibili import router as bili_router
from publish.xiaohongshu import router as xhs_router
from publish.kuaishou import router as ks_router
from db.database import init_db
app = FastAPI(title="AI Video Platform API")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
@app.on_event("startup")
async def startup():
init_db()
app.include_router(video_router, prefix="/api/video", tags=["video"])
app.include_router(llm_router, prefix="/api/llm", tags=["llm"])
app.include_router(tts_router, prefix="/api/tts", tags=["tts"])
app.include_router(douyin_router, prefix="/api/publish/douyin", tags=["publish"])
app.include_router(bili_router, prefix="/api/publish/bilibili", tags=["publish"])
app.include_router(xhs_router, prefix="/api/publish/xiaohongshu", tags=["publish"])
app.include_router(ks_router, prefix="/api/publish/kuaishou", tags=["publish"])
backend/video/generator.py(调度器)
from fastapi import APIRouter
from pydantic import BaseModel
from typing import Literal
import asyncio, uuid
router = APIRouter()
tasks = {}
class VideoRequest(BaseModel):
topic: str
mode: Literal["stock", "kling", "jimeng", "wan21", "runway"]
aspect_ratio: Literal["9:16", "16:9"] = "9:16"
duration: int = 30
script: str = ""
@router.post("/generate")
async def generate_video(req: VideoRequest):
task_id = str(uuid.uuid4())
tasks[task_id] = {"status": "pending", "progress": 0, "url": None}
asyncio.create_task(_run_generation(task_id, req))
return {"task_id": task_id}
@router.get("/status/{task_id}")
async def get_status(task_id: str):
return tasks.get(task_id, {"status": "not_found"})
async def _run_generation(task_id: str, req: VideoRequest):
tasks[task_id]["status"] = "running"
try:
if req.mode == "stock":
from .stock import generate
elif req.mode == "kling":
from .kling import generate
elif req.mode == "jimeng":
from .jimeng import generate
elif req.mode == "wan21":
from .wan21 import generate
elif req.mode == "runway":
from .runway import generate
output_path = await generate(req)
tasks[task_id] = {"status": "done", "progress": 100, "url": output_path}
except Exception as e:
tasks[task_id] = {"status": "error", "error": str(e)}
backend/video/stock.py(拼素材,基于MoneyPrinterTurbo)
import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../vendor/MoneyPrinterTurbo'))
from app.services.video import VideoService
from app.services.material import MaterialService
async def generate(req) -> str:
"""复用 MoneyPrinterTurbo 的核心逻辑生成拼素材视频"""
materials = MaterialService.search_videos(req.topic, count=5)
output = VideoService.combine(
materials=materials,
script=req.script,
aspect_ratio=req.aspect_ratio,
output_dir="output/"
)
return output
backend/video/kling.py(可灵API)
import requests, time
KLING_BASE = "https://api.klingai.com/v1"
def
import requests, time, jwt
KLING_BASE = 'https://api.klingai.com/v1'
def _get_token(ak, sk):
payload = {'iss': ak, 'exp': int(time.time())+1800, 'nbf': int(time.time())-5}
return jwt.encode(payload, sk, algorithm='HS256')
async def generate(req):
from config import get_config
cfg = get_config()
token = _get_token(cfg.kling_ak, cfg.kling_sk)
headers = {'Authorization': f'Bearer {token}'}
body = {'model_name': 'kling-v1', 'prompt': req.script or req.topic,
'aspect_ratio': req.aspect_ratio, 'duration': str(min(req.duration,10))}
r = requests.post(f'{KLING_BASE}/videos/text2video', headers=headers, json=body)
task_id = r.json()['data']['task_id']
for _ in range(120):
time.sleep(5)
r = requests.get(f'{KLING_BASE}/videos/text2video/{task_id}', headers=headers)
d = r.json()['data']
if d['task_status'] == 'succeed': return d['task_result']['videos'][0]['url']
if d['task_status'] == 'failed': raise Exception('Kling failed')
raise TimeoutError('Kling timeout')
async def generate(req):
from config import get_config
import requests, time
cfg = get_config()
headers = {'Authorization': f'Bearer {cfg.jimeng_api_key}'}
body = {'prompt': req.script or req.topic, 'ratio': req.aspect_ratio.replace(':','_')}
r = requests.post('https://jimeng.jianying.com/mweb/v1/generate_video', headers=headers, json=body)
task_id = r.json()['data']['task_id']
for _ in range(120):
time.sleep(5)
r = requests.get(f'https://jimeng.jianying.com/mweb/v1/task/{task_id}', headers=headers)
d = r.json()['data']
if d['status'] == 'done': return d['video_url']
raise TimeoutError('Jimeng timeout')
async def generate(req):
import runwayml, time
from config import get_config
cfg = get_config()
client = runwayml.RunwayML(api_key=cfg.runway_api_key)
ratio = '768:1280' if req.aspect_ratio == '9:16' else '1280:768'
task = client.image_to_video.create(model='gen4_turbo', prompt_text=req.script or req.topic, ratio=ratio)
for _ in range(120):
time.sleep(5)
result = client.tasks.retrieve(task.id)
if result.status == 'SUCCEEDED': return result.output[0]
if result.status == 'FAILED': raise Exception('Runway failed')
raise TimeoutError('Runway timeout')
async def generate(req):
from diffusers import AutoPipelineForText2Video
import torch
from config import get_config
cfg = get_config()
pipe = AutoPipelineForText2Video.from_pretrained(cfg.wan21_model_path, torch_dtype=torch.bfloat16).to('cuda')
output = pipe(prompt=req.script or req.topic, num_frames=req.duration*16).frames[0]
out_path = f'output/wan21_{req.topic[:10]}.mp4'
output.save(out_path)
return out_path
---
PROMPT_TEMPLATE:
主题:{topic} 平台:{platform} 风格:{style}
要求: 1.开头钩子 2.口语化 3.结尾引导 4.生成{count}个版本用---分隔
模型支持: DeepSeek(默认)/OpenAI/Gemini/通义千问
配置: llm_api_key + llm_base_url + llm_model
---
音色选项:
zh-female: zh-CN-XiaoxiaoNeural
zh-male: zh-CN-YunxiNeural
en-female: en-US-JennyNeural
en-male: en-US-GuyNeural
调用: edge_tts.Communicate(text, voice).save(output_path)
---
model = WhisperModel('base', device='cpu', compute_type='int8')
segments, _ = model.transcribe(audio_path, language='zh')
返回: [{start, end, text}, ...]
---
基于 social-auto-upload 直接复用各平台 uploader:
douyin: uploader.douyin_uploader.main.DouYinVideo
bilibili: uploader.bilibili_uploader.main.BilibiliUploader
xiaohongshu: uploader.xhs_uploader.main.XhsUploader
kuaishou: uploader.ks_uploader.main.KSVideo
tiktok: uploader.tiktok_uploader.main.TiktokVideo
账号Cookie存储路径: accounts/{platform}_{account_name}.json
首次登录: 运行 setup 函数打开浏览器扫码/登录
---
启动流程:
1. app.whenReady() -> startPythonBackend()
2. Python: uvicorn main:app --host 127.0.0.1 --port 8765
3. 等待后端就绪(轮询 /health 接口)
4. 加载前端页面
前后端通信: HTTP REST (http://127.0.0.1:8765)
---
LLM: llm_api_key, llm_base_url, llm_model
视频生成: kling_ak, kling_sk, jimeng_api_key, runway_api_key, wan21_model_path
素材: pexels_api_key, pixabay_api_key
代理: proxy (海外平台用)
---
Python后端: pyinstaller --onefile backend/main.py -> dist/backend_server.exe
Electron: electron-builder --win -> dist/AI视频平台 Setup.exe
extraResources: backend/dist/, accounts/, output/
---
POST /api/llm/copywrite - 生成文案
POST /api/video/generate - 提交视频生成任务
GET /api/video/status/{id} - 查询进度
POST /api/tts/synthesize - 合成配音
POST /api/subtitle/generate - 生成字幕
POST /api/publish/douyin/publish - 发布抖音
POST /api/publish/bilibili/publish - 发布B站
POST /api/publish/xiaohongshu/publish - 发布小红书
POST /api/publish/kuaishou/publish - 发布快手
POST /api/publish/tiktok/publish - 发布TikTok
POST /api/publish/youtube/publish - 发布YouTube
GET /api/accounts/list - 获取账号列表
GET /api/queue/list - 获取任务队列
POST /api/settings/save - 保存配置
---
1. Python后端端口 8765,Electron启动时自动拉起,退出时自动关闭
2. 账号Cookie加密存储,不上传
3. 可灵/即梦/Runway均为异步任务,轮询间隔5秒,超时10分钟
4. social-auto-upload需要Playwright+Chrome,首次运行自动安装
5. Wan2.1需要CUDA GPU,无GPU时前端自动禁用该选项
6. 海外平台(TikTok/YouTube/Instagram)需配置代理
7. 拼素材需要Pexels API Key(免费注册)