Примеры скриптов для мониторинга и управления вызовами

Мы уже изучили архитектуру AMI (Asterisk Manager Interface), научились подключаться к нему через telnet и разобрали структуру событий. Но ручной ввод команд годится только для отладки. Чтобы АТС работала автономно, нужны AMI-скрипт, автоматизация вызовов и мониторинг через AMI.

В современной разработке стандартом для таких скриптов стал Python с асинхронным подходом. Это позволяет скрипту обрабатывать поток событий в реальном времени, не «зависая» в ожидании ответов от сервера.

Настройка доступа

Для безопасности создайте отдельную учетную запись в /etc/asterisk/manager.conf. Не используйте логин администратора для скриптов.

; /etc/asterisk/manager.conf
[py_script_user]
secret = your_strong_password
deny = 0.0.0.0/0
permit = 127.0.0.1/32 ; Разрешаем только локальные подключения
read = system,call,log,verbose,agent,user,config,dtmf,reporting,cdr,dialplan
write = system,call,agent,user,config,originate

Автоматизация вызовов: команда Originate

Автоматизация вызовов — это программная генерация звонков. Типичный сценарий: система набирает номер менеджера и, как только он снимает трубку, соединяет его с клиентом.

Asterisk работает эффективнее, когда управление передается в конкретный контекст диалплана. Ниже — пример Click-to-Call скрипта на Python с библиотекой panoramisk.

import asyncio
from panoramisk import Manager

async def call_manager():
    manager = Manager(
        host='127.0.0.1',
        port=5038,
        username='py_script_user',
        secret='your_strong_password'
    )
    
    await manager.connect()

    # Формируем команду Originate
    action = {
        'Action': 'Originate',
        'Channel': 'PJSIP/101',
        'WaitTime': '30',
        'Context': 'from-internal',
        'Exten': '89991234567',
        'Priority': '1',
        'CallerID': 'Automation <100>',
        'Async': 'true' # Обязательно: не блокируем выполнение
    }

    response = await manager.send_action(action)
    print(f"Ответ Asterisk: {response}")
    
    await manager.close()

asyncio.run(call_manager())

Параметр 'Async': 'true' критичен. Если поставить false, Asterisk не ответит на запрос, пока вызов не завершится. Это заблокирует работу вашего приложения.

Мониторинг через AMI в реальном времени

Мониторинг через AMI позволяет реагировать на события мгновенно. Вам не нужно ждать записи в CDR или парсить лог-файлы — вы получаете данные о регистрациях транков или начале разговора в момент события.

На Схеме 1 показано, как скрипт фильтрует поток событий Asterisk.

Пример скрипта для отслеживания статуса SIP-транков. Это поможет сразу узнать, если связь с провайдером упала.

import asyncio
from panoramisk import Manager

async def monitor_registry():
    manager = Manager(host='127.0.0.1', username='py_script_user', secret='your_strong_password')
    await manager.connect()

    # Обработчик события Registry
    @manager.on_event('Registry')
    def handle_registry(manager, event):
        print(f"Статус: {event.ChannelType}/{event.Username} изменился на {event.Status}")

    while True:
        await asyncio.sleep(1)

asyncio.run(monitor_registry())

Типичные ошибки

При разработке автоматизации часто допускают три ошибки:

1. Блокирующие вызовы. Использование синхронных библиотек (вроде telnetlib). Если Asterisk задержит ответ, скрипт «замрет».
2. Отсутствие ActionID. Без уникального ID невозможно понять, к какой команде относится пришедший ответ в асинхронном потоке.
3. Утечка соединений. Открытие нового подключения на каждый чих. Это быстро исчерпает лимит maxsessions.

Практическое задание

Мы научились управлять Asterisk «снаружи». Это удобно для интеграции с CRM, но AMI бывает избыточен и сложен для кастомизации логики.

В следующей теме мы изучим ARI (Asterisk REST Interface) — современный способ управления ресурсами через REST API и WebSocket.