429 lines
15 KiB
Python
429 lines
15 KiB
Python
"""
|
||
测试文档生成接口
|
||
1. 测试模板读取功能
|
||
2. 测试占位符识别和替换
|
||
3. 验证生成的文档
|
||
"""
|
||
import os
|
||
import pymysql
|
||
from pathlib import Path
|
||
from typing import Dict, List, Set
|
||
from dotenv import load_dotenv
|
||
import re
|
||
from docx import Document
|
||
import requests
|
||
import json
|
||
import tempfile
|
||
import shutil
|
||
|
||
# 加载环境变量
|
||
load_dotenv()
|
||
|
||
# 数据库配置
|
||
DB_CONFIG = {
|
||
'host': os.getenv('DB_HOST', '152.136.177.240'),
|
||
'port': int(os.getenv('DB_PORT', 5012)),
|
||
'user': os.getenv('DB_USER', 'finyx'),
|
||
'password': os.getenv('DB_PASSWORD', '6QsGK6MpePZDE57Z'),
|
||
'database': os.getenv('DB_NAME', 'finyx'),
|
||
'charset': 'utf8mb4'
|
||
}
|
||
|
||
# API配置
|
||
API_BASE_URL = os.getenv('API_BASE_URL', 'http://localhost:5000')
|
||
|
||
# 项目根目录
|
||
PROJECT_ROOT = Path(__file__).parent
|
||
TEMPLATES_DIR = PROJECT_ROOT / "template_finish"
|
||
|
||
|
||
def print_section(title):
|
||
"""打印章节标题"""
|
||
print("\n" + "="*70)
|
||
print(f" {title}")
|
||
print("="*70)
|
||
|
||
|
||
def print_result(success, message):
|
||
"""打印结果"""
|
||
status = "[OK]" if success else "[FAIL]"
|
||
print(f"{status} {message}")
|
||
|
||
|
||
def get_actual_tenant_id(conn) -> int:
|
||
"""获取数据库中的实际tenant_id"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
try:
|
||
cursor.execute("SELECT DISTINCT tenant_id FROM f_polic_file_config LIMIT 1")
|
||
result = cursor.fetchone()
|
||
if result:
|
||
return result['tenant_id']
|
||
return 1
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def get_test_templates(conn, tenant_id: int, limit: int = 5) -> List[Dict]:
|
||
"""获取测试用的模板列表"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
try:
|
||
sql = """
|
||
SELECT id, name, file_path
|
||
FROM f_polic_file_config
|
||
WHERE tenant_id = %s
|
||
AND file_path IS NOT NULL
|
||
AND file_path != ''
|
||
AND file_path LIKE 'template_finish/%%'
|
||
AND state = 1
|
||
ORDER BY id
|
||
LIMIT %s
|
||
"""
|
||
cursor.execute(sql, (tenant_id, limit))
|
||
templates = cursor.fetchall()
|
||
return templates
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def extract_placeholders_from_docx(file_path: Path) -> Set[str]:
|
||
"""从docx文件中提取所有占位符"""
|
||
placeholders = set()
|
||
placeholder_pattern = re.compile(r'\{\{([^}]+)\}\}')
|
||
|
||
try:
|
||
doc = Document(file_path)
|
||
|
||
# 从段落中提取
|
||
for paragraph in doc.paragraphs:
|
||
text = paragraph.text
|
||
matches = placeholder_pattern.findall(text)
|
||
for match in matches:
|
||
field_code = match.strip()
|
||
if field_code:
|
||
placeholders.add(field_code)
|
||
|
||
# 从表格中提取
|
||
for table in doc.tables:
|
||
try:
|
||
for row in table.rows:
|
||
for cell in row.cells:
|
||
for paragraph in cell.paragraphs:
|
||
text = paragraph.text
|
||
matches = placeholder_pattern.findall(text)
|
||
for match in matches:
|
||
field_code = match.strip()
|
||
if field_code:
|
||
placeholders.add(field_code)
|
||
except:
|
||
continue
|
||
except Exception as e:
|
||
print(f" [错误] 读取文件失败: {str(e)}")
|
||
|
||
return placeholders
|
||
|
||
|
||
def check_template_file_exists(file_path: str) -> bool:
|
||
"""检查模板文件是否存在"""
|
||
local_file = PROJECT_ROOT / file_path
|
||
return local_file.exists() and local_file.is_file()
|
||
|
||
|
||
def test_template_reading(file_path: str) -> tuple[bool, str, Set[str]]:
|
||
"""测试模板读取功能"""
|
||
local_file = PROJECT_ROOT / file_path
|
||
|
||
if not local_file.exists():
|
||
return False, f"文件不存在: {local_file}", set()
|
||
|
||
try:
|
||
# 尝试读取文档
|
||
doc = Document(local_file)
|
||
|
||
# 提取占位符
|
||
placeholders = extract_placeholders_from_docx(local_file)
|
||
|
||
return True, "读取成功", placeholders
|
||
except Exception as e:
|
||
return False, f"读取失败: {str(e)}", set()
|
||
|
||
|
||
def test_document_generation_api(template_id: int, template_name: str, file_path: str, placeholders: Set[str], api_url: str) -> Dict:
|
||
"""测试文档生成API"""
|
||
# 构建测试数据
|
||
input_data = []
|
||
test_values = {
|
||
'target_name': '张三',
|
||
'target_organization': '测试单位',
|
||
'target_position': '测试职务',
|
||
'target_organization_and_position': '测试单位-测试职务',
|
||
'target_age': '35',
|
||
'target_gender': '男',
|
||
'target_contact': '13800138000',
|
||
'target_address': '测试地址',
|
||
'target_political_status': '中共党员',
|
||
'investigation_team_code': 'DC2025001',
|
||
'investigation_team_leader_name': '李四',
|
||
'investigation_team_member_names': '王五、赵六',
|
||
'investigation_unit_name': '调查单位',
|
||
'appointment_time': '2025-12-16 14:00',
|
||
'appointment_location': '会议室A',
|
||
'approval_time': '2025-12-16',
|
||
'clue_source': '群众举报',
|
||
'handler_name': '处理人',
|
||
'handling_department': '处理部门',
|
||
'department_opinion': '同意',
|
||
'investigation_location': '调查地点',
|
||
'target_basic_info': '基本信息',
|
||
'target_contact': '联系方式',
|
||
'target_date_of_birth': '1990-01-01',
|
||
'target_date_of_birth_full': '1990年1月1日',
|
||
'target_education': '本科',
|
||
'target_education_level': '大学',
|
||
'target_ethnicity': '汉族',
|
||
'target_family_situation': '家庭情况',
|
||
'target_id_number': '110101199001011234',
|
||
'target_issue_description': '问题描述',
|
||
'target_place_of_origin': '北京',
|
||
'target_professional_rank': '正科级',
|
||
'target_registered_address': '户籍地址',
|
||
'target_social_relations': '社会关系',
|
||
'target_work_basic_info': '工作基本信息'
|
||
}
|
||
|
||
for placeholder in sorted(placeholders):
|
||
value = test_values.get(placeholder, f'测试值_{placeholder}')
|
||
input_data.append({
|
||
"fieldCode": placeholder,
|
||
"fieldValue": value
|
||
})
|
||
|
||
payload = {
|
||
"fileId": template_id,
|
||
"inputData": input_data
|
||
}
|
||
|
||
try:
|
||
print(f" 请求URL: {api_url}/api/document/generate")
|
||
print(f" 请求体: {json.dumps(payload, ensure_ascii=False, indent=2)}")
|
||
|
||
response = requests.post(
|
||
f"{api_url}/api/document/generate",
|
||
json=payload,
|
||
timeout=60
|
||
)
|
||
|
||
result = {
|
||
'success': response.status_code == 200,
|
||
'status_code': response.status_code,
|
||
'response': None,
|
||
'error': None
|
||
}
|
||
|
||
if response.status_code == 200:
|
||
result['response'] = response.json()
|
||
else:
|
||
result['error'] = response.text
|
||
|
||
return result
|
||
except requests.exceptions.ConnectionError:
|
||
return {
|
||
'success': False,
|
||
'status_code': None,
|
||
'response': None,
|
||
'error': '无法连接到API服务器，请确保服务已启动'
|
||
}
|
||
except requests.exceptions.Timeout:
|
||
return {
|
||
'success': False,
|
||
'status_code': None,
|
||
'response': None,
|
||
'error': '请求超时'
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
'success': False,
|
||
'status_code': None,
|
||
'response': None,
|
||
'error': f'请求异常: {str(e)}'
|
||
}
|
||
|
||
|
||
def verify_generated_document(file_path: str, placeholders: Set[str]) -> Dict:
|
||
"""验证生成的文档（如果能够下载）"""
|
||
# 这里可以添加下载和验证逻辑
|
||
# 目前只返回占位符验证信息
|
||
return {
|
||
'placeholders_found': len(placeholders),
|
||
'placeholders_list': sorted(placeholders)
|
||
}
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print_section("文档生成接口测试")
|
||
|
||
# 1. 连接数据库
|
||
print_section("1. 连接数据库")
|
||
try:
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
print_result(True, "数据库连接成功")
|
||
except Exception as e:
|
||
print_result(False, f"数据库连接失败: {str(e)}")
|
||
return
|
||
|
||
try:
|
||
# 2. 获取实际的tenant_id
|
||
tenant_id = get_actual_tenant_id(conn)
|
||
print(f" 实际tenant_id: {tenant_id}")
|
||
|
||
# 3. 获取测试模板
|
||
print_section("2. 获取测试模板")
|
||
test_templates = get_test_templates(conn, tenant_id, limit=5)
|
||
print_result(True, f"找到 {len(test_templates)} 个测试模板")
|
||
|
||
if not test_templates:
|
||
print_result(False, "没有找到可测试的模板")
|
||
return
|
||
|
||
# 4. 测试每个模板
|
||
print_section("3. 测试模板读取和占位符识别")
|
||
|
||
test_results = []
|
||
|
||
for i, template in enumerate(test_templates, 1):
|
||
template_id = template['id']
|
||
template_name = template['name']
|
||
file_path = template['file_path']
|
||
|
||
print(f"\n 模板 {i}: {template_name}")
|
||
print(f" ID: {template_id}")
|
||
print(f" 路径: {file_path}")
|
||
|
||
# 检查文件是否存在
|
||
if not check_template_file_exists(file_path):
|
||
print_result(False, f"文件不存在: {file_path}")
|
||
test_results.append({
|
||
'template_id': template_id,
|
||
'template_name': template_name,
|
||
'file_path': file_path,
|
||
'file_exists': False,
|
||
'read_success': False,
|
||
'placeholders': set(),
|
||
'api_test': None
|
||
})
|
||
continue
|
||
|
||
print_result(True, "文件存在")
|
||
|
||
# 测试模板读取
|
||
read_success, read_message, placeholders = test_template_reading(file_path)
|
||
print_result(read_success, read_message)
|
||
|
||
if read_success:
|
||
print(f" 占位符数量: {len(placeholders)}")
|
||
if placeholders:
|
||
print(f" 占位符: {sorted(placeholders)}")
|
||
else:
|
||
print(f" [警告] 未找到占位符")
|
||
|
||
test_results.append({
|
||
'template_id': template_id,
|
||
'template_name': template_name,
|
||
'file_path': file_path,
|
||
'file_exists': True,
|
||
'read_success': read_success,
|
||
'placeholders': placeholders,
|
||
'api_test': None
|
||
})
|
||
|
||
# 5. 测试API接口
|
||
print_section("4. 测试文档生成API接口")
|
||
|
||
# 检查API是否可用
|
||
try:
|
||
response = requests.get(f"{API_BASE_URL}/api/file-configs", timeout=5)
|
||
api_available = response.status_code == 200
|
||
except:
|
||
api_available = False
|
||
|
||
if not api_available:
|
||
print_result(False, f"API服务不可用: {API_BASE_URL}")
|
||
print(" [提示] 请确保Flask服务已启动")
|
||
print(" [提示] 可以手动测试API，使用以下命令:")
|
||
print(f" python app.py")
|
||
else:
|
||
print_result(True, f"API服务可用: {API_BASE_URL}")
|
||
|
||
# 选择有占位符的模板进行API测试
|
||
templates_with_placeholders = [r for r in test_results if r['read_success'] and r['placeholders']]
|
||
|
||
if templates_with_placeholders:
|
||
test_template = templates_with_placeholders[0]
|
||
print(f"\n 测试模板: {test_template['template_name']}")
|
||
print(f" 模板ID: {test_template['template_id']}")
|
||
print(f" 占位符: {sorted(test_template['placeholders'])}")
|
||
|
||
api_result = test_document_generation_api(
|
||
test_template['template_id'],
|
||
test_template['template_name'],
|
||
test_template['file_path'],
|
||
test_template['placeholders'],
|
||
API_BASE_URL
|
||
)
|
||
|
||
test_template['api_test'] = api_result
|
||
|
||
if api_result['success']:
|
||
print_result(True, "API调用成功")
|
||
print(f" 响应: {json.dumps(api_result['response'], ensure_ascii=False, indent=2)}")
|
||
else:
|
||
print_result(False, f"API调用失败: {api_result.get('error', '未知错误')}")
|
||
if api_result.get('status_code'):
|
||
print(f" 状态码: {api_result['status_code']}")
|
||
else:
|
||
print_result(False, "没有找到有占位符的模板进行API测试")
|
||
|
||
# 6. 生成测试报告
|
||
print_section("5. 测试结果汇总")
|
||
|
||
total_templates = len(test_results)
|
||
file_exists_count = sum(1 for r in test_results if r['file_exists'])
|
||
read_success_count = sum(1 for r in test_results if r['read_success'])
|
||
with_placeholders_count = sum(1 for r in test_results if r['placeholders'])
|
||
api_tested_count = sum(1 for r in test_results if r.get('api_test') is not None)
|
||
api_success_count = sum(1 for r in test_results if r.get('api_test') and r.get('api_test', {}).get('success', False))
|
||
|
||
print(f" 测试模板总数: {total_templates}")
|
||
print(f" 文件存在: {file_exists_count}/{total_templates}")
|
||
print(f" 读取成功: {read_success_count}/{total_templates}")
|
||
print(f" 有占位符: {with_placeholders_count}/{total_templates}")
|
||
print(f" API测试: {api_tested_count}/{total_templates}")
|
||
if api_tested_count > 0:
|
||
print(f" API成功: {api_success_count}/{api_tested_count}")
|
||
|
||
# 7. 详细结果
|
||
print_section("6. 详细测试结果")
|
||
|
||
for i, result in enumerate(test_results, 1):
|
||
print(f"\n 模板 {i}: {result['template_name']}")
|
||
print(f" ID: {result['template_id']}")
|
||
print(f" 文件存在: {'是' if result['file_exists'] else '否'}")
|
||
print(f" 读取成功: {'是' if result['read_success'] else '否'}")
|
||
print(f" 占位符数量: {len(result['placeholders'])}")
|
||
|
||
if result['api_test']:
|
||
api_test = result['api_test']
|
||
print(f" API测试: {'成功' if api_test['success'] else '失败'}")
|
||
if not api_test['success']:
|
||
print(f" 错误: {api_test.get('error', '未知错误')}")
|
||
|
||
finally:
|
||
conn.close()
|
||
print_result(True, "数据库连接已关闭")
|
||
|
||
print_section("测试完成")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|