#!/usr/bin/env python3 """ 修复验证测试 - 验证所有修复是否成功 """ import os import sys import requests import json import time from pathlib import Path # 项目路径 PROJECT_ROOT = Path(__file__).parent.parent DJANGO_ROOT = PROJECT_ROOT / "信息抽取+数据检验/Django123/atc_extractor/backend" API_BASE = "http://127.0.0.1:8000/api" class FixValidation: def __init__(self): self.results = { 'security_fixes': {}, 'bug_fixes': {}, 'architecture_improvements': {}, 'overall_score': 0 } def check_security_fixes(self): """检查安全修复""" print("🔒 检查安全修复...") # 1. 检查settings.py的环境变量配置 settings_file = DJANGO_ROOT / "backend/settings.py" if settings_file.exists(): with open(settings_file, 'r', encoding='utf-8') as f: content = f.read() # 检查是否使用环境变量 has_env_vars = 'get_env_variable' in content has_secret_key_gen = 'get_random_secret_key' in content has_debug_env = "get_env_variable('DEBUG'" in content self.results['security_fixes'] = { 'environment_variables': has_env_vars, 'dynamic_secret_key': has_secret_key_gen, 'debug_env_control': has_debug_env, 'score': sum([has_env_vars, has_secret_key_gen, has_debug_env]) } print(f" ✓ 环境变量配置: {'是' if has_env_vars else '否'}") print(f" ✓ 动态SECRET_KEY: {'是' if has_secret_key_gen else '否'}") print(f" ✓ DEBUG环境控制: {'是' if has_debug_env else '否'}") # 2. 检查.env.example文件 env_example = DJANGO_ROOT / ".env.example" if env_example.exists(): print(" ✓ .env.example文件已创建") self.results['security_fixes']['env_example'] = True else: print(" ✗ .env.example文件未找到") self.results['security_fixes']['env_example'] = False def check_bug_fixes(self): """检查bug修复""" print("\n🐛 检查bug修复...") # 检查表名拼写修复 views_file = DJANGO_ROOT / "extractor/views.py" services_file = DJANGO_ROOT / "extractor/services.py" table_name_fixes = 0 if views_file.exists(): with open(views_file, 'r', encoding='utf-8') as f: content = f.read() if 'processed_table' in content and 'precessed_table' in content: print(" ✓ views.py: 表名修复且保持向后兼容") table_name_fixes += 1 elif 'processed_table' in content: print(" ⚠ views.py: 表名已修复但可能缺少向后兼容") table_name_fixes += 0.5 if services_file.exists(): with open(services_file, 'r', encoding='utf-8') as f: content = f.read() if 'processed_table' in content and 'precessed_table' in content: print(" ✓ services.py: 表名修复且保持向后兼容") table_name_fixes += 1 elif 'processed_table' in content: print(" ⚠ services.py: 表名已修复但可能缺少向后兼容") table_name_fixes += 0.5 self.results['bug_fixes'] = { 'table_name_fixes': table_name_fixes, 'score': table_name_fixes } def check_architecture_improvements(self): """检查架构改进""" print("\n🏗️ 检查架构改进...") # 1. 检查requirements.txt req_file = DJANGO_ROOT / "requirements.txt" has_requirements = req_file.exists() print(f" ✓ requirements.txt: {'已创建' if has_requirements else '未创建'}") # 2. 检查models.py改进 models_file = DJANGO_ROOT / "extractor/models.py" has_models = False if models_file.exists(): with open(models_file, 'r', encoding='utf-8') as f: content = f.read() has_models = len(content) > 500 and 'class' in content print(f" ✓ models.py改进: {'已完成' if has_models else '未完成'}") # 3. 检查测试文件 test_files = [ PROJECT_ROOT / "test/comprehensive_backend_tests.py", DJANGO_ROOT / "extractor/test_views.py", PROJECT_ROOT / "test/run_tests.py" ] test_coverage = sum(1 for f in test_files if f.exists()) print(f" ✓ 测试覆盖: {test_coverage}/3 个测试文件") self.results['architecture_improvements'] = { 'requirements_file': has_requirements, 'models_enhancement': has_models, 'test_coverage': test_coverage, 'score': sum([has_requirements, has_models, test_coverage > 0]) } def test_api_functionality(self): """测试API功能是否正常""" print("\n🌐 测试API功能...") try: # 测试健康检查 response = requests.get(f"{API_BASE}/health/", timeout=5) if response.status_code == 200: print(" ✓ 健康检查API正常") api_health = True else: print(f" ✗ 健康检查API异常: {response.status_code}") api_health = False except Exception as e: print(f" ✗ API测试失败: {e}") api_health = False self.results['api_functionality'] = { 'health_check': api_health, 'score': 1 if api_health else 0 } return api_health def calculate_overall_score(self): """计算总体评分""" security_score = self.results['security_fixes'].get('score', 0) bug_score = self.results['bug_fixes'].get('score', 0) arch_score = self.results['architecture_improvements'].get('score', 0) api_score = self.results['api_functionality'].get('score', 0) # 权重计算 total_score = ( security_score * 0.3 + # 安全修复30% bug_score * 0.3 + # bug修复30% arch_score * 0.3 + # 架构改进30% api_score * 0.1 # API功能10% ) max_score = (3 * 0.3) + (2 * 0.3) + (3 * 0.3) + (1 * 0.1) # 最大可能分数 percentage = (total_score / max_score) * 100 self.results['overall_score'] = { 'total': total_score, 'max': max_score, 'percentage': percentage } return percentage def generate_report(self): """生成修复验证报告""" print("\n" + "="*60) print("📊 修复验证报告") print("="*60) # 安全修复 sec_fixes = self.results['security_fixes'] print(f"\n🔒 安全修复 ({sec_fixes.get('score', 0)}/3):") print(f" 环境变量配置: {'✓' if sec_fixes.get('environment_variables') else '✗'}") print(f" 动态SECRET_KEY: {'✓' if sec_fixes.get('dynamic_secret_key') else '✗'}") print(f" DEBUG环境控制: {'✓' if sec_fixes.get('debug_env_control') else '✗'}") # Bug修复 bug_fixes = self.results['bug_fixes'] print(f"\n🐛 Bug修复 ({bug_fixes.get('score', 0)}/2):") print(f" 表名拼写修复: {bug_fixes.get('table_name_fixes', 0)}/2 个文件") # 架构改进 arch_imp = self.results['architecture_improvements'] print(f"\n🏗️ 架构改进 ({arch_imp.get('score', 0)}/3):") print(f" requirements.txt: {'✓' if arch_imp.get('requirements_file') else '✗'}") print(f" models.py增强: {'✓' if arch_imp.get('models_enhancement') else '✗'}") print(f" 测试覆盖: {arch_imp.get('test_coverage', 0)}/3") # API功能 api_func = self.results['api_functionality'] print(f"\n🌐 API功能 ({api_func.get('score', 0)}/1):") print(f" 健康检查: {'✓' if api_func.get('health_check') else '✗'}") # 总体评分 overall = self.results['overall_score'] percentage = overall.get('percentage', 0) print(f"\n🎯 总体评分: {percentage:.1f}%") if percentage >= 90: print("🏆 优秀！所有关键问题已修复") elif percentage >= 70: print("✅ 良好！大部分问题已修复") elif percentage >= 50: print("⚠️ 一般，还有部分问题需要修复") else: print("❌ 需要继续修复更多问题") # 保存详细报告 report_file = PROJECT_ROOT / "test/fix_validation_report.json" with open(report_file, 'w', encoding='utf-8') as f: json.dump(self.results, f, indent=2, ensure_ascii=False) print(f"\n📄 详细报告已保存: {report_file}") def run_validation(self): """运行完整验证""" print("🔍 开始修复验证...") self.check_security_fixes() self.check_bug_fixes() self.check_architecture_improvements() # 检查服务器是否运行 try: requests.get(f"{API_BASE}/health/", timeout=3) self.test_api_functionality() except: print("\n⚠️ Django服务器未运行，跳过API测试") self.results['api_functionality'] = {'health_check': False, 'score': 0} self.calculate_overall_score() self.generate_report() def main(): validator = FixValidation() validator.run_validation() score = validator.results['overall_score'].get('percentage', 0) return 0 if score >= 70 else 1 # 70分以上算成功 if __name__ == "__main__": sys.exit(main())