92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from .config import settings
|
|
|
|
|
|
class SmsChannel(str, Enum):
|
|
TRANSACTIONAL = "Trans"
|
|
PROMOTIONAL = "Promo"
|
|
|
|
|
|
@dataclass
|
|
class SendResult:
|
|
success: bool
|
|
job_id: str | None
|
|
message_id: str | None
|
|
number: str | None
|
|
raw: dict[str, Any]
|
|
|
|
|
|
class UltronsmsClient:
|
|
def __init__(
|
|
self,
|
|
base_url: str | None = None,
|
|
user: str | None = None,
|
|
password: str | None = None,
|
|
) -> None:
|
|
self.base_url = (base_url or settings.ultronsms_base_url).rstrip("/")
|
|
self.user = user or settings.ultronsms_user
|
|
self.password = password or settings.ultronsms_password
|
|
|
|
@staticmethod
|
|
def normalize_phone(number: str) -> str:
|
|
digits = "".join(character for character in number if character.isdigit())
|
|
if len(digits) == 10:
|
|
return f"91{digits}"
|
|
return digits
|
|
|
|
@staticmethod
|
|
def route_for_channel(channel: SmsChannel) -> str:
|
|
return "2" if channel == SmsChannel.TRANSACTIONAL else "1"
|
|
|
|
@staticmethod
|
|
def channel_for_sender_type(sender_type: str) -> SmsChannel:
|
|
return SmsChannel.PROMOTIONAL if sender_type == "promotional" else SmsChannel.TRANSACTIONAL
|
|
|
|
def _get(self, path: str, params: dict[str, str]) -> dict[str, Any]:
|
|
if not self.user or not self.password:
|
|
raise RuntimeError("Ultronsms credentials are not configured")
|
|
with httpx.Client(timeout=30.0, follow_redirects=True) as client:
|
|
response = client.get(f"{self.base_url}{path}", params=params)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def send_template_sms(
|
|
self,
|
|
mobile: str,
|
|
message: str,
|
|
*,
|
|
sender_id: str,
|
|
pe_id: str,
|
|
dlt_template_id: str,
|
|
sender_type: str,
|
|
dcs: str = "0",
|
|
flash_sms: str = "0",
|
|
) -> SendResult:
|
|
channel = self.channel_for_sender_type(sender_type)
|
|
params = {
|
|
"user": self.user,
|
|
"password": self.password,
|
|
"senderid": sender_id,
|
|
"channel": channel.value,
|
|
"route": self.route_for_channel(channel),
|
|
"DCS": dcs,
|
|
"flashsms": flash_sms,
|
|
"number": self.normalize_phone(mobile),
|
|
"text": message,
|
|
"peid": pe_id,
|
|
"dlttemplateid": dlt_template_id,
|
|
}
|
|
raw = self._get("/api/mt/SendSMS", params)
|
|
message_data = (raw.get("MessageData") or [{}])[0]
|
|
return SendResult(
|
|
success=raw.get("ErrorCode") == "000",
|
|
job_id=raw.get("JobId"),
|
|
message_id=message_data.get("MessageId"),
|
|
number=message_data.get("Number"),
|
|
raw=raw,
|
|
)
|