Initial commit: SenSu Alpha 0.2.0
- 13-service async plugin framework - Textual TUI with CLI fallback - Plugin hot-reload + permission system - Web management panel (aiohttp) - Bridge-based inter-module communication - 10 regression tests Fixes applied: - PBKDF2-SHA256 auth (was plain SHA256) - Auth bypass removed (was allow-all on fail) - Bare excepts replaced with logged errors - CatFramework/DreamSu -> SenSu naming unified - ServiceManager: health checks + startup_order - Env var credentials (SENSU_ADMIN_PASSWORD etc)
This commit is contained in:
@@ -0,0 +1,203 @@
|
||||
#!/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
|
||||
Reference in New Issue
Block a user