T_Sentiment-Analysis-System-for-Aerial-Tweets/test_api_connection.py

"""测试 DeepSeek API 连接"""

import os
import asyncio
import httpx
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

async def test_deepseek_api():
    """测试 DeepSeek API 连接"""
    
    api_key = os.getenv("DEEPSEEK_API_KEY")
    base_url = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1")
    
    print("🔍 开始测试 DeepSeek API 连接...")
    print(f"API Key: {'已配置' if api_key else '未配置'}")
    print(f"Base URL: {base_url}")
    
    if not api_key:
        print("❌ 错误: 未找到 DEEPSEEK_API_KEY 环境变量")
        return False
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    # 简单的测试消息
    data = {
        "model": "deepseek-chat",
        "messages": [
            {"role": "system", "content": "你是一个测试助手，只需要回复'连接成功'即可。"},
            {"role": "user", "content": "请回复'连接成功'"}
        ],
        "temperature": 0.1,
        "max_tokens": 10,
        "stream": False
    }
    
    try:
        print("🔄 发送测试请求...")
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                f"{base_url}/chat/completions",
                headers=headers,
                json=data
            )
            
        print(f"📊 响应状态码: {response.status_code}")
        
        if response.status_code == 200:
            result = response.json()
            content = result["choices"][0]["message"]["content"]
            print(f"✅ API 连接成功!")
            print(f"📝 响应内容: {content}")
            print(f"🔢 使用令牌数: {result.get('usage', {}).get('total_tokens', '未知')}")
            return True
        else:
            print(f"❌ API 调用失败: {response.status_code}")
            print(f"📋 错误信息: {response.text}")
            return False
            
    except httpx.ConnectError as e:
        print(f"❌ 连接错误: {e}")
        print("💡 请检查网络连接和API地址")
        return False
    except httpx.TimeoutException as e:
        print(f"❌ 请求超时: {e}")
        print("💡 请检查网络连接或增加超时时间")
        return False
    except Exception as e:
        print(f"❌ 未知错误: {e}")
        return False

async def test_agent_integration():
    """测试 Agent 集成"""
    
    print("\n🧪 测试 Agent 集成...")
    
    try:
        from src.deepseek_agent import DeepSeekTweetAgent
        
        agent = DeepSeekTweetAgent()
        
        # 测试推文分析
        test_tweet = "@United This is the worst airline ever! My flight was delayed for 5 hours and the staff was rude."
        test_airline = "United"
        
        print(f"📝 测试推文: {test_tweet}")
        print(f"✈️ 航空公司: {test_airline}")
        
        result = await agent.analyze_sentiment(test_tweet, test_airline)
        
        print("✅ Agent 集成测试成功!")
        print(f"😊 情感分析结果: {result.sentiment}")
        print(f"📊 置信度: {result.confidence:.1%}")
        print(f"🔍 关键因素: {result.key_factors}")
        
        return True
        
    except ImportError as e:
        print(f"❌ 导入错误: {e}")
        return False
    except Exception as e:
        print(f"❌ Agent 测试失败: {e}")
        return False

async def main():
    """主测试函数"""
    
    print("=" * 50)
    print("🚀 DeepSeek API 连接测试")
    print("=" * 50)
    
    # 测试基础 API 连接
    api_success = await test_deepseek_api()
    
    # 测试 Agent 集成
    agent_success = await test_agent_integration()
    
    print("\n" + "=" * 50)
    print("📋 测试结果汇总")
    print("=" * 50)
    
    if api_success and agent_success:
        print("🎉 所有测试通过! DeepSeek API 集成成功!")
        print("💡 您现在可以正常使用增强版应用了")
    else:
        print("⚠️ 部分测试失败，请检查配置")
        
        if not api_success:
            print("❌ API 连接测试失败")
            print("💡 请检查:")
            print("  1. API Key 是否正确配置")
            print("  2. 网络连接是否正常")
            print("  3. API 地址是否正确")
            
        if not agent_success:
            print("❌ Agent 集成测试失败")
            print("💡 请检查模块导入和依赖安装")

if __name__ == "__main__":
    asyncio.run(main())