""" 配置健康检查工具 用于验证配置文件的完整性和正确性 """ import sys from typing import List, Dict, Any from core.utils.config_manager import get_config_manager from core.utils.spider_config import SpiderConfig from config import settings class ConfigHealthCheck: """ 配置健康检查工具 """ def __init__(self): self.config_manager = get_config_manager() self.errors = [] self.warnings = [] def check_env_config(self) -> bool: """ 检查环境配置 """ try: # 检查必要配置是否存在 required_settings = [ 'DB_HOST', 'DB_USER', 'DB_PASSWORD', 'DB_NAME', 'ROCKETMQ_ENDPOINT', 'ROCKETMQ_ACCESS_KEY_ID', 'ROCKETMQ_ACCESS_KEY_SECRET', 'FEISHU_APPID', 'FEISHU_APPSECRET', 'ALIYUN_ACCESS_KEY_ID', 'ALIYUN_ACCESS_KEY_SECRET', 'REDIS_HOST', 'REDIS_PASSWORD' ] for setting in required_settings: if not getattr(settings, setting, None): self.errors.append(f"环境配置缺失: {setting}") # 检查URL格式 url_settings = ['ROCKETMQ_ENDPOINT'] for setting in url_settings: url = getattr(settings, setting, None) if url and not isinstance(url, str): self.errors.append(f"URL配置格式错误: {setting}") return len(self.errors) == 0 except Exception as e: self.errors.append(f"环境配置检查异常: {str(e)}") return False def check_spider_configs(self) -> bool: """ 检查所有爬虫配置 """ try: platforms = self.config_manager.list_platforms() if not platforms: self.warnings.append("未找到任何平台配置") return True valid_count = 0 for platform in platforms: try: config = self.config_manager.get_platform_config(platform) # 验证配置字段 if not config.platform: self.errors.append(f"平台 {platform} 缺少 platform 字段") if not config.mode: self.errors.append(f"平台 {platform} 缺少 mode 字段") if not config.url: self.errors.append(f"平台 {platform} 缺少 url 字段") valid_count += 1 except Exception as e: self.errors.append(f"平台 {platform} 配置验证失败: {str(e)}") return len(self.errors) == 0 except Exception as e: self.errors.append(f"爬虫配置检查异常: {str(e)}") return False def check_file_permissions(self) -> bool: """ 检查配置文件权限 """ import os from core.utils.path_utils import spiders_config_path try: # 检查爬虫配置文件是否存在 if not os.path.exists(spiders_config_path): self.errors.append(f"爬虫配置文件不存在: {spiders_config_path}") return False # 检查文件是否可读 if not os.access(spiders_config_path, os.R_OK): self.errors.append(f"爬虫配置文件不可读: {spiders_config_path}") return len(self.errors) == 0 except Exception as e: self.errors.append(f"文件权限检查异常: {str(e)}") return False def run_all_checks(self) -> Dict[str, Any]: """ 运行所有检查 """ self.errors.clear() self.warnings.clear() env_ok = self.check_env_config() spider_ok = self.check_spider_configs() file_ok = self.check_file_permissions() overall_ok = env_ok and spider_ok and file_ok return { "success": overall_ok, "errors": self.errors.copy(), "warnings": self.warnings.copy(), "details": { "env_config": env_ok, "spider_configs": spider_ok, "file_permissions": file_ok } } def print_report(self): """ 打印健康检查报告 """ result = self.run_all_checks() print("=" * 50) print("配置健康检查报告") print("=" * 50) if result["success"]: print("✓ 所有配置检查通过") else: print("✗ 配置存在问题") print(f"\n详细信息:") print(f" 环境配置: {'✓' if result['details']['env_config'] else '✗'}") print(f" 爬虫配置: {'✓' if result['details']['spider_configs'] else '✗'}") print(f" 文件权限: {'✓' if result['details']['file_permissions'] else '✗'}") if result["warnings"]: print(f"\n警告:") for warning in result["warnings"]: print(f" - {warning}") if result["errors"]: print(f"\n错误:") for error in result["errors"]: print(f" - {error}") print("\n统计信息:") try: stats = self.config_manager.get_config_stats() print(f" 平台数量: {stats['total_platforms']}") print(f" 运行环境: {stats['env']}") except Exception as e: print(f" 统计信息获取失败: {e}") print("=" * 50) return result def run_health_check(): """ 运行配置健康检查 """ checker = ConfigHealthCheck() return checker.print_report() if __name__ == "__main__": result = run_health_check() sys.exit(0 if result["success"] else 1)