xiaoxia-saas/scripts/seed_test_voice_clones.py

#!/usr/bin/env python3
"""Seed test voice clone profiles for staging E2E testing.

Usage:
    # On staging (inside Docker container):
    docker exec -it staging-api python scripts/seed_test_voice_clones.py

    # Locally (requires DATABASE_URL env):
    DATABASE_URL=postgresql+psycopg://... python scripts/seed_test_voice_clones.py

Creates test VoiceCloneProfile records for known staging test accounts,
so that the frontend E2E tests can verify clone voice list and preview.
"""

from __future__ import annotations

import logging
import os
import sys
from datetime import datetime, timezone
from uuid import uuid4

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# Test accounts and their seed data
SEED_DATA = [
    {
        "email": "18314979086@163.com",
        "clones": [
            {
                "name": "测试音色-温柔女声",
                "description": "用于E2E测试的克隆音色（温柔女声）",
                "voice_id": "longxiaochun_v3",
                "gender": "female",
                "language": "zh-CN",
            },
            {
                "name": "测试音色-磁性男声",
                "description": "用于E2E测试的克隆音色（磁性男声）",
                "voice_id": "longsanshu_v3",
                "gender": "male",
                "language": "zh-CN",
            },
        ],
    },
]


def get_db_session():
    """Create a SQLAlchemy session from environment."""
    try:
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        database_url = os.environ.get("DATABASE_URL", "")
        if not database_url:
            # Try loading from shared config
            try:
                from packages.shared.config import get_shared_settings

                settings = get_shared_settings()
                database_url = settings.effective_database_url
            except Exception:
                logger.error("DATABASE_URL not set and could not load from config")
                sys.exit(1)

        engine = create_engine(database_url)
        Session = sessionmaker(bind=engine)
        return Session()
    except ImportError:
        logger.error("sqlalchemy not installed")
        sys.exit(1)


def find_user(session, email: str) -> dict | None:
    """Find user by email."""
    from sqlalchemy import text

    result = session.execute(text("SELECT id, email, display_name FROM users WHERE email = :email"), {"email": email})
    row = result.fetchone()
    if row:
        return {"id": row[0], "email": row[1], "display_name": row[2]}
    return None


def clone_exists(session, user_id: str, name: str) -> bool:
    """Check if a clone profile with this name already exists for the user."""
    from sqlalchemy import text

    result = session.execute(
        text("SELECT COUNT(*) FROM voice_clone_profiles WHERE user_id = :uid AND name = :name AND status != 'deleted'"),
        {"uid": user_id, "name": name},
    )
    return result.scalar() > 0


def create_clone_profile(session, user_id: str, clone_data: dict) -> str:
    """Create a VoiceCloneProfile record."""
    from sqlalchemy import text

    profile_id = uuid4().hex
    now = datetime.now(timezone.utc)

    session.execute(
        text("""
        INSERT INTO voice_clone_profiles
            (id, user_id, name, description, source_audio_url, voice_id,
             voice_model, language, gender, status, error_message,
             retry_count, max_retries, metadata, created_at, updated_at)
        VALUES
            (:id, :uid, :name, :desc, :audio_url, :voice_id,
             :voice_model, :lang, :gender, :status, '',
             0, 3, '{}', :now, :now)
        """),
        {
            "id": profile_id,
            "uid": user_id,
            "name": clone_data["name"],
            "desc": clone_data.get("description", ""),
            "audio_url": "",
            "voice_id": clone_data["voice_id"],
            "voice_model": "cosyvoice-v3-flash",
            "lang": clone_data.get("language", "zh-CN"),
            "gender": clone_data.get("gender", "unknown"),
            "status": "ready",
            "now": now,
        },
    )
    return profile_id


def main():
    logger.info("=== Seed Test Voice Clones ===")

    session = get_db_session()

    total_created = 0
    total_skipped = 0

    for account in SEED_DATA:
        email = account["email"]
        user = find_user(session, email)

        if not user:
            logger.warning("User not found: %s — skipping", email)
            continue

        logger.info("Found user: %s (id=%s, name=%s)", email, user["id"], user["display_name"])

        for clone_data in account.get("clones", []):
            name = clone_data["name"]

            if clone_exists(session, user["id"], name):
                logger.info("  Clone '%s' already exists — skipping", name)
                total_skipped += 1
                continue

            try:
                profile_id = create_clone_profile(session, user["id"], clone_data)
                logger.info("  Created clone '%s' (id=%s, voice_id=%s)", name, profile_id, clone_data["voice_id"])
                total_created += 1
            except Exception as e:
                logger.error("  Failed to create clone '%s': %s", name, e)
                session.rollback()
                continue

    try:
        session.commit()
        logger.info("=== Done: %d created, %d skipped ===", total_created, total_skipped)
    except Exception as e:
        logger.error("Failed to commit: %s", e)
        session.rollback()
        sys.exit(1)
    finally:
        session.close()


if __name__ == "__main__":
    main()