Spaces:
Sleeping
Sleeping
File size: 3,652 Bytes
cd570f9 39678c7 cd570f9 39678c7 cd570f9 39678c7 cd570f9 39678c7 cd570f9 |
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 |
import psycopg2
from psycopg2 import sql
import cloudinary.uploader
from user_auth.user import User
from settings.base import setup_logger, DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER, cloudinary
logger = setup_logger()
class UserManager:
def __init__(self):
self.conn = psycopg2.connect(
dbname=DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT
)
self.cursor = self.conn.cursor()
def create_user(self, email, password, profile):
if self.get_user(email):
return False, "User already exists!"
user = User(email, password, profile)
try:
self.cursor.execute(
"""
INSERT INTO users (username, password, name, age, gender, height, weight, location, allergies, spec_diet_pref, primary_goal, health_condition, activity_level, daily_calorie_intake)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(email, user.password_hash.decode("utf-8"), profile['name'], profile['age'], profile['gender'], profile['height'], profile['weight'], profile['location'], profile['allergies'], profile['spec_diet_pref'], profile['primary_goal'], profile['health_condition'], profile['activity_level'], profile['daily_calorie_intake'])
)
self.conn.commit()
logger.info(f'User created successfully: {email}')
return True, "Account created successfully!"
except Exception as e:
self.conn.rollback()
logger.error(f'Error creating user: {e}')
return False, str(e)
def get_user(self, email):
self.cursor.execute("SELECT * FROM users WHERE username = %s", (email,))
result = self.cursor.fetchone()
if result:
profile = {
"name": result[3],
"age": result[4],
"gender": result[5],
"height": result[6],
"weight": result[7],
"location": result[8],
"allergies": result[9],
"spec_diet_pref": result[10],
"primary_goal": result[11],
"health_condition": result[12],
"activity_level": result[13],
"daily_calorie_intake": result[14]
}
logger.info(f'User retrieved from database: {result[1]}')
return User(email=result[1], password=result[2], profile=profile)
return None
def save_history(self, email, history_json):
try:
with self.conn.cursor() as cur:
query = sql.SQL("""
INSERT INTO chat_history (username, data)
VALUES (%s, %s)
""")
cur.execute(query, (email, history_json))
self.conn.commit()
logger.info(f"Chat history saved to database for user: {email}")
except Exception as e:
self.conn.rollback()
logger.error(f"Error saving chat history to database: {e}")
def upload_to_cloudinary(self, file_path):
try:
# Upload file to Cloudinary
upload_result = cloudinary.uploader.upload(file_path)
uploaded_file_url = upload_result['url']
logger.info(f"Uploaded file to Cloudinary: {uploaded_file_url}")
return uploaded_file_url
except Exception as e:
logger.exception(f"Error uploading file to Cloudinary: {repr(e)}")
return None
def close(self):
self.cursor.close()
self.conn.close()
|