Skip to content

Instantly share code, notes, and snippets.

@AlonsoMackenlly
Last active December 2, 2020 13:21
Show Gist options
  • Save AlonsoMackenlly/b7106c3a4befa7148ea4fcb9fe5f6aaf to your computer and use it in GitHub Desktop.
Save AlonsoMackenlly/b7106c3a4befa7148ea4fcb9fe5f6aaf to your computer and use it in GitHub Desktop.
Оригинация звонка через Asterisk Freebpx ARI с помощью библиотеки ari-py (https://github.com/asterisk/ari-py). Важно! Для запуска кода необходимо иметь Python 2 версии!
# coding=utf-8
# Пример диалплана
# [custom-context]
# same => n,Answer()
# exten => _XXXX.,n,Stasis(ari_test)
# ;exten => _XXXX.,n,Dial(${TDIAL_STRING}/${OUTNUM}${TDIAL_SUFFIX},${TRUNK_RING_TIMER},${DIAL_TRUNK_OPTIONS})
# exten => _XXXX.,n,NoOp(Left Stasis)
# exten => _XXXX.,n,Hangup
import argparse
import ari
import gevent
from gevent.monkey import patch_all
patch_all()
from gevent.event import Event
import logging
from requests.exceptions import (
HTTPError,
ConnectionError
)
import socket
import time
logging.basicConfig() # Important!
# Otherwise you get No handlers could be found for
# logger "ari.client"
ARI_URL = 'http://<ip asterisk Freebpx>:8088/ari'
ARI_USER = '######### YOUR ARI USERNAME ########'
ARI_PASSWORD = '########### YOUR ARI USER PASSWORD ##########'
client = ari.connect(ARI_URL, ARI_USER, ARI_PASSWORD)
def run():
try:
client.run('ari_test')
except socket.error as e:
if e.errno == 32: # Broken pipe as we close the client.
pass
except ValueError as e:
if e.message == 'No JSON object could be decoded': # client.close()
pass
def originate(endpoint = None, callerid = None, context = None, extension = None,
priority = None, timeout = None):
# Go!
evt = Event() # Wait flag for origination
result = {}
gevent.sleep(0.1) # Hack to let run() arrange all.
start_time = time.time()
try:
channel = client.channels.originate(
endpoint = endpoint,
callerId = callerid,
app = 'ari_test',
timeout = timeout
)
def state_change(channel, event):
state = event['channel']['state']
if state == 'Up':
channel = channel.continueInDialplan(context = context, extension = extension, priority = priority)
def destroyed(channel, event):
end_time = time.time()
result['status'] = 'success'
result['message'] = '%s (%s)' % (
event.get('cause_txt'),
event.get('cause'))
result['duration'] = '%0.2f' % (end_time - start_time)
evt.set()
channel.on_event('ChannelDestroyed', destroyed)
channel.on_event('ChannelStateChange', state_change)
# Wait until we get origination result
evt.wait()
client.close()
return
except HTTPError as e:
result['status'] = 'error'
try:
error = e.response.json().get('error')
result['message'] = e.response.json().get('error')
except Exception:
result['message'] = e.response.content
finally:
print result
client.close()
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument('endpoint',
type = str,
help = 'Endpoint, e.g. SIP/operator/123456789',
default = "SIP/zadarma-577161")
parser.add_argument('callerid', type = str, help = 'CallerID, e.g. 111111', default = "13")
parser.add_argument('context', type = str, help = 'Asterisk context to connect call, e.g. default',
default = 'default')
parser.add_argument('extension', type = str, help = 'Context\'s extension, e.g. s', default = '13')
parser.add_argument('priority', type = str, help = 'Context\'s priority, e.g. 1', default = '1')
parser.add_argument('timeout', type = int, help = 'Originate timeout, e.g. 60', default = 60)
return parser.parse_args()
if __name__ == '__main__':
args = parse_args()
runner = gevent.spawn(run)
originator = gevent.spawn(originate, endpoint = args.endpoint, callerid = args.callerid,
context = args.context, extension = args.extension,
priority = args.priority, timeout = args.timeout
)
gevent.joinall([originator, runner])
@AlonsoMackenlly
Copy link
Author

Жаль, что библиотека перестала поддерживаться, и совместим только с Python 2. В роде как начали в 14 году его приспосабливать, но так и не закончили

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment