Maas-Model-platform/backend/scripts/test_photo_answer.py

"""
拍照解题功能测试脚本

测试百炼EduTutor API的调用
注意：需要提供用户的 DASHSCOPE_API_KEY
"""
import os
import sys
import asyncio
from pathlib import Path

# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

from app.services.edututor_client import BailianEduTutorClient


async def test_answer_stream(api_key: str):
    """测试流式解答"""
    print("=" * 50)
    print("测试拍照解题流式API")
    print("=" * 50)
    
    # 创建客户端（使用用户的 API Key）
    client = BailianEduTutorClient(api_key=api_key)
    
    # 测试图片URL（需要替换为实际的图片URL）
    test_image_url = input("\n请输入题目图片URL: ").strip()
    if not test_image_url:
        test_image_url = "https://your-oss-bucket.oss-cn-hangzhou.aliyuncs.com/test-question.jpg"
    
    print(f"\n图片URL: {test_image_url}")
    print(f"年级: 6 (六年级)")
    print(f"学段: primary (小学)")
    print(f"学科: math (数学)")
    print("\n开始解题...\n")
    
    try:
        full_answer = ""
        async for event in client.answer_sse_async(
            image_url=test_image_url,
            grade=6,
            stage='primary',
            subject='math'
        ):
            if event['type'] == 'chunk':
                content = event['content']
                print(content, end='', flush=True)
                full_answer += content
            elif event['type'] == 'finish':
                tokens = event.get('tokens', {})
                print(f"\n\n{'=' * 50}")
                print(f"解题完成！")
                print(f"输入Token: {tokens.get('input', 0)}")
                print(f"输出Token: {tokens.get('output', 0)}")
                print(f"{'=' * 50}")
        
        print(f"\n完整答案长度: {len(full_answer)} 字符")
        
    except Exception as e:
        print(f"\n错误: {e}")
        import traceback
        traceback.print_exc()


def test_answer_sync(api_key: str):
    """测试同步解答"""
    print("=" * 50)
    print("测试拍照解题同步API")
    print("=" * 50)
    
    # 创建客户端（使用用户的 API Key）
    client = BailianEduTutorClient(api_key=api_key)
    
    # 测试图片URL
    test_image_url = input("\n请输入题目图片URL: ").strip()
    if not test_image_url:
        test_image_url = "https://your-oss-bucket.oss-cn-hangzhou.aliyuncs.com/test-question.jpg"
    
    print(f"\n图片URL: {test_image_url}")
    print(f"年级: 6 (六年级)")
    print(f"学段: primary (小学)")
    print(f"学科: math (数学)")
    print("\n开始解题...\n")
    
    try:
        result = client.answer_sync(
            image_url=test_image_url,
            grade=6,
            stage='primary',
            subject='math'
        )
        
        print(f"解答内容:\n{result['answer']}")
        print(f"\n{'=' * 50}")
        print(f"输入Token: {result['input_tokens']}")
        print(f"输出Token: {result['output_tokens']}")
        print(f"{'=' * 50}")
        
    except Exception as e:
        print(f"\n错误: {e}")
        import traceback
        traceback.print_exc()


def check_environment():
    """检查环境配置"""
    print("=" * 50)
    print("检查环境配置")
    print("=" * 50)
    
    required_vars = [
        'BAILIAN_WORKSPACE_ID'
    ]
    
    missing_vars = []
    for var in required_vars:
        value = os.getenv(var)
        if value:
            print(f"✓ {var}: {value}")
        else:
            print(f"✗ {var}: 未设置")
            missing_vars.append(var)
    
    print("=" * 50)
    
    if missing_vars:
        print(f"\n警告: 缺少以下必需环境变量: {', '.join(missing_vars)}")
        print("请在 .env 文件中配置这些变量")
        return False
    
    print("\n✓ 环境配置检查通过")
    return True


def main():
    """主函数"""
    print("\n拍照解题功能测试\n")
    
    # 检查环境
    if not check_environment():
        return
    
    # 获取用户 API Key
    print("\n" + "=" * 50)
    api_key = input("请输入用户的 DASHSCOPE_API_KEY: ").strip()
    if not api_key:
        print("错误: API Key 不能为空")
        return
    
    # 选择测试模式
    print("\n请选择测试模式:")
    print("1. 流式解答测试 (推荐)")
    print("2. 同步解答测试")
    print("3. 退出")
    
    choice = input("\n请输入选项 (1-3): ").strip()
    
    if choice == '1':
        asyncio.run(test_answer_stream(api_key))
    elif choice == '2':
        test_answer_sync(api_key)
    elif choice == '3':
        print("退出测试")
    else:
        print("无效选项")


if __name__ == '__main__':
    main()