PropBot / database.py
zenaight
Enhance property handling and update server configuration
0854d25
Raw
History Blame
12.5 kB
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 - prioritize recent activity, then featured, then price
query = query.order("updated_at", desc=True).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 []