From 0bb7fb94cdecb646be264e222f108b40f88a2a1b Mon Sep 17 00:00:00 2001 From: Maff Date: Tue, 21 May 2019 20:02:25 +0100 Subject: [PATCH] Further rejiggering of classes --- Py3CX/__init__.py | 551 +++++++++++++++++++++++----------------------- 1 file changed, 275 insertions(+), 276 deletions(-) diff --git a/Py3CX/__init__.py b/Py3CX/__init__.py index 23e9e25..db342a1 100644 --- a/Py3CX/__init__.py +++ b/Py3CX/__init__.py @@ -1,276 +1,275 @@ -# Environment var definitions -ENV_URI='TCX_SITE_URI' -ENV_AUTH_USER='TCX_API_AUTH_USERNAME' -ENV_AUTH_PASS='TCX_API_AUTH_PASSWORD' - -class APIError(Exception): - pass -class ValidationError(Exception): - pass - -class Request: - from requests import ConnectionError, exceptions - def __init__(self, uri, verify): - from requests import Session - self.uri=uri - self.base="%s/api/{}" % uri - self.last=None - self.sess=Session() - self.sess.verify=verify - - def get(self, api, params=None, expect=200): - try: - self.last=resp=self.sess.get(url=self.base.format(api), params=params) - assert (resp.status_code==expect) - except AssertionError: - raise APIError("Assertion error when handling API response; response code was %s, but expected response was %s" % (resp.status_code, expect)) - except self.exceptions.SSLError as e: - raise APIError("TLS error returned when communicating; use tls_verify=False or check leaf certs: %s" % str(e)) - except self.exceptions.BaseHTTPError as e: - raise APIError("HTTP error raised when communicating: %s" % str(e)) - except ConnectionError as e: - raise APIError("ConnectionError raised when communicating: %s" % str(e)) - return resp - - def post(self, api, params, expect=200): - try: - self.last=resp=self.sess.post(url=self.base.format(api), json=params) - assert (resp.status_code==expect) - except AssertionError: - raise APIError("Assertion error when handling API response; response code was %s, but expected response was %s" % (resp.status_code, expect)) - except self.exceptions.SSLError as e: - raise APIError("TLS error returned when communicating; use tls_verify=False or check leaf certs: %s" % str(e)) - except self.exceptions.BaseHTTPError as e: - raise APIError("HTTP error raised when communicating: %s" % str(e)) - except ConnectionError as e: - raise APIError("ConnectionError raised when communicating: %s" % str(e)) - except Exception as e: - raise APIError("Other exception raised during API call: %s" % str(e)) - return resp - -class BasicObject(object): - pass -class ReadOnlyObject(object): - def __init__(self, parent, api): - self.tcx=parent - self.api=api - def refresh(self, params={}): - self._result=self.tcx.rq.get(self.api, params=params) - self.active=self._result.json() -class TransactionalObject(object): - def __init__(self, parent, api, save='edit/save', discard='edit/cancel'): - self.tcx=parent - self.api=api - self.save=save - self.discard=discard - def create(self, params): - self._result=self.tcx.rq.post(self.api, params=params) - self._session=self._result.json()['Id'] - self.active=self._result.json()['ActiveObject'] - def submit(self, params): - raise ValidationError("NotImplemented error") - def cancel(self): - self.tcx.rq.post(self.discard, params={ - 'Id':self._session}) - -class Call(ReadOnlyObject): - def __init__(self, tcx, callid): - super().__init__(tcx, 'activeCalls') - self.params=callid - def refresh(self): - self.refresh() - self.timestamp=self._result.headers.get('date') - res=list(filter(lambda cid: cid['Id'] == self.params, self._result.json()['list'])) - assert (len(res)>0), "No call found in currently-active call list for ID %s" % self.params - assert(len(res)==1), "More than one active call found for ID %s" % self.params - res=res[0] - self.id=res['Id'] - self.caller=res['Caller'] - self.callee=res['Callee'] - self.state=res['Status'] - self.duration=res['Duration'] - self.since=res['LastChangeStatus'] - def hangup(self): - self.tcx.rq.post('activeCalls/drop', params={ - 'Id': self.params}) - self.params=None - -class User(TransactionalObject): - def __init__(self, parent, params): - super().__init__(parent, 'ExtensionList/set') - self.params=params - self.load() - def load(self): - self.create(params={ - 'Id':self.params}) - parms=self.active - self.id=parms['Id'] - self.enabled=not parms['Disabled']['_value'] - self.sip_id=parms['SIPId']['_value'] - #RecordCallsOption - enum - self.callrecs=parms['RecordCalls']['selected'] - self.sip_host=parms['MyPhoneLocalInterface']['selected'] - self.sip_host_addrs=parms['MyPhoneLocalInterface']['possibleValues'] - self.sip_authid=parms['AuthId']['_value'] - self.sip_authpw=parms['AuthPassword']['_value'] - self.provision_uri=parms['MyPhoneProvLink']['_value'] - self.webpw=parms['AccessPassword']['_value'] - self.extension=Extension(self.tcx, self.params) - self.cancel() -class Extension(ReadOnlyObject): - def __init__(self, parent, params, populate=True): - super().__init__(parent, api='ExtensionList') - self.params=params - if populate: - self.load() - def load(self): - self.refresh(params=self.params) - res=self._result - self.timestamp=res.headers.get('date') - res=list(filter(lambda ext: ext['Number'] == self.params, res.json()['list'])) - assert (len(res)>0), "No extension found for: %s" % self.params - assert (len(res)==1), "More than one extension found for %s" % self.params - res=res[0] - self.number=res['Number'] - self.firstname=res['FirstName'] - self.surname=res['LastName'] - self.name="%s %s" % (self.firstname, self.surname) - self.dnd=res['DND'] - self.status=res['CurrentProfile'] - self.mail=res['Email'] - self.cli=res['OutboundCallerId'] - self.mobile=res['MobileNumber'] - self.online=res['IsRegistered'] -class PhoneSystem: - class System(object): - class License(object): - pass - class Status(object): - pass - def __init__(self, tcx): - self.tcx=tcx - self.extlist=None - def refresh_sysstat(self): - sysstat=self.tcx.rq.get('SystemStatus').json() - self.Status.exts_online=sysstat['ExtensionsRegistered'] - self.Status.trunks_online=sysstat['TrunksRegistered'] - self.Status.calls_active=sysstat['CallsActive'] - self.Status.diskbytes_free=sysstat['FreeDiskSpace'] - self.Status.membytes_free=sysstat['FreePhysicalMemory'] - self.Status.banned_ips=sysstat['BlacklistedIpCount'] - self.Status.last_backup=sysstat['LastBackupDateTime'] - self.Status.sysexts_online=not sysstat['HasUnregisteredSystemExtensions'] - self.Status.services_online=not sysstat['HasNotRunningServices'] - self.System.routes_out=sysstat['OutboundRules'] - self.System.fqdn=sysstat['FQDN'] - self.System.webconf=sysstat['WebMeetingFQDN'] - self.System.version=sysstat['Version'] - self.System.diskbytes_total=sysstat['TotalDiskSpace'] - self.System.membytes_total=sysstat['TotalPhysicalMemory'] - self.System.exts_total=sysstat['ExtensionsTotal'] - self.System.trunks_total=sysstat['TrunksTotal'] - self.System.firebase_confed=sysstat['OwnPush'] - self.System.backups_enabled=sysstat['BackupScheduled'] - self.System.License.activated=sysstat['Activated'] - self.System.License.calls=sysstat['MaxSimCalls'] - self.System.License.meetingcalls=sysstat['MaxSimMeetingParticipants'] - self.System.License.supported=sysstat['Support'] - self.System.License.expire=sysstat['ExpirationDate'] - self.System.License.maintexpire=sysstat['MaintenanceExpiresAt'] - self.System.License.reseller=sysstat['ResellerName'] - self.System.License.key=sysstat['LicenseKey'] - self.System.License.sku=sysstat['ProductCode'] - self.System.License.spla=sysstat['IsSpla'] - def refresh_addl(self): - addl=self.tcx.rq.get('SystemStatus/AdditionalStatus').json() - self.Status.callrecs_bytes=addl['RecordingUsedSpace'] - self.System.callrecs_enabled=not addl['RecordingStopped'] - self.System.callrecs_quota=addl['RecordingQuota'] - def refresh_inroute(self): - inroute=self.tcx.rq.get('InboundRulesList').json()['list'] - self.System.routes_in=len(inroute) - def refresh_hists(self): - hists=self.tcx.rq.post('SystemStatus/getDbInformation', params={}).json() - self.System.calls_total=hists['CallHistoryCount'] - self.System.chats_total=hists['ChatMessagesCount'] - def refresh_certified(self): - certstat=self.tcx.rq.get('SystemStatus/GetSingleStatus').json() - self.System.License.firewallsupported=certstat['Health']['Firewall'] - self.System.License.trunkssupported=certstat['Health']['Trunks'] - self.System.License.phonessupported=certstat['Health']['Phones'] - def refresh(self): - self.refresh_addl() - self.refresh_certified() - self.refresh_hists() - self.refresh_inroute() - self.refresh_sysstat() - @property - def extensionlist(self): - if self.extlist is None: - self.extlist=ReadOnlyObject(self.tcx, api='ExtensionList') - self.extlist.refresh() - res=self.extlist._result.json()['list'] - assert (len(res)>0), "No extensions found" - ret=[] - for extension in res: - this=Extension(self.tcx, params=extension['Number'], populate=False) - this.firstname=extension['FirstName'] - this.surname=extension['LastName'] - this.name="%s %s" % (this.firstname, this.surname) - this.dnd=extension['DND'] - this.status=extension['CurrentProfile'] - this.mail=extension['Email'] - this.cli=extension['OutboundCallerId'] - this.mobile=extension['MobileNumber'] - this.online=extension['IsRegistered'] - ret.append(this) - return ret - @property - def callslist(self): - if self.calllist is None: - self.calllist=ReadOnlyObject(self.tcx, api='activeCalls') - self.calllist.refresh() - res=self.calllist._result.json()['list'] - if len(res)==0: return [] - ret=[] - for call in res: - this=Call(self.tcx, call['Id']) - this.caller=call['Caller'] - this.callee=call['Callee'] - this.state=call['Status'] - this.duration=res['Duration'] - this.since=res['LastChangeStatus'] - ret.append(this) - return ret - -class Py3CX: - def __init__(self, uri=None, tls_verify=True): - from os import getenv - self.uri=uri if uri is not None and uri.startswith('http') else getenv(ENV_URI, None) - assert (self.uri is not None and self.uri.startswith('http')), "Please provide URI" - self.uname=getenv(ENV_AUTH_USER, None) - self.passw=getenv(ENV_AUTH_PASS, None) - self.rq=Request(uri=self.uri, verify=tls_verify) - def authenticate(self, username=None, password=None): - if username is not None and password is not None: - self.uname=username - self.passw=password - assert (self.uname is not None and self.passw is not None), "Authentication information needed" - rs=self.rq.post('login', params={ - 'Username': self.uname, - 'Password': self.passw}) - self.rq.sess.headers.update({'x-xsrf-token':rs.cookies['XSRF-TOKEN']}) - @property - def authenticated(self): - try: - self.rq.get('CurrentUser') - except APIError: - return False - return True - @property - def System(self): - assert self.authenticated, "Py3CX not authenticated yet!" - if self.tcxsystem is None: - self.tcxsystem=PhoneSystem(self) - return self.tcxsystem - +# Environment var definitions +ENV_URI='TCX_SITE_URI' +ENV_AUTH_USER='TCX_API_AUTH_USERNAME' +ENV_AUTH_PASS='TCX_API_AUTH_PASSWORD' + +class APIError(Exception): + pass +class ValidationError(Exception): + pass + +class Request: + from requests import ConnectionError, exceptions + def __init__(self, uri, verify): + from requests import Session + self.uri=uri + self.base="%s/api/{}" % uri + self.last=None + self.sess=Session() + self.sess.verify=verify + + def get(self, api, params=None, expect=200): + try: + self.last=resp=self.sess.get(url=self.base.format(api), params=params) + assert (resp.status_code==expect) + except AssertionError: + raise APIError("Assertion error when handling API response; response code was %s, but expected response was %s" % (resp.status_code, expect)) + except self.exceptions.SSLError as e: + raise APIError("TLS error returned when communicating; use tls_verify=False or check leaf certs: %s" % str(e)) + except self.exceptions.BaseHTTPError as e: + raise APIError("HTTP error raised when communicating: %s" % str(e)) + except ConnectionError as e: + raise APIError("ConnectionError raised when communicating: %s" % str(e)) + return resp + + def post(self, api, params, expect=200): + try: + self.last=resp=self.sess.post(url=self.base.format(api), json=params) + assert (resp.status_code==expect) + except AssertionError: + raise APIError("Assertion error when handling API response; response code was %s, but expected response was %s" % (resp.status_code, expect)) + except self.exceptions.SSLError as e: + raise APIError("TLS error returned when communicating; use tls_verify=False or check leaf certs: %s" % str(e)) + except self.exceptions.BaseHTTPError as e: + raise APIError("HTTP error raised when communicating: %s" % str(e)) + except ConnectionError as e: + raise APIError("ConnectionError raised when communicating: %s" % str(e)) + except Exception as e: + raise APIError("Other exception raised during API call: %s" % str(e)) + return resp + +class BasicObject(object): + pass +class APIObject(object): + def __init__(self, parent, api): + self.tcx=parent + self.api=api +class ReadOnlyObject(APIObject): + def refresh(self, params={}): + self._result=self.tcx.rq.get(self.api, params=params) + self.active=self._result.json() +class TransactionalObject(APIObject): + def __init__(self, parent, api, save='edit/save', discard='edit/cancel'): + super().__init__(parent, api) + self.save=save + self.discard=discard + def create(self, params): + self._result=self.tcx.rq.post(self.api, params=params) + self._session=self._result.json()['Id'] + self.active=self._result.json()['ActiveObject'] + def submit(self, params): + raise ValidationError("NotImplemented error") + def cancel(self): + self.tcx.rq.post(self.discard, params={ + 'Id':self._session}) + +class Py3CX: + class __Call(ReadOnlyObject): + def __init__(self, tcx, callid): + super().__init__(tcx, 'activeCalls') + self.params=callid + def refresh(self): + self.refresh() + self.timestamp=self._result.headers.get('date') + res=list(filter(lambda cid: cid['Id'] == self.params, self._result.json()['list'])) + assert (len(res)>0), "No call found in currently-active call list for ID %s" % self.params + assert(len(res)==1), "More than one active call found for ID %s" % self.params + res=res[0] + self.id=res['Id'] + self.caller=res['Caller'] + self.callee=res['Callee'] + self.state=res['Status'] + self.duration=res['Duration'] + self.since=res['LastChangeStatus'] + def hangup(self): + self.tcx.rq.post('activeCalls/drop', params={ + 'Id': self.params}) + self.params=None + class __User(TransactionalObject): + def __init__(self, parent, params): + super().__init__(parent, 'ExtensionList/set') + self.params=params + self.load() + def load(self): + self.create(params={ + 'Id':self.params}) + parms=self.active + self.id=parms['Id'] + self.enabled=not parms['Disabled']['_value'] + self.sip_id=parms['SIPId']['_value'] + #RecordCallsOption - enum + self.callrecs=parms['RecordCalls']['selected'] + self.sip_host=parms['MyPhoneLocalInterface']['selected'] + self.sip_host_addrs=parms['MyPhoneLocalInterface']['possibleValues'] + self.sip_authid=parms['AuthId']['_value'] + self.sip_authpw=parms['AuthPassword']['_value'] + self.provision_uri=parms['MyPhoneProvLink']['_value'] + self.webpw=parms['AccessPassword']['_value'] + self.extension=Py3CX.__Extension(self.tcx, self.params) + self.cancel() + class __Extension(ReadOnlyObject): + def __init__(self, parent, params, populate=True): + super().__init__(parent, api='ExtensionList') + self.params=params + if populate: + self.load() + def load(self): + self.refresh(params=self.params) + res=self._result + self.timestamp=res.headers.get('date') + res=list(filter(lambda ext: ext['Number'] == self.params, res.json()['list'])) + assert (len(res)>0), "No extension found for: %s" % self.params + assert (len(res)==1), "More than one extension found for %s" % self.params + res=res[0] + self.number=res['Number'] + self.firstname=res['FirstName'] + self.surname=res['LastName'] + self.name="%s %s" % (self.firstname, self.surname) + self.dnd=res['DND'] + self.status=res['CurrentProfile'] + self.mail=res['Email'] + self.cli=res['OutboundCallerId'] + self.mobile=res['MobileNumber'] + self.online=res['IsRegistered'] + class __PhoneSystem: + class System(object): + class License(object): + pass + class Status(object): + pass + def __init__(self, tcx): + self.tcx=tcx + self.extlist=None + def refresh_sysstat(self): + sysstat=self.tcx.rq.get('SystemStatus').json() + self.Status.exts_online=sysstat['ExtensionsRegistered'] + self.Status.trunks_online=sysstat['TrunksRegistered'] + self.Status.calls_active=sysstat['CallsActive'] + self.Status.diskbytes_free=sysstat['FreeDiskSpace'] + self.Status.membytes_free=sysstat['FreePhysicalMemory'] + self.Status.banned_ips=sysstat['BlacklistedIpCount'] + self.Status.last_backup=sysstat['LastBackupDateTime'] + self.Status.sysexts_online=not sysstat['HasUnregisteredSystemExtensions'] + self.Status.services_online=not sysstat['HasNotRunningServices'] + self.System.routes_out=sysstat['OutboundRules'] + self.System.fqdn=sysstat['FQDN'] + self.System.webconf=sysstat['WebMeetingFQDN'] + self.System.version=sysstat['Version'] + self.System.diskbytes_total=sysstat['TotalDiskSpace'] + self.System.membytes_total=sysstat['TotalPhysicalMemory'] + self.System.exts_total=sysstat['ExtensionsTotal'] + self.System.trunks_total=sysstat['TrunksTotal'] + self.System.firebase_confed=sysstat['OwnPush'] + self.System.backups_enabled=sysstat['BackupScheduled'] + self.System.License.activated=sysstat['Activated'] + self.System.License.calls=sysstat['MaxSimCalls'] + self.System.License.meetingcalls=sysstat['MaxSimMeetingParticipants'] + self.System.License.supported=sysstat['Support'] + self.System.License.expire=sysstat['ExpirationDate'] + self.System.License.maintexpire=sysstat['MaintenanceExpiresAt'] + self.System.License.reseller=sysstat['ResellerName'] + self.System.License.key=sysstat['LicenseKey'] + self.System.License.sku=sysstat['ProductCode'] + self.System.License.spla=sysstat['IsSpla'] + def refresh_addl(self): + addl=self.tcx.rq.get('SystemStatus/AdditionalStatus').json() + self.Status.callrecs_bytes=addl['RecordingUsedSpace'] + self.System.callrecs_enabled=not addl['RecordingStopped'] + self.System.callrecs_quota=addl['RecordingQuota'] + def refresh_inroute(self): + inroute=self.tcx.rq.get('InboundRulesList').json()['list'] + self.System.routes_in=len(inroute) + def refresh_hists(self): + hists=self.tcx.rq.post('SystemStatus/getDbInformation', params={}).json() + self.System.calls_total=hists['CallHistoryCount'] + self.System.chats_total=hists['ChatMessagesCount'] + def refresh_certified(self): + certstat=self.tcx.rq.get('SystemStatus/GetSingleStatus').json() + self.System.License.firewallsupported=certstat['Health']['Firewall'] + self.System.License.trunkssupported=certstat['Health']['Trunks'] + self.System.License.phonessupported=certstat['Health']['Phones'] + def refresh(self): + self.refresh_addl() + self.refresh_certified() + self.refresh_hists() + self.refresh_inroute() + self.refresh_sysstat() + @property + def extensionlist(self): + if self.extlist is None: + self.extlist=ReadOnlyObject(self.tcx, api='ExtensionList') + self.extlist.refresh() + res=self.extlist._result.json()['list'] + assert (len(res)>0), "No extensions found" + ret=[] + for extension in res: + this=Py3CX.__Extension(self.tcx, params=extension['Number'], populate=False) + this.firstname=extension['FirstName'] + this.surname=extension['LastName'] + this.name="%s %s" % (this.firstname, this.surname) + this.dnd=extension['DND'] + this.status=extension['CurrentProfile'] + this.mail=extension['Email'] + this.cli=extension['OutboundCallerId'] + this.mobile=extension['MobileNumber'] + this.online=extension['IsRegistered'] + ret.append(this) + return ret + @property + def callslist(self): + if self.calllist is None: + self.calllist=ReadOnlyObject(self.tcx, api='activeCalls') + self.calllist.refresh() + res=self.calllist._result.json()['list'] + if len(res)==0: return [] + ret=[] + for call in res: + this=Py3CX.__Call(self.tcx, call['Id']) + this.caller=call['Caller'] + this.callee=call['Callee'] + this.state=call['Status'] + this.duration=res['Duration'] + this.since=res['LastChangeStatus'] + ret.append(this) + return ret + def __init__(self, uri=None, tls_verify=True): + from os import getenv + self.uri=uri if uri is not None and uri.startswith('http') else getenv(ENV_URI, None) + assert (self.uri is not None and self.uri.startswith('http')), "Please provide URI" + self.uname=getenv(ENV_AUTH_USER, None) + self.passw=getenv(ENV_AUTH_PASS, None) + self.rq=Request(uri=self.uri, verify=tls_verify) + self.__tcxsystem=None + def authenticate(self, username=None, password=None): + if username is not None and password is not None: + self.uname=username + self.passw=password + assert (self.uname is not None and self.passw is not None), "Authentication information needed" + rs=self.rq.post('login', params={ + 'Username': self.uname, + 'Password': self.passw}) + self.rq.sess.headers.update({'x-xsrf-token':rs.cookies['XSRF-TOKEN']}) + @property + def authenticated(self): + try: + self.rq.get('CurrentUser') + except APIError: + return False + return True + @property + def System(self): + assert self.authenticated, "Py3CX not authenticated yet!" + if self.__tcxsystem is None: + self.__tcxsystem=__PhoneSystem(self) + return self.__tcxsystem +