Initial commit: AI Interview System
This commit is contained in:
167
backend/test_coze.py
Normal file
167
backend/test_coze.py
Normal file
@@ -0,0 +1,167 @@
|
||||
"""
|
||||
Coze API 配置测试脚本
|
||||
运行: python test_coze.py
|
||||
"""
|
||||
import asyncio
|
||||
import httpx
|
||||
|
||||
# 配置
|
||||
COZE_API_BASE = "https://api.coze.cn"
|
||||
COZE_PAT_TOKEN = "pat_nd1wU47WyPS9GCIyJ1clnH8h1WOQXGrYELX8w73TnSZaYbFdYD4swIhzcETBUbfT"
|
||||
COZE_BOT_ID = "7595113005181386792"
|
||||
COZE_DATABASE_ID = "7595077053909712922"
|
||||
|
||||
|
||||
async def test_bot_info():
|
||||
"""测试 Bot 信息"""
|
||||
print("\n" + "=" * 50)
|
||||
print("1. 测试 Bot 信息")
|
||||
print("=" * 50)
|
||||
|
||||
url = f"{COZE_API_BASE}/v1/bot/get_online_info"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {COZE_PAT_TOKEN}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
try:
|
||||
response = await client.get(
|
||||
url,
|
||||
params={"bot_id": COZE_BOT_ID},
|
||||
headers=headers
|
||||
)
|
||||
data = response.json()
|
||||
print(f"Status: {response.status_code}")
|
||||
print(f"Response: {data}")
|
||||
|
||||
if data.get("code") == 0:
|
||||
bot_info = data.get("data", {})
|
||||
print(f"\n✅ Bot 名称: {bot_info.get('name', 'N/A')}")
|
||||
print(f"✅ Bot ID: {COZE_BOT_ID}")
|
||||
return True
|
||||
else:
|
||||
print(f"\n❌ 错误: {data.get('msg', 'Unknown error')}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"❌ 请求失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def test_database():
|
||||
"""测试数据库查询"""
|
||||
print("\n" + "=" * 50)
|
||||
print("2. 测试数据库连接和字段结构")
|
||||
print("=" * 50)
|
||||
|
||||
url = f"{COZE_API_BASE}/v1/database/query"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {COZE_PAT_TOKEN}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {
|
||||
"database_id": COZE_DATABASE_ID,
|
||||
"page": 1,
|
||||
"page_size": 5,
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
try:
|
||||
response = await client.post(url, json=payload, headers=headers)
|
||||
data = response.json()
|
||||
print(f"Status: {response.status_code}")
|
||||
|
||||
if data.get("code") == 0:
|
||||
db_data = data.get("data", {})
|
||||
records = db_data.get("records", [])
|
||||
total = db_data.get("total", 0)
|
||||
|
||||
print(f"\n✅ 数据库连接成功!")
|
||||
print(f"✅ 数据库 ID: {COZE_DATABASE_ID}")
|
||||
print(f"✅ 总记录数: {total}")
|
||||
|
||||
if records:
|
||||
print(f"\n📋 数据库字段列表:")
|
||||
first_record = records[0]
|
||||
for key in first_record.keys():
|
||||
value = first_record[key]
|
||||
value_preview = str(value)[:50] + "..." if len(str(value)) > 50 else value
|
||||
print(f" - {key}: {value_preview}")
|
||||
else:
|
||||
print("\n⚠️ 数据库为空,无法显示字段结构")
|
||||
print(" 这是正常的,首次面试后会有数据")
|
||||
|
||||
return True
|
||||
else:
|
||||
print(f"\n❌ 错误: {data.get('msg', 'Unknown error')}")
|
||||
print(f" 完整响应: {data}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"❌ 请求失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def test_audio_room():
|
||||
"""测试语音房间创建(不实际创建,只检查 API 可用性)"""
|
||||
print("\n" + "=" * 50)
|
||||
print("3. 测试语音房间 API(模拟请求)")
|
||||
print("=" * 50)
|
||||
|
||||
# 我们只测试 API 是否可访问,不实际创建房间
|
||||
url = f"{COZE_API_BASE}/v1/audio/rooms"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {COZE_PAT_TOKEN}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
# 使用测试参数
|
||||
payload = {
|
||||
"bot_id": COZE_BOT_ID,
|
||||
"user_id": "test_user_001",
|
||||
}
|
||||
|
||||
print(f"API 端点: {url}")
|
||||
print(f"请求参数: {payload}")
|
||||
print("\n⚠️ 跳过实际创建房间测试(避免产生费用)")
|
||||
print("✅ API 配置看起来正确")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def main():
|
||||
print("\n🔍 Coze API 配置测试")
|
||||
print("=" * 50)
|
||||
print(f"API Base: {COZE_API_BASE}")
|
||||
print(f"Bot ID: {COZE_BOT_ID}")
|
||||
print(f"Database ID: {COZE_DATABASE_ID}")
|
||||
print(f"Token: {COZE_PAT_TOKEN[:20]}...")
|
||||
|
||||
results = []
|
||||
|
||||
# 测试 Bot
|
||||
results.append(await test_bot_info())
|
||||
|
||||
# 测试数据库
|
||||
results.append(await test_database())
|
||||
|
||||
# 测试语音房间
|
||||
results.append(await test_audio_room())
|
||||
|
||||
# 汇总
|
||||
print("\n" + "=" * 50)
|
||||
print("📊 测试结果汇总")
|
||||
print("=" * 50)
|
||||
|
||||
tests = ["Bot 信息", "数据库连接", "语音房间 API"]
|
||||
for i, (test_name, result) in enumerate(zip(tests, results)):
|
||||
status = "✅ 通过" if result else "❌ 失败"
|
||||
print(f"{i + 1}. {test_name}: {status}")
|
||||
|
||||
if all(results):
|
||||
print("\n🎉 所有测试通过!配置正确!")
|
||||
else:
|
||||
print("\n⚠️ 部分测试失败,请检查配置")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user