generated from westfarn/web_django_template
808 lines
25 KiB
Python
808 lines
25 KiB
Python
"""SMTP2GO email/SMS event webhooks → Message + ProviderEvent updates."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from django.core.exceptions import ValidationError
|
|
from django.http import HttpRequest
|
|
|
|
from contacts.models import Channel, Contact
|
|
from directmail.models import Message, ProviderEvent
|
|
from directmail.services import set_channel_consent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROVIDER_EMAIL = "smtp2go_email"
|
|
PROVIDER_SMS = "smtp2go_sms"
|
|
PROVIDER_PCM = "pcm"
|
|
PROVIDER = PROVIDER_EMAIL # backward-compatible alias
|
|
|
|
# Do not move a message backward to a weaker delivery / engagement state.
|
|
_STATUS_RANK = {
|
|
Message.Status.DRAFT: 0,
|
|
Message.Status.SCHEDULED: 1,
|
|
Message.Status.QUEUED: 2,
|
|
Message.Status.SENT: 3,
|
|
Message.Status.FAILED: 3,
|
|
Message.Status.DELIVERED: 4,
|
|
Message.Status.OPENED: 5,
|
|
Message.Status.CLICKED: 6,
|
|
Message.Status.BOUNCED: 7,
|
|
Message.Status.SUPPRESSED: 7,
|
|
}
|
|
|
|
_ENGAGED_OR_DELIVERED = frozenset(
|
|
{
|
|
Message.Status.DELIVERED,
|
|
Message.Status.OPENED,
|
|
Message.Status.CLICKED,
|
|
}
|
|
)
|
|
_SENT_OR_BETTER = frozenset(
|
|
{
|
|
Message.Status.SENT,
|
|
Message.Status.DELIVERED,
|
|
Message.Status.OPENED,
|
|
Message.Status.CLICKED,
|
|
}
|
|
)
|
|
|
|
_MONICA_HEADER_KEYS = (
|
|
"X-Monica-Message-Id",
|
|
"x-monica-message-id",
|
|
"X_Monica_Message_Id",
|
|
"monica-message-id",
|
|
)
|
|
|
|
# SMTP2GO UI labels → canonical event strings from their docs.
|
|
_EMAIL_EVENT_ALIASES = {
|
|
"bounced": "bounce",
|
|
"rejected": "reject",
|
|
"opened": "open",
|
|
"clicked": "click",
|
|
"unsubscribed": "unsubscribe",
|
|
"resubscribed": "resubscribe",
|
|
}
|
|
|
|
# API sms_events use short names (delivered); webhook body often uses sms_delivered.
|
|
# UI test labels / Opt-Out may arrive without the sms_ prefix.
|
|
_SMS_EVENT_ALIASES = {
|
|
"submitted": "sms_submitted",
|
|
"sending": "sms_sending",
|
|
"delivered": "sms_delivered",
|
|
"failed": "sms_failed",
|
|
"rejected": "sms_rejected",
|
|
"opt_out": "sms_opt_out",
|
|
"optout": "sms_opt_out",
|
|
"sms_optout": "sms_opt_out",
|
|
}
|
|
|
|
_SMS_OPT_OUT_EVENTS = frozenset({"sms_opt_out"})
|
|
|
|
|
|
def _as_str(value: Any) -> str:
|
|
"""Coerce webhook field values to a stripped string (lists / None safe)."""
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, (list, tuple)):
|
|
if not value:
|
|
return ""
|
|
value = value[0]
|
|
if isinstance(value, bytes):
|
|
value = value.decode("utf-8", errors="replace")
|
|
return str(value).strip()
|
|
|
|
|
|
def _json_safe(value: Any) -> Any:
|
|
"""Ensure ProviderEvent.payload can be stored as JSON."""
|
|
if value is None or isinstance(value, (str, int, float, bool)):
|
|
return value
|
|
if isinstance(value, dict):
|
|
return {str(k): _json_safe(v) for k, v in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [_json_safe(v) for v in value]
|
|
return str(value)
|
|
|
|
|
|
def _normalize_email_event(event: str) -> str:
|
|
event = (event or "").strip().lower()
|
|
return _EMAIL_EVENT_ALIASES.get(event, event)
|
|
|
|
|
|
def _normalize_sms_event(event: str) -> str:
|
|
"""Map UI / short API names to docs canonical sms_* event strings."""
|
|
event = (event or "").strip().lower().replace("-", "_").replace(" ", "_")
|
|
return _SMS_EVENT_ALIASES.get(event, event)
|
|
|
|
|
|
def parse_webhook_payload(request: HttpRequest) -> dict[str, Any]:
|
|
"""Accept JSON or form-encoded SMTP2GO webhook bodies."""
|
|
content_type = (request.content_type or "").lower()
|
|
if "application/json" in content_type:
|
|
try:
|
|
data = json.loads(request.body.decode() or "{}")
|
|
except (json.JSONDecodeError, UnicodeDecodeError):
|
|
return {}
|
|
return data if isinstance(data, dict) else {}
|
|
# Form-encoded (SMTP2GO default)
|
|
return {key: request.POST.get(key) for key in request.POST.keys()}
|
|
|
|
|
|
def extract_monica_message_id(payload: dict[str, Any]) -> str:
|
|
"""Pull our correlation id from flat keys or a nested headers object."""
|
|
for key in _MONICA_HEADER_KEYS:
|
|
value = _as_str(payload.get(key))
|
|
if value:
|
|
return value
|
|
|
|
headers = payload.get("headers") or payload.get("email_headers") or {}
|
|
if isinstance(headers, dict):
|
|
for key in _MONICA_HEADER_KEYS:
|
|
value = _as_str(headers.get(key))
|
|
if value:
|
|
return value
|
|
# Case-insensitive scan
|
|
lower_map = {str(k).lower(): v for k, v in headers.items()}
|
|
for key in _MONICA_HEADER_KEYS:
|
|
value = _as_str(lower_map.get(key.lower()))
|
|
if value:
|
|
return value
|
|
elif isinstance(headers, list):
|
|
# Some ESP shapes send [["X-Monica-Message-Id", "..."], ...]
|
|
for item in headers:
|
|
if isinstance(item, (list, tuple)) and len(item) >= 2:
|
|
if _as_str(item[0]).lower() in {
|
|
k.lower() for k in _MONICA_HEADER_KEYS
|
|
}:
|
|
value = _as_str(item[1])
|
|
if value:
|
|
return value
|
|
elif isinstance(item, str) and ":" in item:
|
|
name, _, rest = item.partition(":")
|
|
if name.strip().lower() in {k.lower() for k in _MONICA_HEADER_KEYS}:
|
|
value = rest.strip()
|
|
if value:
|
|
return value
|
|
return ""
|
|
|
|
|
|
def _message_by_pk(pk: str) -> Message | None:
|
|
"""Lookup Message by UUID pk without raising on malformed ids."""
|
|
try:
|
|
uuid.UUID(str(pk))
|
|
except (ValueError, AttributeError, TypeError):
|
|
return None
|
|
try:
|
|
return (
|
|
Message.objects.select_related("contact", "campaign")
|
|
.filter(pk=pk)
|
|
.first()
|
|
)
|
|
except (ValidationError, ValueError):
|
|
return None
|
|
|
|
|
|
def find_message_for_email_event(payload: dict[str, Any]) -> Message | None:
|
|
monica_id = extract_monica_message_id(payload)
|
|
if monica_id:
|
|
message = _message_by_pk(monica_id)
|
|
if message:
|
|
return message
|
|
|
|
email_id = _as_str(payload.get("email_id") or payload.get("email-id"))
|
|
if email_id:
|
|
message = (
|
|
Message.objects.select_related("contact", "campaign")
|
|
.filter(provider_message_id=email_id)
|
|
.first()
|
|
)
|
|
if message:
|
|
return message
|
|
|
|
rcpt = _as_str(payload.get("rcpt")).lower()
|
|
if not rcpt:
|
|
recipients = payload.get("recipients")
|
|
if isinstance(recipients, str) and recipients.strip():
|
|
rcpt = recipients.split(",")[0].strip().lower()
|
|
elif isinstance(recipients, list) and recipients:
|
|
rcpt = _as_str(recipients[0]).lower()
|
|
|
|
if not rcpt:
|
|
return None
|
|
|
|
contact = Contact.objects.filter(email__iexact=rcpt).first()
|
|
if not contact:
|
|
return None
|
|
|
|
return (
|
|
Message.objects.select_related("contact", "campaign")
|
|
.filter(
|
|
contact=contact,
|
|
channel=Channel.EMAIL,
|
|
status__in=[
|
|
Message.Status.QUEUED,
|
|
Message.Status.SENT,
|
|
Message.Status.DELIVERED,
|
|
Message.Status.OPENED,
|
|
Message.Status.CLICKED,
|
|
Message.Status.FAILED,
|
|
Message.Status.BOUNCED,
|
|
],
|
|
)
|
|
.order_by("-sent_at", "-updated_at")
|
|
.first()
|
|
)
|
|
|
|
|
|
def _maybe_upgrade_status(message: Message, new_status: str, *, error: str = "") -> None:
|
|
current_rank = _STATUS_RANK.get(message.status, 0)
|
|
new_rank = _STATUS_RANK.get(new_status, 0)
|
|
# Always allow bounce/suppress to overwrite delivered; allow delivered over sent.
|
|
if new_rank < current_rank and new_status not in {
|
|
Message.Status.BOUNCED,
|
|
Message.Status.SUPPRESSED,
|
|
Message.Status.FAILED,
|
|
}:
|
|
return
|
|
if message.status in {Message.Status.BOUNCED, Message.Status.SUPPRESSED} and new_status in {
|
|
Message.Status.SENT,
|
|
Message.Status.DELIVERED,
|
|
Message.Status.OPENED,
|
|
Message.Status.CLICKED,
|
|
}:
|
|
return
|
|
|
|
fields = ["status", "updated_at"]
|
|
message.status = new_status
|
|
if error:
|
|
message.error = error[:2000]
|
|
fields.append("error")
|
|
elif new_status in _ENGAGED_OR_DELIVERED:
|
|
message.error = ""
|
|
fields.append("error")
|
|
message.save(update_fields=fields)
|
|
|
|
|
|
def _apply_email_event(message: Message, event: str, payload: dict[str, Any]) -> None:
|
|
event = _normalize_email_event(event)
|
|
bounce_kind = _as_str(payload.get("bounce")).lower()
|
|
err = _as_str(payload.get("message") or payload.get("context"))
|
|
|
|
email_id = _as_str(payload.get("email_id") or payload.get("email-id"))
|
|
if email_id and message.provider_message_id != email_id:
|
|
message.provider_message_id = email_id
|
|
message.provider = PROVIDER_EMAIL
|
|
message.save(
|
|
update_fields=["provider_message_id", "provider", "updated_at"]
|
|
)
|
|
|
|
if event == "processed":
|
|
if message.status in {Message.Status.QUEUED, Message.Status.DRAFT}:
|
|
_maybe_upgrade_status(message, Message.Status.SENT)
|
|
return
|
|
|
|
if event == "delivered":
|
|
_maybe_upgrade_status(message, Message.Status.DELIVERED)
|
|
return
|
|
|
|
if event == "bounce":
|
|
status = Message.Status.BOUNCED
|
|
_maybe_upgrade_status(
|
|
message,
|
|
status,
|
|
error=err or f"{bounce_kind or 'unknown'} bounce",
|
|
)
|
|
if bounce_kind == "hard":
|
|
set_channel_consent(
|
|
message.contact,
|
|
Channel.EMAIL,
|
|
opted_in=False,
|
|
reason="smtp2go_hard_bounce",
|
|
)
|
|
return
|
|
|
|
if event == "reject":
|
|
_maybe_upgrade_status(
|
|
message, Message.Status.FAILED, error=err or "rejected by provider"
|
|
)
|
|
return
|
|
|
|
if event == "spam":
|
|
_maybe_upgrade_status(
|
|
message, Message.Status.SUPPRESSED, error=err or "spam complaint"
|
|
)
|
|
set_channel_consent(
|
|
message.contact,
|
|
Channel.EMAIL,
|
|
opted_in=False,
|
|
reason="smtp2go_spam",
|
|
)
|
|
return
|
|
|
|
if event == "unsubscribe":
|
|
_maybe_upgrade_status(
|
|
message, Message.Status.SUPPRESSED, error="provider unsubscribe"
|
|
)
|
|
set_channel_consent(
|
|
message.contact,
|
|
Channel.EMAIL,
|
|
opted_in=False,
|
|
reason="smtp2go_unsubscribe",
|
|
)
|
|
return
|
|
|
|
if event == "open":
|
|
# Open implies delivery; do not overwrite a stronger click status.
|
|
if message.status != Message.Status.CLICKED:
|
|
_maybe_upgrade_status(message, Message.Status.OPENED)
|
|
return
|
|
|
|
if event == "click":
|
|
_maybe_upgrade_status(message, Message.Status.CLICKED)
|
|
return
|
|
|
|
# resubscribe — event row only (status unchanged)
|
|
|
|
|
|
def process_smtp2go_email_webhook(payload: dict[str, Any]) -> ProviderEvent | None:
|
|
"""
|
|
Persist ProviderEvent and update Message delivery status when possible.
|
|
|
|
Returns the stored event (even if message could not be matched).
|
|
"""
|
|
event = _normalize_email_event(_as_str(payload.get("event")))
|
|
if not event:
|
|
logger.warning("SMTP2GO webhook missing event: %s", payload)
|
|
return None
|
|
|
|
message = find_message_for_email_event(payload)
|
|
if message:
|
|
_apply_email_event(message, event, payload)
|
|
message.refresh_from_db()
|
|
else:
|
|
logger.info(
|
|
"SMTP2GO webhook unmatched event=%s rcpt=%s email_id=%s",
|
|
event,
|
|
payload.get("rcpt"),
|
|
payload.get("email_id"),
|
|
)
|
|
|
|
return ProviderEvent.objects.create(
|
|
message=message,
|
|
provider=PROVIDER_EMAIL,
|
|
event_type=event[:64],
|
|
payload=_json_safe(payload) if isinstance(payload, dict) else {},
|
|
)
|
|
|
|
|
|
def normalize_phone(value: str) -> str:
|
|
return "".join(ch for ch in (value or "") if ch.isdigit())
|
|
|
|
|
|
def _sms_provider_message_id(payload: dict[str, Any]) -> str:
|
|
"""SMS unique id from docs (`message_id`). Never use webhook `id`."""
|
|
return _as_str(payload.get("message_id") or payload.get("sms_id"))
|
|
|
|
|
|
def find_message_for_sms_event(payload: dict[str, Any]) -> Message | None:
|
|
provider_id = _sms_provider_message_id(payload)
|
|
if provider_id:
|
|
message = (
|
|
Message.objects.select_related("contact", "campaign")
|
|
.filter(channel=Channel.SMS, provider_message_id=provider_id)
|
|
.first()
|
|
)
|
|
if message:
|
|
return message
|
|
|
|
# Outbound delivery events use destination_number (recipient).
|
|
# Do not use source_number / from — those are the pool or inbound reply.
|
|
raw_phone = (
|
|
payload.get("destination_number")
|
|
or payload.get("to")
|
|
or payload.get("phone")
|
|
or ""
|
|
)
|
|
digits = normalize_phone(_as_str(raw_phone))
|
|
if len(digits) < 7:
|
|
return None
|
|
|
|
# Match last 10 digits so +1 / formatting differences still hit.
|
|
tail = digits[-10:]
|
|
contacts = Contact.objects.exclude(phone="").only("id", "phone")
|
|
contact = None
|
|
for row in contacts.iterator():
|
|
if normalize_phone(row.phone).endswith(tail):
|
|
contact = row
|
|
break
|
|
if not contact:
|
|
return None
|
|
|
|
return (
|
|
Message.objects.select_related("contact", "campaign")
|
|
.filter(
|
|
contact=contact,
|
|
channel=Channel.SMS,
|
|
status__in=[
|
|
Message.Status.QUEUED,
|
|
Message.Status.SENT,
|
|
Message.Status.DELIVERED,
|
|
Message.Status.FAILED,
|
|
Message.Status.SUPPRESSED,
|
|
],
|
|
)
|
|
.order_by("-sent_at", "-updated_at")
|
|
.first()
|
|
)
|
|
|
|
|
|
def _apply_sms_event(message: Message, event: str, payload: dict[str, Any]) -> None:
|
|
event = _normalize_sms_event(event)
|
|
err = _as_str(
|
|
payload.get("message")
|
|
or payload.get("status_code")
|
|
or payload.get("context")
|
|
)
|
|
|
|
provider_id = _sms_provider_message_id(payload)
|
|
if provider_id and message.provider_message_id != provider_id:
|
|
message.provider_message_id = provider_id
|
|
message.provider = PROVIDER_SMS
|
|
message.save(
|
|
update_fields=["provider_message_id", "provider", "updated_at"]
|
|
)
|
|
|
|
if event in {"sms_sending", "sms_submitted"}:
|
|
if message.status in {Message.Status.QUEUED, Message.Status.DRAFT}:
|
|
_maybe_upgrade_status(message, Message.Status.SENT)
|
|
return
|
|
|
|
if event == "sms_delivered":
|
|
_maybe_upgrade_status(message, Message.Status.DELIVERED)
|
|
return
|
|
|
|
if event in {"sms_failed", "sms_rejected"}:
|
|
_maybe_upgrade_status(
|
|
message,
|
|
Message.Status.FAILED,
|
|
error=err or event,
|
|
)
|
|
return
|
|
|
|
if event in _SMS_OPT_OUT_EVENTS:
|
|
_maybe_upgrade_status(
|
|
message, Message.Status.SUPPRESSED, error="sms opt-out"
|
|
)
|
|
set_channel_consent(
|
|
message.contact,
|
|
Channel.SMS,
|
|
opted_in=False,
|
|
reason="smtp2go_sms_opt_out",
|
|
)
|
|
return
|
|
|
|
|
|
def process_smtp2go_sms_webhook(payload: dict[str, Any]) -> ProviderEvent | None:
|
|
"""Persist SMS delivery/opt-out ProviderEvent and update Message when matched."""
|
|
event = _normalize_sms_event(_as_str(payload.get("event")))
|
|
if not event:
|
|
logger.warning("SMTP2GO SMS webhook missing event: %s", payload)
|
|
return None
|
|
|
|
message = find_message_for_sms_event(payload)
|
|
if message:
|
|
_apply_sms_event(message, event, payload)
|
|
message.refresh_from_db()
|
|
else:
|
|
# Opt-out with no matched campaign message still suppresses by phone.
|
|
if event in _SMS_OPT_OUT_EVENTS:
|
|
phone = _as_str(
|
|
payload.get("destination_number")
|
|
or payload.get("from")
|
|
or payload.get("source_number")
|
|
)
|
|
if phone:
|
|
from directmail.services import record_sms_stop
|
|
|
|
record_sms_stop(phone)
|
|
logger.info(
|
|
"SMTP2GO SMS webhook unmatched event=%s phone=%s message_id=%s",
|
|
event,
|
|
payload.get("destination_number"),
|
|
payload.get("message_id"),
|
|
)
|
|
|
|
return ProviderEvent.objects.create(
|
|
message=message,
|
|
provider=PROVIDER_SMS,
|
|
event_type=event[:64],
|
|
payload=_json_safe(payload) if isinstance(payload, dict) else {},
|
|
)
|
|
|
|
|
|
def is_inbound_sms_stop(payload: dict[str, Any]) -> bool:
|
|
"""True for gateway-style inbound reply payloads (STOP / UNSUBSCRIBE).
|
|
|
|
SMTP2GO auto-handles STOP/UNSUB/UNSUBSCRIBE replies; this catches the
|
|
inbound gateway POST shape (no ``event`` field) when configured to
|
|
forward replies. Prefer this over relying on a webhook Opt-Out event —
|
|
the API ``sms_events`` list does not include opt-out.
|
|
"""
|
|
if payload.get("event"):
|
|
return False
|
|
text = _as_str(
|
|
payload.get("text")
|
|
or payload.get("message")
|
|
or payload.get("message_content")
|
|
).upper()
|
|
return text in {"STOP", "UNSUBSCRIBE", "UNSUB", "CANCEL", "END", "QUIT"}
|
|
|
|
|
|
def classify_smtp2go_payload(payload: dict[str, Any]) -> str:
|
|
"""
|
|
Decide email vs SMS vs inbound STOP for a unified SMTP2GO webhook URL.
|
|
|
|
Returns one of: ``sms_inbound``, ``sms``, ``email``, ``unknown``.
|
|
"""
|
|
if not isinstance(payload, dict) or not payload:
|
|
return "unknown"
|
|
|
|
if is_inbound_sms_stop(payload):
|
|
return "sms_inbound"
|
|
|
|
raw = _as_str(payload.get("event")).lower().replace("-", "_").replace(" ", "_")
|
|
if not raw:
|
|
return "unknown"
|
|
|
|
# Explicit SMS forms (docs sms_* + API/UI short names that are SMS-only).
|
|
if raw.startswith("sms_") or raw in {
|
|
"submitted",
|
|
"sending",
|
|
"opt_out",
|
|
"optout",
|
|
}:
|
|
return "sms"
|
|
|
|
email_event = _normalize_email_event(raw)
|
|
if email_event in {
|
|
"processed",
|
|
"open",
|
|
"click",
|
|
"bounce",
|
|
"spam",
|
|
"unsubscribe",
|
|
"resubscribe",
|
|
}:
|
|
return "email"
|
|
|
|
has_dest = bool(_as_str(payload.get("destination_number")))
|
|
has_rcpt = bool(_as_str(payload.get("rcpt")))
|
|
has_email_id = bool(
|
|
_as_str(payload.get("email_id") or payload.get("email-id"))
|
|
)
|
|
has_monica = bool(extract_monica_message_id(payload))
|
|
has_from_address = bool(_as_str(payload.get("from_address")))
|
|
has_sms_id = bool(_as_str(payload.get("message_id") or payload.get("sms_id")))
|
|
has_sms_body = bool(
|
|
_as_str(payload.get("message_content") or payload.get("source_number"))
|
|
)
|
|
|
|
email_leaning = has_rcpt or has_email_id or has_monica or has_from_address
|
|
sms_leaning = has_dest or has_sms_body or (
|
|
has_sms_id and not email_leaning
|
|
)
|
|
|
|
# Ambiguous short names shared by email + SMS API lists.
|
|
if raw in {"delivered", "failed", "rejected", "reject"}:
|
|
if sms_leaning and not email_leaning:
|
|
return "sms"
|
|
if email_leaning:
|
|
return "email"
|
|
# Bare reject without fields → email docs name; rejected alone → sms lean default
|
|
if raw == "reject":
|
|
return "email"
|
|
if raw == "rejected":
|
|
return "sms"
|
|
return "email"
|
|
|
|
if email_leaning:
|
|
return "email"
|
|
if sms_leaning:
|
|
return "sms"
|
|
# Default: email (historical primary SMTP2GO traffic).
|
|
return "email"
|
|
|
|
|
|
def _pcm_event_type(payload: dict[str, Any]) -> str:
|
|
for key in ("event", "eventType", "event_type", "type", "status"):
|
|
value = payload.get(key)
|
|
if value:
|
|
return str(value).strip()
|
|
return "unknown"
|
|
|
|
|
|
def find_message_for_pcm_event(payload: dict[str, Any]) -> Message | None:
|
|
"""Correlate PCM webhook to Message via extRefNbr or orderID."""
|
|
ext = (
|
|
payload.get("extRefNbr")
|
|
or payload.get("ext_ref_nbr")
|
|
or payload.get("externalReference")
|
|
or ""
|
|
)
|
|
if not ext and isinstance(payload.get("recipient"), dict):
|
|
ext = payload["recipient"].get("extRefNbr") or ""
|
|
ext = str(ext).strip()
|
|
if ext:
|
|
message = _message_by_pk(ext)
|
|
if message:
|
|
return message
|
|
|
|
order_id = (
|
|
payload.get("orderID")
|
|
or payload.get("orderId")
|
|
or payload.get("order_id")
|
|
or ""
|
|
)
|
|
order_id = str(order_id).strip()
|
|
if order_id:
|
|
message = (
|
|
Message.objects.select_related("contact", "campaign")
|
|
.filter(provider_message_id=order_id, channel=Channel.POSTCARD)
|
|
.first()
|
|
)
|
|
if message:
|
|
return message
|
|
return None
|
|
|
|
|
|
def _apply_pcm_status(message: Message, status: str, payload: dict[str, Any]) -> None:
|
|
status_norm = (status or "").strip().lower()
|
|
err = (
|
|
payload.get("message")
|
|
or payload.get("error")
|
|
or payload.get("reason")
|
|
or ""
|
|
)
|
|
err = str(err).strip()
|
|
|
|
order_id = (
|
|
payload.get("orderID")
|
|
or payload.get("orderId")
|
|
or payload.get("order_id")
|
|
or ""
|
|
)
|
|
if order_id and message.provider_message_id != str(order_id):
|
|
message.provider_message_id = str(order_id)
|
|
message.provider = PROVIDER_PCM
|
|
message.save(
|
|
update_fields=["provider_message_id", "provider", "updated_at"]
|
|
)
|
|
|
|
if status_norm in {"delivered"}:
|
|
_maybe_upgrade_status(message, Message.Status.DELIVERED)
|
|
return
|
|
if status_norm in {"undeliverable", "returned"}:
|
|
_maybe_upgrade_status(
|
|
message,
|
|
Message.Status.BOUNCED,
|
|
error=err or "undeliverable",
|
|
)
|
|
return
|
|
if status_norm in {"canceled", "cancelled"}:
|
|
_maybe_upgrade_status(
|
|
message, Message.Status.FAILED, error=err or "canceled"
|
|
)
|
|
return
|
|
if status_norm in {"pending", "processing", "processed", "mailed", "intransit", "in_transit"}:
|
|
if message.status in {
|
|
Message.Status.QUEUED,
|
|
Message.Status.DRAFT,
|
|
Message.Status.SCHEDULED,
|
|
}:
|
|
_maybe_upgrade_status(message, Message.Status.SENT)
|
|
return
|
|
|
|
|
|
def process_pcm_postcard_webhook(payload: dict[str, Any]) -> ProviderEvent | None:
|
|
"""Record a PCM Integrations postcard event and advance Message status."""
|
|
if not payload:
|
|
return None
|
|
|
|
# Nested data wrappers some webhook UIs use.
|
|
if "data" in payload and isinstance(payload["data"], dict):
|
|
inner = dict(payload["data"])
|
|
for key in ("event", "eventType", "type"):
|
|
if key in payload and key not in inner:
|
|
inner[key] = payload[key]
|
|
payload = inner
|
|
|
|
event_type = _pcm_event_type(payload)
|
|
message = find_message_for_pcm_event(payload)
|
|
if message:
|
|
status_for_apply = (
|
|
payload.get("status")
|
|
or payload.get("orderStatus")
|
|
or event_type
|
|
)
|
|
_apply_pcm_status(message, str(status_for_apply), payload)
|
|
|
|
return ProviderEvent.objects.create(
|
|
message=message,
|
|
provider=PROVIDER_PCM,
|
|
event_type=event_type[:64],
|
|
payload=payload,
|
|
)
|
|
|
|
|
|
def _engagement_chart_bars(metrics: list[tuple[str, int]]) -> list[dict]:
|
|
"""Build bar heights (percent) for the campaign engagement chart."""
|
|
peak = max((value for _, value in metrics), default=0)
|
|
bars: list[dict] = []
|
|
for label, value in metrics:
|
|
if peak <= 0:
|
|
pct = 12 if value == 0 else 100
|
|
else:
|
|
pct = max(12, int(round((value / peak) * 100))) if value else 8
|
|
bars.append({"label": label, "value": value, "pct": pct})
|
|
return bars
|
|
|
|
|
|
def campaign_engagement_stats(campaign) -> dict:
|
|
"""Aggregate delivery + open/click counts for the campaign report."""
|
|
messages_qs = campaign.messages.all()
|
|
statuses = list(messages_qs.values_list("status", flat=True))
|
|
message_ids = list(messages_qs.values_list("pk", flat=True))
|
|
|
|
events = ProviderEvent.objects.filter(message_id__in=message_ids)
|
|
open_message_ids = set(
|
|
events.filter(event_type__iexact="open").values_list("message_id", flat=True)
|
|
)
|
|
click_message_ids = set(
|
|
events.filter(event_type__iexact="click").values_list("message_id", flat=True)
|
|
)
|
|
|
|
# Status-based engagement also counts (webhook may set opened/clicked).
|
|
status_opened = sum(1 for s in statuses if s in {"opened", "clicked"})
|
|
status_clicked = sum(1 for s in statuses if s == "clicked")
|
|
opens = max(len(open_message_ids), status_opened)
|
|
clicks = max(len(click_message_ids), status_clicked)
|
|
|
|
sent = sum(1 for s in statuses if s in _SENT_OR_BETTER)
|
|
delivered = sum(1 for s in statuses if s in _ENGAGED_OR_DELIVERED)
|
|
failed = sum(1 for s in statuses if s in {"failed", "bounced"})
|
|
suppressed = sum(1 for s in statuses if s == "suppressed")
|
|
|
|
if campaign.channel == Channel.EMAIL:
|
|
chart_metrics = [
|
|
("Sent", sent),
|
|
("Delivered", delivered),
|
|
("Opens", opens),
|
|
("Clicks", clicks),
|
|
("Failed", failed),
|
|
]
|
|
else:
|
|
chart_metrics = [
|
|
("Sent", sent),
|
|
("Delivered", delivered),
|
|
("Failed", failed),
|
|
("Suppressed", suppressed),
|
|
]
|
|
|
|
return {
|
|
"total": len(statuses),
|
|
"sent": sent,
|
|
"delivered": delivered,
|
|
"failed": failed,
|
|
"bounced": sum(1 for s in statuses if s == "bounced"),
|
|
"suppressed": suppressed,
|
|
"opens": opens,
|
|
"clicks": clicks,
|
|
"open_events": events.filter(event_type__iexact="open").count(),
|
|
"click_events": events.filter(event_type__iexact="click").count(),
|
|
"chart_bars": _engagement_chart_bars(chart_metrics),
|
|
}
|