Voice Cloning Using API

# API configuration
API_KEY = "your_api_key_here"
BASE_URL = "https://mirako.co"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

def start_voice_cloning(audio_files, annotation_file, voice_name, clean_data=False, webhook_url=None):
    """Start voice cloning process"""

    # Prepare files for upload
    files = {}

    # Add audio files
    for i, audio_file in enumerate(audio_files):
        with open(audio_file, 'rb') as f:
            files[f'audio_samples'] = (audio_file.name, f.read(), 'audio/wav')

    # Add annotation file
    with open(annotation_file, 'rb') as f:
        files['annotation_list'] = (annotation_file, f.read(), 'text/plain')

    # Prepare form data
    data = {
        'name': voice_name,
        'clean_data': str(clean_data).lower()
    }

    # Add webhook if provided
    if webhook_url:
        data['webhook'] = webhook_url

    # Make request
    response = requests.post(
        f"{BASE_URL}/v1/voice/clone",
        headers=headers,
        files=files,
        data=data
    )

    if response.status_code == 200:
        result = response.json()
        task_id = result['data']['task_id']
        print(f"✅ Voice cloning started!")
        print(f"Task ID: {task_id}")
        return task_id
    else:
        print(f"❌ Error: {response.status_code}")
        print(response.text)
        return None

# Start cloning
task_id = start_voice_cloning(
    audio_files=audio_files,
    annotation_file="voice_annotations.txt",
    voice_name="My Custom Voice",
    # clean_data=True  # Optionally apply denoising
)

import time

def check_cloning_status(task_id):
    """Check voice cloning task status"""

    response = requests.get(
        f"{BASE_URL}/v1/voice/clone/{task_id}/status",
        headers=headers
    )

    if response.status_code == 200:
        result = response.json()['data']
        status = result['status']

        print(f"Status: {status}")

        if status == "COMPLETED":
            profile_id = result.get('profile_id')
            print(f"🎉 Voice cloning completed!")
            print(f"Profile ID: {profile_id}")
            return {"status": "completed", "profile_id": profile_id}

        elif status in ["IN_QUEUE", "IN_PROGRESS"]:
            print("⏳ Voice cloning in progress...")
            return {"status": "processing"}

        elif status in ["FAILED", "CANCELED", "TIMED_OUT"]:
            error = result.get('error', 'Unknown error')
            print(f"❌ Voice cloning failed: {error}")
            return {"status": "failed", "error": error}

        else:
            print(f"Unknown status: {status}")
            return {"status": "unknown"}
    else:
        print(f"Error checking status: {response.text}")
        return {"status": "error"}

def wait_for_cloning_completion(task_id, max_wait_time=1800):  # 30 minutes
    """Wait for voice cloning to complete"""

    start_time = time.time()

    while time.time() - start_time < max_wait_time:
        result = check_cloning_status(task_id)

        if result["status"] == "completed":
            return result["profile_id"]
        elif result["status"] == "failed":
            print(f"Cloning failed: {result.get('error', 'Unknown error')}")
            return None

        # Wait 30 seconds before next check
        time.sleep(30)

    print("⏰ Timeout: Voice cloning didn't complete within time limit")
    return None

# Wait for completion
if task_id:
    profile_id = wait_for_cloning_completion(task_id)

    if profile_id:
        print(f"✅ Voice profile ready! ID: {profile_id}")
    else:
        print("❌ Voice cloning failed or timed out")