""" 完整测试脚本 - 测试所有功能 """ import requests import json import sys import time BASE_URL = "http://localhost:7500" def print_section(title): """打印章节标题""" print("\n" + "="*60) print(f" {title}") print("="*60) def print_result(success, message): """打印测试结果""" status = "✓" if success else "✗" print(f"{status} {message}") def test_server_connection(): """测试服务连接""" print_section("1. 测试服务连接") try: response = requests.get(f"{BASE_URL}/", timeout=5) if response.status_code == 200: print_result(True, f"服务连接成功 (状态码: {response.status_code})") return True else: print_result(False, f"服务连接异常 (状态码: {response.status_code})") return False except requests.exceptions.ConnectionError: print_result(False, "无法连接到服务，请确保服务已启动 (运行 python app.py)") return False except Exception as e: print_result(False, f"连接失败: {str(e)}") return False def test_extract_api(): """测试解析接口""" print_section("2. 测试解析接口 (/ai/extract)") url = f"{BASE_URL}/ai/extract" # 测试数据 test_data = { "inputData": [ { "fieldCode": "clue_info", "fieldValue": "被举报用户名称是张三，年龄30岁，某公司总经理，男性，1980年5月出生，中共党员，正处级" } ], "outputData": [ {"fieldCode": "target_name"}, {"fieldCode": "target_gender"}, {"fieldCode": "target_organization_and_position"} ] } print(f"\n请求URL: {url}") print(f"请求数据:") print(json.dumps(test_data, ensure_ascii=False, indent=2)) try: print("\n正在发送请求...") response = requests.post(url, json=test_data, timeout=30) print(f"响应状态码: {response.status_code}") result = response.json() print(f"\n响应数据:") print(json.dumps(result, ensure_ascii=False, indent=2)) if result.get('isSuccess'): print_result(True, "解析接口调用成功") out_data = result.get('data', {}).get('outData', []) if out_data: print("\n提取的字段：") for item in out_data: field_code = item.get('fieldCode', '') field_value = item.get('fieldValue', '') if field_value: print(f" - {field_code}: {field_value}") else: print(f" - {field_code}: (空)") return True else: error_msg = result.get('errorMsg', '未知错误') error_code = result.get('code', 'N/A') print_result(False, f"解析失败 - 错误码: {error_code}, 错误信息: {error_msg}") return False except requests.exceptions.Timeout: print_result(False, "请求超时（超过30秒）") return False except requests.exceptions.ConnectionError: print_result(False, "无法连接到服务") return False except json.JSONDecodeError as e: print_result(False, f"响应JSON解析失败: {str(e)}") return False except Exception as e: print_result(False, f"测试失败: {str(e)}") import traceback traceback.print_exc() return False def test_generate_document_api(): """测试文档生成接口（需要MinIO配置）""" print_section("3. 测试文档生成接口 (/ai/generate-document)") url = f"{BASE_URL}/ai/generate-document" # 测试数据 test_data = { "inputData": [ { "fieldCode": "target_name", "fieldValue": "张三" }, { "fieldCode": "target_gender", "fieldValue": "男" }, { "fieldCode": "target_organization_and_position", "fieldValue": "某公司总经理" } ], "fpolicFieldParamFileList": [ { "fileId": 1, "fileName": "测试文档.doc", "templateCode": "PRELIMINARY_VERIFICATION_APPROVAL" } ] } print(f"\n请求URL: {url}") print(f"请求数据:") print(json.dumps(test_data, ensure_ascii=False, indent=2)) print("\n⚠ 注意：此接口需要MinIO配置和模板文件，可能会失败") print("按 Enter 继续，或 Ctrl+C 跳过此测试...") try: input() except KeyboardInterrupt: print("\n已跳过文档生成接口测试") return None try: print("\n正在发送请求...") response = requests.post(url, json=test_data, timeout=60) print(f"响应状态码: {response.status_code}") result = response.json() print(f"\n响应数据:") print(json.dumps(result, ensure_ascii=False, indent=2)) if result.get('isSuccess'): print_result(True, "文档生成接口调用成功") file_list = result.get('data', {}).get('fpolicFieldParamFileList', []) if file_list: print("\n生成的文件：") for file_info in file_list: print(f" - {file_info.get('fileName')}: {file_info.get('filePath', '(无路径)')}") return True else: error_msg = result.get('errorMsg', '未知错误') error_code = result.get('code', 'N/A') print_result(False, f"文档生成失败 - 错误码: {error_code}, 错误信息: {error_msg}") # 提供错误诊断 if error_code == 1001: print("\n 诊断：模板不存在，请检查：") print(" 1. templateCode 是否正确") print(" 2. 数据库中是否有对应的文件配置") print(" 3. 运行 python update_template_code_field.py 更新配置") elif error_code == 3001: print("\n 诊断：文件生成失败，请检查：") print(" 1. MinIO配置是否正确") print(" 2. 模板文件是否存在于MinIO") elif error_code == 3002: print("\n 诊断：文件保存失败，请检查：") print(" 1. MinIO服务是否可访问") print(" 2. 存储桶是否存在") return False except requests.exceptions.Timeout: print_result(False, "请求超时（超过60秒）") return False except requests.exceptions.ConnectionError: print_result(False, "无法连接到服务") return False except Exception as e: print_result(False, f"测试失败: {str(e)}") import traceback traceback.print_exc() return False def test_api_docs(): """测试API文档页面""" print_section("4. 测试API文档页面") try: response = requests.get(f"{BASE_URL}/api-docs", timeout=5) if response.status_code == 200: print_result(True, f"API文档页面可访问 (状态码: {response.status_code})") print(f" URL: {BASE_URL}/api-docs") return True else: print_result(False, f"API文档页面访问异常 (状态码: {response.status_code})") return False except Exception as e: print_result(False, f"访问失败: {str(e)}") return False def main(): """主函数""" print("\n" + "="*60) print(" 智慧监督AI文书写作服务 - 完整功能测试") print("="*60) print(f"\n测试目标: {BASE_URL}") print("请确保服务已启动（运行 python app.py）\n") results = { 'server': False, 'extract': False, 'generate_document': None, 'api_docs': False } # 1. 测试服务连接 results['server'] = test_server_connection() if not results['server']: print("\n" + "="*60) print(" 服务未启动，无法继续测试") print("="*60) print("\n请先启动服务：") print(" python app.py") return # 2. 测试解析接口 results['extract'] = test_extract_api() # 3. 测试文档生成接口（可选） results['generate_document'] = test_generate_document_api() # 4. 测试API文档 results['api_docs'] = test_api_docs() # 总结 print_section("测试总结") print("\n测试结果：") print(f" 服务连接: {'✓ 通过' if results['server'] else '✗ 失败'}") print(f" 解析接口: {'✓ 通过' if results['extract'] else '✗ 失败'}") if results['generate_document'] is not None: print(f" 文档生成: {'✓ 通过' if results['generate_document'] else '✗ 失败'}") else: print(f" 文档生成: ⊘ 跳过") print(f" API文档: {'✓ 通过' if results['api_docs'] else '✗ 失败'}") # 计算成功率 test_count = sum(1 for v in results.values() if v is not None) pass_count = sum(1 for v in results.values() if v is True) print(f"\n通过率: {pass_count}/{test_count} ({pass_count*100//test_count if test_count > 0 else 0}%)") print("\n" + "="*60) print(" 测试完成") print("="*60) print("\n提示：") if not results['extract']: print(" - 如果解析接口失败，请检查：") print(" 1. 数据库连接是否正常") print(" 2. 字段是否已初始化（运行 python init_all_fields_from_excel.py）") print(" 3. AI服务配置是否正确（检查 .env 文件中的 SILICONFLOW_API_KEY）") if results['generate_document'] is False: print(" - 文档生成接口需要：") print(" 1. MinIO服务配置") print(" 2. 模板文件存在于MinIO") print(" 3. 数据库中有正确的文件配置") if __name__ == '__main__': try: main() except KeyboardInterrupt: print("\n\n测试已中断") sys.exit(1)