Ubuntu
102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
"""
|
|
测试深度分析功能
|
|
"""
|
|
import requests
|
|
import time
|
|
|
|
API_BASE = "http://129.211.173.87:10001"
|
|
|
|
# 获取token (使用默认admin/admin)
|
|
def login():
|
|
resp = requests.post(f"{API_BASE}/auth/login", json={
|
|
"username": "admin",
|
|
"password": "abcd1234"
|
|
})
|
|
data = resp.json()
|
|
print(f"Login response: {data}")
|
|
return data.get("access_token") or data.get("token")
|
|
|
|
def test_deep_analysis():
|
|
token = login()
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
print("=" * 60)
|
|
print("1. 测试获取模型配置")
|
|
print("=" * 60)
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/config/model", headers=headers)
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.json() if resp.status_code == 200 else resp.text}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("2. 设置测试模型配置")
|
|
print("=" * 60)
|
|
# 设置一个假的API key进行测试（会失败但能看到流程）
|
|
resp = requests.post(f"{API_BASE}/deep-analysis/config/model",
|
|
headers=headers,
|
|
json={"model": "deepseek-chat", "api_key": "sk-test123456789"}
|
|
)
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.json() if resp.status_code == 200 else resp.text}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("3. 测试获取客户分析状态")
|
|
print("=" * 60)
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/status/1", headers=headers)
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.json() if resp.status_code == 200 else resp.text}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("4. 测试创建分析任务")
|
|
print("=" * 60)
|
|
resp = requests.post(f"{API_BASE}/deep-analysis/create",
|
|
headers=headers,
|
|
json={"customer_ids": [1, 2, 3]}
|
|
)
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.json() if resp.status_code == 200 else resp.text}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("5. 测试获取分析列表")
|
|
print("=" * 60)
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/list?page=1&page_size=10", headers=headers)
|
|
print(f"Status: {resp.status_code}")
|
|
data = resp.json() if resp.status_code == 200 else {}
|
|
print(f"Total: {data.get('total', 0)}")
|
|
print(f"Items: {len(data.get('items', []))}")
|
|
if data.get('items'):
|
|
for item in data['items'][:3]:
|
|
print(f" - {item['company_name']}: {item['status']}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("6. 测试获取队列状态")
|
|
print("=" * 60)
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/queue/status", headers=headers)
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.json() if resp.status_code == 200 else resp.text}")
|
|
|
|
# 等待几秒看队列状态
|
|
print("\n等待3秒查看队列处理...")
|
|
time.sleep(3)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("7. 再次获取队列状态（应该显示失败，因为API key是测试值）")
|
|
print("=" * 60)
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/queue/status", headers=headers)
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.json() if resp.status_code == 200 else resp.text}")
|
|
|
|
# 获取分析列表看状态变化
|
|
print("\n" + "=" * 60)
|
|
print("8. 再次获取分析列表看状态")
|
|
print("=" * 60)
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/list?page=1&page_size=10", headers=headers)
|
|
data = resp.json() if resp.status_code == 200 else {}
|
|
if data.get('items'):
|
|
for item in data['items'][:3]:
|
|
print(f" - {item['company_name']}: {item['status']}")
|
|
if item.get('error_message'):
|
|
print(f" Error: {item['error_message'][:100]}...")
|
|
|
|
if __name__ == "__main__":
|
|
test_deep_analysis()
|