70 lines
1.9 KiB
Python
70 lines
1.9 KiB
Python
from core.classes.sms import BaseSMSProvider, SMSNotSent
|
|
|
|
from dbsettings.functions import getValue
|
|
|
|
from urllib.parse import urlencode
|
|
from urllib.request import urlopen, Request
|
|
from typing import Union
|
|
|
|
import json
|
|
|
|
# Create your models here.
|
|
|
|
class KumiSMSServer(BaseSMSProvider):
|
|
@property
|
|
def is_active(self):
|
|
return bool(self.get_balance)
|
|
|
|
@property
|
|
def get_name(self):
|
|
return "Kumi SMS"
|
|
|
|
@property
|
|
def get_key(self):
|
|
return getValue("kumisms.apikey")
|
|
|
|
@staticmethod
|
|
def getError(status):
|
|
if "error" in status.keys():
|
|
return status["error"]
|
|
|
|
def sendSMS(self, recipients: Union[str, list], message: str):
|
|
'''Send an SMS message to one or more recipients
|
|
|
|
:param recipients: Recipient phone number as a string, or a list of multiple phone number strings
|
|
:param message: Message to be sent as a string
|
|
'''
|
|
|
|
if isinstance(recipients, str):
|
|
recipients = [recipients]
|
|
|
|
url = 'https://kumisms.com/api/v1/send/'
|
|
|
|
for recipient in recipients:
|
|
vars = {"key": self.get_key, "text": message, "recipient": recipient}
|
|
|
|
request = Request(url, urlencode(vars).encode())
|
|
response = urlopen(request)
|
|
|
|
status = json.loads(response.read().decode())
|
|
|
|
error = KumiSMSServer.getError(status)
|
|
|
|
if error:
|
|
raise SMSNotSent(f'An error occurred trying to send the SMS: {error}')
|
|
|
|
@property
|
|
def get_balance(self):
|
|
url = 'https://kumisms.com/api/v1/balance/'
|
|
|
|
vars = {"key": self.get_key}
|
|
|
|
request = Request(url, urlencode(vars).encode())
|
|
response = urlopen(request)
|
|
|
|
status = json.loads(response.read().decode())
|
|
|
|
if not KumiSMSServer.getError(status):
|
|
return status["balance"]
|
|
|
|
SMSPROVIDERS = [KumiSMSServer()] |