如何基于Asterisk ARI实现Python批量外呼并播放音频后挂断

问题描述 #

我用Asterisk ARI编写了一个Python 3脚本，目前仅支持单次外呼：被叫接通后播放音频随即挂断。想修改脚本实现对多个号码的外呼操作，已尝试修改ARIApp类，但效果不符合预期。原脚本和修改后的ARIApp类如下：

原脚本 #

#!/usr/bin/env python3

import json
import sys
import websocket
import requests
import time


class ARIInterface(object):
    def __init__(self, server_addr, username, password):
        self._req_base = "http://%s:8088/ari/" % server_addr
        self._username = username
        self._password = password

    def answer_call(self, channel_id):
        req_str = self._req_base + "channels/%s/answer" % channel_id
        self._send_post_request(req_str)

    def play_sound(self, channel_id, sound_name):
        req_str = self._req_base + ("channels/%s/play?media=sound:%s" % (channel_id, sound_name))
        self._send_post_request(req_str)

    def cr_channel(self, number):
        req_str = self._req_base + ("channels/create?endpoint=SIP/25002/%s&app=hello" % (number))
        self._send_post_request(req_str)

    def dial_channel(self, channel_id):
        req_str = self._req_base + ("channels/%s/dial?timeout=30" % (channel_id))
        self._send_post_request(req_str)

    def del_channel(self, channel_id):
        req_str = self._req_base + ("channels/%s" % (channel_id))
        self._send_delete_request(req_str)

    def _send_post_request(self, req_str):
        r = requests.post(req_str, auth=(self._username, self._password))

    def _send_delete_request(self, req_str):
        r = requests.delete(req_str, auth=(self._username, self._password))


class ARIApp(object):
    def __init__(self, server_addr):
        app_name = 'hello'
        username = 'ari-user'
        password = 'ari-ewsd-135'
        url = "ws://%s:8088/ari/events?app=%s&api_key=%s:%s" % (server_addr, app_name, username, password)

        ari = ARIInterface(server_addr, username, password)
        ws = websocket.create_connection(url)

        numbers = [21018, 35000]

        ari.cr_channel('21018')

        channel_id = json.loads(ws.recv())['channel']['id']

        ari.dial_channel(channel_id)

        try:
            for event_str in iter(lambda: ws.recv(), None):
                event_json = json.loads(event_str)
                json.dump(event_json, sys.stdout, indent=2, sort_keys=True,
                          separators=(',', ': '))
                print("\n\nWebsocket event***************************************************\n")

                try:
                    if event_json['dialstatus'] == 'ANSWER':
                        peer_id = event_json['peer']['id']
                        ari.play_sound(peer_id, 'hello-world')
                        time.sleep(2)
                        ari.del_channel(peer_id)
                except:
                    pass

        except websocket.WebSocketConnectionClosedException:
            print("Websocket connection closed")
        except KeyboardInterrupt:
            print("Keyboard interrupt")
        finally:
            if ws:
                ws.close()


if __name__ == "__main__":
    app = ARIApp('localhost')

修改后的ARIApp类 #

class ARIApp(object):
    def __init__(self, server_addr):
        app_name = 'hello'
        username = 'ari-user'
        password = 'ari-ewsd-135'
        url = "ws://%s:8088/ari/events?app=%s&api_key=%s:%s" % (server_addr, app_name, username, password)

        ari = ARIInterface(server_addr, username, password)
        ws = websocket.create_connection(url)

        numbers = [21018, 35000]
        
        for number in numbers:          

            ari.cr_channel(number)

            channel_id = json.loads(ws.recv())['channel']['id']

            ari.dial_channel(channel_id)

        try:
            for event_str in iter(lambda: ws.recv(), None):
                event_json = json.loads(event_str)
                json.dump(event_json, sys.stdout, indent=2, sort_keys=True,
                          separators=(',', ': '))
                print("\n\nWebsocket event***************************************************\n")

                try:
                    if event_json['dialstatus'] == 'ANSWER':
                        peer_id = event_json['peer']['id']
                        ari.play_sound(peer_id, 'hello-world')
                        time.sleep(2)
                        ari.del_channel(peer_id)
                except:
                    pass

        except websocket.WebSocketConnectionClosedException:
            print("Websocket connection closed")
        except KeyboardInterrupt:
            print("Keyboard interrupt")
        finally:
            if ws:
                ws.close()

问题分析 #

当前修改存在两个核心问题：

• 事件处理混乱：循环创建所有通道后再统一处理事件，多个通道的事件会混杂在一起，ws.recv()在创建通道的循环中只能拿到当前通道的事件，后续其他通道的事件会积压，容易引发逻辑错误。
• 阻塞事件循环：time.sleep(2)会阻塞整个脚本的事件处理流程，期间无法响应其他通道的事件（比如另一个号码接通的通知），导致多号码外呼的并发处理失效。

解决方案 #

要实现可靠的多号码外呼，需要跟踪每个通道的状态，采用事件驱动的方式处理流程，避免阻塞。具体修改如下：

修改后的完整ARIApp类 #

class ARIApp(object):
    def __init__(self, server_addr):
        app_name = 'hello'
        username = 'ari-user'
        password = 'ari-ewsd-135'
        url = "ws://%s:8088/ari/events?app=%s&api_key=%s:%s" % (server_addr, app_name, username, password)

        self.ari = ARIInterface(server_addr, username, password)
        self.ws = websocket.create_connection(url)
        # 跟踪通道与号码的对应关系：key为通道ID，value为目标号码
        self.channel_map = {}
        # 记录已完成的外呼任务数
        self.completed_tasks = 0
        self.target_numbers = [21018, 35000]

        # 批量创建通道并触发拨号
        for number in self.target_numbers:
            self.ari.cr_channel(number)
            # 等待通道创建事件，记录通道与号码的关联
            event_str = self.ws.recv()
            event_json = json.loads(event_str)
            if event_json['type'] == 'ChannelCreated':
                channel_id = event_json['channel']['id']
                self.channel_map[channel_id] = number
                self.ari.dial_channel(channel_id)

        try:
            self.run_event_loop()
        except websocket.WebSocketConnectionClosedException:
            print("Websocket connection closed")
        except KeyboardInterrupt:
            print("Keyboard interrupt")
        finally:
            if self.ws:
                self.ws.close()

    def run_event_loop(self):
        for event_str in iter(lambda: self.ws.recv(), None):
            event_json = json.loads(event_str)
            json.dump(event_json, sys.stdout, indent=2, sort_keys=True, separators=(',', ': '))
            print("\n\nWebsocket event***************************************************\n")

            event_type = event_json.get('type')
            # 处理拨号接通事件
            if event_type == 'Dial' and event_json.get('dialstatus') == 'ANSWER':
                peer_id = event_json['peer']['id']
                self.ari.play_sound(peer_id, 'hello-world')
            # 处理音频播放完成事件，触发挂断
            elif event_type == 'PlaybackFinished':
                channel_id = event_json['playback']['target_uri'].split('/')[-1]
                self.ari.del_channel(channel_id)
                self.completed_tasks += 1
                # 所有外呼任务完成后退出循环
                if self.completed_tasks == len(self.target_numbers):
                    print("所有外呼任务执行完成")
                    break

关键改进点 #

• 状态跟踪：用channel_map字典关联通道ID和目标号码，确保每个事件能对应到正确的外呼任务。
• 非阻塞处理：通过监听PlaybackFinished事件触发挂断操作，替代time.sleep，保证事件循环始终能响应新的事件。
• 任务结束判断：记录已完成的任务数，所有号码处理完成后自动退出事件循环，避免脚本无意义持续运行。

内容的提问来源于stack exchange，提问作者royun