import logging import json import urllib.request import urllib.error import ssl from typing import List logger = logging.getLogger(__name__) class MetaWhatsAppProvider: def __init__(self, phone_id: str, token: str): self.phone_id = phone_id self.token = token def send_message(self, admin_numbers: List[str], text: str) -> List[str]: if not self.phone_id or not self.token: return ["Meta WhatsApp settings incomplete."] url = f"https://graph.facebook.com/v17.0/{self.phone_id}/messages" headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json" } ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE errors = [] for number in admin_numbers: payload = { "messaging_product": "whatsapp", "to": number, "type": "text", "text": {"body": text} } data = json.dumps(payload).encode('utf-8') req = urllib.request.Request(url, data=data, headers=headers, method='POST') try: with urllib.request.urlopen(req, timeout=15.0, context=ctx) as response: logger.info("Meta WhatsApp notification sent successfully to %s", number) except urllib.error.HTTPError as e: err_msg = e.read().decode('utf-8') detail = err_msg try: detail = json.loads(err_msg).get("error", {}).get("message", err_msg) except Exception: pass errors.append(f"Meta Error for {number}: {detail}") logger.error("Meta WhatsApp API error for %s (%d): %s", number, e.code, err_msg) except Exception as e: errors.append(f"Meta Network error for {number}: {type(e).__name__} - {str(e)}") logger.error("Meta Network error sending WhatsApp to %s: %s - %s", number, type(e).__name__, str(e)) return errors class BaileysWhatsAppProvider: def __init__(self, url: str, token: str): self.url = url self.token = token def send_message(self, admin_numbers: List[str], text: str) -> List[str]: if not self.url: return ["Baileys API URL not configured."] headers = { "Content-Type": "application/json" } if self.token: # Common headers for various Baileys REST APIs (Evolution API uses apikey) headers["apikey"] = self.token headers["Authorization"] = f"Bearer {self.token}" ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE errors = [] for number in admin_numbers: # Payload tailored to work with common wrappers like Evolution API payload = { "number": number, "options": { "delay": 0, "presence": "composing" }, "textMessage": { "text": text }, "text": text, # generic fallback "message": text # generic fallback } data = json.dumps(payload).encode('utf-8') req = urllib.request.Request(self.url, data=data, headers=headers, method='POST') try: with urllib.request.urlopen(req, timeout=15.0, context=ctx) as response: logger.info("Baileys WhatsApp notification sent successfully to %s", number) except urllib.error.HTTPError as e: err_msg = e.read().decode('utf-8') errors.append(f"Baileys Error for {number}: {err_msg}") logger.error("Baileys WhatsApp API error for %s (%d): %s", number, e.code, err_msg) except Exception as e: errors.append(f"Baileys Network error for {number}: {type(e).__name__} - {str(e)}") logger.error("Baileys Network error sending WhatsApp to %s: %s - %s", number, type(e).__name__, str(e)) return errors class ProviderFactory: @staticmethod def get_provider(settings: dict): provider_type = settings.get("provider", "meta") if provider_type == "baileys": return BaileysWhatsAppProvider( url=settings.get("baileys_url", ""), token=settings.get("baileys_token", "") ) else: return MetaWhatsAppProvider( phone_id=settings.get("phone_id", ""), token=settings.get("token", "") )