012-kaopeilian/tests/test_ai_functions.py

#!/usr/bin/env python3
"""
考培练系统 AI 功能完整测试脚本

测试所有 AI 功能是否正常：
1. AI 基础服务连通性（4sapi + OpenRouter）
2. 试题生成功能（V2 引擎）
3. 答案判断功能（V2 引擎）
4. 知识点分析功能（V2 引擎）
5. 课程对话功能（V2 引擎）
6. 陪练场景提取功能（V2 引擎）
7. 能力分析功能（V2 引擎）

测试方式：通过 HTTP API 调用后端服务
"""

import asyncio
import httpx
import json
import sys
import time
from datetime import datetime
from typing import Optional, Dict, Any, List

# ==================== 配置 ====================

# 测试基础URL（演示版后端）
BASE_URL = "http://localhost:8000/api/v1"

# 测试账户（从数据库获取有效用户）
TEST_USERNAME = "admin"
TEST_PASSWORD = "admin123"

# 测试超时时间（秒）
TIMEOUT = 120

# 测试结果统计
test_results: List[Dict[str, Any]] = []

# ==================== 工具函数 ====================

def print_header(title: str):
    """打印标题"""
    print("\n" + "=" * 60)
    print(f"  {title}")
    print("=" * 60)

def print_result(test_name: str, success: bool, message: str, duration_ms: int = 0):
    """打印测试结果"""
    status = "✅ 通过" if success else "❌ 失败"
    duration_str = f" ({duration_ms}ms)" if duration_ms > 0 else ""
    print(f"  {status} | {test_name}{duration_str}")
    if not success:
        print(f"       └─ {message}")
    
    # 记录结果
    test_results.append({
        "name": test_name,
        "success": success,
        "message": message,
        "duration_ms": duration_ms,
        "timestamp": datetime.now().isoformat()
    })

async def get_auth_token(client: httpx.AsyncClient) -> Optional[str]:
    """获取认证 token"""
    try:
        response = await client.post(
            f"{BASE_URL}/auth/login",
            data={"username": TEST_USERNAME, "password": TEST_PASSWORD}
        )
        if response.status_code == 200:
            data = response.json()
            return data.get("data", {}).get("access_token")
        else:
            print(f"  ⚠️ 登录失败: {response.status_code} - {response.text}")
            return None
    except Exception as e:
        print(f"  ⚠️ 登录异常: {e}")
        return None

# ==================== 测试用例 ====================

async def test_1_ai_basic_connectivity():
    """测试1: AI 基础服务连通性"""
    print_header("测试1: AI 基础服务连通性")
    
    # 直接测试 4sapi
    async with httpx.AsyncClient(timeout=30) as client:
        # 4sapi 测试
        start = time.time()
        try:
            response = await client.post(
                "https://4sapi.com/v1/chat/completions",
                headers={
                    "Authorization": "Bearer sk-9yMCXjRGANbacz20kJY8doSNy6Rf446aYwmgGIuIXQ7DAyBw",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gemini-3-flash-preview",
                    "messages": [{"role": "user", "content": "你好，请回复'测试成功'"}],
                    "max_tokens": 50
                }
            )
            duration = int((time.time() - start) * 1000)
            
            if response.status_code == 200:
                data = response.json()
                content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
                print_result("4sapi.com 连通性", True, f"响应: {content[:50]}...", duration)
            else:
                print_result("4sapi.com 连通性", False, f"HTTP {response.status_code}: {response.text[:100]}", duration)
        except Exception as e:
            duration = int((time.time() - start) * 1000)
            print_result("4sapi.com 连通性", False, str(e), duration)
        
        # OpenRouter 测试
        start = time.time()
        try:
            response = await client.post(
                "https://openrouter.ai/api/v1/chat/completions",
                headers={
                    "Authorization": "Bearer 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",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "google/gemini-3-flash-preview",
                    "messages": [{"role": "user", "content": "你好，请回复'测试成功'"}],
                    "max_tokens": 50
                }
            )
            duration = int((time.time() - start) * 1000)
            
            if response.status_code == 200:
                data = response.json()
                content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
                print_result("OpenRouter 连通性", True, f"响应: {content[:50]}...", duration)
            else:
                print_result("OpenRouter 连通性", False, f"HTTP {response.status_code}: {response.text[:100]}", duration)
        except Exception as e:
            duration = int((time.time() - start) * 1000)
            print_result("OpenRouter 连通性", False, str(e), duration)

async def test_2_exam_generation():
    """测试2: 试题生成功能"""
    print_header("测试2: 试题生成功能 (V2 引擎)")
    
    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        # 获取 token
        token = await get_auth_token(client)
        if not token:
            print_result("试题生成", False, "无法获取认证 token")
            return
        
        headers = {"Authorization": f"Bearer {token}"}
        
        # 先查询一个有知识点的课程
        try:
            response = await client.get(f"{BASE_URL}/courses", headers=headers)
            if response.status_code != 200:
                print_result("试题生成 - 查询课程", False, f"HTTP {response.status_code}")
                return
            
            courses = response.json().get("data", {}).get("items", [])
            if not courses:
                print_result("试题生成", False, "没有可用的课程")
                return
            
            course_id = courses[0].get("id")
            print(f"  📝 使用课程ID: {course_id}")
        except Exception as e:
            print_result("试题生成 - 查询课程", False, str(e))
            return
        
        # 测试试题生成 API
        start = time.time()
        try:
            response = await client.post(
                f"{BASE_URL}/exams/generate?engine=v2",
                headers=headers,
                json={
                    "course_id": course_id,
                    "single_choice_count": 2,
                    "multiple_choice_count": 1,
                    "true_false_count": 1,
                    "fill_blank_count": 0,
                    "essay_count": 0,
                    "difficulty_level": 3,
                    "current_round": 1
                }
            )
            duration = int((time.time() - start) * 1000)
            
            if response.status_code == 200:
                data = response.json()
                result = data.get("data", {}).get("result", "")
                try:
                    questions = json.loads(result) if result else []
                    print_result("试题生成 (V2)", True, f"生成了 {len(questions)} 道题目", duration)
                except:
                    print_result("试题生成 (V2)", True, f"返回结果: {result[:100]}...", duration)
            else:
                print_result("试题生成 (V2)", False, f"HTTP {response.status_code}: {response.text[:200]}", duration)
        except Exception as e:
            duration = int((time.time() - start) * 1000)
            print_result("试题生成 (V2)", False, str(e), duration)

async def test_3_answer_judge():
    """测试3: 答案判断功能"""
    print_header("测试3: 答案判断功能 (V2 引擎)")
    
    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        # 获取 token
        token = await get_auth_token(client)
        if not token:
            print_result("答案判断", False, "无法获取认证 token")
            return
        
        headers = {"Authorization": f"Bearer {token}"}
        
        # 测试正确答案
        start = time.time()
        try:
            response = await client.post(
                f"{BASE_URL}/exams/judge-answer?engine=v2",
                headers=headers,
                json={
                    "question": "玻尿酸的主要作用是什么？",
                    "correct_answer": "保湿、填充、塑形",
                    "user_answer": "保湿和填充",
                    "analysis": "玻尿酸具有强大的保水能力，可用于皮肤保湿和面部填充"
                }
            )
            duration = int((time.time() - start) * 1000)
            
            if response.status_code == 200:
                data = response.json()
                is_correct = data.get("data", {}).get("is_correct")
                print_result("答案判断 (V2)", True, f"判断结果: {'正确' if is_correct else '错误'}", duration)
            else:
                print_result("答案判断 (V2)", False, f"HTTP {response.status_code}: {response.text[:200]}", duration)
        except Exception as e:
            duration = int((time.time() - start) * 1000)
            print_result("答案判断 (V2)", False, str(e), duration)

async def test_4_knowledge_analysis():
    """测试4: 知识点分析功能"""
    print_header("测试4: 知识点分析功能 (V2 引擎)")
    
    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        # 获取 token
        token = await get_auth_token(client)
        if not token:
            print_result("知识点分析", False, "无法获取认证 token")
            return
        
        headers = {"Authorization": f"Bearer {token}"}
        
        # 先查询课程和资料
        try:
            response = await client.get(f"{BASE_URL}/courses", headers=headers)
            if response.status_code != 200:
                print_result("知识点分析 - 查询课程", False, f"HTTP {response.status_code}")
                return
            
            courses = response.json().get("data", {}).get("items", [])
            if not courses:
                print_result("知识点分析", False, "没有可用的课程")
                return
            
            course_id = courses[0].get("id")
            
            # 查询课程资料
            response = await client.get(f"{BASE_URL}/courses/{course_id}/materials", headers=headers)
            if response.status_code != 200:
                print_result("知识点分析 - 查询资料", False, f"HTTP {response.status_code}")
                return
            
            materials = response.json().get("data", {}).get("items", [])
            if not materials:
                print_result("知识点分析", False, "课程没有可用的资料")
                return
            
            material_id = materials[0].get("id")
            print(f"  📝 使用课程ID: {course_id}, 资料ID: {material_id}")
        except Exception as e:
            print_result("知识点分析 - 查询资料", False, str(e))
            return
        
        # 测试知识点分析 API
        start = time.time()
        try:
            response = await client.post(
                f"{BASE_URL}/courses/{course_id}/materials/{material_id}/analyze?engine=v2",
                headers=headers
            )
            duration = int((time.time() - start) * 1000)
            
            if response.status_code == 200:
                data = response.json()
                kps = data.get("data", {}).get("knowledge_points", [])
                print_result("知识点分析 (V2)", True, f"提取了 {len(kps)} 个知识点", duration)
            else:
                # 可能返回 400 因为没有文件内容
                error_detail = response.json().get("detail", response.text[:200])
                print_result("知识点分析 (V2)", False, f"HTTP {response.status_code}: {error_detail}", duration)
        except Exception as e:
            duration = int((time.time() - start) * 1000)
            print_result("知识点分析 (V2)", False, str(e), duration)

async def test_5_course_chat():
    """测试5: 课程对话功能"""
    print_header("测试5: 课程对话功能 (V2 引擎)")
    
    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        # 获取 token
        token = await get_auth_token(client)
        if not token:
            print_result("课程对话", False, "无法获取认证 token")
            return
        
        headers = {"Authorization": f"Bearer {token}"}
        
        # 先查询课程
        try:
            response = await client.get(f"{BASE_URL}/courses", headers=headers)
            if response.status_code != 200:
                print_result("课程对话 - 查询课程", False, f"HTTP {response.status_code}")
                return
            
            courses = response.json().get("data", {}).get("items", [])
            if not courses:
                print_result("课程对话", False, "没有可用的课程")
                return
            
            course_id = courses[0].get("id")
            print(f"  📝 使用课程ID: {course_id}")
        except Exception as e:
            print_result("课程对话 - 查询课程", False, str(e))
            return
        
        # 测试课程对话 API（流式响应）
        start = time.time()
        try:
            # 注意：课程对话是流式响应，需要特殊处理
            async with client.stream(
                "POST",
                f"{BASE_URL}/course/chat?engine=v2",
                headers=headers,
                json={
                    "course_id": course_id,
                    "query": "请简单介绍一下这门课程的主要内容"
                }
            ) as response:
                if response.status_code == 200:
                    content = ""
                    async for line in response.aiter_lines():
                        if line.startswith("data:"):
                            try:
                                data = json.loads(line[5:].strip())
                                if data.get("event") == "message_chunk":
                                    content += data.get("content", "")
                            except:
                                pass
                    
                    duration = int((time.time() - start) * 1000)
                    if content:
                        print_result("课程对话 (V2)", True, f"响应: {content[:100]}...", duration)
                    else:
                        print_result("课程对话 (V2)", True, "收到流式响应", duration)
                else:
                    duration = int((time.time() - start) * 1000)
                    body = await response.aread()
                    print_result("课程对话 (V2)", False, f"HTTP {response.status_code}: {body.decode()[:200]}", duration)
        except Exception as e:
            duration = int((time.time() - start) * 1000)
            print_result("课程对话 (V2)", False, str(e), duration)

async def test_6_practice_scene():
    """测试6: 陪练场景提取功能"""
    print_header("测试6: 陪练场景提取功能 (V2 引擎)")
    
    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        # 获取 token
        token = await get_auth_token(client)
        if not token:
            print_result("陪练场景提取", False, "无法获取认证 token")
            return
        
        headers = {"Authorization": f"Bearer {token}"}
        
        # 先查询课程
        try:
            response = await client.get(f"{BASE_URL}/courses", headers=headers)
            if response.status_code != 200:
                print_result("陪练场景提取 - 查询课程", False, f"HTTP {response.status_code}")
                return
            
            courses = response.json().get("data", {}).get("items", [])
            if not courses:
                print_result("陪练场景提取", False, "没有可用的课程")
                return
            
            course_id = courses[0].get("id")
            print(f"  📝 使用课程ID: {course_id}")
        except Exception as e:
            print_result("陪练场景提取 - 查询课程", False, str(e))
            return
        
        # 测试陪练场景提取 API
        start = time.time()
        try:
            response = await client.post(
                f"{BASE_URL}/practice/extract-scene?engine=v2",
                headers=headers,
                json={
                    "course_id": course_id,
                    "scene_type": "consultation"  # 咨询场景
                }
            )
            duration = int((time.time() - start) * 1000)
            
            if response.status_code == 200:
                data = response.json()
                scenes = data.get("data", {}).get("scenes", [])
                print_result("陪练场景提取 (V2)", True, f"提取了 {len(scenes)} 个场景", duration)
            else:
                error_detail = response.json().get("detail", response.text[:200])
                print_result("陪练场景提取 (V2)", False, f"HTTP {response.status_code}: {error_detail}", duration)
        except Exception as e:
            duration = int((time.time() - start) * 1000)
            print_result("陪练场景提取 (V2)", False, str(e), duration)

async def test_7_ability_analysis():
    """测试7: 能力分析功能"""
    print_header("测试7: 能力分析功能 (V2 引擎)")
    
    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        # 获取 token
        token = await get_auth_token(client)
        if not token:
            print_result("能力分析", False, "无法获取认证 token")
            return
        
        headers = {"Authorization": f"Bearer {token}"}
        
        # 测试能力分析 API（需要员工的言迹数据）
        start = time.time()
        try:
            # 先查询是否有员工数据
            response = await client.get(f"{BASE_URL}/users/me", headers=headers)
            if response.status_code != 200:
                print_result("能力分析 - 获取用户信息", False, f"HTTP {response.status_code}")
                return
            
            user = response.json().get("data", {})
            user_id = user.get("id")
            print(f"  📝 当前用户ID: {user_id}")
            
            # 测试能力分析 API
            response = await client.get(
                f"{BASE_URL}/ability/user/{user_id}/analysis?engine=v2",
                headers=headers
            )
            duration = int((time.time() - start) * 1000)
            
            if response.status_code == 200:
                data = response.json()
                dimensions = data.get("data", {}).get("dimensions", [])
                print_result("能力分析 (V2)", True, f"分析了 {len(dimensions)} 个能力维度", duration)
            elif response.status_code == 404:
                # 可能没有言迹数据
                print_result("能力分析 (V2)", True, "接口正常（无言迹数据）", duration)
            else:
                error_detail = response.json().get("detail", response.text[:200])
                print_result("能力分析 (V2)", False, f"HTTP {response.status_code}: {error_detail}", duration)
        except Exception as e:
            duration = int((time.time() - start) * 1000)
            print_result("能力分析 (V2)", False, str(e), duration)

async def test_backend_health():
    """测试后端健康状态"""
    print_header("后端服务健康检查")
    
    backends = [
        ("演示版 (aiedu)", "http://localhost:8000"),
        ("瑞小美 (kpl)", "http://localhost:8001"),
        ("华尔倍丽 (hua)", "http://localhost:8010"),
        ("杨扬宠物 (yy)", "http://localhost:8011"),
        ("武汉禾丽 (hl)", "http://localhost:8012"),
        ("芯颜定制 (xy)", "http://localhost:8013"),
        ("飞沃 (fw)", "http://localhost:8014"),
        ("恩喜成都 (ex)", "http://localhost:8015"),
        ("陪练试用版 (pl)", "http://localhost:8020"),
    ]
    
    async with httpx.AsyncClient(timeout=10) as client:
        for name, url in backends:
            start = time.time()
            try:
                response = await client.get(f"{url}/health")
                duration = int((time.time() - start) * 1000)
                
                if response.status_code == 200:
                    print_result(name, True, "健康", duration)
                else:
                    print_result(name, False, f"HTTP {response.status_code}", duration)
            except Exception as e:
                duration = int((time.time() - start) * 1000)
                print_result(name, False, str(e), duration)

def print_summary():
    """打印测试汇总"""
    print_header("测试汇总")
    
    total = len(test_results)
    passed = sum(1 for r in test_results if r["success"])
    failed = total - passed
    
    print(f"\n  总计: {total} 项测试")
    print(f"  ✅ 通过: {passed}")
    print(f"  ❌ 失败: {failed}")
    
    if failed > 0:
        print("\n  失败项目:")
        for r in test_results:
            if not r["success"]:
                print(f"    - {r['name']}: {r['message']}")
    
    # 计算平均延迟
    durations = [r["duration_ms"] for r in test_results if r["duration_ms"] > 0]
    if durations:
        avg_duration = sum(durations) / len(durations)
        print(f"\n  平均响应时间: {avg_duration:.0f}ms")
    
    print("\n" + "=" * 60)
    
    return failed == 0

async def main():
    """主测试入口"""
    print("\n" + "=" * 60)
    print("   考培练系统 AI 功能完整测试")
    print(f"   时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 60)
    
    # 检查后端健康状态
    await test_backend_health()
    
    # 运行所有测试
    await test_1_ai_basic_connectivity()
    await test_2_exam_generation()
    await test_3_answer_judge()
    await test_4_knowledge_analysis()
    await test_5_course_chat()
    await test_6_practice_scene()
    await test_7_ability_analysis()
    
    # 打印汇总
    success = print_summary()
    
    sys.exit(0 if success else 1)

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