SenSu/utils/config_utils.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import logging
import yaml
import json
from pathlib import Path
from typing import Dict, Any, Optional
import copy

logger = logging.getLogger(__name__)

class ConfigUtils:
    """配置工具类"""
    
    @staticmethod
    def load_yaml_config(file_path: str, default_config: Dict = None) -> Dict:
        """加载YAML配置文件"""
        try:
            path = Path(file_path)
            if not path.exists():
                logger.warning(f"YAML配置文件不存在: {file_path}")
                if default_config:
                    ConfigUtils.save_yaml_config(file_path, default_config)
                    logger.debug(f"已创建默认YAML配置: {file_path}")
                return default_config or {}
            
            with open(path, 'r', encoding='utf-8') as f:
                config = yaml.safe_load(f)
            
            logger.debug(f"YAML配置加载成功: {file_path}")
            return config or {}
            
        except yaml.YAMLError as e:
            logger.error(f"YAML配置文件解析错误 {file_path}: {str(e)}", exc_info=True)
            return default_config or {}
        except Exception as e:
            logger.error(f"加载YAML配置时出错 {file_path}: {str(e)}", exc_info=True)
            return default_config or {}
    
    @staticmethod
    def save_yaml_config(file_path: str, config: Dict) -> bool:
        """保存YAML配置文件"""
        try:
            path = Path(file_path)
            path.parent.mkdir(parents=True, exist_ok=True)
            
            with open(path, 'w', encoding='utf-8') as f:
                yaml.dump(config, f, default_flow_style=False, allow_unicode=True, indent=2)
            
            logger.debug(f"YAML配置保存成功: {file_path}")
            return True
            
        except Exception as e:
            logger.error(f"保存YAML配置时出错 {file_path}: {str(e)}", exc_info=True)
            return False
    
    @staticmethod
    def load_json_config(file_path: str, default_config: Dict = None) -> Dict:
        """加载JSON配置文件"""
        try:
            path = Path(file_path)
            if not path.exists():
                logger.warning(f"JSON配置文件不存在: {file_path}")
                if default_config:
                    ConfigUtils.save_json_config(file_path, default_config)
                    logger.debug(f"已创建默认JSON配置: {file_path}")
                return default_config or {}
            
            with open(path, 'r', encoding='utf-8') as f:
                config = json.load(f)
            
            logger.debug(f"JSON配置加载成功: {file_path}")
            return config
            
        except json.JSONDecodeError as e:
            logger.error(f"JSON配置文件解析错误 {file_path}: {str(e)}", exc_info=True)
            return default_config or {}
        except Exception as e:
            logger.error(f"加载JSON配置时出错 {file_path}: {str(e)}", exc_info=True)
            return default_config or {}
    
    @staticmethod
    def save_json_config(file_path: str, config: Dict) -> bool:
        """保存JSON配置文件"""
        try:
            path = Path(file_path)
            path.parent.mkdir(parents=True, exist_ok=True)
            
            with open(path, 'w', encoding='utf-8') as f:
                json.dump(config, f, ensure_ascii=False, indent=2)
            
            logger.debug(f"JSON配置保存成功: {file_path}")
            return True
            
        except Exception as e:
            logger.error(f"保存JSON配置时出错 {file_path}: {str(e)}", exc_info=True)
            return False
    
    @staticmethod
    def get_nested_value(config: Dict, key_path: str, default: Any = None) -> Any:
        """获取嵌套配置值"""
        try:
            keys = key_path.split('.')
            current = config
            
            for key in keys:
                if isinstance(current, dict) and key in current:
                    current = current[key]
                else:
                    logger.debug(f"配置键不存在: {key_path}")
                    return default
            
            logger.debug(f"获取嵌套配置值: {key_path} -> {current}")
            return current
            
        except Exception as e:
            logger.error(f"获取嵌套配置值时出错 {key_path}: {str(e)}", exc_info=True)
            return default
    
    @staticmethod
    def set_nested_value(config: Dict, key_path: str, value: Any) -> bool:
        """设置嵌套配置值"""
        try:
            keys = key_path.split('.')
            current = config
            
            # 遍历到最后一个键的父级
            for key in keys[:-1]:
                if key not in current or not isinstance(current[key], dict):
                    current[key] = {}
                current = current[key]
            
            # 设置值
            current[keys[-1]] = value
            logger.debug(f"设置嵌套配置值: {key_path} -> {value}")
            return True
            
        except Exception as e:
            logger.error(f"设置嵌套配置值时出错 {key_path}: {str(e)}", exc_info=True)
            return False
    
    @staticmethod
    def merge_configs(base_config: Dict, override_config: Dict) -> Dict:
        """合并配置（深度合并）"""
        try:
            result = copy.deepcopy(base_config)
            
            for key, value in override_config.items():
                if (key in result and 
                    isinstance(result[key], dict) and 
                    isinstance(value, dict)):
                    # 递归合并字典
                    result[key] = ConfigUtils.merge_configs(result[key], value)
                else:
                    # 直接覆盖
                    result[key] = copy.deepcopy(value)
            
            logger.debug("配置合并完成")
            return result
            
        except Exception as e:
            logger.error(f"合并配置时出错: {str(e)}", exc_info=True)
            return base_config
    
    @staticmethod
    def validate_config_structure(config: Dict, schema: Dict) -> bool:
        """验证配置结构"""
        try:
            def _validate(current_config, current_schema, path=""):
                for key, expected_type in current_schema.items():
                    full_path = f"{path}.{key}" if path else key
                    
                    if key not in current_config:
                        logger.error(f"配置缺少必要字段: {full_path}")
                        return False
                    
                    actual_value = current_config[key]
                    expected_type_name = expected_type.__name__ if hasattr(expected_type, '__name__') else str(expected_type)
                    
                    if not isinstance(actual_value, expected_type):
                        logger.error(f"配置类型错误 {full_path}: 期望 {expected_type_name}, 实际 {type(actual_value).__name__}")
                        return False
                    
                    # 如果是字典且schema有嵌套定义，递归验证
                    if (isinstance(expected_type, dict) and 
                        isinstance(actual_value, dict)):
                        if not _validate(actual_value, expected_type, full_path):
                            return False
                
                return True
            
            result = _validate(config, schema)
            if result:
                logger.debug("配置结构验证通过")
            else:
                logger.error("配置结构验证失败")
            
            return result
            
        except Exception as e:
            logger.error(f"验证配置结构时出错: {str(e)}", exc_info=True)
            return False