""" Action Dispatcher — Executes actions generated by AI agents. ============================================================= When an agent produces actions (send_whatsapp, create_meeting, etc.), this dispatcher routes them to the correct integration service. """ import logging from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger("dealix.agents.dispatcher") class ActionDispatcher: """ Receives actions from AgentExecutor and dispatches them to the appropriate integration service (WhatsApp, Email, DB, etc.). """ def __init__(self, db: AsyncSession): self.db = db async def dispatch(self, actions: list[dict], tenant_id: str = None) -> list[dict]: """Execute a list of agent-generated actions.""" results = [] for action in actions: action_type = action.get("type", "") try: result = await self._execute_action(action_type, action, tenant_id) results.append({ "type": action_type, "status": "success", "result": result, }) logger.info(f"Action dispatched: {action_type} → success") except Exception as e: results.append({ "type": action_type, "status": "error", "error": str(e), }) logger.error(f"Action dispatch failed: {action_type} → {e}") return results async def _execute_action(self, action_type: str, action: dict, tenant_id: str) -> dict: """Route action to the correct handler.""" handlers = { "send_whatsapp": self._handle_send_whatsapp, "send_email": self._handle_send_email, "queue_message": self._handle_queue_message, "queue_ab_variant": self._handle_queue_ab_variant, "create_meeting": self._handle_create_meeting, "update_lead_score": self._handle_update_lead_score, "trigger_event": self._handle_trigger_event, "generate_payment_link": self._handle_generate_payment_link, "create_proposal": self._handle_create_proposal, "block_action": self._handle_block_action, "suspend_entity": self._handle_suspend_entity, "process_refund": self._handle_process_refund, "send_retention_offer": self._handle_send_retention_offer, } handler = handlers.get(action_type) if not handler: logger.warning(f"No handler for action type: {action_type}") return {"status": "skipped", "reason": f"Unknown action type: {action_type}"} return await handler(action, tenant_id) # ── WhatsApp ───────────────────────────────────── async def _handle_send_whatsapp(self, action: dict, tenant_id: str) -> dict: """Send a WhatsApp message.""" from app.integrations.whatsapp import send_whatsapp_message phone = action.get("phone", "") message = action.get("message", "") if not phone or not message: return {"status": "skipped", "reason": "Missing phone or message"} result = await send_whatsapp_message(phone, message) # Log to messages table try: from app.models.message import Message import uuid msg = Message( tenant_id=uuid.UUID(tenant_id) if tenant_id else None, channel="whatsapp", direction="outbound", content=message, status="sent" if result.get("status") == "success" else "failed", extra_metadata={"action": "agent_auto_send", "result": result}, ) self.db.add(msg) await self.db.flush() except Exception as e: logger.error(f"Failed to log WhatsApp message: {e}") return result # ── Email ──────────────────────────────────────── async def _handle_send_email(self, action: dict, tenant_id: str) -> dict: """Send an email.""" from app.integrations.email_sender import send_email email = action.get("email", "") message = action.get("message", "") subject = action.get("subject", "Dealix — رسالة جديدة") if not email or not message: return {"status": "skipped", "reason": "Missing email or message"} result = await send_email(email, subject, message) # Log to messages table try: from app.models.message import Message import uuid msg = Message( tenant_id=uuid.UUID(tenant_id) if tenant_id else None, channel="email", direction="outbound", content=message, status="sent" if result.get("status") == "sent" else "failed", extra_metadata={"action": "agent_auto_send", "subject": subject}, ) self.db.add(msg) await self.db.flush() except Exception as e: logger.error(f"Failed to log email message: {e}") return result # ── Message Queue ──────────────────────────────── async def _handle_queue_message(self, action: dict, tenant_id: str) -> dict: """Queue a message for scheduled sending.""" channel = action.get("channel", "whatsapp") message = action.get("message", "") optimal_time = action.get("optimal_send_time") if optimal_time: # Schedule for later — use Celery task try: from app.workers.message_tasks import send_scheduled_message send_scheduled_message.apply_async( args=[channel, message, tenant_id], countdown=self._calculate_delay(optimal_time), ) return {"status": "queued", "send_time": optimal_time} except Exception: pass # Send immediately if no schedule if channel == "whatsapp": return await self._handle_send_whatsapp(action, tenant_id) elif channel == "email": return await self._handle_send_email(action, tenant_id) return {"status": "queued", "channel": channel} async def _handle_queue_ab_variant(self, action: dict, tenant_id: str) -> dict: """Store an A/B variant for testing.""" return { "status": "stored", "variant": "B", "message_preview": action.get("message", "")[:100], } # ── Meeting ────────────────────────────────────── async def _handle_create_meeting(self, action: dict, tenant_id: str) -> dict: """Create a meeting booking.""" try: from app.models.ai_conversation import AutoBooking import uuid from datetime import datetime dt_str = action.get("datetime", "") meeting_dt = datetime.fromisoformat(dt_str) if dt_str else datetime.utcnow() booking = AutoBooking( tenant_id=uuid.UUID(tenant_id) if tenant_id else None, lead_id=uuid.UUID(action["lead_id"]) if action.get("lead_id") else None, meeting_type=action.get("type", "demo"), meeting_datetime=meeting_dt, duration_minutes=action.get("duration_minutes", 30), client_name=action.get("client_name", ""), status="scheduled", ) self.db.add(booking) await self.db.flush() return {"status": "booked", "booking_id": str(booking.id), "datetime": dt_str} except Exception as e: return {"status": "error", "detail": str(e)} # ── Lead Score ─────────────────────────────────── async def _handle_update_lead_score(self, action: dict, tenant_id: str) -> dict: """Update lead score in database.""" try: from app.models.lead import Lead from sqlalchemy import update import uuid lead_id = action.get("lead_id") if not lead_id: return {"status": "skipped", "reason": "No lead_id"} await self.db.execute( update(Lead) .where(Lead.id == uuid.UUID(lead_id)) .values( score=action.get("score", 0), status=action.get("status", "contacted"), ) ) await self.db.flush() return { "status": "updated", "lead_id": lead_id, "score": action.get("score"), "classification": action.get("classification"), } except Exception as e: return {"status": "error", "detail": str(e)} # ── Event Trigger ──────────────────────────────── async def _handle_trigger_event(self, action: dict, tenant_id: str) -> dict: """Trigger a new event in the agent system.""" event_type = action.get("event", "") lead_id = action.get("lead_id", "") try: from app.workers.agent_tasks import process_agent_event process_agent_event.delay( event_type=event_type, input_data={"lead_id": lead_id, "auto_triggered": True}, tenant_id=tenant_id, lead_id=lead_id, ) return {"status": "triggered", "event": event_type} except Exception: # Fallback: execute synchronously return {"status": "queued_fallback", "event": event_type} # ── Payment ────────────────────────────────────── async def _handle_generate_payment_link(self, action: dict, tenant_id: str) -> dict: """Generate a payment link via Stripe or manual.""" amount = action.get("amount_sar", 0) lead_id = action.get("lead_id", "") # TODO: Integrate with Stripe when configured return { "status": "generated", "amount_sar": amount, "payment_link": f"https://pay.dealix.sa/invoice/{lead_id}", "note": "Mock payment link — Stripe integration pending", } # ── Proposal ───────────────────────────────────── async def _handle_create_proposal(self, action: dict, tenant_id: str) -> dict: """Create a proposal in the database.""" try: from app.models.proposal import Proposal import uuid proposal_data = action.get("proposal_data", {}) proposal = Proposal( tenant_id=uuid.UUID(tenant_id) if tenant_id else None, lead_id=uuid.UUID(action["lead_id"]) if action.get("lead_id") else None, title=proposal_data.get("id", "Auto-Generated Proposal"), content=proposal_data, status="draft", ) self.db.add(proposal) await self.db.flush() return {"status": "created", "proposal_id": str(proposal.id)} except Exception as e: return {"status": "error", "detail": str(e)} # ── Compliance ─────────────────────────────────── async def _handle_block_action(self, action: dict, tenant_id: str) -> dict: """Block an action due to compliance failure.""" logger.warning( f"⚠️ ACTION BLOCKED by compliance: {action.get('reason')} " f"Issues: {action.get('issues')}" ) return { "status": "blocked", "reason": action.get("reason"), "issues_count": len(action.get("issues", [])), } # ── Fraud ──────────────────────────────────────── async def _handle_suspend_entity(self, action: dict, tenant_id: str) -> dict: """Suspend an entity flagged for fraud.""" entity_type = action.get("entity_type", "unknown") risk_score = action.get("risk_score", 0) logger.critical( f"🚨 FRAUD ALERT: Suspending {entity_type} — risk_score={risk_score} " f"affected={action.get('affected')}" ) # TODO: Update entity status in DB return { "status": "suspended", "entity_type": entity_type, "risk_score": risk_score, } # ── Refund ─────────────────────────────────────── async def _handle_process_refund(self, action: dict, tenant_id: str) -> dict: """Process a guarantee refund.""" amount = action.get("amount_sar", 0) customer_id = action.get("customer_id", "") logger.info(f"💰 Refund initiated: {amount} SAR for customer {customer_id}") # TODO: Integrate with Stripe refund API return { "status": "initiated", "amount_sar": amount, "customer_id": customer_id, "note": "Refund processing — manual verification required for amounts > 5000 SAR", } # ── Retention ──────────────────────────────────── async def _handle_send_retention_offer(self, action: dict, tenant_id: str) -> dict: """Send a retention offer to a churning customer.""" offer = action.get("offer", {}) customer_id = action.get("customer_id", "") logger.info( f"🎁 Retention offer for customer {customer_id}: " f"{offer.get('discount_percent', 0)}% discount + " f"{offer.get('free_months', 0)} free months" ) return { "status": "sent", "offer": offer, "customer_id": customer_id, } # ── Helpers ────────────────────────────────────── @staticmethod def _calculate_delay(optimal_time: str) -> int: """Calculate delay in seconds until optimal send time.""" from datetime import datetime, timezone try: now = datetime.now(timezone.utc) # Parse HH:MM format hour, minute = map(int, optimal_time.split(":")) target = now.replace(hour=hour, minute=minute, second=0) if target <= now: # Next day from datetime import timedelta target += timedelta(days=1) return max(0, int((target - now).total_seconds())) except Exception: return 0 # Send immediately on parse error