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, )