#!/usr/bin/env python3 """ CAE Mesh Generator - Deployment Check Script 部署前系统检查脚本 """ import os import sys import subprocess import importlib import json import tempfile from pathlib import Path from datetime import datetime class DeploymentChecker: def __init__(self): self.project_root = Path(__file__).parent.parent # 上一级目录 self.checks_passed = 0 self.checks_failed = 0 self.warnings = [] self.errors = [] def print_header(self): """打印检查头部信息""" print("=" * 70) print("🔍 CAE网格生成助手 - 部署检查") print("=" * 70) print(f"检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"项目目录: {self.project_root}") print("=" * 70) def check_python_environment(self): """检查Python环境""" print("\n📋 检查Python环境...") # Python版本检查 version = sys.version_info if version.major < 3 or (version.major == 3 and version.minor < 8): self.add_error(f"Python版本过低: {version.major}.{version.minor}.{version.micro}") self.add_error("需要Python 3.8或更高版本") return False else: print(f"✅ Python版本: {version.major}.{version.minor}.{version.micro}") # 虚拟环境检查 if hasattr(sys, 'real_prefix') or (hasattr(sys, 'base_prefix') and sys.base_prefix != sys.prefix): print("✅ 运行在虚拟环境中") else: self.add_warning("未使用虚拟环境，建议使用虚拟环境") return True def check_dependencies(self): """检查依赖包""" print("\n📦 检查依赖包...") # 读取requirements.txt requirements_file = self.project_root / 'requirements.txt' if not requirements_file.exists(): self.add_error("requirements.txt文件不存在") return False # 解析依赖 required_packages = [] try: with open(requirements_file, 'r') as f: for line in f: line = line.strip() if line and not line.startswith('#'): package = line.split('==')[0].split('>=')[0].split('<=')[0] required_packages.append(package) except Exception as e: self.add_error(f"读取requirements.txt失败: {e}") return False # 检查每个包 missing_packages = [] for package in required_packages: try: importlib.import_module(package.replace('-', '_')) print(f"✅ {package}: 已安装") except ImportError: print(f"❌ {package}: 未安装") missing_packages.append(package) if missing_packages: self.add_error(f"缺少依赖包: {', '.join(missing_packages)}") self.add_error("请运行: pip install -r requirements.txt") return False return True def check_project_structure(self): """检查项目结构""" print("\n📁 检查项目结构...") required_files = [ 'app.py', 'config.py', 'requirements.txt', 'README.md' ] required_dirs = [ 'backend', 'backend/api', 'backend/core', 'backend/models', 'backend/utils', 'backend/pymechanical', 'templates', 'static', 'static/css', 'static/js' ] # 检查必需文件 for file_path in required_files: full_path = self.project_root / file_path if full_path.exists(): print(f"✅ {file_path}") else: print(f"❌ {file_path}") self.add_error(f"缺少必需文件: {file_path}") # 检查必需目录 for dir_path in required_dirs: full_path = self.project_root / dir_path if full_path.exists() and full_path.is_dir(): print(f"✅ {dir_path}/") else: print(f"❌ {dir_path}/") self.add_error(f"缺少必需目录: {dir_path}") return len(self.errors) == 0 def check_configuration(self): """检查配置文件""" print("\n⚙️ 检查配置...") try: from config import FLASK_CONFIG, ANSYS_CONFIG, ALLOWED_EXTENSIONS # 检查Flask配置 required_flask_keys = ['MAX_CONTENT_LENGTH', 'UPLOAD_FOLDER', 'RESULT_FOLDER'] for key in required_flask_keys: if key in FLASK_CONFIG: print(f"✅ Flask配置 {key}: {FLASK_CONFIG[key]}") else: self.add_warning(f"Flask配置缺少 {key}") # 检查ANSYS配置 if ANSYS_CONFIG: print(f"✅ ANSYS配置已定义") else: self.add_warning("ANSYS配置为空，将使用默认设置") # 检查允许的文件扩展名 if ALLOWED_EXTENSIONS: print(f"✅ 允许的文件扩展名: {ALLOWED_EXTENSIONS}") else: self.add_error("未定义允许的文件扩展名") except ImportError as e: self.add_error(f"无法导入配置: {e}") return False except Exception as e: self.add_error(f"配置检查失败: {e}") return False return True def check_directories(self): """检查运行时目录""" print("\n📂 检查运行时目录...") runtime_dirs = [ 'uploads', 'results', 'temp', 'logs', 'static/visualizations' ] for dir_name in runtime_dirs: dir_path = self.project_root / dir_name try: dir_path.mkdir(parents=True, exist_ok=True) if os.access(dir_path, os.W_OK): print(f"✅ {dir_name}/ (可写)") else: print(f"⚠️ {dir_name}/ (只读)") self.add_warning(f"目录 {dir_name} 不可写") except Exception as e: print(f"❌ {dir_name}/ (创建失败)") self.add_error(f"无法创建目录 {dir_name}: {e}") return True def check_flask_app(self): """检查Flask应用""" print("\n🌐 检查Flask应用...") try: # 添加项目根目录到Python路径 sys.path.insert(0, str(self.project_root)) from app import create_app app = create_app() print("✅ Flask应用创建成功") # 检查路由 routes = [] for rule in app.url_map.iter_rules(): routes.append(f"{rule.methods} {rule.rule}") print(f"✅ 注册路由数量: {len(routes)}") # 检查关键路由 key_routes = [ 'GET /', 'POST /api/upload', 'GET /api/mesh/status', 'POST /api/mesh/generate', 'GET /api/mesh/result' ] for route in key_routes: if any(route in r for r in routes): print(f"✅ 关键路由: {route}") else: print(f"❌ 缺少路由: {route}") self.add_error(f"缺少关键路由: {route}") except Exception as e: self.add_error(f"Flask应用检查失败: {e}") return False return True def check_ansys_availability(self): """检查ANSYS可用性""" print("\n🔧 检查ANSYS可用性...") try: import ansys.mechanical.core as mech print("✅ ANSYS Mechanical Core 可用") # 尝试获取版本信息 try: # 这里可以添加更详细的ANSYS检查 print("✅ ANSYS集成模块正常") except Exception as e: self.add_warning(f"ANSYS详细检查失败: {e}") except ImportError: print("⚠️ ANSYS Mechanical Core 不可用") self.add_warning("ANSYS不可用，系统将在演示模式下运行") return True def check_port_availability(self, port=5000): """检查端口可用性""" print(f"\n🌐 检查端口 {port} 可用性...") import socket try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(1) result = s.connect_ex(('localhost', port)) if result == 0: print(f"⚠️ 端口 {port} 已被占用") self.add_warning(f"端口 {port} 被占用，可能需要更改配置") else: print(f"✅ 端口 {port} 可用") except Exception as e: self.add_warning(f"端口检查失败: {e}") return True def check_disk_space(self): """检查磁盘空间""" print("\n💾 检查磁盘空间...") try: import shutil total, used, free = shutil.disk_usage(self.project_root) free_gb = free / (1024**3) total_gb = total / (1024**3) print(f"✅ 总空间: {total_gb:.1f} GB") print(f"✅ 可用空间: {free_gb:.1f} GB") if free_gb < 1: self.add_error("磁盘空间不足 (< 1GB)") return False elif free_gb < 5: self.add_warning("磁盘空间较少 (< 5GB)") except Exception as e: self.add_warning(f"磁盘空间检查失败: {e}") return True def run_basic_tests(self): """运行基本测试""" print("\n🧪 运行基本测试...") try: # 测试数据模型 from backend.models.data_models import UploadedFile, ProcessingStatus, MeshResult from datetime import datetime # 测试UploadedFile test_file = UploadedFile( id='test', filename='test.step', file_path='/tmp/test.step', upload_time=datetime.now(), status='UPLOADED' ) test_dict = test_file.to_dict() print("✅ UploadedFile模型测试通过") # 测试ProcessingStatus test_status = ProcessingStatus( status='PROCESSING', message='Test processing', start_time=datetime.now() ) status_dict = test_status.to_dict() print("✅ ProcessingStatus模型测试通过") # 测试MeshResult test_result = MeshResult( element_count=1000, node_count=1500, quality_score=85.0, quality_status='GOOD', generation_time=60.0 ) result_dict = test_result.to_dict() print("✅ MeshResult模型测试通过") except Exception as e: self.add_error(f"基本测试失败: {e}") return False return True def add_error(self, message): """添加错误""" self.errors.append(message) self.checks_failed += 1 def add_warning(self, message): """添加警告""" self.warnings.append(message) def print_summary(self): """打印检查总结""" print("\n" + "=" * 70) print("📊 检查结果总结") print("=" * 70) total_checks = self.checks_passed + self.checks_failed if total_checks > 0: success_rate = (self.checks_passed / total_checks) * 100 print(f"检查通过率: {success_rate:.1f}%") print(f"错误数量: {len(self.errors)}") print(f"警告数量: {len(self.warnings)}") if self.errors: print("\n❌ 错误列表:") for i, error in enumerate(self.errors, 1): print(f" {i}. {error}") if self.warnings: print("\n⚠️ 警告列表:") for i, warning in enumerate(self.warnings, 1): print(f" {i}. {warning}") print("\n" + "=" * 70) if len(self.errors) == 0: print("🎉 所有检查通过！系统已准备好部署。") return True else: print("❌ 存在错误，请修复后重新检查。") return False def run_all_checks(self): """运行所有检查""" self.print_header() checks = [ ("Python环境", self.check_python_environment), ("依赖包", self.check_dependencies), ("项目结构", self.check_project_structure), ("配置文件", self.check_configuration), ("运行时目录", self.check_directories), ("Flask应用", self.check_flask_app), ("ANSYS可用性", self.check_ansys_availability), ("端口可用性", self.check_port_availability), ("磁盘空间", self.check_disk_space), ("基本测试", self.run_basic_tests) ] for check_name, check_func in checks: try: if check_func(): self.checks_passed += 1 else: self.checks_failed += 1 except Exception as e: self.add_error(f"{check_name}检查异常: {e}") self.checks_failed += 1 return self.print_summary() def main(): """主函数""" checker = DeploymentChecker() success = checker.run_all_checks() return 0 if success else 1 if __name__ == '__main__': sys.exit(main())