单人部署数字人直播系统的60天实践：从环境配置到多账号运营的技术记录

import json
import os
from openai import OpenAI

client = OpenAI(api_key="your_api_key")

class ScriptGenerator:
    def __init__(self, knowledge_base_path: str):
        with open(knowledge_base_path, 'r', encoding='utf-8') as f:
            self.kb = json.load(f)

    def generate_batch(self, topic_list: list, output_dir: str) -> list:
        """批量生成脚本"""
        results = []
        for topic in topic_list:
            script = self._generate_single(topic)
            if script:
                filename = f"{topic[:20].replace('/', '_')}.json"
                filepath = os.path.join(output_dir, filename)
                with open(filepath, 'w', encoding='utf-8') as f:
                    json.dump(script, f, ensure_ascii=False, indent=2)
                results.append(script)
        return results

    def _generate_single(self, topic: str) -> dict:
        """生成单个脚本"""
        prompt = f"""
        为以下话题生成一个60秒直播/短视频脚本：
        话题：{topic}

        要求：
        1. 三段式结构：开场（吸引注意）→ 核心（信息输出）→ 结尾（引导互动）
        2. 开场用一句反常识提问或数据对比
        3. 核心内容信息密度适中，每段不超过3句话
        4. 结尾引导用户点赞、评论或关注
        5. 输出JSON格式：{{"opening": "", "core": "", "closing": ""}}
        """

        try:
            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"] = topic
            return script
        except Exception as e:
            print(f"生成脚本失败：{topic}，错误：{e}")
            return None

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

class VideoComposer:
    def __init__(self, avatar_path: str, background_path: str):
        self.avatar_path = avatar_path
        self.background_path = background_path

    async def synthesize_audio(self, text: str, output_path: str):
        """合成语音"""
        tts = edge_tts.Communicate(text, "zh-CN-XiaoxiaoNeural")
        await tts.save(output_path)

    def compose_video(self, script: dict, audio_path: str, output_path: str):
        """合成视频"""
        audio = AudioFileClip(audio_path)
        duration = audio.duration

        # 加载背景和形象
        if self.background_path:
            bg = ImageClip(self.background_path).set_duration(duration).resize(height=1080)
        else:
            bg = ImageClip(self.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 i, sentence in enumerate(sentences):
            clip = TextClip(
                sentence + "。",
                fontsize=38,
                color='white',
                stroke_color='black',
                stroke_width=2,
                method='caption',
                size=(bg.w * 0.85, None)
            ).set_start(i * segment_duration).set_duration(min(segment_duration, duration - i * segment_duration))
            subtitle_clips.append(clip)

        final = CompositeVideoClip([bg, *subtitle_clips])
        final = final.set_audio(audio)
        final.write_videofile(output_path, codec='libx264', audio_codec='aac', fps=24)

    def batch_compose(self, scripts: list, output_dir: str):
        """批量合成视频"""
        for i, script in enumerate(scripts):
            audio_path = f"{output_dir}/audio_{i}.wav"
            video_path = f"{output_dir}/video_{i}.mp4"

            # 合成语音
            full_text = script["opening"] + script["core"] + script["closing"]
            asyncio.run(self.synthesize_audio(full_text, audio_path))

            # 合成视频
            self.compose_video(script, audio_path, video_path)

import subprocess
import time
import json
from datetime import datetime, timedelta
from threading import Thread

class StreamScheduler:
    def __init__(self, config_path: str):
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = json.load(f)
        self.running_streams = {}

    def start_stream(self, account_id: str, video_path: str, duration_seconds: int = 1800):
        """为指定账号启动推流"""
        account = self.config["accounts"][account_id]
        stream_url = f"{account['rtmp_url']}/{account['stream_key']}"

        cmd = [
            "ffmpeg",
            "-re",
            "-stream_loop", "-1",
            "-i", video_path,
            "-c", "copy",
            "-f", "flv",
            "-t", str(duration_seconds),
            stream_url
        ]

        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        self.running_streams[account_id] = {
            "process": process,
            "start_time": datetime.now(),
            "duration": duration_seconds
        }
        return process

    def stop_stream(self, account_id: str):
        """停止指定账号的推流"""
        if account_id in self.running_streams:
            process = self.running_streams[account_id]["process"]
            process.terminate()
            del self.running_streams[account_id]

    def run_daily_schedule(self):
        """执行每日排期"""
        today = datetime.now().strftime("%Y-%m-%d")
        for account_id, schedule in self.config["daily_schedule"].items():
            if schedule["date"] == today:
                video_path = schedule.get("video_path")
                if video_path and account_id not in self.running_streams:
                    self.start_stream(account_id, video_path, schedule.get("duration", 1800))

    def stop_all_streams(self):
        """停止所有推流"""
        for account_id in list(self.running_streams.keys()):
            self.stop_stream(account_id)

    def get_status(self) -> dict:
        """获取所有推流状态"""
        status = {}
        for account_id, info in self.running_streams.items():
            elapsed = (datetime.now() - info["start_time"]).seconds
            status[account_id] = {
                "status": "running",
                "elapsed_seconds": elapsed,
                "duration": info["duration"]
            }
        return status