56 lines
2 KiB
Python
56 lines
2 KiB
Python
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,
|
|
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.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 _messages_url(self) -> str:
|
|
return (
|
|
f"https://graph.facebook.com/{self.api_version}/"
|
|
f"{self.phone_number_id}/messages"
|
|
)
|
|
|
|
async def send_text(self, to: str, body: str) -> dict:
|
|
if not self.configured:
|
|
raise WhatsAppGraphError("WhatsApp Graph API is not configured", status_code=503)
|
|
|
|
payload = {
|
|
"messaging_product": "whatsapp",
|
|
"recipient_type": "individual",
|
|
"to": to,
|
|
"type": "text",
|
|
"text": {"preview_url": False, "body": body},
|
|
}
|
|
headers = {"Authorization": f"Bearer {self.access_token}"}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
response = await client.post(self._messages_url(), 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
|