""" caller_memory.py — Persistent Caller Memory for Maya (Neon/Postgres Version) """ import asyncio from dataclasses import dataclass, field from typing import Optional, List from datetime import datetime import pytz from src.database import _is_valid_uuid IST = pytz.timezone("Asia/Kolkata") @dataclass class CallerProfile: id: Optional[str] = None # UUID phone_number: str = "" name: Optional[str] = None tenant_id: str = "" call_count: int = 0 first_call_at: Optional[datetime] = None last_call_at: Optional[datetime] = None last_service: Optional[str] = None is_vip: bool = False notes: Optional[str] = None recent_calls: list = field(default_factory=list) @property def is_return_caller(self) -> bool: return self.call_count > 0 def get_context_for_llm(self) -> str: # If we have a name, provide it regardless of call count (could be imported) # If we have no name AND no history, return empty if not self.name and not self.is_return_caller: return "" lines = [] if self.name: lines.append(f"CALLER NAME: {self.name}") lines.append(f" -> Address them as '{self.name}' naturally.") if self.is_return_caller: lines.append(f"CALL HISTORY: This is their call #{self.call_count + 1}") if self.last_call_at: # Simple "days ago" logic delta = datetime.now(pytz.utc) - self.last_call_at.astimezone(pytz.utc) days = delta.days lines.append(f"LAST CALLED: {days} days ago" if days > 0 else "LAST CALLED: Today") if self.last_service: lines.append(f"LAST SERVICE: {self.last_service}") if self.is_vip: lines.append("VIP CALLER: Yes. Give them priority treatment.") if self.recent_calls: last = self.recent_calls[0] if last.get("summary"): lines.append(f"LAST CALL SUMMARY: {last['summary']}") return "\n\n[CALLER MEMORY]\n" + "\n".join(lines) + "\n[END CALLER MEMORY]" def get_personalized_greeting(self, language: str = "gujarati") -> Optional[str]: if not self.name: return None # Standardizing on Devanagari for Hindi for better TTS if language == "hindi": if self.is_return_caller: return f"नमस्ते {self.name} जी! स्माइलकेयर डेंटल क्लिनिक में आपका फिर से स्वागत है। आप कैसे हैं? मैं आपकी क्या मदद कर सकती हूँ?" else: return f"नमस्ते {self.name} जी! स्माइलकेयर डेंटल क्लिनिक में आपका स्वागत है। मैं माया हूँ, आपकी क्या मदद कर सकती हूँ?" if language == "gujarati": if self.is_return_caller: return f"નમસ્તે {self.name} ભાઈ/બહેન! સ્માઇલકેર ડેન્ટલ ક્લિનિકમાં તમારું ફરી સ્વાગત છે. તમે કેમ છો? હું તમારી શું મદદ કરી શકું?" else: return f"નમસ્તે {self.name} ભાઈ/બહેન! સ્માઇલકેર ડેન્ટલ ક્લિનિકમાં તમારું સ્વાગત છે. હું માયા છું, હું તમારી શું મદદ કરી શકું?" # English Fallback if self.is_return_caller: return f"Hello {self.name}! Welcome back to SmileCare Dental. How are you doing today? How can I help you?" else: return f"Hello {self.name}! Welcome to SmileCare Dental. I'm Maya, how can I assist you today?" class CallerMemoryService: def __init__(self, db_pool): self.pool = db_pool async def lookup_caller(self, phone_number: str, tenant_id: str) -> CallerProfile: # Return blank profile if tenant_id is not a valid UUID (demo/test tenants) if not _is_valid_uuid(tenant_id): if tenant_id == "demo-tenant-123" and phone_number == "client:demo-caller": # Mock return caller for testing Phase E4 return CallerProfile( phone_number=phone_number, tenant_id=tenant_id, name="Rudra", call_count=5, is_vip=True, last_service="Dental Cleaning", last_call_at=datetime.now(IST) ) return CallerProfile(phone_number=phone_number, tenant_id=tenant_id) async with self.pool.acquire() as conn: # 1. Fetch Profile row = await conn.fetchrow( "SELECT * FROM caller_profiles WHERE tenant_id = $1 AND phone_number = $2", tenant_id, phone_number ) if not row: return CallerProfile(phone_number=phone_number, tenant_id=tenant_id) profile = CallerProfile( id=str(row["id"]), phone_number=row["phone_number"], tenant_id=str(row["tenant_id"]), name=row["name"], call_count=row["call_count"], first_call_at=row["first_call_at"], last_call_at=row["last_call_at"], last_service=row["last_service"], is_vip=row["is_vip"], notes=row["notes"] ) # 2. Fetch Recent History history_rows = await conn.fetch( "SELECT called_at, summary FROM caller_history WHERE caller_id = $1 ORDER BY called_at DESC LIMIT 3", row["id"] ) profile.recent_calls = [dict(r) for r in history_rows] return profile async def update_caller_after_call(self, profile: CallerProfile, session_data: dict): # Skip if tenant_id is not a valid UUID (demo/test tenants) if not _is_valid_uuid(profile.tenant_id): print(f"[Memory] Skipping DB update for non-UUID tenant: {profile.tenant_id}") return async with self.pool.acquire() as conn: # 1. Upsert Profile if profile.id: await conn.execute( """ UPDATE caller_profiles SET call_count = call_count + 1, last_call_at = NOW(), name = COALESCE($1, name), last_service = COALESCE($2, last_service) WHERE id = $3 """, session_data.get("name"), session_data.get("service"), profile.id ) caller_id = profile.id else: caller_id = await conn.fetchval( """ INSERT INTO caller_profiles (tenant_id, phone_number, name, call_count, last_service) VALUES ($1, $2, $3, 1, $4) RETURNING id """, profile.tenant_id, profile.phone_number, session_data.get("name"), session_data.get("service") ) # 2. Add History await conn.execute( """ INSERT INTO caller_history (caller_id, tenant_id, call_sid, summary, emotion, appointment_booked) VALUES ($1, $2, $3, $4, $5, $6) """, caller_id, profile.tenant_id, session_data.get("call_sid"), session_data.get("summary"), session_data.get("emotion"), session_data.get("appointment_booked", False) ) print(f"[Memory] Updated profile for {profile.phone_number}")