218 lines
8.1 KiB
Python
218 lines
8.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, WhatsAppPhoneAsset
|
|
from .whatsapp_config import resolved_whatsapp_phone_number_id
|
|
|
|
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,
|
|
"contact_name": message.contact_name,
|
|
"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
|
|
|
|
platform_number = business_phone()
|
|
platform_phone_id = resolved_whatsapp_phone_number_id(db)
|
|
created = 0
|
|
for entry in payload.get("entry", []):
|
|
for change in entry.get("changes", []):
|
|
if change.get("field") != "messages":
|
|
continue
|
|
value = change.get("value", {})
|
|
metadata = value.get("metadata") if isinstance(value.get("metadata"), dict) else {}
|
|
phone_number_id = str(metadata.get("phone_number_id") or "") or None
|
|
owner_id = None
|
|
line_number = platform_number
|
|
if phone_number_id and phone_number_id != platform_phone_id:
|
|
asset = db.scalar(
|
|
select(WhatsAppPhoneAsset).where(WhatsAppPhoneAsset.phone_number_id == phone_number_id)
|
|
)
|
|
if asset is None:
|
|
continue
|
|
owner_id = asset.user_id
|
|
raw_line = asset.display_phone or str(metadata.get("display_phone_number") or "")
|
|
try:
|
|
line_number = normalize_wa_number(raw_line)
|
|
except ValueError:
|
|
line_number = re.sub(r"\D", "", raw_line) or platform_number
|
|
names: dict[str, str] = {}
|
|
for contact in value.get("contacts") or []:
|
|
if not isinstance(contact, dict):
|
|
continue
|
|
wa_id = str(contact.get("wa_id") or "")
|
|
profile = contact.get("profile") if isinstance(contact.get("profile"), dict) else {}
|
|
name = profile.get("name")
|
|
if wa_id and isinstance(name, str) and name.strip():
|
|
names[wa_id] = name.strip()
|
|
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=line_number,
|
|
body=body,
|
|
message_type=message.get("type") or "text",
|
|
status="received",
|
|
payload=message,
|
|
user_id=owner_id,
|
|
phone_number_id=phone_number_id if owner_id else None,
|
|
contact_name=names.get(from_number),
|
|
)
|
|
)
|
|
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=line_number,
|
|
to_number=status.get("recipient_id") or "",
|
|
body=None,
|
|
message_type="status",
|
|
status=status.get("status") or "unknown",
|
|
payload=status,
|
|
user_id=owner_id,
|
|
phone_number_id=phone_number_id if owner_id else None,
|
|
)
|
|
)
|
|
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,
|
|
from_number: str | None = None,
|
|
user_id: Any = None,
|
|
phone_number_id: str | None = None,
|
|
) -> WhatsAppMessage:
|
|
message = WhatsAppMessage(
|
|
direction="outbound",
|
|
wa_message_id=wa_message_id,
|
|
from_number=from_number or business_phone(),
|
|
to_number=to_number,
|
|
body=body,
|
|
message_type=message_type,
|
|
status=status,
|
|
payload=payload,
|
|
user_id=user_id,
|
|
phone_number_id=phone_number_id,
|
|
)
|
|
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
|