agent-Specialization/scripts/qq_bot/test_connection.py

102 lines
2.9 KiB
Python
Executable File

#!/usr/bin/env python3
"""测试 NapCat 和 Agents API 连接"""
import asyncio
import sys
from pathlib import Path
project_root = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(project_root))
from scripts.qq_bot.napcat_client import NapCatClient
from scripts.qq_bot.web_api_client import WebAPIClient
from scripts.qq_bot.config import (
NAPCAT_HOST,
NAPCAT_TOKEN,
AGENTS_HOST,
AGENTS_EMAIL,
TARGET_GROUP_ID,
)
async def test_napcat():
"""测试 NapCat 连接"""
print("=" * 60)
print("测试 NapCat 连接...")
print(f"Host: {NAPCAT_HOST}")
print(f"Token: {NAPCAT_TOKEN[:10]}...")
print("=" * 60)
try:
async with NapCatClient() as client:
# 获取登录信息
result = await client.get_login_info()
if result.get("status") == "ok":
data = result.get("data", {})
print(f"✅ NapCat 连接成功")
print(f" 机器人 QQ 号: {data.get('user_id')}")
print(f" 昵称: {data.get('nickname')}")
return True
else:
print(f"❌ NapCat 连接失败: {result}")
return False
except Exception as e:
print(f"❌ NapCat 连接异常: {e}")
return False
async def test_agents():
"""测试 Agents API 连接"""
print("\n" + "=" * 60)
print("测试 Agents API 连接...")
print(f"Host: {AGENTS_HOST}")
print(f"Email: {AGENTS_EMAIL}")
print("=" * 60)
try:
async with WebAPIClient() as client:
print(f"✅ Agents API 连接成功")
print(f" 已登录为: {AGENTS_EMAIL}")
# 测试创建对话
result = await client.create_conversation()
if result.get("success"):
conv_id = result.get("conversation_id")
print(f"✅ 创建对话成功: {conv_id}")
return True
else:
print(f"❌ 创建对话失败: {result}")
return False
except Exception as e:
print(f"❌ Agents API 连接异常: {e}")
import traceback
traceback.print_exc()
return False
async def main():
"""主测试函数"""
print("\n🤖 QQ Bot 配置测试\n")
napcat_ok = await test_napcat()
agents_ok = await test_agents()
print("\n" + "=" * 60)
print("测试结果:")
print(f" NapCat: {'✅ 通过' if napcat_ok else '❌ 失败'}")
print(f" Agents: {'✅ 通过' if agents_ok else '❌ 失败'}")
print(f" 目标群号: {TARGET_GROUP_ID}")
print("=" * 60)
if napcat_ok and agents_ok:
print("\n✅ 所有测试通过,可以启动 QQ Bot")
return 0
else:
print("\n❌ 部分测试失败,请检查配置")
return 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)