数字人直播技术在传统产业中的应用场景调研：助农、文旅、专业服务三类领域的部署实践

{
  "products": [
    {
      "name": "产品名称",
      "origin": "产地",
      "story": "种植故事或农户故事",
      "season": "上市季节",
      "features": ["特点1", "特点2"]
    }
  ]
}

{
  "attractions": [
    {
      "name": "景点名称",
      "location": "地理位置",
      "history": "历史背景",
      "highlights": ["亮点1", "亮点2"],
      "tips": ["游玩建议1", "游玩建议2"]
    }
  ]
}

{
  "faqs": [
    {
      "category": "分类",
      "question": "常见问题",
      "answer": "专业解答",
      "disclaimer": "免责声明"
    }
  ]
}

import json
import os
from typing import List, Dict

class KnowledgeBaseManager:
    def __init__(self, base_path: str):
        self.base_path = base_path
        self.industry_type = self._detect_industry()
    
    def _detect_industry(self) -> str:
        """根据文件结构自动识别行业类型"""
        if os.path.exists(os.path.join(self.base_path, "products.json")):
            return "agriculture"
        elif os.path.exists(os.path.join(self.base_path, "attractions.json")):
            return "tourism"
        elif os.path.exists(os.path.join(self.base_path, "faqs.json")):
            return "professional_service"
        return "general"
    
    def load_knowledge(self) -> Dict:
        """加载对应行业的知识库"""
        file_map = {
            "agriculture": "products.json",
            "tourism": "attractions.json",
            "professional_service": "faqs.json"
        }
        filename = file_map.get(self.industry_type, "general.json")
        filepath = os.path.join(self.base_path, filename)
        
        with open(filepath, 'r', encoding='utf-8') as f:
            return json.load(f)
    
    def get_script_template(self) -> str:
        """根据行业类型返回对应的Prompt模板"""
        templates = {
            "agriculture": """
                你是一名农产品推广内容编辑。请根据以下产品信息生成一个60秒短视频脚本：
                产品名称：{name}
                产地：{origin}
                产品故事：{story}
                特点：{features}
                要求：开场用一句反常识提问吸引注意，核心部分讲述产品故事，结尾引导关注。
            """,
            "tourism": """
                你是一名旅游推荐内容编辑。请根据以下景点信息生成一个60秒短视频脚本：
                景点名称：{name}
                位置：{location}
                历史背景：{history}
                亮点：{highlights}
                游玩建议：{tips}
                要求：开场用一句悬念提问，核心部分讲述景点独特之处，结尾推荐游览时间。
            """,
            "professional_service": """
                你是一名专业领域科普内容编辑。请根据以下问答生成一个60秒短视频脚本：
                问题：{question}
                专业解答：{answer}
                要求：开场用一句反问句，核心部分用通俗语言解释，结尾必须包含免责声明。
            """
        }
        return templates.get(self.industry_type, templates["professional_service"])

import json
from openai import OpenAI

client = OpenAI(api_key="your_api_key")

def generate_scene_script(knowledge_item: Dict, industry_type: str) -> Dict:
    """根据行业类型和知识条目生成脚本"""
    
    prompt_templates = {
        "agriculture": f"""
            请根据以下农产品信息生成一个60秒短视频脚本：
            产品名称：{knowledge_item.get('name', '')}
            产地：{knowledge_item.get('origin', '')}
            种植故事：{knowledge_item.get('story', '')}
            产品特点：{', '.join(knowledge_item.get('features', []))}
            
            要求：
            1. 输出JSON格式：{{"opening": "开场", "core": "核心内容", "closing": "结尾"}}
            2. 开场需使用反常识提问或数据对比
            3. 核心内容重点讲述产地故事和产品特色
            4. 结尾引导用户关注或下单
        """,
        
        "tourism": f"""
            请根据以下景点信息生成一个60秒短视频脚本：
            景点名称：{knowledge_item.get('name', '')}
            地理位置：{knowledge_item.get('location', '')}
            历史背景：{knowledge_item.get('history', '')}
            游览亮点：{', '.join(knowledge_item.get('highlights', []))}
            
            要求：
            1. 输出JSON格式：{{"opening": "开场", "core": "核心内容", "closing": "结尾"}}
            2. 开场用悬念式提问
            3. 核心内容讲述景点的独特文化或自然价值
            4. 结尾提供游览建议
        """,
        
        "professional_service": f"""
            请根据以下法律问答生成一个60秒科普短视频脚本：
            问题：{knowledge_item.get('question', '')}
            专业解答：{knowledge_item.get('answer', '')}
            
            要求：
            1. 输出JSON格式：{{"opening": "开场", "core": "核心内容", "closing": "结尾"}}
            2. 开场用一句反问句吸引注意力
            3. 核心内容用通俗语言解释，避免专业术语堆砌
            4. 结尾必须包含：本内容仅供参考，不构成法律意见
        """
    }
    
    prompt = prompt_templates.get(industry_type, prompt_templates["professional_service"])
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0.75
    )
    
    script = json.loads(response.choices[0].message.content)
    return script

class AssetManager:
    def __init__(self, asset_path: str):
        self.asset_path = asset_path
    
    def get_background_video(self, industry_type: str) -> str:
        """根据行业类型获取默认背景视频"""
        bg_map = {
            "agriculture": "farmland_background.mp4",
            "tourism": "landscape_background.mp4",
            "professional_service": "office_background.mp4"
        }
        filename = bg_map.get(industry_type, "default_background.mp4")
        return os.path.join(self.asset_path, filename)
    
    def get_scene_images(self, knowledge_item: Dict) -> List[str]:
        """根据知识条目获取相关图片素材"""
        # 实际实现中通过文件名匹配或标签检索
        # 此处仅为示意逻辑
        image_dir = os.path.join(self.asset_path, knowledge_item.get("category", "general"))
        if os.path.exists(image_dir):
            return [os.path.join(image_dir, f) for f in os.listdir(image_dir) if f.endswith(('.jpg', '.png'))]
        return []

import schedule
import time
from datetime import datetime
import subprocess

class MultiSceneScheduler:
    def __init__(self, config_path: str):
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = json.load(f)
    
    def rotate_scenes(self):
        """轮播不同行业的直播内容"""
        scenes = self.config.get("scenes", [])
        current_index = 0
        
        while True:
            scene = scenes[current_index % len(scenes)]
            self.push_scene(scene)
            current_index += 1
            time.sleep(scene.get("duration", 1800))  # 默认30分钟
    
    def push_scene(self, scene: Dict):
        """推送单个场景内容"""
        video_path = scene["video_path"]
        rtmp_url = scene["rtmp_url"]
        stream_key = scene["stream_key"]
        
        cmd = [
            "ffmpeg",
            "-re",
            "-stream_loop", "-1",
            "-i", video_path,
            "-c", "copy",
            "-f", "flv",
            "-t", str(scene.get("duration", 1800)),
            f"{rtmp_url}/{stream_key}"
        ]
        
        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        print(f"[{datetime.now()}] 已推送场景：{scene.get('name')}")
        return process