98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
|
import requests
|
|||
|
|
import json
|
|||
|
|
|
|||
|
|
# 配置
|
|||
|
|
BASE_URL = "http://localhost:8080"
|
|||
|
|
LOGIN_URL = f"{BASE_URL}/login/test"
|
|||
|
|
IMPORT_URL = f"{BASE_URL}/dpc/intermediary/importPersonData"
|
|||
|
|
|
|||
|
|
# 登录获取token
|
|||
|
|
print("=" * 60)
|
|||
|
|
print("测试个人中介Excel导入功能")
|
|||
|
|
print("=" * 60)
|
|||
|
|
|
|||
|
|
login_data = {
|
|||
|
|
"username": "admin",
|
|||
|
|
"password": "admin123"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
print("\n1. 登录系统...")
|
|||
|
|
try:
|
|||
|
|
response = requests.post(LOGIN_URL, json=login_data)
|
|||
|
|
print(f" 登录响应状态码: {response.status_code}")
|
|||
|
|
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
result = response.json()
|
|||
|
|
print(f" 登录响应: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
|||
|
|
|
|||
|
|
token = result.get("token")
|
|||
|
|
if token:
|
|||
|
|
print(f" ✓ 成功获取token: {token[:50]}...")
|
|||
|
|
else:
|
|||
|
|
print(" ✗ 登录失败：未获取到token")
|
|||
|
|
exit(1)
|
|||
|
|
else:
|
|||
|
|
print(f" ✗ 登录失败：{response.text}")
|
|||
|
|
exit(1)
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" ✗ 登录异常: {e}")
|
|||
|
|
exit(1)
|
|||
|
|
|
|||
|
|
# 测试导入
|
|||
|
|
print("\n2. 准备导入Excel文件...")
|
|||
|
|
files = {
|
|||
|
|
'file': ('个人中介黑名单测试数据_1000条.xlsx',
|
|||
|
|
open('doc/个人中介黑名单测试数据_1000条.xlsx', 'rb'),
|
|||
|
|
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
headers = {
|
|||
|
|
'Authorization': f'Bearer {token}'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
params = {
|
|||
|
|
'updateSupport': 'false'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
print(f" 导入URL: {IMPORT_URL}")
|
|||
|
|
print(f" 文件名: 个人中介黑名单测试数据_1000条.xlsx")
|
|||
|
|
print(f" 更新支持: false")
|
|||
|
|
|
|||
|
|
print("\n3. 发送导入请求...")
|
|||
|
|
try:
|
|||
|
|
response = requests.post(IMPORT_URL, files=files, headers=headers, params=params)
|
|||
|
|
print(f" 导入响应状态码: {response.status_code}")
|
|||
|
|
print(f" 响应头: {dict(response.headers)}")
|
|||
|
|
|
|||
|
|
if response.status_code == 200:
|
|||
|
|
result = response.json()
|
|||
|
|
print(f"\n 导入结果:")
|
|||
|
|
print(f" {json.dumps(result, ensure_ascii=False, indent=2)}")
|
|||
|
|
|
|||
|
|
if result.get("code") == 200:
|
|||
|
|
print(f"\n ✓ 导入成功!")
|
|||
|
|
print(f" 消息: {result.get('msg')}")
|
|||
|
|
else:
|
|||
|
|
print(f"\n ✗ 导入失败!")
|
|||
|
|
print(f" 错误代码: {result.get('code')}")
|
|||
|
|
print(f" 错误消息: {result.get('msg')}")
|
|||
|
|
else:
|
|||
|
|
print(f"\n ✗ HTTP错误: {response.status_code}")
|
|||
|
|
print(f" 响应内容: {response.text}")
|
|||
|
|
|
|||
|
|
except requests.exceptions.ConnectionError:
|
|||
|
|
print(f"\n ✗ 连接失败：无法连接到后端服务器")
|
|||
|
|
print(f" 请确认后端服务已启动在 {BASE_URL}")
|
|||
|
|
except requests.exceptions.Timeout:
|
|||
|
|
print(f"\n ✗ 请求超时")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"\n ✗ 导入异常: {e}")
|
|||
|
|
import traceback
|
|||
|
|
traceback.print_exc()
|
|||
|
|
finally:
|
|||
|
|
files['file'][1].close()
|
|||
|
|
|
|||
|
|
print("\n" + "=" * 60)
|
|||
|
|
print("测试完成")
|
|||
|
|
print("=" * 60)
|