Spaces:
Running
Running
| 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: | |
| 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", "") | |
| ) | |