RRT
/
pbx


			
				
					
						
						
							123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173
							#!/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','')
externalContexts = ('from-trunk',
                    'from-digital',
                    'from-pstn',
                    'from-did-direct')
# 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 freepbxExternalContext(context):
  if ((context in externalContexts) or
      (context.startswith('from-trunk-')) or
      (context.startswith('from-did-')) or
      (context.startswith('ivr-'))):
    return True
  return False

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 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))