Spaces:
Sleeping
Sleeping
File size: 8,145 Bytes
d1d1019 3bd77b6 d1d1019 | 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 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 |
import os
import pymongo
from datetime import datetime
from bson.objectid import ObjectId
# Config
MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017/mental_health_db")
class Database:
def __init__(self):
self.memory_db = {
'users': [],
'journals': [],
'streaks': {}
}
self.db = None
self.local_file = "data/local_db.json"
try:
self.client = pymongo.MongoClient(MONGO_URI, serverSelectionTimeoutMS=2000)
self.client.admin.command('ping')
self.db = self.client.get_database()
self.users = self.db.users
self.journals = self.db.journals
self.streaks = self.db.streaks
print(f"✓ Connected to MongoDB at {MONGO_URI}")
except Exception as e:
print(f"✗ Failed to connect to MongoDB ({e}). Using Persistent JSON Fallback.")
self.db = None
self._load_local_db()
def _load_local_db(self):
if os.path.exists(self.local_file):
try:
import json
with open(self.local_file, 'r') as f:
data = json.load(f)
# Convert isoformat str back to datetime where needed might be complex,
# but for now we load raw. Dates might need handling.
self.memory_db = data
# Deserialize dates
for user in self.memory_db.get('users', []):
if isinstance(user.get('created_at'), str):
user['created_at'] = datetime.fromisoformat(user['created_at'])
for assessment in user.get('assessments', []):
if isinstance(assessment.get('date'), str):
assessment['date'] = datetime.fromisoformat(assessment['date'])
for journal in self.memory_db.get('journals', []):
if isinstance(journal.get('date'), str):
journal['date'] = datetime.fromisoformat(journal['date'])
print(f"✓ Loaded data from {self.local_file}")
except Exception as e:
print(f"Error loading local db: {e}")
def _save_local_db(self):
if self.db is None:
import json
# Helper to serialize datetime
def json_serial(obj):
if isinstance(obj, datetime):
return obj.isoformat()
raise TypeError ("Type not serializable")
try:
os.makedirs(os.path.dirname(self.local_file), exist_ok=True)
with open(self.local_file, 'w') as f:
json.dump(self.memory_db, f, default=json_serial, indent=2)
except Exception as e:
print(f"Error saving local db: {e}")
# User CRUD
def create_user(self, name, email, password_hash):
if self.db is None:
# Fallback
self.memory_db['users'].append({
"name": name,
"email": email,
"password_hash": password_hash,
"created_at": datetime.now(),
"assessments": []
})
self._save_local_db()
return True
try:
user = {
"name": name,
"email": email,
"password_hash": password_hash,
"created_at": datetime.now(),
"assessments": []
}
self.users.insert_one(user)
return True
except Exception as e:
print(f"Error creating user: {e}")
return False
def get_user(self, email):
if self.db is None:
return next((u for u in self.memory_db['users'] if u['email'] == email), None)
return self.users.find_one({"email": email})
def update_user(self, email, data):
if self.db is None: return False # Simple fallback doesn't support advanced update yet
self.users.update_one({"email": email}, {"$set": data})
return True
# Assessment
def save_assessment(self, email, assessment_data):
if self.db is None:
user = self.get_user(email)
if user:
user['assessments'].append(assessment_data)
self._save_local_db()
return True
assessment_data['date'] = datetime.now()
self.users.update_one(
{"email": email},
{"$push": {"assessments": assessment_data}}
)
return True
def get_history(self, email):
user = self.get_user(email)
return user.get('assessments', []) if user else []
# Journal
def save_journal(self, email, content, mood):
entry = {
"email": email,
"content": content,
"mood": mood,
"date": datetime.now()
}
if self.db is None:
self.memory_db['journals'].append(entry)
self.update_streak(email)
self._save_local_db()
return True
self.journals.insert_one(entry)
self.update_streak(email)
return True
def get_journals(self, email):
if self.db is None:
return sorted([j for j in self.memory_db['journals'] if j['email'] == email], key=lambda x: x['date'], reverse=True)
return list(self.journals.find({"email": email}).sort("date", -1))
# Gamification
def update_streak(self, email):
today = datetime.now().date()
date_str = today.isoformat()
if self.db is None:
streak_data = self.memory_db['streaks'].get(email, {"current_streak": 0, "last_active": "", "max_streak": 0})
if streak_data['last_active']:
last_active = datetime.fromisoformat(streak_data['last_active']).date()
if (today - last_active).days == 1:
streak_data['current_streak'] += 1
elif (today - last_active).days > 1:
streak_data['current_streak'] = 1
else:
streak_data['current_streak'] = 1
streak_data['last_active'] = date_str
streak_data['max_streak'] = max(streak_data['current_streak'], streak_data['max_streak'])
self.memory_db['streaks'][email] = streak_data
return
streak_doc = self.streaks.find_one({"email": email})
if not streak_doc:
self.streaks.insert_one({
"email": email,
"current_streak": 1,
"last_active": date_str,
"max_streak": 1
})
else:
last_active = datetime.fromisoformat(streak_doc['last_active']).date()
if (today - last_active).days == 1:
# Consecutive day
new_streak = streak_doc['current_streak'] + 1
self.streaks.update_one(
{"email": email},
{
"$set": {
"current_streak": new_streak,
"last_active": date_str,
"max_streak": max(new_streak, streak_doc['max_streak'])
}
}
)
elif (today - last_active).days > 1:
# Streak broken
self.streaks.update_one(
{"email": email},
{"$set": {"current_streak": 1, "last_active": date_str}}
)
def get_streak(self, email):
if self.db is None:
return self.memory_db['streaks'].get(email, {}).get('current_streak', 0)
doc = self.streaks.find_one({"email": email})
return doc['current_streak'] if doc else 0
# Global Instance
db = Database()
|