Spaces:
Sleeping
Sleeping
File size: 5,848 Bytes
6d6b8af |
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 |
"""
This module manages user profiles and preferences.
"""
import json
from typing import Dict, Any, List, Optional
from datetime import datetime
from utils.database import Database
class UserProfile:
"""Manages user profiles and preferences"""
def __init__(self, db: Database):
self.db = db
self.cache = {}
self.last_update = {}
def get_profile(self, user_id: int) -> Dict[str, Any]:
"""
Get a user's profile data.
Args:
user_id: The ID of the user
Returns:
User profile data
"""
# Check cache first
if user_id in self.cache:
return self.cache[user_id]
# Query database
cursor = self.db.connection.cursor()
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
result = cursor.fetchone()
if result:
profile = {
"user_id": result[0],
"preferences": json.loads(result[1]) if result[1] else {},
"settings": json.loads(result[2]) if result[2] else {},
"created_at": result[3],
"updated_at": result[4]
}
# Update cache
self.cache[user_id] = profile
self.last_update[user_id] = datetime.now()
return profile
else:
return self._create_default_profile(user_id)
def update_profile(self, user_id: int, updates: Dict[str, Any]) -> bool:
"""
Update a user's profile data.
Args:
user_id: The ID of the user
updates: Dictionary of updates to apply
Returns:
Success status
"""
try:
current = self.get_profile(user_id)
# Update preferences
if "preferences" in updates:
current["preferences"].update(updates["preferences"])
# Update settings
if "settings" in updates:
current["settings"].update(updates["settings"])
# Save to database
cursor = self.db.connection.cursor()
cursor.execute(
"""
UPDATE users
SET preferences = ?, settings = ?, updated_at = ?
WHERE id = ?
""",
(
json.dumps(current["preferences"]),
json.dumps(current["settings"]),
datetime.now().isoformat(),
user_id
)
)
self.db.connection.commit()
# Update cache
self.cache[user_id] = current
self.last_update[user_id] = datetime.now()
return True
except Exception as e:
print(f"Error updating profile: {e}")
return False
def _create_default_profile(self, user_id: int) -> Dict[str, Any]:
"""
Create a default profile for a new user.
Args:
user_id: The ID of the user
Returns:
Default profile data
"""
now = datetime.now().isoformat()
profile = {
"user_id": user_id,
"preferences": {
"language": "en",
"theme": "light",
"notifications": True
},
"settings": {
"privacy_level": "standard",
"auto_save": True,
"sync_enabled": True
},
"created_at": now,
"updated_at": now
}
# Save to database
cursor = self.db.connection.cursor()
cursor.execute(
"""
INSERT INTO users (id, preferences, settings, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
""",
(
user_id,
json.dumps(profile["preferences"]),
json.dumps(profile["settings"]),
profile["created_at"],
profile["updated_at"]
)
)
self.db.connection.commit()
# Update cache
self.cache[user_id] = profile
self.last_update[user_id] = datetime.now()
return profile
def get_preference(self, user_id: int, key: str) -> Any:
"""
Get a specific user preference.
Args:
user_id: The ID of the user
key: The preference key to retrieve
Returns:
The preference value if found, None otherwise
"""
profile = self.get_profile(user_id)
return profile["preferences"].get(key)
def get_setting(self, user_id: int, key: str) -> Any:
"""
Get a specific user setting.
Args:
user_id: The ID of the user
key: The setting key to retrieve
Returns:
The setting value if found, None otherwise
"""
profile = self.get_profile(user_id)
return profile["settings"].get(key)
def clear_cache(self, user_id: Optional[int] = None):
"""
Clear the profile cache.
Args:
user_id: Optional specific user ID to clear
"""
if user_id is not None:
self.cache.pop(user_id, None)
self.last_update.pop(user_id, None)
else:
self.cache = {}
self.last_update = {} |