smart-question-bank/SERVER/app/config/loader_config.py

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()