单人部署AI数字人直播系统的60天实践笔记：工具选型与效率分析

import json
from openai import OpenAI

client = OpenAI(api_key="your_api_key")

def generate_live_script(product_name: str, duration_minutes: int = 30) -> dict:
    # 第一级：生成话术提纲
    outline_prompt = f"""你是一名直播运营策划。请为{product_name}设计一个{duration_minutes}分钟的数字人直播话术提纲。
    要求：
    1. 按时间线划分为开场、产品介绍、互动答疑、促单成交四个阶段
    2. 每个阶段标注核心话术目标
    3. 输出格式为JSON，包含phase（阶段名）、duration（时长）、objective（目标）、key_points（关键话术列表）四个字段
    """
    
    outline_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": outline_prompt}],
        response_format={"type": "json_object"},
        temperature=0.7
    )
    
    outline = json.loads(outline_response.choices[0].message.content)
    
    # 第二级：逐阶段扩展为完整话术
    full_script = {"phases": []}
    for phase in outline["phases"]:
        expand_prompt = f"""请将以下直播阶段的关键话术点扩展为完整的口播文案：
        阶段名称：{phase['phase']}
        目标：{phase['objective']}
        关键话术点：{', '.join(phase['key_points'])}
        要求：文案自然流畅，适合语音播报，总字数控制在{int(phase['duration'] * 150)}字左右（按每分钟150字语速估算）
        """
        
        expand_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": expand_prompt}],
            temperature=0.85
        )
        
        phase["full_text"] = expand_response.choices[0].message.content
        full_script["phases"].append(phase)
    
    return full_script

import edge_tts
import asyncio
from pydub import AudioSegment
import os

async def text_to_speech(text: str, output_path: str, voice: str = "zh-CN-XiaoxiaoNeural"):
    """
    使用Edge TTS将文本转为语音
    voice参数可选：zh-CN-XiaoxiaoNeural（女声）、zh-CN-YunxiNeural（男声）
    """
    tts = edge_tts.Communicate(text, voice)
    await tts.save(output_path)

def normalize_audio(input_path: str, output_path: str, target_dbfs: int = -20):
    """
    将音频归一化到目标响度，避免多段音频音量不一致
    """
    audio = AudioSegment.from_wav(input_path)
    change_in_dbfs = target_dbfs - audio.dbfs
    normalized = audio.apply_gain(change_in_dbfs)
    normalized.export(output_path, format="wav")

def concatenate_audios(segment_paths: list, output_path: str, gap_ms: int = 300):
    """
    拼接多段音频，段间插入静音间隔
    """
    combined = AudioSegment.empty()
    silence = AudioSegment.silent(duration=gap_ms)
    for path in segment_paths:
        segment = AudioSegment.from_wav(path)
        combined += segment + silence
    combined.export(output_path, format="wav")

import numpy as np
import cv2
from moviepy.editor import VideoFileClip, AudioFileClip, ImageClip, CompositeVideoClip
from moviepy.video.VideoClip import VideoClip

def create_avatar_video(
    avatar_image_path: str, 
    audio_path: str, 
    output_path: str, 
    duration: float,
    mouth_open_ratio: float = 0.08
):
    """
    生成基础数字人视频：静态形象 + 口型模拟
    mouth_open_ratio控制口部开合幅度，建议范围0.05-0.12
    """
    # 加载音频获取波形
    audio = AudioFileClip(audio_path)
    samples = audio.to_soundarray(fps=30)
    # 计算帧级音频能量（简化为RMS值）
    rms = np.sqrt(np.mean(samples**2, axis=1))
    rms_normalized = rms / np.max(rms)  # 归一化到0-1
    
    # 加载静态形象
    avatar = ImageClip(avatar_image_path).set_duration(duration).resize(height=1080)
    
    # 创建口型遮罩动画（模拟口部开合）
    def make_frame(t):
        frame = avatar.get_frame(t)
        # 获取当前帧对应的音频能量值
        idx = int(t * 30)  # 30fps
        if idx < len(rms_normalized):
            energy = rms_normalized[idx]
        else:
            energy = 0
        
        # 计算口部开合程度：基础开度 + 能量驱动
        mouth_scale = 1.0 + mouth_open_ratio * energy
        
        # 在画面下半部分绘制一个椭圆遮罩模拟口型
        h, w = frame.shape[:2]
        center = (w // 2, int(h * 0.72))  # 口部位置（需根据具体形象调整）
        axes = (int(40 * mouth_scale), int(25 * mouth_scale))
        
        # 使用椭圆遮罩叠加在口部区域
        mask = np.zeros((h, w), dtype=np.uint8)
        cv2.ellipse(mask, center, axes, 0, 0, 360, 255, -1)
        
        # 将遮罩区域变暗，模拟口部开合
        frame[mask > 0] = (frame[mask > 0] * 0.6).astype(np.uint8)
        return frame
    
    avatar_with_mouth = VideoClip(make_frame, duration=duration)
    avatar_with_mouth = avatar_with_mouth.set_audio(audio)
    
    # 导出
    avatar_with_mouth.write_videofile(
        output_path, 
        codec='libx264', 
        audio_codec='aac',
        fps=30,
        bitrate='3000k'
    )

import subprocess
import time
from datetime import datetime

def stream_to_platform(video_path: str, rtmp_url: str, stream_key: str, duration_seconds: int = 1800):
    """
    使用FFmpeg将本地视频推流至直播平台
    rtmp_url和stream_key需从目标平台的后台获取
    """
    full_rtmp = f"{rtmp_url}/{stream_key}"
    
    # 构建FFmpeg推流命令
    # 使用-loop 1实现循环推流，-t限制推流总时长
    cmd = [
        "ffmpeg",
        "-re",  # 以实时速率读取
        "-stream_loop", "-1",  # 循环播放
        "-i", video_path,
        "-c", "copy",  # 不重新编码，降低CPU占用
        "-f", "flv",
        "-t", str(duration_seconds),
        full_rtmp
    ]
    
    # 启动子进程
    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    print(f"[{datetime.now()}] 推流已启动，目标地址：{full_rtmp}")
    
    return process

def schedule_stream(video_path: str, rtmp_url: str, stream_key: str, start_time: str):
    """
    定时启动推流，配合系统任务调度器使用
    start_time格式："14:30"
    """
    # 此处仅为演示逻辑，实际部署需结合schedule库或系统cron
    print(f"已调度推流任务，预定启动时间：{start_time}")
    # 实际执行时需实现时间判断循环

import sqlite3
import json
from datetime import datetime

class LocalDataStore:
    def __init__(self, db_path: str = "live_data.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_table()
    
    def _init_table(self):
        cursor = self.conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS live_sessions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_date TEXT,
                start_time TEXT,
                end_time TEXT,
                duration_minutes INTEGER,
                peak_viewers INTEGER,
                avg_viewers INTEGER,
                total_messages INTEGER,
                api_source TEXT
            )
        """)
        self.conn.commit()
    
    def insert_session(self, data: dict):
        cursor = self.conn.cursor()
        cursor.execute("""
            INSERT INTO live_sessions (
                session_date, start_time, end_time, duration_minutes,
                peak_viewers, avg_viewers, total_messages, api_source
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            data.get('session_date'),
            data.get('start_time'),
            data.get('end_time'),
            data.get('duration_minutes'),
            data.get('peak_viewers'),
            data.get('avg_viewers'),
            data.get('total_messages'),
            data.get('api_source')
        ))
        self.conn.commit()
    
    def export_summary(self, days: int = 7):
        """导出最近N天的汇总统计"""
        cursor = self.conn.cursor()
        cursor.execute("""
            SELECT 
                COUNT(*) as session_count,
                AVG(peak_viewers) as avg_peak,
                SUM(total_messages) as total_interactions
            FROM live_sessions
            WHERE session_date >= date('now', ?)
        """, (f'-{days} days',))
        return cursor.fetchone()

# 增加以下参数
"-reconnect", "1",
"-reconnect_at_eof", "1",
"-reconnect_streamed", "1",
"-reconnect_delay_max", "5"

# 在标点符号后插入停顿标记
def insert_ssml_breaks(text: str) -> str:
    text = text.replace("。", '。<break time="800ms"/>')
    text = text.replace("？", '？<break time="700ms"/>')
    text = text.replace("！", '！<break time="700ms"/>')
    text = text.replace("，", '，<break time="300ms"/>')
    return text

# 使用FFmpeg拼接多个片段
# 首先创建文件列表 list.txt
# file segment1.mp4
# file segment2.mp4
# 然后执行拼接
# ffmpeg -f concat -safe 0 -i list.txt -c copy output.mp4