#!/usr/bin/env python3 """ 完整端到端流程测试 验证从数据上传到最终结果的完整数据处理链路 """ import os import sys import django import requests import json import time import io from pathlib import Path # 设置Django环境 sys.path.append('/home/hzk/项目/moxun-1/信息抽取+数据检验/Django123/atc_extractor/backend') os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'backend.settings') django.setup() from django.db import connection API_BASE = "http://127.0.0.1:8080/api" class CompleteEndToEndTest: def __init__(self): self.test_results = { 'system_health': {}, 'data_upload': {}, 'processing_pipeline': {}, 'data_retrieval': {}, 'final_verification': {} } self.test_data_file = None def step_1_system_health_check(self): """步骤1: 系统健康检查""" print("🚀 步骤1: 系统健康检查") print("-" * 50) try: # 检查Django服务器 response = requests.get(f"{API_BASE}/health/", timeout=5) if response.status_code == 200: data = response.json() print(f" ✓ Django服务器: {data['message']}") # 检查数据库连接 with connection.cursor() as cursor: cursor.execute("SELECT 1") cursor.fetchone() print(" ✓ 数据库连接: 正常") self.test_results['system_health'] = { 'django_server': True, 'database': True, 'status': 'healthy' } return True else: print(f" ✗ Django服务器响应异常: {response.status_code}") return False except Exception as e: print(f" ✗ 系统健康检查失败: {e}") self.test_results['system_health'] = { 'status': 'failed', 'error': str(e) } return False def step_2_prepare_test_data(self): """步骤2: 准备测试数据文件""" print("\n📁 步骤2: 准备测试数据文件") print("-" * 50) # 创建真实的ATC对话CSV测试文件 test_data_content = """id,text e2e_001,"CCA123 climb to flight level 350" e2e_002,"China Southern 456 descend to flight level 280" e2e_003,"Air China 789 turn left heading 270 degrees" e2e_004,"Hainan Airlines 888 contact Shanghai approach 121.9" e2e_005,"Spring Airlines 999 cleared for takeoff runway 36L" e2e_006,"United 997 maintain flight level 330" e2e_007,"Lufthansa 672 descending to flight level 250" e2e_008,"Singapore Airlines 106 approach runway 18R" e2e_009,"Japan Airlines 550 hold position" e2e_010,"KLM 695 taxi to gate A15" e2e_011,"Emirates 203 climb and maintain flight level 380" e2e_012,"Cathay Pacific 747 contact tower 118.1" e2e_013,"Turkish Airlines 1 turn right heading 090" e2e_014,"Delta 88 cleared to land runway 24L" e2e_015,"British Airways 15 hold short of runway 06R" """ try: # 保存测试文件 self.test_data_file = Path(__file__).parent / "e2e_test_data.csv" with open(self.test_data_file, 'w', encoding='utf-8') as f: f.write(test_data_content) print(f" ✓ 测试文件创建: {self.test_data_file}") print(f" 📊 测试数据: 15条ATC对话记录") print(" 📋 数据样例:") lines = test_data_content.strip().split('\n') for i, line in enumerate(lines[1:4], 1): # 显示前3行数据 parts = line.split(',', 1) if len(parts) >= 2: text = parts[1].strip('"') print(f" {i}. {text}") self.test_results['data_upload'] = { 'file_created': True, 'record_count': 15, 'file_path': str(self.test_data_file) } return True except Exception as e: print(f" ✗ 测试文件创建失败: {e}") self.test_results['data_upload'] = { 'file_created': False, 'error': str(e) } return False def step_3_file_upload(self): """步骤3: 文件上传测试""" print("\n📤 步骤3: 文件上传测试") print("-" * 50) try: # 上传文件到API with open(self.test_data_file, 'rb') as f: files = {'file': ('e2e_test_data.csv', f, 'text/csv')} response = requests.post(f"{API_BASE}/upload/", files=files, timeout=30) if response.status_code == 200: data = response.json() if data['status'] == 'success': print(f" ✓ 文件上传成功") print(f" 📊 处理记录数: {data['data']['inserted_rows']}") print(f" 📁 文件名: {data['data']['filename']}") # 验证数据是否正确插入数据库 with connection.cursor() as cursor: cursor.execute("SELECT COUNT(*) FROM prewashed_table") db_count = cursor.fetchone()[0] print(f" 📋 数据库记录数: {db_count}") # 显示插入的数据样例 cursor.execute("SELECT id, text FROM prewashed_table LIMIT 3") samples = cursor.fetchall() print(" 📋 数据库样例:") for id_val, text_val in samples: print(f" {id_val}: {text_val}") self.test_results['data_upload'].update({ 'upload_success': True, 'inserted_count': data['data']['inserted_rows'], 'db_verified_count': db_count }) return True else: print(f" ✗ 文件上传失败: {data.get('message', '未知错误')}") return False else: print(f" ✗ 上传请求失败: {response.status_code}") return False except Exception as e: print(f" ✗ 文件上传过程失败: {e}") self.test_results['data_upload'].update({ 'upload_success': False, 'error': str(e) }) return False def step_4_processing_pipeline(self): """步骤4: 数据处理流水线测试""" print("\n🔄 步骤4: 数据处理流水线测试") print("-" * 50) pipeline_steps = [ ('预处理', 'preprocess'), ('格式合并', 'merge'), ('单词纠错', 'correct'), ('AI分析', 'analyze') ] pipeline_results = {} for step_name, endpoint in pipeline_steps: try: print(f" 🔧 执行{step_name}...") response = requests.post(f"{API_BASE}/{endpoint}/", json={}, headers={'Content-Type': 'application/json'}, timeout=30) if response.status_code == 200: data = response.json() if data['status'] == 'success': print(f" ✓ {step_name}成功") # 根据不同步骤显示不同的结果信息 if endpoint == 'preprocess': processed_count = data['data']['processed_count'] cleaning_rate = data['data']['cleaning_rate'] print(f" 📊 处理记录: {processed_count}, 清理率: {cleaning_rate}%") elif endpoint == 'merge': merged_records = data['data']['merged_records'] success_rate = data['data']['merge_success_rate'] print(f" 📊 合并记录: {merged_records}, 成功率: {success_rate}%") elif endpoint == 'correct': corrected_words = data['data']['corrected_words'] print(f" 📊 纠正单词: {corrected_words}个") elif endpoint == 'analyze': if 'analysis_summary' in data['data']: # 模拟分析结果 summary = data['data']['analysis_summary'] print(f" 📊 呼号提取: {summary.get('call_signs_extracted', 0)}") print(f" 📊 行为识别: {summary.get('behaviors_identified', 0)}") else: # 实际AI分析结果 print(f" 📊 AI分析完成") pipeline_results[endpoint] = { 'success': True, 'data': data['data'] } else: print(f" ✗ {step_name}失败: {data.get('message', '未知错误')}") pipeline_results[endpoint] = { 'success': False, 'error': data.get('message', '未知错误') } else: print(f" ✗ {step_name}请求失败: {response.status_code}") pipeline_results[endpoint] = { 'success': False, 'error': f"HTTP {response.status_code}" } # 步骤间暂停 time.sleep(1) except Exception as e: print(f" ✗ {step_name}异常: {e}") pipeline_results[endpoint] = { 'success': False, 'error': str(e) } self.test_results['processing_pipeline'] = pipeline_results # 检查是否所有步骤都成功 all_success = all(result.get('success', False) for result in pipeline_results.values()) if all_success: print(f"\n 🎉 数据处理流水线全部完成！") return True else: failed_steps = [step for step, result in pipeline_results.items() if not result.get('success', False)] print(f"\n ⚠️ 部分步骤失败: {', '.join(failed_steps)}") return False def step_5_complete_processing(self): """步骤5: 完整处理流程测试""" print("\n🚀 步骤5: 完整处理流程测试") print("-" * 50) try: print(" 🔧 执行完整处理流程...") response = requests.post(f"{API_BASE}/process-data/", timeout=60) if response.status_code == 200: data = response.json() if data['status'] == 'success': print(" ✓ 完整处理流程成功") print(f" 📊 原始记录: {data.get('original_count', 0)}") print(f" 📊 提取结果: {data.get('extracted_count', 0)}") print(f" 📊 处理记录: {data.get('processed_count', 0)}") print(f" 📊 有效记录: {data.get('valid_count', 0)}") print(f" 📊 无效记录: {data.get('invalid_count', 0)}") print(f" 📊 提取率: {data.get('extraction_rate', 0)}%") print(f" 📊 验证率: {data.get('validation_rate', 0)}%") self.test_results['processing_pipeline']['complete_process'] = { 'success': True, 'metrics': { 'original_count': data.get('original_count', 0), 'extracted_count': data.get('extracted_count', 0), 'processed_count': data.get('processed_count', 0), 'valid_count': data.get('valid_count', 0), 'invalid_count': data.get('invalid_count', 0), 'extraction_rate': data.get('extraction_rate', 0), 'validation_rate': data.get('validation_rate', 0) } } return True else: print(f" ✗ 完整处理失败: {data.get('message', '未知错误')}") return False else: print(f" ✗ 完整处理请求失败: {response.status_code}") return False except Exception as e: print(f" ✗ 完整处理异常: {e}") return False def step_6_data_retrieval_test(self): """步骤6: 数据获取测试""" print("\n📋 步骤6: 数据获取API测试") print("-" * 50) data_apis = [ ('原始数据', 'original-data'), ('处理数据', 'processed-data'), ('最终数据', 'final-data'), ('隔离数据', 'quarantine-data'), ('统计信息', 'statistics') ] retrieval_results = {} for api_name, endpoint in data_apis: try: print(f" 📊 测试{api_name}API...") response = requests.get(f"{API_BASE}/{endpoint}/", timeout=10) if response.status_code == 200: data = response.json() if data['status'] == 'success': count = data.get('count', 0) print(f" ✓ {api_name}: {count} 条记录") # 显示数据样例 if endpoint == 'statistics': stats = data.get('statistics', {}) print(f" 📈 提取率: {stats.get('extraction_rate', 0)}%") print(f" 📈 验证率: {stats.get('validation_rate', 0)}%") elif count > 0 and 'data' in data: sample_count = min(3, len(data['data'])) print(f" 📋 样例数据 (显示{sample_count}条):") for i, record in enumerate(data['data'][:sample_count], 1): if endpoint in ['processed-data', 'final-data']: call_sign = record.get('call_sign', 'N/A') behavior = record.get('behavior', 'N/A') print(f" {i}. 呼号: {call_sign} | 行为: {behavior}") elif endpoint == 'original-data': text = record.get('text', '')[:50] + '...' if len(record.get('text', '')) > 50 else record.get('text', '') print(f" {i}. {record.get('id', 'N/A')}: {text}") retrieval_results[endpoint] = { 'success': True, 'count': count, 'data_sample': data.get('data', [])[:3] if 'data' in data else [] } else: print(f" ⚠️ {api_name}: {data.get('message', '无数据')}") retrieval_results[endpoint] = { 'success': False, 'message': data.get('message', '无数据') } else: print(f" ✗ {api_name}请求失败: {response.status_code}") retrieval_results[endpoint] = { 'success': False, 'error': f"HTTP {response.status_code}" } except Exception as e: print(f" ✗ {api_name}异常: {e}") retrieval_results[endpoint] = { 'success': False, 'error': str(e) } self.test_results['data_retrieval'] = retrieval_results # 检查关键API是否成功 critical_apis = ['original-data', 'processed-data', 'statistics'] critical_success = all(retrieval_results.get(api, {}).get('success', False) for api in critical_apis) if critical_success: print(f"\n 🎉 数据获取API测试完成！") return True else: print(f"\n ⚠️ 部分关键API测试失败") return False def step_7_final_verification(self): """步骤7: 最终验证""" print("\n✅ 步骤7: 最终系统验证") print("-" * 50) try: # 获取最终统计信息 response = requests.get(f"{API_BASE}/statistics/", timeout=10) if response.status_code == 200: data = response.json() if data['status'] == 'success': stats = data['statistics'] print(" 📊 最终系统状态:") print(f" 原始数据: {stats.get('original_count', 0)} 条") print(f" 提取数据: {stats.get('extracted_count', 0)} 条") print(f" 有效数据: {stats.get('valid_count', 0)} 条") print(f" 无效数据: {stats.get('invalid_count', 0)} 条") print(f" 提取效率: {stats.get('extraction_rate', 0)}%") print(f" 验证通过率: {stats.get('validation_rate', 0)}%") # 计算系统健康度 extraction_rate = stats.get('extraction_rate', 0) validation_rate = stats.get('validation_rate', 0) data_quality = (extraction_rate + validation_rate) / 2 print(f" 系统健康度: {data_quality:.1f}%") # 验证数据库表状态 with connection.cursor() as cursor: tables_status = {} for table in ['prewashed_table', 'processed_table', 'final_table', 'quarantine_table']: try: cursor.execute(f"SELECT COUNT(*) FROM {table}") count = cursor.fetchone()[0] tables_status[table] = count print(f" {table}: {count} 条") except Exception: tables_status[table] = 0 print(f" {table}: 不存在或为空") self.test_results['final_verification'] = { 'success': True, 'statistics': stats, 'tables_status': tables_status, 'system_health': data_quality } return True else: print(f" ✗ 统计信息获取失败: {data.get('message', '未知错误')}") return False else: print(f" ✗ 最终验证请求失败: {response.status_code}") return False except Exception as e: print(f" ✗ 最终验证异常: {e}") self.test_results['final_verification'] = { 'success': False, 'error': str(e) } return False def generate_comprehensive_report(self): """生成综合测试报告""" print("\n" + "="*80) print("📋 完整端到端测试 - 综合报告") print("="*80) # 测试步骤总结 test_steps = [ ('系统健康检查', self.test_results.get('system_health', {})), ('数据文件上传', self.test_results.get('data_upload', {})), ('处理流水线', self.test_results.get('processing_pipeline', {})), ('数据获取API', self.test_results.get('data_retrieval', {})), ('最终验证', self.test_results.get('final_verification', {})) ] passed_steps = 0 total_steps = len(test_steps) print(f"\n🔍 测试步骤总结:") for step_name, step_result in test_steps: if step_result.get('success', False) or step_result.get('status') == 'healthy': print(f" ✅ {step_name}: 通过") passed_steps += 1 elif step_result: print(f" ❌ {step_name}: 失败") else: print(f" ⏭️ {step_name}: 跳过") success_rate = (passed_steps / total_steps * 100) if total_steps > 0 else 0 print(f"\n📊 总体成功率: {passed_steps}/{total_steps} ({success_rate:.1f}%)") # 性能指标总结 final_verification = self.test_results.get('final_verification', {}) if final_verification.get('success'): stats = final_verification.get('statistics', {}) print(f"\n📈 性能指标总结:") print(f" 数据处理效率: {stats.get('extraction_rate', 0)}%") print(f" 数据验证率: {stats.get('validation_rate', 0)}%") print(f" 系统健康度: {final_verification.get('system_health', 0):.1f}%") # 数据流总结 print(f"\n🔄 数据流总结:") print(f" 原始数据 → 提取数据 → 有效数据") print(f" {stats.get('original_count', 0)} → {stats.get('extracted_count', 0)} → {stats.get('valid_count', 0)}") # 系统评级 print(f"\n🏆 系统评级:") if success_rate >= 90: grade = "A+" status = "🏆 优秀！系统已达到生产级别" elif success_rate >= 80: grade = "A" status = "✅ 良好！系统运行稳定" elif success_rate >= 70: grade = "B+" status = "📈 可用！需要少量优化" elif success_rate >= 60: grade = "B" status = "⚠️ 基本可用，需要改进" else: grade = "C" status = "❌ 需要重大修复" print(f" 等级: {grade}") print(f" 状态: {status}") # 保存详细报告 report_file = Path(__file__).parent / "complete_e2e_test_report.json" with open(report_file, 'w', encoding='utf-8') as f: json.dump(self.test_results, f, indent=2, ensure_ascii=False) print(f"\n📄 详细报告已保存: {report_file}") return success_rate >= 80 def cleanup(self): """清理测试文件""" try: if self.test_data_file and self.test_data_file.exists(): self.test_data_file.unlink() print(f"🗑️ 清理测试文件: {self.test_data_file}") except Exception as e: print(f"⚠️ 清理文件失败: {e}") def run_complete_test(self): """运行完整的端到端测试""" print("🎯 开始完整端到端流程测试") print("="*80) print("📝 测试范围: 数据上传 → 处理流水线 → AI分析 → 数据获取 → 验证") print("="*80) try: # 执行所有测试步骤 steps = [ self.step_1_system_health_check, self.step_2_prepare_test_data, self.step_3_file_upload, self.step_4_processing_pipeline, self.step_5_complete_processing, self.step_6_data_retrieval_test, self.step_7_final_verification ] for step in steps: if not step(): print(f"\n⚠️ 测试步骤失败，但继续执行后续步骤...") time.sleep(1) # 步骤间暂停 # 生成综合报告 success = self.generate_comprehensive_report() return success except KeyboardInterrupt: print(f"\n⏹️ 测试被用户中断") return False except Exception as e: print(f"\n❌ 测试过程发生异常: {e}") return False finally: self.cleanup() def main(): """主函数""" tester = CompleteEndToEndTest() success = tester.run_complete_test() print("\n" + "="*80) if success: print("🎉 完整端到端测试成功完成！") print("✅ 系统已准备好投入生产使用") else: print("⚠️ 端到端测试完成，但存在需要关注的问题") print("💡 请查看详细报告了解具体问题") return 0 if success else 1 if __name__ == "__main__": sys.exit(main())