generated from westfarn/web_django_template
91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
"""SMTP2GO SMS REST API."""
|
|
|
|
import logging
|
|
import re
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
|
|
from email_sms.services import render_merge_tags
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _format_destination(phone: str) -> str:
|
|
"""Normalize stored phone to E.164-ish string SMTP2GO accepts."""
|
|
digits = re.sub(r"\D", "", phone or "")
|
|
if not digits:
|
|
raise ValueError("Contact has no phone number")
|
|
if phone.strip().startswith("+") and digits:
|
|
return f"+{digits}"
|
|
# US 10-digit local numbers → +1…
|
|
if len(digits) == 10:
|
|
return f"+1{digits}"
|
|
if len(digits) == 11 and digits.startswith("1"):
|
|
return f"+{digits}"
|
|
return f"+{digits}"
|
|
|
|
|
|
def _provider_error_detail(response: requests.Response) -> str:
|
|
"""Prefer SMTP2GO JSON error text over bare HTTP reason."""
|
|
try:
|
|
data = response.json()
|
|
except ValueError:
|
|
text = (response.text or "").strip()
|
|
return text[:500] if text else response.reason
|
|
nested = data.get("data") if isinstance(data.get("data"), dict) else {}
|
|
err = nested.get("error") or data.get("error") or ""
|
|
code = nested.get("error_code") or data.get("error_code") or ""
|
|
if err and code:
|
|
return f"{err} ({code})"
|
|
return str(err or code or response.reason)
|
|
|
|
|
|
def send_sms(message) -> str:
|
|
contact = message.contact
|
|
if not contact.phone:
|
|
raise ValueError("Contact has no phone number")
|
|
|
|
api_key = settings.SMTP2GO_SMS_API_KEY
|
|
if not api_key:
|
|
raise RuntimeError("SMTP2GO_SMS_API_KEY is not configured")
|
|
|
|
campaign = message.campaign
|
|
body = message.body_snapshot or campaign.body_override or (
|
|
campaign.template.body if campaign.template else ""
|
|
)
|
|
body = render_merge_tags(body, contact)
|
|
destination = _format_destination(contact.phone)
|
|
|
|
# Current SMTP2GO /v3/sms/send schema: destination[] + content.
|
|
payload = {
|
|
"api_key": api_key,
|
|
"destination": [destination],
|
|
"content": body[:1600],
|
|
}
|
|
response = requests.post(
|
|
settings.SMTP2GO_SMS_API_URL,
|
|
json=payload,
|
|
timeout=30,
|
|
)
|
|
if not response.ok:
|
|
detail = _provider_error_detail(response)
|
|
raise requests.HTTPError(
|
|
f"{response.status_code} Client Error: {detail} for url: {response.url}",
|
|
response=response,
|
|
)
|
|
|
|
data = response.json() if response.content else {}
|
|
nested = data.get("data") if isinstance(data.get("data"), dict) else {}
|
|
messages = nested.get("messages") if isinstance(nested.get("messages"), list) else []
|
|
first = messages[0] if messages and isinstance(messages[0], dict) else {}
|
|
return str(
|
|
first.get("message_id")
|
|
or nested.get("sms_id")
|
|
or nested.get("message_id")
|
|
or data.get("sms_id")
|
|
or data.get("message_id")
|
|
or data.get("request_id")
|
|
or f"sms-{message.pk}"
|
|
)
|