#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 风七六平台 - 阿里云开发环境部署验证脚本 按照部署文档要求验证部署完成情况 符合大厂标准的自动化验证工具 """ import os import sys import subprocess import json import configparser from datetime import datetime class DeploymentVerifier: def __init__(self): self.config = self._load_config() self.results = [] self.passed = 0 self.failed = 0 def _load_config(self): """加载部署配置文件""" config = configparser.ConfigParser() config_path = os.path.join(os.path.dirname(__file__), 'deploy_config.ini') if os.path.exists(config_path): config.read(config_path) else: print(f"警告: 配置文件 {config_path} 不存在，使用默认配置") return config def _add_result(self, category, check_name, status, message): """添加验证结果""" self.results.append({ 'category': category, 'check_name': check_name, 'status': status, 'message': message }) if status == 'PASS': self.passed += 1 else: self.failed += 1 def _run_command(self, cmd, cwd=None): """执行命令并返回结果""" try: result = subprocess.run( cmd, shell=True, cwd=cwd, capture_output=True, text=True, timeout=30 ) return result.returncode, result.stdout.strip(), result.stderr.strip() except subprocess.TimeoutExpired: return -1, '', '命令执行超时' except Exception as e: return -1, '', str(e) def check_git_status(self): """检查Git仓库状态""" print("🔍 检查Git仓库状态...") # 检查当前分支 code, stdout, stderr = self._run_command('git branch --show-current', cwd='.') if code == 0: current_branch = stdout expected_branch = self.config.get('deploy', 'branch', fallback='feature/security-enhance') if current_branch == expected_branch: self._add_result('代码仓库', '当前分支', 'PASS', f'当前分支: {current_branch}') else: self._add_result('代码仓库', '当前分支', 'FAIL', f'期望分支: {expected_branch}, 实际分支: {current_branch}') else: self._add_result('代码仓库', '当前分支', 'FAIL', f'无法获取分支: {stderr}') # 检查是否有未提交的更改 code, stdout, stderr = self._run_command('git status --porcelain', cwd='.') if code == 0: if stdout == '': self._add_result('代码仓库', '工作目录状态', 'PASS', '工作目录干净，无未提交更改') else: self._add_result('代码仓库', '工作目录状态', 'WARN', f'存在未提交的更改:\n{stdout}') else: self._add_result('代码仓库', '工作目录状态', 'FAIL', f'无法检查状态: {stderr}') def check_version_file(self): """检查版本文件""" print("🔍 检查版本文件...") version_file = 'VERSION' if os.path.exists(version_file): with open(version_file, 'r', encoding='utf-8') as f: content = f.read().strip() if content.startswith('BASE_VERSION='): version = content.replace('BASE_VERSION=', '') self._add_result('版本管理', 'VERSION文件', 'PASS', f'版本号: v{version}') else: self._add_result('版本管理', 'VERSION文件', 'FAIL', 'VERSION文件格式不正确') else: self._add_result('版本管理', 'VERSION文件', 'FAIL', 'VERSION文件不存在') # 检查version.json json_file = 'version.json' if os.path.exists(json_file): try: with open(json_file, 'r') as f: data = json.load(f) required_fields = ['version', 'baseVersion', 'commit', 'branch', 'buildTime'] missing = [f for f in required_fields if f not in data] if not missing: self._add_result('版本管理', 'version.json', 'PASS', f'完整版本: {data.get("version")}') else: self._add_result('版本管理', 'version.json', 'FAIL', f'缺少字段: {", ".join(missing)}') except json.JSONDecodeError: self._add_result('版本管理', 'version.json', 'FAIL', 'JSON格式错误') else: self._add_result('版本管理', 'version.json', 'WARN', 'version.json不存在') def check_env_config(self): """检查环境配置文件""" print("🔍 检查环境配置文件...") env_path = self.config.get('paths', 'env_file', fallback='infrastructure/compose/.env.aliyun') if os.path.exists(env_path): self._add_result('配置文件', '环境变量文件', 'PASS', f'配置文件存在: {env_path}') with open(env_path, 'r', encoding='utf-8') as f: content = f.read() # 检查关键配置项 required_vars = [ 'MARIADB_HOST', 'MARIADB_PASSWORD', 'REDIS_HOST', 'REDIS_PASSWORD', 'NACOS_SERVER_ADDR', 'JWT_SECRET_KEY' ] missing_vars = [] for var in required_vars: if var not in content: missing_vars.append(var) if not missing_vars: self._add_result('配置文件', '关键配置项', 'PASS', '所有关键配置项已设置') else: self._add_result('配置文件', '关键配置项', 'FAIL', f'缺少配置项: {", ".join(missing_vars)}') else: self._add_result('配置文件', '环境变量文件', 'FAIL', f'配置文件不存在: {env_path}') def check_compose_file(self): """检查Compose文件""" print("🔍 检查Compose文件...") compose_path = self.config.get('paths', 'compose_file', fallback='infrastructure/compose/podman-compose.aliyun-dev.yml') if os.path.exists(compose_path): self._add_result('配置文件', 'Compose文件', 'PASS', f'Compose文件存在: {compose_path}') with open(compose_path, 'r') as f: content = f.read() # 检查必需服务 required_services = {'mariadb': 'fzx-aliyun-mariadb', 'redis': 'fzx-aliyun-redis', 'nacos': 'fzx-aliyun-nacos', 'api-gateway': 'fzx-aliyun-gateway'} for service, container_name in required_services.items(): if f'container_name: {container_name}' in content: self._add_result('配置文件', f'{service}服务配置', 'PASS', f'{service}服务配置存在') else: self._add_result('配置文件', f'{service}服务配置', 'FAIL', f'{service}服务配置缺失') else: self._add_result('配置文件', 'Compose文件', 'FAIL', f'Compose文件不存在: {compose_path}') def check_nginx_config(self): """检查Nginx配置""" print("🔍 检查Nginx配置...") nginx_conf_path = 'infrastructure/nginx/conf/aliyun-dev.conf' if os.path.exists(nginx_conf_path): self._add_result('Nginx配置', '配置文件', 'PASS', f'配置文件存在: {nginx_conf_path}') with open(nginx_conf_path, 'r') as f: content = f.read() # 检查域名配置 domains = ['platform.fengzhongxing.com', 'merchant.fengzhongxing.com', 'api.fengzhongxing.com', 'www.fengzhongxing.com'] for domain in domains: if domain in content: self._add_result('Nginx配置', f'{domain}配置', 'PASS', f'{domain}配置存在') else: self._add_result('Nginx配置', f'{domain}配置', 'WARN', f'{domain}配置缺失') else: self._add_result('Nginx配置', '配置文件', 'FAIL', f'配置文件不存在: {nginx_conf_path}') def check_frontend_builds(self): """检查前端构建产物""" print("🔍 检查前端构建产物...") frontend_dirs = [ 'infrastructure/nginx/html/platform', 'infrastructure/nginx/html/merchant', 'infrastructure/nginx/html/official' ] for frontend_dir in frontend_dirs: if os.path.exists(frontend_dir) and os.listdir(frontend_dir): self._add_result('前端构建', frontend_dir, 'PASS', '构建产物存在') else: self._add_result('前端构建', frontend_dir, 'WARN', '构建产物不存在或为空') def check_backend_builds(self): """检查后端构建产物""" print("🔍 检查后端构建产物...") services = ['api-gateway', 'user-service', 'auth-service', 'merchant-service', 'order-service', 'rider-service'] for service in services: jar_path = f'services/{service}/target/*.jar' code, stdout, stderr = self._run_command(f'ls {jar_path} 2>/dev/null | head -1', cwd='.') if code == 0 and stdout: self._add_result('后端构建', f'{service}.jar', 'PASS', f'JAR文件: {stdout}') else: self._add_result('后端构建', f'{service}.jar', 'WARN', 'JAR文件不存在') def check_scripts(self): """检查部署脚本""" print("🔍 检查部署脚本...") scripts = [ 'scripts/deploy_to_aliyun_enhanced.py', 'scripts/version-manager.ps1', 'scripts/version-manager.sh', 'scripts/check-config-consistency.ps1' ] for script in scripts: if os.path.exists(script): self._add_result('部署脚本', script, 'PASS', '脚本存在') else: self._add_result('部署脚本', script, 'FAIL', f'脚本不存在: {script}') def generate_report(self): """生成验证报告""" print("\n" + "="*80) print("📋 风七六平台 - 阿里云开发环境部署验证报告") print("="*80) print(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"环境: {self.config.get('deploy', 'branch', fallback='未知')}") print("-"*80) # 按类别分组 categories = {} for result in self.results: category = result['category'] if category not in categories: categories[category] = [] categories[category].append(result) for category, items in categories.items(): print(f"\n📌 {category}") print("-"*40) for item in items: status_icon = '✅' if item['status'] == 'PASS' else '❌' if item['status'] == 'FAIL' else '⚠️' print(f" {status_icon} {item['check_name']}: {item['message']}") print("\n" + "="*80) print(f"📊 验证结果: {self.passed} 通过, {self.failed} 失败, {len(self.results) - self.passed - self.failed} 警告") print("="*80) # 输出建议 if self.failed > 0: print("\n🔧 需要修复的问题:") for result in self.results: if result['status'] == 'FAIL': print(f" - {result['check_name']}: {result['message']}") print("\n📝 下一步建议:") print(" 1. 修复所有失败项") print(" 2. 处理警告项（如需要）") print(" 3. 运行版本管理工具更新版本号") print(" 4. 执行部署脚本进行部署") print(" 5. 部署完成后运行验证脚本确认状态") return self.failed == 0 def run(self): """执行所有验证检查""" print("🚀 开始验证阿里云开发环境部署配置...") print("="*80) self.check_git_status() self.check_version_file() self.check_env_config() self.check_compose_file() self.check_nginx_config() self.check_frontend_builds() self.check_backend_builds() self.check_scripts() return self.generate_report() if __name__ == '__main__': verifier = DeploymentVerifier() success = verifier.run() sys.exit(0 if success else 1)