You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
152 lines
5.0 KiB
152 lines
5.0 KiB
from email.policy import default
|
|
from flask import Blueprint, request, jsonify, current_app
|
|
from werkzeug.utils import secure_filename
|
|
from datetime import datetime
|
|
from db.model import db
|
|
from db.model import User, UserRole, Session # Adjust based on your model's location
|
|
from werkzeug.security import generate_password_hash,check_password_hash
|
|
import uuid
|
|
import os
|
|
from config import *
|
|
from utils.utils import password_check_sanity,is_valid_email,InsecurePasswordException
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
profile = Blueprint('profile', __name__)
|
|
|
|
# Function to check allowed file extensions
|
|
def allowed_file(filename):
|
|
"""Check if the uploaded file has an allowed extension."""
|
|
return '.' in filename and filename.rsplit('.', 1)[1].lower() in current_app.config['ALLOWED_EXTENSIONS']
|
|
|
|
@profile.route('/register', methods=['POST'])
|
|
def register():
|
|
"""Handle user registration."""
|
|
data = request.form
|
|
|
|
# Extract form data
|
|
email = data.get('email')
|
|
first_name = data.get('firstName')
|
|
last_name = data.get('lastName')
|
|
username = data.get('username')
|
|
password = data.get('password')
|
|
bio = data.get('bio', '')
|
|
dob = data.get('dob') # Optional field for date of birth
|
|
role = int(data.get('role', UserRole.USER.value)) # Default to 'USER' role if not provided
|
|
is_activated = True # New user will be activated initially
|
|
profile_picture = request.files.get('profile_picture')
|
|
|
|
# Validate email
|
|
if not is_valid_email(email):
|
|
return jsonify({"error": "Invalid email address"}), 400
|
|
|
|
# Validate password
|
|
try:
|
|
password_check_sanity(password)
|
|
except InsecurePasswordException as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
# Validate required fields
|
|
if not all([email, first_name, last_name, username, password]):
|
|
return jsonify({"error": "Missing required fields"}), 400
|
|
|
|
# Check if the file is allowed and save it
|
|
if profile_picture and allowed_file(profile_picture.filename):
|
|
filename = secure_filename(profile_picture.filename)
|
|
# file_path = os.path.join(USER_UPLOADS_DIR, filename)
|
|
profile_picture.save(
|
|
os.path.join(USER_UPLOADS_DIR, filename)
|
|
)
|
|
else:
|
|
filename = DEFAULT_PROFILE_FILE # Use a default image if no file is uploaded
|
|
# Hash the password
|
|
hashed_password = generate_password_hash(password)
|
|
|
|
# Generate activation key (a UUID for example)
|
|
activation_key = str(uuid.uuid4())
|
|
|
|
# Create a new user
|
|
new_user = User(
|
|
email=email,
|
|
firstName=first_name,
|
|
lastName=last_name,
|
|
username=username,
|
|
hash_password=hashed_password,
|
|
bio=bio,
|
|
dob=datetime.fromisoformat(dob) if dob else datetime(2002, 1, 1),
|
|
pfpFilename=filename,
|
|
role=role,
|
|
isActivated=is_activated,
|
|
activationKey=activation_key,
|
|
sessions=[],
|
|
user_badges=[],
|
|
enrollments=[],
|
|
quizzes=[],
|
|
quiz_attempts=[],
|
|
chats=[],
|
|
notifications=[]
|
|
)
|
|
|
|
# Save the user to the database
|
|
try:
|
|
db.session.add(new_user)
|
|
db.session.commit()
|
|
|
|
return jsonify({"message": "User registered successfully."}), 201
|
|
except IntegrityError as e:
|
|
db.session.rollback()
|
|
return jsonify({"error": "User with this email or username already exists."}), 400
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return jsonify({"error": "Registration failed, please try again later."}), 500
|
|
|
|
|
|
@profile.route('/login', methods=['POST'])
|
|
def login():
|
|
"""
|
|
Handle user login.
|
|
"""
|
|
data = request.json # Expecting JSON body
|
|
|
|
# Extract credentials from request
|
|
username = data.get('username')
|
|
password = data.get('password')
|
|
user_agent = request.headers.get('User-Agent', 'Unknown')
|
|
|
|
# Validate required fields
|
|
if not username or not password:
|
|
return jsonify({"error": "Username and password are required"}), 400
|
|
|
|
# Find the user by username
|
|
user = User.query.filter_by(username=username).first()
|
|
|
|
if not user:
|
|
return jsonify({"error": "Invalid username or password"}), 401
|
|
|
|
# Verify the password
|
|
if not check_password_hash(user.hash_password, password):
|
|
return jsonify({"error": "Invalid username or password"}), 401
|
|
|
|
# Create a new session
|
|
session_key = str(uuid.uuid4()) # Generate a unique session key
|
|
new_session = Session(
|
|
userID=user.id,
|
|
user=user, # Pass the user object here
|
|
key=session_key,
|
|
ua=user_agent,
|
|
creationDate=datetime.utcnow(),
|
|
lastUsed=datetime.utcnow(),
|
|
isValid=True
|
|
)
|
|
|
|
|
|
try:
|
|
db.session.add(new_session)
|
|
db.session.commit()
|
|
return jsonify({
|
|
"message": "Login successful",
|
|
"session_key": session_key,
|
|
"user_id": str(user.id)
|
|
}), 200
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
return jsonify({"error": "Login failed, please try again later."}), 500 |