bakery-admin/bakery-ia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add whatsapp feature

Browse Source
This commit is contained in:
Urtzi Alfaro
2025-11-13 16:01:08 +01:00
parent d7df2b0853
commit 9bc048d360
74 changed files with 9765 additions and 533 deletions
Download Patch File Download Diff File Expand all files Collapse all files

300
services/notification/app/api/whatsapp_webhooks.py Normal file
View File

@@ -0,0 +1,300 @@
# ================================================================
# services/notification/app/api/whatsapp_webhooks.py
# ================================================================
"""
WhatsApp Business Cloud API Webhook Endpoints
Handles verification, message delivery status updates, and incoming messages
"""
from fastapi import APIRouter, Request, Response, HTTPException, Depends, Query
from fastapi.responses import PlainTextResponse
from sqlalchemy.ext.asyncio import AsyncSession
import structlog
from typing import Dict, Any
from datetime import datetime
from app.core.config import settings
from app.repositories.whatsapp_message_repository import WhatsAppMessageRepository
from app.models.whatsapp_messages import WhatsAppMessageStatus
from app.core.database import get_db
from shared.monitoring.metrics import MetricsCollector
logger = structlog.get_logger()
metrics = MetricsCollector("notification-service")
router = APIRouter(prefix="/api/v1/whatsapp", tags=["whatsapp-webhooks"])
@router.get("/webhook")
async def verify_webhook(
request: Request,
hub_mode: str = Query(None, alias="hub.mode"),
hub_token: str = Query(None, alias="hub.verify_token"),
hub_challenge: str = Query(None, alias="hub.challenge")
) -> PlainTextResponse:
"""
Webhook verification endpoint for WhatsApp Cloud API
Meta sends a GET request with hub.mode, hub.verify_token, and hub.challenge
to verify the webhook URL when you configure it in the Meta Business Suite.
Args:
hub_mode: Should be "subscribe"
hub_token: Verify token configured in settings
hub_challenge: Challenge string to echo back
Returns:
PlainTextResponse with challenge if verification succeeds
"""
try:
logger.info(
"WhatsApp webhook verification request received",
mode=hub_mode,
token_provided=bool(hub_token),
challenge_provided=bool(hub_challenge)
)
# Verify the mode and token
if hub_mode == "subscribe" and hub_token == settings.WHATSAPP_WEBHOOK_VERIFY_TOKEN:
logger.info("WhatsApp webhook verification successful")
# Respond with the challenge token
return PlainTextResponse(content=hub_challenge, status_code=200)
else:
logger.warning(
"WhatsApp webhook verification failed",
mode=hub_mode,
token_match=hub_token == settings.WHATSAPP_WEBHOOK_VERIFY_TOKEN
)
raise HTTPException(status_code=403, detail="Verification token mismatch")
except Exception as e:
logger.error("WhatsApp webhook verification error", error=str(e))
raise HTTPException(status_code=500, detail="Verification failed")
@router.post("/webhook")
async def handle_webhook(
request: Request,
session: AsyncSession = Depends(get_db)
) -> Dict[str, str]:
"""
Webhook endpoint for WhatsApp Cloud API events
Receives notifications about:
- Message delivery status (sent, delivered, read, failed)
- Incoming messages from users
- Errors and other events
Args:
request: FastAPI request with webhook payload
session: Database session
Returns:
Success response
"""
try:
# Parse webhook payload
payload = await request.json()
logger.info(
"WhatsApp webhook received",
object_type=payload.get("object"),
entries_count=len(payload.get("entry", []))
)
# Verify it's a WhatsApp webhook
if payload.get("object") != "whatsapp_business_account":
logger.warning("Unknown webhook object type", object_type=payload.get("object"))
return {"status": "ignored"}
# Process each entry
for entry in payload.get("entry", []):
entry_id = entry.get("id")
for change in entry.get("changes", []):
field = change.get("field")
value = change.get("value", {})
if field == "messages":
# Handle incoming messages or status updates
await _handle_message_change(value, session)
else:
logger.debug("Unhandled webhook field", field=field)
# Record metric
metrics.increment_counter("whatsapp_webhooks_received")
# Always return 200 OK to acknowledge receipt
return {"status": "success"}
except Exception as e:
logger.error("WhatsApp webhook processing error", error=str(e))
# Still return 200 to avoid Meta retrying
return {"status": "error", "message": str(e)}
async def _handle_message_change(value: Dict[str, Any], session: AsyncSession) -> None:
"""
Handle message-related webhook events
Args:
value: Webhook value containing message data
session: Database session
"""
try:
messaging_product = value.get("messaging_product")
metadata = value.get("metadata", {})
# Handle status updates
statuses = value.get("statuses", [])
if statuses:
await _handle_status_updates(statuses, session)
# Handle incoming messages
messages = value.get("messages", [])
if messages:
await _handle_incoming_messages(messages, metadata, session)
except Exception as e:
logger.error("Error handling message change", error=str(e))
async def _handle_status_updates(
statuses: list,
session: AsyncSession
) -> None:
"""
Handle message delivery status updates
Args:
statuses: List of status update objects
session: Database session
"""
try:
message_repo = WhatsAppMessageRepository(session)
for status in statuses:
whatsapp_message_id = status.get("id")
status_value = status.get("status") # sent, delivered, read, failed
timestamp = status.get("timestamp")
errors = status.get("errors", [])
logger.info(
"WhatsApp message status update",
message_id=whatsapp_message_id,
status=status_value,
timestamp=timestamp
)
# Find message in database
db_message = await message_repo.get_by_whatsapp_id(whatsapp_message_id)
if not db_message:
logger.warning(
"Received status for unknown message",
whatsapp_message_id=whatsapp_message_id
)
continue
# Map WhatsApp status to our enum
status_mapping = {
"sent": WhatsAppMessageStatus.SENT,
"delivered": WhatsAppMessageStatus.DELIVERED,
"read": WhatsAppMessageStatus.READ,
"failed": WhatsAppMessageStatus.FAILED
}
new_status = status_mapping.get(status_value)
if not new_status:
logger.warning("Unknown status value", status=status_value)
continue
# Extract error information if failed
error_message = None
error_code = None
if errors:
error = errors[0]
error_code = error.get("code")
error_message = error.get("title", error.get("message"))
# Update message status
await message_repo.update_message_status(
message_id=str(db_message.id),
status=new_status,
error_message=error_message,
provider_response=status
)
# Record metric
metrics.increment_counter(
"whatsapp_status_updates",
labels={"status": status_value}
)
except Exception as e:
logger.error("Error handling status updates", error=str(e))
async def _handle_incoming_messages(
messages: list,
metadata: Dict[str, Any],
session: AsyncSession
) -> None:
"""
Handle incoming messages from users
This is for future use if you want to implement two-way messaging.
For now, we just log incoming messages.
Args:
messages: List of message objects
metadata: Metadata about the phone number
session: Database session
"""
try:
for message in messages:
message_id = message.get("id")
from_number = message.get("from")
message_type = message.get("type")
timestamp = message.get("timestamp")
# Extract message content based on type
content = None
if message_type == "text":
content = message.get("text", {}).get("body")
elif message_type == "image":
content = message.get("image", {}).get("caption")
logger.info(
"Incoming WhatsApp message",
message_id=message_id,
from_number=from_number,
message_type=message_type,
content=content[:100] if content else None
)
# Record metric
metrics.increment_counter(
"whatsapp_incoming_messages",
labels={"type": message_type}
)
# TODO: Implement incoming message handling logic
# For example:
# - Create a new conversation session
# - Route to customer support
# - Auto-reply with acknowledgment
except Exception as e:
logger.error("Error handling incoming messages", error=str(e))
@router.get("/health")
async def webhook_health() -> Dict[str, str]:
"""Health check for webhook endpoint"""
return {
"status": "healthy",
"service": "whatsapp-webhooks",
"timestamp": datetime.utcnow().isoformat()
}

93
services/notification/app/consumers/po_event_consumer.py
View File

@@ -11,6 +11,7 @@ from datetime import datetime
from shared.messaging.rabbitmq import RabbitMQClient
from app.services.email_service import EmailService
from app.services.whatsapp_service import WhatsAppService
logger = structlog.get_logger()
@@ -18,10 +19,12 @@ logger = structlog.get_logger()
class POEventConsumer:
"""
Consumes purchase order events from RabbitMQ and sends notifications
Sends both email and WhatsApp notifications to suppliers
"""
def __init__(self, email_service: EmailService):
def __init__(self, email_service: EmailService, whatsapp_service: WhatsAppService = None):
self.email_service = email_service
self.whatsapp_service = whatsapp_service
# Setup Jinja2 template environment
template_dir = Path(__file__).parent.parent / 'templates'
@@ -50,17 +53,24 @@ class POEventConsumer:
)
# Send notification email
success = await self.send_po_approved_email(event_data)
email_success = await self.send_po_approved_email(event_data)
if success:
# Send WhatsApp notification if service is available
whatsapp_success = False
if self.whatsapp_service:
whatsapp_success = await self.send_po_approved_whatsapp(event_data)
if email_success:
logger.info(
"PO approved email sent successfully",
po_id=event_data.get('data', {}).get('po_id')
po_id=event_data.get('data', {}).get('po_id'),
whatsapp_sent=whatsapp_success
)
else:
logger.error(
"Failed to send PO approved email",
po_id=event_data.get('data', {}).get('po_id')
po_id=event_data.get('data', {}).get('po_id'),
whatsapp_sent=whatsapp_success
)
except Exception as e:
@@ -276,3 +286,76 @@ This is an automated email from your Bakery Management System.
return dt.strftime('%B %d, %Y')
except Exception:
return iso_date
async def send_po_approved_whatsapp(self, event_data: Dict[str, Any]) -> bool:
"""
Send PO approved WhatsApp notification to supplier
This sends a WhatsApp Business template message notifying the supplier
of a new purchase order. The template must be pre-approved in Meta Business Suite.
Args:
event_data: Full event payload from RabbitMQ
Returns:
bool: True if WhatsApp message sent successfully
"""
try:
# Extract data from event
data = event_data.get('data', {})
# Check for supplier phone number
supplier_phone = data.get('supplier_phone')
if not supplier_phone:
logger.debug(
"No supplier phone in event, skipping WhatsApp notification",
po_id=data.get('po_id')
)
return False
# Extract tenant ID for tracking
tenant_id = data.get('tenant_id')
# Prepare template parameters
# Template: "Hola {{1}}, has recibido una nueva orden de compra {{2}} por un total de {{3}}."
# Parameters: supplier_name, po_number, total_amount
template_params = [
data.get('supplier_name', 'Estimado proveedor'),
data.get('po_number', 'N/A'),
f"€{data.get('total_amount', 0):.2f}"
]
# Send WhatsApp template message
# The template must be named 'po_notification' and approved in Meta Business Suite
success = await self.whatsapp_service.send_message(
to_phone=supplier_phone,
message="", # Not used for template messages
template_name="po_notification", # Must match template name in Meta
template_params=template_params,
tenant_id=tenant_id
)
if success:
logger.info(
"PO approved WhatsApp sent successfully",
po_id=data.get('po_id'),
supplier_phone=supplier_phone,
template="po_notification"
)
else:
logger.warning(
"Failed to send PO approved WhatsApp",
po_id=data.get('po_id'),
supplier_phone=supplier_phone
)
return success
except Exception as e:
logger.error(
"Error sending PO approved WhatsApp",
error=str(e),
po_id=data.get('po_id'),
exc_info=True
)
return False

15
services/notification/app/core/config.py
View File

@@ -53,11 +53,18 @@ class NotificationSettings(BaseServiceSettings):
DEFAULT_FROM_NAME: str = os.getenv("DEFAULT_FROM_NAME", "Bakery Forecast")
EMAIL_TEMPLATES_PATH: str = os.getenv("EMAIL_TEMPLATES_PATH", "/app/templates/email")
# WhatsApp Configuration
WHATSAPP_API_KEY: str = os.getenv("WHATSAPP_API_KEY", "")
WHATSAPP_BASE_URL: str = os.getenv("WHATSAPP_BASE_URL", "https://api.twilio.com")
WHATSAPP_FROM_NUMBER: str = os.getenv("WHATSAPP_FROM_NUMBER", "")
# WhatsApp Business Cloud API Configuration (Meta/Facebook)
WHATSAPP_ACCESS_TOKEN: str = os.getenv("WHATSAPP_ACCESS_TOKEN", "")
WHATSAPP_PHONE_NUMBER_ID: str = os.getenv("WHATSAPP_PHONE_NUMBER_ID", "")
WHATSAPP_BUSINESS_ACCOUNT_ID: str = os.getenv("WHATSAPP_BUSINESS_ACCOUNT_ID", "")
WHATSAPP_API_VERSION: str = os.getenv("WHATSAPP_API_VERSION", "v18.0")
WHATSAPP_WEBHOOK_VERIFY_TOKEN: str = os.getenv("WHATSAPP_WEBHOOK_VERIFY_TOKEN", "")
WHATSAPP_TEMPLATES_PATH: str = os.getenv("WHATSAPP_TEMPLATES_PATH", "/app/templates/whatsapp")
# Legacy Twilio Configuration (deprecated, for backward compatibility)
WHATSAPP_API_KEY: str = os.getenv("WHATSAPP_API_KEY", "") # Deprecated
WHATSAPP_BASE_URL: str = os.getenv("WHATSAPP_BASE_URL", "https://api.twilio.com") # Deprecated
WHATSAPP_FROM_NUMBER: str = os.getenv("WHATSAPP_FROM_NUMBER", "") # Deprecated
# Notification Queuing
MAX_RETRY_ATTEMPTS: int = int(os.getenv("MAX_RETRY_ATTEMPTS", "3"))

19
services/notification/app/main.py
View File

@@ -14,6 +14,7 @@ from app.api.notifications import router as notification_router
from app.api.notification_operations import router as notification_operations_router
from app.api.analytics import router as analytics_router
from app.api.audit import router as audit_router
from app.api.whatsapp_webhooks import router as whatsapp_webhooks_router
from app.services.messaging import setup_messaging, cleanup_messaging
from app.services.sse_service import SSEService
from app.services.notification_orchestrator import NotificationOrchestrator
@@ -21,13 +22,14 @@ from app.services.email_service import EmailService
from app.services.whatsapp_service import WhatsAppService
from app.consumers.po_event_consumer import POEventConsumer
from shared.service_base import StandardFastAPIService
from shared.clients.tenant_client import TenantServiceClient
import asyncio
class NotificationService(StandardFastAPIService):
"""Notification Service with standardized setup"""
expected_migration_version = "359991e24ea2"
expected_migration_version = "whatsapp001"
async def verify_migrations(self):
"""Verify database schema matches the latest migrations."""
@@ -47,13 +49,14 @@ class NotificationService(StandardFastAPIService):
# Define expected database tables for health checks
notification_expected_tables = [
'notifications', 'notification_templates', 'notification_preferences',
'notification_logs', 'email_templates', 'whatsapp_templates'
'notification_logs', 'email_templates', 'whatsapp_messages', 'whatsapp_templates'
]
self.sse_service = None
self.orchestrator = None
self.email_service = None
self.whatsapp_service = None
self.tenant_client = None
self.po_consumer = None
self.po_consumer_task = None
@@ -172,9 +175,13 @@ class NotificationService(StandardFastAPIService):
# Call parent startup (includes database, messaging, etc.)
await super().on_startup(app)
# Initialize tenant client for fetching tenant-specific settings
self.tenant_client = TenantServiceClient(settings)
self.logger.info("Tenant service client initialized")
# Initialize services
self.email_service = EmailService()
self.whatsapp_service = WhatsAppService()
self.whatsapp_service = WhatsAppService(tenant_client=self.tenant_client)
# Initialize SSE service
self.sse_service = SSEService()
@@ -195,7 +202,10 @@ class NotificationService(StandardFastAPIService):
app.state.whatsapp_service = self.whatsapp_service
# Initialize and start PO event consumer
self.po_consumer = POEventConsumer(self.email_service)
self.po_consumer = POEventConsumer(
email_service=self.email_service,
whatsapp_service=self.whatsapp_service
)
# Start consuming PO approved events in background
# Use the global notification_publisher from messaging module
@@ -284,6 +294,7 @@ service.setup_custom_endpoints()
# IMPORTANT: Register audit router FIRST to avoid route matching conflicts
# where {notification_id} would match literal paths like "audit-logs"
service.add_router(audit_router, tags=["audit-logs"])
service.add_router(whatsapp_webhooks_router, tags=["whatsapp-webhooks"])
service.add_router(notification_operations_router, tags=["notification-operations"])
service.add_router(analytics_router, tags=["notifications-analytics"])
service.add_router(notification_router, tags=["notifications"])

8
services/notification/app/models/__init__.py
View File

@@ -23,7 +23,12 @@ from .notifications import (
)
from .templates import (
EmailTemplate,
)
from .whatsapp_messages import (
WhatsAppTemplate,
WhatsAppMessage,
WhatsAppMessageStatus,
WhatsAppMessageType,
)
# List all models for easier access
@@ -37,5 +42,8 @@ __all__ = [
"NotificationLog",
"EmailTemplate",
"WhatsAppTemplate",
"WhatsAppMessage",
"WhatsAppMessageStatus",
"WhatsAppMessageType",
"AuditLog",
]

66
services/notification/app/models/templates.py
View File

@@ -48,35 +48,37 @@ class EmailTemplate(Base):
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
class WhatsAppTemplate(Base):
"""WhatsApp-specific templates"""
__tablename__ = "whatsapp_templates"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
tenant_id = Column(UUID(as_uuid=True), nullable=True, index=True)
# Template identification
template_key = Column(String(100), nullable=False, unique=True)
name = Column(String(255), nullable=False)
# WhatsApp template details
whatsapp_template_name = Column(String(255), nullable=False) # Template name in WhatsApp Business API
whatsapp_template_id = Column(String(255), nullable=True)
language_code = Column(String(10), default="es")
# Template content
header_text = Column(String(60), nullable=True) # WhatsApp header limit
body_text = Column(Text, nullable=False)
footer_text = Column(String(60), nullable=True) # WhatsApp footer limit
# Template parameters
parameter_count = Column(Integer, default=0)
parameters = Column(JSON, nullable=True) # Parameter definitions
# Status
approval_status = Column(String(20), default="pending") # pending, approved, rejected
is_active = Column(Boolean, default=True)
# Timestamps
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# NOTE: WhatsAppTemplate has been moved to app/models/whatsapp_messages.py
# This old definition is commented out to avoid duplicate table definition errors
# class WhatsAppTemplate(Base):
# """WhatsApp-specific templates"""
# __tablename__ = "whatsapp_templates"
#
# id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
# tenant_id = Column(UUID(as_uuid=True), nullable=True, index=True)
#
# # Template identification
# template_key = Column(String(100), nullable=False, unique=True)
# name = Column(String(255), nullable=False)
#
# # WhatsApp template details
# whatsapp_template_name = Column(String(255), nullable=False) # Template name in WhatsApp Business API
# whatsapp_template_id = Column(String(255), nullable=True)
# language_code = Column(String(10), default="es")
#
# # Template content
# header_text = Column(String(60), nullable=True) # WhatsApp header limit
# body_text = Column(Text, nullable=False)
# footer_text = Column(String(60), nullable=True) # WhatsApp footer limit
#
# # Template parameters
# parameter_count = Column(Integer, default=0)
# parameters = Column(JSON, nullable=True) # Parameter definitions
#
# # Status
# approval_status = Column(String(20), default="pending") # pending, approved, rejected
# is_active = Column(Boolean, default=True)
#
# # Timestamps
# created_at = Column(DateTime, default=datetime.utcnow)
# updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

135
services/notification/app/models/whatsapp_messages.py Normal file
View File

@@ -0,0 +1,135 @@
# ================================================================
# services/notification/app/models/whatsapp_messages.py
# ================================================================
"""
WhatsApp message tracking models for WhatsApp Business Cloud API
"""
from sqlalchemy import Column, String, Text, Boolean, DateTime, JSON, Enum, Integer
from sqlalchemy.dialects.postgresql import UUID
from datetime import datetime
import uuid
import enum
from shared.database.base import Base
class WhatsAppMessageStatus(enum.Enum):
"""WhatsApp message delivery status"""
PENDING = "pending"
SENT = "sent"
DELIVERED = "delivered"
READ = "read"
FAILED = "failed"
class WhatsAppMessageType(enum.Enum):
"""WhatsApp message types"""
TEMPLATE = "template"
TEXT = "text"
IMAGE = "image"
DOCUMENT = "document"
INTERACTIVE = "interactive"
class WhatsAppMessage(Base):
"""Track WhatsApp messages sent via Cloud API"""
__tablename__ = "whatsapp_messages"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True)
notification_id = Column(UUID(as_uuid=True), nullable=True, index=True) # Link to notification if exists
# Message identification
whatsapp_message_id = Column(String(255), nullable=True, index=True) # WhatsApp's message ID
# Recipient details
recipient_phone = Column(String(20), nullable=False, index=True) # E.164 format
recipient_name = Column(String(255), nullable=True)
# Message details
message_type = Column(Enum(WhatsAppMessageType), nullable=False)
status = Column(Enum(WhatsAppMessageStatus), default=WhatsAppMessageStatus.PENDING, index=True)
# Template details (for template messages)
template_name = Column(String(255), nullable=True)
template_language = Column(String(10), default="es")
template_parameters = Column(JSON, nullable=True) # Template variable values
# Message content (for non-template messages)
message_body = Column(Text, nullable=True)
media_url = Column(String(512), nullable=True)
# Delivery tracking
sent_at = Column(DateTime, nullable=True)
delivered_at = Column(DateTime, nullable=True)
read_at = Column(DateTime, nullable=True)
failed_at = Column(DateTime, nullable=True)
# Error tracking
error_code = Column(String(50), nullable=True)
error_message = Column(Text, nullable=True)
# Provider response
provider_response = Column(JSON, nullable=True)
# Additional data (renamed from metadata to avoid SQLAlchemy reserved word)
additional_data = Column(JSON, nullable=True) # Additional context (PO number, order ID, etc.)
# Conversation tracking
conversation_id = Column(String(255), nullable=True, index=True) # WhatsApp conversation ID
conversation_category = Column(String(50), nullable=True) # business_initiated, user_initiated
# Timestamps
created_at = Column(DateTime, default=datetime.utcnow, index=True)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
class WhatsAppTemplate(Base):
"""Store WhatsApp message templates metadata"""
__tablename__ = "whatsapp_templates"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
tenant_id = Column(UUID(as_uuid=True), nullable=True, index=True) # Null for system templates
# Template identification
template_name = Column(String(255), nullable=False, index=True) # Name in WhatsApp
template_key = Column(String(100), nullable=False, unique=True) # Internal key
display_name = Column(String(255), nullable=False)
description = Column(Text, nullable=True)
category = Column(String(50), nullable=False) # MARKETING, UTILITY, AUTHENTICATION
# Template configuration
language = Column(String(10), default="es")
status = Column(String(20), default="PENDING") # PENDING, APPROVED, REJECTED
# Template structure
header_type = Column(String(20), nullable=True) # TEXT, IMAGE, DOCUMENT, VIDEO
header_text = Column(String(60), nullable=True)
body_text = Column(Text, nullable=False)
footer_text = Column(String(60), nullable=True)
# Parameters
parameters = Column(JSON, nullable=True) # List of parameter definitions
parameter_count = Column(Integer, default=0)
# Buttons (for interactive templates)
buttons = Column(JSON, nullable=True)
# Metadata
is_active = Column(Boolean, default=True)
is_system = Column(Boolean, default=False)
# Usage tracking
sent_count = Column(Integer, default=0)
last_used_at = Column(DateTime, nullable=True)
# WhatsApp metadata
whatsapp_template_id = Column(String(255), nullable=True)
approved_at = Column(DateTime, nullable=True)
rejected_at = Column(DateTime, nullable=True)
rejection_reason = Column(Text, nullable=True)
# Timestamps
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

379
services/notification/app/repositories/whatsapp_message_repository.py Normal file
View File

@@ -0,0 +1,379 @@
"""
WhatsApp Message Repository
"""
from typing import Optional, List, Dict, Any
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text, select, and_
from datetime import datetime, timedelta
import structlog
from app.repositories.base import NotificationBaseRepository
from app.models.whatsapp_messages import WhatsAppMessage, WhatsAppMessageStatus, WhatsAppTemplate
from shared.database.exceptions import DatabaseError
logger = structlog.get_logger()
class WhatsAppMessageRepository(NotificationBaseRepository):
"""Repository for WhatsApp message operations"""
def __init__(self, session: AsyncSession):
super().__init__(WhatsAppMessage, session, cache_ttl=60) # 1 minute cache
async def create_message(self, message_data: Dict[str, Any]) -> WhatsAppMessage:
"""Create a new WhatsApp message record"""
try:
# Validate required fields
validation = self._validate_notification_data(
message_data,
["tenant_id", "recipient_phone", "message_type"]
)
if not validation["is_valid"]:
raise DatabaseError(f"Validation failed: {', '.join(validation['errors'])}")
message = await self.create(message_data)
logger.info(
"WhatsApp message created",
message_id=str(message.id),
recipient=message.recipient_phone,
message_type=message.message_type.value
)
return message
except Exception as e:
logger.error("Failed to create WhatsApp message", error=str(e))
raise DatabaseError(f"Failed to create message: {str(e)}")
async def update_message_status(
self,
message_id: str,
status: WhatsAppMessageStatus,
whatsapp_message_id: Optional[str] = None,
error_message: Optional[str] = None,
provider_response: Optional[Dict] = None
) -> Optional[WhatsAppMessage]:
"""Update message status and related fields"""
try:
update_data = {
"status": status,
"updated_at": datetime.utcnow()
}
# Update timestamps based on status
if status == WhatsAppMessageStatus.SENT:
update_data["sent_at"] = datetime.utcnow()
elif status == WhatsAppMessageStatus.DELIVERED:
update_data["delivered_at"] = datetime.utcnow()
elif status == WhatsAppMessageStatus.READ:
update_data["read_at"] = datetime.utcnow()
elif status == WhatsAppMessageStatus.FAILED:
update_data["failed_at"] = datetime.utcnow()
if whatsapp_message_id:
update_data["whatsapp_message_id"] = whatsapp_message_id
if error_message:
update_data["error_message"] = error_message
if provider_response:
update_data["provider_response"] = provider_response
message = await self.update(message_id, update_data)
logger.info(
"WhatsApp message status updated",
message_id=message_id,
status=status.value,
whatsapp_message_id=whatsapp_message_id
)
return message
except Exception as e:
logger.error(
"Failed to update message status",
message_id=message_id,
error=str(e)
)
return None
async def get_by_whatsapp_id(self, whatsapp_message_id: str) -> Optional[WhatsAppMessage]:
"""Get message by WhatsApp's message ID"""
try:
messages = await self.get_multi(
filters={"whatsapp_message_id": whatsapp_message_id},
limit=1
)
return messages[0] if messages else None
except Exception as e:
logger.error(
"Failed to get message by WhatsApp ID",
whatsapp_message_id=whatsapp_message_id,
error=str(e)
)
return None
async def get_by_notification_id(self, notification_id: str) -> Optional[WhatsAppMessage]:
"""Get message by notification ID"""
try:
messages = await self.get_multi(
filters={"notification_id": notification_id},
limit=1
)
return messages[0] if messages else None
except Exception as e:
logger.error(
"Failed to get message by notification ID",
notification_id=notification_id,
error=str(e)
)
return None
async def get_messages_by_phone(
self,
tenant_id: str,
phone: str,
skip: int = 0,
limit: int = 50
) -> List[WhatsAppMessage]:
"""Get all messages for a specific phone number"""
try:
return await self.get_multi(
filters={"tenant_id": tenant_id, "recipient_phone": phone},
skip=skip,
limit=limit,
order_by="created_at",
order_desc=True
)
except Exception as e:
logger.error(
"Failed to get messages by phone",
phone=phone,
error=str(e)
)
return []
async def get_pending_messages(
self,
tenant_id: str,
limit: int = 100
) -> List[WhatsAppMessage]:
"""Get pending messages for retry processing"""
try:
return await self.get_multi(
filters={
"tenant_id": tenant_id,
"status": WhatsAppMessageStatus.PENDING
},
limit=limit,
order_by="created_at",
order_desc=False # Oldest first
)
except Exception as e:
logger.error("Failed to get pending messages", error=str(e))
return []
async def get_conversation_messages(
self,
conversation_id: str,
skip: int = 0,
limit: int = 50
) -> List[WhatsAppMessage]:
"""Get all messages in a conversation"""
try:
return await self.get_multi(
filters={"conversation_id": conversation_id},
skip=skip,
limit=limit,
order_by="created_at",
order_desc=False # Chronological order
)
except Exception as e:
logger.error(
"Failed to get conversation messages",
conversation_id=conversation_id,
error=str(e)
)
return []
async def get_delivery_stats(
self,
tenant_id: str,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None
) -> Dict[str, Any]:
"""Get delivery statistics for WhatsApp messages"""
try:
# Default to last 30 days
if not start_date:
start_date = datetime.utcnow() - timedelta(days=30)
if not end_date:
end_date = datetime.utcnow()
query = text("""
SELECT
COUNT(*) as total_messages,
COUNT(CASE WHEN status = 'SENT' THEN 1 END) as sent,
COUNT(CASE WHEN status = 'DELIVERED' THEN 1 END) as delivered,
COUNT(CASE WHEN status = 'READ' THEN 1 END) as read,
COUNT(CASE WHEN status = 'FAILED' THEN 1 END) as failed,
COUNT(CASE WHEN status = 'PENDING' THEN 1 END) as pending,
COUNT(DISTINCT recipient_phone) as unique_recipients,
COUNT(DISTINCT conversation_id) as total_conversations
FROM whatsapp_messages
WHERE tenant_id = :tenant_id
AND created_at BETWEEN :start_date AND :end_date
""")
result = await self.session.execute(
query,
{
"tenant_id": tenant_id,
"start_date": start_date,
"end_date": end_date
}
)
row = result.fetchone()
if row:
total = row.total_messages or 0
delivered = row.delivered or 0
return {
"total_messages": total,
"sent": row.sent or 0,
"delivered": delivered,
"read": row.read or 0,
"failed": row.failed or 0,
"pending": row.pending or 0,
"unique_recipients": row.unique_recipients or 0,
"total_conversations": row.total_conversations or 0,
"delivery_rate": round((delivered / total * 100), 2) if total > 0 else 0,
"period": {
"start": start_date.isoformat(),
"end": end_date.isoformat()
}
}
return {
"total_messages": 0,
"sent": 0,
"delivered": 0,
"read": 0,
"failed": 0,
"pending": 0,
"unique_recipients": 0,
"total_conversations": 0,
"delivery_rate": 0,
"period": {
"start": start_date.isoformat(),
"end": end_date.isoformat()
}
}
except Exception as e:
logger.error("Failed to get delivery stats", error=str(e))
return {}
class WhatsAppTemplateRepository(NotificationBaseRepository):
"""Repository for WhatsApp template operations"""
def __init__(self, session: AsyncSession):
super().__init__(WhatsAppTemplate, session, cache_ttl=300) # 5 minute cache
async def get_by_template_name(
self,
template_name: str,
language: str = "es"
) -> Optional[WhatsAppTemplate]:
"""Get template by name and language"""
try:
templates = await self.get_multi(
filters={
"template_name": template_name,
"language": language,
"is_active": True
},
limit=1
)
return templates[0] if templates else None
except Exception as e:
logger.error(
"Failed to get template by name",
template_name=template_name,
error=str(e)
)
return None
async def get_by_template_key(self, template_key: str) -> Optional[WhatsAppTemplate]:
"""Get template by internal key"""
try:
templates = await self.get_multi(
filters={"template_key": template_key},
limit=1
)
return templates[0] if templates else None
except Exception as e:
logger.error(
"Failed to get template by key",
template_key=template_key,
error=str(e)
)
return None
async def get_active_templates(
self,
tenant_id: Optional[str] = None,
category: Optional[str] = None
) -> List[WhatsAppTemplate]:
"""Get all active templates"""
try:
filters = {"is_active": True, "status": "APPROVED"}
if tenant_id:
filters["tenant_id"] = tenant_id
if category:
filters["category"] = category
return await self.get_multi(
filters=filters,
limit=1000,
order_by="created_at",
order_desc=True
)
except Exception as e:
logger.error("Failed to get active templates", error=str(e))
return []
async def increment_usage(self, template_id: str) -> None:
"""Increment template usage counter"""
try:
template = await self.get(template_id)
if template:
await self.update(
template_id,
{
"sent_count": (template.sent_count or 0) + 1,
"last_used_at": datetime.utcnow()
}
)
except Exception as e:
logger.error(
"Failed to increment template usage",
template_id=template_id,
error=str(e)
)

370
services/notification/app/schemas/whatsapp.py Normal file
View File

@@ -0,0 +1,370 @@
"""
WhatsApp Business Cloud API Schemas
"""
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Dict, Any
from datetime import datetime
from enum import Enum
# ============================================================
# Enums
# ============================================================
class WhatsAppMessageType(str, Enum):
"""WhatsApp message types"""
TEMPLATE = "template"
TEXT = "text"
IMAGE = "image"
DOCUMENT = "document"
INTERACTIVE = "interactive"
class WhatsAppMessageStatus(str, Enum):
"""WhatsApp message delivery status"""
PENDING = "pending"
SENT = "sent"
DELIVERED = "delivered"
READ = "read"
FAILED = "failed"
class TemplateCategory(str, Enum):
"""WhatsApp template categories"""
MARKETING = "MARKETING"
UTILITY = "UTILITY"
AUTHENTICATION = "AUTHENTICATION"
# ============================================================
# Template Message Schemas
# ============================================================
class TemplateParameter(BaseModel):
"""Template parameter for dynamic content"""
type: str = Field(default="text", description="Parameter type (text, currency, date_time)")
text: Optional[str] = Field(None, description="Text value for the parameter")
class Config:
json_schema_extra = {
"example": {
"type": "text",
"text": "PO-2024-001"
}
}
class TemplateComponent(BaseModel):
"""Template component (header, body, buttons)"""
type: str = Field(..., description="Component type (header, body, button)")
parameters: Optional[List[TemplateParameter]] = Field(None, description="Component parameters")
sub_type: Optional[str] = Field(None, description="Button sub_type (quick_reply, url)")
index: Optional[int] = Field(None, description="Button index")
class Config:
json_schema_extra = {
"example": {
"type": "body",
"parameters": [
{"type": "text", "text": "PO-2024-001"},
{"type": "text", "text": "100.50"}
]
}
}
class TemplateMessageRequest(BaseModel):
"""Request to send a template message"""
template_name: str = Field(..., description="WhatsApp template name")
language: str = Field(default="es", description="Template language code")
components: List[TemplateComponent] = Field(..., description="Template components with parameters")
class Config:
json_schema_extra = {
"example": {
"template_name": "po_notification",
"language": "es",
"components": [
{
"type": "body",
"parameters": [
{"type": "text", "text": "PO-2024-001"},
{"type": "text", "text": "Supplier XYZ"},
{"type": "text", "text": "€1,250.00"}
]
}
]
}
}
# ============================================================
# Send Message Schemas
# ============================================================
class SendWhatsAppMessageRequest(BaseModel):
"""Request to send a WhatsApp message"""
tenant_id: str = Field(..., description="Tenant ID")
recipient_phone: str = Field(..., description="Recipient phone number (E.164 format)")
recipient_name: Optional[str] = Field(None, description="Recipient name")
message_type: WhatsAppMessageType = Field(..., description="Message type")
template: Optional[TemplateMessageRequest] = Field(None, description="Template details (required for template messages)")
text: Optional[str] = Field(None, description="Text message body (for text messages)")
media_url: Optional[str] = Field(None, description="Media URL (for image/document messages)")
metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata (PO number, order ID, etc.)")
notification_id: Optional[str] = Field(None, description="Link to existing notification")
@validator('recipient_phone')
def validate_phone(cls, v):
"""Validate E.164 phone format"""
if not v.startswith('+'):
raise ValueError('Phone number must be in E.164 format (starting with +)')
if len(v) < 10 or len(v) > 16:
raise ValueError('Phone number length must be between 10 and 16 characters')
return v
@validator('template')
def validate_template(cls, v, values):
"""Validate template is provided for template messages"""
if values.get('message_type') == WhatsAppMessageType.TEMPLATE and not v:
raise ValueError('Template details required for template messages')
return v
class Config:
json_schema_extra = {
"example": {
"tenant_id": "123e4567-e89b-12d3-a456-426614174000",
"recipient_phone": "+34612345678",
"recipient_name": "Supplier ABC",
"message_type": "template",
"template": {
"template_name": "po_notification",
"language": "es",
"components": [
{
"type": "body",
"parameters": [
{"type": "text", "text": "PO-2024-001"},
{"type": "text", "text": "€1,250.00"}
]
}
]
},
"metadata": {
"po_number": "PO-2024-001",
"po_id": "123e4567-e89b-12d3-a456-426614174111"
}
}
}
class SendWhatsAppMessageResponse(BaseModel):
"""Response after sending a WhatsApp message"""
success: bool = Field(..., description="Whether message was sent successfully")
message_id: str = Field(..., description="Internal message ID")
whatsapp_message_id: Optional[str] = Field(None, description="WhatsApp's message ID")
status: WhatsAppMessageStatus = Field(..., description="Message status")
error_message: Optional[str] = Field(None, description="Error message if failed")
class Config:
json_schema_extra = {
"example": {
"success": True,
"message_id": "123e4567-e89b-12d3-a456-426614174222",
"whatsapp_message_id": "wamid.HBgNMzQ2MTIzNDU2Nzg5FQIAERgSMzY5RTFFNTdEQzZBRkVCODdBAA==",
"status": "sent",
"error_message": None
}
}
# ============================================================
# Webhook Schemas
# ============================================================
class WebhookValue(BaseModel):
"""Webhook notification value"""
messaging_product: str
metadata: Dict[str, Any]
contacts: Optional[List[Dict[str, Any]]] = None
messages: Optional[List[Dict[str, Any]]] = None
statuses: Optional[List[Dict[str, Any]]] = None
class WebhookEntry(BaseModel):
"""Webhook entry"""
id: str
changes: List[Dict[str, Any]]
class WhatsAppWebhook(BaseModel):
"""WhatsApp webhook payload"""
object: str
entry: List[WebhookEntry]
class WebhookVerification(BaseModel):
"""Webhook verification request"""
mode: str = Field(..., alias="hub.mode")
token: str = Field(..., alias="hub.verify_token")
challenge: str = Field(..., alias="hub.challenge")
class Config:
populate_by_name = True
# ============================================================
# Message Status Schemas
# ============================================================
class MessageStatusUpdate(BaseModel):
"""Message status update"""
whatsapp_message_id: str = Field(..., description="WhatsApp message ID")
status: WhatsAppMessageStatus = Field(..., description="New status")
timestamp: datetime = Field(..., description="Status update timestamp")
error_code: Optional[str] = Field(None, description="Error code if failed")
error_message: Optional[str] = Field(None, description="Error message if failed")
# ============================================================
# Template Management Schemas
# ============================================================
class WhatsAppTemplateCreate(BaseModel):
"""Create a WhatsApp template"""
tenant_id: Optional[str] = Field(None, description="Tenant ID (null for system templates)")
template_name: str = Field(..., description="Template name in WhatsApp")
template_key: str = Field(..., description="Internal template key")
display_name: str = Field(..., description="Display name")
description: Optional[str] = Field(None, description="Template description")
category: TemplateCategory = Field(..., description="Template category")
language: str = Field(default="es", description="Template language")
header_type: Optional[str] = Field(None, description="Header type (TEXT, IMAGE, DOCUMENT, VIDEO)")
header_text: Optional[str] = Field(None, max_length=60, description="Header text (max 60 chars)")
body_text: str = Field(..., description="Body text with {{1}}, {{2}} placeholders")
footer_text: Optional[str] = Field(None, max_length=60, description="Footer text (max 60 chars)")
parameters: Optional[List[Dict[str, Any]]] = Field(None, description="Parameter definitions")
buttons: Optional[List[Dict[str, Any]]] = Field(None, description="Button definitions")
class Config:
json_schema_extra = {
"example": {
"template_name": "po_notification",
"template_key": "po_notification_v1",
"display_name": "Purchase Order Notification",
"description": "Notify supplier of new purchase order",
"category": "UTILITY",
"language": "es",
"body_text": "Hola {{1}}, has recibido una nueva orden de compra {{2}} por un total de {{3}}.",
"parameters": [
{"name": "supplier_name", "example": "Proveedor ABC"},
{"name": "po_number", "example": "PO-2024-001"},
{"name": "total_amount", "example": "€1,250.00"}
]
}
}
class WhatsAppTemplateResponse(BaseModel):
"""WhatsApp template response"""
id: str
tenant_id: Optional[str]
template_name: str
template_key: str
display_name: str
description: Optional[str]
category: str
language: str
status: str
body_text: str
parameter_count: int
is_active: bool
sent_count: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
json_schema_extra = {
"example": {
"id": "123e4567-e89b-12d3-a456-426614174333",
"tenant_id": None,
"template_name": "po_notification",
"template_key": "po_notification_v1",
"display_name": "Purchase Order Notification",
"description": "Notify supplier of new purchase order",
"category": "UTILITY",
"language": "es",
"status": "APPROVED",
"body_text": "Hola {{1}}, has recibido una nueva orden de compra {{2}} por un total de {{3}}.",
"parameter_count": 3,
"is_active": True,
"sent_count": 125,
"created_at": "2024-01-15T10:30:00",
"updated_at": "2024-01-15T10:30:00"
}
}
# ============================================================
# Message Query Schemas
# ============================================================
class WhatsAppMessageResponse(BaseModel):
"""WhatsApp message response"""
id: str
tenant_id: str
notification_id: Optional[str]
whatsapp_message_id: Optional[str]
recipient_phone: str
recipient_name: Optional[str]
message_type: str
status: str
template_name: Optional[str]
template_language: Optional[str]
message_body: Optional[str]
sent_at: Optional[datetime]
delivered_at: Optional[datetime]
read_at: Optional[datetime]
failed_at: Optional[datetime]
error_message: Optional[str]
metadata: Optional[Dict[str, Any]]
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class WhatsAppDeliveryStats(BaseModel):
"""WhatsApp delivery statistics"""
total_messages: int
sent: int
delivered: int
read: int
failed: int
pending: int
unique_recipients: int
total_conversations: int
delivery_rate: float
period: Dict[str, str]
class Config:
json_schema_extra = {
"example": {
"total_messages": 1500,
"sent": 1480,
"delivered": 1450,
"read": 1200,
"failed": 20,
"pending": 0,
"unique_recipients": 350,
"total_conversations": 400,
"delivery_rate": 96.67,
"period": {
"start": "2024-01-01T00:00:00",
"end": "2024-01-31T23:59:59"
}
}
}

555
services/notification/app/services/whatsapp_business_service.py Normal file
View File

@@ -0,0 +1,555 @@
# ================================================================
# services/notification/app/services/whatsapp_business_service.py
# ================================================================
"""
WhatsApp Business Cloud API Service
Direct integration with Meta's WhatsApp Business Cloud API
Supports template messages for proactive notifications
"""
import structlog
import httpx
from typing import Optional, Dict, Any, List
import asyncio
from datetime import datetime
import uuid
from app.core.config import settings
from app.schemas.whatsapp import (
SendWhatsAppMessageRequest,
SendWhatsAppMessageResponse,
TemplateComponent,
WhatsAppMessageStatus,
WhatsAppMessageType
)
from app.repositories.whatsapp_message_repository import (
WhatsAppMessageRepository,
WhatsAppTemplateRepository
)
from app.models.whatsapp_messages import WhatsAppMessage
from shared.monitoring.metrics import MetricsCollector
from sqlalchemy.ext.asyncio import AsyncSession
logger = structlog.get_logger()
metrics = MetricsCollector("notification-service")
class WhatsAppBusinessService:
"""
WhatsApp Business Cloud API Service
Direct integration with Meta/Facebook WhatsApp Business Cloud API
"""
def __init__(self, session: Optional[AsyncSession] = None, tenant_client=None):
# Global configuration (fallback)
self.global_access_token = settings.WHATSAPP_ACCESS_TOKEN
self.global_phone_number_id = settings.WHATSAPP_PHONE_NUMBER_ID
self.global_business_account_id = settings.WHATSAPP_BUSINESS_ACCOUNT_ID
self.api_version = settings.WHATSAPP_API_VERSION or "v18.0"
self.base_url = f"https://graph.facebook.com/{self.api_version}"
self.enabled = settings.ENABLE_WHATSAPP_NOTIFICATIONS
# Tenant client for fetching per-tenant settings
self.tenant_client = tenant_client
# Repository dependencies (will be injected)
self.session = session
self.message_repo = WhatsAppMessageRepository(session) if session else None
self.template_repo = WhatsAppTemplateRepository(session) if session else None
async def _get_whatsapp_credentials(self, tenant_id: str) -> Dict[str, str]:
"""
Get WhatsApp credentials for a tenant
Tries tenant-specific settings first, falls back to global config
Args:
tenant_id: Tenant ID
Returns:
Dictionary with access_token, phone_number_id, business_account_id
"""
# Try to fetch tenant-specific settings
if self.tenant_client:
try:
notification_settings = await self.tenant_client.get_notification_settings(tenant_id)
if notification_settings and notification_settings.get('whatsapp_enabled'):
tenant_access_token = notification_settings.get('whatsapp_access_token', '').strip()
tenant_phone_id = notification_settings.get('whatsapp_phone_number_id', '').strip()
tenant_business_id = notification_settings.get('whatsapp_business_account_id', '').strip()
# Use tenant credentials if all are configured
if tenant_access_token and tenant_phone_id:
logger.info(
"Using tenant-specific WhatsApp credentials",
tenant_id=tenant_id
)
return {
'access_token': tenant_access_token,
'phone_number_id': tenant_phone_id,
'business_account_id': tenant_business_id
}
else:
logger.info(
"Tenant WhatsApp enabled but credentials incomplete, falling back to global",
tenant_id=tenant_id
)
except Exception as e:
logger.warning(
"Failed to fetch tenant notification settings, using global config",
error=str(e),
tenant_id=tenant_id
)
# Fallback to global configuration
logger.info(
"Using global WhatsApp credentials",
tenant_id=tenant_id
)
return {
'access_token': self.global_access_token,
'phone_number_id': self.global_phone_number_id,
'business_account_id': self.global_business_account_id
}
async def send_message(
self,
request: SendWhatsAppMessageRequest
) -> SendWhatsAppMessageResponse:
"""
Send WhatsApp message via Cloud API
Args:
request: Message request with all details
Returns:
SendWhatsAppMessageResponse with status
"""
try:
if not self.enabled:
logger.info("WhatsApp notifications disabled")
return SendWhatsAppMessageResponse(
success=False,
message_id=str(uuid.uuid4()),
status=WhatsAppMessageStatus.FAILED,
error_message="WhatsApp notifications are disabled"
)
# Get tenant-specific or global credentials
credentials = await self._get_whatsapp_credentials(request.tenant_id)
access_token = credentials['access_token']
phone_number_id = credentials['phone_number_id']
# Validate configuration
if not access_token or not phone_number_id:
logger.error("WhatsApp Cloud API not configured properly", tenant_id=request.tenant_id)
return SendWhatsAppMessageResponse(
success=False,
message_id=str(uuid.uuid4()),
status=WhatsAppMessageStatus.FAILED,
error_message="WhatsApp Cloud API credentials not configured"
)
# Create message record in database
message_data = {
"tenant_id": request.tenant_id,
"notification_id": request.notification_id,
"recipient_phone": request.recipient_phone,
"recipient_name": request.recipient_name,
"message_type": request.message_type,
"status": WhatsAppMessageStatus.PENDING,
"metadata": request.metadata,
"created_at": datetime.utcnow(),
"updated_at": datetime.utcnow()
}
# Add template details if template message
if request.message_type == WhatsAppMessageType.TEMPLATE and request.template:
message_data["template_name"] = request.template.template_name
message_data["template_language"] = request.template.language
message_data["template_parameters"] = [
comp.model_dump() for comp in request.template.components
]
# Add text details if text message
if request.message_type == WhatsAppMessageType.TEXT and request.text:
message_data["message_body"] = request.text
# Save to database
if self.message_repo:
db_message = await self.message_repo.create_message(message_data)
message_id = str(db_message.id)
else:
message_id = str(uuid.uuid4())
# Send message via Cloud API
if request.message_type == WhatsAppMessageType.TEMPLATE:
result = await self._send_template_message(
recipient_phone=request.recipient_phone,
template=request.template,
message_id=message_id,
access_token=access_token,
phone_number_id=phone_number_id
)
elif request.message_type == WhatsAppMessageType.TEXT:
result = await self._send_text_message(
recipient_phone=request.recipient_phone,
text=request.text,
message_id=message_id,
access_token=access_token,
phone_number_id=phone_number_id
)
else:
logger.error(f"Unsupported message type: {request.message_type}")
result = {
"success": False,
"error_message": f"Unsupported message type: {request.message_type}"
}
# Update database with result
if self.message_repo and result.get("success"):
await self.message_repo.update_message_status(
message_id=message_id,
status=WhatsAppMessageStatus.SENT,
whatsapp_message_id=result.get("whatsapp_message_id"),
provider_response=result.get("provider_response")
)
elif self.message_repo:
await self.message_repo.update_message_status(
message_id=message_id,
status=WhatsAppMessageStatus.FAILED,
error_message=result.get("error_message"),
provider_response=result.get("provider_response")
)
# Record metrics
status = "success" if result.get("success") else "failed"
metrics.increment_counter("whatsapp_sent_total", labels={"status": status})
return SendWhatsAppMessageResponse(
success=result.get("success", False),
message_id=message_id,
whatsapp_message_id=result.get("whatsapp_message_id"),
status=WhatsAppMessageStatus.SENT if result.get("success") else WhatsAppMessageStatus.FAILED,
error_message=result.get("error_message")
)
except Exception as e:
logger.error("Failed to send WhatsApp message", error=str(e))
metrics.increment_counter("whatsapp_sent_total", labels={"status": "failed"})
return SendWhatsAppMessageResponse(
success=False,
message_id=str(uuid.uuid4()),
status=WhatsAppMessageStatus.FAILED,
error_message=str(e)
)
async def _send_template_message(
self,
recipient_phone: str,
template: Any,
message_id: str,
access_token: str,
phone_number_id: str
) -> Dict[str, Any]:
"""Send template message via WhatsApp Cloud API"""
try:
# Build template payload
payload = {
"messaging_product": "whatsapp",
"to": recipient_phone,
"type": "template",
"template": {
"name": template.template_name,
"language": {
"code": template.language
},
"components": [
{
"type": comp.type,
"parameters": [
param.model_dump() for param in (comp.parameters or [])
]
}
for comp in template.components
]
}
}
# Send request to WhatsApp Cloud API
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/{phone_number_id}/messages",
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
},
json=payload
)
response_data = response.json()
if response.status_code == 200:
whatsapp_message_id = response_data.get("messages", [{}])[0].get("id")
logger.info(
"WhatsApp template message sent successfully",
message_id=message_id,
whatsapp_message_id=whatsapp_message_id,
template=template.template_name,
recipient=recipient_phone
)
# Increment template usage count
if self.template_repo:
template_obj = await self.template_repo.get_by_template_name(
template.template_name,
template.language
)
if template_obj:
await self.template_repo.increment_usage(str(template_obj.id))
return {
"success": True,
"whatsapp_message_id": whatsapp_message_id,
"provider_response": response_data
}
else:
error_message = response_data.get("error", {}).get("message", "Unknown error")
error_code = response_data.get("error", {}).get("code")
logger.error(
"WhatsApp Cloud API error",
status_code=response.status_code,
error_code=error_code,
error_message=error_message,
template=template.template_name
)
return {
"success": False,
"error_message": f"{error_code}: {error_message}",
"provider_response": response_data
}
except Exception as e:
logger.error(
"Failed to send template message",
template=template.template_name,
error=str(e)
)
return {
"success": False,
"error_message": str(e)
}
async def _send_text_message(
self,
recipient_phone: str,
text: str,
message_id: str,
access_token: str,
phone_number_id: str
) -> Dict[str, Any]:
"""Send text message via WhatsApp Cloud API"""
try:
payload = {
"messaging_product": "whatsapp",
"to": recipient_phone,
"type": "text",
"text": {
"body": text
}
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/{phone_number_id}/messages",
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
},
json=payload
)
response_data = response.json()
if response.status_code == 200:
whatsapp_message_id = response_data.get("messages", [{}])[0].get("id")
logger.info(
"WhatsApp text message sent successfully",
message_id=message_id,
whatsapp_message_id=whatsapp_message_id,
recipient=recipient_phone
)
return {
"success": True,
"whatsapp_message_id": whatsapp_message_id,
"provider_response": response_data
}
else:
error_message = response_data.get("error", {}).get("message", "Unknown error")
error_code = response_data.get("error", {}).get("code")
logger.error(
"WhatsApp Cloud API error",
status_code=response.status_code,
error_code=error_code,
error_message=error_message
)
return {
"success": False,
"error_message": f"{error_code}: {error_message}",
"provider_response": response_data
}
except Exception as e:
logger.error("Failed to send text message", error=str(e))
return {
"success": False,
"error_message": str(e)
}
async def send_bulk_messages(
self,
requests: List[SendWhatsAppMessageRequest],
batch_size: int = 20
) -> Dict[str, Any]:
"""
Send bulk WhatsApp messages with rate limiting
Args:
requests: List of message requests
batch_size: Number of messages to send per batch
Returns:
Dict containing success/failure counts
"""
results = {
"total": len(requests),
"sent": 0,
"failed": 0,
"errors": []
}
try:
# Process in batches to respect WhatsApp rate limits
for i in range(0, len(requests), batch_size):
batch = requests[i:i + batch_size]
# Send messages concurrently within batch
tasks = [self.send_message(req) for req in batch]
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
for req, result in zip(batch, batch_results):
if isinstance(result, Exception):
results["failed"] += 1
results["errors"].append({
"phone": req.recipient_phone,
"error": str(result)
})
elif result.success:
results["sent"] += 1
else:
results["failed"] += 1
results["errors"].append({
"phone": req.recipient_phone,
"error": result.error_message
})
# Rate limiting delay between batches
if i + batch_size < len(requests):
await asyncio.sleep(2.0)
logger.info(
"Bulk WhatsApp completed",
total=results["total"],
sent=results["sent"],
failed=results["failed"]
)
return results
except Exception as e:
logger.error("Bulk WhatsApp failed", error=str(e))
results["errors"].append({"error": str(e)})
return results
async def health_check(self) -> bool:
"""
Check if WhatsApp Cloud API is healthy
Returns:
bool: True if service is healthy
"""
try:
if not self.enabled:
return True # Service is "healthy" if disabled
if not self.global_access_token or not self.global_phone_number_id:
logger.warning("WhatsApp Cloud API not configured")
return False
# Test API connectivity
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.base_url}/{self.global_phone_number_id}",
headers={
"Authorization": f"Bearer {self.global_access_token}"
},
params={
"fields": "verified_name,code_verification_status"
}
)
if response.status_code == 200:
logger.info("WhatsApp Cloud API health check passed")
return True
else:
logger.error(
"WhatsApp Cloud API health check failed",
status_code=response.status_code
)
return False
except Exception as e:
logger.error("WhatsApp Cloud API health check failed", error=str(e))
return False
def _format_phone_number(self, phone: str) -> Optional[str]:
"""
Format phone number for WhatsApp (E.164 format)
Args:
phone: Input phone number
Returns:
Formatted phone number or None if invalid
"""
if not phone:
return None
# If already in E.164 format, return as is
if phone.startswith('+'):
return phone
# Remove spaces, dashes, and other non-digit characters
clean_phone = "".join(filter(str.isdigit, phone))
# Handle Spanish phone numbers
if clean_phone.startswith("34"):
return f"+{clean_phone}"
elif clean_phone.startswith(("6", "7", "9")) and len(clean_phone) == 9:
return f"+34{clean_phone}"
else:
# Try to add + if it looks like a complete international number
if len(clean_phone) > 10:
return f"+{clean_phone}"
logger.warning("Unrecognized phone format", phone=phone)
return None

445
services/notification/app/services/whatsapp_service.py
View File

@@ -3,60 +3,59 @@
# ================================================================
"""
WhatsApp service for sending notifications
Integrates with WhatsApp Business API via Twilio
Integrates with WhatsApp Business Cloud API (Meta/Facebook)
This is a backward-compatible wrapper around the new WhatsAppBusinessService
"""
import structlog
import httpx
from typing import Optional, Dict, Any, List
import asyncio
from urllib.parse import quote
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import settings
from app.services.whatsapp_business_service import WhatsAppBusinessService
from app.schemas.whatsapp import (
SendWhatsAppMessageRequest,
TemplateMessageRequest,
TemplateComponent,
TemplateParameter,
WhatsAppMessageType
)
from shared.monitoring.metrics import MetricsCollector
logger = structlog.get_logger()
metrics = MetricsCollector("notification-service")
class WhatsAppService:
"""
WhatsApp service for sending notifications via Twilio WhatsApp API
Supports text messages and template messages
WhatsApp service for sending notifications via WhatsApp Business Cloud API
Backward-compatible wrapper for existing code
"""
def __init__(self):
self.api_key = settings.WHATSAPP_API_KEY
self.base_url = settings.WHATSAPP_BASE_URL
self.from_number = settings.WHATSAPP_FROM_NUMBER
def __init__(self, session: Optional[AsyncSession] = None, tenant_client=None):
self.enabled = settings.ENABLE_WHATSAPP_NOTIFICATIONS
def _parse_api_credentials(self):
"""Parse API key into username and password for Twilio basic auth"""
if not self.api_key or ":" not in self.api_key:
raise ValueError("WhatsApp API key must be in format 'username:password'")
api_parts = self.api_key.split(":", 1)
if len(api_parts) != 2:
raise ValueError("Invalid WhatsApp API key format")
return api_parts[0], api_parts[1]
self.business_service = WhatsAppBusinessService(session, tenant_client)
async def send_message(
self,
to_phone: str,
message: str,
template_name: Optional[str] = None,
template_params: Optional[List[str]] = None
template_params: Optional[List[str]] = None,
tenant_id: Optional[str] = None
) -> bool:
"""
Send WhatsApp message
Send WhatsApp message (backward-compatible wrapper)
Args:
to_phone: Recipient phone number (with country code)
message: Message text
template_name: WhatsApp template name (optional)
template_params: Template parameters (optional)
tenant_id: Tenant ID (optional, defaults to system tenant)
Returns:
bool: True if message was sent successfully
"""
@@ -64,47 +63,71 @@ class WhatsAppService:
if not self.enabled:
logger.info("WhatsApp notifications disabled")
return True # Return success to avoid blocking workflow
if not self.api_key:
logger.error("WhatsApp API key not configured")
return False
# Validate phone number
# Format phone number
phone = self._format_phone_number(to_phone)
if not phone:
logger.error("Invalid phone number", phone=to_phone)
return False
# Send template message if template specified
# Use default tenant if not provided
if not tenant_id:
tenant_id = "00000000-0000-0000-0000-000000000000" # System tenant
# Build request
if template_name:
success = await self._send_template_message(
phone, template_name, template_params or []
# Template message
components = []
if template_params:
# Build body component with parameters
parameters = [
TemplateParameter(type="text", text=param)
for param in template_params
]
components.append(
TemplateComponent(type="body", parameters=parameters)
)
template_request = TemplateMessageRequest(
template_name=template_name,
language="es",
components=components
)
request = SendWhatsAppMessageRequest(
tenant_id=tenant_id,
recipient_phone=phone,
message_type=WhatsAppMessageType.TEMPLATE,
template=template_request
)
else:
# Send regular text message
success = await self._send_text_message(phone, message)
if success:
logger.info("WhatsApp message sent successfully",
to=phone,
template=template_name)
# Record success metrics
metrics.increment_counter("whatsapp_sent_total", labels={"status": "success"})
else:
# Record failure metrics
metrics.increment_counter("whatsapp_sent_total", labels={"status": "failed"})
return success
# Text message
request = SendWhatsAppMessageRequest(
tenant_id=tenant_id,
recipient_phone=phone,
message_type=WhatsAppMessageType.TEXT,
text=message
)
# Send via business service
response = await self.business_service.send_message(request)
if response.success:
logger.info(
"WhatsApp message sent successfully",
to=phone,
template=template_name
)
return response.success
except Exception as e:
logger.error("Failed to send WhatsApp message",
to=to_phone,
error=str(e))
# Record failure metrics
logger.error(
"Failed to send WhatsApp message",
to=to_phone,
error=str(e)
)
metrics.increment_counter("whatsapp_sent_total", labels={"status": "failed"})
return False
async def send_bulk_messages(
@@ -112,17 +135,21 @@ class WhatsAppService:
recipients: List[str],
message: str,
template_name: Optional[str] = None,
batch_size: int = 20
template_params: Optional[List[str]] = None,
batch_size: int = 20,
tenant_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Send bulk WhatsApp messages with rate limiting
Send bulk WhatsApp messages with rate limiting (backward-compatible wrapper)
Args:
recipients: List of recipient phone numbers
message: Message text
template_name: WhatsApp template name (optional)
template_params: Template parameters (optional)
batch_size: Number of messages to send per batch
tenant_id: Tenant ID (optional)
Returns:
Dict containing success/failure counts
"""
@@ -132,45 +159,76 @@ class WhatsAppService:
"failed": 0,
"errors": []
}
try:
# Process in batches to respect WhatsApp rate limits
for i in range(0, len(recipients), batch_size):
batch = recipients[i:i + batch_size]
# Send messages concurrently within batch
tasks = [
self.send_message(
to_phone=phone,
message=message,
template_name=template_name
# Use default tenant if not provided
if not tenant_id:
tenant_id = "00000000-0000-0000-0000-000000000000"
# Build requests for all recipients
requests = []
for phone in recipients:
formatted_phone = self._format_phone_number(phone)
if not formatted_phone:
results["failed"] += 1
results["errors"].append({"phone": phone, "error": "Invalid phone format"})
continue
if template_name:
# Template message
components = []
if template_params:
parameters = [
TemplateParameter(type="text", text=param)
for param in template_params
]
components.append(
TemplateComponent(type="body", parameters=parameters)
)
template_request = TemplateMessageRequest(
template_name=template_name,
language="es",
components=components
)
for phone in batch
]
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
for phone, result in zip(batch, batch_results):
if isinstance(result, Exception):
results["failed"] += 1
results["errors"].append({"phone": phone, "error": str(result)})
elif result:
results["sent"] += 1
else:
results["failed"] += 1
results["errors"].append({"phone": phone, "error": "Unknown error"})
# Rate limiting delay between batches (WhatsApp has strict limits)
if i + batch_size < len(recipients):
await asyncio.sleep(2.0) # 2 second delay between batches
logger.info("Bulk WhatsApp completed",
total=results["total"],
sent=results["sent"],
failed=results["failed"])
request = SendWhatsAppMessageRequest(
tenant_id=tenant_id,
recipient_phone=formatted_phone,
message_type=WhatsAppMessageType.TEMPLATE,
template=template_request
)
else:
# Text message
request = SendWhatsAppMessageRequest(
tenant_id=tenant_id,
recipient_phone=formatted_phone,
message_type=WhatsAppMessageType.TEXT,
text=message
)
requests.append(request)
# Send via business service
bulk_result = await self.business_service.send_bulk_messages(
requests,
batch_size=batch_size
)
# Update results
results["sent"] = bulk_result.get("sent", 0)
results["failed"] += bulk_result.get("failed", 0)
results["errors"].extend(bulk_result.get("errors", []))
logger.info(
"Bulk WhatsApp completed",
total=results["total"],
sent=results["sent"],
failed=results["failed"]
)
return results
except Exception as e:
logger.error("Bulk WhatsApp failed", error=str(e))
results["errors"].append({"error": str(e)})
@@ -179,203 +237,20 @@ class WhatsAppService:
async def health_check(self) -> bool:
"""
Check if WhatsApp service is healthy
Returns:
bool: True if service is healthy
"""
try:
if not self.enabled:
return True # Service is "healthy" if disabled
if not self.api_key:
logger.warning("WhatsApp API key not configured")
return False
# Test API connectivity with a simple request
# Parse API key (expected format: username:password for Twilio basic auth)
if ":" not in self.api_key:
logger.error("WhatsApp API key must be in format 'username:password'")
return False
api_parts = self.api_key.split(":", 1) # Split on first : only
if len(api_parts) != 2:
logger.error("Invalid WhatsApp API key format")
return False
username, password = api_parts
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.base_url}/v1/Account", # Twilio account info endpoint
auth=(username, password)
)
if response.status_code == 200:
logger.info("WhatsApp service health check passed")
return True
else:
logger.error("WhatsApp service health check failed",
status_code=response.status_code)
return False
except Exception as e:
logger.error("WhatsApp service health check failed", error=str(e))
return False
# ================================================================
# PRIVATE HELPER METHODS
# ================================================================
async def _send_text_message(self, to_phone: str, message: str) -> bool:
"""Send regular text message via Twilio"""
try:
# Parse API credentials
try:
username, password = self._parse_api_credentials()
except ValueError as e:
logger.error(f"WhatsApp API key configuration error: {e}")
return False
# Prepare request data
data = {
"From": f"whatsapp:{self.from_number}",
"To": f"whatsapp:{to_phone}",
"Body": message
}
# Send via Twilio API
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/2010-04-01/Accounts/{username}/Messages.json",
data=data,
auth=(username, password)
)
if response.status_code == 201:
response_data = response.json()
logger.debug("WhatsApp message sent",
message_sid=response_data.get("sid"),
status=response_data.get("status"))
return True
else:
logger.error("WhatsApp API error",
status_code=response.status_code,
response=response.text)
return False
except Exception as e:
logger.error("Failed to send WhatsApp text message", error=str(e))
return False
async def _send_template_message(
self,
to_phone: str,
template_name: str,
parameters: List[str]
) -> bool:
"""Send WhatsApp template message via Twilio"""
try:
# Parse API credentials
try:
username, password = self._parse_api_credentials()
except ValueError as e:
logger.error(f"WhatsApp API key configuration error: {e}")
return False
# Prepare template data
content_variables = {str(i+1): param for i, param in enumerate(parameters)}
data = {
"From": f"whatsapp:{self.from_number}",
"To": f"whatsapp:{to_phone}",
"ContentSid": template_name, # Template SID in Twilio
"ContentVariables": str(content_variables) if content_variables else "{}"
}
# Send via Twilio API
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/2010-04-01/Accounts/{username}/Messages.json",
data=data,
auth=(username, password)
)
if response.status_code == 201:
response_data = response.json()
logger.debug("WhatsApp template message sent",
message_sid=response_data.get("sid"),
template=template_name)
return True
else:
logger.error("WhatsApp template API error",
status_code=response.status_code,
response=response.text,
template=template_name)
return False
except Exception as e:
logger.error("Failed to send WhatsApp template message",
template=template_name,
error=str(e))
return False
return await self.business_service.health_check()
def _format_phone_number(self, phone: str) -> Optional[str]:
"""
Format phone number for WhatsApp (Spanish format)
Format phone number for WhatsApp (E.164 format)
Args:
phone: Input phone number
Returns:
Formatted phone number or None if invalid
"""
if not phone:
return None
# Remove spaces, dashes, and other non-digit characters
clean_phone = "".join(filter(str.isdigit, phone.replace("+", "")))
# Handle Spanish phone numbers
if clean_phone.startswith("34"):
# Already has country code
return f"+{clean_phone}"
elif clean_phone.startswith(("6", "7", "9")) and len(clean_phone) == 9:
# Spanish mobile/landline without country code
return f"+34{clean_phone}"
elif len(clean_phone) == 9 and clean_phone[0] in "679":
# Likely Spanish mobile
return f"+34{clean_phone}"
else:
logger.warning("Unrecognized phone format", phone=phone)
return None
async def _get_message_status(self, message_sid: str) -> Optional[str]:
"""Get message delivery status from Twilio"""
try:
# Parse API credentials
try:
username, password = self._parse_api_credentials()
except ValueError as e:
logger.error(f"WhatsApp API key configuration error: {e}")
return None
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{self.base_url}/2010-04-01/Accounts/{username}/Messages/{message_sid}.json",
auth=(username, password)
)
if response.status_code == 200:
data = response.json()
return data.get("status")
else:
logger.error("Failed to get message status",
message_sid=message_sid,
status_code=response.status_code)
return None
except Exception as e:
logger.error("Failed to check message status",
message_sid=message_sid,
error=str(e))
return None
return self.business_service._format_phone_number(phone)