"""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_test_… ou ccs_live_…) — requis CHAPCHAPSEND_API_SECRET secret d'API — requis pour le SMS CHAPCHAPSEND_BASE_URL surcharge de l'URL de base — optionnel Le WhatsApp n'exige que la clé ; le SMS exige la clé et le secret. Le préfixe de la clé décide du mode : « ccs_test_ » n'entame pas vos crédits. 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, api_secret: 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._api_secret = api_secret or os.getenv("CHAPCHAPSEND_API_SECRET", "") 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." ) @property def is_test_mode(self) -> bool: return self._api_key.startswith("ccs_test_") # ------------------------------------------------------------------ SMS def send_sms( self, to: str, message: str, sender_name: Optional[str] = None, callback_url: Optional[str] = None, ) -> Dict[str, Any]: """Envoie un SMS. Bloque jusqu'au verdict de l'opérateur.""" payload: Dict[str, Any] = {"to": to, "message": message} if sender_name: payload["sender_name"] = sender_name if callback_url: payload["callback_url"] = callback_url return self._request("POST", "/api/integrations/sms/send", payload, secret=True) 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``. """ return self._run_bulk( messages, lambda item: self.send_sms(**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}", secret=True ) def get_credits(self) -> Dict[str, Any]: return self._request("GET", "/api/integrations/sms/credits", secret=True) # ------------------------------------------------------------- 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_async(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return await asyncio.to_thread(self.send_sms, *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, 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, secret: bool) -> Dict[str, str]: headers = { "X-Api-Key": self._api_key, "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "chapchapsend-python/1.0", } if secret: if not self._api_secret: raise ChapChapSendConfigError( "Secret d'API absent : l'envoi de SMS exige " "CHAPCHAPSEND_API_SECRET." ) headers["X-Api-Secret"] = self._api_secret return headers def _request( self, method: str, path: str, payload: Optional[Dict[str, Any]] = None, secret: bool = False, ) -> 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(secret), 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 vos credentials, copiés depuis le portail ChapChapSend # (SMS → Intégration (API), ou WhatsApp → Intégration (API)). # # export CHAPCHAPSEND_API_KEY="ccs_test_xxxxxxxxxxxxxxxx" # export CHAPCHAPSEND_API_SECRET="sk_xxxxxxxxxxxxxxxx" # # # 1. Un SMS, en bloquant # # from chapchapsend_service import ChapChapSendService, ChapChapSendError # # ccs = ChapChapSendService() # # try: # result = ccs.send_sms( # to="+221771234567", # message="Votre code de confirmation est 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", "message": "Votre commande est prête."}, # {"to": "+221779876543", "message": "Votre commande est prête."}, # ]) # 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_async("+221771234567", "Bonjour")) # # … votre traitement continue ; pensez à `await task` avant de quitter.