husanjon
1daf4f3569
Some checks failed
Build and Push to Docker Hub / build-test-push (push) Failing after 2m4s
143 lines
4.5 KiB
Python
Executable File
143 lines
4.5 KiB
Python
Executable File
import requests
|
|
import logging
|
|
from django.core.cache import cache
|
|
from common.env import env
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class SendService:
|
|
GET = "GET"
|
|
POST = "POST"
|
|
PATCH = "PATCH"
|
|
CONTACT = "contact"
|
|
CACHE_KEY = "eskiz_token"
|
|
TOKEN_LIFETIME = 86400 * 20 # Eskiz tokens usually last 30 days, we cache for 20
|
|
|
|
def __init__(
|
|
self, api_url=None, email=None, password=None, callback_url=None
|
|
):
|
|
self.api_url = (api_url or env("SMS_API_URL")).rstrip("/")
|
|
self.email = email or env("SMS_LOGIN")
|
|
self.password = password or env("SMS_PASSWORD")
|
|
self.callback_url = callback_url
|
|
self.headers = {
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
self.methods = {
|
|
"auth_user": "auth/user",
|
|
"auth_login": "auth/login",
|
|
"auth_refresh": "auth/refresh",
|
|
"send_message": "message/sms/send",
|
|
}
|
|
|
|
def request(self, api_path, data=None, method=None, headers=None):
|
|
url = f"{self.api_url}/{api_path}"
|
|
method = method or self.POST
|
|
final_headers = self.headers.copy()
|
|
if headers:
|
|
final_headers.update(headers)
|
|
|
|
try:
|
|
# Reverting to data=data (form-encoded) as it was working before
|
|
response = requests.request(
|
|
method,
|
|
url,
|
|
data=data if method != self.GET else None,
|
|
params=data if method == self.GET else None,
|
|
headers=final_headers,
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
logger.error(f"SMS API HTTP Error: {e.response.status_code} - {e.response.text}")
|
|
return e.response.json()
|
|
except Exception as e:
|
|
logger.error(f"SMS API Request Exception: {str(e)}")
|
|
return {"status": "error", "message": str(e)}
|
|
|
|
def get_token(self):
|
|
token = cache.get(self.CACHE_KEY)
|
|
if not token:
|
|
logger.info("Eskiz token not found in cache, authenticating...")
|
|
auth_res = self.auth()
|
|
if auth_res.get("status") == "success" or "data" in auth_res:
|
|
token = auth_res.get("data", {}).get("token")
|
|
if token:
|
|
cache.set(self.CACHE_KEY, token, self.TOKEN_LIFETIME)
|
|
else:
|
|
logger.error(f"Eskiz Auth Failed: {auth_res}")
|
|
return token
|
|
|
|
def auth(self):
|
|
data = {"email": self.email, "password": self.password}
|
|
# Eskiz login expects JSON
|
|
return self.request(
|
|
self.methods["auth_login"], data=data, method=self.POST
|
|
)
|
|
|
|
def refresh_token(self):
|
|
token = self.get_token()
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
res = self.request(
|
|
self.methods["auth_refresh"],
|
|
method=self.PATCH,
|
|
headers=headers,
|
|
)
|
|
if res.get("status") == "success":
|
|
# Some Eskiz versions return a new token on refresh
|
|
new_token = res.get("data", {}).get("token")
|
|
if new_token:
|
|
cache.set(self.CACHE_KEY, new_token, self.TOKEN_LIFETIME)
|
|
return res
|
|
|
|
def get_my_user_info(self):
|
|
token = self.get_token()
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
return self.request(
|
|
self.methods["auth_user"], method=self.GET, headers=headers
|
|
)
|
|
|
|
def add_sms_contact(self, first_name, phone_number, group):
|
|
token = self.get_token()
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
data = {
|
|
"name": first_name,
|
|
"email": self.email,
|
|
"group": group,
|
|
"mobile_phone": phone_number,
|
|
}
|
|
|
|
return self.request(
|
|
self.CONTACT,
|
|
data=data,
|
|
method=self.POST,
|
|
headers=headers,
|
|
)
|
|
|
|
def send_sms(self, phone_number, message):
|
|
token = self.get_token()
|
|
if not token:
|
|
return {"status": "error", "message": "Failed to obtain auth token"}
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# Use integer 4546 for sender name
|
|
data = {
|
|
"from": 4546,
|
|
"mobile_phone": phone_number,
|
|
"callback_url": self.callback_url,
|
|
"message": message,
|
|
}
|
|
|
|
res = self.request(
|
|
self.methods["send_message"],
|
|
data=data,
|
|
method=self.POST,
|
|
headers=headers,
|
|
)
|
|
return res |