90 lines
No EOL
2.9 KiB
Python
90 lines
No EOL
2.9 KiB
Python
import bcrypt
|
|
from bson import ObjectId
|
|
from app.db import get_db
|
|
|
|
class User:
|
|
def __init__(self, username, email, password_hash, role="user"):
|
|
self.username = username
|
|
self.email = email
|
|
self.password_hash = password_hash
|
|
self.role = role
|
|
|
|
@staticmethod
|
|
def hash_password(password):
|
|
salt = bcrypt.gensalt()
|
|
hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
|
|
return hashed.decode('utf-8')
|
|
|
|
@staticmethod
|
|
def check_password(password_hash, password):
|
|
return bcrypt.checkpw(password.encode('utf-8'), password_hash.encode('utf-8'))
|
|
|
|
@staticmethod
|
|
def find_by_username(username):
|
|
db = get_db()
|
|
user_data = db.users.find_one({"username": username})
|
|
return user_data
|
|
|
|
@staticmethod
|
|
def find_by_email(email):
|
|
db = get_db()
|
|
user_data = db.users.find_one({"email": email})
|
|
return user_data
|
|
|
|
@staticmethod
|
|
def find_by_id(user_id):
|
|
db = get_db()
|
|
user_data = db.users.find_one({"_id": ObjectId(user_id)})
|
|
return user_data
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"username": self.username,
|
|
"email": self.email,
|
|
"role": self.role
|
|
}
|
|
|
|
def save(self):
|
|
db = get_db()
|
|
user_data = {
|
|
"username": self.username,
|
|
"email": self.email,
|
|
"password_hash": self.password_hash,
|
|
"role": self.role
|
|
}
|
|
result = db.users.insert_one(user_data)
|
|
return result.inserted_id
|
|
|
|
@staticmethod
|
|
def create_default_user():
|
|
try:
|
|
db = get_db()
|
|
|
|
# First check if users collection exists
|
|
collections = db.list_collection_names()
|
|
if "users" not in collections:
|
|
print("Creating users collection")
|
|
db.create_collection("users")
|
|
|
|
# Safely check if user exists, handling potential auth errors
|
|
try:
|
|
user_exists = db.users.count_documents({"username": "user"}) > 0
|
|
except Exception as e:
|
|
print(f"Error checking for default user: {e}")
|
|
# If we can't query, assume we need to create the user
|
|
user_exists = False
|
|
|
|
if not user_exists:
|
|
default_user = User(
|
|
username="user",
|
|
email="user@example.com",
|
|
password_hash=User.hash_password("pass"),
|
|
role="admin"
|
|
)
|
|
default_user.save()
|
|
print("Default user created successfully")
|
|
else:
|
|
print("Default user already exists")
|
|
except Exception as e:
|
|
print(f"Error creating default user: {e}")
|
|
# Don't raise the exception - allow the app to continue even if we can't create the user |