180 lines
6.1 KiB
Python
180 lines
6.1 KiB
Python
import re
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .config import settings
|
|
from .models import WhatsAppMessage
|
|
|
|
PHONE_DIGITS_RE = re.compile(r"^\d{10,15}$")
|
|
|
|
|
|
def normalize_wa_number(value: str) -> str:
|
|
digits = re.sub(r"\D", "", value.strip())
|
|
if len(digits) == 10:
|
|
digits = f"91{digits}"
|
|
if not PHONE_DIGITS_RE.match(digits):
|
|
raise ValueError("Invalid WhatsApp number")
|
|
return digits
|
|
|
|
|
|
def business_phone() -> str:
|
|
return normalize_wa_number(settings.whatsapp_business_phone)
|
|
|
|
|
|
def _media_from_payload(message_type: str, payload: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
if not payload or message_type not in {"image", "audio", "video", "document", "sticker"}:
|
|
return None
|
|
media = payload.get(message_type) or {}
|
|
if not isinstance(media, dict):
|
|
return None
|
|
return {
|
|
"id": media.get("id"),
|
|
"mime_type": media.get("mime_type"),
|
|
"caption": media.get("caption"),
|
|
"filename": media.get("filename"),
|
|
}
|
|
|
|
|
|
def serialize_message(message: WhatsAppMessage) -> dict:
|
|
payload = message.payload if isinstance(message.payload, dict) else None
|
|
media = _media_from_payload(message.message_type, payload)
|
|
return {
|
|
"id": str(message.id),
|
|
"direction": message.direction,
|
|
"wa_message_id": message.wa_message_id,
|
|
"from": message.from_number,
|
|
"to": message.to_number,
|
|
"body": message.body,
|
|
"message_type": message.message_type,
|
|
"status": message.status,
|
|
"media": media,
|
|
"created_at": message.created_at,
|
|
}
|
|
|
|
|
|
def _extract_text_body(message: dict[str, Any]) -> str | None:
|
|
msg_type = message.get("type")
|
|
if msg_type == "text":
|
|
return (message.get("text") or {}).get("body")
|
|
if msg_type == "button":
|
|
return (message.get("button") or {}).get("text")
|
|
if msg_type == "interactive":
|
|
interactive = message.get("interactive") or {}
|
|
if interactive.get("type") == "button_reply":
|
|
return (interactive.get("button_reply") or {}).get("title")
|
|
if interactive.get("type") == "list_reply":
|
|
return (interactive.get("list_reply") or {}).get("title")
|
|
if msg_type in {"image", "video", "document"}:
|
|
return (message.get(msg_type) or {}).get("caption")
|
|
if msg_type == "reaction":
|
|
reaction = message.get("reaction") or {}
|
|
emoji = reaction.get("emoji")
|
|
return f"Reacted {emoji}" if emoji else "Reaction"
|
|
if msg_type == "location":
|
|
location = message.get("location") or {}
|
|
name = location.get("name") or location.get("address")
|
|
if name:
|
|
return f"Location: {name}"
|
|
lat = location.get("latitude")
|
|
lng = location.get("longitude")
|
|
if lat is not None and lng is not None:
|
|
return f"Location: {lat}, {lng}"
|
|
if msg_type == "contacts":
|
|
contacts = message.get("contacts") or []
|
|
if contacts:
|
|
name = ((contacts[0].get("name") or {}).get("formatted_name")) or "Contact"
|
|
return f"Contact: {name}"
|
|
return None
|
|
|
|
|
|
def record_inbound_messages(db: Session, payload: dict[str, Any]) -> int:
|
|
if payload.get("object") != "whatsapp_business_account":
|
|
return 0
|
|
|
|
business = business_phone()
|
|
created = 0
|
|
for entry in payload.get("entry", []):
|
|
for change in entry.get("changes", []):
|
|
if change.get("field") != "messages":
|
|
continue
|
|
value = change.get("value", {})
|
|
for message in value.get("messages", []):
|
|
from_number = message.get("from") or ""
|
|
body = _extract_text_body(message)
|
|
db.add(
|
|
WhatsAppMessage(
|
|
direction="inbound",
|
|
wa_message_id=message.get("id"),
|
|
from_number=from_number,
|
|
to_number=business,
|
|
body=body,
|
|
message_type=message.get("type") or "text",
|
|
status="received",
|
|
payload=message,
|
|
)
|
|
)
|
|
created += 1
|
|
for status in value.get("statuses", []):
|
|
wa_id = status.get("id")
|
|
if not wa_id:
|
|
continue
|
|
existing = db.scalar(
|
|
select(WhatsAppMessage).where(WhatsAppMessage.wa_message_id == wa_id)
|
|
)
|
|
if existing:
|
|
existing.status = status.get("status") or existing.status
|
|
else:
|
|
db.add(
|
|
WhatsAppMessage(
|
|
direction="outbound",
|
|
wa_message_id=wa_id,
|
|
from_number=business,
|
|
to_number=status.get("recipient_id") or "",
|
|
body=None,
|
|
message_type="status",
|
|
status=status.get("status") or "unknown",
|
|
payload=status,
|
|
)
|
|
)
|
|
created += 1
|
|
if created:
|
|
db.commit()
|
|
return created
|
|
|
|
|
|
def record_outbound_message(
|
|
db: Session,
|
|
*,
|
|
to_number: str,
|
|
body: str | None,
|
|
message_type: str = "text",
|
|
wa_message_id: str | None,
|
|
status: str = "sent",
|
|
payload: dict | None = None,
|
|
) -> WhatsAppMessage:
|
|
message = WhatsAppMessage(
|
|
direction="outbound",
|
|
wa_message_id=wa_message_id,
|
|
from_number=business_phone(),
|
|
to_number=to_number,
|
|
body=body,
|
|
message_type=message_type,
|
|
status=status,
|
|
payload=payload,
|
|
)
|
|
db.add(message)
|
|
db.commit()
|
|
db.refresh(message)
|
|
return message
|
|
|
|
|
|
def extract_wa_message_id(result: dict[str, Any] | None) -> str | None:
|
|
if not isinstance(result, dict):
|
|
return None
|
|
messages = result.get("messages")
|
|
if isinstance(messages, list) and messages:
|
|
message_id = messages[0].get("id")
|
|
return message_id if isinstance(message_id, str) else None
|
|
return None
|