Ai_GirlFriend/test_music_library_sing.py

"""
测试音乐库唱歌视频功能
"""
import requests
import json
import time

# 配置
BASE_URL = "http://localhost:30101"
# 需要替换为实际的 token
TOKEN = "your_token_here"

headers = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json"
}


def test_convert_music_to_song():
    """测试转换音乐为系统歌曲"""
    print("\n=== 测试转换音乐为系统歌曲 ===")
    
    # 测试直链音乐（Bensound）
    response = requests.post(
        f"{BASE_URL}/music/convert-to-song",
        headers=headers,
        params={"music_id": 1}
    )
    print(f"状态码: {response.status_code}")
    print(f"响应: {json.dumps(response.json(), indent=2, ensure_ascii=False)}")
    
    if response.status_code == 200:
        data = response.json()
        if data.get("code") == 1:
            song_id = data["data"]["song_id"]
            print(f"✅ 转换成功，song_id: {song_id}")
            return song_id
        else:
            print(f"❌ 转换失败: {data.get('message')}")
            return None
    else:
        print(f"❌ 请求失败")
        return None


def test_convert_external_music():
    """测试转换外部链接音乐（应该失败）"""
    print("\n=== 测试转换外部链接音乐（应该失败） ===")
    
    response = requests.post(
        f"{BASE_URL}/music/convert-to-song",
        headers=headers,
        params={"music_id": 31}  # 假设 31 是网易云音乐
    )
    print(f"状态码: {response.status_code}")
    print(f"响应: {json.dumps(response.json(), indent=2, ensure_ascii=False)}")
    
    if response.status_code == 200:
        data = response.json()
        if data.get("code") != 1:
            print(f"✅ 正确拒绝外部链接音乐")
        else:
            print(f"❌ 不应该允许转换外部链接音乐")
    else:
        print(f"✅ 正确拒绝外部链接音乐")


def test_generate_sing_video(song_id):
    """测试生成唱歌视频"""
    print("\n=== 测试生成唱歌视频 ===")
    
    if not song_id:
        print("❌ 没有 song_id，跳过测试")
        return None
    
    response = requests.post(
        f"{BASE_URL}/sing/generate",
        headers=headers,
        json={"song_id": song_id}
    )
    print(f"状态码: {response.status_code}")
    print(f"响应: {json.dumps(response.json(), indent=2, ensure_ascii=False)}")
    
    if response.status_code == 200:
        data = response.json()
        if data.get("code") == 1:
            task_data = data["data"]
            task_id = task_data.get("generation_task_id")
            status = task_data.get("status")
            
            print(f"✅ 任务创建成功")
            print(f"   任务ID: {task_id}")
            print(f"   状态: {status}")
            
            if status == "succeeded":
                print(f"   视频URL: {task_data.get('video_url')}")
                print(f"   ✅ 立即成功（有缓存）")
            else:
                print(f"   ⏳ 生成中...")
            
            return task_id
        else:
            print(f"❌ 生成失败: {data.get('message')}")
            return None
    else:
        print(f"❌ 请求失败")
        return None


def test_check_task_status(task_id):
    """测试查询任务状态"""
    print("\n=== 测试查询任务状态 ===")
    
    if not task_id:
        print("❌ 没有 task_id，跳过测试")
        return
    
    response = requests.get(
        f"{BASE_URL}/sing/task/{task_id}",
        headers=headers
    )
    print(f"状态码: {response.status_code}")
    print(f"响应: {json.dumps(response.json(), indent=2, ensure_ascii=False)}")
    
    if response.status_code == 200:
        data = response.json()
        if data.get("code") == 1:
            task_data = data["data"]
            status = task_data.get("status")
            print(f"✅ 任务状态: {status}")
            
            if status == "succeeded":
                print(f"   视频URL: {task_data.get('video_url')}")
        else:
            print(f"❌ 查询失败: {data.get('message')}")
    else:
        print(f"❌ 请求失败")


def test_get_sing_history():
    """测试获取唱歌历史记录"""
    print("\n=== 测试获取唱歌历史记录 ===")
    
    response = requests.get(
        f"{BASE_URL}/sing/history",
        headers=headers,
        params={"page": 1, "size": 5}
    )
    print(f"状态码: {response.status_code}")
    
    if response.status_code == 200:
        data = response.json()
        if data.get("code") == 1:
            history_list = data.get("data", [])
            print(f"✅ 历史记录数量: {len(history_list)}")
            
            if history_list:
                print("\n最近的视频:")
                for i, item in enumerate(history_list[:3], 1):
                    print(f"{i}. {item.get('song_title', '未知')} - {item.get('status')}")
                    if item.get('video_url'):
                        print(f"   视频: {item.get('video_url')[:50]}...")
        else:
            print(f"❌ 获取失败: {data.get('message')}")
    else:
        print(f"❌ 请求失败")


def main():
    """主函数"""
    print("=" * 60)
    print("音乐库唱歌视频功能测试")
    print("=" * 60)
    
    # 检查 token
    if TOKEN == "your_token_here":
        print("\n⚠️  警告: 请先设置有效的 TOKEN")
        print("   1. 登录应用获取 token")
        print("   2. 修改此脚本中的 TOKEN 变量")
        print("   3. 重新运行测试")
        return
    
    # 运行测试
    results = []
    
    # 测试 1: 转换音乐
    song_id = test_convert_music_to_song()
    results.append(("转换音乐", song_id is not None))
    
    # 测试 2: 转换外部链接（应该失败）
    test_convert_external_music()
    results.append(("拒绝外部链接", True))  # 手动判断
    
    # 测试 3: 生成视频
    if song_id:
        task_id = test_generate_sing_video(song_id)
        results.append(("生成视频", task_id is not None))
        
        # 测试 4: 查询任务状态
        if task_id:
            time.sleep(2)  # 等待 2 秒
            test_check_task_status(task_id)
            results.append(("查询任务", True))
    
    # 测试 5: 获取历史记录
    test_get_sing_history()
    results.append(("获取历史", True))
    
    # 显示测试结果
    print("\n" + "=" * 60)
    print("测试结果汇总")
    print("=" * 60)
    
    for name, success in results:
        status = "✅ 通过" if success else "❌ 失败"
        print(f"{status} - {name}")
    
    # 统计
    passed = sum(1 for _, success in results if success)
    total = len(results)
    print(f"\n总计: {passed}/{total} 通过")
    
    if passed == total:
        print("\n🎉 所有测试通过！")
    else:
        print(f"\n⚠️  有 {total - passed} 个测试失败")


if __name__ == "__main__":
    main()