certflow.config.loader 源代码

"""YAML 配置加载器

支持环境变量替换、文件包含和配置验证.
"""

import os
import re
from pathlib import Path
from typing import Any

import yaml
from loguru import logger

from certflow.config.models import RootConfig


# YAML 自定义标签: !include filename.yaml
class _IncludeTag(yaml.YAMLObject):
    """YAML !include 自定义标签，用于包含外部配置文件

    该标签允许在YAML配置文件中直接引用其他YAML文件,
    实现配置的模块化和复用.

    Attributes:
        yaml_tag: YAML标签名称，用于识别!include标记
        file_path: 被包含文件的路径，相对于配置目录
    """

    yaml_tag = "!include"

    def __init__(self, file_path: str):
        """初始化包含标签实例

        Args:
            file_path: 被包含文件的路径，可以是相对路径或绝对路径
        """
        self.file_path = file_path

    @classmethod
    def from_yaml(cls, loader, node):
        """从YAML节点构造_IncludeTag实例

        该方法是PyYAML自定义标签的标准构造器接口.

        Args:
            loader: YAML加载器实例，用于解析YAML数据
            node: YAML节点对象，包含待解析的标量数据

        Returns:
            _IncludeTag: 使用节点中的文件路径初始化后的标签实例
        """
        return cls(loader.construct_scalar(node))

    def __repr__(self) -> str:
        """返回标签的字符串表示

        Returns:
            str: 标签的字符串形式，如 "!include config/database.yaml"
        """
        return f"!include {self.file_path}"


# 自定义 YAML 构造器，处理 !include 裸标签
class _IncludeLoader(yaml.SafeLoader):
    """自定义YAML加载器，支持!include标签语法

    继承自yaml.SafeLoader，添加了对!include自定义标签的支持.
    用于解析包含文件引用语法的YAML配置文件.
    """

    pass


def _include_constructor(loader: yaml.Loader, node: yaml.nodes.ScalarNode) -> _IncludeTag:
    """处理!include标签的YAML构造器函数

    当YAML解析器遇到!include标签时调用此函数，
    创建对应的_IncludeTag对象.

    Args:
        loader: YAML加载器实例
        node: YAML标量节点，包含被引用文件的路径

    Returns:
        _IncludeTag: 包含文件路径信息的标签对象
    """
    return _IncludeTag(loader.construct_scalar(node))


_IncludeLoader.add_constructor("!include", _include_constructor)




class ConfigLoader:
    """配置加载器，负责加载和处理YAML配置文件

    提供完整的配置文件加载功能，包括环境变量替换、配置文件包含和配置验证.
    支持从指定目录加载YAML格式的配置文件，并自动处理配置中的动态内容.

    Attributes:
        config_dir: 配置文件所在目录的路径，默认为项目根目录下的config文件夹
        _config: 内部存储的配置对象，仅在调用load()后有效

    Examples:
        >>> from pathlib import Path
        >>> loader = ConfigLoader(Path("/path/to/config"))
        >>> config = loader.load("app_config.yaml")
        >>> print(config.database.host)
    """



    def __init__(self, config_dir: Path | None = None):
        """初始化配置加载器实例

        如果不指定配置目录，会自动定位到项目根目录下的config文件夹.
        开发环境: 通过当前文件向上查找4级父目录确定.
        打包环境: 使用可执行文件所在目录 (sys.frozen).

        Args:
            config_dir: 配置文件所在目录路径，默认为None时自动使用项目config目录
        """
        if config_dir is None:
            import sys

            if getattr(sys, "frozen", False):
                # PyInstaller 打包环境: config 目录在 exe 同级
                project_root = Path(sys.executable).parent
            else:
                # 开发环境: 从当前文件定位项目根目录
                current_file = Path(__file__).resolve()
                project_root = current_file.parent.parent.parent.parent
            config_dir = project_root / "config"

        self.config_dir = Path(config_dir)
        self._config: RootConfig | None = None


    def _preprocess_top_level_includes(self, content: str) -> str:
        """预处理顶层!include指令，展开被引用文件内容

        由于YAML规范不允许文档顶层直接使用裸标签!include，
        该方法在YAML解析前将顶层!include行替换为对应文件的文本内容.
        仅处理行首的!include指令，不影响嵌套在数据结构中的标签.

        Args:
            content: 原始的YAML文本内容

        Returns:
            str: 展开顶层!include后的YAML文本内容

        Example:
            输入: "!include database.yaml"
            输出: "host: localhost\nport: 5432"
        """
        lines = content.splitlines(keepends=True)
        result_lines: list[str] = []
        include_pattern = re.compile(r"^\s*!include\s+(.+\.ya?ml)\s*$")

        for line in lines:
            m = include_pattern.match(line)
            if m:
                file_path = m.group(1).strip()
                full_path = self.config_dir / file_path
                if full_path.exists():
                    logger.debug(f"Including (top-level): {file_path}")
                    with open(full_path, encoding="utf-8") as f:
                        result_lines.append(f.read())
                else:
                    logger.warning(f"Included file not found: {full_path}")
            else:
                result_lines.append(line)

        return "".join(result_lines)



    def load(self, config_file: str = "config.yaml") -> RootConfig:
        """加载配置文件并进行处理

        执行完整的配置加载流程: 读取文件 -> 展开顶层包含 -> 解析YAML ->
        处理嵌套包含 -> 替换环境变量 -> 验证配置结构.

        Args:
            config_file: 配置文件名，默认为"config.yaml"

        Returns:
            RootConfig: 经过验证的配置对象，包含所有处理后的配置数据

        Raises:
            FileNotFoundError: 指定的配置文件不存在
            yaml.YAMLError: YAML语法错误或格式不正确
            ValidationError: 配置内容不符合定义的数据模型
        """
        config_path = self.config_dir / config_file

        if not config_path.exists():
            raise FileNotFoundError(f"Configuration file not found: {config_path}")

        logger.info(f"Loading configuration from: {config_path}")

        # 读取原始内容
        with open(config_path, encoding="utf-8") as f:
            raw_content = f.read()

        # 预处理: 展开顶层的 !include 裸标签
        raw_content = self._preprocess_top_level_includes(raw_content)

        # 使用自定义 Loader 解析 YAML (处理映射值中的 !include 标签)
        raw_config = yaml.load(raw_content, Loader=_IncludeLoader) or {}

        # 递归处理 !include 标签对象，加载外部文件并合并
        raw_config = self._process_includes(raw_config)

        # 替换环境变量
        raw_config = self._resolve_env_vars(raw_config)

        # 验证并转换为 Pydantic 模型
        self._config = RootConfig(**raw_config)

        logger.info("Configuration loaded successfully")
        return self._config


    def _process_includes(self, data: Any) -> Any:
        """递归处理配置数据中的!include指令

        支持两种形式的!include语法:
        1. 裸标签形式: !include file.yaml (解析为_IncludeTag对象)
        2. 字典键形式: {"!include": "file.yaml"} (向后兼容)

        Args:
            data: 待处理的配置数据，可以是字典、列表或基本类型

        Returns:
            Any: 处理后的配置数据，所有!include标签已被替换为实际内容
        """
        if isinstance(data, _IncludeTag):
            # 裸标签形式: !include file.yaml
            included = self._load_included_file(data.file_path)
            return self._process_includes(included)

        if isinstance(data, dict):
            result = {}
            for key, value in data.items():
                if key == "!include":
                    # 字典键形式(向后兼容): some_key: !include file.yaml
                    included = self._load_included_file(value)
                    result.update(self._process_includes(included))
                elif isinstance(value, _IncludeTag):
                    # 字典值形式: some_key: !include file.yaml
                    included = self._load_included_file(value.file_path)
                    result[key] = self._process_includes(included)
                else:
                    result[key] = self._process_includes(value)
            return result

        if isinstance(data, list):
            return [self._process_includes(item) for item in data]

        return data

    def _load_included_file(self, file_path: str) -> dict[str, Any]:
        """加载被包含的配置文件

        从配置目录读取指定的YAML文件并解析为字典对象.
        如果文件不存在，记录警告并返回空字典.

        Args:
            file_path: 文件路径，相对于配置目录(config_dir)

        Returns:
            Dict[str, Any]: 解析后的配置字典，文件不存在时返回空字典
        """
        full_path = self.config_dir / file_path

        if not full_path.exists():
            logger.warning(f"Included file not found: {full_path}")
            return {}

        with open(full_path, encoding="utf-8") as f:
            return yaml.safe_load(f)

    def _resolve_env_vars(self, data: Any) -> Any:
        """递归替换配置中的环境变量

        支持的环境变量语法:
        - ${ENV_VAR}: 直接替换为环境变量的值
        - ${ENV_VAR:default_value}: 环境变量不存在时使用默认值

        Args:
            data: 待处理的配置数据，可以是任意嵌套结构

        Returns:
            Any: 替换环境变量后的配置数据

        Example:
            输入: {"host": "${DB_HOST:localhost}", "port": "${DB_PORT:5432}"}
            输出: {"host": "192.168.1.100", "port": "5432"}
        """
        pattern = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(?::([^}]*))?\}")

        def replace_value(value: Any) -> Any:
            if isinstance(value, str):

                def replacer(match):
                    var_name = match.group(1)
                    default_value = match.group(2) or ""
                    return os.environ.get(var_name, default_value)

                return pattern.sub(replacer, value)
            if isinstance(value, dict):
                return {k: replace_value(v) for k, v in value.items()}
            if isinstance(value, list):
                return [replace_value(item) for item in value]
            return value

        return replace_value(data)

    @property
    def config(self) -> RootConfig:
        """获取已加载的配置对象

        该属性提供对加载后的配置对象的只读访问.
        必须先调用load()方法，否则会抛出运行时错误.

        Returns:
            RootConfig: 已加载的配置对象

        Raises:
            RuntimeError: 配置尚未加载，需要先调用load()方法
        """
        if self._config is None:
            raise RuntimeError("Configuration not loaded. Call load() first.")
        return self._config



# 全局配置实例
_global_loader: ConfigLoader | None = None
_global_config: RootConfig | None = None




def get_config(config_dir: Path | None = None, reload: bool = False) -> RootConfig:
    """获取全局配置实例(单例模式)

    提供便捷的全局配置访问接口，避免在多处重复创建ConfigLoader实例.
    首次调用时会自动加载配置，后续调用返回缓存的配置对象.

    Args:
        config_dir: 配置文件所在目录路径，首次调用时可指定，后续调用此参数无效
        reload: 是否强制重新加载配置，设为True时会重新读取配置文件

    Returns:
        RootConfig: 全局配置对象实例

    Example:
        >>> from certflow.config.loader import get_config
        >>> config = get_config()  # 首次调用，自动加载配置
        >>> db_config = get_config().database  # 使用缓存配置
        >>> new_config = get_config(reload=True)  # 强制重新加载
    """
    global _global_loader, _global_config

    if _global_loader is None or reload:
        _global_loader = ConfigLoader(config_dir)
        _global_config = _global_loader.load()

    return _global_config