server/sync_tts_aligned.py

#!/usr/bin/env python3
"""
TTS Alignment Tool - Adjusts espeak rhythm to match ElevenLabs word-by-word
"""

import subprocess
import asyncio
import json
import os
import sys
import wave
import struct
import tempfile

WHISPER_CLI = "/workspace/MuseTalk1.5/vendor/whisper/build/bin/whisper-cli"
WHISPER_MODEL = "/workspace/MuseTalk1.5/vendor/whisper/models/ggml-large-v3-turbo-q5_0.bin"

def get_audio_duration(filepath):
    """Get audio duration in milliseconds"""
    if filepath.endswith('.wav'):
        with wave.open(filepath, 'r') as f:
            return int(f.getnframes() / f.getframerate() * 1000)
    else:
        result = subprocess.run([
            'ffprobe', '-v', 'error', '-show_entries', 'format=duration',
            '-of', 'default=noprint_wrappers=1:nokey=1', filepath
        ], capture_output=True, text=True)
        return int(float(result.stdout.strip()) * 1000)

def extract_word_timestamps(audio_path):
    """Extract word timestamps using local Whisper"""
    wav_path = audio_path
    if not audio_path.endswith('.wav'):
        wav_path = audio_path + '.wav'
        subprocess.run([
            'ffmpeg', '-y', '-v', 'quiet', '-i', audio_path, '-ar', '16000', '-ac', '1', wav_path
        ], capture_output=True)
    
    result = subprocess.run([
        WHISPER_CLI, '-m', WHISPER_MODEL, '-f', wav_path,
        '--output-json', '--max-len', '1', '-l', 'pt', '--no-prints'
    ], capture_output=True, text=True)
    
    json_path = wav_path + '.json'
    if os.path.exists(json_path):
        with open(json_path) as f:
            data = json.load(f)
        os.remove(json_path)
        
        words = []
        for item in data.get('transcription', []):
            text = item['text'].strip()
            if text:
                words.append({
                    'text': text,
                    'start': item['offsets']['from'],
                    'end': item['offsets']['to']
                })
        return words
    return []

def extract_audio_segment(input_path, output_path, start_ms, end_ms):
    """Extract audio segment using sox"""
    start_sec = start_ms / 1000
    duration_sec = (end_ms - start_ms) / 1000
    subprocess.run([
        'sox', input_path, output_path,
        'trim', str(start_sec), str(duration_sec)
    ], capture_output=True)

def time_stretch_audio(input_path, output_path, ratio):
    """Time stretch audio using rubberband (ratio > 1 = slower, < 1 = faster)"""
    subprocess.run([
        'rubberband', '-t', str(ratio), input_path, output_path
    ], capture_output=True)

def generate_silence(output_path, duration_ms, sample_rate=16000):
    """Generate silence WAV file"""
    num_samples = int(sample_rate * duration_ms / 1000)
    with wave.open(output_path, 'w') as f:
        f.setnchannels(1)
        f.setsampwidth(2)
        f.setframerate(sample_rate)
        f.writeframes(b'\x00\x00' * num_samples)

def concatenate_audio(input_files, output_path):
    """Concatenate audio files using sox"""
    subprocess.run(['sox'] + input_files + [output_path], capture_output=True)

def generate_espeak(text, output_path, speed=175):
    """Generate espeak audio"""
    subprocess.run([
        "espeak-ng", "-v", "pt-br", "-s", str(speed), "-w", output_path, text
    ], capture_output=True)

async def generate_elevenlabs(text, output_path):
    """Generate ElevenLabs audio"""
    import httpx
    
    ELEVENLABS_API_KEY = "sk_857e9e6f2412ddf3ff5334b736e4b571641d26225c0d8d62"
    ELEVENLABS_VOICE_ID = "21m00Tcm4TlvDq8ikWAM"
    
    async with httpx.AsyncClient(timeout=60.0) as client:
        response = await client.post(
            f"https://api.elevenlabs.io/v1/text-to-speech/{ELEVENLABS_VOICE_ID}/stream",
            headers={"xi-api-key": ELEVENLABS_API_KEY, "Content-Type": "application/json"},
            json={"text": text, "model_id": "eleven_flash_v2_5", "voice_settings": {"stability": 0.5, "similarity_boost": 0.75}}
        )
        with open(output_path, "wb") as f:
            f.write(response.content)

async def align_espeak_to_elevenlabs(text, output_path):
    """
    Generate espeak audio aligned to ElevenLabs timing.
    Returns the aligned espeak audio path.
    """
    print(f"\n{'='*70}")
    print(f"ALINHAMENTO DE RITMO: \"{text[:50]}...\"")
    print('='*70)
    
    with tempfile.TemporaryDirectory() as tmpdir:
        eleven_mp3 = os.path.join(tmpdir, "eleven.mp3")
        eleven_wav = os.path.join(tmpdir, "eleven.wav")
        espeak_wav = os.path.join(tmpdir, "espeak.wav")
        
        # 1. Generate ElevenLabs
        print("\n1. Gerando ElevenLabs...")
        await generate_elevenlabs(text, eleven_mp3)
        subprocess.run(['ffmpeg', '-y', '-v', 'quiet', '-i', eleven_mp3, '-ar', '16000', '-ac', '1', eleven_wav], capture_output=True)
        
        # 2. Generate espeak
        print("2. Gerando espeak...")
        generate_espeak(text, espeak_wav)
        
        # 3. Extract word timestamps
        print("3. Extraindo timestamps...")
        eleven_words = extract_word_timestamps(eleven_wav)
        espeak_words = extract_word_timestamps(espeak_wav)
        
        if not eleven_words or not espeak_words:
            print("   ERRO: Não foi possível extrair timestamps")
            return None
        
        print(f"   ElevenLabs: {len(eleven_words)} palavras")
        print(f"   espeak: {len(espeak_words)} palavras")
        
        # 4. Align word by word
        print("\n4. Alinhando palavra por palavra...")
        segments = []
        
        # Match words by index (simple approach)
        min_words = min(len(eleven_words), len(espeak_words))
        
        for i in range(min_words):
            ew = eleven_words[i]
            sw = espeak_words[i]
            
            # Calculate target duration and stretch ratio
            target_dur = ew['end'] - ew['start']
            source_dur = sw['end'] - sw['start']
            
            if source_dur > 0:
                ratio = target_dur / source_dur
            else:
                ratio = 1.0
            
            # Clamp ratio to reasonable range
            ratio = max(0.5, min(2.0, ratio))
            
            # Extract segment
            segment_in = os.path.join(tmpdir, f"seg_{i}_in.wav")
            segment_out = os.path.join(tmpdir, f"seg_{i}_out.wav")
            
            extract_audio_segment(espeak_wav, segment_in, sw['start'], sw['end'])
            
            # Time stretch
            if abs(ratio - 1.0) > 0.05:  # Only stretch if > 5% difference
                time_stretch_audio(segment_in, segment_out, ratio)
            else:
                subprocess.run(['cp', segment_in, segment_out], capture_output=True)
            
            # Add silence gap if needed
            if i > 0:
                prev_ew = eleven_words[i-1]
                gap = ew['start'] - prev_ew['end']
                if gap > 10:  # Add silence for gaps > 10ms
                    silence_path = os.path.join(tmpdir, f"silence_{i}.wav")
                    generate_silence(silence_path, gap)
                    segments.append(silence_path)
            
            segments.append(segment_out)
            
            print(f"   {sw['text']:<12} {source_dur:>4}ms → {target_dur:>4}ms (x{ratio:.2f})")
        
        # 5. Concatenate
        print("\n5. Concatenando...")
        if segments:
            concatenate_audio(segments, output_path)
            
            # Compare final durations
            eleven_dur = get_audio_duration(eleven_wav)
            aligned_dur = get_audio_duration(output_path)
            
            print(f"\n   Duração ElevenLabs: {eleven_dur} ms")
            print(f"   Duração Alinhado:   {aligned_dur} ms")
            print(f"   Diferença:          {abs(eleven_dur - aligned_dur)} ms")
            
            return output_path
    
    return None

async def test_alignment(text):
    """Test alignment and compare timestamps"""
    aligned_path = "/tmp/espeak_aligned.wav"
    eleven_path = "/tmp/eleven_test.mp3"
    
    # Generate aligned espeak
    result = await align_espeak_to_elevenlabs(text, aligned_path)
    
    if result:
        print("\n" + "="*70)
        print("VERIFICAÇÃO FINAL")
        print("="*70)
        
        # Generate ElevenLabs for comparison
        await generate_elevenlabs(text, eleven_path)
        
        # Extract timestamps from both
        eleven_words = extract_word_timestamps(eleven_path)
        aligned_words = extract_word_timestamps(aligned_path)
        
        print(f"\n{'PALAVRA':<15} {'ELEVENLABS':<20} {'ALINHADO':<20} {'DIFF':<10}")
        print("-"*70)
        
        total_diff = 0
        count = 0
        
        for i, ew in enumerate(eleven_words):
            if i < len(aligned_words):
                aw = aligned_words[i]
                diff = abs(ew['start'] - aw['start'])
                total_diff += diff
                count += 1
                print(f"{ew['text']:<15} {ew['start']:>5}-{ew['end']:<5} ms    {aw['start']:>5}-{aw['end']:<5} ms    {diff:>5} ms")
        
        if count > 0:
            avg_diff = total_diff / count
            print("-"*70)
            print(f"\nMÉDIA DE DIFERENÇA: {avg_diff:.0f} ms")
            
            if avg_diff < 30:
                print("✓ EXCELENTE SINCRONIZAÇÃO!")
            elif avg_diff < 50:
                print("✓ MUITO BOA SINCRONIZAÇÃO")
            elif avg_diff < 100:
                print("✓ BOA SINCRONIZAÇÃO")
            else:
                print("⚠ SINCRONIZAÇÃO ACEITÁVEL")
        
        print(f"\nArquivo alinhado: {aligned_path}")

if __name__ == "__main__":
    text = sys.argv[1] if len(sys.argv) > 1 else "Olá, tudo bem? Como posso te ajudar hoje?"
    asyncio.run(test_alignment(text))