#!/usr/bin/env python3 """ 中间处理过程测试文件 测试完整的数据处理流水线，展示每个中间步骤的处理结果 """ import requests import json import time from datetime import datetime API_BASE = "http://127.0.0.1:8080/api" class IntermediateProcessTest: def __init__(self): self.test_results = { 'pipeline_steps': [], 'data_samples': {}, 'processing_summary': {} } def print_header(self, step_num, title, description=""): """打印步骤标题""" print(f"\n{'='*80}") print(f"🎯 步骤 {step_num}: {title}") if description: print(f"📝 说明: {description}") print(f"{'='*80}") def print_data_sample(self, data, max_items=3, max_text_length=100): """打印数据样例""" if not data: print(" 📋 暂无数据") return print(f" 📋 数据样例 (显示前{min(max_items, len(data))}条):") for i, item in enumerate(data[:max_items], 1): if isinstance(item, dict): # 显示关键字段 key_fields = ['id', 'text', 'original_text', 'cleaned_text', 'preprocessed_text', 'merged_text', 'corrected_text', 'call_sign', 'behavior', 'Call Sign', 'Behavior'] displayed_fields = [] for field in key_fields: if field in item and item[field]: value = str(item[field]) if len(value) > max_text_length: value = value[:max_text_length] + "..." displayed_fields.append(f"{field}: {value}") if displayed_fields: print(f" {i}. {' | '.join(displayed_fields[:2])}") else: print(f" {i}. {item}") else: print(f" {i}. {item}") def test_system_health(self): """检查系统健康状态""" self.print_header(0, "系统健康检查", "确认Django服务器正常运行") try: response = requests.get(f"{API_BASE}/health/", timeout=5) if response.status_code == 200: data = response.json() print(f" ✅ Django服务器状态: {data['message']}") return True else: print(f" ❌ 服务器响应异常: {response.status_code}") return False except Exception as e: print(f" ❌ 无法连接服务器: {e}") return False def test_step_1_original_data(self): """步骤1: 查看原始数据""" self.print_header(1, "原始数据 (prewashed_table)", "查看文件上传后的原始ATC对话数据") try: response = requests.get(f"{API_BASE}/original-data/", timeout=10) if response.status_code == 200: data = response.json() count = data['count'] records = data['data'] print(f" 📊 数据统计: {count} 条原始记录") self.print_data_sample(records) self.test_results['data_samples']['original'] = records[:3] self.test_results['pipeline_steps'].append({ 'step': 1, 'name': '原始数据', 'table': 'prewashed_table', 'count': count, 'status': 'success' }) return True else: print(f" ❌ 获取原始数据失败: {response.status_code}") return False except Exception as e: print(f" ❌ 原始数据检查异常: {e}") return False def test_step_2_preprocess(self): """步骤2: 数据预处理""" self.print_header(2, "数据预处理 (preprocessed_table)", "执行数据清理和格式统一") try: # 执行预处理 print(" 🔧 执行预处理操作...") response = requests.post(f"{API_BASE}/preprocess/", json={}, timeout=30) if response.status_code == 200: process_result = response.json() print(f" ✅ 预处理完成: {process_result['message']}") print(f" 📊 处理统计: 原始{process_result['data']['raw_count']}条 → 清理后{process_result['data']['processed_count']}条") print(f" 📈 清理率: {process_result['data']['cleaning_rate']}%") # 获取预处理结果 print("\n 📋 查看预处理结果...") response = requests.get(f"{API_BASE}/preprocessed-data/", timeout=10) if response.status_code == 200: data = response.json() count = data['count'] records = data['data'] print(f" 📊 预处理数据: {count} 条记录") self.print_data_sample(records) # 显示清理操作统计 if records: cleaning_stats = {} for record in records: ops = record.get('cleaning_operations', '无需清理') cleaning_stats[ops] = cleaning_stats.get(ops, 0) + 1 print(f" 🔧 清理操作统计:") for op, count_op in cleaning_stats.items(): print(f" - {op}: {count_op} 条") self.test_results['data_samples']['preprocessed'] = records[:3] self.test_results['pipeline_steps'].append({ 'step': 2, 'name': '数据预处理', 'table': 'preprocessed_table', 'count': count, 'status': 'success' }) return True else: print(f" ❌ 获取预处理数据失败: {response.status_code}") return False else: print(f" ❌ 预处理执行失败: {response.status_code}") return False except Exception as e: print(f" ❌ 预处理过程异常: {e}") return False def test_step_3_merge(self): """步骤3: 格式合并""" self.print_header(3, "格式合并 (merged_table)", "统一航空术语和数据格式") try: # 执行格式合并 print(" 🔧 执行格式合并操作...") response = requests.post(f"{API_BASE}/merge/", json={}, timeout=30) if response.status_code == 200: process_result = response.json() print(f" ✅ 格式合并完成: {process_result['message']}") print(f" 📊 合并统计: 总记录{process_result['data']['total_records']}条 → 合并{process_result['data']['merged_records']}条") print(f" 📈 合并成功率: {process_result['data']['merge_success_rate']}%") # 获取格式合并结果 print("\n 📋 查看格式合并结果...") response = requests.get(f"{API_BASE}/merged-data/", timeout=10) if response.status_code == 200: data = response.json() count = data['count'] records = data['data'] print(f" 📊 格式合并数据: {count} 条记录") self.print_data_sample(records) # 显示格式化操作统计 if records: format_stats = {} for record in records: ops = record.get('format_operations', '无需格式化') format_stats[ops] = format_stats.get(ops, 0) + 1 print(f" 🔧 格式化操作统计:") for op, count_op in format_stats.items(): print(f" - {op}: {count_op} 条") self.test_results['data_samples']['merged'] = records[:3] self.test_results['pipeline_steps'].append({ 'step': 3, 'name': '格式合并', 'table': 'merged_table', 'count': count, 'status': 'success' }) return True else: print(f" ❌ 获取格式合并数据失败: {response.status_code}") return False else: print(f" ❌ 格式合并执行失败: {response.status_code}") return False except Exception as e: print(f" ❌ 格式合并过程异常: {e}") return False def test_step_4_correct(self): """步骤4: 单词纠错""" self.print_header(4, "单词纠错 (corrected_table)", "修正拼写错误和语音字母转换") try: # 执行单词纠错 print(" 🔧 执行单词纠错操作...") response = requests.post(f"{API_BASE}/correct/", json={}, timeout=30) if response.status_code == 200: process_result = response.json() print(f" ✅ 单词纠错完成: {process_result['message']}") print(f" 📊 纠错统计: 总记录{process_result['data']['total_records']}条，纠正{process_result['data']['corrected_words']}个单词") print(f" 🔧 纠错类型: {', '.join(process_result['data']['correction_types'])}") # 获取单词纠错结果 print("\n 📋 查看单词纠错结果...") response = requests.get(f"{API_BASE}/corrected-data/", timeout=10) if response.status_code == 200: data = response.json() count = data['count'] records = data['data'] print(f" 📊 单词纠错数据: {count} 条记录") self.print_data_sample(records) # 显示纠错操作统计 if records: correction_stats = {} total_corrections = 0 for record in records: ops = record.get('corrections_made', '无需纠错') count_corrections = record.get('correction_count', 0) total_corrections += count_corrections correction_stats[ops] = correction_stats.get(ops, 0) + 1 print(f" 🔧 纠错操作统计 (总计纠正{total_corrections}处):") for op, count_op in list(correction_stats.items())[:5]: # 只显示前5种 print(f" - {op}: {count_op} 条") self.test_results['data_samples']['corrected'] = records[:3] self.test_results['pipeline_steps'].append({ 'step': 4, 'name': '单词纠错', 'table': 'corrected_table', 'count': count, 'status': 'success' }) return True else: print(f" ❌ 获取单词纠错数据失败: {response.status_code}") return False else: print(f" ❌ 单词纠错执行失败: {response.status_code}") return False except Exception as e: print(f" ❌ 单词纠错过程异常: {e}") return False def test_step_5_analyze(self): """步骤5: AI分析""" self.print_header(5, "AI分析 (processed_table)", "使用AI提取结构化飞行信息") try: # 执行AI分析 print(" 🔧 执行AI分析操作...") response = requests.post(f"{API_BASE}/analyze/", json={}, timeout=60) if response.status_code == 200: process_result = response.json() print(f" ✅ AI分析完成: {process_result['message']}") # 检查返回数据格式 if 'data' in process_result: result_data = process_result['data'] # 处理不同的返回格式 if 'original_count' in result_data: # 实际AI处理结果 print(f" 📊 AI处理统计:") print(f" 原始记录: {result_data.get('original_count', 0)} 条") print(f" 提取结果: {result_data.get('extracted_count', 0)} 条") print(f" 有效数据: {result_data.get('valid_count', 0)} 条") print(f" 提取效率: {result_data.get('extraction_rate', 0)}%") print(f" 验证率: {result_data.get('validation_rate', 0)}%") elif 'analysis_summary' in result_data: # 模拟分析结果 summary = result_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)} 个") print(f" 位置识别: {summary.get('locations_identified', 0)} 个") # 获取AI处理结果 print("\n 📋 查看AI处理结果...") response = requests.get(f"{API_BASE}/processed-data/", timeout=10) if response.status_code == 200: data = response.json() count = data['count'] records = data['data'] print(f" 📊 AI处理数据: {count} 条记录") self.print_data_sample(records) # 显示提取字段统计 if records: field_stats = {} for record in records: for field in ['call_sign', 'behavior', 'flight_level', 'location', 'time']: value = record.get(field) if value and value not in ['N/A', 'null', None]: field_stats[field] = field_stats.get(field, 0) + 1 print(f" 📈 AI提取字段统计:") for field, count_field in field_stats.items(): percentage = round(count_field / count * 100, 1) if count > 0 else 0 print(f" - {field}: {count_field}/{count} ({percentage}%)") self.test_results['data_samples']['processed'] = records[:3] self.test_results['pipeline_steps'].append({ 'step': 5, 'name': 'AI分析', 'table': 'processed_table', 'count': count, 'status': 'success' }) return True else: print(f" ❌ 获取AI处理数据失败: {response.status_code}") return False else: print(f" ❌ AI分析执行失败: {response.status_code}") return False except Exception as e: print(f" ❌ AI分析过程异常: {e}") return False def test_step_6_final_validation(self): """步骤6: 最终验证""" self.print_header(6, "最终验证 (final_table + quarantine_table)", "验证数据有效性并分离异常数据") try: # 获取最终有效数据 print(" 📋 查看最终有效数据...") response = requests.get(f"{API_BASE}/final-data/", timeout=10) final_count = 0 if response.status_code == 200: data = response.json() final_count = data['count'] records = data['data'] print(f" 📊 最终有效数据: {final_count} 条记录") self.print_data_sample(records) self.test_results['data_samples']['final'] = records[:3] else: print(f" ❌ 获取最终数据失败: {response.status_code}") # 获取异常数据 print("\n 📋 查看异常隔离数据...") response = requests.get(f"{API_BASE}/quarantine-data/", timeout=10) quarantine_count = 0 if response.status_code == 200: data = response.json() quarantine_count = data['count'] records = data['data'] print(f" 📊 异常隔离数据: {quarantine_count} 条记录") if quarantine_count > 0: self.print_data_sample(records) else: print(" ✅ 无异常数据，所有记录均通过验证") self.test_results['data_samples']['quarantine'] = records[:3] else: print(f" ❌ 获取异常数据失败: {response.status_code}") # 获取统计信息 print("\n 📋 查看处理统计...") response = requests.get(f"{API_BASE}/statistics/", timeout=10) if response.status_code == 200: data = response.json() 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['processing_summary'] = stats self.test_results['pipeline_steps'].append({ 'step': 6, 'name': '最终验证', 'table': 'final_table + quarantine_table', 'count': f"{final_count} + {quarantine_count}", 'status': 'success' }) return True except Exception as e: print(f" ❌ 最终验证过程异常: {e}") return False def generate_summary_report(self): """生成总结报告""" print(f"\n{'='*80}") print("📋 中间处理过程测试 - 总结报告") print(f"{'='*80}") # 流水线步骤总结 print(f"\n🔄 数据处理流水线总结:") for step in self.test_results['pipeline_steps']: status_icon = "✅" if step['status'] == 'success' else "❌" print(f" {status_icon} 步骤{step['step']}: {step['name']} ({step['table']}) - {step['count']} 条") # 数据质量分析 if 'processing_summary' in self.test_results and self.test_results['processing_summary']: stats = self.test_results['processing_summary'] print(f"\n📊 数据质量分析:") print(f" 📈 数据流向: {stats.get('original_count', 0)} → {stats.get('extracted_count', 0)} → {stats.get('valid_count', 0)}") print(f" 🎯 提取效率: {stats.get('extraction_rate', 0)}%") print(f" ✅ 验证通过率: {stats.get('validation_rate', 0)}%") print(f" ❌ 错误率: {stats.get('error_rate', 0)}%") # 关键成果 print(f"\n🎉 关键成果:") print(f" ✅ 完整的6步数据处理流水线已验证") print(f" ✅ 每个中间步骤都有对应的数据表存储") print(f" ✅ 所有API接口均正常工作") print(f" ✅ 数据在各步骤间正确传递和转换") # 前端集成建议 print(f"\n💡 前端集成建议:") print(f" 1. 使用分步处理模式展示中间过程") print(f" 2. 每个步骤完成后调用对应的GET API获取结果") print(f" 3. 展示数据转换过程和处理统计") print(f" 4. 支持查看各步骤的详细数据样例") # 保存详细报告 report_file = "/home/hzk/项目/moxun-1/test/intermediate_process_report.json" with open(report_file, 'w', encoding='utf-8') as f: json.dump({ 'test_time': datetime.now().isoformat(), 'results': self.test_results }, f, indent=2, ensure_ascii=False) print(f"\n📄 详细报告已保存: {report_file}") return len([s for s in self.test_results['pipeline_steps'] if s['status'] == 'success']) >= 5 def run_complete_test(self): """运行完整的中间处理过程测试""" print("🎯 开始中间处理过程测试") print("📝 目标: 测试数据处理流水线的每个中间步骤") print(f"🕒 测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # 执行测试步骤 test_steps = [ self.test_system_health, self.test_step_1_original_data, self.test_step_2_preprocess, self.test_step_3_merge, self.test_step_4_correct, self.test_step_5_analyze, self.test_step_6_final_validation ] success_count = 0 for step in test_steps: try: if step(): success_count += 1 time.sleep(1) # 步骤间暂停 except Exception as e: print(f" ❌ 测试步骤异常: {e}") # 生成总结报告 overall_success = self.generate_summary_report() print(f"\n{'='*80}") if overall_success: print("🎉 中间处理过程测试成功完成！") print("✅ 所有数据处理步骤均正常工作，可以向前端提供完整的中间过程展示") else: print("⚠️ 中间处理过程测试完成，但存在需要关注的问题") return overall_success def main(): """主函数""" tester = IntermediateProcessTest() success = tester.run_complete_test() return 0 if success else 1 if __name__ == "__main__": exit(main())