File size: 3,411 Bytes
102dd4f dc8cb49 102dd4f d4f6fe7 1cd53ba 102dd4f 0467de1 dc8cb49 102dd4f d4f6fe7 102dd4f dc8cb49 0467de1 1cd53ba dc8cb49 102dd4f d4f6fe7 dc8cb49 102dd4f d4f6fe7 102dd4f dc8cb49 0467de1 1cd53ba 102dd4f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | from typing import Any
from app.models.domain import WhatsAppInboundMessage
SUPPORTED_MESSAGE_TYPES = {"text", "interactive"}
def parse_inbound_messages(payload: dict[str, Any]) -> list[WhatsAppInboundMessage]:
messages: list[WhatsAppInboundMessage] = []
for entry in payload.get("entry", []):
for change in entry.get("changes", []):
value = change.get("value", {})
contacts_by_wa_id = {
contact.get("wa_id"): contact.get("profile", {}).get("name")
for contact in value.get("contacts", [])
}
phone_number_id = value.get("metadata", {}).get("phone_number_id")
for message in value.get("messages", []):
message_type = message.get("type")
if message_type not in SUPPORTED_MESSAGE_TYPES:
continue
remoteJid = message.get("from")
is_group = isinstance(remoteJid, str) and remoteJid.endswith("@g.us")
message_id = message.get("id")
phone_number = message.get("remote_jid_alt")
context_message_id = message.get("context", {}).get("id")
text = message.get("text", {}).get("body")
if not remoteJid or not message_id or text is None:
continue
if message_type == "text":
text = message.get("text", {}).get("body")
if text is None:
continue
messages.append(
WhatsAppInboundMessage(
message_id=message_id,
remoteJid=remoteJid,
text=text,
timestamp=message.get("timestamp"),
profile_name=contacts_by_wa_id.get(remoteJid),
phone_number_id=phone_number_id,
message_type="text",
context_message_id=context_message_id,
phone_number=phone_number,
is_group=is_group,
raw=message,
)
)
continue
interactive = message.get("interactive", {})
if interactive.get("type") != "list_reply":
continue
list_reply = interactive.get("list_reply", {})
reply_id = list_reply.get("id")
if not reply_id:
continue
title = list_reply.get("title") or reply_id
messages.append(
WhatsAppInboundMessage(
message_id=message_id,
remoteJid=remoteJid,
text=title,
timestamp=message.get("timestamp"),
profile_name=contacts_by_wa_id.get(remoteJid),
phone_number_id=phone_number_id,
message_type="interactive",
interactive_reply_id=reply_id,
context_message_id=context_message_id,
phone_number=phone_number,
is_group=is_group,
raw=message,
)
)
return messages
|