meta whatsapp chat
This commit is contained in:
parent
54f75387eb
commit
5ced0d0812
10 changed files with 1015 additions and 25 deletions
|
|
@ -8,6 +8,7 @@ addressed by another tenant.
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import secrets
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
|
@ -22,11 +23,14 @@ from .models import (
|
|||
UserAccount,
|
||||
WhatsAppBusinessAccount,
|
||||
WhatsAppConnectionAudit,
|
||||
WhatsAppMessage,
|
||||
WhatsAppMetaConnection,
|
||||
WhatsAppPhoneAsset,
|
||||
WhatsAppWebhookEvent,
|
||||
)
|
||||
from .whatsapp_config import resolved_whatsapp_app_secret
|
||||
from .whatsapp_graph_client import WhatsAppGraphClient, WhatsAppGraphError
|
||||
from .whatsapp_messages import extract_wa_message_id, normalize_wa_number, record_outbound_message, serialize_message
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -199,6 +203,7 @@ def _owned_account(db: Session, user: UserAccount, account_id: Any) -> WhatsAppB
|
|||
|
||||
|
||||
def _serialize_phone(phone: WhatsAppPhoneAsset) -> dict[str, Any]:
|
||||
verified = (phone.code_verification_status or "").upper()
|
||||
return {
|
||||
"id": str(phone.id),
|
||||
"phone_number_id": phone.phone_number_id,
|
||||
|
|
@ -207,6 +212,8 @@ def _serialize_phone(phone: WhatsAppPhoneAsset) -> dict[str, Any]:
|
|||
"quality": phone.quality,
|
||||
"status": phone.status,
|
||||
"code_verification_status": phone.code_verification_status,
|
||||
"needs_otp": phone.status != "connected" and verified in {"NOT_VERIFIED", "EXPIRED"},
|
||||
"needs_registration": phone.status != "connected",
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -344,6 +351,75 @@ def _apply_phones(
|
|||
row.code_verification_status = item.get("code_verification_status")
|
||||
|
||||
|
||||
def _pin_for_phone(phone: WhatsAppPhoneAsset, requested: str | None = None) -> str:
|
||||
pin = (requested or phone.two_step_pin or "").strip()
|
||||
if pin.isdigit() and len(pin) == 6:
|
||||
phone.two_step_pin = pin
|
||||
return pin
|
||||
pin = "".join(secrets.choice("0123456789") for _ in range(6))
|
||||
phone.two_step_pin = pin
|
||||
return pin
|
||||
|
||||
|
||||
async def _register_phone(token: str, phone: WhatsAppPhoneAsset, pin: str) -> str | None:
|
||||
try:
|
||||
await _graph(
|
||||
"POST",
|
||||
f"{phone.phone_number_id}/register",
|
||||
token=token,
|
||||
json_body={"messaging_product": "whatsapp", "pin": pin},
|
||||
)
|
||||
return None
|
||||
except HTTPException as exc:
|
||||
detail = str(exc.detail)
|
||||
lowered = detail.lower()
|
||||
if "already" in lowered and "register" in lowered:
|
||||
return None
|
||||
return detail[:500]
|
||||
|
||||
|
||||
def _set_account_health(db: Session, user: UserAccount, account: WhatsAppBusinessAccount) -> None:
|
||||
connected = any(phone.status == "connected" for phone in account.phones)
|
||||
otp_pending = any(
|
||||
phone.status != "connected"
|
||||
and (phone.code_verification_status or "").upper() in {"NOT_VERIFIED", "EXPIRED"}
|
||||
for phone in account.phones
|
||||
)
|
||||
if connected and account.management_status == "active":
|
||||
account.status = "connected"
|
||||
account.messaging_status = "active"
|
||||
account.last_error = None
|
||||
_audit(
|
||||
db,
|
||||
user,
|
||||
"connected",
|
||||
"WhatsApp Business connection verified",
|
||||
connection_id=account.connection_id,
|
||||
waba_id=account.waba_id,
|
||||
)
|
||||
return
|
||||
account.status = "incomplete"
|
||||
account.messaging_status = "unavailable"
|
||||
if otp_pending:
|
||||
account.last_error = (
|
||||
"This number still needs the SMS or voice code from Meta. "
|
||||
"It cannot stay logged into the WhatsApp or WhatsApp Business app while Cloud API uses it."
|
||||
)
|
||||
else:
|
||||
account.last_error = account.last_error or (
|
||||
"This number is attached to the WhatsApp Business Account, but Meta has not registered it for Cloud API yet. "
|
||||
"Enter the 6-digit two-step PIN if this number already has one, then register it."
|
||||
)
|
||||
_audit(
|
||||
db,
|
||||
user,
|
||||
"provision_incomplete",
|
||||
account.last_error,
|
||||
connection_id=account.connection_id,
|
||||
waba_id=account.waba_id,
|
||||
)
|
||||
|
||||
|
||||
async def _provision_waba(
|
||||
db: Session,
|
||||
user: UserAccount,
|
||||
|
|
@ -416,29 +492,31 @@ async def _provision_waba(
|
|||
account.last_error = str(exc.detail)[:500]
|
||||
_apply_phones(db, user, account, phones, session_phone_id)
|
||||
|
||||
for phone in account.phones:
|
||||
verified = (phone.code_verification_status or "").upper()
|
||||
if phone.status == "connected":
|
||||
continue
|
||||
if verified in {"NOT_VERIFIED", "EXPIRED"}:
|
||||
continue
|
||||
pin = _pin_for_phone(phone)
|
||||
register_error = await _register_phone(token, phone, pin)
|
||||
if register_error:
|
||||
account.last_error = register_error
|
||||
else:
|
||||
phone.status = "connected"
|
||||
|
||||
try:
|
||||
phones = await _load_phones(token, waba_id)
|
||||
_apply_phones(db, user, account, phones, session_phone_id)
|
||||
except HTTPException:
|
||||
pass
|
||||
|
||||
subscribed = await _subscribe(token, waba_id)
|
||||
template_count, templates_ok = await _template_count(token, waba_id)
|
||||
account.template_count = template_count
|
||||
account.templates_status = "active" if templates_ok else "unavailable"
|
||||
account.webhooks_status = "active" if subscribed else "unavailable"
|
||||
|
||||
has_connected_phone = any(phone.status == "connected" for phone in account.phones)
|
||||
if has_connected_phone and account.management_status == "active":
|
||||
account.status = "connected"
|
||||
account.messaging_status = "active"
|
||||
_audit(db, user, "connected", "WhatsApp Business connection verified", connection_id=connection.id, waba_id=waba_id)
|
||||
else:
|
||||
account.status = "incomplete"
|
||||
account.messaging_status = "unavailable"
|
||||
account.last_error = account.last_error or "Phone number verification is not complete"
|
||||
_audit(
|
||||
db,
|
||||
user,
|
||||
"provision_incomplete",
|
||||
account.last_error,
|
||||
connection_id=connection.id,
|
||||
waba_id=waba_id,
|
||||
)
|
||||
_set_account_health(db, user, account)
|
||||
return account
|
||||
|
||||
|
||||
|
|
@ -547,6 +625,49 @@ async def refresh_account(db: Session, user: UserAccount, account_id: Any) -> di
|
|||
return serialize_account(account)
|
||||
|
||||
|
||||
async def register_phone_number(
|
||||
db: Session,
|
||||
user: UserAccount,
|
||||
account_id: Any,
|
||||
phone_id: Any,
|
||||
pin: str | None,
|
||||
) -> dict[str, Any]:
|
||||
account = _owned_account(db, user, account_id)
|
||||
connection = account.connection
|
||||
token = connection.access_token if connection else None
|
||||
if connection is None or not token or connection.status == "disconnected":
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Reconnect this WhatsApp account before registering the number")
|
||||
phone = next((item for item in account.phones if str(item.id) == str(phone_id)), None)
|
||||
if phone is None:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Phone number not found on this account")
|
||||
verified = (phone.code_verification_status or "").upper()
|
||||
if verified in {"NOT_VERIFIED", "EXPIRED"}:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail="Complete the SMS or voice code for this number in Meta first. Then register it for Cloud API.",
|
||||
)
|
||||
submitted = (pin or "").strip()
|
||||
if submitted and (not submitted.isdigit() or len(submitted) != 6):
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Enter the 6-digit two-step PIN for this number")
|
||||
register_pin = _pin_for_phone(phone, submitted or None)
|
||||
register_error = await _register_phone(token, phone, register_pin)
|
||||
if register_error:
|
||||
account.last_error = register_error
|
||||
_audit(db, user, "register_failed", register_error, connection_id=connection.id, waba_id=account.waba_id)
|
||||
db.commit()
|
||||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=register_error)
|
||||
try:
|
||||
phones = await _load_phones(token, account.waba_id)
|
||||
_apply_phones(db, user, account, phones, phone.phone_number_id)
|
||||
except HTTPException:
|
||||
phone.status = "connected"
|
||||
_audit(db, user, "number_registered", "Registered this number for Cloud API", connection_id=connection.id, waba_id=account.waba_id)
|
||||
_set_account_health(db, user, account)
|
||||
db.commit()
|
||||
db.refresh(account)
|
||||
return serialize_account(account)
|
||||
|
||||
|
||||
async def disconnect_account(db: Session, user: UserAccount, account_id: Any) -> dict[str, Any]:
|
||||
account = _owned_account(db, user, account_id)
|
||||
connection = account.connection
|
||||
|
|
@ -692,3 +813,133 @@ def customer_token_for_phone(db: Session, user: UserAccount, phone_number_id: st
|
|||
if not token:
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Reconnect this WhatsApp account before sending")
|
||||
return token, phone.phone_number_id
|
||||
|
||||
|
||||
def list_account_messages(db: Session, user: UserAccount, account_id: Any, limit: int = 300) -> list[dict[str, Any]]:
|
||||
account = _owned_account(db, user, account_id)
|
||||
phone_ids = [phone.phone_number_id for phone in account.phones]
|
||||
if not phone_ids:
|
||||
return []
|
||||
rows = db.scalars(
|
||||
select(WhatsAppMessage)
|
||||
.where(WhatsAppMessage.user_id == user.id, WhatsAppMessage.phone_number_id.in_(phone_ids))
|
||||
.order_by(WhatsAppMessage.created_at.desc())
|
||||
.limit(min(limit, 500))
|
||||
).all()
|
||||
return [serialize_message(row) for row in rows]
|
||||
|
||||
|
||||
def _messaging_phone(account: WhatsAppBusinessAccount) -> tuple[WhatsAppPhoneAsset, str]:
|
||||
connection = account.connection
|
||||
token = connection.access_token if connection else None
|
||||
phone = next((item for item in account.phones if item.status == "connected"), None)
|
||||
if phone is None or account.messaging_status != "active" or not token:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail=account.last_error or "Messaging is unavailable until this number is connected on Cloud API",
|
||||
)
|
||||
return phone, token
|
||||
|
||||
|
||||
async def send_account_message(db: Session, user: UserAccount, account_id: Any, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
account = _owned_account(db, user, account_id)
|
||||
phone, token = _messaging_phone(account)
|
||||
try:
|
||||
recipient = normalize_wa_number(str(payload.get("to") or ""))
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
|
||||
client = WhatsAppGraphClient(access_token=token, phone_number_id=phone.phone_number_id, business_account_id=account.waba_id)
|
||||
message_type = str(payload.get("message_type") or "text")
|
||||
reply_to = payload.get("reply_to") or None
|
||||
try:
|
||||
if message_type == "text":
|
||||
body = str(payload.get("text") or "").strip()
|
||||
if not body:
|
||||
raise HTTPException(status_code=422, detail="text is required")
|
||||
result = await client.send_text(recipient, body, reply_to=reply_to)
|
||||
elif message_type == "template":
|
||||
name = str(payload.get("template_name") or "").strip()
|
||||
if not name:
|
||||
raise HTTPException(status_code=422, detail="template_name is required")
|
||||
result = await client.send_template(recipient, name, language_code=str(payload.get("template_language") or "en"))
|
||||
body = f"Template: {name}"
|
||||
message_type = "template"
|
||||
elif message_type in {"image", "document", "audio", "video"}:
|
||||
link = str(payload.get("media_link") or "").strip()
|
||||
if not link:
|
||||
raise HTTPException(status_code=422, detail="media_link is required")
|
||||
result = await client.send_media(
|
||||
recipient,
|
||||
message_type,
|
||||
link=link,
|
||||
caption=payload.get("media_caption"),
|
||||
filename=payload.get("media_filename"),
|
||||
reply_to=reply_to,
|
||||
)
|
||||
body = payload.get("media_caption") or f"[{message_type}]"
|
||||
elif message_type == "interactive_buttons":
|
||||
buttons = payload.get("interactive_buttons") or []
|
||||
text = str(payload.get("interactive_body") or "").strip()
|
||||
if not text or not buttons:
|
||||
raise HTTPException(status_code=422, detail="interactive_body and interactive_buttons are required")
|
||||
result = await client.send_interactive_buttons(recipient, text, buttons, reply_to=reply_to)
|
||||
body = text
|
||||
message_type = "interactive"
|
||||
elif message_type == "reaction":
|
||||
target = str(payload.get("reaction_message_id") or "")
|
||||
emoji = str(payload.get("reaction_emoji") or "")
|
||||
if not target or not emoji:
|
||||
raise HTTPException(status_code=422, detail="reaction_message_id and reaction_emoji are required")
|
||||
result = await client.send_reaction(recipient, target, emoji)
|
||||
body = emoji
|
||||
else:
|
||||
raise HTTPException(status_code=422, detail="Unsupported message type")
|
||||
except WhatsAppGraphError as exc:
|
||||
raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
|
||||
try:
|
||||
from_number = normalize_wa_number(phone.display_phone or "")
|
||||
except ValueError:
|
||||
from_number = phone.display_phone or phone.phone_number_id
|
||||
message = record_outbound_message(
|
||||
db,
|
||||
to_number=recipient,
|
||||
body=body,
|
||||
message_type=message_type,
|
||||
wa_message_id=extract_wa_message_id(result),
|
||||
status="sent",
|
||||
payload=result if isinstance(result, dict) else None,
|
||||
from_number=from_number,
|
||||
user_id=user.id,
|
||||
phone_number_id=phone.phone_number_id,
|
||||
)
|
||||
return serialize_message(message)
|
||||
|
||||
|
||||
async def download_account_media(db: Session, user: UserAccount, account_id: Any, media_id: str) -> tuple[bytes, str]:
|
||||
account = _owned_account(db, user, account_id)
|
||||
token = account.connection.access_token if account.connection else None
|
||||
if not token:
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Reconnect this WhatsApp account before opening media")
|
||||
phone_ids = [phone.phone_number_id for phone in account.phones]
|
||||
if not phone_ids:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Media not found")
|
||||
rows = db.scalars(
|
||||
select(WhatsAppMessage).where(
|
||||
WhatsAppMessage.user_id == user.id,
|
||||
WhatsAppMessage.phone_number_id.in_(phone_ids),
|
||||
)
|
||||
).all()
|
||||
allowed = False
|
||||
for row in rows:
|
||||
payload = row.payload if isinstance(row.payload, dict) else {}
|
||||
media = payload.get(row.message_type)
|
||||
if isinstance(media, dict) and str(media.get("id") or "") == media_id:
|
||||
allowed = True
|
||||
break
|
||||
if not allowed:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Media not found")
|
||||
client = WhatsAppGraphClient(access_token=token, phone_number_id=phone_ids[0] if phone_ids else None)
|
||||
try:
|
||||
return await client.download_media(media_id)
|
||||
except WhatsAppGraphError as exc:
|
||||
raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue