Ejercicio: Síntesis de Voz con Amazon Polly ​

# config.py
POLLY_CONFIG = {
    'region_name': 'us-east-1',
    'output_format': 'mp3',
    'sample_rate': '22050',
    'voice_id': 'Mia',  # Voz en español
    'language_code': 'es-MX'
}

S3_CONFIG = {
    'bucket_name': 'my-polly-audio-[tu-sufijo-único]',
    'input_prefix': 'input/',
    'output_prefix': 'output/'
}

VOICES = {
    'es-MX': ['Mia'],
    'es-ES': ['Lucia', 'Enrique'],
    'en-US': ['Joanna', 'Matthew'],
    'fr-FR': ['Celine', 'Mathieu']
}

# speech_synthesizer.py
import boto3
import os
import time
from contextlib import closing

class SpeechSynthesizer:
    def __init__(self, region_name='us-east-1'):
        self.polly = boto3.client('polly', region_name=region_name)
        self.s3 = boto3.client('s3')
        
    def synthesize_speech(self, text, voice_id='Mia', output_format='mp3'):
        """Sintetiza texto a voz"""
        response = self.polly.synthesize_speech(
            Text=text,
            OutputFormat=output_format,
            VoiceId=voice_id,
            Engine='neural'  # Usar motor neural para mejor calidad
        )
        
        # Guardar el audio
        if "AudioStream" in response:
            with closing(response["AudioStream"]) as stream:
                return stream.read()
                
    def synthesize_ssml(self, ssml_text, voice_id='Mia', output_format='mp3'):
        """Sintetiza texto SSML a voz"""
        response = self.polly.synthesize_speech(
            Text=ssml_text,
            TextType='ssml',
            OutputFormat=output_format,
            VoiceId=voice_id,
            Engine='neural'
        )
        
        if "AudioStream" in response:
            with closing(response["AudioStream"]) as stream:
                return stream.read()
                
    def start_task(self, text, voice_id='Mia', output_format='mp3'):
        """Inicia una tarea asíncrona para textos largos"""
        response = self.polly.start_speech_synthesis_task(
            Text=text,
            OutputFormat=output_format,
            OutputS3BucketName=S3_CONFIG['bucket_name'],
            OutputS3KeyPrefix=S3_CONFIG['output_prefix'],
            VoiceId=voice_id,
            Engine='neural'
        )
        return response['SynthesisTask']['TaskId']
        
    def get_task_status(self, task_id):
        """Obtiene el estado de una tarea asíncrona"""
        response = self.polly.get_speech_synthesis_task(TaskId=task_id)
        return response['SynthesisTask']['TaskStatus']
        
    def list_voices(self, language_code=None):
        """Lista las voces disponibles"""
        if language_code:
            response = self.polly.describe_voices(LanguageCode=language_code)
        else:
            response = self.polly.describe_voices()
        return response['Voices']

# utils.py
import os
import json
import time

def save_audio(audio_data, filename, output_dir='audio/mp3'):
    """Guarda el audio en un archivo"""
    os.makedirs(output_dir, exist_ok=True)
    output_path = os.path.join(output_dir, filename)
    
    with open(output_path, 'wb') as file:
        file.write(audio_data)
    return output_path

def process_long_text(text, max_chars=3000):
    """Divide texto largo en segmentos procesables"""
    segments = []
    while text:
        if len(text) <= max_chars:
            segments.append(text)
            break
            
        # Buscar el último punto dentro del límite
        split_point = text.rfind('.', 0, max_chars)
        if split_point == -1:
            split_point = max_chars
            
        segments.append(text[:split_point + 1])
        text = text[split_point + 1:].strip()
    
    return segments

def create_ssml(text, voice_id='Mia', rate='medium', pitch='medium'):
    """Crea texto SSML formateado"""
    ssml = f"""
    <speak>
        <prosody rate="{rate}" pitch="{pitch}">
            {text}
        </prosody>
    </speak>
    """
    return ssml.strip()

def wait_for_task(synthesizer, task_id, check_interval=30):
    """Espera a que una tarea asíncrona se complete"""
    while True:
        status = synthesizer.get_task_status(task_id)
        if status == 'completed':
            return True
        elif status in ['failed', 'error']:
            return False
            
        time.sleep(check_interval)

# main.py
from speech_synthesizer import SpeechSynthesizer
from utils import save_audio, process_long_text, create_ssml, wait_for_task
import config

def main():
    # Inicializar sintetizador
    synthesizer = SpeechSynthesizer(
        region_name=config.POLLY_CONFIG['region_name']
    )
    
    # Ejemplo 1: Síntesis simple
    text = "Hola, esto es una prueba de Amazon Polly."
    audio_data = synthesizer.synthesize_speech(
        text,
        voice_id=config.POLLY_CONFIG['voice_id']
    )
    save_audio(audio_data, 'test1.mp3')
    
    # Ejemplo 2: Síntesis con SSML
    ssml_text = create_ssml(
        text,
        rate='slow',
        pitch='high'
    )
    audio_data = synthesizer.synthesize_ssml(
        ssml_text,
        voice_id=config.POLLY_CONFIG['voice_id']
    )
    save_audio(audio_data, 'test2.mp3')
    
    # Ejemplo 3: Procesar texto largo
    with open('texts/input/long_text.txt', 'r', encoding='utf-8') as f:
        long_text = f.read()
        
    task_id = synthesizer.start_task(
        long_text,
        voice_id=config.POLLY_CONFIG['voice_id']
    )
    
    if wait_for_task(synthesizer, task_id):
        print(f"Texto largo procesado. Verificar en S3: {config.S3_CONFIG['bucket_name']}")

if __name__ == '__main__':
    main()

python main.py