zhanting_AI_MCP/calculator.py

# server.py
from mcp.server.fastmcp import FastMCP
import sys
import logging
import os
from typing import List, Dict, Any, Optional
import requests
from utils.logger_config import logger
try:
    import yaml
except ImportError:  # 允许在未安装 PyYAML 的环境下给出清晰日志
    yaml = None

# Fix UTF-8 encoding for Windows console
if sys.platform == 'win32':
    sys.stderr.reconfigure(encoding='utf-8')
    sys.stdout.reconfigure(encoding='utf-8')

# Create an MCP server
mcp = FastMCP("展厅交互模块")

# YAML 配置路径
VIDEO_CONFIG_PATH = os.path.join(os.path.dirname(__file__), 'video_config_prod.yaml')

# 内存状态
_is_running: bool = False
_current_video_id: Optional[int] = None

def _load_videos_from_yaml(config_path: str) -> List[Dict[str, Any]]:
    if yaml is None:
        logger.error("未安装 PyYAML，请在环境中安装 pyyaml 以读取配置文件")
        return []
    if not os.path.exists(config_path):
        logger.error(f"未找到配置文件: {config_path}")
        return []
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f) or {}
        videos = data.get('video_infos', [])
        if not isinstance(videos, list):
            logger.error("配置文件格式错误: 'video_infos' 应为列表")
            return []
        # 规范化字段
        normalized = []
        for v in videos:
            if not isinstance(v, dict):
                continue
            if 'id' not in v or 'name' not in v:
                continue
            normalized.append({
                'id': v.get('id'),
                'name': v.get('name'),
                'formula': v.get('formula'),
                'description': v.get('description'),
                'video_duration': v.get('video_duration'),
                'video_path': v.get('video_path'),
                'code': v.get('code'),
            })
        return normalized
    except Exception as e:
        logger.exception(f"读取配置失败: {e}")
        return []

def _index_videos(videos: List[Dict[str, Any]]) -> Dict[int, Dict[str, Any]]:
    return {int(v['id']): v for v in videos if isinstance(v.get('id'), int)}

# 启动时加载
_videos: List[Dict[str, Any]] = _load_videos_from_yaml(VIDEO_CONFIG_PATH)
_videos_by_id: Dict[int, Dict[str, Any]] = _index_videos(_videos)

# Flask API 基础地址（与 README_API.md 一致）
FLASK_BASE = os.environ.get('FLASK_API_BASE', 'http://192.168.254.242:5050')

def _flask_get(path: str) -> Dict[str, Any]:
    url = f"{FLASK_BASE}{path}"
    try:
        resp = requests.get(url, timeout=3)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        logger.error(f"GET {url} 失败: {e}")
        return {"success": False, "message": str(e)}

def _flask_post(path: str, json_body: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    url = f"{FLASK_BASE}{path}"
    try:
        resp = requests.post(url, json=json_body or {}, timeout=3)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        logger.error(f"POST {url} 失败: {e}")
        return {"success": False, "message": str(e)}

@mcp.tool(name="获取所有可用的视频列表")
def list_videos() -> dict:
    """获取所有可用的视频列表（来自 video_config.yaml）。

    使用建议：
    - 当用户只说出展品名称且可能存在谐音/近音时，先调用本工具获取候选列表，
      在客户端侧做相似度/拼音/编辑距离匹配，选取"最接近"的名称对应的 video_id，
      再调用 kodi_start(video_id)。
    
    返回的每个视频包含以下字段，可用于匹配：
    - id: 视频ID（整数）
    - name: 展品名称（如"氯化铷"、"硫酸钙新材料"等）
    - code: 展品编号（如"1号展品"、"2号展品"等）
    - formula: 化学式或公式
    - description: 描述信息
    """
    logger.info(f"获取视频列表，共 {_videos.__len__()} 个")
    return {"success": True, "videos": _videos, "count": len(_videos)}

@mcp.tool(name="播放到指定的展品视频")
def kodi_start(video_id: int) -> dict:
    """启动或切换 Kodi 播放到指定 video_id（转发到 Flask: POST /api/kodi/start）。

    参数：
    - video_id: 整数类型的视频ID（从 list_videos() 获取）

    使用场景和匹配策略：
    
    1. 用户说编号类表达（如"一号展品"、"1号产品"、"展品1"、"产品1"等）：
       - 提取数字部分（如"一号"→1、"1号"→1、"展品1"→1）
       - 直接使用该数字作为 video_id
    
    2. 用户直接说物质名称（如"播放氯化铷"、"播放硫酸钙"、"氯化铯介绍视频"等）：
       - 先调用 list_videos() 获取所有视频列表
       - 从返回的列表中匹配：
         a) 优先匹配 name 字段（精确或模糊匹配）
         b) 其次匹配 formula 字段
         c) 再次匹配 code 字段中的名称部分
       - 匹配时考虑：
         * 精确匹配（如"氯化铷"完全匹配）
         * 包含匹配（如输入"氯化铷"匹配到 name="氯化铷"）
         * 拼音/谐音匹配（如"氯化铯"与"氯化铷"的相似度）
         * 编辑距离（允许轻微口误）
       - 选择"最接近"的条目，使用其 id 字段作为 video_id 调用本接口
    
    3. 匹配示例：
       - "播放1号展品" → video_id=1
       - "播放一号产品" → video_id=1
       - "播放氯化铷" → 匹配 name="氯化铷" → video_id=1
       - "播放氯化铯介绍视频" → 匹配 name="氯化铯" → video_id=6
       - "硫酸钙" → 匹配 name="硫酸钙新材料"（包含匹配） → video_id=2
    
    提示：如果用户给的是展品"名称"而非 id，且存在谐音/口误，
    - 先调用 list_videos() 获取所有视频元数据
    - 在调用侧进行名称相似度匹配（可用拼音/编辑距离/分词等方法）
    - 选择"最接近"的条目并传入其 video_id 调用本接口
    """
    if not isinstance(video_id, int) or video_id < 0:
        return {"success": False, "message": "video_id 必须为大于等于 0 的整数"}
    if video_id not in _videos_by_id:
        return {"success": False, "message": f"未找到视频ID={video_id}"}
    video = _videos_by_id[video_id]
    logger.info(f"请求 Flask 启动/切换播放: 视频ID={video_id}，名称={video.get('name')}")
    api = _flask_post('/api/kodi/start', {"video_id": video_id})
    # 同步内存状态
    if api.get('success') is True:
        global _is_running, _current_video_id
        _is_running = True
        _current_video_id = video_id
    return api

# Start the server
if __name__ == "__main__":
    mcp.run(transport="stdio")