consultoria-as/Autoparts-DB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(pos): replace Meta Cloud API WhatsApp with Evolution API (self-hosted)

Browse Source 
Switch from Meta Business Cloud API to Evolution API for WhatsApp integration.
Evolution API is self-hosted, free, and connects via WhatsApp Web QR code scan.

- Add docker-compose for Evolution API deployment
- Rewrite whatsapp_service.py for Evolution API endpoints
- Add instance management (create, QR, status, logout) to blueprint
- Add QR code scanning UI with connection status indicator
- Add duplicate message prevention in webhook handler
- Update config.py with EVOLUTION_API_URL/KEY (remove old Meta vars)
- Add setup documentation

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
consultoria-as
2026-04-05 03:15:52 +00:00
parent 04340f2f29
commit 5f92fe83ba
7 changed files with 672 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

404
pos/services/whatsapp_service.py
View File

@@ -1,137 +1,190 @@
"""WhatsApp Business API service via Meta Cloud API.
"""WhatsApp service via Evolution API (self-hosted, free).
Handles sending and receiving WhatsApp messages.
Uses Meta's Cloud API (free for 1,000 service conversations/month).
Setup:
1. Create a Meta Business account at business.facebook.com
2. Create a WhatsApp Business app at developers.facebook.com
3. Get a permanent access token and Phone Number ID
4. Set environment variables:
WHATSAPP_TOKEN=<your-permanent-token>
WHATSAPP_PHONE_ID=<your-phone-number-id>
WHATSAPP_VERIFY_TOKEN=<your-chosen-verify-string>
5. Configure the webhook URL in Meta Developer Console:
https://<your-domain>/pos/api/whatsapp/webhook
Evolution API connects to WhatsApp Web via QR code scan.
Docs: https://doc.evolution-api.com/
"""
import logging
import requests
from config import WHATSAPP_TOKEN, WHATSAPP_PHONE_ID
from config import EVOLUTION_API_URL, EVOLUTION_API_KEY
logger = logging.getLogger(__name__)
API_VERSION = "v18.0"
BASE_URL = f"https://graph.facebook.com/{API_VERSION}"
def _headers():
return {
"Authorization": f"Bearer {WHATSAPP_TOKEN}",
"Content-Type": "application/json",
}
HEADERS = {
'apikey': EVOLUTION_API_KEY,
'Content-Type': 'application/json'
}
def is_configured():
"""Return True if WhatsApp credentials are set."""
return bool(WHATSAPP_TOKEN and WHATSAPP_PHONE_ID)
"""Return True if Evolution API credentials are set."""
return bool(EVOLUTION_API_URL and EVOLUTION_API_KEY)
# ── Sending ──────────────────────────────────────────────────────────────
# -- Instance management -----------------------------------------------------
def send_message(to_phone, message_text):
"""Send a plain-text message via WhatsApp Cloud API.
def create_instance(instance_name):
"""Create a WhatsApp instance (one per tenant/phone number)."""
try:
resp = requests.post(
f'{EVOLUTION_API_URL}/instance/create',
headers=HEADERS,
json={
'instanceName': instance_name,
'qrcode': True,
'integration': 'WHATSAPP-BAILEYS'
},
timeout=15
)
return resp.json()
except Exception as e:
logger.exception("Failed to create Evolution instance")
return {'error': str(e)}
def get_qr_code(instance_name):
"""Get QR code to connect WhatsApp.
Returns dict with 'base64' key containing data:image/png;base64,... string.
"""
try:
resp = requests.get(
f'{EVOLUTION_API_URL}/instance/connect/{instance_name}',
headers=HEADERS,
timeout=15
)
return resp.json()
except Exception as e:
logger.exception("Failed to get QR code")
return {'error': str(e)}
def get_instance_status(instance_name):
"""Check if instance is connected.
Returns dict with 'state' key: 'open' | 'close' | 'connecting'.
"""
try:
resp = requests.get(
f'{EVOLUTION_API_URL}/instance/connectionState/{instance_name}',
headers=HEADERS,
timeout=10
)
return resp.json()
except Exception as e:
logger.exception("Failed to get instance status")
return {'error': str(e), 'state': 'close'}
def logout_instance(instance_name):
"""Disconnect WhatsApp instance."""
try:
resp = requests.delete(
f'{EVOLUTION_API_URL}/instance/logout/{instance_name}',
headers=HEADERS,
timeout=10
)
return resp.json()
except Exception as e:
logger.exception("Failed to logout instance")
return {'error': str(e)}
def delete_instance(instance_name):
"""Delete instance completely."""
try:
resp = requests.delete(
f'{EVOLUTION_API_URL}/instance/delete/{instance_name}',
headers=HEADERS,
timeout=10
)
return resp.json()
except Exception as e:
logger.exception("Failed to delete instance")
return {'error': str(e)}
# -- Sending messages ---------------------------------------------------------
def send_message(instance_name, to_phone, message_text):
"""Send text message.
Args:
to_phone: recipient phone in E.164 format (e.g. '5215512345678')
instance_name: Evolution instance name (tenant identifier)
to_phone: phone in format 5214421234567 (country code + number, no +)
message_text: text content
Returns:
dict with 'wa_message_id' on success or 'error' on failure.
dict with response or 'error' key on failure.
"""
if not is_configured():
return {"error": "WhatsApp not configured — set WHATSAPP_TOKEN and WHATSAPP_PHONE_ID"}
url = f"{BASE_URL}/{WHATSAPP_PHONE_ID}/messages"
payload = {
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": to_phone,
"type": "text",
"text": {"preview_url": False, "body": message_text},
}
return {'error': 'Evolution API not configured'}
try:
resp = requests.post(url, json=payload, headers=_headers(), timeout=15)
resp = requests.post(
f'{EVOLUTION_API_URL}/message/sendText/{instance_name}',
headers=HEADERS,
json={'number': to_phone, 'text': message_text},
timeout=15
)
data = resp.json()
if resp.status_code in (200, 201):
wa_id = data.get("messages", [{}])[0].get("id", "")
return {"wa_message_id": wa_id}
return data
else:
err = data.get("error", {}).get("message", resp.text)
logger.error("WhatsApp send failed: %s", err)
return {"error": err}
err = data.get('message', data.get('error', resp.text))
logger.error("Evolution send failed: %s", err)
return {'error': err}
except Exception as e:
logger.exception("WhatsApp send exception")
return {"error": str(e)}
logger.exception("Evolution send exception")
return {'error': str(e)}
def send_template(to_phone, template_name, parameters=None):
"""Send a pre-approved template message (for initiating conversations).
Args:
to_phone: E.164 phone number
template_name: approved template name (e.g. 'order_confirmation')
parameters: list of {"type": "text", "text": "value"} dicts
"""
if not is_configured():
return {"error": "WhatsApp not configured"}
url = f"{BASE_URL}/{WHATSAPP_PHONE_ID}/messages"
components = []
if parameters:
components.append({
"type": "body",
"parameters": parameters,
})
payload = {
"messaging_product": "whatsapp",
"to": to_phone,
"type": "template",
"template": {
"name": template_name,
"language": {"code": "es_MX"},
"components": components,
},
}
def send_image(instance_name, to_phone, image_url, caption=''):
"""Send image message."""
try:
resp = requests.post(url, json=payload, headers=_headers(), timeout=15)
data = resp.json()
if resp.status_code in (200, 201):
wa_id = data.get("messages", [{}])[0].get("id", "")
return {"wa_message_id": wa_id}
else:
err = data.get("error", {}).get("message", resp.text)
logger.error("WhatsApp template send failed: %s", err)
return {"error": err}
resp = requests.post(
f'{EVOLUTION_API_URL}/message/sendMedia/{instance_name}',
headers=HEADERS,
json={
'number': to_phone,
'mediatype': 'image',
'media': image_url,
'caption': caption
},
timeout=30
)
return resp.json()
except Exception as e:
logger.exception("WhatsApp template send exception")
return {"error": str(e)}
logger.exception("Evolution image send exception")
return {'error': str(e)}
def send_quote(to_phone, quote_data):
"""Send a formatted quotation to a customer.
def send_document(instance_name, to_phone, doc_url, filename, caption=''):
"""Send document (PDF, etc)."""
try:
resp = requests.post(
f'{EVOLUTION_API_URL}/message/sendMedia/{instance_name}',
headers=HEADERS,
json={
'number': to_phone,
'mediatype': 'document',
'media': doc_url,
'fileName': filename,
'caption': caption
},
timeout=30
)
return resp.json()
except Exception as e:
logger.exception("Evolution document send exception")
return {'error': str(e)}
Args:
to_phone: E.164 phone number
quote_data: dict with keys: business_name, quote_number, items[], subtotal,
tax, total, validity_days, notes (optional)
"""
def send_quote(instance_name, to_phone, quote_data):
"""Send a formatted quotation to a customer."""
biz = quote_data.get("business_name", "Nexus Autoparts")
qnum = quote_data.get("quote_number", "—")
qnum = quote_data.get("quote_number", "--")
items = quote_data.get("items", [])
subtotal = quote_data.get("subtotal", 0)
tax = quote_data.get("tax", 0)
@@ -141,15 +194,15 @@ def send_quote(to_phone, quote_data):
lines = [
f"*{biz}*",
f"Cotizacion #{qnum}",
"─────────────────",
"---",
]
for it in items:
name = it.get("name", "Articulo")
qty = it.get("quantity", 1)
price = it.get("unit_price", 0)
lines.append(f" {qty}x {name} — ${price:,.2f}")
lines.append(f" {qty}x {name} -- ${price:,.2f}")
lines.append("─────────────────")
lines.append("---")
lines.append(f"Subtotal: ${subtotal:,.2f}")
lines.append(f"IVA: ${tax:,.2f}")
lines.append(f"*Total: ${total:,.2f}*")
@@ -159,46 +212,36 @@ def send_quote(to_phone, quote_data):
if notes:
lines.append(f"Nota: {notes}")
return send_message(to_phone, "\n".join(lines))
return send_message(instance_name, to_phone, "\n".join(lines))
def send_order_confirmation(to_phone, sale_data):
"""Send order / sale confirmation.
Args:
to_phone: E.164 phone number
sale_data: dict with keys: business_name, folio, items[], total, payment_method
"""
def send_order_confirmation(instance_name, to_phone, sale_data):
"""Send order / sale confirmation."""
biz = sale_data.get("business_name", "Nexus Autoparts")
folio = sale_data.get("folio", "—")
folio = sale_data.get("folio", "--")
total = sale_data.get("total", 0)
method = sale_data.get("payment_method", "efectivo")
lines = [
f"*{biz}*",
f"Confirmacion de venta #{folio}",
"─────────────────",
"---",
]
for it in sale_data.get("items", []):
name = it.get("name", "Articulo")
qty = it.get("quantity", 1)
lines.append(f" {qty}x {name}")
lines.append("─────────────────")
lines.append("---")
lines.append(f"*Total: ${total:,.2f}*")
lines.append(f"Pago: {method}")
lines.append("\nGracias por su compra!")
return send_message(to_phone, "\n".join(lines))
return send_message(instance_name, to_phone, "\n".join(lines))
def send_stock_alert(to_phone, alert_data):
"""Send stock alert to owner/manager.
Args:
to_phone: E.164 phone
alert_data: dict with keys: items[] (each has name, current_stock, min_stock)
"""
def send_stock_alert(instance_name, to_phone, alert_data):
"""Send stock alert to owner/manager."""
lines = [
"*ALERTA DE INVENTARIO*",
"Los siguientes articulos estan bajos en stock:",
@@ -211,97 +254,90 @@ def send_stock_alert(to_phone, alert_data):
lines.append(f" {name}: {current} uds (min: {minimum})")
lines.append("\nRevisa el inventario en tu POS.")
return send_message(to_phone, "\n".join(lines))
return send_message(instance_name, to_phone, "\n".join(lines))
# ── Incoming webhook processing ──────────────────────────────────────────
# -- Incoming webhook processing ----------------------------------------------
def process_incoming(webhook_data):
"""Process incoming WhatsApp webhook message.
"""Process incoming Evolution API webhook.
Routes to AI chatbot for auto-response or queues for employee.
Args:
webhook_data: the full webhook payload from Meta
Evolution sends a different format than Meta Cloud API.
Returns:
dict with parsed message info: phone, text, wa_message_id, response (if auto-replied)
dict with parsed message info.
"""
result = {"handled": False}
result = {'handled': False}
try:
entry = webhook_data.get("entry", [])
if not entry:
data = webhook_data.get('data', {})
key = data.get('key', {})
message = data.get('message', {})
from_me = key.get('fromMe', False)
phone = key.get('remoteJid', '').replace('@s.whatsapp.net', '')
if not phone:
return result
changes = entry[0].get("changes", [])
if not changes:
return result
# Extract text from different message types
text = ''
msg_type = 'text'
if 'conversation' in message:
text = message['conversation']
elif 'extendedTextMessage' in message:
text = message['extendedTextMessage'].get('text', '')
elif 'imageMessage' in message:
text = message['imageMessage'].get('caption', '[Imagen]')
msg_type = 'image'
elif 'documentMessage' in message:
text = message['documentMessage'].get('caption', '[Documento]')
msg_type = 'document'
elif 'audioMessage' in message:
text = '[Audio]'
msg_type = 'audio'
elif 'videoMessage' in message:
text = message['videoMessage'].get('caption', '[Video]')
msg_type = 'video'
elif 'contactMessage' in message:
text = '[Contacto]'
msg_type = 'contact'
elif 'locationMessage' in message:
text = '[Ubicacion]'
msg_type = 'location'
value = changes[0].get("value", {})
# Handle status updates (delivered, read, etc.)
statuses = value.get("statuses", [])
if statuses:
return {
"type": "status_update",
"statuses": statuses,
"handled": True,
}
messages = value.get("messages", [])
if not messages:
return result
msg = messages[0]
phone = msg.get("from", "")
wa_message_id = msg.get("id", "")
msg_type = msg.get("type", "text")
text = ""
if msg_type == "text":
text = msg.get("text", {}).get("body", "")
elif msg_type == "interactive":
interactive = msg.get("interactive", {})
if interactive.get("type") == "button_reply":
text = interactive.get("button_reply", {}).get("title", "")
elif interactive.get("type") == "list_reply":
text = interactive.get("list_reply", {}).get("title", "")
else:
text = f"[{msg_type} message]"
# Contact info
contacts = value.get("contacts", [])
contact_name = ""
if contacts:
profile = contacts[0].get("profile", {})
contact_name = profile.get("name", "")
# Contact info from pushName
contact_name = data.get('pushName', '')
result = {
"type": "message",
"phone": phone,
"contact_name": contact_name,
"text": text,
"message_type": msg_type,
"wa_message_id": wa_message_id,
"handled": True,
'type': 'message',
'phone': phone,
'contact_name': contact_name,
'text': text,
'message_type': msg_type,
'from_me': from_me,
'message_id': key.get('id', ''),
'timestamp': data.get('messageTimestamp', 0),
'handled': True,
}
# Skip auto-reply for our own outgoing messages
if from_me:
result['type'] = 'outgoing'
return result
# Attempt AI auto-response if the ai_chat service is available
try:
from services.ai_chat import chat as ai_chat_fn
ai_result = ai_chat_fn(text, [])
ai_reply = ai_result.get("message", "")
if ai_reply:
send_result = send_message(phone, ai_reply)
result["auto_reply"] = ai_reply
result["auto_reply_wa_id"] = send_result.get("wa_message_id", "")
except Exception:
# AI not available — message stays in queue for employee
logger.debug("AI auto-reply not available, message queued for employee")
return result
except Exception as e:
logger.exception("Error processing incoming WhatsApp message")
return {"error": str(e), "handled": False}
logger.exception("Error processing incoming Evolution webhook")
return {'error': str(e), 'handled': False}