tv_show_v1/hardware/mitvs_module.py

import yaml
import requests
import os
import json
from utils.logger_config import logger

class MiTVController:
    def __init__(self, kodi_config_path='kodi_config_prod.yaml', ha_config_path='ha_config_prod.yaml'):
        self.kodi_config_path = kodi_config_path
        self.ha_config_path = ha_config_path
        
        self.kodi_config = self._load_config(self.kodi_config_path)
        self.ha_config_full = self._load_config(self.ha_config_path)
        
        self.ha_config = self.ha_config_full.get('home_assistant', {})
        self.base_url = self.ha_config.get('url', '').rstrip('/')
        self.token = self.ha_config.get('token', '')
        self.headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
        }
        self.kodi_servers = self.kodi_config.get('kodi_servers', [])

    def _load_config(self, path):
        if not os.path.exists(path):
            logger.error(f"配置文件不存在: {path}")
            return {}
        try:
            with open(path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f) or {}
        except Exception as e:
            logger.error(f"加载配置文件失败 {path}: {e}")
            return {}

    def _call_ha_service(self, entity_id, service="press", domain="button"):
        if not self.base_url or not self.token:
            logger.error("Home Assistant 配置缺失 (url or token)")
            return False
        
        if not entity_id:
            logger.warning("未配置 entity_id")
            return False

        url = f"{self.base_url}/api/services/{domain}/{service}"
        payload = {"entity_id": entity_id}
        
        logger.info(f"准备调用HA接口: URL={url}, Entity={entity_id}")
        
        try:
            response = requests.post(url, headers=self.headers, json=payload, timeout=5)
            if response.status_code in [200, 201]:
                logger.info(f"HA调用成功: {entity_id} -> {domain}.{service}, 响应: {response.text}")
                return True
            else:
                logger.error(f"HA调用失败: {response.status_code} - {response.text}")
                return False
        except Exception as e:
            logger.error(f"HA请求异常: {e}")
            return False

    def turn_on_display(self, kodi_id):
        """唤醒指定ID的电视"""
        server = next((s for s in self.kodi_servers if s.get('id') == kodi_id), None)
        if not server:
            logger.error(f"未找到ID为 {kodi_id} 的Kodi服务器配置")
            return False
        
        entity_id = server.get('ha_turn_on_entity_id')
        if not entity_id:
            logger.error(f"Kodi ID {kodi_id} 未配置唤醒实例ID (ha_turn_on_entity_id)")
            return False
            
        # 智能调用，不强制指定service，让 _call_ha_service 判断
        return self._call_ha_service(entity_id)

    def turn_off_display(self, kodi_id):
        """息屏指定ID的电视"""
        server = next((s for s in self.kodi_servers if s.get('id') == kodi_id), None)
        if not server:
            logger.error(f"未找到ID为 {kodi_id} 的Kodi服务器配置")
            return False
        
        entity_id = server.get('ha_turn_off_entity_id')
        if not entity_id:
            logger.error(f"Kodi ID {kodi_id} 未配置息屏实例ID (ha_turn_off_entity_id)")
            return False
            
        # 智能调用，不强制指定service
        return self._call_ha_service(entity_id)

    def turn_on_all_displays(self):
        """唤醒所有电视"""
        success_count = 0
        for server in self.kodi_servers:
            if self.turn_on_display(server.get('id')):
                success_count += 1
        return success_count > 0

    def turn_off_all_displays(self):
        """息屏所有电视"""
        success_count = 0
        for server in self.kodi_servers:
            if self.turn_off_display(server.get('id')):
                success_count += 1
        return success_count > 0

# 单例实例
mitv_controller = MiTVController()

def turn_on_display(kodi_id):
    return mitv_controller.turn_on_display(kodi_id)

def turn_off_display(kodi_id):
    return mitv_controller.turn_off_display(kodi_id)

def turn_on_all_displays():
    return mitv_controller.turn_on_all_displays()

def turn_off_all_displays():
    return mitv_controller.turn_off_all_displays()