DataOps-platform/app/core/system/config.py

"""
系统配置管理模块
提供系统配置的获取、验证和安全过滤功能
"""

import logging
import os

from flask import current_app

logger = logging.getLogger(__name__)


def get_system_config():
    """
    获取系统配置信息
    过滤掉敏感的配置项

    Returns:
        dict: 过滤后的系统配置信息
    """
    try:
        # 收集系统配置信息（去除敏感信息）
        config_info = {
            "environment": current_app.config["FLASK_ENV"],
            "debug_mode": current_app.config["DEBUG"],
            "platform": current_app.config["PLATFORM"],
            "port": current_app.config["PORT"],
            "allowed_extensions": list(current_app.config["ALLOWED_EXTENSIONS"]),
            "bucket_name": current_app.config["BUCKET_NAME"],
            "prefix": current_app.config["PREFIX"],
        }

        return config_info
    except Exception as e:
        logger.error(f"获取系统配置失败: {str(e)}")
        return {"error": str(e)}


def validate_config():
    """
    验证系统配置的有效性
    检查必要的配置项是否存在且有效

    Returns:
        tuple: (是否有效, 错误信息)
    """
    errors = []

    # 检查Neo4j配置
    if "NEO4J_URI" not in current_app.config or not current_app.config["NEO4J_URI"]:
        errors.append("NEO4J_URI未配置")
    if "NEO4J_USER" not in current_app.config or not current_app.config["NEO4J_USER"]:
        errors.append("NEO4J_USER未配置")
    if (
        "NEO4J_PASSWORD" not in current_app.config
        or not current_app.config["NEO4J_PASSWORD"]
    ):
        errors.append("NEO4J_PASSWORD未配置")

    # 检查MinIO配置
    if "MINIO_HOST" not in current_app.config or not current_app.config["MINIO_HOST"]:
        errors.append("MINIO_HOST未配置")
    if "MINIO_USER" not in current_app.config or not current_app.config["MINIO_USER"]:
        errors.append("MINIO_USER未配置")
    if (
        "MINIO_PASSWORD" not in current_app.config
        or not current_app.config["MINIO_PASSWORD"]
    ):
        errors.append("MINIO_PASSWORD未配置")

    # 检查其他必要配置
    if "BUCKET_NAME" not in current_app.config or not current_app.config["BUCKET_NAME"]:
        errors.append("BUCKET_NAME未配置")
    if "PREFIX" not in current_app.config:
        errors.append("PREFIX未配置")

    return (len(errors) == 0, errors)


def get_config_file_paths():
    """
    获取系统所有配置文件的路径

    Returns:
        list: 配置文件路径列表
    """
    base_dir = os.path.dirname(
        os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    )
    config_dir = os.path.join(base_dir, "config")

    if not os.path.exists(config_dir):
        logger.warning(f"配置目录不存在: {config_dir}")
        return []

    config_files = []
    for file in os.listdir(config_dir):
        if file.endswith(".py") or file.endswith(".yaml") or file.endswith(".json"):
            config_files.append(os.path.join(config_dir, file))

    return config_files