¿Cómo organizar el marcaje automático de una cola mediante SIP y Python?

La nota analiza la arquitectura de marcación automática: cómo se organiza la canalización de procesamiento, cómo funciona la marcación a través de Asterisk AMI y cómo se cierra el ciclo desde la generación de voz hasta la fijación del resultado.

Arquitectura: canalización de colas

El sistema está construido como una canalización con etapas explícitas: generación de audio, llamada, almacenamiento, reconocimiento de la respuesta. Cada cliente las atraviesa secuencialmente, pero las etapas funcionan en paralelo entre sí: mientras una llamada está en curso, para otras ya se prepara el audio y se registran los resultados.

Para cada etapa — su propia cola y un pool de trabajadores:

self.queues  = { "generate_voice_message": DelayedQueue(), "create_call": DelayedQueue(), ... }
self.workers = { "generate_voice_message": 5, "create_call": 1, "recognition": 3, "store_data": 10 }
self.stages  = {
    CallStatus.CREATED:   'generate_voice_message',
    CallStatus.GENERATED: 'create_call',
    CallStatus.PENDING:   'recognition'
}

La transición entre etapas se gestiona mediante CallStatus: el estado de la llamada determina en qué cola entrará a continuación. DelayedQueue permite aplazar reintentos sin un planificador separado: la demora se pasa directamente en el momento de encolar la tarea.

Conexión a Asterisk AMI

El servicio levanta panoramisk.Manager y se suscribe a eventos del canal — Hangup y BridgeEnter. El procesamiento de llamadas no arranca mientras AMI no esté disponible: antes de iniciar los workers se ejecuta un ciclo que comprueba la disponibilidad mediante Ping.

Del texto a la llamada

La primera etapa es la generación de audio. El texto del mensaje (individual o plantilla desde la BD) se envía a Yandex Cloud SpeechKit; como resultado se obtiene un archivo con nombre uuid4().hex, que se guarda en el servidor en un directorio accesible por Asterisk. El nombre del archivo viaja con el objeto de la llamada y en el momento oportuno se pasa como variable al dialplan — Asterisk lo localizará y lo reproducirá al abonado.

Cuando el audio está listo, el worker create_call forma un Originate de AMI:

action = {
    'Action':   'Originate',
    'Channel':  f'PJSIP/11{cl.client.phone}@{trunk}',
    'Context':  'applications',
    'Exten':    2200101,
    'Priority': 1,
    'Async':    'true',
    'Variable': f'file_name={cl.generated},file_record={cl.recorded},lang={lang_value}'
}

A través de Variable en el dialplan se pasan tres cosas: qué archivo reproducir, dónde guardar la grabación de la conversación y qué rama de idioma del guion usar. Async: true significa que AMI no queda bloqueado esperando la respuesta del abonado — la orden se envía y devuelve respuesta de inmediato.

Espera de finalización e intentos repetidos

Tras un Originate exitoso el worker guarda el identificador del canal y queda en espera: se ejecuta en un bucle con asyncio.sleep(1) hasta que el manejador del evento Hangup vacíe ese identificador. Es una forma simple pero efectiva de sincronizar el flujo asíncrono de AMI con la lógica del worker.

En caso de Failure en la respuesta — la llamada se vuelve a colocar en la misma cola a través de DelayedQueue con la demora tomada de la configuración. El límite son 6 intentos, tras lo cual el estado se fija como FAILED.

Almacenamiento, reconocimiento y cierre

Tras cada cambio de estado, store_data hace un upsert en la BD y, según el valor del estado, decide qué sigue: estado final (SUCCESSFUL / FAILED) — actualizar la fila en Google Sheet; cualquier otro — colocar en la siguiente cola según la tabla self.stages.

La última etapa, recognition, lee el archivo WAV de la grabación y lo envía a SpeechKit STT. Si el archivo no se encuentra o es demasiado pequeño, se marca inmediatamente como FAILED. De lo contrario, según el resultado del reconocimiento se fija el estado final y se cierra el ciclo para ese cliente.

Cuándo es aplicable

El enfoque descrito encaja bien en cualquier tarea de notificaciones salientes masivas: recordatorios de citas, confirmaciones de pedidos, llamadas a deudores, envíos de mensajes de voz a un segmento. La condición clave es disponer de Asterisk con AMI configurado y un trunk SIP. Todo lo demás — Python, colas y TTS/STT en la nube — se adapta al stack concreto.

Un enfoque basado en una máquina de estados y colas con retraso demuestra ser fiable donde la resistencia a fallos es importante: si cae el TTS, la tarea esperará y se reintentará; si el abonado no contesta, la llamada se reintentará tras el intervalo correspondiente sin intervención manual. Toda la lógica de transiciones, reintentos y paralelismo se concentra en tres estructuras — self.stages, self.workers y DelayedQueue — y se lee fácilmente como un esquema único del sistema.