# -*- coding: utf-8 -*- import logging import requests import json import hmac import hashlib import re from odoo import models, fields, api, _ _logger = logging.getLogger(__name__) WHATSAPP_API_VERSION = "v19.0" WHATSAPP_API_BASE = f"https://graph.facebook.com/{WHATSAPP_API_VERSION}" class SocialWhatsAppApi(models.AbstractModel): _name = 'social.whatsapp.api' _description = 'WhatsApp Business Cloud API Service' @api.model def _mask_secret(self, secret): if not secret: return "" if len(secret) <= 8: return "******" return f"{secret[:4]}...{secret[-4:]}" @api.model def format_phone(self, phone, partner=None): """Cleans and formats phone number into international digits without + or whitespace.""" if not phone: return "" # Strip all non-digit characters cleaned = re.sub(r'\D', '', phone) # If partner has a country with phone code, ensure it has international code if starting with 0 if cleaned.startswith('0') and partner and partner.country_id and partner.country_id.phone_code: cleaned = f"{partner.country_id.phone_code}{cleaned[1:]}" return cleaned @api.model def test_connection(self, company=None): company = company or self.env.company phone_number_id = company.whatsapp_phone_number_id access_token = company.whatsapp_access_token if not phone_number_id or not access_token: company.sudo().write({ 'whatsapp_connection_status': 'error', 'whatsapp_last_connection_test': fields.Datetime.now(), 'whatsapp_last_test_error': _("Phone Number ID and Access Token are required.") }) return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _("WhatsApp Configuration Missing"), 'message': _("Please specify both WhatsApp Phone Number ID and Access Token."), 'type': 'warning', 'sticky': False, } } url = f"{WHATSAPP_API_BASE}/{phone_number_id}" headers = { 'Authorization': f"Bearer {access_token}", 'Content-Type': 'application/json' } params = { 'fields': 'id,display_phone_number,verified_name,code_verification_status,quality_rating' } try: _logger.info("Testing WhatsApp connection for Phone Number ID %s", phone_number_id) response = requests.get(url, headers=headers, params=params, timeout=12) if response.status_code == 200: data = response.json() phone_num = data.get('display_phone_number', phone_number_id) verified_name = data.get('verified_name', 'Verified Account') rating = data.get('quality_rating', 'UNKNOWN') company.sudo().write({ 'whatsapp_connection_status': 'connected', 'whatsapp_last_connection_test': fields.Datetime.now(), 'whatsapp_last_test_error': False }) return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _("WhatsApp Connection Successful!"), 'message': _("Connected: %s (%s). Quality: %s") % (verified_name, phone_num, rating), 'type': 'success', 'sticky': False, } } else: error_data = response.json().get('error', {}) error_msg = error_data.get('message', response.text) company.sudo().write({ 'whatsapp_connection_status': 'error', 'whatsapp_last_connection_test': fields.Datetime.now(), 'whatsapp_last_test_error': f"HTTP {response.status_code}: {error_msg}" }) return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _("WhatsApp Connection Failed"), 'message': _("WhatsApp Cloud API error (HTTP %s): %s") % (response.status_code, error_msg), 'type': 'danger', 'sticky': True, } } except requests.exceptions.Timeout: err = _("Connection to WhatsApp Cloud API timed out after 12 seconds.") company.sudo().write({ 'whatsapp_connection_status': 'error', 'whatsapp_last_connection_test': fields.Datetime.now(), 'whatsapp_last_test_error': err }) return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': {'title': _("Timeout"), 'message': err, 'type': 'danger', 'sticky': True} } except Exception as e: err = str(e) company.sudo().write({ 'whatsapp_connection_status': 'error', 'whatsapp_last_connection_test': fields.Datetime.now(), 'whatsapp_last_test_error': err }) return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': {'title': _("Connection Error"), 'message': err, 'type': 'danger', 'sticky': True} } @api.model def send_template_message(self, company, to_phone, template_name, language_code='en_US', parameters=None, header_params=None): """ Sends an approved WhatsApp Template message via WhatsApp Cloud API. parameters: list of strings [param1, param2, ...] """ phone_number_id = company.whatsapp_phone_number_id access_token = company.whatsapp_access_token if not company.whatsapp_enabled: return {'success': False, 'error': _("WhatsApp integration is disabled in settings.")} if not phone_number_id or not access_token: return {'success': False, 'error': _("Missing WhatsApp Phone Number ID or Access Token.")} cleaned_to = self.format_phone(to_phone) if not cleaned_to: return {'success': False, 'error': _("Recipient phone number is invalid or empty.")} components = [] if header_params: components.append({ "type": "header", "parameters": [{"type": "text", "text": str(p)} for p in header_params] }) if parameters: components.append({ "type": "body", "parameters": [{"type": "text", "text": str(p)} for p in parameters] }) payload = { "messaging_product": "whatsapp", "recipient_type": "individual", "to": cleaned_to, "type": "template", "template": { "name": template_name, "language": { "code": language_code }, "components": components } } url = f"{WHATSAPP_API_BASE}/{phone_number_id}/messages" headers = { 'Authorization': f"Bearer {access_token}", 'Content-Type': 'application/json' } try: _logger.info("Sending WhatsApp template '%s' to phone %s...", template_name, cleaned_to) response = requests.post(url, headers=headers, json=payload, timeout=12) if response.status_code in (200, 201): res_data = response.json() messages = res_data.get('messages', []) msg_id = messages[0].get('id') if messages else '' _logger.info("WhatsApp template message sent successfully. ID: %s", msg_id) return {'success': True, 'message_id': msg_id, 'response': res_data} else: error_data = response.json().get('error', {}) error_msg = error_data.get('message', response.text) _logger.error("WhatsApp template send failed (HTTP %s): %s", response.status_code, error_msg) return {'success': False, 'error': f"HTTP {response.status_code}: {error_msg}"} except requests.exceptions.Timeout: err = _("WhatsApp Cloud API send request timed out.") _logger.error(err) return {'success': False, 'error': err} except Exception as e: err = str(e) _logger.exception("Unexpected error sending WhatsApp template message: %s", err) return {'success': False, 'error': err} @api.model def send_text_message(self, company, to_phone, body_text): """ Sends a standard text message within the 24-hour customer service window. """ phone_number_id = company.whatsapp_phone_number_id access_token = company.whatsapp_access_token if not company.whatsapp_enabled or not phone_number_id or not access_token: return {'success': False, 'error': _("WhatsApp API not configured.")} cleaned_to = self.format_phone(to_phone) if not cleaned_to or not body_text: return {'success': False, 'error': _("Invalid phone or empty message.")} payload = { "messaging_product": "whatsapp", "recipient_type": "individual", "to": cleaned_to, "type": "text", "text": { "preview_url": False, "body": body_text[:4096] } } url = f"{WHATSAPP_API_BASE}/{phone_number_id}/messages" headers = { 'Authorization': f"Bearer {access_token}", 'Content-Type': 'application/json' } try: response = requests.post(url, headers=headers, json=payload, timeout=12) if response.status_code in (200, 201): res_data = response.json() msg_id = res_data.get('messages', [{}])[0].get('id', '') return {'success': True, 'message_id': msg_id, 'response': res_data} else: error_msg = response.json().get('error', {}).get('message', response.text) return {'success': False, 'error': error_msg} except Exception as e: return {'success': False, 'error': str(e)} @api.model def verify_webhook_signature(self, raw_body, signature_header, app_secret): """ Validates the HMAC-SHA256 signature from Meta webhook request (X-Hub-Signature-256). """ if not app_secret or not signature_header: return False if not signature_header.startswith('sha256='): return False expected_sig = signature_header[7:] mac = hmac.new(app_secret.encode('utf-8'), raw_body, hashlib.sha256) computed_sig = mac.hexdigest() return hmac.compare_digest(computed_sig, expected_sig)