"""Client ChapChapSend — SMS et templates WhatsApp. Fichier autonome, sans dépendance : déposez-le dans votre projet et importez-le. Compatible Python 3.9+. Configuration par variables d'environnement : CHAPCHAPSEND_API_KEY clé d'API (ccs_live_…) — requis CHAPCHAPSEND_BASE_URL surcharge de l'URL de base — optionnel La clé suffit, pour le SMS comme pour le WhatsApp. Chaque envoi débite vos crédits : réclamez vos crédits offerts depuis la facturation du portail pour essayer sans payer. Un SMS part d'un template validé par ChapChapSend, comme un message WhatsApp part d'un template approuvé par Meta : soumettez le texte depuis le portail (SMS → Templates), puis envoyez-le avec son identifiant. Des exemples d'utilisation figurent en bas de ce fichier. """ from __future__ import annotations import asyncio import json import os import urllib.error import urllib.request from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Sequence DEFAULT_BASE_URL = "https://api.chapchapsend.com" DEFAULT_TIMEOUT = 30.0 DEFAULT_BULK_CONCURRENCY = 4 class ChapChapSendError(Exception): """Erreur retournée par l'API. ``detail`` reprend le message exact du service — y compris celui de Meta ou de l'opérateur quand c'est lui qui a refusé le message (statut 502). """ def __init__(self, status_code: int, detail: str) -> None: super().__init__(f"[{status_code}] {detail}") self.status_code = status_code self.detail = detail @property def is_rejected_by_provider(self) -> bool: return self.status_code == 502 @property def is_out_of_credits(self) -> bool: return self.status_code == 402 class ChapChapSendConfigError(Exception): pass @dataclass class BulkResult: """Issue d'un envoi de masse : les succès d'un côté, les échecs de l'autre.""" sent: List[Dict[str, Any]] = field(default_factory=list) failed: List[Dict[str, Any]] = field(default_factory=list) @property def total(self) -> int: return len(self.sent) + len(self.failed) @property def all_sent(self) -> bool: return not self.failed class ChapChapSendService: def __init__( self, api_key: Optional[str] = None, base_url: Optional[str] = None, timeout: float = DEFAULT_TIMEOUT, ) -> None: self._api_key = api_key or os.getenv("CHAPCHAPSEND_API_KEY", "") self._base_url = ( base_url or os.getenv("CHAPCHAPSEND_BASE_URL") or DEFAULT_BASE_URL ).rstrip("/") self._timeout = timeout if not self._api_key: raise ChapChapSendConfigError( "Clé d'API absente : définissez CHAPCHAPSEND_API_KEY ou passez " "api_key au constructeur." ) # ------------------------------------------------------------------ SMS def list_sms_templates(self) -> List[Dict[str, Any]]: """Liste les templates SMS validés, avec leurs variables obligatoires.""" response = self._request("GET", "/api/integrations/sms/templates") items = response.get("items") return list(items) if isinstance(items, list) else [] def send_sms_template( self, to: str, template_id: str, variables: Optional[Dict[str, str]] = None, sender_name: Optional[str] = None, ) -> Dict[str, Any]: """Envoie un template SMS validé. Bloque jusqu'au verdict de l'opérateur.""" payload: Dict[str, Any] = {"to": to, "template_id": template_id} if variables: payload["variables"] = dict(variables) if sender_name: payload["sender_name"] = sender_name return self._request("POST", "/api/integrations/sms/send-template", payload) def send_sms_bulk( self, messages: Sequence[Dict[str, Any]], stop_on_error: bool = False, ) -> BulkResult: """Envoie une liste de SMS, un par un, et rapporte chaque issue. Chaque entrée accepte les mêmes champs que ``send_sms_template``. """ return self._run_bulk( messages, lambda item: self.send_sms_template(**item), stop_on_error ) def get_sms_status(self, message_id: str) -> Dict[str, Any]: return self._request("GET", f"/api/integrations/sms/messages/{message_id}") def get_credits(self) -> Dict[str, Any]: return self._request("GET", "/api/integrations/sms/credits") # ------------------------------------------------------------- WhatsApp def list_whatsapp_templates(self) -> List[Dict[str, Any]]: """Liste les templates approuvés, avec leurs variables obligatoires.""" response = self._request("GET", "/api/integrations/whatsapp/templates") items = response.get("items") return list(items) if isinstance(items, list) else [] def send_whatsapp_template( self, to: str, template_id: str, variables: Optional[Dict[str, str]] = None, product_ids: Optional[Sequence[str]] = None, ) -> Dict[str, Any]: """Envoie un template approuvé. Bloque jusqu'au verdict de WhatsApp.""" payload: Dict[str, Any] = {"to": to, "template_id": template_id} if variables: payload["variables"] = dict(variables) if product_ids: payload["product_ids"] = list(product_ids) return self._request("POST", "/api/integrations/whatsapp/send", payload) def send_whatsapp_bulk( self, messages: Sequence[Dict[str, Any]], stop_on_error: bool = False, ) -> BulkResult: """Envoie une liste de templates WhatsApp et rapporte chaque issue. Chaque entrée accepte les mêmes champs que ``send_whatsapp_template``. """ return self._run_bulk( messages, lambda item: self.send_whatsapp_template(**item), stop_on_error ) def get_whatsapp_status(self, message_id: str) -> Dict[str, Any]: return self._request( "GET", f"/api/integrations/whatsapp/messages/{message_id}" ) # ------------------------------------------- variantes non bloquantes async def send_sms_template_async( self, *args: Any, **kwargs: Any ) -> Dict[str, Any]: return await asyncio.to_thread(self.send_sms_template, *args, **kwargs) async def send_whatsapp_template_async( self, *args: Any, **kwargs: Any ) -> Dict[str, Any]: return await asyncio.to_thread(self.send_whatsapp_template, *args, **kwargs) async def send_sms_bulk_async( self, messages: Sequence[Dict[str, Any]], concurrency: int = DEFAULT_BULK_CONCURRENCY, ) -> BulkResult: return await self._run_bulk_async( messages, self.send_sms_template, concurrency ) async def send_whatsapp_bulk_async( self, messages: Sequence[Dict[str, Any]], concurrency: int = DEFAULT_BULK_CONCURRENCY, ) -> BulkResult: return await self._run_bulk_async( messages, self.send_whatsapp_template, concurrency ) # ------------------------------------------------------------ internes @staticmethod def _run_bulk( messages: Sequence[Dict[str, Any]], send: Any, stop_on_error: bool, ) -> BulkResult: outcome = BulkResult() for item in messages: try: outcome.sent.append(send(item)) except ChapChapSendError as error: outcome.failed.append( { "to": item.get("to"), "status_code": error.status_code, "detail": error.detail, } ) if stop_on_error: break return outcome async def _run_bulk_async( self, messages: Sequence[Dict[str, Any]], send: Any, concurrency: int, ) -> BulkResult: semaphore = asyncio.Semaphore(max(1, concurrency)) async def one(item: Dict[str, Any]) -> Any: async with semaphore: try: return await asyncio.to_thread(lambda: send(**item)) except ChapChapSendError as error: return { "to": item.get("to"), "status_code": error.status_code, "detail": error.detail, } results = await asyncio.gather(*(one(item) for item in messages)) outcome = BulkResult() for result in results: if "status_code" in result: outcome.failed.append(result) else: outcome.sent.append(result) return outcome def _headers(self) -> Dict[str, str]: return { "X-Api-Key": self._api_key, "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "chapchapsend-python/2.0", } def _request( self, method: str, path: str, payload: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: body = json.dumps(payload).encode("utf-8") if payload is not None else None request = urllib.request.Request( f"{self._base_url}{path}", data=body, headers=self._headers(), method=method, ) try: with urllib.request.urlopen(request, timeout=self._timeout) as response: raw = response.read().decode("utf-8") except urllib.error.HTTPError as error: raise ChapChapSendError(error.code, _detail(error.read())) from error except urllib.error.URLError as error: raise ChapChapSendError(0, f"Service injoignable : {error.reason}") from error return json.loads(raw) if raw else {} def _detail(raw: bytes) -> str: try: parsed = json.loads(raw.decode("utf-8")) except (ValueError, UnicodeDecodeError): return raw.decode("utf-8", "replace") or "Erreur inconnue." if isinstance(parsed, dict) and "detail" in parsed: return str(parsed["detail"]) return str(parsed) # --------------------------------------------------------------------------- # EXEMPLES D'UTILISATION # --------------------------------------------------------------------------- # # Prérequis : exportez votre clé, créée depuis le portail ChapChapSend # (Développeurs → Clés d'API). # # export CHAPCHAPSEND_API_KEY="ccs_live_xxxxxxxxxxxxxxxx" # # # 1. Un SMS, en bloquant # # Le texte « Votre code de confirmation est {{code}}. » a été soumis et # validé depuis le portail (SMS → Templates) ; son identifiant s'y copie. # # from chapchapsend_service import ChapChapSendService, ChapChapSendError # # ccs = ChapChapSendService() # # for template in ccs.list_sms_templates(): # print(template["template_id"], template["name"], # template["required_variables"]) # # try: # result = ccs.send_sms_template( # to="+221771234567", # template_id="0198e3a1-9b20-7f41-8c3e-5d2a1f6b7c90", # variables={"code": "4821"}, # ) # print(result["message_id"], result["status"]) # except ChapChapSendError as error: # if error.is_rejected_by_provider: # print("L'opérateur a refusé :", error.detail) # elif error.is_out_of_credits: # print("Rechargez votre compte :", error.detail) # else: # raise # # # 2. Un template WhatsApp, en bloquant # # for template in ccs.list_whatsapp_templates(): # print(template["template_id"], template["name"], # template["required_variables"]) # # result = ccs.send_whatsapp_template( # to="+221771234567", # template_id="0198e3a1-7c44-7c2a-9f10-2b6d5e4a8c11", # variables={"numero_commande": "CMD-1042", "montant": "24 500 FCFA"}, # ) # print(result["whatsapp_message_id"]) # # # 3. Un envoi de masse, séquentiel # # outcome = ccs.send_sms_bulk([ # {"to": "+221771234567", # "template_id": "0198e3a1-9b20-7f41-8c3e-5d2a1f6b7c90", # "variables": {"code": "4821"}}, # {"to": "+221779876543", # "template_id": "0198e3a1-9b20-7f41-8c3e-5d2a1f6b7c90", # "variables": {"code": "7310"}}, # ]) # print(f"{len(outcome.sent)} envoyés, {len(outcome.failed)} en échec") # for failure in outcome.failed: # print(failure["to"], failure["detail"]) # # # 4. Non bloquant : le reste de votre traitement continue # # import asyncio # # async def notifier_client(): # ccs = ChapChapSendService() # result = await ccs.send_whatsapp_template_async( # to="+221771234567", # template_id="0198e3a1-7c44-7c2a-9f10-2b6d5e4a8c11", # variables={"numero_commande": "CMD-1042", "montant": "24 500 FCFA"}, # ) # return result["message_id"] # # asyncio.run(notifier_client()) # # # 5. Masse non bloquante, 8 envois en parallèle # # async def campagne(destinataires): # ccs = ChapChapSendService() # return await ccs.send_whatsapp_bulk_async( # [ # { # "to": phone, # "template_id": "0198e3a1-7c44-7c2a-9f10-2b6d5e4a8c11", # "variables": {"numero_commande": ref, "montant": "24 500 FCFA"}, # } # for phone, ref in destinataires # ], # concurrency=8, # ) # # # 6. Vraiment « fire and forget » (vous renoncez au verdict) # # task = asyncio.create_task( # ccs.send_sms_template_async( # "+221771234567", "0198e3a1-9b20-7f41-8c3e-5d2a1f6b7c90", {"code": "4821"} # ) # ) # # … votre traitement continue ; pensez à `await task` avant de quitter.