243 lines
8.2 KiB
Python
243 lines
8.2 KiB
Python
"""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() |