NebulaShell/oss/config/config.py

"""配置管理 - 支持环境变量和配置文件"""
import os
import json
from pathlib import Path
from typing import Any, Optional


class Config:
    """全局配置管理
    
    优先级：环境变量 > 配置文件 > 默认值
    """
    
    # 隐藏成就系统标志（在类级别定义，避免循环导入）
    _ACHIEVEMENTS_ENABLED = False
    
    DEFAULTS = {
        # 服务器配置
        "HTTP_API_PORT": 10086,
        "HTTP_TCP_PORT": 10086,
        "HOST": "127.0.0.1",
        
        # 数据目录
        "DATA_DIR": "./data",
        "PLUGIN_STORAGE_DIR": "./data/plugin-storage",
        "SIGNATURE_KEYS_DIR": "./data/signature-verifier/keys",
        "DCIM_DIR": "./data/DCIM",
        "WEB_TOOLKIT_DIR": "./data/web-toolkit",
        "HTML_RENDER_DIR": "./data/html-render",
        
        # 插件配置
        "STORE_DIR": "./store",
        "MODS_DIR": "./mods",
        "PLUGINS_DIR": "./oss/plugins",
        
        # 日志配置
        "LOG_LEVEL": "INFO",
        "LOG_FORMAT": "text",  # text 或 json
        
        # 安全配置
        "PERMISSION_CHECK": True,
        "ENFORCE_SIGNATURE": True,
        "JWT_SECRET": "",
        "CSRF_SECRET": "",
        "CSRF_TOKEN_TTL": 3600,
        "TLS_CERT_DIR": "./data/tls",
        "PUBLIC_PATHS": ["/health", "/favicon.ico", "/api/status", "/api/health", "/api/login", "/metrics"],
        "ADMIN_USER": "admin",
        "ADMIN_PASS": "admin123",
        "CORS_ALLOWED_ORIGINS": ["http://localhost:3000", "http://127.0.0.1:3000"],
        "CSRF_ENABLED": True,
        "INPUT_VALIDATION_ENABLED": True,
        "RATE_LIMIT_ENABLED": True,
        "RATE_LIMIT_MAX_REQUESTS": 100,
        "RATE_LIMIT_TIME_WINDOW": 60,
        
        # 性能配置
        "MAX_WORKERS": 4,
        "ENABLE_ASYNC": False,
        
        # NBPF 配置
        "NBPF_KEYS_DIR": "./data/nbpf-keys",
        "NBPF_TRUSTED_KEYS_DIR": "./data/nbpf-keys/trusted",
        "NBPF_RSA_KEYS_DIR": "./data/nbpf-keys/rsa",
        "NBPF_ENCRYPTION_ENABLED": True,
        "NBPF_SIGNATURE_REQUIRED": True,
    }
    
    def __init__(self, config_file: Optional[str] = None):
        self._config: dict[str, Any] = dict(self.DEFAULTS)
        self._config_file = config_file
        
        # 加载配置文件
        if config_file:
            self._load_from_file(config_file)
        
        # 环境变量覆盖
        self._load_from_env()
    
    def _load_from_file(self, path: str):
        """从 JSON 配置文件加载"""
        config_path = Path(path)
        if config_path.exists():
            try:
                with open(config_path, 'r', encoding='utf-8') as f:
                    file_config = json.load(f)
                for key, value in file_config.items():
                    if key in self.DEFAULTS:
                        self._config[key] = value
                
                # 隐藏成就：配置黑客 - 记录配置修改
                if Config._ACHIEVEMENTS_ENABLED:
                    try:
                        from oss.core.achievements import get_validator
                        validator = get_validator()
                        validator.record_config_modify()
                    except Exception as e:
                        print(f"[Config] 配置加载错误: {e}")
            except Exception as e:
                print(f"[Config] 加载配置文件失败：{type(e).__name__}: {e}")
    
    def _load_from_env(self):
        """从环境变量加载"""
        for key in self.DEFAULTS.keys():
            env_value = os.environ.get(key)
            if env_value is not None:
                # 类型转换
                default_value = self.DEFAULTS[key]
                if isinstance(default_value, bool):
                    self._config[key] = env_value.lower() in ('true', '1', 'yes')
                elif isinstance(default_value, int):
                    try:
                        self._config[key] = int(env_value)
                    except ValueError as e:
                        print(f"[Config] 类型转换错误: {e}")
                else:
                    self._config[key] = env_value
    
    def get(self, key: str, default: Any = None) -> Any:
        """获取配置值"""
        return self._config.get(key, default)
    
    def set(self, key: str, value: Any):
        """设置配置值"""
        if key in self.DEFAULTS:
            self._config[key] = value
    
    def all(self) -> dict[str, Any]:
        """获取所有配置"""
        return dict(self._config)
    
    @property
    def http_api_port(self) -> int:
        return int(self._config["HTTP_API_PORT"])
    
    @property
    def http_tcp_port(self) -> int:
        return int(self._config["HTTP_TCP_PORT"])
    
    @property
    def host(self) -> str:
        return str(self._config["HOST"])
    
    @property
    def data_dir(self) -> Path:
        return Path(self._config["DATA_DIR"])
    
    @property
    def store_dir(self) -> Path:
        return Path(self._config["STORE_DIR"])

    @property
    def mods_dir(self) -> Path:
        return Path(self._config["MODS_DIR"])
    
    @property
    def log_level(self) -> str:
        return str(self._config["LOG_LEVEL"])
    
    @property
    def permission_check(self) -> bool:
        return bool(self._config["PERMISSION_CHECK"])


# 全局配置实例
_global_config: Optional[Config] = None


def get_config() -> Config:
    """获取全局配置实例"""
    global _global_config
    if _global_config is None:
        _global_config = Config()
    return _global_config


def init_config(config_file: Optional[str] = None) -> Config:
    """初始化全局配置"""
    global _global_config
    _global_config = Config(config_file)
    return _global_config