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

"""
执行V2版本数据库迁移脚本

运行方式：
    python scripts/run_migrations_v2.py

功能：
    - 执行4个V2版本的数据库迁移文件
    - 创建新表：asr_recognition_v2, audio_synthesis_v2, voice_clone_v2, long_text_audio
    - 不影响现有表和服务
"""
import os
import sys
from pathlib import Path

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

from app.database import engine
from sqlalchemy import text


def run_migration_file(filepath: str) -> bool:
    """执行单个迁移文件"""
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            sql_content = f.read()
        
        with engine.connect() as conn:
            # 执行SQL（可能包含多个语句）
            conn.execute(text(sql_content))
            conn.commit()
        
        print(f"✅ 成功执行: {os.path.basename(filepath)}")
        return True
    except Exception as e:
        print(f"❌ 执行失败: {os.path.basename(filepath)}")
        print(f"   错误信息: {e}")
        return False


def main():
    """主函数"""
    print("=" * 60)
    print("开始执行V2版本数据库迁移")
    print("=" * 60)
    
    # 迁移文件列表（按顺序执行）
    migration_files = [
        "migrations/030_create_asr_recognition_v2_table.sql",
        "migrations/031_create_audio_synthesis_v2_table.sql",
        "migrations/032_create_voice_clone_v2_table.sql",
        "migrations/033_create_long_text_audio_table.sql",
    ]
    
    success_count = 0
    failed_count = 0
    
    for migration_file in migration_files:
        filepath = project_root / migration_file
        
        if not filepath.exists():
            print(f"⚠️  文件不存在: {migration_file}")
            failed_count += 1
            continue
        
        print(f"\n执行迁移: {migration_file}")
        if run_migration_file(str(filepath)):
            success_count += 1
        else:
            failed_count += 1
    
    print("\n" + "=" * 60)
    print(f"迁移完成: 成功 {success_count} 个, 失败 {failed_count} 个")
    print("=" * 60)
    
    if failed_count > 0:
        print("\n⚠️  部分迁移失败，请检查错误信息")
        sys.exit(1)
    else:
        print("\n✅ 所有迁移成功执行！")
        print("\n新表已创建：")
        print("  - asr_recognition_v2 (语音识别V2)")
        print("  - audio_synthesis_v2 (语音合成V2)")
        print("  - voice_clone_v2 (声音克隆V2)")
        print("  - long_text_audio (长文本转音频)")
        print("\n现在可以启动服务并测试V2 API端点")


if __name__ == "__main__":
    main()