zhanting_AI_MCP/calculator.py

# server.py
from mcp.server.fastmcp import FastMCP
import sys
import logging
import os
import threading
import time
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')
CONFIG_PATH = os.path.join(os.path.dirname(__file__), 'config.yaml')

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

def _load_config(config_path: str) -> Dict[str, Any]:
    if yaml is None:
        return {}
    if not os.path.exists(config_path):
        return {}
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f) or {}
    except Exception as e:
        logger.error(f"读取配置文件失败 {config_path}: {e}")
        return {}

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 一致）
_app_config = _load_config(CONFIG_PATH)
FLASK_BASE = _app_config.get('flask_api_base', os.environ.get('FLASK_API_BASE', 'http://192.168.254.242:5050'))

# 登录配置
ADMIN_USERNAME = 'admin'
ADMIN_PASSWORD = 'HNYZ0821'
_session = requests.Session()
_has_logged_in = False

def _ensure_login():
    global _has_logged_in
    if _has_logged_in:
        return
        
    # 尝试常用登录路径 (先尝试 /auth/login, 再尝试 /login)
    for login_path in ['/auth/login', '/login']:
        url = f"{FLASK_BASE}{login_path}"
        try:
            # 只有当响应不是 404 时才认为是有效的登录端点
            resp = _session.post(url, data={
                'username': ADMIN_USERNAME,
                'password': ADMIN_PASSWORD
            }, timeout=5)
            
            if resp.status_code != 404:
                resp.raise_for_status()
                logger.info(f"登录成功: {url}")
                _has_logged_in = True
                return
        except Exception as e:
            logger.warning(f"尝试登录 {url} 失败: {e}")
            
    logger.error("无法登录 Flask API，后续请求可能会失败")

def _flask_get(path: str) -> Dict[str, Any]:
    _ensure_login()
    url = f"{FLASK_BASE}{path}"
    try:
        resp = _session.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]:
    _ensure_login()
    url = f"{FLASK_BASE}{path}"
    try:
        resp = _session.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

@mcp.tool(name="设置电视播放视频音量全局音量")
def set_global_volume(volume: int) -> dict:
    """设置电视播放视频的全局音量 (转发到 Flask: POST /api/kodi/set_volume)。

    使用场景：
    - 在播放产品视频或公司宣传视频时候，播放声音太小或太大就可以通过这个来控制。

    参数:
    - volume: 音量值，整数 0-100
    """
    # 参数校验
    if not isinstance(volume, int):
        try:
            volume = int(volume)
        except (ValueError, TypeError):
            return {"success": False, "message": "音量值必须为整数"}
            
    if volume < 0 or volume > 100:
        return {"success": False, "message": "音量值必须在 0-100 之间"}

    logger.info(f"请求设置全局音量: {volume}")
    return _flask_post('/api/kodi/set_volume', {"volume": volume})

@mcp.tool(name="打开办公楼大门")
def open_door(door_id: int = 1) -> dict:
    """打开办公楼大门 (转发到 Flask: POST /api/door/open)。

    使用场景：
    - 当用户说"打开大门"、"帮我开门"时调用此工具。
    - 目前只控制一个门，默认 door_id 为 1。

    参数:
    - door_id: 门ID，整数，默认为 1
    """
    if not isinstance(door_id, int):
         return {"success": False, "message": "door_id 必须为整数"}

    logger.info(f"请求打开大门: door_id={door_id}")
    return _flask_post('/api/door/open', {"door_id": door_id})

@mcp.tool(name="设置办公楼大门模式")
def set_door_mode(control_way: int) -> dict:
    """设置办公楼大门模式 (转发到 Flask: POST /api/door/control)。

    参数:
    - control_way: 模式，整数 (0:正常模式, 1:常开模式, 2:常闭模式)
    """
    if not isinstance(control_way, int):
        return {"success": False, "message": "control_way 必须为整数"}
    
    if control_way not in [0, 1, 2]:
        return {"success": False, "message": "control_way 必须为 0, 1 或 2"}

    logger.info(f"请求设置大门模式: control_way={control_way}")
    return _flask_post('/api/door/control', {"control_way": control_way})

@mcp.tool(name="打开指定电视")
def turn_on_tv(kodi_id: int) -> dict:
    """打开指定 ID 的电视 (转发到 Flask: POST /api/mitv/turn_on)。

    参数:
    - kodi_id: 整数 (必填) 对应 Kodi 客户端的 ID/索引

    使用说明：
    - 展厅共有 6 台电视，顺序从左到右。
    - kodi_id 从 0 开始计数：
      * "第一台"、"1号电视"、"左边第一台" -> kodi_id=0
      * "第二台"、"2号电视" -> kodi_id=1
      * ...
      * "第六台"、"6号电视"、"最右边那台" -> kodi_id=5
    """
    if not isinstance(kodi_id, int):
        return {"success": False, "message": "kodi_id 必须为整数"}

    logger.info(f"请求打开电视: kodi_id={kodi_id}")
    return _flask_post('/api/mitv/turn_on', {"kodi_id": kodi_id})

@mcp.tool(name="关闭指定电视")
def turn_off_tv(kodi_id: int) -> dict:
    """关闭指定 ID 的电视 (转发到 Flask: POST /api/mitv/turn_off)。

    参数:
    - kodi_id: 整数 (必填) 对应 Kodi 客户端的 ID/索引

    使用说明：
    - 展厅共有 6 台电视，顺序从左到右。
    - kodi_id 从 0 开始计数：
      * "第一台"、"1号电视"、"左边第一台" -> kodi_id=0
      * "第二台"、"2号电视" -> kodi_id=1
      * ...
      * "第六台"、"6号电视"、"最右边那台" -> kodi_id=5
    """
    if not isinstance(kodi_id, int):
        return {"success": False, "message": "kodi_id 必须为整数"}

    logger.info(f"请求关闭电视: kodi_id={kodi_id}")
    return _flask_post('/api/mitv/turn_off', {"kodi_id": kodi_id})

@mcp.tool(name="打开所有电视")
def turn_on_all_tvs() -> dict:
    """打开所有电视 (转发到 Flask: POST /api/mitv/turn_on_all)。

    使用场景：
    - 当用户明确提到"所有"、"全部"时使用，如"打开所有电视"。
    - 如果用户仅说"开电视"、"帮我把电视打开"，未明确指明是"所有"还是"某一台"，请不要直接调用本工具，而是先向用户确认："请问是打开所有电视，还是指定哪一台？"
    """
    logger.info("请求打开所有电视")
    return _flask_post('/api/mitv/turn_on_all')

@mcp.tool(name="关闭所有电视")
def turn_off_all_tvs() -> dict:
    """关闭所有电视 (转发到 Flask: POST /api/mitv/turn_off_all)。

    使用场景：
    - 当用户明确提到"所有"、"全部"时使用，如"关闭所有电视"。
    - 用户说"关闭电视电源"通常也理解为关闭所有电视。
    - 如果用户仅说"关电视"、"把电视关了"，未明确指明是"所有"还是"某一台"，请不要直接调用本工具，而是先向用户确认："请问是关闭所有电视，还是指定哪一台？"
    """
    logger.info("请求关闭所有电视")
    return _flask_post('/api/mitv/turn_off_all')

# 灯光控制相关工具
@mcp.tool(name="控制一楼大门玄关顶灯")
def control_entrance_lights(action: str) -> dict:
    """控制一楼大门玄关顶灯 (转发到 Flask: POST /api/ha/entrance_lights/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制玄关顶灯: {action}")
    return _flask_post(f'/api/ha/entrance_lights/{action}')

@mcp.tool(name="控制一楼大门玄关射灯")
def control_exhibition_spotlight(action: str) -> dict:
    """控制一楼大门玄关射灯 (转发到 Flask: POST /api/ha/exhibition_spotlight/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制玄关射灯: {action}")
    return _flask_post(f'/api/ha/exhibition_spotlight/{action}')

@mcp.tool(name="控制一楼展厅顶灯")
def control_exhibition_ceiling_light(action: str) -> dict:
    """控制一楼展厅顶灯 (转发到 Flask: POST /api/ha/exhibition_ceiling_light/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制展厅顶灯: {action}")
    return _flask_post(f'/api/ha/exhibition_ceiling_light/{action}')

@mcp.tool(name="控制展厅桌面灯座总开关")
def control_exhibition_desktop_switch(action: str) -> dict:
    """控制展厅桌面灯座总开关 (转发到 Flask: POST /api/ha/exhibition_desktop_switch/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制桌面灯座总开关: {action}")
    return _flask_post(f'/api/ha/exhibition_desktop_switch/{action}')

@mcp.tool(name="控制展厅桌面3D风扇投影")
def control_exhibition_3d_fan(action: str) -> dict:
    """控制展厅桌面3D风扇投影 (转发到 Flask: POST /api/ha/exhibition_3d_fan/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制3D风扇投影: {action}")
    return _flask_post(f'/api/ha/exhibition_3d_fan/{action}')

@mcp.tool(name="控制展台桌子灯带")
def control_exhibition_stand_light_strip(action: str) -> dict:
    """控制展台桌子灯带 (转发到 Flask: POST /api/ha/exhibition_stand_light_strip/turn_on|off)。

    参数:
    - action: "turn_on" (打开) 或 "turn_off" (关闭)
    """
    if action not in ["turn_on", "turn_off"]:
        return {"success": False, "message": "action 必须为 'turn_on' 或 'turn_off'"}
    
    logger.info(f"请求控制展台灯带: {action}")
    return _flask_post(f'/api/ha/exhibition_stand_light_strip/{action}')

@mcp.tool(name="重启展厅所有设备")
def restart_all_devices() -> dict:
    """重启展厅所有设备。

    使用场景：
    - 当用户反馈设备故障、卡顿、无反应时调用。
    - 例如用户说："电视播放卡住了"、"灯光控制没反应"、"设备出问题了"、"系统好像死机了"。
    - 此时建议主动询问用户："检测到设备可能出现异常，建议重启展厅所有设备进行恢复，是否现在执行重启？"
    
    功能：
    - 依次关闭所有电视、桌面3D风扇投影、展品灯座。
    - 等待 1 分钟。
    - 依次重新启动上述设备。
    - 此操作异步执行。
    """
    def _restart_process():
        try:
            logger.info("【重启任务】开始执行...")
            
            # 1. 关闭设备
            logger.info("【重启任务】正在关闭设备...")
            # 关闭所有电视
            _flask_post('/api/mitv/turn_off_all')
            # 关闭桌面3D风扇投影
            _flask_post('/api/ha/exhibition_3d_fan/turn_off')
            # 关闭展品灯座
            _flask_post('/api/ha/exhibition_desktop_switch/turn_off')
            
            # 2. 等待
            logger.info("【重启任务】设备已关闭，等待 60 秒...")
            time.sleep(60)
            
            # 3. 启动设备
            logger.info("【重启任务】正在启动设备...")
            # 启动所有电视
            _flask_post('/api/mitv/turn_on_all')
            # 启动桌面3D风扇投影
            _flask_post('/api/ha/exhibition_3d_fan/turn_on')
            # 启动展品灯座
            _flask_post('/api/ha/exhibition_desktop_switch/turn_on')
            # 启动视频播放
            _flask_post('/api/kodi/free_time/control', {"action": "start"})
            
            logger.info("【重启任务】执行完成")
        except Exception as e:
            logger.error(f"【重启任务】发生异常: {e}")

    # 启动后台线程
    t = threading.Thread(target=_restart_process)
    t.daemon = True
    t.start()
    
    return {"success": True, "message": "已开始重启展厅所有设备，全程约需 1 分钟，请耐心等待。"}

@mcp.tool(name="启动迎宾模式")
def start_welcome_mode() -> dict:
    """启动迎宾模式。

    使用场景：
    - 当用户明确要求"启动迎宾模式"时调用。
    - 如果用户提到"有客人要来"、"准备接待客人"、"开启接待模式"等含义，请先主动询问用户："是否需要为您开启迎宾模式？"
    
    功能（异步执行）：
    1. 打开一楼大门玄关顶灯
    2. 打开一楼大门玄关射灯
    3. 打开一楼展厅顶灯
    4. 打开展台桌子灯带
    5. 把大门模式设置为常开模式
    """
    def _process():
        try:
            logger.info("【迎宾模式】正在启动...")
            # 1.打开一楼大门玄关顶灯
            _flask_post('/api/ha/entrance_lights/turn_on')
            # 2.打开一楼大门玄关射灯
            _flask_post('/api/ha/exhibition_spotlight/turn_on')
            # 3.打开一楼展厅顶灯
            _flask_post('/api/ha/exhibition_ceiling_light/turn_on')
            # 6.打开展台桌子灯带
            _flask_post('/api/ha/exhibition_stand_light_strip/turn_on')
            # 7.把大门模式设置为常开模式 (1:常开)
            _flask_post('/api/door/control', {"control_way": 1})
            logger.info("【迎宾模式】启动完成")
        except Exception as e:
            logger.error(f"【迎宾模式】启动失败: {e}")

    t = threading.Thread(target=_process)
    t.daemon = True
    t.start()
    return {"success": True, "message": "正在启动迎宾模式..."}

@mcp.tool(name="关闭迎宾模式")
def stop_welcome_mode() -> dict:
    """关闭迎宾模式。
    
    功能（异步执行）：
    1. 关闭一楼大门玄关顶灯
    2. 关闭一楼大门玄关射灯
    3. 关闭一楼展厅顶灯
    4. 关闭展台桌子灯带
    5. 把大门模式设置为正常模式
    """
    def _process():
        try:
            logger.info("【迎宾模式】正在关闭...")
            # 1.关闭一楼大门玄关顶灯
            _flask_post('/api/ha/entrance_lights/turn_off')
            # 2.关闭一楼大门玄关射灯
            _flask_post('/api/ha/exhibition_spotlight/turn_off')
            # 3.关闭一楼展厅顶灯
            _flask_post('/api/ha/exhibition_ceiling_light/turn_off')
            # 6.关闭展台桌子灯带
            _flask_post('/api/ha/exhibition_stand_light_strip/turn_off')
            # 7.把大门模式设置为正常模式 (0:正常)
            _flask_post('/api/door/control', {"control_way": 0})
            logger.info("【迎宾模式】关闭完成")
        except Exception as e:
            logger.error(f"【迎宾模式】关闭失败: {e}")

    t = threading.Thread(target=_process)
    t.daemon = True
    t.start()
    return {"success": True, "message": "正在关闭迎宾模式..."}

@mcp.tool(name="控制展品以及宣传视频自动播放")
def control_free_time_playback(action: str) -> dict:
    """控制展品以及宣传视频自动播放功能 (转发到 Flask: POST /api/kodi/free_time/control)。
    
    功能别名：继续电视播放、开启/关闭公司宣传视频播放。

    使用场景：
    - 当用户说"继续播放电视"、"电视继续播放"时，通常意味着恢复默认的宣传视频循环播放，请调用 action="start"。
    - 当用户说"播放宣传视频"、"开启公司宣传视频"、"开启自动播放"时，调用 action="start"。
    - 当用户说"停止自动播放"、"关闭宣传视频"、"不要播放宣传片了"时，调用 action="stop"。

    参数:
    - action: "start" (开启/强制开启) 或 "stop" (停止)

    功能说明:
    - 开启后，如果在 07:30-18:00 时间段内，会自动循环播放视频 ID 0 (通常为公司宣传片)。
    - 手动发送 "start" 会立即强制开启播放，无论是否在时间段内。
    - 发送 "stop" 会立即强制停止播放。
    """
    if action not in ["start", "stop"]:
        return {"success": False, "message": "action 必须为 'start' 或 'stop'"}

    logger.info(f"请求控制展品以及宣传视频自动播放: {action}")
    return _flask_post('/api/kodi/free_time/control', {"action": action})

@mcp.tool(name="获取展品以及宣传视频自动播放状态")
def get_free_time_playback_status() -> dict:
    """获取展品以及宣传视频自动播放功能的状态 (转发到 Flask: GET /api/kodi/free_time/status)。

    返回包含:
    - is_running: 功能开关状态
    - enabled: 同 is_running
    - is_thread_alive: 底层线程存活状态
    """
    logger.info("请求获取展品以及宣传视频自动播放状态")
    return _flask_get('/api/kodi/free_time/status')

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