test_env.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""测试环境配置和 API 连接"""

import os
from pathlib import Path
from dotenv import load_dotenv

# 加载 .env 文件
env_file = Path('.env')
if env_file.exists():
    load_dotenv(env_file)
    print(f"✓ 已加载 .env 文件")
else:
    print(f"✗ 未找到 .env 文件")

# 检查必要的环境变量
print("\n【环境变量检查】")
print("-" * 60)

api_keys = {
    'DASHSCOPE_API_KEY': os.getenv('DASHSCOPE_API_KEY'),
    'QWEN_API_KEY': os.getenv('QWEN_API_KEY'),
    'LLM_API_KEY': os.getenv('LLM_API_KEY'),
}

for key_name, key_value in api_keys.items():
    if key_value:
        # 显示前缀，隐藏大部分内容
        masked = key_value[:10] + '...' + key_value[-5:] if len(key_value) > 15 else '***'
        print(f"✓ {key_name}: {masked}")
    else:
        print(f"✗ {key_name}: 未配置")

# 检查 LLM 配置
print("\n【LLM 配置】")
print("-" * 60)

llm_config = {
    'LLM_MODEL': os.getenv('LLM_MODEL', 'qwen-vl-max'),
    'LLM_BASE_URL': os.getenv('LLM_BASE_URL', 'https://dashscope.aliyuncs.com/compatible-mode/v1'),
}

for config_name, config_value in llm_config.items():
    print(f"  {config_name}: {config_value}")

# 检查是否能导入必要的模块
print("\n【模块导入检查】")
print("-" * 60)

modules_to_check = [
    'dashscope',
    'langchain',
    'pandas',
    'fastapi',
    'PIL',
]

for module_name in modules_to_check:
    try:
        __import__(module_name)
        print(f"✓ {module_name} 已安装")
    except ImportError:
        print(f"✗ {module_name} 未安装")

# 测试 DashScope 连接
print("\n【DashScope API 测试】")
print("-" * 60)

api_key = os.getenv('DASHSCOPE_API_KEY') or os.getenv('QWEN_API_KEY')

if not api_key:
    print("✗ 未配置 API Key，跳过 API 测试")
    print("\n【解决方案】")
    print("-" * 60)
    print("1. 创建 .env 文件在项目根目录")
    print("2. 添加配置:")
    print("   DASHSCOPE_API_KEY=sk_xxx (从阿里云获取)")
    print("3. 重启服务")
else:
    try:
        from dashscope import Generation
        
        # 简单的测试调用
        response = Generation.call(
            model="qwen-plus",
            messages=[{"role": "user", "content": "Hello"}],
            api_key=api_key,
            max_tokens=10
        )
        
        if response.status_code == 200:
            print(f"✓ API 连接成功")
            print(f"  模型: qwen-plus")
            print(f"  响应: {response.output.choices[0].message.content[:50]}...")
        else:
            print(f"✗ API 返回错误: {response.status_code}")
            print(f"  {response.message}")
    except Exception as e:
        print(f"✗ API 调用失败: {str(e)}")

print("\n" + "=" * 60)