"""Async client for the voip.ms SMS/MMS REST API. Wraps the seven documented methods against https://voip.ms/api/v1/rest.php: sendSMS, sendMMS, getSMS, getMMS, getMediaMMS, deleteSMS, deleteMMS """ from __future__ import annotations import re from typing import Any import httpx from .config import Settings from .exceptions import VoipMsApiError, VoipMsAuthError, VoipMsError from .models import DeleteResult, MediaResult, MmsRecord, SendResult, SmsRecord _DIGITS = re.compile(r"\d+") def _digits(value: str) -> str: return "".join(_DIGITS.findall(value or "")) def normalize_number(value: str, mode: str) -> str: """Render a US/Canada number in the configured dialing mode. nanpa -> 10 digits (NPANXXXXXX) e164 -> +1 then 10 digits """ digits = _digits(value) # Tolerate an 11-digit number with leading 1. if len(digits) == 11 and digits.startswith("1"): digits = digits[1:] if len(digits) != 10: # Don't silently mangle unexpected shapes; pass through normalized digits. return digits if mode == "e164": return f"+1{digits}" return digits # voip.ms returns one of these `status` values (with a "There are no ... messages" # message) when a getSMS/getMMS query matches nothing. Treat as an empty list, # not an error. _EMPTY_STATUSES = {"no_sms", "no_mms"} def _is_empty(payload: dict[str, Any]) -> bool: status = str(payload.get("status", "")).lower() if status in _EMPTY_STATUSES: return True message = str(payload.get("message") or "").lower() return "there are no" in message and "message" in message _AUTH_HINTS = ( "invalid api", "incorrect api", "api_username", "api_password", "ip", "authorized", "whitelist", "incorrect", "invalid_credentials", "username or password", "credentials", ) class VoipMsSMSClient: """Thin async wrapper over the voip.ms SMS/MMS REST API.""" def __init__(self, settings: Settings) -> None: self._settings = settings self._http = httpx.AsyncClient( base_url=settings.base_url, timeout=settings.timeout, ) async def aclose(self) -> None: await self._http.aclose() # --- internals -------------------------------------------------------- async def _call(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]: """Perform one GET (or POST for base64 media) and return parsed JSON.""" query: dict[str, Any] = { "api_username": self._settings.api_username, "api_password": self._settings.api_password, "method": method, } if params: query.update({k: v for k, v in params.items() if v is not None}) try: response = await self._http.get("", params=query) except httpx.HTTPError as exc: raise VoipMsError(f"transport error calling {method}: {exc}") from exc if response.status_code >= 400: raise VoipMsError(f"HTTP {response.status_code} from voip.ms calling {method}") try: payload = response.json() except ValueError as exc: raise VoipMsError(f"non-JSON response from {method}: {response.text[:200]}") from exc return payload @staticmethod def _check_status(payload: dict[str, Any], method: str) -> None: status = str(payload.get("status", "")).lower() if status == "success": return message = str(payload.get("message") or payload or "unknown error") # Match against both the message and the status string (e.g. # "invalid_credentials") so auth failures classify as VoipMsAuthError. haystack = f"{status} {message}".lower() if any(hint in haystack for hint in _AUTH_HINTS): raise VoipMsAuthError(f"{method}: {message}") raise VoipMsApiError(f"{method}: {message}", code=payload.get("code")) # --- outbound --------------------------------------------------------- async def send_sms(self, *, did: str, dst: str, message: str) -> SendResult: if len(message) > 160: raise VoipMsError("SMS message exceeds 160 characters") mode = self._settings.dialing_mode payload = await self._call("sendSMS", { "did": normalize_number(did, mode), "dst": normalize_number(dst, mode), "message": message, }) self._check_status(payload, "sendSMS") return SendResult(id=int(payload["sms"])) async def send_mms( self, *, did: str, dst: str, message: str, media1: str | None = None, media2: str | None = None, media3: str | None = None, ) -> SendResult: if len(message) > 2048: raise VoipMsError("MMS message exceeds 2048 characters") mode = self._settings.dialing_mode payload = await self._call("sendMMS", { "did": normalize_number(did, mode), "dst": normalize_number(dst, mode), "message": message, "media1": media1, "media2": media2, "media3": media3, }) self._check_status(payload, "sendMMS") # sendMMS returns the new id under the "sms" key per the docs examples. return SendResult(id=int(payload.get("sms", payload.get("mms", 0)))) # --- inbound / history ------------------------------------------------ async def get_sms( self, *, sms: int | None = None, from_: str | None = None, to: str | None = None, type: int | None = None, did: str | None = None, contact: str | None = None, limit: int | None = None, timezone: int | None = None, all_messages: int | None = None, ) -> list[SmsRecord]: payload = await self._call("getSMS", { "sms": sms, "from": from_, "to": to, "type": type, "did": did, "contact": contact, "limit": limit, "timezone": timezone, "all_messages": all_messages, }) if _is_empty(payload): return [] self._check_status(payload, "getSMS") rows = payload.get("sms") or [] return [SmsRecord.model_validate(row) for row in rows] async def get_mms( self, *, id: int | None = None, from_: str | None = None, to: str | None = None, type: int | None = None, did: str | None = None, contact: str | None = None, limit: int | None = None, timezone: int | None = None, all_messages: int | None = None, ) -> list[MmsRecord]: payload = await self._call("getMMS", { "mms": id, "from": from_, "to": to, "type": type, "did": did, "contact": contact, "limit": limit, "timezone": timezone, "all_messages": all_messages, }) if _is_empty(payload): return [] self._check_status(payload, "getMMS") rows = payload.get("mms") or [] return [MmsRecord.model_validate(row) for row in rows] async def get_media_mms(self, *, id: int, media_as_array: bool = False) -> MediaResult: payload = await self._call("getMediaMMS", { "id": id, "media_as_array": 1 if media_as_array else 0, }) self._check_status(payload, "getMediaMMS") media = payload.get("media") if isinstance(media, dict): media_list = [v for v in media.values() if v] elif isinstance(media, list): media_list = [v for v in media if v] else: media_list = [] return MediaResult(id=payload.get("id", id), date=payload.get("date"), media=media_list) # --- delete ----------------------------------------------------------- async def delete_sms(self, *, id: int) -> DeleteResult: payload = await self._call("deleteSMS", {"id": id}) self._check_status(payload, "deleteSMS") return DeleteResult() async def delete_mms(self, *, id: int) -> DeleteResult: payload = await self._call("deleteMMS", {"id": id}) self._check_status(payload, "deleteMMS") return DeleteResult()