AI_Agent_Server / backend /rag_utils.py
Sidreds06's picture
bug fixes
81c40fc
import os
from dotenv import load_dotenv
from backend.mood_extraction import get_recent_mood_entries
import time
from backend.cache_utils import get_cached_user_data, cache_user_data
from backend.credentials import setup_google_credentials
setup_google_credentials()
# Load from .env file
load_dotenv()
service_account_path = os.getenv("GOOGLE_APPLICATION_CREDENTIALS")
from google.cloud import firestore
db = firestore.Client()
def get_user_profile(user_id: str):
doc_ref = db.collection("users").document(user_id).collection("profile").document("general")
doc = doc_ref.get()
return doc.to_dict() if doc.exists else {}
def get_user_goals(user_id: str):
goals_ref = db.collection("goals")
query_ref = goals_ref.where("user_id", "==", user_id)
results = query_ref.stream()
return [doc.to_dict() for doc in results]
def get_user_data(user_id: str):
start_time = time.time()
# Try to get from cache first
cached_data = get_cached_user_data(user_id)
if cached_data:
print(f"[TIMING] User data (cached): {(time.time() - start_time) * 1000:.2f}ms")
return cached_data
# Cache miss - fetch fresh data
print("[CACHE] User data cache miss, fetching fresh data...")
profile = get_user_profile(user_id)
goals = get_user_goals(user_id)
recent_moods = get_recent_mood_entries(user_id, days=60)
result = {
"profile": profile,
"goals": goals,
"recent_moods": recent_moods
}
# Cache the result
cache_user_data(user_id, result)
fetch_time = (time.time() - start_time) * 1000
print(f"[TIMING] User data fetch (fresh): {fetch_time:.2f}ms")
return result
def format_profile_goals_and_moods(user_data):
profile = user_data.get("profile", {})
goals = user_data.get("goals", [])
moods = user_data.get("recent_moods", [])
profile_text = (
f"User Profile:\n"
f"Name: {profile.get('name', '[unknown]')}\n"
f"Age: {profile.get('age', '[unknown]')}\n"
f"Gender: {profile.get('gender', '[unknown]')}\n"
)
goals_text = ""
if goals:
goals_text = "User Goals:\n" + "\n".join(
[f"- {g.get('goalName', '[No name]')}: {g.get('goalDescription', '[No description]')}" for g in goals]
) + "\n"
moods_text = ""
if moods:
moods_text = "Recent Mood Entries:\n" + "\n".join(
[f"{m.get('endDate', '[no date]')}: {m.get('mood', '[no mood]')} | Emotions: {', '.join(m.get('emotions', []))} | Note: {m.get('note', '')[:40]}..." for m in moods]
) + "\n"
return profile_text + goals_text + moods_text
def format_profile_and_goals(user_data):
profile = user_data.get("profile", {})
goals = user_data.get("goals", [])
profile_text = (
f"User Profile:\n"
f"Name: {profile.get('name', '[unknown]')}\n"
f"Age: {profile.get('age', '[unknown]')}\n"
f"Gender: {profile.get('gender', '[unknown]')}\n"
)
goals_text = ""
if goals:
goals_text = "User Goals:\n" + "\n".join(
[f"- {g.get('goalName', '[No name]')}: {g.get('goalDescription', '[No description]')}" for g in goals]
) + "\n"
return profile_text + goals_text