|
@@ -0,0 +1,205 @@
|
|
|
+import os, yaml, datetime
|
|
|
+from typing import Any, List, Callable
|
|
|
+from watchdog.observers import Observer
|
|
|
+from watchdog.events import FileSystemEventHandler
|
|
|
+from .config import Config
|
|
|
+
|
|
|
+
|
|
|
+class ConfigReloadHandler(FileSystemEventHandler):
|
|
|
+ """配置文件变更处理器"""
|
|
|
+
|
|
|
+ def __init__(self, callback: Callable):
|
|
|
+ self.callback = callback
|
|
|
+
|
|
|
+ def on_modified(self, event):
|
|
|
+ if not event.is_directory:
|
|
|
+ self.callback()
|
|
|
+
|
|
|
+
|
|
|
+class ConfigLoader:
|
|
|
+ """配置加载器"""
|
|
|
+ _instance = None
|
|
|
+ _observers = []
|
|
|
+
|
|
|
+ # 环境变量映射配置
|
|
|
+ ENV_MAPPING = {
|
|
|
+ 'APP_NAME': 'app.name',
|
|
|
+ 'APP_VERSION': 'app.version',
|
|
|
+ 'APP_DEBUG': 'app.debug',
|
|
|
+ 'DB_HOST': 'database.host',
|
|
|
+ 'DB_PORT': 'database.port',
|
|
|
+ 'DB_USER': 'database.user',
|
|
|
+ 'DB_PASSWORD': 'database.password',
|
|
|
+ 'DB_NAME': 'database.name',
|
|
|
+ }
|
|
|
+
|
|
|
+ def __new__(cls):
|
|
|
+ if cls._instance is None:
|
|
|
+ cls._instance = super().__new__(cls)
|
|
|
+ cls._instance._load_config()
|
|
|
+ return cls._instance
|
|
|
+
|
|
|
+ def _load_env_vars(self, config: Config) -> Config:
|
|
|
+ """加载环境变量配置"""
|
|
|
+ for env_key, config_path in self.ENV_MAPPING.items():
|
|
|
+ value = os.getenv(env_key)
|
|
|
+ if value is None:
|
|
|
+ continue
|
|
|
+
|
|
|
+ try:
|
|
|
+ # 按路径分割
|
|
|
+ keys = config_path.split('.')
|
|
|
+ current = config
|
|
|
+
|
|
|
+ # 遍历配置路径
|
|
|
+ for key in keys[:-1]:
|
|
|
+ if not hasattr(current, key):
|
|
|
+ setattr(current, key, Config())
|
|
|
+ current = getattr(current, key)
|
|
|
+
|
|
|
+ # 获取目标字段类型
|
|
|
+ last_key = keys[-1]
|
|
|
+ target_type = type(getattr(current, last_key, str))
|
|
|
+
|
|
|
+ # 根据目标类型转换值
|
|
|
+ if target_type == bool:
|
|
|
+ # 支持多种布尔值表示
|
|
|
+ value_lower = value.lower()
|
|
|
+ if value_lower in ('true', 'yes', 'y', '1'):
|
|
|
+ setattr(current, last_key, True)
|
|
|
+ elif value_lower in ('false', 'no', 'n', '0'):
|
|
|
+ setattr(current, last_key, False)
|
|
|
+ else:
|
|
|
+ raise ValueError(f"环境变量 {env_key} 的值 {value} 不是有效的布尔值")
|
|
|
+ elif target_type == int:
|
|
|
+ # 支持数字字符串
|
|
|
+ try:
|
|
|
+ setattr(current, last_key, int(float(value)))
|
|
|
+ except (ValueError, TypeError):
|
|
|
+ raise ValueError(f"环境变量 {env_key} 的值 {value} 不是有效的整数")
|
|
|
+ elif target_type == float:
|
|
|
+ # 支持浮点数字符串
|
|
|
+ try:
|
|
|
+ setattr(current, last_key, float(value))
|
|
|
+ except (ValueError, TypeError):
|
|
|
+ raise ValueError(f"环境变量 {env_key} 的值 {value} 不是有效的浮点数")
|
|
|
+ elif target_type == list:
|
|
|
+ # 支持数组类型,格式:value1,value2,value3
|
|
|
+ try:
|
|
|
+ setattr(current, last_key,
|
|
|
+ [item.strip() for item in value.split(',')])
|
|
|
+ except Exception:
|
|
|
+ raise ValueError(
|
|
|
+ f"环境变量 {env_key} 的值 {value} 不是有效的数组格式")
|
|
|
+ elif target_type == set:
|
|
|
+ # 支持集合类型,格式:value1,value2,value3
|
|
|
+ try:
|
|
|
+ setattr(current, last_key,
|
|
|
+ set(item.strip() for item in value.split(',')))
|
|
|
+ except Exception:
|
|
|
+ raise ValueError(
|
|
|
+ f"环境变量 {env_key} 的值 {value} 不是有效的集合格式")
|
|
|
+ elif target_type == dict:
|
|
|
+ # 支持字典类型,格式:key1:value1,key2:value2
|
|
|
+ try:
|
|
|
+ setattr(
|
|
|
+ current, last_key,
|
|
|
+ dict(item.split(':') for item in value.split(',')))
|
|
|
+ except Exception:
|
|
|
+ raise ValueError(
|
|
|
+ f"环境变量 {env_key} 的值 {value} 不是有效的字典格式")
|
|
|
+ elif target_type == datetime.datetime:
|
|
|
+ # 支持时间类型,格式:YYYY-MM-DD HH:MM:SS
|
|
|
+ try:
|
|
|
+ setattr(
|
|
|
+ current, last_key,
|
|
|
+ datetime.datetime.strptime(value,
|
|
|
+ '%Y-%m-%d %H:%M:%S'))
|
|
|
+ except Exception:
|
|
|
+ raise ValueError(
|
|
|
+ f"环境变量 {env_key} 的值 {value} 不是有效的时间格式")
|
|
|
+ else:
|
|
|
+ setattr(current, last_key, value)
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[ERROR] 加载环境变量 {env_key} 时出错: {str(e)}")
|
|
|
+ continue
|
|
|
+
|
|
|
+ return config
|
|
|
+
|
|
|
+ def _load_yaml_config(self, file_path: str) -> dict:
|
|
|
+ """加载YAML配置文件"""
|
|
|
+ try:
|
|
|
+ if os.path.exists(file_path):
|
|
|
+ with open(file_path, 'r', encoding='utf-8') as f:
|
|
|
+ config = yaml.safe_load(f) or {}
|
|
|
+ if not isinstance(config, dict):
|
|
|
+ raise ValueError(f"配置文件 {file_path} 格式错误:应为字典格式")
|
|
|
+ return config
|
|
|
+ else:
|
|
|
+ print(f"配置文件 {file_path} 不存在")
|
|
|
+ return {}
|
|
|
+ except yaml.YAMLError as e:
|
|
|
+ print(f"YAML解析错误:{str(e)}")
|
|
|
+ return {}
|
|
|
+ except Exception as e:
|
|
|
+ print(f"加载配置文件 {file_path} 时发生错误:{str(e)}")
|
|
|
+ return {}
|
|
|
+
|
|
|
+ def _load_config(self):
|
|
|
+ """加载配置"""
|
|
|
+ # 初始化配置实例
|
|
|
+ self._config = Config()
|
|
|
+ self._callbacks: List[Callable] = []
|
|
|
+
|
|
|
+ # 1. 加载默认值(通过Config类初始化)
|
|
|
+
|
|
|
+ # 2. 加载通用配置
|
|
|
+ self._config.update_from_dict(
|
|
|
+ self._load_yaml_config('SERVER/app/application.yml'))
|
|
|
+
|
|
|
+ # 3. 加载指定环境配置
|
|
|
+ env = os.getenv('ENV', 'dev')
|
|
|
+ self._config.update_from_dict(
|
|
|
+ self._load_yaml_config(f'SERVER/app/application-{env}.yml'))
|
|
|
+
|
|
|
+ # 4. 最后加载环境变量(最高优先级)
|
|
|
+ self._config = self._load_env_vars(self._config)
|
|
|
+
|
|
|
+ # 启动配置监控
|
|
|
+ self._start_config_watcher()
|
|
|
+
|
|
|
+ def _start_config_watcher(self):
|
|
|
+ """启动配置文件监控"""
|
|
|
+ event_handler = ConfigReloadHandler(self.reload_config)
|
|
|
+ observer = Observer()
|
|
|
+ observer.schedule(event_handler, path='SERVER/app/', recursive=False)
|
|
|
+ observer.start()
|
|
|
+ self._observers.append(observer)
|
|
|
+
|
|
|
+ def reload_config(self):
|
|
|
+ """重新加载配置"""
|
|
|
+ self._load_config()
|
|
|
+ for callback in self._callbacks:
|
|
|
+ callback()
|
|
|
+
|
|
|
+ def get(self, key: str, default=None) -> Any:
|
|
|
+ """获取配置值"""
|
|
|
+ keys = key.split('.')
|
|
|
+ current = self._config
|
|
|
+ try:
|
|
|
+ for k in keys:
|
|
|
+ current = getattr(current, k)
|
|
|
+ return current
|
|
|
+ except (KeyError, AttributeError):
|
|
|
+ return default
|
|
|
+
|
|
|
+ def on_config_change(self, callback: Callable):
|
|
|
+ """注册配置变更回调"""
|
|
|
+ self._callbacks.append(callback)
|
|
|
+
|
|
|
+ def __del__(self):
|
|
|
+ """清理资源"""
|
|
|
+ for observer in self._observers:
|
|
|
+ observer.stop()
|
|
|
+ observer.join()
|