from datetime import datetime, timedelta from config import supabase async def get_or_create_user(wa_id: str, name: str = None) -> dict: """Get existing user or create new one""" if not supabase: return {"wa_id": wa_id, "name": name or "Unknown"} try: # Try to get existing user response = supabase.table("users").select("*").eq("wa_id", wa_id).execute() if response.data: # User exists, update if name provided user = response.data[0] if name and name != user.get("name"): supabase.table("users").update({"name": name, "updated_at": datetime.utcnow().isoformat()}).eq("wa_id", wa_id).execute() user["name"] = name return user else: # Create new user new_user = { "wa_id": wa_id, "name": name or "Unknown", "created_at": datetime.utcnow().isoformat(), "updated_at": datetime.utcnow().isoformat() } response = supabase.table("users").insert(new_user).execute() return response.data[0] if response.data else new_user except Exception as e: print(f"Error in get_or_create_user: {e}") return {"wa_id": wa_id, "name": name or "Unknown"} async def update_user_activity(wa_id: str): """Update user's last activity timestamp""" if not supabase: return try: supabase.table("users").update({"updated_at": datetime.utcnow().isoformat()}).eq("wa_id", wa_id).execute() except Exception as e: print(f"Error updating user activity: {e}") # --- Session Management --- async def get_active_session(wa_id: str) -> dict: """Get the active session for a user (not ended and within 3 hours)""" if not supabase: return None try: # Calculate 3 hours ago three_hours_ago = datetime.utcnow() - timedelta(hours=3) response = supabase.table("chat_sessions").select("*").eq("wa_id", wa_id).eq("ended", False).gte("last_activity", three_hours_ago.isoformat()).order("last_activity", desc=True).limit(1).execute() return response.data[0] if response.data else None except Exception as e: print(f"Error getting active session: {e}") return None async def create_new_session(wa_id: str) -> dict: """Create a new chat session for a user""" if not supabase: return {"id": "local_session", "wa_id": wa_id, "started_at": datetime.utcnow().isoformat(), "last_activity": datetime.utcnow().isoformat(), "ended": False} try: new_session = { "wa_id": wa_id, "started_at": datetime.utcnow().isoformat(), "last_activity": datetime.utcnow().isoformat(), "ended": False } response = supabase.table("chat_sessions").insert(new_session).execute() return response.data[0] if response.data else new_session except Exception as e: print(f"Error creating new session: {e}") return {"id": "local_session", "wa_id": wa_id, "started_at": datetime.utcnow().isoformat(), "last_activity": datetime.utcnow().isoformat(), "ended": False} async def update_session_activity(session_id: str): """Update session's last activity timestamp""" if not supabase or session_id == "local_session": return try: supabase.table("chat_sessions").update({"last_activity": datetime.utcnow().isoformat()}).eq("id", session_id).execute() except Exception as e: print(f"Error updating session activity: {e}") async def end_session(session_id: str): """Mark a session as ended""" if not supabase or session_id == "local_session": return try: supabase.table("chat_sessions").update({"ended": True}).eq("id", session_id).execute() except Exception as e: print(f"Error ending session: {e}") # --- Message Management --- async def save_message(session_id: str, wa_id: str, wamid: str, role: str, content: str): """Save a message to the database""" if not supabase or session_id == "local_session": return try: message = { "session_id": session_id, "wa_id": wa_id, "wamid": wamid, "role": role, "content": content, "created_at": datetime.utcnow().isoformat() } supabase.table("messages").insert(message).execute() except Exception as e: print(f"Error saving message: {e}") async def get_session_messages(session_id: str, limit: int = 10) -> list: """Get recent messages for a session (for memory context)""" if not supabase or session_id == "local_session": return [] try: response = supabase.table("messages").select("*").eq("session_id", session_id).order("created_at", desc=True).limit(limit).execute() # Reverse to get chronological order messages = response.data[::-1] if response.data else [] return messages except Exception as e: print(f"Error getting session messages: {e}") return [] # --- Session Management Helper --- async def get_or_create_active_session(wa_id: str) -> dict: """Get active session or create new one if needed""" active_session = await get_active_session(wa_id) if active_session: # Update activity await update_session_activity(active_session["id"]) return active_session else: # Create new session return await create_new_session(wa_id) async def get_user(wa_id: str): """Get user by WhatsApp ID""" if not supabase: return None try: response = supabase.table("users").select("*").eq("wa_id", wa_id).execute() return response.data[0] if response.data else None except Exception as e: print(f"Error getting user: {e}") return None async def list_users(limit: int = 50, offset: int = 0): """List all users with pagination""" if not supabase: return {"users": [], "count": 0} try: response = supabase.table("users").select("*").range(offset, offset + limit - 1).order("created_at", desc=True).execute() return {"users": response.data, "count": len(response.data)} except Exception as e: print(f"Error listing users: {e}") return {"users": [], "count": 0} async def update_user_name(wa_id: str, name: str): """Update user name""" if not supabase: return None try: response = supabase.table("users").update({ "name": name, "updated_at": datetime.utcnow().isoformat() }).eq("wa_id", wa_id).execute() return response.data[0] if response.data else None except Exception as e: print(f"Error updating user: {e}") return None # --- Persona Management --- async def get_user_persona(wa_id: str) -> dict: """ Fetch the persona row for this user. Returns an empty dict if no data exists. """ if not supabase: return {} try: resp = supabase\ .table("user_personas")\ .select("*")\ .eq("wa_id", wa_id)\ .single()\ .execute() return resp.data or {} except Exception as e: print(f"Error getting user persona: {e}") return {} async def update_user_persona(wa_id: str, updates: dict): """ Apply partial updates to the persona row. Automatically sets `updated_at` to now(). """ if not supabase: return try: updates_with_ts = {**updates, "updated_at": datetime.utcnow().isoformat()} # First try to get existing persona existing = supabase.table("user_personas").select("*").eq("wa_id", wa_id).execute() if existing.data: # Update existing record supabase\ .table("user_personas")\ .update(updates_with_ts)\ .eq("wa_id", wa_id)\ .execute() else: # Create new record new_persona = { "wa_id": wa_id, "created_at": datetime.utcnow().isoformat(), **updates_with_ts } supabase.table("user_personas").insert(new_persona).execute() except Exception as e: print(f"Error updating user persona: {e}") # --- Intent Management --- async def get_or_create_user_intent(session_id: str, wa_id: str) -> dict: """ Fetch or create the intent record for this session. """ try: resp = supabase \ .table("user_intents") \ .select("*") \ .eq("session_id", session_id) \ .single() \ .execute() if resp.data: return resp.data except Exception as e: # No record exists, create one pass new_intent = { "session_id": session_id, "wa_id": wa_id, "location_preference": None, "budget": None, "size_preference_sqm": None, "must_have": None, "created_at": datetime.utcnow().isoformat(), "updated_at": datetime.utcnow().isoformat() } insert = supabase.table("user_intents").insert(new_intent).execute() return insert.data[0] if insert.data else new_intent async def update_user_intent(session_id: str, updates: dict): """ Patch the intent record with any changed fields. """ if not supabase or session_id == "local_session": return try: updates_with_ts = { **updates, "updated_at": datetime.utcnow().isoformat() } supabase \ .table("user_intents") \ .update(updates_with_ts) \ .eq("session_id", session_id) \ .execute() except Exception as e: pass # --- Property Search --- async def search_properties(filters: dict) -> list: """ Returns up to 5 active properties matching the user's filters with flexible ranges. """ if not supabase: return [] try: # Start with base query query = supabase.table("properties").select("*").eq("is_active", True) # a. Location: try city first, then location field loc = filters.get("location_preference") if loc: # Try city match first city_query = query.ilike("city", f"%{loc}%") city_resp = city_query.execute() if city_resp.data: query = city_query else: # Fallback to location field query = supabase.table("properties").select("*").eq("is_active", True).ilike("location", f"%{loc}%") # b. Budget range (target ± 50% tolerance) budget = filters.get("budget") if budget is not None: # Allow properties within 50% of the target budget min_budget = int(budget * 0.5) # 50% below target max_budget = int(budget * 1.5) # 50% above target query = query.gte("price", min_budget).lte("price", max_budget) # c. Size range (target ± 50% tolerance) size = filters.get("size_preference_sqm") if size is not None: # Allow properties within 50% of the target size min_size = int(size * 0.5) # 50% below target max_size = int(size * 1.5) # 50% above target query = query.gte("size_sqm", min_size).lte("size_sqm", max_size) # Note: Must-have features are removed from database search # They will be handled by the LLM in the chat response # e. Order & limit query = query.order("is_featured", desc=True).order("price").limit(5) # f. Execute and return resp = query.execute() print(f"DEBUG - search_properties filters: {filters}") print(f"DEBUG - properties returned: {len(resp.data) if resp.data else 0}") if resp.data: print(f"DEBUG - first property: {resp.data[0].get('title')} in {resp.data[0].get('city')}") return resp.data or [] except Exception as e: print(f"DEBUG - search_properties error: {e}") return []