tv_show_v1/hardware/kodi_module.py

import json
from typing import Any, Dict
import requests
import time
import threading
import yaml  # Add yaml import
import os
import sys
from utils.logger_config import logger

class VideoInfo:
    def __init__(self, name, formula, description, video_duration, video_path, id=0):
        self.name = name
        self.formula = formula
        self.description = description
        self.video_duration = video_duration
        self.video_path = video_path
        self.id = id

class KodiClient:
    def __init__(self, host="localhost", port=8080, username=None, password=None, id=0):
        self.host = host
        self.port = port
        self.id = id
        self.url = f"http://{host}:{port}/jsonrpc"
        self.headers = {'Content-Type': 'application/json'}
        # 非独立的客户端默认播放展品视频，独立情况下可能会播放图片或监控
        self.isIndividual = False
        # 启动kodi的url
        self.my_app_name = "org.xbmc.kodi" # kodi的包名
        self.mitv_startapp_url = f"http://{host}:6095/controller?action=startapp&&type=packagename&packagename={self.my_app_name}"
        
        if username and password:
            self.auth = (username, password)
        else:
            self.auth = None
        self.ready_event = threading.Event()

    def _send_request(self, method, params=None):
        data = {
            "jsonrpc": "2.0",
            "method": method,
            "id": 1,
        }
        if params:
            data["params"] = params
        response = requests.post(
            self.url,
            data=json.dumps(data),
            headers=self.headers,
            auth=self.auth
        )
        return response.json()

    def _send_request_async(self, method, params=None):
        """在后台线程中发送JSON-RPC请求，立即返回，不阻塞当前调用。"""
        def _worker():
            try:
                resp = self._send_request(method, params)
                logger.debug(f"[async] {method} 响应: {resp}")
            except Exception as e:
                logger.warning(f"[async] {method} 调用异常: {e}")
        t = threading.Thread(target=_worker, daemon=True)
        t.start()
        return True

    # 播放url图片
    def play_url_image(self, image_url):
        """播放指定url的图片文件（异步派发）。"""
        params = {"item": {"file": image_url}}
        self._send_request_async("Player.Open", params)
        logger.info(f"[async] Player.Open(url image) 已派发: {image_url}")
        return {"queued": True, "file": image_url}

    # 播放rtsp视频流
    def play_rtsp_video(self, rtsp_url):
        """播放指定rtsp视频流（异步派发）。"""
        params = {"item": {"file": rtsp_url}}
        self._send_request_async("Player.Open", params)
        logger.info(f"[async] Player.Open(rtsp) 已派发: {rtsp_url}")
        return {"queued": True, "file": rtsp_url}

    def play_video(self, video_path, loop=False):
        """播放指定路径的视频文件（异步派发），loop为是否循环播放"""
        params = {"item": {"file": video_path}}
        
        def _play_video_worker():
            """异步工作线程：播放视频并根据loop参数设置循环"""
            try:
                response = self._send_request("Player.Open", params)
                logger.info(f"[async] Player.Open(video) 响应: {response}")
                
                if loop:
                    # 如果需要循环播放，等待视频加载后设置循环
                    if response and response.get('result') == 'OK':
                        logger.info("[async] Player.Open 调用成功，等待视频加载后设置循环...")
                        # 等待视频加载 (根据观察到的 5 秒加载时间设置)
                        wait_time = 5.5
                        logger.info(f"[async] 等待 {wait_time} 秒让视频加载...")
                        time.sleep(wait_time)
                        # 尝试直接使用 playerid: 1，并验证其类型
                        player_id_to_try = 1
                        try:
                            properties_to_get = ["speed", "type"]
                            get_props_params = {"playerid": player_id_to_try, "properties": properties_to_get}
                            props_response = self._send_request("Player.GetProperties", get_props_params)
                            logger.info(f"[async] Player.GetProperties (ID: {player_id_to_try}) 响应: {props_response}")

                            if props_response and props_response.get('result') and props_response['result'].get('type') == 'video':
                                logger.info(f"[async] 播放器 {player_id_to_try} 确认为视频播放器。尝试设置循环...")
                                repeat_params = {"playerid": player_id_to_try, "repeat": "all"}
                                repeat_response = self._send_request("Player.SetRepeat", repeat_params)
                                logger.info(f"[async] Player.SetRepeat 响应: {repeat_response}")
                                if repeat_response and repeat_response.get('result') == 'OK':
                                    logger.info("[async] 循环播放设置成功。")
                                else:
                                    logger.warning(f"[async] 设置循环播放失败: {repeat_response}")
                            elif props_response and props_response.get('error'):
                                logger.warning(f"[async] 获取播放器 {player_id_to_try} 属性时出错: {props_response['error']} - 可能播放器尚未就绪或 ID 不正确")
                                logger.warning("[async] 无法确认播放器状态，不设置循环。")
                            else:
                                logger.warning(f"[async] 播放器 {player_id_to_try} 不是预期的视频播放器或未找到。类型: {props_response.get('result', {}).get('type')}，不设置循环。")
                        except Exception as e:
                            logger.error(f"[async] 在尝试获取属性或设置循环时发生错误: {str(e)}")
                else:
                    logger.info(f"[async] Player.Open(video) 已派发，不设置循环: {video_path}")
            except Exception as e:
                logger.warning(f"[async] Player.Open(video) 调用异常: {e}")
        
        t = threading.Thread(target=_play_video_worker, daemon=True)
        t.start()
        logger.info(f"[async] Player.Open(video) 已派发: {video_path}, 循环: {loop}")
        return {"queued": True, "file": video_path, "loop": loop}

    def play_playlist_looped(self, video_paths):
        """清空播放列表，添加多个视频，并循环播放（整体异步派发）。"""
        if not isinstance(video_paths, list) or not video_paths:
            logger.error("错误：video_paths 必须是一个非空列表。")
            return None
        playlist_id = 1
        def _playlist_worker():
            try:
                logger.info(f"[async] Playlist.Clear -> {playlist_id}")
                self._send_request("Playlist.Clear", {"playlistid": playlist_id})
                for vp in video_paths:
                    logger.info(f"[async] Playlist.Add -> {vp}")
                    self._send_request("Playlist.Add", {"playlistid": playlist_id, "item": {"file": vp}})
                logger.info(f"[async] Player.Open playlist -> position 0")
                self._send_request("Player.Open", {"item": {"playlistid": playlist_id, "position": 0}})
                logger.info(f"[async] Player.SetRepeat(all) -> playerid 1")
                self._send_request("Player.SetRepeat", {"playerid": 1, "repeat": "all"})
            except Exception as e:
                logger.warning(f"[async] 播放列表派发异常: {e}")
        threading.Thread(target=_playlist_worker, daemon=True).start()
        return {"queued": True, "playlist": len(video_paths)}

    def _get_active_player_id(self):
        """获取当前活动的播放器ID"""
        try:
            response = self._send_request("Player.GetActivePlayers")
            logger.debug(f"Player.GetActivePlayers 响应: {response}")
            if response and response.get('result'):
                players = response['result']
                if players:
                    return players[0].get('playerid')
            logger.warning("未能从响应中找到有效的播放器ID。")
            return None
        except Exception as e:
            logger.error(f"获取活动播放器ID时出错: {str(e)}")
            return None

    def stop_playback(self):
        """停止当前播放（异步派发）。"""
        # 直接尝试默认播放器1，避免阻塞
        self._send_request_async("Player.Stop", {"playerid": 1})
        logger.info("[async] Player.Stop 已派发 (playerid=1)")
        return {"queued": True}


    def pause_playback(self):
        """暂停/继续播放（异步派发）。"""
        self._send_request_async("Player.PlayPause", {"playerid": 1})
        logger.info("[async] Player.PlayPause 已派发 (playerid=1)")
        return {"queued": True}

    def set_volume(self, volume):
        """设置Kodi音量 (0-100)"""
        if not isinstance(volume, int) or not 0 <= volume <= 100:
            logger.error("错误：音量必须是 0 到 100 之间的整数。")
            return None
        params = {"volume": volume}
        # 异步发送，不阻塞调用方
        self._send_request_async("Application.SetVolume", params)
        logger.info(f"Application.SetVolume ({volume}%) 已异步派发")
        return {"queued": True, "volume": volume}

    def get_player_state(self):
        """获取当前播放器状态（为避免阻塞，改为异步派发查询，并返回占位数据）。"""
        self._send_request_async('Player.GetActivePlayers')
        logger.info("[async] Player.GetActivePlayers 已派发（不阻塞）")
        # 占位返回，避免调用方阻塞；如需真实状态，应改造为回调/轮询机制
        return {'queued': True}

    def set_ready(self):
        """设置客户端准备就绪"""
        self.ready_event.set()

    def wait_for_ready(self, timeout=10):
        """等待客户端准备就绪"""
        return self.ready_event.wait(timeout)

    def set_individual(self, isIndividual=False):
        """设置客户端是否为独立客户端"""
        self.isIndividual = isIndividual

    def get_individual(self):
        """获取客户端是否为独立客户端"""
        return self.isIndividual

    # 启动kodi
    def start_kodi(self):
        """启动kodi"""
        res = requests.get(self.mitv_startapp_url, timeout=3).json()
        logger.info(f"启动kodi响应: {res}")
        return res

    # kodi心跳检测,检查kodi客户端是否在线
    def kodi_heartbeat_check(self):
        """检查kodi客户端是否在线"""
        try:
            # 使用JSON-RPC请求进行心跳检测，支持认证
            data = {
                "jsonrpc": "2.0",
                "method": "JSONRPC.Version",
                "id": 1
            }
            response = requests.post(
                self.url,
                data=json.dumps(data),
                headers=self.headers,
                auth=self.auth,
                timeout=3
            )
            # 检查HTTP状态码
            if response.status_code != 200:
                logger.warning(f"kodi心跳检测失败: HTTP状态码 {response.status_code}")
                return False
            # 检查响应内容是否为空
            if not response.text or response.text.strip() == '':
                logger.warning(f"kodi心跳检测失败: 响应内容为空")
                return False
            # 尝试解析JSON
            try:
                json_data = response.json()
                # JSONRPC.Version返回的结果包含version字段表示成功
                if json_data.get('result') is not None:
                    return True
                else:
                    return False
            except json.JSONDecodeError as json_err:
                logger.warning(f"kodi心跳检测失败: JSON解析错误 - {json_err}, 响应内容: {response.text[:100]}")
                return False
        except requests.exceptions.Timeout:
            logger.warning(f"kodi心跳检测超时: 连接 {self.url} 超时")
            return False
        except requests.exceptions.ConnectionError:
            logger.warning(f"kodi心跳检测失败: 无法连接到 {self.url}")
            return False
        except Exception as e:
            logger.error(f"kodi心跳检测异常: {e}")
            return False    

class KodiClientManager():

    def __init__(self):
        self.kodi_clients = []
        self.video_infos = []
        # 生产环境配置文件路径
        self.kodi_config_path = 'kodi_config_prod.yaml'
        self.video_config_path = 'video_config_prod.yaml'
        # 开发环境配置文件路径
        # self.kodi_config_path = 'kodi_config_test.yaml'
        # self.video_config_path = 'video_config_test.yaml'
        # 只有一台可以有声音，其他没有声音，这是音量
        self.volume = 65
        self._init_kodi_clients_from_config()
        self._init_video_infos_from_config()

    def set_volume(self, volume):
        # 设置播放视频的音量
        self.volume = volume
        
        # 立即应用到符合条件的客户端
        # 逻辑与 sync_play_video 中一致：第一台非独立设备设置音量，其他静音
        # 注意：这里假设 kodi_clients 的顺序即为物理排列顺序
        found_first_non_individual = False
        
        for client in self.kodi_clients:
            if client.get_individual():
                continue
            
            if not found_first_non_individual:
                client.set_volume(self.volume)
                found_first_non_individual = True
            else:
                client.set_volume(0)

    def get_volume(self):
        """获取当前全局音量"""
        return self.volume
    
    def _init_video_infos_from_config(self):
        config = self._load_config(self.video_config_path)
        video_infos_config = config.get('video_infos', [])
        for video_info_config in video_infos_config:
            video_info = VideoInfo(video_info_config.get('name'), video_info_config.get('formula'), video_info_config.get('description'), video_info_config.get('video_duration'), video_info_config.get('video_path'), video_info_config.get('id'))
            logger.info(f"成功加载视频信息: {video_info.name}")
            self.video_infos.append(video_info)

    def _load_config(self, config_path) -> Dict[str, Any]:
        """
        从YAML配置文件加载配置
        
        Returns:
            dict: 配置字典
            
        Raises:
            FileNotFoundError: 配置文件不存在
            yaml.YAMLError: YAML解析错误
            Exception: 其他加载错误
        """

        if not os.path.exists(config_path):
            error_msg = f"配置文件不存在: {config_path}"
            logger.error(error_msg)
            raise FileNotFoundError(error_msg)
        
        try:
            with open(config_path, 'r', encoding='utf-8') as file:
                config = yaml.safe_load(file)
                if config is None:
                    error_msg = f"配置文件为空或格式错误: {config_path}"
                    logger.error(error_msg)
                    raise ValueError(error_msg)
                logger.info(f"成功加载配置文件: {config_path}")
                return config
        except yaml.YAMLError as e:
            error_msg = f"YAML解析错误: {e}"
            logger.error(error_msg)
            raise
        except Exception as e:
            error_msg = f"加载配置文件失败: {e}"
            logger.error(error_msg)
            raise

    def _init_kodi_clients_from_config(self):
        config = self._load_config(self.kodi_config_path)
        kodi_servers_config = config.get('kodi_servers', [])

        if not kodi_servers_config:
            logger.error("未在 config.yaml 中找到有效的 Kodi 服务器配置，脚本将退出。")
            exit()

        # 创建 Kodi 客户端实例列表
        
        for server_config in kodi_servers_config:
            client = KodiClient(
                host=server_config.get('ip', 'localhost'),
                port=server_config.get('port', 8080),
                username=server_config.get('username'),
                password=server_config.get('password'),
                id=server_config.get('id', 0)
            )
            try:
                if hasattr(client, 'set_volume') and callable(getattr(client, 'set_volume')):
                    client.set_volume(65)
                else:
                    # 兼容旧构建：直接通过 JSON-RPC 设置音量
                    client._send_request("Application.SetVolume", {"volume": 65})
            except Exception as e:
                logger.warning(f"设置音量时出现问题（已忽略以继续）：{e}")
            self.kodi_clients.append(client)



    def _resolve_config_path(self, filename = "kodi_config.yaml"):
        """Return absolute path to config file, located next to the script/exe.

        When bundled with PyInstaller, sys.frozen is True and sys.executable points to the exe.
        In normal execution, use the directory of this file.
        """
        try:
            if getattr(sys, "frozen", False):
                base_dir = os.path.dirname(sys.executable)
            else:
                base_dir = os.path.dirname(os.path.abspath(__file__))
            return os.path.join(base_dir, filename)
        except Exception:
            # Fallback to current working directory
            return filename

    def sync_play_video(self, clients, video_path, loop=False):
        """同步播放视频"""
        # 创建一个共享的Event来控制所有客户端同时开始播放
        start_event = threading.Event()
        logger.info(f"开始同步播放视频: {video_path}")
        
        # 创建播放线程
        def play_thread(client):
            # 等待开始信号
            start_event.wait()
            # 执行播放
            result = client.play_video(video_path, loop=loop)
            logger.info(f"播放视频: {video_path}, 循环: {loop}")
            logger.info(f"播放视频结果: {result}")

        # 启动所有播放线程
        threads = []
        client_index = 0
        for client in clients:
            
            if client.get_individual():
                client_index += 1
                continue
            # 只对第一台设置音量
            if client == clients[client_index]:
                client.set_volume(self.volume)
            else:
                client.set_volume(0)
            thread = threading.Thread(target=play_thread, args=(client,))
            threads.append(thread)
            thread.start()

        # 等待所有线程准备就绪
        time.sleep(0.1)  # 给线程一点时间启动
        
        # 同时触发所有客户端开始播放
        start_event.set()

        # 等待所有线程完成
        for thread in threads:
            thread.join()

    # 指定某台播放url图片
    def play_url_image_on_client(self, client, image_url):
        """指定某台播放url图片"""
        client.isIndividual = True
        return client.play_url_image(image_url)
    # 指定某台播放rtsp视频流
    def play_rtsp_video_on_client(self, client, rtsp_url,volume=0):
        """指定某台播放rtsp视频流"""
        client.isIndividual = True
        client.set_volume(volume)
        return client.play_rtsp_video(rtsp_url)
    # 撤销所有客户端的独立状态
    def revoke_individual_state(self):
        """撤销所有客户端的独立状态"""
        for client in self.kodi_clients:
            client.isIndividual = False
            client.stop_playback()

    # 启动所有kodi应用程序
    def start_all_kodi_apps(self):
        """启动所有kodi应用程序"""
        for client in self.kodi_clients:
            client.start_kodi()

    # 检查所有kodi客户端是否在线，返回所有不在线的client_index集合
    def check_all_kodi_clients_online(self):
        """检查所有kodi客户端是否在线，返回所有不在线的客户端索引集合"""
        offline_indices = set()
        client_index = 0
        for client in self.kodi_clients:
            if not client.kodi_heartbeat_check():
                offline_indices.add(client_index)
            client_index += 1
        return offline_indices