单人数字人直播系统的自动化部署实践：从内容批量生产到多平台分发

选题库 → [内容生产管线] → 脚本/素材 → [视频合成] → 素材库
                ↓
          统一IP定位与话术体系
                ↓
          数字人直播系统 → RTMP推流 → 多平台分发

{
  "topics": [
    {
      "id": "T001",
      "category": "劳动纠纷",
      "question": "公司单方面降薪是否合法",
      "target_platform": ["抖音", "视频号"],
      "tags": ["劳动法", "职场权益"],
      "publish_frequency": "每周2次"
    }
  ]
}

import json
from openai import OpenAI

client = OpenAI(api_key="your_api_key")

def generate_batch_scripts(topics: list, platform: str = "general") -> list:
    """批量生成短视频脚本"""
    results = []
    for topic in topics:
        prompt = f"""
        请根据以下话题生成一个60秒短视频脚本。
        话题：{topic['question']}
        分类：{topic.get('category', '')}
        目标平台：{platform}

        要求：
        1. 三段式结构：开场提问→核心解答→结尾引导
        2. 开场用反常识提问或数据对比抓住注意力
        3. 核心部分分3个要点，每点不超过2句话
        4. 结尾引导点赞、评论或关注
        5. 输出JSON格式：{{"opening": "", "core": "", "closing": ""}}
        """

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.8
        )

        script = json.loads(response.choices[0].message.content)
        script["topic_id"] = topic["id"]
        script["category"] = topic.get("category", "")
        results.append(script)

    return results

import asyncio
import edge_tts
from moviepy.editor import ImageClip, AudioFileClip, CompositeVideoClip, TextClip
import os

async def batch_synthesize(scripts: list, output_dir: str):
    """批量合成语音"""
    semaphore = asyncio.Semaphore(5)

    async def synthesize_one(text: str, path: str):
        async with semaphore:
            tts = edge_tts.Communicate(text, "zh-CN-XiaoxiaoNeural")
            await tts.save(path)

    tasks = []
    for i, script in enumerate(scripts):
        full_text = script["opening"] + script["core"] + script["closing"]
        path = os.path.join(output_dir, f"audio_{i}.wav")
        tasks.append(synthesize_one(full_text, path))

    await asyncio.gather(*tasks)

def batch_compose_videos(scripts: list, audio_dir: str, avatar_path: str, output_dir: str):
    """批量合成视频"""
    for i, script in enumerate(scripts):
        audio_path = os.path.join(audio_dir, f"audio_{i}.wav")
        output_path = os.path.join(output_dir, f"video_{i}.mp4")

        audio = AudioFileClip(audio_path)
        duration = audio.duration

        avatar = ImageClip(avatar_path).set_duration(duration).resize(height=1080)

        full_text = script["opening"] + script["core"] + script["closing"]
        sentences = [s for s in full_text.split("。") if s.strip()]
        if not sentences:
            sentences = [full_text]

        segment_duration = duration / len(sentences)
        subtitle_clips = []

        for j, sentence in enumerate(sentences):
            clip = TextClip(
                sentence + "。",
                fontsize=38,
                color='white',
                stroke_color='black',
                stroke_width=2,
                method='caption',
                size=(avatar.w * 0.85, None)
            ).set_start(j * segment_duration).set_duration(segment_duration)
            subtitle_clips.append(clip)

        final = CompositeVideoClip([avatar, *subtitle_clips])
        final = final.set_audio(audio)
        final.write_videofile(output_path, codec='libx264', audio_codec='aac', fps=24)
        print(f"已生成：{output_path}")

{
  "session_name": "法律知识科普",
  "duration_minutes": 30,
  "segments": [
    {"time": "0-3min", "type": "开场", "template": "欢迎来到直播间，今天我们来聊一个大家都很关心的问题..."},
    {"time": "3-20min", "type": "核心内容", "template": "关于{topic}，很多朋友都问过我..."},
    {"time": "20-27min", "type": "互动", "template": "如果你也遇到了类似的情况，欢迎在评论区留言..."},
    {"time": "27-30min", "type": "结尾", "template": "感谢大家的观看，我们下期再见..."}
  ]
}

{
  "platforms": [
    {
      "name": "抖音",
      "rtmp_url": "rtmp://push.douyin.com/live",
      "stream_key": "your_stream_key",
      "schedule": "19:00-20:00",
      "enabled": true
    },
    {
      "name": "视频号",
      "rtmp_url": "rtmp://push.weixin.qq.com/live",
      "stream_key": "your_stream_key",
      "schedule": "20:30-21:30",
      "enabled": true
    }
  ]
}

import subprocess
import time
import json
from datetime import datetime

class AutoStream:
    def __init__(self, config_path: str):
        with open(config_path, 'r') as f:
            self.config = json.load(f)
        self.running = {}

    def start_stream(self, platform: dict, video_path: str, duration_seconds: int = 3600):
        """启动单个平台推流"""
        url = f"{platform['rtmp_url']}/{platform['stream_key']}"

        cmd = [
            "ffmpeg",
            "-re",
            "-stream_loop", "-1",
            "-i", video_path,
            "-c", "copy",
            "-f", "flv",
            "-reconnect", "1",
            "-reconnect_at_eof", "1",
            "-reconnect_streamed", "1",
            "-reconnect_delay_max", "5",
            "-t", str(duration_seconds),
            url
        ]

        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        self.running[platform['name']] = {
            "process": process,
            "start_time": datetime.now(),
            "platform": platform['name']
        }
        print(f"[{datetime.now()}] 推流已启动：{platform['name']}")

    def stop_all(self):
        """停止所有推流"""
        for name, info in self.running.items():
            info["process"].terminate()
            print(f"[{datetime.now()}] 已停止：{name}")
        self.running.clear()

    def get_status(self):
        """获取运行状态"""
        status = {}
        for name, info in self.running.items():
            elapsed = (datetime.now() - info["start_time"]).seconds
            status[name] = {
                "status": "running",
                "elapsed_minutes": elapsed // 60
            }
        return status

# 创建每日推流任务
schtasks /create /tn "AutoStream_Daily" /tr "python C:\stream\auto_stream.py" /sc daily /st 19:00

CREATE TABLE content_distribution (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    content_id TEXT,
    platform TEXT,
    title TEXT,
    publish_time TEXT,
    views INTEGER DEFAULT 0,
    likes INTEGER DEFAULT 0,
    comments INTEGER DEFAULT 0,
    shares INTEGER DEFAULT 0,
    status TEXT
);

CREATE TABLE live_sessions (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    platform TEXT,
    start_time TEXT,
    end_time TEXT,
    duration_minutes INTEGER,
    peak_viewers INTEGER,
    total_messages INTEGER,
    status TEXT
);

def check_distribution_status(days: int = 7) -> dict:
    """检查最近N天的内容分发状态"""
    cursor = conn.cursor()
    cursor.execute("""
        SELECT platform, COUNT(*) as total,
               SUM(CASE WHEN status = 'published' THEN 1 ELSE 0 END) as published
        FROM content_distribution
        WHERE publish_time >= date('now', ?)
        GROUP BY platform
    """, (f'-{days} days',))

    results = {}
    for row in cursor.fetchall():
        results[row[0]] = {
            "total": row[1],
            "published": row[2],
            "sync_rate": round(row[2] / row[1] * 100, 1)
        }
    return results

# 安装Python依赖
pip install openai edge-tts moviepy opencv-python

# 验证FFmpeg
ffmpeg -version