Ubuntu
206 lines
7.1 KiB
Python
206 lines
7.1 KiB
Python
"""
|
|
深度分析功能完整测试 - 测试 Ht Active Software (ID=1)
|
|
"""
|
|
import requests
|
|
import time
|
|
import sys
|
|
|
|
API_BASE = "http://129.211.173.87:10001"
|
|
|
|
def login():
|
|
"""登录获取token"""
|
|
resp = requests.post(f"{API_BASE}/auth/login", json={
|
|
"username": "admin",
|
|
"password": "abcd1234"
|
|
})
|
|
data = resp.json()
|
|
if "access_token" not in data:
|
|
print(f"❌ 登录失败: {data}")
|
|
sys.exit(1)
|
|
print(f"✅ 登录成功")
|
|
return data["access_token"]
|
|
|
|
def test_config(token):
|
|
"""测试配置接口"""
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# 获取当前配置
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/config/model", headers=headers)
|
|
print(f"\n📊 当前模型配置:")
|
|
print(f" Status: {resp.status_code}")
|
|
if resp.status_code == 200:
|
|
print(f" Config: {resp.json()}")
|
|
|
|
# 设置测试配置（使用测试API key）
|
|
resp = requests.post(f"{API_BASE}/deep-analysis/config/model",
|
|
headers=headers,
|
|
json={"model": "deepseek-chat", "api_key": "sk-test12345678"}
|
|
)
|
|
print(f"\n📝 设置测试配置:")
|
|
print(f" Status: {resp.status_code}")
|
|
if resp.status_code == 200:
|
|
print(f" Response: {resp.json()}")
|
|
|
|
def test_customer_status(token, customer_id):
|
|
"""测试获取客户分析状态"""
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/status/{customer_id}", headers=headers)
|
|
print(f"\n🔍 客户 {customer_id} 分析状态:")
|
|
print(f" Status: {resp.status_code}")
|
|
if resp.status_code == 200:
|
|
print(f" Response: {resp.json()}")
|
|
|
|
def test_create_analysis(token, customer_id):
|
|
"""测试创建分析任务"""
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resp = requests.post(f"{API_BASE}/deep-analysis/create",
|
|
headers=headers,
|
|
json={"customer_ids": [customer_id]}
|
|
)
|
|
print(f"\n✨ 创建分析任务 (客户ID: {customer_id}):")
|
|
print(f" Status: {resp.status_code}")
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
print(f" Created: {data.get('created_count', 0)} 个任务")
|
|
print(f" Skipped: {data.get('skipped_count', 0)} 个已存在")
|
|
print(f" IDs: {data.get('created_ids', [])}")
|
|
return data.get('created_ids', [])
|
|
else:
|
|
print(f" Error: {resp.text}")
|
|
return []
|
|
|
|
def test_list_analyses(token):
|
|
"""测试获取分析列表"""
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/list?page=1&page_size=10", headers=headers)
|
|
print(f"\n📋 分析列表:")
|
|
print(f" Status: {resp.status_code}")
|
|
data = resp.json()
|
|
print(f" Total: {data.get('total', 0)}")
|
|
print(f" Items: {len(data.get('items', []))}")
|
|
if data.get('items'):
|
|
print(f" 最新分析:")
|
|
for item in data['items'][:3]:
|
|
status_icon = {
|
|
'pending': '⏳',
|
|
'processing': '🔄',
|
|
'completed': '✅',
|
|
'failed': '❌'
|
|
}.get(item['status'], '❓')
|
|
print(f" {status_icon} [{item['id']}] {item['company_name']}: {item['status']}")
|
|
return data.get('items', [])
|
|
|
|
def test_queue_status(token):
|
|
"""测试队列状态"""
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/queue/status", headers=headers)
|
|
print(f"\n🎛️ 队列状态:")
|
|
print(f" Status: {resp.status_code}")
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
print(f" 队列中: {data.get('queued_count', 0)} 个")
|
|
print(f" 处理中: {'是' if data.get('is_processing') else '否'}")
|
|
if data.get('current_task'):
|
|
print(f" 当前任务: {data.get('current_task')}")
|
|
return resp.json() if resp.status_code == 200 else {}
|
|
|
|
def test_get_detail(token, analysis_id):
|
|
"""测试获取分析详情"""
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/{analysis_id}", headers=headers)
|
|
print(f"\n📄 分析详情 (ID: {analysis_id}):")
|
|
print(f" Status: {resp.status_code}")
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
print(f" 公司: {data.get('company_name')}")
|
|
print(f" 状态: {data.get('status')}")
|
|
print(f" 模型: {data.get('model_used')}")
|
|
if data.get('canvas_summary'):
|
|
summary = data['canvas_summary']
|
|
print(f" 评分: {summary.get('overall_score', 'N/A')}")
|
|
print(f" 建议: {summary.get('recommendation', 'N/A')[:50]}..." if summary.get('recommendation') else " 建议: N/A")
|
|
if data.get('error_message'):
|
|
print(f" 错误: {data.get('error_message')}")
|
|
return resp.json() if resp.status_code == 200 else None
|
|
|
|
def wait_for_analysis(token, analysis_id, max_wait=60):
|
|
"""等待分析完成"""
|
|
print(f"\n⏱️ 等待分析完成 (最多 {max_wait} 秒)...")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
for i in range(max_wait):
|
|
resp = requests.get(f"{API_BASE}/deep-analysis/{analysis_id}", headers=headers)
|
|
status = "unknown"
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
status = data.get('status', 'unknown')
|
|
if status == 'completed':
|
|
print(f" ✅ 分析完成！")
|
|
return True
|
|
elif status == 'failed':
|
|
print(f" ❌ 分析失败: {data.get('error_message', '未知错误')}")
|
|
return False
|
|
|
|
if i % 5 == 0:
|
|
print(f" ... 已等待 {i} 秒，状态: {status}")
|
|
time.sleep(1)
|
|
|
|
print(f" ⚠️ 超时，分析可能还在进行中")
|
|
return None
|
|
|
|
def test_cancel_queue(token):
|
|
"""测试取消队列"""
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
resp = requests.post(f"{API_BASE}/deep-analysis/queue/cancel", headers=headers)
|
|
print(f"\n🛑 取消队列:")
|
|
print(f" Status: {resp.status_code}")
|
|
if resp.status_code == 200:
|
|
print(f" Response: {resp.json()}")
|
|
|
|
def main():
|
|
print("=" * 70)
|
|
print("🧪 深度分析功能完整测试")
|
|
print("=" * 70)
|
|
|
|
# 1. 登录
|
|
token = login()
|
|
|
|
# 2. 配置
|
|
test_config(token)
|
|
|
|
# 3. 检查客户状态
|
|
test_customer_status(token, 1)
|
|
|
|
# 4. 创建分析任务
|
|
created_ids = test_create_analysis(token, 1)
|
|
|
|
if not created_ids:
|
|
print("\n⚠️ 没有创建新任务（可能已存在进行中/待处理的任务）")
|
|
# 尝试获取现有任务
|
|
items = test_list_analyses(token)
|
|
if items:
|
|
created_ids = [items[0]['id']]
|
|
|
|
# 5. 查看队列状态
|
|
test_queue_status(token)
|
|
|
|
# 6. 查看列表
|
|
test_list_analyses(token)
|
|
|
|
# 7. 等待分析完成
|
|
if created_ids:
|
|
result = wait_for_analysis(token, created_ids[0], max_wait=30)
|
|
|
|
# 8. 查看详情
|
|
test_get_detail(token, created_ids[0])
|
|
|
|
# 9. 最终队列状态
|
|
test_queue_status(token)
|
|
|
|
print("\n" + "=" * 70)
|
|
print("🎉 测试完成!")
|
|
print("=" * 70)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|