"""Core message processing logic""" import re import asyncio from datetime import datetime from typing import Dict, Any, Optional from app.core.logging import get_logger from app.core.security import sanitize_input from app.core.exceptions import RateLimitExceeded, PageNotFound logger = get_logger(__name__) class MessageProcessor: """Process incoming messages and send auto-replies""" def __init__(self, laravel_client, facebook_client): self.laravel_client = laravel_client self.facebook_client = facebook_client async def process( self, page_id: str, event: Dict[str, Any], metrics, rate_limiter, deduplicator ): """Main processing logic""" # Extract event data sender_id = event.get("sender", {}).get("id") message = event.get("message", {}) message_id = message.get("mid") message_text = message.get("text") # Validate required fields if not all([message_text, sender_id, message_id]): logger.debug("Skipping event - missing required fields") return # Sanitize input message_text = sanitize_input(message_text) logger.info(f"Processing message {message_id} from {sender_id}") metrics.increment("messages_received") # Check for duplicates is_dup = deduplicator.is_duplicate(message_id) if asyncio.iscoroutine(is_dup): is_dup = await is_dup if is_dup: logger.info(f"Duplicate message ignored: {message_id}") metrics.increment("duplicates") return # Apply rate limiting is_allowed = rate_limiter.is_allowed(sender_id) if asyncio.iscoroutine(is_allowed): is_allowed = await is_allowed if not is_allowed: logger.warning(f"Rate limit exceeded for: {sender_id}") metrics.increment("rate_limited") await self._log_analytics(page_id, "rate_limited", {"sender_id": sender_id}) return # Fetch page data page_data = await self.laravel_client.get_page(page_id) if not page_data: logger.error(f"Page {page_id} not found") metrics.increment("errors") raise PageNotFound(f"Page {page_id} not found") # Store incoming message await self.laravel_client.create_message({ "page_id": page_id, "message_id": message_id, "sender_id": sender_id, "recipient_id": page_id, "message_text": message_text, "direction": "incoming", "metadata": {"timestamp": event.get("timestamp")} }) # Log analytics await self._log_analytics(page_id, "message_received", { "sender_id": sender_id, "message_length": len(message_text) }) # Find matching auto-reply auto_replies = await self.laravel_client.get_auto_replies(page_id) matched_reply = self._find_match(message_text, auto_replies) if not matched_reply: logger.info(f"No match found for: {message_text}") metrics.increment("no_matches") await self._log_analytics(page_id, "no_match", { "sender_id": sender_id, "message_text": message_text }) return metrics.increment("matches_found") logger.info(f"Matched auto-reply ID {matched_reply['id']}") # Send typing indicator before reply await self.facebook_client.send_typing_indicator( sender_id, page_data["access_token"] ) # Delay for typing indicator to be visible (1-2 seconds is more natural) await asyncio.sleep(1.5) # Send reply with quick replies if available quick_replies = matched_reply.get("quick_replies") sent_message_id = await self.facebook_client.send_message( sender_id, matched_reply["reply_message"], page_data["access_token"], quick_replies=quick_replies ) # Turn off typing await self.facebook_client.send_typing_indicator( sender_id, page_data["access_token"], typing_on=False ) if sent_message_id: metrics.increment("messages_sent") # Store outgoing message await self.laravel_client.create_message({ "page_id": page_id, "message_id": sent_message_id, "sender_id": page_id, "recipient_id": sender_id, "message_text": matched_reply["reply_message"], "direction": "outgoing", "auto_reply_id": matched_reply["id"], "metadata": {"original_message_id": message_id} }) # Log success await self._log_analytics(page_id, "auto_reply_sent", { "auto_reply_id": matched_reply["id"], "sender_id": sender_id, "trigger_type": matched_reply["trigger_type"] }) else: metrics.increment("errors") await self._log_analytics(page_id, "send_failed", { "sender_id": sender_id, "auto_reply_id": matched_reply["id"] }) def _find_match( self, message_text: str, auto_replies: list ) -> Optional[Dict[str, Any]]: """Find matching auto-reply with support for multiple triggers""" message_lower = message_text.lower().strip() # Sort by priority (higher priority first) sorted_replies = sorted( auto_replies, key=lambda x: x.get("priority", 0), reverse=True ) for auto_reply in sorted_replies: trigger_type = auto_reply.get("trigger_type", "exact") # Support both old (trigger_text) and new (trigger_items) format trigger_items = auto_reply.get("trigger_items") if trigger_items is None: # Fallback to old format trigger_text = auto_reply.get("trigger_text", "") trigger_items = [trigger_text] if trigger_text else [] # Check conditions if present conditions = auto_reply.get("conditions") if conditions and not self._check_conditions(message_text, conditions): continue # Try to match any of the trigger items try: for trigger in trigger_items: if not trigger: continue trigger_lower = trigger.lower().strip() if trigger_type == "exact" and message_lower == trigger_lower: return auto_reply elif trigger_type == "contains" and trigger_lower in message_lower: return auto_reply elif trigger_type == "starts_with" and message_lower.startswith(trigger_lower): return auto_reply elif trigger_type == "ends_with" and message_lower.endswith(trigger_lower): return auto_reply elif trigger_type == "regex": pattern = re.compile(trigger, re.IGNORECASE) if pattern.search(message_text): return auto_reply except Exception as e: logger.error(f"Error matching message with auto_reply {auto_reply.get('id')}: {e}") return None def _check_conditions( self, message_text: str, conditions: Dict[str, Any] ) -> bool: """Check if message meets conditions""" try: # Example conditions: min_length, max_length, contains_keywords, etc. if "min_length" in conditions: if len(message_text) < conditions["min_length"]: return False if "max_length" in conditions: if len(message_text) > conditions["max_length"]: return False if "contains_keywords" in conditions: keywords = conditions["contains_keywords"] message_lower = message_text.lower() if not any(kw.lower() in message_lower for kw in keywords): return False if "excludes_keywords" in conditions: keywords = conditions["excludes_keywords"] message_lower = message_text.lower() if any(kw.lower() in message_lower for kw in keywords): return False return True except Exception as e: logger.error(f"Error checking conditions: {e}") return True # Default to allowing if condition check fails async def _log_analytics( self, page_id: str, event_type: str, data: Dict[str, Any] ): """Log analytics event""" try: await self.laravel_client.create_analytics({ "page_id": page_id, "event_type": event_type, "data": { **data, "timestamp": datetime.now().isoformat() } }) except Exception as e: logger.error(f"Failed to log analytics: {e}")