#!/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))