#!/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 = '''
{{ title }}
'''
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 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))