""" 系统配置管理模块 提供系统配置的获取、验证和安全过滤功能 """ 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