252 lines
9.3 KiB
Python
252 lines
9.3 KiB
Python
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from .config import settings
|
|
|
|
|
|
class WhatsAppGraphError(Exception):
|
|
def __init__(self, detail: str, status_code: int = 502) -> None:
|
|
super().__init__(detail)
|
|
self.detail = detail
|
|
self.status_code = status_code
|
|
|
|
|
|
class WhatsAppGraphClient:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
access_token: str | None = None,
|
|
phone_number_id: str | None = None,
|
|
business_account_id: str | None = None,
|
|
api_version: str | None = None,
|
|
) -> None:
|
|
self.access_token = access_token or settings.whatsapp_access_token
|
|
self.phone_number_id = phone_number_id or settings.whatsapp_phone_number_id
|
|
self.business_account_id = business_account_id or settings.whatsapp_business_account_id
|
|
self.api_version = api_version or settings.whatsapp_api_version
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return bool(self.access_token and self.phone_number_id)
|
|
|
|
def _graph_url(self, path: str) -> str:
|
|
return f"https://graph.facebook.com/{self.api_version}/{path.lstrip('/')}"
|
|
|
|
def _messages_url(self) -> str:
|
|
return self._graph_url(f"{self.phone_number_id}/messages")
|
|
|
|
async def _request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
json_payload: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
headers = {"Authorization": f"Bearer {self.access_token}"}
|
|
async with httpx.AsyncClient(timeout=60.0) as client:
|
|
response = await client.request(method, url, json=json_payload, headers=headers)
|
|
try:
|
|
data = response.json()
|
|
except ValueError:
|
|
data = {"raw": response.text}
|
|
if response.status_code >= 400:
|
|
detail = data.get("error", {}).get("message") if isinstance(data, dict) else response.text
|
|
raise WhatsAppGraphError(detail or "WhatsApp Graph API request failed", status_code=response.status_code)
|
|
return data if isinstance(data, dict) else {"data": data}
|
|
|
|
async def _send_message(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
if not self.configured:
|
|
raise WhatsAppGraphError("WhatsApp Graph API is not configured", status_code=503)
|
|
body = {"messaging_product": "whatsapp", **payload}
|
|
return await self._request("POST", self._messages_url(), json_payload=body)
|
|
|
|
async def send_text(self, to: str, body: str, reply_to: str | None = None) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"recipient_type": "individual",
|
|
"to": to,
|
|
"type": "text",
|
|
"text": {"preview_url": False, "body": body},
|
|
}
|
|
if reply_to:
|
|
payload["context"] = {"message_id": reply_to}
|
|
return await self._send_message(payload)
|
|
|
|
async def send_media(
|
|
self,
|
|
to: str,
|
|
media_type: str,
|
|
*,
|
|
link: str | None = None,
|
|
media_id: str | None = None,
|
|
caption: str | None = None,
|
|
filename: str | None = None,
|
|
reply_to: str | None = None,
|
|
) -> dict[str, Any]:
|
|
media: dict[str, Any] = {}
|
|
if media_id:
|
|
media["id"] = media_id
|
|
elif link:
|
|
media["link"] = link
|
|
else:
|
|
raise WhatsAppGraphError("Media requires link or media_id", status_code=422)
|
|
if caption and media_type in {"image", "document", "video"}:
|
|
media["caption"] = caption
|
|
if filename and media_type == "document":
|
|
media["filename"] = filename
|
|
payload: dict[str, Any] = {
|
|
"recipient_type": "individual",
|
|
"to": to,
|
|
"type": media_type,
|
|
media_type: media,
|
|
}
|
|
if reply_to:
|
|
payload["context"] = {"message_id": reply_to}
|
|
return await self._send_message(payload)
|
|
|
|
async def send_template(
|
|
self,
|
|
to: str,
|
|
name: str,
|
|
language_code: str = "en",
|
|
components: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
template: dict[str, Any] = {"name": name, "language": {"code": language_code}}
|
|
if components:
|
|
template["components"] = components
|
|
return await self._send_message(
|
|
{
|
|
"recipient_type": "individual",
|
|
"to": to,
|
|
"type": "template",
|
|
"template": template,
|
|
}
|
|
)
|
|
|
|
async def send_marketing_template(
|
|
self,
|
|
to: str,
|
|
name: str,
|
|
language_code: str = "en",
|
|
components: list[dict[str, Any]] | None = None,
|
|
product_policy: str = "STRICT",
|
|
) -> dict[str, Any]:
|
|
"""Send a marketing template through Meta's marketing endpoint, not /messages."""
|
|
if not self.configured:
|
|
raise WhatsAppGraphError("WhatsApp Graph API is not configured", status_code=503)
|
|
template: dict[str, Any] = {"name": name, "language": {"code": language_code}}
|
|
if components:
|
|
template["components"] = components
|
|
return await self._request(
|
|
"POST",
|
|
self._graph_url(f"{self.phone_number_id}/marketing_messages"),
|
|
json_payload={
|
|
"messaging_product": "whatsapp",
|
|
"recipient_type": "individual",
|
|
"to": to,
|
|
"type": "template",
|
|
"template": template,
|
|
"product_policy": product_policy,
|
|
},
|
|
)
|
|
|
|
async def send_interactive_buttons(
|
|
self,
|
|
to: str,
|
|
body: str,
|
|
buttons: list[dict[str, str]],
|
|
*,
|
|
header: str | None = None,
|
|
footer: str | None = None,
|
|
reply_to: str | None = None,
|
|
) -> dict[str, Any]:
|
|
interactive: dict[str, Any] = {
|
|
"type": "button",
|
|
"body": {"text": body},
|
|
"action": {
|
|
"buttons": [
|
|
{"type": "reply", "reply": {"id": button["id"], "title": button["title"][:20]}}
|
|
for button in buttons[:3]
|
|
]
|
|
},
|
|
}
|
|
if header:
|
|
interactive["header"] = {"type": "text", "text": header[:60]}
|
|
if footer:
|
|
interactive["footer"] = {"text": footer[:60]}
|
|
payload: dict[str, Any] = {
|
|
"recipient_type": "individual",
|
|
"to": to,
|
|
"type": "interactive",
|
|
"interactive": interactive,
|
|
}
|
|
if reply_to:
|
|
payload["context"] = {"message_id": reply_to}
|
|
return await self._send_message(payload)
|
|
|
|
async def send_interactive_list(
|
|
self,
|
|
to: str,
|
|
body: str,
|
|
button_text: str,
|
|
sections: list[dict[str, Any]],
|
|
*,
|
|
header: str | None = None,
|
|
footer: str | None = None,
|
|
reply_to: str | None = None,
|
|
) -> dict[str, Any]:
|
|
interactive: dict[str, Any] = {
|
|
"type": "list",
|
|
"body": {"text": body},
|
|
"action": {"button": button_text[:20], "sections": sections},
|
|
}
|
|
if header:
|
|
interactive["header"] = {"type": "text", "text": header[:60]}
|
|
if footer:
|
|
interactive["footer"] = {"text": footer[:60]}
|
|
payload: dict[str, Any] = {
|
|
"recipient_type": "individual",
|
|
"to": to,
|
|
"type": "interactive",
|
|
"interactive": interactive,
|
|
}
|
|
if reply_to:
|
|
payload["context"] = {"message_id": reply_to}
|
|
return await self._send_message(payload)
|
|
|
|
async def send_reaction(self, to: str, message_id: str, emoji: str) -> dict[str, Any]:
|
|
return await self._send_message(
|
|
{
|
|
"recipient_type": "individual",
|
|
"to": to,
|
|
"type": "reaction",
|
|
"reaction": {"message_id": message_id, "emoji": emoji},
|
|
}
|
|
)
|
|
|
|
async def mark_read(self, message_id: str) -> dict[str, Any]:
|
|
return await self._send_message({"status": "read", "message_id": message_id})
|
|
|
|
async def list_templates(self, limit: int = 50) -> dict[str, Any]:
|
|
if not self.access_token or not self.business_account_id:
|
|
raise WhatsAppGraphError("WhatsApp business account ID is not configured", status_code=503)
|
|
url = self._graph_url(f"{self.business_account_id}/message_templates?limit={limit}")
|
|
return await self._request("GET", url)
|
|
|
|
async def get_media_metadata(self, media_id: str) -> dict[str, Any]:
|
|
if not self.access_token:
|
|
raise WhatsAppGraphError("WhatsApp access token is not configured", status_code=503)
|
|
return await self._request("GET", self._graph_url(media_id))
|
|
|
|
async def download_media(self, media_id: str) -> tuple[bytes, str]:
|
|
metadata = await self.get_media_metadata(media_id)
|
|
download_url = metadata.get("url")
|
|
mime_type = metadata.get("mime_type") or "application/octet-stream"
|
|
if not download_url:
|
|
raise WhatsAppGraphError("Media URL not available", status_code=404)
|
|
headers = {"Authorization": f"Bearer {self.access_token}"}
|
|
async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client:
|
|
response = await client.get(download_url, headers=headers)
|
|
if response.status_code >= 400:
|
|
raise WhatsAppGraphError("Failed to download media", status_code=response.status_code)
|
|
return response.content, mime_type
|