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

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

import logging
import json
import os
from app.config.config import Config

logger = logging.getLogger(__name__)

def get_system_config():
    """
    获取系统配置信息
    过滤掉敏感的配置项
    
    Returns:
        dict: 过滤后的系统配置信息
    """
    try:
        # 收集系统配置信息（去除敏感信息）
        config_info = {
            "environment": Config.ENVIRONMENT,
            "debug_mode": Config.DEBUG,
            "port": Config.PORT,
            "platform": Config.PLATFORM,
            "upload_folder": Config.UPLOAD_FOLDER,
            "bucket_name": Config.BUCKET_NAME,
            "prefix": Config.PREFIX,
            "neo4j_uri": Config.NEO4J_URI,
            "neo4j_encrypted": Config.NEO4J_ENCRYPTED,
            # 不返回敏感信息如密码、密钥等
        }
        
        return config_info
    except Exception as e:
        logger.error(f"获取系统配置失败: {str(e)}")
        return {"error": str(e)}

def validate_config():
    """
    验证系统配置的有效性
    检查必要的配置项是否存在且有效
    
    Returns:
        tuple: (是否有效, 错误信息)
    """
    errors = []
    
    # 检查Neo4j配置
    if not hasattr(Config, 'NEO4J_URI') or not Config.NEO4J_URI:
        errors.append("NEO4J_URI未配置")
    if not hasattr(Config, 'NEO4J_USER') or not Config.NEO4J_USER:
        errors.append("NEO4J_USER未配置")
    if not hasattr(Config, 'NEO4J_PASSWORD') or not Config.NEO4J_PASSWORD:
        errors.append("NEO4J_PASSWORD未配置")
    
    # 检查MinIO配置
    if not hasattr(Config, 'MINIO_HOST') or not Config.MINIO_HOST:
        errors.append("MINIO_HOST未配置")
    if not hasattr(Config, 'MINIO_USER') or not Config.MINIO_USER:
        errors.append("MINIO_USER未配置")
    if not hasattr(Config, 'MINIO_PASSWORD') or not Config.MINIO_PASSWORD:
        errors.append("MINIO_PASSWORD未配置")
    
    # 检查其他必要配置
    if not hasattr(Config, 'UPLOAD_FOLDER') or not Config.UPLOAD_FOLDER:
        errors.append("UPLOAD_FOLDER未配置")
    if not os.path.isdir(Config.UPLOAD_FOLDER):
        errors.append(f"UPLOAD_FOLDER目录不存在: {Config.UPLOAD_FOLDER}")
    
    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