#!/usr/bin/env python3 """ 数据处理流程完整测试 测试从原始数据到最终结果的完整流程 """ import os import sys import django import requests import json import time import pandas as pd 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 DataProcessingFlowTest: def __init__(self): self.test_results = { 'setup': {}, 'processing_steps': {}, 'data_validation': {}, 'api_tests': {} } def setup_test_data(self): """准备测试数据""" print("🔧 准备测试数据...") # 创建测试用的ATC对话数据 test_data = [ ("test_001", "CCA123 climb to flight level 350"), ("test_002", "CSN456 descend to flight level 280"), ("test_003", "CHH789 maintain heading 090 degrees"), ("test_004", "Air China 001 turn left heading 270"), ("test_005", "China Southern 888 contact approach 121.9"), ("test_006", "Hainan Airlines 7777 reduce speed to 250 knots"), ("test_007", "Spring Airlines 9999 cleared for takeoff runway 36L"), ("test_008", "Juneyao Airlines 1234 taxi to gate A15"), ("test_009", "Shanghai Airlines 5678 hold short of runway 18R"), ("test_010", "Tibet Airlines 9876 report when ready for departure") ] try: with connection.cursor() as cursor: # 确保prewashed_table存在 cursor.execute(""" CREATE TABLE IF NOT EXISTS prewashed_table ( id VARCHAR(255) NOT NULL, text TEXT, PRIMARY KEY (id) ) """) # 清空现有数据 cursor.execute("DELETE FROM prewashed_table") # 插入测试数据 for id_val, text_val in test_data: cursor.execute( "INSERT INTO prewashed_table (id, text) VALUES (%s, %s)", [id_val, text_val] ) # 验证插入的数据 cursor.execute("SELECT COUNT(*) FROM prewashed_table") count = cursor.fetchone()[0] print(f" ✓ 成功插入 {count} 条测试数据") # 显示插入的数据样例 cursor.execute("SELECT * FROM prewashed_table LIMIT 3") sample_data = cursor.fetchall() print(" 📋 数据样例:") for row in sample_data: print(f" ID: {row[0]}, Text: {row[1]}") self.test_results['setup'] = { 'success': True, 'data_count': count, 'sample_data': sample_data } return True except Exception as e: print(f" ✗ 测试数据准备失败: {e}") self.test_results['setup'] = { 'success': False, 'error': str(e) } return False def check_server_status(self): """检查Django服务器状态""" print("\n🚀 检查Django服务器状态...") try: response = requests.get(f"{API_BASE}/health/", timeout=5) if response.status_code == 200: print(" ✓ Django服务器运行正常") return True else: print(f" ✗ 服务器响应异常: {response.status_code}") return False except requests.RequestException as e: print(f" ✗ 无法连接到Django服务器: {e}") print(" 💡 请确保Django服务器正在运行: python manage.py runserver") return False def test_original_data_api(self): """测试原始数据获取API""" print("\n📊 测试原始数据获取API...") try: response = requests.get(f"{API_BASE}/original-data/", timeout=10) if response.status_code == 200: data = response.json() if data['status'] == 'success': count = data['count'] records = data['data'] print(f" ✓ 成功获取 {count} 条原始数据") print(" 📋 原始数据样例:") for i, record in enumerate(records[:3]): print(f" {i+1}. ID: {record['id']}, Text: {record['text'][:50]}...") self.test_results['api_tests']['original_data'] = { 'success': True, 'count': count, 'sample': records[:3] } return True else: print(f" ✗ API返回错误: {data.get('message', '未知错误')}") return False else: print(f" ✗ API请求失败: {response.status_code}") return False except Exception as e: print(f" ✗ 原始数据API测试失败: {e}") self.test_results['api_tests']['original_data'] = { 'success': False, 'error': str(e) } return False def test_preprocessing_step(self): """测试数据预处理步骤""" print("\n🔄 测试数据预处理步骤...") try: response = requests.post(f"{API_BASE}/preprocess/", json={}, headers={'Content-Type': 'application/json'}, timeout=10) if response.status_code == 200: data = response.json() if data['status'] == 'success': processed_count = data['data']['processed_count'] cleaning_rate = data['data']['cleaning_rate'] print(f" ✓ 预处理成功: 处理了 {processed_count} 条记录") print(f" 📊 数据清理率: {cleaning_rate}%") self.test_results['processing_steps']['preprocess'] = { 'success': True, 'processed_count': processed_count, 'cleaning_rate': cleaning_rate } 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['processing_steps']['preprocess'] = { 'success': False, 'error': str(e) } return False def test_merge_step(self): """测试格式合并步骤""" print("\n🔗 测试格式合并步骤...") try: response = requests.post(f"{API_BASE}/merge/", json={}, headers={'Content-Type': 'application/json'}, timeout=10) if response.status_code == 200: data = response.json() if data['status'] == 'success': merged_records = data['data']['merged_records'] success_rate = data['data']['merge_success_rate'] print(f" ✓ 格式合并成功: 合并了 {merged_records} 条记录") print(f" 📊 合并成功率: {success_rate}%") self.test_results['processing_steps']['merge'] = { 'success': True, 'merged_records': merged_records, 'success_rate': success_rate } 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['processing_steps']['merge'] = { 'success': False, 'error': str(e) } return False def test_correction_step(self): """测试单词纠错步骤""" print("\n📝 测试单词纠错步骤...") try: response = requests.post(f"{API_BASE}/correct/", json={}, headers={'Content-Type': 'application/json'}, timeout=10) if response.status_code == 200: data = response.json() if data['status'] == 'success': corrected_words = data['data']['corrected_words'] correction_types = data['data']['correction_types'] print(f" ✓ 单词纠错成功: 纠正了 {corrected_words} 个单词") print(f" 📋 纠错类型: {', '.join(correction_types)}") self.test_results['processing_steps']['correction'] = { 'success': True, 'corrected_words': corrected_words, 'correction_types': correction_types } 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['processing_steps']['correction'] = { 'success': False, 'error': str(e) } return False def test_ai_analysis_step(self): """测试AI分析步骤""" print("\n🤖 测试AI分析步骤...") try: response = requests.post(f"{API_BASE}/analyze/", json={}, headers={'Content-Type': 'application/json'}, timeout=30) # AI处理可能需要更长时间 if response.status_code == 200: data = response.json() if data['status'] == 'success': if 'analysis_summary' in data['data']: # 模拟分析结果 summary = data['data']['analysis_summary'] print(f" ✓ AI分析完成（模拟结果）") print(f" 📊 呼号提取: {summary.get('call_signs_extracted', 0)}") print(f" 📊 行为识别: {summary.get('behaviors_identified', 0)}") print(f" 📊 高度检测: {summary.get('flight_levels_detected', 0)}") else: # 实际AI处理结果 print(f" ✓ AI分析完成（实际处理）") print(f" 📊 处理结果: {data['data']}") self.test_results['processing_steps']['ai_analysis'] = { 'success': True, 'result': data['data'] } return True else: print(f" ✗ AI分析失败: {data.get('message', '未知错误')}") return False else: print(f" ✗ AI分析请求失败: {response.status_code}") return False except Exception as e: print(f" ✗ AI分析测试失败: {e}") self.test_results['processing_steps']['ai_analysis'] = { 'success': False, 'error': str(e) } return False def test_processed_data_api(self): """测试处理后数据获取API""" print("\n📋 测试处理后数据获取API...") try: response = requests.get(f"{API_BASE}/processed-data/", timeout=10) if response.status_code == 200: data = response.json() if data['status'] == 'success': count = data['count'] records = data['data'] print(f" ✓ 成功获取 {count} 条处理后数据") if records: print(" 📋 处理后数据样例:") for i, record in enumerate(records[:3]): print(f" {i+1}. 呼号: {record.get('Call Sign', 'N/A')}") print(f" 行为: {record.get('Behavior', 'N/A')}") print(f" 高度: {record.get('Flight Level', 'N/A')}") print(f" 位置: {record.get('Location', 'N/A')}") print() self.test_results['api_tests']['processed_data'] = { 'success': True, 'count': count, 'sample': records[:3] if records else [] } return True else: print(f" ⚠️ 处理后数据为空或获取失败: {data.get('message', '未知错误')}") return False else: print(f" ✗ 处理后数据API请求失败: {response.status_code}") return False except Exception as e: print(f" ✗ 处理后数据API测试失败: {e}") self.test_results['api_tests']['processed_data'] = { 'success': False, 'error': str(e) } return False def test_statistics_api(self): """测试统计信息API""" print("\n📊 测试统计信息API...") 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(f" ✓ 统计信息获取成功") 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)}%") self.test_results['api_tests']['statistics'] = { 'success': True, 'statistics': stats } return True else: print(f" ✗ 统计信息获取失败: {data.get('message', '未知错误')}") return False else: print(f" ✗ 统计信息API请求失败: {response.status_code}") return False except Exception as e: print(f" ✗ 统计信息API测试失败: {e}") self.test_results['api_tests']['statistics'] = { 'success': False, 'error': str(e) } return False def test_complete_processing_api(self): """测试完整处理流程API""" print("\n🚀 测试完整处理流程API...") try: response = requests.post(f"{API_BASE}/process-data/", timeout=60) # 完整处理可能需要较长时间 if response.status_code == 200: data = response.json() if data['status'] == 'success': print(f" ✓ 完整处理流程成功") # 显示处理结果 if 'processed_count' in data: print(f" 📊 处理数量: {data['processed_count']}") if 'valid_count' in data: print(f" 📊 有效数量: {data['valid_count']}") if 'invalid_count' in data: print(f" 📊 无效数量: {data['invalid_count']}") self.test_results['processing_steps']['complete_process'] = { 'success': True, 'result': data } return True else: print(f" ✗ 完整处理失败: {data.get('message', '未知错误')}") return False else: print(f" ✗ 完整处理API请求失败: {response.status_code}") return False except Exception as e: print(f" ✗ 完整处理API测试失败: {e}") self.test_results['processing_steps']['complete_process'] = { 'success': False, 'error': str(e) } return False def check_database_tables(self): """检查数据库表状态""" print("\n🗄️ 检查数据库表状态...") tables_to_check = [ 'prewashed_table', 'processed_table', 'precessed_table', # 旧的拼写错误表名 'final_table', 'quarantine_table' ] table_status = {} try: with connection.cursor() as cursor: for table in tables_to_check: try: cursor.execute(f"SELECT COUNT(*) FROM {table}") count = cursor.fetchone()[0] table_status[table] = {'exists': True, 'count': count} print(f" ✓ {table}: {count} 条记录") except Exception: table_status[table] = {'exists': False, 'count': 0} print(f" - {table}: 表不存在") self.test_results['data_validation']['table_status'] = table_status except Exception as e: print(f" ✗ 数据库表检查失败: {e}") self.test_results['data_validation']['table_status'] = { 'error': str(e) } def generate_test_report(self): """生成测试报告""" print("\n" + "="*80) print("📋 数据处理流程测试报告") print("="*80) # 测试数据准备 setup = self.test_results.get('setup', {}) if setup.get('success'): print(f"\n✅ 测试数据准备: 成功 ({setup.get('data_count', 0)} 条)") else: print(f"\n❌ 测试数据准备: 失败") # 处理步骤测试 steps = self.test_results.get('processing_steps', {}) print(f"\n🔄 处理步骤测试:") step_names = { 'preprocess': '数据预处理', 'merge': '格式合并', 'correction': '单词纠错', 'ai_analysis': 'AI分析', 'complete_process': '完整处理流程' } for step_key, step_name in step_names.items(): if step_key in steps: if steps[step_key].get('success'): print(f" ✅ {step_name}: 成功") else: print(f" ❌ {step_name}: 失败") else: print(f" ⏭️ {step_name}: 跳过") # API测试 api_tests = self.test_results.get('api_tests', {}) print(f"\n🌐 API测试:") api_names = { 'original_data': '原始数据获取', 'processed_data': '处理后数据获取', 'statistics': '统计信息获取' } for api_key, api_name in api_names.items(): if api_key in api_tests: if api_tests[api_key].get('success'): print(f" ✅ {api_name}: 成功") else: print(f" ❌ {api_name}: 失败") else: print(f" ⏭️ {api_name}: 跳过") # 数据验证 validation = self.test_results.get('data_validation', {}) if 'table_status' in validation: print(f"\n🗄️ 数据库表状态:") for table, status in validation['table_status'].items(): if status.get('exists'): print(f" ✅ {table}: {status.get('count', 0)} 条记录") else: print(f" ➖ {table}: 不存在") # 总体评估 total_tests = 0 passed_tests = 0 # 统计各类测试 if setup.get('success'): passed_tests += 1 total_tests += 1 for step_result in steps.values(): if step_result.get('success'): passed_tests += 1 total_tests += 1 for api_result in api_tests.values(): if api_result.get('success'): passed_tests += 1 total_tests += 1 success_rate = (passed_tests / total_tests * 100) if total_tests > 0 else 0 print(f"\n🎯 总体测试结果: {passed_tests}/{total_tests} 通过 ({success_rate:.1f}%)") if success_rate >= 80: print("🏆 优秀！数据处理流程运行良好") elif success_rate >= 60: print("✅ 良好！大部分功能正常") elif success_rate >= 40: print("⚠️ 一般，部分功能需要检查") else: print("❌ 需要修复多个问题") # 保存详细报告 report_file = Path(__file__).parent / "data_processing_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}") def run_complete_test(self): """运行完整的数据处理流程测试""" print("🎯 开始数据处理流程完整测试") print("="*80) # 1. 准备测试数据 if not self.setup_test_data(): print("❌ 测试数据准备失败，无法继续测试") return False # 2. 检查服务器状态 if not self.check_server_status(): print("❌ Django服务器未运行，无法进行API测试") print("💡 请先启动服务器: cd Django123/atc_extractor/backend && python manage.py runserver") self.check_database_tables() self.generate_test_report() return False # 3. 测试原始数据API self.test_original_data_api() # 4. 按顺序测试各个处理步骤 self.test_preprocessing_step() time.sleep(1) # 避免请求过快 self.test_merge_step() time.sleep(1) self.test_correction_step() time.sleep(1) self.test_ai_analysis_step() time.sleep(2) # 5. 测试完整处理流程 self.test_complete_processing_api() time.sleep(2) # 6. 测试处理结果API self.test_processed_data_api() time.sleep(1) # 7. 测试统计信息API self.test_statistics_api() # 8. 检查数据库表状态 self.check_database_tables() # 9. 生成测试报告 self.generate_test_report() return True def main(): """主函数""" tester = DataProcessingFlowTest() success = tester.run_complete_test() print("\n" + "="*80) if success: print("✅ 数据处理流程测试完成！") else: print("⚠️ 测试完成，但可能存在问题需要解决") return 0 if success else 1 if __name__ == "__main__": sys.exit(main())