plutus/splynx.py

import time
import json
import requests # type: ignore
import hmac
import hashlib

SPLYNX_URL = 'https://billing.interphone.com.au'
#SPLYNX_KEY = 'c189c78b155ee8e4d389bbcb34bebc05'
#SPLYNX_SECRET = '1454679ddf5c97ea347766709d3ca3bd'
SPLYNX_KEY = 'b4cd90cbea15e7692c940484a9637fc4'
SPLYNX_SECRET = '297ce5c6b7cd5aaf93d8c725fcb49f8f'

class Splynx():
    def __init__(self, url, key, secret):
        self.url = url
        self.key = key
        self.secret = secret
        self.token = None
        self.refresh = None
        self.refreshtime = None
        self.refreshexpire = None

    def _authenticate(self):
        nonce = str(round(time.time() * 1000))
        sig = hmac.new(bytes(self.secret, 'UTF-8'),msg=bytes(nonce+self.key, 'UTF-8'), digestmod = hashlib.sha256).hexdigest().upper()
        data = { 'auth_type': 'api_key', 'key': self.key, 'nonce': nonce, 'signature': sig}
        headers = {'Content-Type': 'application/json'}
        ret = requests.post(url=self.url+'/api/2.0/admin/auth/tokens', data=json.dumps(data), headers=headers)
        jsonret = json.loads(ret.content)
        self.token = jsonret['access_token']
        self.refresh = jsonret['refresh_token']
        self.refreshtime = jsonret['access_token_expiration']
        self.refreshexpire = jsonret['refresh_token_expiration']

    def _refresh(self):
        headers = {'Content-Type': 'application/json', 'Authorization': 'Splynx-EA (access_token={token})'.format(token=self.token)}
        ret = requests.get(url=self.url+'/api/2.0/admin/auth/tokens/{refresh}'.format(refresh=self.refresh),headers=headers)
        jsonret = json.loads(ret.content)
        self.token = jsonret['access_token']
        self.refresh = jsonret['refresh_token']
        self.refreshtime = jsonret['access_token_expiration']
        self.refreshexpire = jsonret['refresh_token_expiration']

    def getToken(self):
        if self.token is None:
            self._authenticate()
        if self.token is not None and (self.refreshexpire <= int(time.time())):
            self._authenticate()
        if self.token is not None and (self.refreshtime <= int(time.time())):
            self._refresh()
        return self.token

    def get(self, url):
        headers = {'Content-Type': 'application/json', 'Authorization': 'Splynx-EA (access_token={token})'.format(token=self.getToken())}
        ret = requests.get(url=self.url+url, headers=headers)
        return json.loads(ret.content)

    def put(self, url, params):
        headers = {'Content-Type': 'application/json', 'Authorization': 'Splynx-EA (access_token={token})'.format(token=self.getToken())}
        ret = requests.put(url=self.url+url, headers=headers, data=json.dumps(params))
        if ret.status_code == 202:
            return True
        return False

    def post(self, url, params):
        headers = {'Content-Type': 'application/json', 'Authorization': 'Splynx-EA (access_token={token})'.format(token=self.getToken())}
        ret = requests.post(url=self.url+url, headers=headers, data=json.dumps(params))
        if ret.status_code == 201:
            return json.loads(ret.content)
        return False

    def delete(self, url):
        headers = {'Content-Type': 'application/json', 'Authorization': 'Splynx-EA (access_token={token})'.format(token=self.getToken())}
        ret = requests.delete(url=self.url+url, headers=headers)
        if ret.status_code == 204:
            return True
        return False


    def ServiceStatus(self, service_login):
        try:
            s1 = self.get(url=f"/api/2.0/admin/customers/customer/0/internet-services?main_attributes[login]={service_login}")

            if not s1:
                return { 'status': 'no service found', 'customer_name': 'none', 'customer_id': 'none' }
            
            service_status = s1[-1].get('status')
            
            if service_status == "active":
                s2 = self.get(url="/api/2.0/admin/customers/customers-online")
                
                if s2:
                    online_services = [d for d in s2 if str(service_login) in d.values()]
                    #print(f"online_services: {json.dumps(online_services,indent=2)}")
                    if online_services:
                        detail = online_services[0]
                        cust = self.Customer(detail['customer_id'])
                        detail['status'] = "Online"
                        detail['customer_name'] = cust['name']
                        return detail
                    else:
                        cust = self.Customer(s1[-1].get('customer_id'))
                        return { 'status': 'Offline', 'customer_name': cust['name'], 'customer_id': cust['id'] }
                else:
                    # No online customers data available
                    return { 'status': 'Offline' }
            else:
                # Service exists but is not active (could be suspended, terminated, etc.)
                cust = self.Customer(s1[-1].get('customer_id'))
                return { 'status': service_status.capitalize(), 'customer_name': cust['name'], 'customer_id': cust['id'] }
                
        except Exception as e:
            print(f"Error checking service status for {service_login}: {str(e)}")
            return { 'status': 'no service found', 'customer_name': 'none', 'customer_id': 'none' }
        
    def Customer(self, customer_id):
        try:
            result = self.get(url=f"/api/2.0/admin/customers/customer/{customer_id}/")
            #print(json.dumps(result,indent=2))
            return result
        except:
            return 'unknown'
        
    def GetInternetTariffs(self, tariff_id=None):
        try:
            if tariff_id:
                tariffs = self.get(url=f"/api/2.0/admin/tariffs/internet/{tariff_id}")
            else:
                tariffs = self.get(url=f"/api/2.0/admin/tariffs/internet")
            return tariffs
        except Exception as e:
            print(f"Error getting Internet Tariffs: {str(e)}")
            return { 'status': 'no Internet Tariff found'}