test stuff

app/Dockerfile

@@ -0,0 +1,12 @@
+FROM jfloff/alpine-python:latest-slim
+RUN /entrypoint.sh \
+-p aiohttp \
+-p aiodns \
+-p databases[mysql] \
+-p quart \
+-p quart-openapi \
+-p hypercorn \
+-p git+https://github.com/gawel/panoramisk.git@master \
+-b libffi-dev 
+
+ENTRYPOINT ["python", "/app/app.py"]

app0/Dockerfile

@@ -0,0 +1,12 @@
+FROM jfloff/alpine-python:latest-slim
+RUN /entrypoint.sh \
+-p aiohttp \
+-p aiodns \
+-p databases[mysql] \
+-p quart \
+-p quart-openapi \
+-p hypercorn \
+-p git+https://github.com/gawel/panoramisk.git@master \
+-b libffi-dev 
+
+ENTRYPOINT ["python", "/app/app.py"]

app0/app.py

@@ -0,0 +1,1335 @@
+#!/usr/bin/env python3
+import asyncio
+import logging
+import os
+import re
+import json
+from datetime import datetime as dt
+from datetime import timedelta as td
+from typing import Any, Optional
+from functools import wraps
+from secrets import compare_digest
+from databases import Database
+from quart import jsonify, request, render_template_string, abort, current_app
+from quart.json import JSONEncoder
+from quart_openapi import Pint, Resource
+from http import HTTPStatus
+from panoramisk import Manager, Message
+from utils import *
+from cel import *
+from logging.config import dictConfig
+from pprint import pformat
+from inspect import getmembers
+
+class ApiJsonEncoder(JSONEncoder):
+  def default(self, o):
+    if isinstance(o, dt):
+      return o.isoformat()
+    if isinstance(o, CdrChannel):
+      return str(o)
+    if isinstance(o, CdrEvent):
+      return o.__dict__
+    if isinstance(o, CdrEvents) or isinstance(o, CelEvents):
+      return o.all
+    if isinstance(o, CdrCall) or isinstance(o, CelCall):
+      return o.__dict__
+    return JSONEncoder.default(self, o)
+
+class PintDB:
+  def __init__(self, app: Optional[Pint] = None) -> None:
+    self.init_app(app)
+    self._db = Database(app.config["DB_URI"])
+
+  def init_app(self, app: Pint) -> None:
+    app.before_serving(self._before_serving)
+    app.after_serving(self._after_serving)
+
+  async def _before_serving(self) -> None:
+    await self._db.connect()
+
+  async def _after_serving(self) -> None:
+    await self._db.disconnect()
+
+  def __getattr__(self, name: str) -> Any:
+    return getattr(self._db, name)
+
+# One asyncio event loop is used for AMI communication and HTTP requests routing with Quart
+main_loop = asyncio.get_event_loop()
+
+app = Pint(__name__, title=os.getenv('APP_TITLE', 'PBX API'), no_openapi=True)
+app.json_encoder = ApiJsonEncoder
+app.config.update({
+  'TITLE':                  os.getenv('APP_TITLE', 'PBX API'),
+  'APPLICATION_ROOT':       os.getenv('APP_APPLICATION_ROOT', None),
+  'SCHEME':                 os.getenv('APP_SCHEME', 'http'),
+  'FQDN':                   os.getenv('APP_FQDN', '127.0.0.1'),
+  'PORT':               int(os.getenv('APP_API_PORT', 8000)),
+  'BODY_TIMEOUT':       int(os.getenv('APP_BODY_TIMEOUT', 60)),
+  'DEBUG':                  os.getenv('APP_DEBUG', 'False').lower() in TRUEs,
+  'MAX_CONTENT_LENGTH': int(os.getenv('APP_MAX_CONTENT_LENGTH', 16777216)),
+  'AMI_HOST':               os.getenv('APP_AMI_HOST', '127.0.0.1'),
+  'AMI_PORT':           int(os.getenv('APP_AMI_PORT', 5038)),
+  'AMI_USERNAME':           os.getenv('APP_AMI_USERNAME', 'app'),
+  'AMI_SECRET':             os.getenv('APP_AMI_SECRET', 'secret'),
+  'AMI_PING_DELAY':     int(os.getenv('APP_AMI_PING_DELAY', 10)),
+  'AMI_PING_INTERVAL':  int(os.getenv('APP_AMI_PING_INTERVAL', 10)),
+  'AMI_TIMEOUT':        int(os.getenv('APP_AMI_TIMEOUT', 5)),
+  'AUTH_HEADER':            os.getenv('APP_AUTH_HEADER', 'APP-auth-token'),
+  'AUTH_SECRET':            os.getenv('APP_AUTH_SECRET', '3bfbeaabf363dd64fe263bd36830a6b6'),
+  'SWAGGER_JS_URL':         os.getenv('APP_SWAGGER_JS_URL', SWAGGER_JS_URL),
+  'SWAGGER_CSS_URL':        os.getenv('APP_SWAGGER_CSS_URL', SWAGGER_CSS_URL),
+  'STATE_CALLBACK_URL':     os.getenv('APP_STATE_CALLBACK_URL', None),
+  'DB_URI':                 'mysql://{}:{}@{}:{}/{}'.format(os.getenv('MYSQL_USER', 'asterisk'),
+                                                            os.getenv('MYSQL_PASSWORD', 'secret'),
+                                                            os.getenv('MYSQL_SERVER', 'db'),
+                                                            os.getenv('APP_PORT_MYSQL', '3306'),
+                                                            os.getenv('FREEPBX_CDRDBNAME', None)),
+  'EXTRA_API_URL':          os.getenv('APP_EXTRA_API_URL', None)})
+
+app.cache = {'devices':{},
+             'usermap':{},
+             'devicemap':{},
+             'ustates':{},
+             'pstates':{},
+             'queues':{},
+             'calls':{},
+             'cel_queue_calls':{},
+             'cel_calls':{}}
+
+manager = Manager(
+  loop=main_loop,
+  host=app.config['AMI_HOST'],
+  port=app.config['AMI_PORT'],
+  username=app.config['AMI_USERNAME'],
+  secret=app.config['AMI_SECRET'],
+  ping_delay=app.config['AMI_PING_DELAY'],
+  ping_interval=app.config['AMI_PING_INTERVAL'],
+  reconnect_timeout=app.config['AMI_TIMEOUT'],
+)
+
+def authRequired(func):
+  @wraps(func)
+  async def authWrapper(*args, **kwargs):
+    request.user = None
+    request.device = None
+    request.admin = False
+    auth = request.authorization
+    headers = request.headers
+    if ((auth is not None) and
+        (auth.type == "basic") and
+        (auth.username in current_app.cache['devices']) and
+        (compare_digest(auth.password, current_app.cache['devices'][auth.username]))):
+      request.device = auth.username
+      if request.device in current_app.cache['usermap']:
+        request.user = current_app.cache['usermap'][request.device]
+      return await func(*args, **kwargs)
+    elif ((current_app.config['AUTH_HEADER'].lower() in headers) and
+          (headers[current_app.config['AUTH_HEADER'].lower()] == current_app.config['AUTH_SECRET'])):
+      request.admin = True
+      return await func(*args, **kwargs)
+    else:
+      abort(401)
+  return authWrapper
+
+db = PintDB(app)
+
+@manager.register_event('FullyBooted')
+@manager.register_event('Reload')
+async def reloadCallback(mngr: Manager, msg: Message):
+  await refreshDevicesCache()
+  await refreshStatesCache()
+  await refreshQueuesCache()
+  await rebindLostDevices()
+  # await db.execute(query='CREATE TABLE IF NOT EXISTS callback_urls (device VARCHAR(16) PRIMARY KEY, url VARCHAR(255))')
+
+@manager.register_event('ExtensionStatus')
+async def extensionStatusCallback(mngr: Manager, msg: Message):
+  user = msg.exten
+  state = msg.statustext.lower()
+  app.logger.warning('ExtensionStatus({}, {})'.format(user, state))
+  if user in app.cache['ustates']:
+    prevState = getUserStateCombined(user)
+    app.cache['ustates'][user] = state
+    combinedState = getUserStateCombined(user)
+    if combinedState != prevState:
+      await userStateChangeCallback(user, combinedState, prevState)
+
+@manager.register_event('PresenceStatus')
+async def presenceStatusCallback(mngr: Manager, msg: Message):
+  user = msg.exten #hint = msg.hint
+  state = msg.status.lower()
+  if user in app.cache['ustates']:
+    prevState = getUserStateCombined(user)
+    app.cache['pstates'][user] = state
+    combinedState = getUserStateCombined(user)
+    if combinedState != prevState:
+      await userStateChangeCallback(user, combinedState, prevState)
+
+@manager.register_event('Hangup')
+async def hangupCallback(mngr: Manager, msg: Message):
+  if msg.uniqueid in app.cache['calls']:
+    del app.cache['calls'][msg.uniqueid]
+
+@manager.register_event('Newchannel')
+async def newchannelCallback(mngr: Manager, msg: Message):
+  if (msg.channelstate == '4'):
+    did = None
+    cid = None
+    user = None
+    device = None
+    uid = None
+    if msg.context in ('from-pstn'):
+      app.cache['calls'][msg.uniqueid]=msg
+    elif ((msg.context in ('from-queue')) and
+          (msg.linkedid in app.cache['calls']) and
+          (msg.exten in app.cache['devicemap'])):
+      did = app.cache['calls'][msg.linkedid].exten
+      cid = app.cache['calls'][msg.linkedid].calleridnum
+      user = msg.exten
+      device = app.cache['devicemap'][user]
+      uid = msg.linkedid
+    elif ((msg.context in ('from-internal')) and
+          (msg.exten in app.cache['devicemap'])):
+      user = msg.exten
+      device = app.cache['devicemap'][user]
+      if msg.calleridnum in app.cache['usermap']:
+        cid = app.cache['usermap'][msg.calleridnum]
+      else:
+        cid = msg.calleridnum
+      uid = msg.uniqueid
+    if device is not None:
+      _cb = {'user': user,
+             'device': device,
+             'state': 'ringing',
+             'callerId': cid,
+             'did': did,
+             'callId': uid}
+      if ('WebCallId' in app.cache['calls'][msg.linkedid]):
+        _cb['WebCallId'] = app.cache['calls'][msg.linkedid]['WebCallId']
+      reply = await doCallback(device, _cb)
+
+@manager.register_event('CEL')
+async def celCallback(mngr: Manager, msg: Message):
+  app.logger.warning('CEL {}'.format(msg))
+  lid = msg.LinkedID
+  if ((msg.EventName == 'CHAN_START') and (lid == msg.UniqueID)):    #save first msg
+    app.cache['cel_calls'][lid] = msg
+    app.cache['cel_calls'][lid]['current_channels'] = {}
+    app.cache['cel_calls'][lid]['all_channels'] = {}
+  if (lid in app.cache['cel_calls']):
+    firstMessage = app.cache['cel_calls'][lid]
+    cid = firstMessage.CallerIDnum
+    if firstMessage.CallerIDnum in app.cache['usermap']:
+        cid = app.cache['usermap'][firstMessage.CallerIDnum]
+    uid = firstMessage.LinkedID
+    if ((msg.Application == 'Queue') and
+        (msg.EventName == 'APP_START') and 
+        (firstMessage.Context == 'from-internal')):
+      app.cache['cel_calls'][lid]['groupCall'] = True
+    if ((msg.Application == 'Queue') and
+        (msg.EventName == 'APP_END') and 
+        (firstMessage.Context == 'from-internal')):
+      app.cache['cel_calls'][lid]['groupCall'] = False
+    if (cid is not None) and (len(cid) < 7):                         #for local calls only
+      if msg.Context in ('from-queue'):
+        if ((msg.EventName == 'CHAN_START') or
+            ((msg.EventName == 'CHAN_END') and ('answered' not in firstMessage))):
+          old_count = len(app.cache['cel_calls'][lid]['current_channels'])
+          channel = msg.Channel.split(';')[0]
+          if msg.EventName == 'CHAN_START':                          #start dial
+            app.cache['cel_calls'][lid]['current_channels'][channel] = msg.Exten
+            app.cache['cel_calls'][lid]['all_channels'][channel] = msg.Exten
+          else:                                                      #end dial
+            app.cache['cel_calls'][uid]['current_channels'].pop(channel, False)
+          if old_count != len(app.cache['cel_calls'][lid]['current_channels']):
+            _cb = {'users': list(app.cache['cel_calls'][uid]['current_channels'].values()),
+                 'state': 'group_ringing',
+                 'callerId': cid,
+                 'callId': uid}
+            if ('WebCallId' in app.cache['cel_calls'][msg.linkedid]):
+              _cb['WebCallId'] = app.cache['cel_calls'][msg.linkedid]['WebCallId']
+            reply = await doCallback('groupRinging', _cb)
+      if ((msg.EventName == 'ANSWER') and
+          (msg.Application == 'AppDial') and
+          firstMessage.get('groupCall',False) and 
+          (lid in app.cache['cel_calls'])):
+        called = msg.Exten
+        app.cache['cel_calls'][lid]['answered'] = True
+        _cb = {'user': called,
+               'users': list(app.cache['cel_calls'][uid]['all_channels'].keys()),
+               'state': 'group_answer',
+               'callerId': cid,
+               'callId': uid}
+        if ('WebCallId' in app.cache['cel_calls'][msg.linkedid]):
+          _cb['WebCallId'] = app.cache['cel_calls'][msg.linkedid]['WebCallId']
+        reply = await doCallback('groupAnswered', _cb)
+    if ((msg.Application == 'Queue') and
+        (firstMessage.Context == 'from-pstn')):
+      if (msg.EventName == 'APP_START'):
+        app.cache['cel_queue_calls'][lid] = {'caller': msg.CallerIDnum, 'start': parseDatetime(msg.EventTime).isoformat()}
+        _cb = {'callid': lid, 
+            'caller': msg.CallerIDnum, 
+            'start': parseDatetime(msg.EventTime).isoformat(),
+            'callerfrom': firstMessage.Exten,
+            'queue': msg.Exten,
+            'agents': [q.user for q in app.cache['queues'][msg.Exten]]}
+        reply = await doCallback('queueEnter', _cb)
+      if (msg.EventName in ('APP_END', 'BRIDGE_ENTER')):
+        call = app.cache['cel_queue_calls'].pop(lid,False)
+        queue_changed = (call != None)
+        if queue_changed :
+          _cb = {'callid': lid, 
+              'queue': msg.Exten,
+              'agents': [q.user for q in app.cache['queues'][msg.Exten]]}
+          reply = await doCallback('queueLeave', _cb)
+    if (msg.EventName == 'LINKEDID_END'):
+      app.cache['cel_calls'].pop(lid, False)
+      app.cache['cel_queue_calls'].pop(lid, False)
+    if (msg.EventName == 'USER_DEFINED') and (msg.UserDefType == 'SETVARIABLE'):
+      varname, value = msg.AppData.split(',')[1].split('=')[0:2]
+      app.cache['cel_calls'][lid][varname]=value
+      if (lid in app.cache['calls']):
+        app.cache['calls'][lid][varname]=value
+
+async def getCDR(start=None,
+                 end=None,
+                 table='cdr',
+                 field='calldate',
+                 sort='calldate, SUBSTR(uniqueid,1,10), sequence'):
+  _cdr = {}
+  if end is None:
+    end = dt.now()
+  if start is None:
+    start=(end - td(hours=24))
+  async for row in db.iterate(query='''SELECT *
+                                       FROM {table}
+                                       WHERE linkedid
+                                       IN (SELECT DISTINCT(linkedid)
+                                       FROM {table}
+                                       WHERE {field}
+                                       BETWEEN :start AND :end)
+                                       ORDER BY {sort};'''.format(table=table,
+                                                                  field=field,
+                                                                  sort=sort),
+                              values={'start':start,
+                                      'end':end}):
+    if row['linkedid'] in _cdr:
+      _cdr[row['linkedid']].events.add(row)
+    else:
+      _cdr[row['linkedid']]=CdrCall(row)
+  cdr = []
+  for _id in sorted(_cdr.keys()):
+    cdr.append(_cdr[_id])
+  return cdr
+
+async def getUserCDR(user,
+                     start=None,
+                     end=None,
+                     direction=None,
+                     limit=None,
+                     offset=None,
+                     order='ASC'):
+  _q = f'''SELECT * FROM cdr AS c INNER JOIN (SELECT linkedid FROM cdr WHERE'''
+  if direction:
+    direction=direction.lower()
+  if direction in ('in', True, '1', 'incoming', 'inbound'):
+    direction = 'inbound'
+    _q += f''' dst="{user}"'''
+  elif direction in ('out', False, '0', 'outgoing', 'outbound'):
+    direction = 'outbound'
+    _q += f''' src="{user}"'''
+  else:
+    direction = None
+    _q += f''' (src="{user}" or dst="{user}")'''
+  if end is None:
+    end = dt.now()
+  if start is None:
+    start=(end - td(hours=24))
+  _q += f''' AND calldate BETWEEN "{start}" AND "{end}" GROUP BY linkedid'''
+  if None not in (limit, offset):
+    _q += f''' LIMIT {offset},{limit}'''
+  _q += f''') AS c2 ON c.linkedid = c2.linkedid;'''
+  app.logger.warning('SQL: {}'.format(_q))
+  _cdr = {}
+  async for row in db.iterate(query=_q):
+    if (row['disposition']=='FAILED' and row['lastapp']=='Queue'):
+        continue
+    if row['linkedid'] in _cdr:
+      _cdr[row['linkedid']].events.add(row)
+    else:
+      _cdr[row['linkedid']]=CdrUserCall(user, row)
+  cdr = []
+  for _id in sorted(_cdr.keys(), reverse = True if (order.lower() == 'desc') else False):
+    record = _cdr[_id].simple
+    if (direction is not None) and (record['src'] == record['dst']) and (record['direction'] != direction):
+      record['direction'] = direction
+    if record['file'] is not None:
+      record['file'] = '/static/records/{d.year}/{d.month:02}/{d.day:02}/{filename}'.format(d=record['start'],
+                                                                                            filename=record['file'])
+    cdr.append(record)
+  return cdr
+
+async def getCEL(start=None, end=None, table='cel', field='eventtime', sort='id'):
+  return await getCDR(start, end, table, field, sort)
+
+async def doCallback(entity, msg):
+  row = await db.fetch_one(query='SELECT url FROM callback_urls WHERE device = :device', values={'device': entity})
+  if (row is not None) and (row['url'].startswith('http')):
+    app.logger.warning(f'''POST {row['url']} data: {str(msg)}''')
+  else:
+    app.logger.warning('No callback url defined for {}'.format(entity))
+  return None
+
+
+@app.route('/openapi.json')
+async def openapi():
+  '''Generates JSON that conforms OpenAPI Specification
+  '''
+  schema = app.__schema__
+  schema['servers'] = [{'url':'http://aster.rrt.ru:8000'},
+                       {'url':'{}://{}:{}'.format(app.config['SCHEME'],
+                                                  app.config['FQDN'],
+                                                  app.config['PORT'])}]
+  if app.config['EXTRA_API_URL'] is not None:
+    schema['servers'].append({'url':app.config['EXTRA_API_URL']})
+  schema['components'] = {'securitySchemes':{'ApiKey':{'type': 'apiKey',
+                                                       'name': app.config['AUTH_HEADER'],
+                                                       'in': 'header'}}}
+  schema['security'] = [{'ApiKey':[]}]
+  return jsonify(schema)
+
+@app.route('/ui')
+async def ui():
+  '''Swagger UI
+  '''
+  return await render_template_string(SWAGGER_TEMPLATE,
+                                      title=app.config['TITLE'],
+                                      js_url=app.config['SWAGGER_JS_URL'],
+                                      css_url=app.config['SWAGGER_CSS_URL'])
+
+async def action():
+  _payload = await request.get_data()
+  reply = await manager.send_action(json.loads(_payload))
+  return str(reply)
+
+async def amiGetVar(variable):
+  '''AMI GetVar
+  Returns value of requested variable using AMI action GetVar in background.
+
+  Parameters:
+    variable (string): Variable to query for
+
+  Returns:
+    string: Variable value or empty string if variable not found
+  '''
+  reply = await manager.send_action({'Action': 'GetVar',
+                                     'Variable': variable})
+  app.logger.warning('GetVar({})->{}'.format(variable, reply.value))
+  return reply.value
+
+@app.route('/ami/auths')
+@authRequired
+async def amiPJSIPShowAuths():
+  if not request.admin:
+    abort(401)
+  return successReply(app.cache['devices'])
+
+@app.route('/blackhole', methods=['GET','POST'])
+async def blackhole():
+  return ''
+
+@app.route('/ami/aors')
+@authRequired
+async def amiPJSIPShowAors():
+  if not request.admin:
+    abort(401)
+  aors = {}
+  reply = await manager.send_action({'Action':'PJSIPShowAors'})
+  if len(reply) >= 2:
+    for message in reply:
+      if ((message.event == 'AorList') and
+          ('objecttype' in message) and
+          (message.objecttype == 'aor') and
+          (int(message.maxcontacts) > 0)):
+        aors[message.objectname] = message.contacts
+    app.logger.warning('AorsList: {}'.format(','.join(aors.keys())))
+  return successReply(aors)
+
+async def amiUserEvent(name, data):
+  '''AMI UserEvent
+  Generates AMI Event using AMI action UserEvent with name and data supplied.
+
+  Parameters:
+    name (string): UserEvent name
+    data (dict): UserEvent data
+
+  Returns:
+    string: None if UserEvent was successfull, error message overwise
+  '''
+  reply = await manager.send_action({**{'Action': 'UserEvent',
+                                        'UserEvent': name},
+                                     **data})
+  app.logger.warning('UserEvent({})'.format(name))
+  if isinstance(reply, Message):
+    if reply.success:
+      return None
+    else:
+      return reply.message
+  return 'AMI error'
+
+async def amiSetVar(variable, value):
+  '''AMI SetVar
+  Sets variable using AMI action SetVar to value in background.
+
+  Parameters:
+    variable (string): Variable to set
+    value (string): Value to set for variable
+
+  Returns:
+    string: None if SetVar was successfull, error message overwise
+  '''
+  reply = await manager.send_action({'Action': 'SetVar',
+                                     'Variable': variable,
+                                     'Value': value})
+  app.logger.warning('SetVar({}, {})'.format(variable, value))
+  if isinstance(reply, Message):
+    if reply.success:
+      return None
+    else:
+      return reply.message
+  return 'AMI error'
+
+async def amiDBGet(family, key):
+  '''AMI DBGet
+  Returns value of requested astdb key using AMI action DBGet in background.
+
+  Parameters:
+    family (string): astdb key family to query for
+    key (string): astdb key to query for
+
+  Returns:
+    string: Value or empty string if variable not found
+  '''
+  reply = await manager.send_action({'Action': 'DBGet',
+                                     'Family': family,
+                                     'Key': key})
+  if (isinstance(reply, list) and
+      (len(reply) > 1)):
+    for message in reply:
+      if (message.event == 'DBGetResponse'):
+        app.logger.warning('DBGet(/{}/{})->{}'.format(family, key, message.val))
+        return message.val
+  app.logger.warning('DBGet(/{}/{})->Error!'.format(family, key))
+  return None
+
+async def amiDBPut(family, key, value):
+  '''AMI DBPut
+  Writes value to astdb by family and key using AMI action DBPut in background.
+
+  Parameters:
+    family (string): astdb key family to write to
+    key (string): astdb key to write to
+    value (string): value to write
+
+  Returns:
+    boolean: True if DBPut action was successfull, False overwise
+  '''
+  reply = await manager.send_action({'Action': 'DBPut',
+                                     'Family': family,
+                                     'Key': key,
+                                     'Val': value})
+  app.logger.warning('DBPut(/{}/{}, {})'.format(family, key, value))
+  if (isinstance(reply, Message) and reply.success):
+    return True
+  return False
+
+async def amiDBDel(family, key):
+  '''AMI DBDel
+  Deletes key from family in astdb using AMI action DBDel in background.
+
+  Parameters:
+    family (string): astdb key family
+    key (string): astdb key to delete
+
+  Returns:
+    boolean: True if DBDel action was successfull, False overwise
+  '''
+  reply = await manager.send_action({'Action': 'DBDel',
+                                     'Family': family,
+                                     'Key': key})
+  app.logger.warning('DBDel(/{}/{})'.format(family, key))
+  if (isinstance(reply, Message) and reply.success):
+    return True
+  return False
+
+async def amiSetHint(context, user, hint):
+  '''AMI SetHint
+  Sets hint for user in context using AMI action DialplanUserAdd with Replace=true in background.
+
+  Parameters:
+    context (string): dialplan context
+    user (string): user
+    hint (string): hint for user
+
+  Returns:
+    boolean: True if DialplanUserAdd action was successfull, False overwise
+  '''
+  reply = await manager.send_action({'Action': 'DialplanExtensionAdd',
+                                     'Context': context,
+                                     'Extension': user,
+                                     'Priority': 'hint',
+                                     'Application': hint,
+                                     'Replace': 'yes'})
+  app.logger.warning('SetHint({},{},{})'.format(context, user, hint))
+  if (isinstance(reply, Message) and reply.success):
+    return True
+  return False
+
+async def amiPresenceState(user):
+  '''AMI PresenceState request for CustomPresence provider
+
+  Parameters:
+    user (string): user
+
+  Returns:
+    boolean, string: True and state or False and error message
+  '''
+  reply = await manager.send_action({'Action': 'PresenceState',
+                                     'Provider': 'CustomPresence:{}'.format(user)})
+  app.logger.warning('PresenceState({})'.format(user))
+  if isinstance(reply, Message):
+    if reply.success:
+      return True, reply.state
+    else:
+      return False, reply.message
+  return False, 'AMI error'
+
+async def amiPresenceStateList():
+  states = {}
+  reply = await manager.send_action({'Action':'PresenceStateList'})
+  if len(reply) >= 2:
+    for message in reply:
+      if message.event == 'PresenceStateChange':
+        user = re.search('CustomPresence:(\d+)', message.presentity).group(1)
+        states[user] = message.status
+    app.logger.warning('PresenceStateList: {}'.format(','.join(states.keys())))
+  return states
+
+async def amiExtensionStateList():
+  states = {}
+  reply = await manager.send_action({'Action':'ExtensionStateList'})
+  if len(reply) >= 2:
+    for message in reply:
+      if ((message.event == 'ExtensionStatus') and
+          (message.context == 'ext-local')):
+        states[message.exten] = message.statustext.lower()
+    app.logger.warning('ExtensionStateList: {}'.format(','.join(states.keys())))
+  return states
+
+async def amiCommand(command):
+  '''AMI Command
+  Runs specified command using AMI action Command in background.
+
+  Parameters:
+    command (string): command to run
+
+  Returns:
+    boolean, list: tuple representing the boolean result of request and list of lines of command output
+  '''
+  reply = await manager.send_action({'Action': 'Command',
+                                     'Command': command})
+  result = []
+  if (isinstance(reply, Message) and reply.success):
+    if isinstance(reply.output, list):
+      result = reply.output
+    else:
+      result = reply.output.split('\n')
+    app.logger.warning('Command({})->{}'.format(command, '\n'.join(result)))
+    return True, result
+  app.logger.warning('Command({})->Error!'.format(command))
+  return False, result
+
+async def amiReload(module='core'):
+  '''AMI Reload
+  Reload specified asterisk module using AMI action reload in background.
+
+  Parameters:
+    module (string): module to reload, defaults to core
+
+  Returns:
+    boolean: True if Reload action was successfull, False overwise
+  '''
+  reply = await manager.send_action({'Action': 'Reload',
+                                     'Module': module})
+  app.logger.warning('Reload({})'.format(module))
+  if (isinstance(reply, Message) and reply.success):
+    return True
+  return False
+
+async def getGlobalVars():
+  globalVars = GlobalVars()
+  for _var in globalVars.d():
+    setattr(globalVars, _var, await amiGetVar(_var))
+  return globalVars
+
+async def setUserHint(user, dial, ast):
+  if dial in NONEs:
+    hint = 'CustomPresence:{}'.format(user)
+  else:
+    _dial= [dial]
+    if (ast.DNDDEVSTATE == 'TRUE'):
+      _dial.append('Custom:DND{}'.format(user))
+    hint = '{},CustomPresence:{}'.format('&'.join(_dial), user)
+  return await amiSetHint('ext-local', user, hint)
+
+async def amiQueues():
+  queues = {}
+  reply = await manager.send_action({'Action':'QueueStatus'})
+  if len(reply) >= 2:
+    for message in reply:
+      if message.event == 'QueueMember':
+        _qm = QueueMember(re.search('Local\/(\d+)', message.location).group(1))
+        queues.setdefault(message.queue, []).append(_qm.fromMessage(message))
+    app.logger.warning('QueuesList: {}'.format(','.join(queues.keys())))
+  return queues
+
+async def amiDeviceChannel(device):
+  reply = await manager.send_action({'Action':'CoreShowChannels'})
+  if len(reply) >= 2:
+    for message in reply:
+      if message.event == 'CoreShowChannel':
+        if message.calleridnum == device:
+          return message.channel
+  return None
+
+async def getUserChannel(user):
+  device = await getUserDevice(user)
+  if device in NONEs:
+    return False
+  channel = await amiDeviceChannel(device)
+  if channel in NONEs:
+    return False
+  return channel
+
+async def setQueueStates(user, device, state):
+  for queue in [_q for _q, _ma in app.cache['queues'].items() for _m in _ma if _m.user == user]:
+    await amiSetVar('DEVICE_STATE(Custom:QUEUE{}*{})'.format(device, queue), state)
+
+async def getDeviceUser(device):
+  return await amiDBGet('DEVICE', '{}/user'.format(device))
+
+async def getDeviceType(device):
+  return await amiDBGet('DEVICE', '{}/type'.format(device))
+
+async def getDeviceDial(device):
+  return await amiDBGet('DEVICE', '{}/dial'.format(device))
+
+async def getUserCID(user):
+  return await amiDBGet('AMPUSER', '{}/cidnum'.format(user))
+
+async def setDeviceUser(device, user):
+  return await amiDBPut('DEVICE', '{}/user'.format(device), user)
+
+async def getUserDevice(user):
+  return await amiDBGet('AMPUSER', '{}/device'.format(user))
+
+async def setUserDevice(user, device):
+  if device is None:
+    return await amiDBDel('AMPUSER', '{}/device'.format(user))
+  else:
+    return await amiDBPut('AMPUSER', '{}/device'.format(user), device)
+
+async def unbindOtherDevices(user, newDevice, ast):
+  '''Unbinds user from all devices except newDevice and sets
+  all required device states.
+  '''
+  devices = await amiDBGet('AMPUSER', '{}/device'.format(user))
+  if devices not in NONEs:
+    for _device in sorted(set(devices.split('&')), key=int):
+      if _device == user:
+        continue
+      if _device != newDevice:
+        if ast.FMDEVSTATE == 'TRUE':
+          await amiSetVar('DEVICE_STATE(Custom:FOLLOWME{})'.format(_device), 'INVALID')
+        if ast.QUEDEVSTATE == 'TRUE':
+          await setQueueStates(user, _device, 'NOT_INUSE')
+        if ast.DNDDEVSTATE:
+          await amiSetVar('DEVICE_STATE(Custom:DEVDND{})'.format(_device), 'NOT_INUSE')
+        if ast.CFDEVSTATE:
+          await amiSetVar('DEVICE_STATE(Custom:DEVCF{})'.format(_device), 'NOT_INUSE')
+        await amiDBPut('DEVICE', '{}/user'.format(_device), 'none')
+
+async def setUserDeviceStates(user, device, ast):
+  if ast.FMDEVSTATE == 'TRUE':
+    _followMe = await amiDBGet('AMPUSER', '{}/followme/ddial'.format(user))
+    if _followMe is not None:
+      await amiSetVar('DEVICE_STATE(Custom:FOLLOWME{})'.format(device), followMe2DevState(_followMe))
+  if ast.QUEDEVSTATE == 'TRUE':
+    await setQueueStates(user, device, 'INUSE')
+  if ast.DNDDEVSTATE:
+    _dnd = await amiDBGet('DND', user)
+    await amiSetVar('DEVICE_STATE(Custom:DEVDND{})'.format(device), 'INUSE' if _dnd == 'YES' else 'NOT_INUSE')
+  if ast.CFDEVSTATE:
+    _cf = await amiDBGet('CF', user)
+    await amiSetVar('DEVICE_STATE(Custom:DEVCF{})'.format(device), 'INUSE' if _cf != '' else 'NOT_INUSE')
+
+async def refreshStatesCache():
+  app.cache['ustates'] = await amiExtensionStateList()
+  app.cache['pstates'] = await amiPresenceStateList()
+  return len(app.cache['ustates'])
+
+async def refreshDevicesCache():
+  auths = {}
+  reply = await manager.send_action({'Action':'PJSIPShowAuths'})
+  if len(reply) >= 2:
+    for message in reply:
+      if ((message.event == 'AuthList') and
+          ('objecttype' in message) and
+          (message.objecttype == 'auth')):
+        auths[message.username] = message.password
+  app.cache['devices'] = auths
+  return len(app.cache['devices'])
+
+async def refreshQueuesCache():
+  app.cache['queues'] = await amiQueues()
+  return len(app.cache['queues'])
+
+async def rebindLostDevices():
+  app.cache['usermap'] = {}
+  app.cache['devicemap'] = {}
+  ast = await getGlobalVars()
+  for device in app.cache['devices']:
+    user = await getDeviceUser(device)
+    deviceType = await getDeviceType(device)
+    if (deviceType != 'fixed') and (user != 'none') and (user in app.cache['ustates'].keys()):
+      _device = await getUserDevice(user)
+      if _device != device:
+        app.logger.warning('Fixing bind user {} to device {}'.format(user, device))
+        dial = await getDeviceDial(device)
+        await setUserHint(user, dial, ast)           # Set hints for user on new device
+        await setUserDeviceStates(user, device, ast) # Set device states for users device
+        await setUserDevice(user, device)            # Bind device to user
+    app.cache['usermap'][device] = user
+    if user != 'none':
+      app.cache['devicemap'][user] = device
+
+async def userStateChangeCallback(user, state, prevState = None):
+  app.logger.warning('{} changed state to: {}'.format(user, state))
+  return ''
+
+def getUserStateCombined(user):
+  _uCache = app.cache['ustates']
+  _pCache = app.cache['pstates']
+  return combinedStates[_uCache.get(user, 'unavailable')][_pCache.get(user, 'not_set')]
+
+def getUsersStatesCombined():
+  return {user:getUserStateCombined(user) for user in app.cache['ustates']}
+
+@app.route('/atxfer/<userA>/<userB>')
+class AtXfer(Resource):
+  @authRequired
+  @app.param('userA', 'User initiating the attended transfer', 'path')
+  @app.param('userB', 'Transfer destination user', 'path')
+  @app.response(HTTPStatus.OK, 'Json reply')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, userA, userB):
+    '''Attended call transfer
+    '''
+    if (userA != request.user) and (not request.admin):
+      abort(401)
+    channel = await getUserChannel(userA)
+    if not channel:
+      return noUserChannel(userA)
+    reply = await manager.send_action({'Action':'Atxfer',
+                                       'Channel':channel,
+                                       'async':'false',
+                                       'Exten':userB})
+    if isinstance(reply, Message):
+      if reply.success:
+        return successfullyTransfered(userA, userB)
+      else:
+        return errorReply(reply.message)
+
+@app.route('/bxfer/<userA>/<userB>')
+class BXfer(Resource):
+  @authRequired
+  @app.param('userA', 'User initiating the blind transfer', 'path')
+  @app.param('userB', 'Transfer destination user', 'path')
+  @app.response(HTTPStatus.OK, 'Json reply')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, userA, userB):
+    '''Blind call transfer
+    '''
+    if (userA != request.user) and (not request.admin):
+      abort(401)
+    channel = await getUserChannel(userA)
+    if not channel:
+      return noUserChannel(userA)
+    reply = await manager.send_action({'Action':'BlindTransfer',
+                                       'Channel':channel,
+                                       'async':'false',
+                                       'Exten':userB})
+    if isinstance(reply, Message):
+      if reply.success:
+        return successfullyTransfered(userA, userB)
+      else:
+        return errorReply(reply.message)
+
+@app.route('/originate/<user>/<number>')
+class Originate(Resource):
+  @authRequired
+  @app.param('user', 'User initiating the call', 'path')
+  @app.param('number', 'Destination number', 'path')
+  @app.response(HTTPStatus.OK, 'Json reply')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, user, number):
+    '''Originate call
+    '''
+    if (user != request.user) and (not request.admin):
+      abort(401)
+    device = await getUserDevice(user)
+    if device in NONEs:
+      return noUserDevice(user)
+    device = device.replace('{}&'.format(user), '')
+    _act = { 'Action':'Originate',
+             'Channel':'PJSIP/{}'.format(device),
+             'Context':'from-internal',
+             'Exten':number,
+             'Priority': '1',
+             'async':'false',
+             'Callerid': '{} <{}>'.format(user, user)}
+    app.logger.warning(_act)
+    reply = await manager.send_action(_act)
+    if isinstance(reply, Message):
+      if reply.success:
+        return successfullyOriginated(user, number)
+      else:
+        return errorReply(reply.message)
+
+@app.route('/hangup/<user>')
+class Hangup(Resource):
+  @authRequired
+  @app.param('user', 'User to hangup', 'path')
+  @app.response(HTTPStatus.OK, 'Json reply')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, user):
+    '''Call hangup
+    '''
+    if (user != request.user) and (not request.admin):
+      abort(401)
+    channel = await getUserChannel(user)
+    if not channel:
+      return noUserChannel(user)
+    reply = await manager.send_action({'Action':'Hangup',
+                                       'Channel':channel})
+    if isinstance(reply, Message):
+      if reply.success:
+        return successfullyHungup(user)
+      else:
+        return errorReply(reply.message)
+
+@app.route('/users/states')
+class UsersStates(Resource):
+  @authRequired
+  @app.response(HTTPStatus.OK, 'JSON reply with user:state map or error message')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns all users with their combined states.
+    Possible states are: available, away, dnd, inuse, busy, unavailable, ringing
+    '''
+    if not request.admin:
+      abort(401)
+    #app.logger.warning('request device: {}'.format(request.device))
+    #usersCount = await refreshStatesCache()
+    #if usersCount == 0:
+    #  return stateCacheEmpty()
+
+    return successReply(getUsersStatesCombined())
+
+@app.route('/users/states/<users_list>')
+class UsersStatesSelected(Resource):
+  @authRequired
+  @app.param('users_list', 'Comma separated list of users to query for combined states', 'path')
+  @app.response(HTTPStatus.OK, 'JSON reply with user:state map or error message')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, users_list):
+    '''Returns selected users with their combined states.
+    Possible states are: available, away, dnd, inuse, busy, unavailable, ringing
+    '''
+    if not request.admin:
+      abort(401)
+    users = users_list.split(',')
+    states = getUsersStatesCombined()
+    result={}
+    for user in states:
+      if user in users:
+        result[user] = states[user]
+    return successReply(result)
+
+@app.route('/user/<user>/state')
+class UserState(Resource):
+  @authRequired
+  @app.param('user', 'User to query for combined state', 'path')
+  @app.response(HTTPStatus.OK, 'JSON data {"user":user,"state":state}')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, user):
+    '''Returns user's combined state.
+    One of: available, away, dnd, inuse, busy, unavailable, ringing
+    '''
+    if (user != request.user) and (not request.admin):
+      abort(401)
+    if user not in app.cache['ustates']:
+      return noUser(user)
+    return successReply({'user':user,'state':getUserStateCombined(user)})
+
+@app.route('/user/<user>/presence')
+class PresenceState(Resource):
+  @authRequired
+  @app.param('user', 'User to query for presence state', 'path')
+  @app.response(HTTPStatus.OK, 'JSON data {"user":user,"state":state}')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, user):
+    '''Returns user's presence state.
+    One of: not_set, unavailable, available, away, xa, chat, dnd
+    '''
+    if (user != request.user) and (not request.admin):
+      abort(401)
+    if user not in app.cache['ustates']:
+      return noUser(user)
+    return successReply({'user':user,'state':app.cache['pstates'].get(user, 'not_set')})
+
+@app.route('/user/<user>/presence/<state>')
+class SetPresenceState(Resource):
+  @authRequired
+  @app.param('user', 'Target user to set the presence state', 'path')
+  @app.param('state',
+             'The presence state for user, one of: not_set, unavailable, available, away, xa, chat or dnd',
+             'path')
+  @app.response(HTTPStatus.OK, 'Json reply')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, user, state):
+    '''Sets user's presence state.
+    Allowed states: not_set | unavailable | available | away | xa | chat | dnd
+    '''
+    if (user != request.user) and (not request.admin):
+      abort(401)
+    if state not in presenceStates:
+      return invalidState(state)
+    if user not in app.cache['ustates']:
+      return noUser(user)
+    # app.logger.warning('state={}, getUserStateCombined({})={}'.format(state, user, getUserStateCombined(user)))
+    if (state.lower() in ('available','away','not_set','xa','chat')) and (getUserStateCombined(user) in ('dnd')):
+      result = await amiDBDel('DND', '{}'.format(user))
+    result = await amiSetVar('PRESENCE_STATE(CustomPresence:{})'.format(user), state)
+    if result is not None:
+      return errorReply(result)
+    if state.lower() in ('dnd'):
+      result = await amiDBPut('DND', '{}'.format(user), 'YES')
+    return successfullySetState(user, state)
+
+@app.route('/users/devices')
+class UsersDevices(Resource):
+  @authRequired
+  @app.response(HTTPStatus.OK, 'JSON reply with user:device map or error message')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns users to device maping.
+    '''
+    if not request.admin:
+      abort(401)
+    data = {}
+    for user in app.cache['ustates']:
+      device = await getUserDevice(user)
+      if ((device in NONEs) or (device == user)):
+        device = None
+      else:
+        device = device.replace('{}&'.format(user), '')
+      data[user]= device
+    return successReply(data)
+
+@app.route('/device/<device>/<user>/on')
+@app.route('/user/<user>/<device>/on')
+class UserDeviceBind(Resource):
+  @authRequired
+  @app.param('device', 'Device number to bind to', 'path')
+  @app.param('user', 'User to bind to device', 'path')
+  @app.response(HTTPStatus.OK, 'JSON reply with fields "success" and "result"')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, device, user):
+    '''Binds user to device.
+    Both user and device numbers are checked for existance.
+    Any device user was previously bound to, is unbound.
+    Any user previously bound to device is unbound also.
+    '''
+    if (device != request.device) and (not request.admin):
+      abort(401)
+    if user not in app.cache['ustates']:
+      return noUser(user)
+    dial = await getDeviceDial(device) # Check if device exists in astdb
+    if dial is None:
+      return noDevice(device)
+    currentUser = await getDeviceUser(device) # Check if any user is already bound to device
+    if currentUser == user:
+      return alreadyBound(user, device)
+    ast = await getGlobalVars()
+    if currentUser not in NONEs: # If any other user is bound to device, unbind him,
+      result = await amiSetVar('PRESENCE_STATE(CustomPresence:{})'.format(user), 'available')
+      result = await amiDBDel('DND', '{}'.format(user))
+      await setUserDevice(currentUser, None)
+      if ast.QUEDEVSTATE == 'TRUE': # set device states for previous user queues
+        await setQueueStates(currentUser, device, 'NOT_INUSE')
+      await setUserHint(currentUser, None, ast) # set hints for previous user
+    await setDeviceUser(device, user) # Bind user to device
+    # If user is bound to some other devices, unbind him and set
+    # device states for those devices
+    await unbindOtherDevices(user, device, ast)
+    if not (await setUserHint(user, dial, ast)): # Set hints for user on new device
+      return hintError(user, device)
+    await setUserDeviceStates(user, device, ast) # Set device states for users new device
+    if not (await setUserDevice(user, device)): # Bind device to user
+      return bindError(user, device)
+    app.cache['usermap'][device] = user
+    app.cache['devicemap'][user] = device
+    await amiUserEvent('DeviceBound',{'device': device, 'newUser': user, 'oldUser': currentUser})
+    return successfullyBound(user, device)
+
+@app.route('/device/<device>/off')
+class DeviceUnBind(Resource):
+  @authRequired
+  @app.param('device', 'Device number to unbind', 'path')
+  @app.response(HTTPStatus.OK, 'JSON reply with fields "success" and "result"')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, device):
+    '''Unbinds any user from device.
+    Device is checked for existance.
+    '''
+    if (device != request.device) and (not request.admin):
+      abort(401)
+    dial = await getDeviceDial(device) # Check if device exists in astdb
+    if dial is None:
+      return noDevice(device)
+    currentUser = await getDeviceUser(device) # Check if any user is bound to device
+    if currentUser in NONEs:
+      return noUserBound(device)
+    else:
+      result = await amiSetVar('PRESENCE_STATE(CustomPresence:{})'.format(currentUser), 'available')
+      result = await amiDBDel('DND', '{}'.format(currentUser))
+      ast = await getGlobalVars()
+      await setUserDevice(currentUser, None) # Unbind device from current user
+      if ast.QUEDEVSTATE == 'TRUE': # set device states for current user queues
+        await setQueueStates(currentUser, device, 'NOT_INUSE')
+      await setUserHint(currentUser, None, ast) # set hints for current user
+    await setDeviceUser(device, 'none') # Unbind user from device
+    del app.cache['usermap'][device]
+    del app.cache['devicemap'][currentUser]
+    await amiUserEvent('DeviceUnbound',{'device': device, 'oldUser': currentUser})
+    return successfullyUnbound(currentUser, device)
+
+@app.route('/cdr')
+class CDR(Resource):
+  @authRequired
+  @app.param('end', 'End of datetime range. Defaults to now. Allowed formats are: timestamp, ISO 8601 or YYYYMMDDhhmmss', 'query')
+  @app.param('start', 'Start of datetime range. Defaults to end-24h. Allowed formats are: timestamp, ISO 8601 or YYYYMMDDhhmmss', 'query')
+  @app.response(HTTPStatus.OK, 'JSON reply')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns CDR data, groupped by logical call id.
+    All request arguments are optional.
+    '''
+    if not request.admin:
+      abort(401)
+    start = parseDatetime(request.args.get('start'))
+    end = parseDatetime(request.args.get('end'))
+    cdr = await getCDR(start, end)
+    return successReply(cdr)
+
+@app.route('/cel')
+class CEL(Resource):
+  @authRequired
+  @app.param('end', 'End of datetime range. Defaults to now. Allowed formats are: timestamp, ISO 8601 or YYYYMMDDhhmmss', 'query')
+  @app.param('start', 'Start of datetime range. Defaults to end-24h. Allowed formats are: timestamp, ISO 8601 or YYYYMMDDhhmmss', 'query')
+  @app.response(HTTPStatus.OK, 'JSON reply')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns CEL data, groupped by logical call id.
+    All request arguments are optional.
+    '''
+    if not request.admin:
+      abort(401)
+    start = parseDatetime(request.args.get('start'))
+    end = parseDatetime(request.args.get('end'))
+    cel = await getCEL(start, end)
+    return successReply(cel)
+
+@app.route('/calls')
+class Calls(Resource):
+  @authRequired
+  @app.param('end', 'End of datetime range. Defaults to now. Allowed formats are: timestamp, ISO 8601 and YYYYMMDDhhmmss', 'query')
+  @app.param('start', 'Start of datetime range. Defaults to end-24h. Allowed formats are: timestamp, ISO 8601 and YYYYMMDDhhmmss', 'query')
+  @app.response(HTTPStatus.OK, 'JSON reply')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns aggregated call data JSON. Draft implementation.
+    All request arguments are optional.
+    '''
+    if not request.admin:
+      abort(401)
+    calls = []
+    start = parseDatetime(request.args.get('start'))
+    end = parseDatetime(request.args.get('end'))
+    cdr = await getCDR(start, end)
+    for c in cdr:
+      _call = {'id':c.linkedid,
+               'start':c.start,
+               'type': c.direction,
+               'numberA': c.src,
+               'numberB': c.dst,
+               'line': c.did,
+               'duration': c.duration,
+               'waiting': c.waiting,
+               'status':c.disposition,
+               'url': c.file }
+      calls.append(_call)
+    return successReply(calls)
+
+@app.route('/user/<user>/calls')
+class UserCalls(Resource):
+  @authRequired
+  @app.param('user', 'User to query for call stats', 'path')
+  @app.param('end', 'End of datetime range. Defaults to now. Allowed formats are: timestamp, ISO 8601 and YYYYMMDDhhmmss', 'query')
+  @app.param('start', 'Start of datetime range. Defaults to end-24h. Allowed formats are: timestamp, ISO 8601 and YYYYMMDDhhmmss', 'query')
+  @app.param('direction', 'Calls direction, in or out. If not specified both are returned', 'query')
+  @app.param('limit', 'Max number of returned records, defaults to unlimited. Use offset parameter together with limit', 'query')
+  @app.param('offset', 'If limit is specified use offset parameter to request more results', 'query')
+  @app.param('order', 'Calls sort order for datetime field. ASC or DESC. Defaults to ASC', 'query')
+  @app.response(HTTPStatus.OK, 'JSON data {"status":status,"data":data,"message":message}')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, user):
+    '''Returns user's call stats.
+    '''
+    if (user != request.user) and (not request.admin):
+      abort(401)
+    if user not in app.cache['ustates']:
+      return noUser(user)
+    cdr = await getUserCDR(user,
+                           parseDatetime(request.args.get('start')),
+                           parseDatetime(request.args.get('end')),
+                           request.args.get('direction', None),
+                           request.args.get('limit', None),
+                           request.args.get('offset', None),
+                           request.args.get('order', 'ASC'))
+    return successReply(cdr)
+
+@app.route('/device/<device>/callback')
+class DeviceCallback(Resource):
+  @authRequired
+  @app.param('device', 'Device to get/set the callback url for', 'path')
+  @app.param('url', 'used to set the Callback url for the device', 'query')
+  @app.response(HTTPStatus.OK, 'JSON data {"user":user,"state":state}')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self, device):
+    '''Returns and sets device's callback url.
+    '''
+    if (device != request.device) and (not request.admin):
+      abort(401)
+    url = request.args.get('url', None)
+    if url is not None:
+      await db.execute(query='REPLACE INTO callback_urls (device, url) VALUES (:device, :url)',
+                       values={'device': device,'url': url})
+    else:
+      row = await db.fetch_one(query='SELECT url FROM callback_urls WHERE device = :device',
+                               values={'device': device})
+      if row is not None:
+        url = row['url']
+    return successCallbackURL(device, url)
+
+@app.route('/group/ringing/callback')
+class GroupRingingCallback(Resource):
+  @authRequired
+  @app.param('url', 'used to set the Callback url for the group ringing callback', 'query')
+  @app.response(HTTPStatus.OK, 'JSON data {"url":url}')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns and sets groupRinging callback url.
+    '''
+    if not request.admin:
+      abort(401)
+    url = request.args.get('url', None)
+    if url is not None:
+      await db.execute(query='REPLACE INTO callback_urls (device, url) VALUES (:device, :url)',
+                       values={'device': 'groupRinging','url': url})
+    else:
+      row = await db.fetch_one(query='SELECT url FROM callback_urls WHERE device = :device',
+                               values={'device': 'groupRinging'})
+      if row is not None:
+        url = row['url']
+    return successCommonCallbackURL('groupRinging', url)
+
+@app.route('/group/answered/callback')
+class GroupAnsweredCallback(Resource):
+  @authRequired
+  @app.param('url', 'used to set the Callback url for the group answered callback', 'query')
+  @app.response(HTTPStatus.OK, 'JSON data {"url":url}')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns and sets groupAnswered callback url.
+    '''
+    if not request.admin:
+      abort(401)
+    url = request.args.get('url', None)
+    if url is not None:
+      await db.execute(query='REPLACE INTO callback_urls (device, url) VALUES (:device, :url)',
+                       values={'device': 'groupAnswered','url': url})
+    else:
+      row = await db.fetch_one(query='SELECT url FROM callback_urls WHERE device = :device',
+                               values={'device': 'groupAnswered'})
+      if row is not None:
+        url = row['url']
+    return successCommonCallbackURL('groupAnswered', url)
+
+@app.route('/queue/enter/callback')
+class QueueEnterCallback(Resource):
+  @authRequired
+  @app.param('url', 'used to set the Callback url for the queue enter callback', 'query')
+  @app.response(HTTPStatus.OK, 'JSON data {"url":url}')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns and sets queueEnter callback url.
+    '''
+    if not request.admin:
+      abort(401)
+    url = request.args.get('url', None)
+    if url is not None:
+      await db.execute(query='REPLACE INTO callback_urls (device, url) VALUES (:device, :url)',
+                       values={'device': 'queueEnter','url': url})
+    else:
+      row = await db.fetch_one(query='SELECT url FROM callback_urls WHERE device = :device',
+                               values={'device': 'queueEnter'})
+      if row is not None:
+        url = row['url']
+    return successCommonCallbackURL('queueEnter', url)
+
+@app.route('/queue/leave/callback')
+class QueueLeaveCallback(Resource):
+  @authRequired
+  @app.param('url', 'used to set the Callback url for the queue leave callback', 'query')
+  @app.response(HTTPStatus.OK, 'JSON data {"url":url}')
+  @app.response(HTTPStatus.UNAUTHORIZED, 'Authorization required')
+  async def get(self):
+    '''Returns and sets queueLeave callback url.
+    '''
+    if not request.admin:
+      abort(401)
+    url = request.args.get('url', None)
+    if url is not None:
+      await db.execute(query='REPLACE INTO callback_urls (device, url) VALUES (:device, :url)',
+                       values={'device': 'queueLeave','url': url})
+    else:
+      row = await db.fetch_one(query='SELECT url FROM callback_urls WHERE device = :device',
+                               values={'device': 'queueLeave'})
+      if row is not None:
+        url = row['url']
+    return successCommonCallbackURL('queueLeave', url)
+
+manager.connect()
+app.run(loop=main_loop, host='0.0.0.0', port=app.config['PORT'])

app0/cdr_test

@@ -0,0 +1,2 @@
+#!/bin/bash
+docker exec -it app0 python /app/cdr_test.py $@

app0/cdr_test.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+import asyncio
+import aiohttp
+import logging
+import os
+import re
+import json
+import sys
+from datetime import datetime as dt
+from datetime import timedelta as td
+from typing import Any, Optional
+from functools import wraps
+from secrets import compare_digest
+from databases import Database
+from cel import *
+from logging.config import dictConfig
+from pprint import pformat
+from inspect import getmembers
+
+_db = Database('mysql://{}:{}@{}:{}/{}'.format(os.getenv('MYSQL_USER', 'asterisk'),
+                                               os.getenv('MYSQL_PASSWORD', 'secret'),
+                                               os.getenv('MYSQL_SERVER', 'db'),
+                                               os.getenv('APP_PORT_MYSQL', '3306'),
+                                               os.getenv('FREEPBX_CDRDBNAME', None)))
+linkedid = sys.argv[1]
+
+async def main():
+  await _db.connect()
+  _q = '''SELECT *
+          FROM cdr
+          WHERE linkedid=:linkedid
+          ORDER BY sequence;'''
+  _v = {'linkedid': linkedid}
+  _f = True
+  async for row in _db.iterate(query=_q, values=_v):
+    print('\t'.join([str(k) for k in row]))
+  await _db.disconnect()
+
+if __name__ ==  '__main__':
+  loop = asyncio.get_event_loop()
+  loop.run_until_complete(main())

app0/cel.py

@@ -0,0 +1,256 @@
+#!/usr/bin/env python3
+import json
+import re
+from datetime import datetime
+from datetime import timedelta as td
+
+fieldsFilter = ('id', 'linkedid', 'appdata', 'lastdata')
+
+class CdrChannel:
+  def __init__(self, chanData):
+    self._str = chanData
+    try:
+      (self.tech,
+       self.peer,
+       self.context,
+       self.id,
+       self.leg) = re.match(r'(\w+)/(\w+)(?:@([\w-]+))?-([0-9a-f]{8})(?:;([0-9]+))?', chanData).groups()
+    except:
+      pass
+  def __repr__(self):
+    return self._str
+
+class CdrEvent:
+  def __init__(self, eventData):
+    eventData = dict(eventData)
+    for key, value in eventData.items():
+      if ((key not in fieldsFilter) and (value is not None) and (str(value) != '')):
+        if key == 'extra':
+          setattr(self, key, json.loads(value))
+        elif key in ('channame','channel','dstchannel'):
+          setattr(self, key, CdrChannel(value))
+        else:
+          setattr(self, key, value)
+  def __getattr__(self, attr):
+    return None
+
+class CdrEvents:
+  def __init__(self):
+    self._events = []
+  def add(self, event):
+    if isinstance(event, CdrEvent):
+      self._events.append(event)
+    else:
+      self._events.append(CdrEvent(event))
+  @property
+  def all(self):
+    return self._events
+  @property
+  def first(self):
+    return self._events[0] if len(self._events) > 0 else None
+  @property
+  def second(self):
+    return self._events[1] if len(self._events) > 1 else None
+  @property
+  def last(self):
+    return self._events[-1] if len(self._events) > 0 else None
+  def filter(self, key, value, func='__eq__'):
+    result = CdrEvents()
+    for event in self._events:
+      if getattr(str(getattr(event, key)), func)(value):
+        result.add(event)
+    return result
+  def has(self, value, key='disposition'):
+    for event in self._events:
+      if getattr(event, key) == value:
+        return True
+    return False
+  def __len__(self):
+    return len(self._events)
+
+class CdrUserEvents(CdrEvents):
+  def __init__(self, user):
+    self._user = user
+    self._events = []
+    self.start = 0
+    self.end = 0
+  def add(self, event):
+    if not isinstance(event, CdrEvent):
+      event = CdrEvent(event)
+    if self._user in (event.src, event.dst, event.cnum):
+      if len(self._events) == 0:
+        self.start = event.calldate
+        self.end = event.calldate + td(seconds=event.duration)
+        self._events.append(event)
+      else:
+        if (event.calldate + td(seconds=event.duration)) > self.end:
+          self.end = event.calldate + td(seconds=event.duration)
+        if event.calldate < self.start:
+          self.start = event.calldate
+        lo = 0
+        hi = len(self._events)
+        while lo < hi:
+          mid = (lo+hi)//2
+          if event.calldate <= self._events[mid].calldate:
+            hi = mid
+          else:
+            lo = mid+1
+        self._events.insert(lo, event)
+       
+  def simple(self):
+    res = []
+    for event in self._events:
+        simple_event = {'start': event.calldate,
+            'answered': event.disposition=='ANSWERED',
+            'duration': event.duration,
+            #'waiting': event.waiting,
+            'application': event.lastapp,
+            'dst': event.dst,
+            'uniqueid': event.uniqueid,
+            'file': event.recordingfile}
+        res.append(simple_event)
+    return res
+
+class CelEvents(CdrEvents):
+  def has(self, value, key='eventtype'):
+    return super().has(value, key)
+
+class CdrCall:
+  def __init__(self, event=None):
+    self.events = CdrEvents()
+    if event is not None:
+      self.events.add(event)
+      self.linkedid = event['linkedid']
+    else:
+      self.linkedid = None
+  @property
+  def start(self):
+    return self.events.first.calldate
+  @property
+  def disposition(self):
+    return self.events.first.disposition
+  @property
+  def file(self):
+    return self.events.first.recordingfile
+  @property
+  def src(self):
+    return self.events.first.cnum
+  @property
+  def direction(self):
+    if self.events.first.dcontext == 'from-internal':
+      if self.events.first.outbound_cnum is not None:
+        return 'outbound'
+      else:
+        return 'local'
+    else:
+      return 'inbound'
+  @property
+  def dst(self):
+    # placeholder
+    # TODO: determine last dst based on disposition
+    if self.direction == 'outbound':
+      return self.events.first.dst
+    elif self.direction == 'local':
+      return self.events.first.dst
+    else:
+      if self.events.first.dstchannel is not None:
+        return self.events.first.dstchannel.peer
+      else:
+        return self.events.first.dst
+  @property
+  def did(self):
+    if self.direction == 'inbound':
+      return self.events.first.did
+    else:
+      return None
+  @property
+  def duration(self):
+    if not self.isAnswered:
+      return 0
+    else:
+      if len(self.events) > 1:
+        return self.events.second.billsec
+      else:
+        return self.events.first.billsec
+  @property
+  def waiting(self):
+    if not self.isAnswered:
+      return self.events.first.duration
+    else:
+      # TODO: exclude time on ivr
+      return self.events.first.duration - self.duration
+  @property
+  def isAnswered(self):
+    return self.disposition == 'ANSWERED';
+
+class CdrUserCall(CdrCall):
+  def __init__(self, user, event=None):
+    self.user = user
+    self.events = CdrUserEvents(user)
+    if event is not None:
+      self.events.add(event)
+      self.linkedid = event['linkedid']
+    else:
+      self.linkedid = None
+  @property
+  def file(self):
+    return self.events.first.recordingfile
+  @property
+  def src(self):
+    return self.events.first.cnum
+  @property
+  def direction(self):
+    if self.user in (self.events.first.src, self.events.first.cnum):
+      return 'outbound'
+    else:
+      return 'inbound'
+  @property
+  def dst(self):
+    return self.events.first.dst
+  @property
+  def did(self):
+    if self.direction == 'inbound':
+      return self.events.first.did
+    else:
+      return None
+  @property
+  def duration(self):
+    if not self.isAnswered:
+      return 0
+    else:
+     return self.events.first.billsec
+  @property
+  def waiting(self):
+    if not self.isAnswered:
+      return self.events.first.duration
+    else:
+      # TODO: exclude time on ivr
+      return self.events.first.duration - self.duration
+  @property
+  def isAnswered(self):
+    return self.disposition == 'ANSWERED';
+  @property
+  def simple(self):
+    return {'start': self.start,
+            'direction': self.direction,
+            'answered': self.isAnswered,
+            'duration': self.duration,
+            'waiting': self.waiting,
+            'src': self.src,
+            'dst': self.dst,
+            'did': self.did,
+            'linkedid': self.linkedid,
+            'file': self.file,
+            'events': self.events.simple()}
+
+class CelCall:
+  def __init__(self, event=None):
+    self.events = CelEvents()
+    if event is not None:
+      self.events.add(event)
+      self.linkedid = event['linkedid']
+    else:
+      self.linkedid = None
+  @property
+  def start(self):
+    return self.events.first.eventtime

app0/utils.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+from dataclasses import dataclass, asdict
+from panoramisk import Message
+from datetime import datetime as dt
+from datetime import timedelta as td
+
+TRUEs = ('true', '1', 'y', 'yes')
+NONEs = (None,'none','')
+# userstate presencestate maping:
+_ustates = ['idle',       'inuse', 'busy', 'unavailable', 'ringing', 'inuse&ringing','hold', 'inuse&hold'] #presence:
+_states = [['available',   'busy', 'busy', 'unavailable', 'ringing', 'busy',         'busy', 'busy'],      #not_set
+           ['unavailable', 'busy', 'busy', 'unavailable', 'ringing', 'busy',         'busy', 'busy'],      #unavailable
+           ['available',   'busy', 'busy', 'unavailable', 'ringing', 'busy',         'busy', 'busy'],      #available
+           ['away',        'busy', 'busy', 'unavailable', 'ringing', 'busy',         'busy', 'busy'],      #away
+           ['available',   'busy', 'busy', 'unavailable', 'ringing', 'busy',         'busy', 'busy'],      #xa
+           ['available',   'busy', 'busy', 'unavailable', 'ringing', 'busy',         'busy', 'busy'],      #chat
+           ['dnd',         'busy', 'busy', 'unavailable', 'ringing', 'busy',         'busy', 'busy']]      #dnd
+_pstates = ['not_set',
+            'unavailable',
+            'available',
+            'away',
+            'xa',
+            'chat',
+            'dnd']
+# combinedStates[userstate][presencestate]=combinedState
+combinedStates = {_u: {_p: _states[_pstates.index(_p)][_ustates.index(_u)] for _p in _pstates} for _u in _ustates}
+presenceStates = _pstates
+NO_AUTH_ROUTES = ('/ui','/openapi.json','/favicon.ico')
+SWAGGER_JS_URL = "https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/3.37.2/swagger-ui-bundle.js"
+SWAGGER_CSS_URL = "https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/3.37.2/swagger-ui.min.css"
+SWAGGER_TEMPLATE = '''
+  <head>
+    <link type="text/css" rel="stylesheet" href="{{ css_url }}">
+    <title>{{ title }}</title>
+  </head>
+  <body>
+    <div id="swagger-ui"></div>
+    <script src="{{ js_url }}"></script>
+    <script>
+      const ui = SwaggerUIBundle({
+        deepLinking: true,
+        dom_id: "#swagger-ui",
+        layout: "BaseLayout",
+        presets: [
+          SwaggerUIBundle.presets.apis,
+          SwaggerUIBundle.SwaggerUIStandalonePreset
+        ],
+        showExtensions: true,
+        showCommonExtensions: true,
+        url: "/openapi.json"
+      });
+    </script>
+  </body>'''
+
+def parseDatetime(dateString):
+  _dt = None
+  if dateString is not None:
+    try:
+      _dt = dt.strptime(dateString, '%Y%m%d%H%M%S')
+    except ValueError:
+      pass
+    if _dt is None:
+      try:
+        _dt = dt.fromtimestamp(int(dateString))
+      except ValueError:
+        pass
+    if _dt is None:
+      try:
+        _dt = dt.fromisoformat(dateString)
+      except ValueError:
+        pass
+  return _dt
+
+def followMe2DevState(followMeState):
+  if followMeState == 'DIRECT':
+    return 'INUSE'
+  if followMeState == 'EXTENSION':
+    return 'NOT_INUSE'
+  return 'INVALID'
+
+@dataclass
+class QueueMember:
+  user: str
+  name: str = ''
+  location: str = ''
+  membership: str = ''
+  stateinterface: str = ''
+  status: str = ''
+  def fromMessage(self, _m: Message):
+    for key in asdict(self).keys():
+      if key in _m:
+        setattr(self, key, _m[key])
+    return self
+
+@dataclass
+class GlobalVars:
+  FMDEVSTATE: str = ''
+  QUEDEVSTATE: str = ''
+  QUEUETOGGLE: str = ''
+  QUEUEPAUSETOGGLE: str = ''
+  INTERCOMCODE: str = ''
+  CAMPONTOGGLE: str = ''
+  DNDDEVSTATE: str = ''
+  CFDEVSTATE: str = ''
+  def d(self):
+    return asdict(self)
+
+def jsonAPIReply(status='success', data=None, message=None):
+  return {'status':status, 'data': data, 'message': message}
+
+def errorReply(message=None):
+  return jsonAPIReply(status='error', data=None, message=message)
+
+def successReply(data=None,message=None):
+  return jsonAPIReply(status='success', data=data, message=message)
+
+def noUser(user):
+  return errorReply('User {} does not exist'.format(user))
+
+def noDevice(device):
+  return errorReply('Device {} does not exist'.format(device))
+
+def noUserDevice(user):
+  return errorReply('User {} does not exist or is not bound to device'.format(user))
+
+def noUserChannel(user):
+  return errorReply('User {} does not have active calls'.format(user))
+
+def stateCacheEmpty():
+  return errorReply('Users states cache update failed')
+
+def invalidState(state):
+  return errorReply('Invalid state "{}" provided'.format(state))
+
+def alreadyBound(user, device):
+  return errorReply('User {} is already bound to device {}'.format(user, device))
+
+def bindError(user, device):
+  return errorReply('Failed binding user {} to device {}'.format(user, device))
+
+def hintError(user, device):
+  return errorReply('Failed setting hint for user {} on device {}'.format(user, device))
+
+def noUserBound(device):
+  return errorReply('No user is bound to device {}'.format(device))
+
+def successfullyTransfered(userA, userB):
+  return successReply({'userA':userA,'userB':userB},
+                      'Call was successfully transfered from user {} to user {}'.format(userA, userB))
+
+def successfullyOriginated(user, number):
+  return successReply({'user':user,'number':number},
+                      'Call was successfully originated from user {} to number {}'.format(user, number))
+
+def successfullyHungup(user):
+  return successReply({'user':user},
+                      'Call was successfully hungup for user {}'.format(user))
+
+def successfullyBound(user, device):
+  return successReply({'user':user,'device':device},
+                      'User {} is successfully bound to device {}'.format(user, device))
+
+def successfullyUnbound(user, device):
+  return successReply({'user':user,'device':device},
+                      'User {} was successfully unbound from device {}'.format(user, device))
+
+def successfullySetState(user, state):
+  return successReply({'user':user,'state':state},
+                      'State "{}" was successfully set for user {}'.format(state, user))
+
+def successCallbackURL(device, url):
+  return successReply({'device':device,'url':url},
+                      'Device {} callback url is: {}'.format(device, url))
+
+def successCommonCallbackURL(dest, url):
+  return successReply({'dest':dest,'url':url},
+                      '{} callback url is: {}'.format(dest, url))

certs.sh

@@ -0,0 +1,5 @@
+#!/bin/sh
+cp /etc/letsencrypt/live/aster.rrt.ru/fullchain.pem /var/storage/pbx/data/pbx/etc/asterisk/keys/integration/certificate.pem
+cp /etc/letsencrypt/live/aster.rrt.ru/privkey.pem /var/storage/pbx/data/pbx/etc/asterisk/keys/integration/webserver.key
+chown ats:ats /var/storage/pbx/data/pbx/etc/asterisk/keys/integration/*
+docker exec -it pbx supervisorctl restart rrtpbx

docker-compose.yml

@@ -60,6 +60,30 @@ services:
     - MYSQL_DATABASE
     - FREEPBX_CDRDBNAME
     - MYSQL_SERVER=db
+  app0:
+    container_name: app0
+    hostname: test-${APP_FQDN}
+    build: ./app0
+    restart: always
+    depends_on:
+    - pbx
+    env_file:
+    - app.env
+    ports:
+    - 1${APP_API_PORT}:${APP_API_PORT}
+    volumes:
+      - /etc/localtime:/etc/localtime:ro
+      - ./app0:/app
+      - ${PERSISTENT_STORAGE_PATH}/pbx/var/spool/asterisk/monitor:/app/static/records
+    environment:
+    - APP_FQDN
+    - APP_API_PORT
+    - APP_PORT_MYSQL
+    - MYSQL_USER
+    - MYSQL_PASSWORD
+    - MYSQL_DATABASE
+    - FREEPBX_CDRDBNAME
+    - MYSQL_SERVER=db
   nfs:
     container_name: nfs
     hostname: nfs.${APP_FQDN}