单人多平台数字人直播矩阵部署：从账号配置到跨平台内容分发的技术实现

{
  "platforms": [
    {
      "id": "douyin",
      "name": "抖音",
      "rtmp_url": "rtmp://push.douyin.com/live",
      "stream_key": "${DOUYIN_STREAM_KEY}",
      "content_spec": {
        "video_aspect": "9:16",
        "resolution": "1080x1920",
        "max_duration": 60,
        "title_max_length": 30,
        "preferred_tags": ["职场", "法律", "科普"]
      },
      "publish_schedule": {
        "days": ["Mon", "Wed", "Fri"],
        "time": "19:00"
      },
      "enabled": true
    },
    {
      "id": "wechat_video",
      "name": "视频号",
      "rtmp_url": "rtmp://push.weixin.qq.com/live",
      "stream_key": "${WECHAT_STREAM_KEY}",
      "content_spec": {
        "video_aspect": "16:9",
        "resolution": "1280x720",
        "max_duration": 120,
        "title_max_length": 40,
        "preferred_tags": ["法律知识", "职场权益"]
      },
      "publish_schedule": {
        "days": ["Tue", "Thu", "Sat"],
        "time": "20:00"
      },
      "enabled": true
    }
  ]
}

import json
import os
from typing import Dict, List, Optional

class PlatformManager:
    def __init__(self, config_path: str):
        self.config_path = config_path
        self.platforms = self._load_config()
        self._expand_env_vars()

    def _load_config(self) -> List[Dict]:
        with open(self.config_path, 'r', encoding='utf-8') as f:
            config = json.load(f)
        return config.get("platforms", [])

    def _expand_env_vars(self):
        """替换环境变量占位符"""
        for platform in self.platforms:
            if "stream_key" in platform:
                key = platform["stream_key"]
                if key.startswith("${") and key.endswith("}"):
                    env_var = key[2:-1]
                    platform["stream_key"] = os.getenv(env_var, "")

    def get_platform(self, platform_id: str) -> Optional[Dict]:
        for p in self.platforms:
            if p["id"] == platform_id:
                return p
        return None

    def get_enabled_platforms(self) -> List[Dict]:
        return [p for p in self.platforms if p.get("enabled", True)]

    def get_publish_schedule(self, platform_id: str) -> Dict:
        platform = self.get_platform(platform_id)
        return platform.get("publish_schedule", {}) if platform else {}

    def update_stream_key(self, platform_id: str, new_key: str):
        """安全更新推流密钥"""
        for platform in self.platforms:
            if platform["id"] == platform_id:
                platform["stream_key"] = new_key
                self._save_config()
                break

    def _save_config(self):
        with open(self.config_path, 'w', encoding='utf-8') as f:
            json.dump({"platforms": self.platforms}, f, ensure_ascii=False, indent=2)

class ContentAdapter:
    """多平台内容适配器"""

    def __init__(self, platform_configs: List[Dict]):
        self.configs = {p["id"]: p["content_spec"] for p in platform_configs}

    def adapt_video(self, input_path: str, platform_id: str, output_path: str):
        """将通用视频适配到指定平台规格"""
        spec = self.configs.get(platform_id, {})
        aspect = spec.get("video_aspect", "9:16")
        resolution = spec.get("resolution", "1080x1920")
        max_duration = spec.get("max_duration", 60)

        # 构建FFmpeg适配命令
        cmd = self._build_adapt_cmd(input_path, output_path, aspect, resolution, max_duration)
        return cmd

    def _build_adapt_cmd(self, input_path: str, output_path: str,
                         aspect: str, resolution: str, max_duration: int) -> list:
        """构建FFmpeg适配命令"""
        # 根据比例计算裁剪参数
        if aspect == "9:16":
            # 竖版：裁剪为9:16
            filter_cmd = "scale=1080:1920:force_original_aspect_ratio=decrease," \
                         "pad=1080:1920:(ow-iw)/2:(oh-ih)/2"
        else:
            # 横版：裁剪为16:9
            filter_cmd = "scale=1280:720:force_original_aspect_ratio=decrease," \
                         "pad=1280:720:(ow-iw)/2:(oh-ih)/2"

        cmd = [
            "ffmpeg",
            "-i", input_path,
            "-vf", filter_cmd,
            "-c:v", "libx264",
            "-b:v", "2000k",
            "-t", str(max_duration),
            "-y",  # 覆盖输出文件
            output_path
        ]
        return cmd

    def adapt_title(self, title: str, platform_id: str) -> str:
        """适配标题长度"""
        spec = self.configs.get(platform_id, {})
        max_len = spec.get("title_max_length", 30)
        if len(title) > max_len:
            return title[:max_len]
        return title

    def adapt_tags(self, tags: List[str], platform_id: str) -> List[str]:
        """适配标签策略"""
        spec = self.configs.get(platform_id, {})
        preferred = spec.get("preferred_tags", [])
        # 保留部分通用标签，优先使用平台偏好标签
        adapted = [t for t in tags if t in preferred]
        if len(adapted) < 3:
            # 补充平台偏好的标签
            adapted.extend(preferred[:3 - len(adapted)])
        return adapted[:5]  # 最多5个标签

import subprocess
import os
from concurrent.futures import ThreadPoolExecutor, as_completed

def batch_adapt_videos(input_dir: str, output_dir: str, platform_manager: PlatformManager):
    """批量适配视频到各平台格式"""
    platforms = platform_manager.get_enabled_platforms()
    adapter = ContentAdapter([p for p in platforms])

    tasks = []
    for platform in platforms:
        platform_id = platform["id"]
        platform_output_dir = os.path.join(output_dir, platform_id)
        os.makedirs(platform_output_dir, exist_ok=True)

        # 获取输入视频列表
        input_files = [f for f in os.listdir(input_dir) if f.endswith('.mp4')]

        for input_file in input_files:
            input_path = os.path.join(input_dir, input_file)
            output_path = os.path.join(platform_output_dir, input_file)
            cmd = adapter.adapt_video(input_path, platform_id, output_path)
            tasks.append((cmd, platform_id, input_file))

    # 串行执行避免资源竞争
    for cmd, platform_id, input_file in tasks:
        print(f"适配：{input_file} → {platform_id}")
        result = subprocess.run(cmd, capture_output=True)
        if result.returncode != 0:
            print(f"适配失败：{input_file}，错误：{result.stderr.decode()}")

from datetime import datetime, timedelta
import random

class DistributionScheduler:
    def __init__(self, platform_manager: PlatformManager):
        self.platform_manager = platform_manager
        self.published_content = {}  # 记录已发布内容，用于去重

    def generate_schedule(self, content_pool: List[Dict], days: int = 7) -> Dict:
        """生成多平台内容排期"""
        schedule = {}
        platforms = self.platform_manager.get_enabled_platforms()

        for day in range(days):
            date = (datetime.now() + timedelta(days=day)).strftime("%Y-%m-%d")
            schedule[date] = {}

            for platform in platforms:
                platform_id = platform["id"]
                # 获取该平台的发布日安排
                pub_schedule = platform.get("publish_schedule", {})
                pub_days = pub_schedule.get("days", [])
                pub_time = pub_schedule.get("time", "19:00")

                # 检查今天是否需要发布
                today_weekday = (datetime.now() + timedelta(days=day)).strftime("%a")
                if today_weekday not in pub_days:
                    continue

                # 从内容池中选择尚未在相同平台发布的内容
                available = [c for c in content_pool
                             if c["id"] not in self.published_content.get(platform_id, [])]

                if available:
                    selected = random.choice(available)
                    schedule[date][platform_id] = {
                        "content_id": selected["id"],
                        "title": selected.get("title", ""),
                        "publish_time": pub_time,
                        "status": "scheduled"
                    }
                    # 记录已排期
                    self.published_content.setdefault(platform_id, []).append(selected["id"])

        return schedule

    def get_today_tasks(self, schedule: Dict) -> List[Dict]:
        """获取今日待发布任务"""
        today = datetime.now().strftime("%Y-%m-%d")
        day_schedule = schedule.get(today, {})
        tasks = []

        for platform_id, task in day_schedule.items():
            platform = self.platform_manager.get_platform(platform_id)
            if platform:
                tasks.append({
                    "platform": platform_id,
                    "platform_name": platform.get("name", platform_id),
                    "content_id": task["content_id"],
                    "title": task["title"],
                    "publish_time": task["publish_time"],
                    "rtmp_url": platform.get("rtmp_url", ""),
                    "stream_key": platform.get("stream_key", "")
                })

        return tasks

import sqlite3
from datetime import datetime

class DistributionTracker:
    def __init__(self, db_path: str = "distribution.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_tables()

    def _init_tables(self):
        cursor = self.conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS distribution_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                platform_id TEXT,
                content_id TEXT,
                title TEXT,
                scheduled_time TEXT,
                actual_time TEXT,
                status TEXT,
                view_count INTEGER DEFAULT 0,
                like_count INTEGER DEFAULT 0,
                comment_count INTEGER DEFAULT 0
            )
        """)
        self.conn.commit()

    def log_publish(self, task: Dict, status: str = "published"):
        """记录发布状态"""
        cursor = self.conn.cursor()
        cursor.execute("""
            INSERT INTO distribution_log
            (platform_id, content_id, title, scheduled_time, actual_time, status)
            VALUES (?, ?, ?, ?, ?, ?)
        """, (
            task["platform"],
            task["content_id"],
            task["title"],
            task.get("scheduled_time", datetime.now().isoformat()),
            datetime.now().isoformat(),
            status
        ))
        self.conn.commit()

    def get_platform_stats(self, days: int = 7) -> Dict:
        """获取各平台近N天统计"""
        cursor = self.conn.cursor()
        cursor.execute("""
            SELECT platform_id,
                   COUNT(*) as total_published,
                   AVG(view_count) as avg_views,
                   SUM(like_count) as total_likes
            FROM distribution_log
            WHERE actual_time >= datetime('now', ?)
            AND status = 'published'
            GROUP BY platform_id
        """, (f'-{days} days',))

        results = {}
        for row in cursor.fetchall():
            results[row[0]] = {
                "total_published": row[1],
                "avg_views": row[2] or 0,
                "total_likes": row[3] or 0
            }
        return results

    def get_content_reach(self, content_id: str) -> Dict:
        """获取单条内容在各平台的覆盖数据"""
        cursor = self.conn.cursor()
        cursor.execute("""
            SELECT platform_id, view_count, like_count, comment_count
            FROM distribution_log
            WHERE content_id = ?
            AND status = 'published'
        """, (content_id,))

        results = {}
        for row in cursor.fetchall():
            results[row[0]] = {
                "views": row[1] or 0,
                "likes": row[2] or 0,
                "comments": row[3] or 0
            }
        return results