Spaces:
Sleeping
Sleeping
File size: 4,249 Bytes
a3c7b61 81c40fc 84d1290 a3c7b61 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 |
import re
from google.cloud import firestore
from datetime import datetime, timedelta, timezone
from backend.config import gpt4o
db = firestore.Client()
from backend.credentials import setup_google_credentials
setup_google_credentials()
COMMON_EMOTIONS = [
"grateful", "hope", "content", "connected", "drained",
"envy", "disappointed", "relief", "happy", "sad", "angry",
"anxious", "excited", "calm", "lonely", "overwhelmed"
]
def get_recent_mood_entries(user_id: str, days: int = 60):
now = datetime.now(timezone.utc)
min_date = now - timedelta(days=days)
entries_ref = db.collection("mood_entries").document("entries").collection(user_id)
docs = entries_ref.stream()
recent_entries = []
for doc in docs:
data = doc.to_dict()
end_date_val = data.get("endDate")
if end_date_val:
try:
if isinstance(end_date_val, datetime):
end_date = end_date_val
else:
end_date = datetime.fromisoformat(str(end_date_val))
if end_date.tzinfo:
end_date_utc = end_date.astimezone(timezone.utc)
else:
end_date_utc = end_date.replace(tzinfo=timezone.utc)
if end_date_utc >= min_date:
recent_entries.append(data)
except Exception as e:
continue
return recent_entries
def _find_emotions(text):
emotions_found = []
for e in COMMON_EMOTIONS:
if re.search(r'\b' + re.escape(e) + r'\b', text, re.IGNORECASE):
emotions_found.append(e)
return list(set(emotions_found))
def _find_mood(text):
moods = ["good", "bad", "neutral", "happy", "sad", "ok", "great", "awful", "fine"]
for mood in moods:
if re.search(r'\b' + re.escape(mood) + r'\b', text, re.IGNORECASE):
return mood
return None
async def extract_mood_details(user_message: str, conversation_history: list = None) -> dict:
details = {
"emotions": [],
"mood": None,
"note": None,
"endDate": None,
"missing_fields": []
}
text = user_message.strip()
details["emotions"] = _find_emotions(text)
details["mood"] = _find_mood(text)
details["note"] = text
# Set endDate to now unless extracted
details["endDate"] = datetime.now(timezone.utc).isoformat()
# Fallback to LLM if missing
if not details["emotions"] or not details["mood"]:
llm_resp = await gpt4o.ainvoke([
{
"role": "system",
"content": (
"Extract the following from the user's message:\n"
"1. A list of specific emotions (words only, as a JSON list)\n"
"2. The overall mood (one word, like 'good', 'bad', or 'neutral')\n"
"Reply in strict JSON:\n"
"{\"emotions\": [...], \"mood\": \"...\"}"
)
},
{
"role": "user",
"content": user_message
}
])
import json
try:
llm_json = json.loads(llm_resp.content)
if not details["emotions"] and "emotions" in llm_json:
details["emotions"] = llm_json["emotions"]
if not details["mood"] and "mood" in llm_json:
details["mood"] = llm_json["mood"]
except Exception:
pass
if not details["mood"]:
details["missing_fields"].append("mood")
if not details["emotions"]:
details["missing_fields"].append("emotions")
return details
def generate_mood_confirmation_prompt(details: dict) -> str:
missing = details["missing_fields"]
if not missing:
return None
prompts = []
if "mood" in missing:
prompts.append("How would you describe your overall mood?")
if "emotions" in missing:
prompts.append("Which emotions did you experience? (e.g., grateful, anxious, calm, etc.)")
if len(prompts) == 1:
return prompts[0]
elif len(prompts) == 2:
return f"{prompts[0]} Also, {prompts[1].lower()}"
else:
return "Could you share more about how you're feeling?"
|