第 2 章：提示词工程基础语法

阶段 1：展示反模式

为什么这是反模式：

• 没说配置格式（JSON? YAML? TOML? INI?）
• 没说配置来源（文件? 环境变量? 命令行?）
• 没说如何处理默认值和验证
• 没说是否需要支持热重载
• 没说错误处理策略

阶段 2：应用四要素模板

## 角色
你是一位 Python 系统工具库开发者，精通配置管理系统的设计。

## 任务
为 NAS 操作系统项目实现一个配置管理模块 ConfigManager。

### 背景
NAS 系统需要管理多种配置：系统设置、网络配置、用户偏好等。
配置可能来自多个来源，需要统一管理。

### 功能需求
1. 支持 YAML 配置文件读取和写入
2. 支持环境变量覆盖（PREFIX_NAS_ 前缀）
3. 支持嵌套键访问（如 config.get("storage.default_pool")）
4. 支持配置验证（类型检查、范围检查、必填检查）
5. 支持配置变更监听（观察者模式）

## 约束
### 技术约束
- Python 3.10+，仅使用标准库 + PyYAML
- 单文件模块，不超过 200 行核心逻辑

### 质量约束
- 线程安全
- 所有公开方法有类型注解和 docstring
- 配置写入使用原子操作（先写临时文件再 rename）
- 解析失败时给出明确的错误位置和修复建议

### 设计约束
- 遵循单一职责：配置读取、验证、监听分离为独立类
- 接口隔离：定义 ConfigBackend 抽象基类，支持未来扩展

## 输出格式
- 单个 Python 文件 config_manager.py
- 包含完整类型注解
- 包含 3 个以上使用示例
- 末尾附测试用例

阶段 3：AI 输出结果（核心代码片段）

"""NAS 系统配置管理模块 — 支持多源加载、验证、热重载。

使用示例:
    # 基本使用
    config = ConfigManager("config.yaml")
    port = config.get("web.port", default=8080)
    config.set("web.port", 9090)

    # 配置验证
    config.add_validator("web.port", lambda v: 1024 <= v <= 65535)

    # 变更监听
    config.watch("storage.default_pool", lambda k, old, new: print(f"{k}: {old} -> {new}"))
"""

import os
import re
import threading
from abc import ABC, abstractmethod
from collections.abc import Callable
from pathlib import Path
from typing import Any, Optional

import yaml


class ConfigBackend(ABC):
    """配置后端抽象 — 支持扩展到 etcd、Redis 等。"""
    @abstractmethod
    def load(self) -> dict[str, Any]: ...
    @abstractmethod
    def save(self, data: dict[str, Any]) -> None: ...


class YamlFileBackend(ConfigBackend):
    """YAML 文件后端，支持原子写入。"""
    def __init__(self, path: str | Path):
        self.path = Path(path)

    def load(self) -> dict[str, Any]:
        if not self.path.exists():
            return {}
        try:
            with open(self.path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f) or {}
        except yaml.YAMLError as e:
            raise ConfigError(f"YAML 解析失败 [{self.path}]: {e}") from e

    def save(self, data: dict[str, Any]) -> None:
        tmp_path = self.path.with_suffix('.tmp')
        with open(tmp_path, 'w', encoding='utf-8') as f:
            yaml.safe_dump(data, f, default_flow_style=False, allow_unicode=True)
        tmp_path.rename(self.path)


class ConfigManager:
    """统一配置管理器。"""

    def __init__(self, backend: ConfigBackend | str, env_prefix: str = "NAS_"):
        self._backend = YamlFileBackend(backend) if isinstance(backend, str) else backend
        self._env_prefix = env_prefix
        self._data: dict[str, Any] = {}
        self._validators: dict[str, list[Callable[[Any], None]]] = {}
        self._watchers: dict[str, list[Callable[[str, Any, Any], None]]] = {}
        self._lock = threading.RLock()
        self.reload()

    def reload(self) -> None:
        with self._lock:
            self._data = self._backend.load()
            self._apply_env_overrides()

    def get(self, key: str, default: Any = None) -> Any:
        with self._lock:
            return self._navigate(key, self._data) if self._navigate(key, self._data) is not NotImplemented else default

    def set(self, key: str, value: Any) -> None:
        with self._lock:
            old = self.get(key)
            self._validate(key, value)
            self._set_nested(key, value)
            self._notify(key, old, value)

    def add_validator(self, key: str, validator: Callable[[Any], None]) -> None:
        self._validators.setdefault(key, []).append(validator)

    def watch(self, key: str, callback: Callable[[str, Any, Any], None]) -> None:
        self._watchers.setdefault(key, []).append(callback)

    def save(self) -> None:
        with self._lock:
            self._backend.save(self._data)

    def _navigate(self, key: str, data: dict) -> Any:
        parts = key.split(".")
        current: Any = data
        for part in parts:
            if isinstance(current, dict) and part in current:
                current = current[part]
            else:
                return NotImplemented
        return current

    def _set_nested(self, key: str, value: Any) -> None:
        parts = key.split(".")
        d = self._data
        for part in parts[:-1]:
            d = d.setdefault(part, {})
        d[parts[-1]] = value

    def _apply_env_overrides(self) -> None:
        prefix = f"{self._env_prefix}"
        for env_key, env_val in os.environ.items():
            if env_key.startswith(prefix):
                config_key = env_key[len(prefix):].lower().replace("__", ".")
                try:
                    self._set_nested(config_key, yaml.safe_load(env_val))
                except Exception:
                    self._set_nested(config_key, env_val)

    def _validate(self, key: str, value: Any) -> None:
        for validator in self._validators.get(key, []):
            validator(value)

    def _notify(self, key: str, old: Any, new: Any) -> None:
        for callback in self._watchers.get(key, []):
            try:
                callback(key, old, new)
            except Exception:
                pass  # 监听器异常不应影响配置写入


class ConfigError(Exception):
    """配置错误基类。"""
    pass

阶段 4：审查要点