HackXlbef
/
FreeBug
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'main' of https://hackethon.ai/hack/HackXlbef/FreeBug

Browse Source
main
Kushal Dotel 6 months ago
parent 4d0d8b94f6 4dc8d6ab64
commit f477cc5577
5 changed files with 103 additions and 191 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 246
      backend/blueprints/chat/__init__.py
  2. 6
      backend/blueprints/course/__init__.py
  3. 33
      backend/blueprints/profile/__init__.py
  4. 3
      backend/config.py
  5. 2
      backend/db/model.py

246
backend/blueprints/chat/__init__.py
Unescape View File

@ -1,122 +1,49 @@
import uuid
from flask import Blueprint, request, jsonify,g from flask import Blueprint, request, jsonify,g
from uuid import UUID from uuid import UUID
from db.model import db, User, Course, Enrollment,Chat from ...db.model import db, User, Course, Enrollment,Chat
from utils.auth import auth_required from utils.auth import auth_required
import requests import requests
from sqlalchemy import desc from config import SPAM_SCORE_THRESHOLD, AI_SPAM_SERVICES_MICROSERVICE
from sqlalchemy import desc, select, and_
chat = Blueprint('chat', __name__) chat = Blueprint('chat', __name__)
SPAM_DETECTION_URL = "http://localhost:5000/test-spam" @chat.route("/send", methods=["POST"])
@chat.route('/course/<uuid:course_id>/users', methods=['POST'])
def get_users_assigned_to_course(course_id: UUID):
"""
Fetch all users assigned to a specific course.
:param course_id: ID of the course to fetch users for (UUID).
:return: JSON response with users assigned to the course.
"""
try:
# Query the course to ensure it exists
course = Course.query.get(course_id)
if not course:
return jsonify({"error": "Course not found."}), 404
# Get the list of users assigned to the course
users = User.query.filter(User.enrollments.any(course_id=course_id)).all()
# Prepare the response data
user_data = [
{
"id": user.id,
"email": user.email,
"username": user.username,
"firstName": user.firstName,
"lastName": user.lastName,
}
for user in users
]
return jsonify({"course": course.name, "users": user_data}), 200
except Exception as e:
return jsonify({"error": f"An error occurred: {str(e)}"}), 500
@chat.route('/course/<uuid:course_id>/chat', methods=['POST'])
def join_chat(course_id: UUID):
"""
Allow users to join the chat only if they are enrolled in the course.
:param course_id: ID of the course (UUID).
:return: JSON response indicating whether the user can join or not.
"""
try:
# Get the user from the request (assume user_id is passed in the request body)
user_id = request.json.get('user_id')
if not user_id:
return jsonify({"error": "User ID is required."}), 400
# Query the user and ensure they exist
user = User.query.get(user_id)
if not user:
return jsonify({"error": "User not found."}), 404
# Check if the user is enrolled in the course
enrollment = Enrollment.query.filter_by(userID=user_id, courseID=course_id).first()
if not enrollment:
return jsonify({"error": "User is not enrolled in this course."}), 403
# If enrolled, allow the user to join the chat
return jsonify({"message": f"User {user.username} is enrolled in the course and can join the chat."}), 200
except Exception as e:
return jsonify({"error": f"An error occurred: {str(e)}"}), 500
@chat.route("/send-message", methods=["POST"])
@auth_required() @auth_required()
def create_chat(): def create_chat():
""" current_user: User = g.current_user # Fetch the logged-in user
Create a chat message for a specific course. data = request.args
""" if not data.get("course_id"):
current_user = g.current_user # Fetch the logged-in user return jsonify({"error": "Unknown Course"}), 400
data = request.get_json() if not data.get("message", "").strip():
return jsonify({'message': 'Silently rejected blank message'}), 200
# Validate the payload
course_id = data.get("course_id")
message = data.get("message")
if not course_id or not message:
return jsonify({"error": "Course ID and message are required."}), 400
try: try:
# Call the spam detection service # Check if the user is enrolled in the course
spam_response = requests.post(SPAM_DETECTION_URL, json={"test_message": message}) enrollment_record: Enrollment = db.session.execute(
select(Enrollment).where(and_(
Enrollment.courseID == data.get("course_id"),
Enrollment.userID == current_user.id)
)
)
if not enrollment_record:
return jsonify({"error": "You are not enrolled in this course."}), 403
if enrollment_record.currentPage < enrollment_record.course.pageForCommunity:
return jsonify({
'message': 'You have not met the requirements to enter the chat.'
}), 403
# Create and save the chat message
spam_response = requests.post(AI_SPAM_SERVICES_MICROSERVICE, json={"test_message": data.get("message").strip()})
if spam_response.status_code != 200: if spam_response.status_code != 200:
return jsonify({"error": "Failed to check message for spam."}), 500 return jsonify({"error": "Failed to check message for spam."}), 500
spam_score = int(spam_response.json().get("spam_score", 0)) spam_score = int(spam_response.json().get("spam_score", 0))
if spam_score > SPAM_SCORE_THRESHOLD:
if spam_score > 6:
return jsonify({"error": "This message contains suspicious links or is vulnerable."}), 400 return jsonify({"error": "This message contains suspicious links or is vulnerable."}), 400
# Verify the course exists
course = db.session.query(Course).filter_by(id=course_id).first()
if not course:
return jsonify({"error": "Invalid course ID."}), 404
# Check if the user is enrolled in the course
is_enrolled = db.session.query(Course).join(Enrollment).filter(
Enrollment.userID == current_user.id,
Enrollment.courseID == course_id
).first()
if not is_enrolled:
return jsonify({"error": "You are not enrolled in this course."}), 403
# Create and save the chat message
new_chat = Chat( new_chat = Chat(
textContent=message, textContent=data.get("message").strip(),
userID=current_user.id, userID=current_user.id,
courseID=course_id, courseID=data.get("course_id"),
) )
db.session.add(new_chat) db.session.add(new_chat)
db.session.commit() db.session.commit()
@ -128,56 +55,71 @@ def create_chat():
@chat.route("/get", methods=["GET"]) @chat.route("/get", methods=["GET"])
@auth_required() @auth_required()
def get_chat_history(): def get_messages():
"""
Fetch chat history for a course.
"""
current_user = g.current_user # Logged-in user
course_id = request.args.get("course_id")
limit = int(request.args.get("limit", 20)) # Default to 20 messages
offset = int(request.args.get("offset", 0)) # Default to no offset (latest chats)
if not course_id:
return jsonify({"error": "Course ID is required."}), 400
try: try:
# Verify the course exists course_id: uuid.UUID = uuid.UUID(request.args.get('course_id'))
course = db.session.query(Course).filter_by(id=course_id).first() current_user: User = g.current_user
if not course: limit = int(request.args.get('limit', 10))
return jsonify({"error": "Invalid course ID."}), 404 before_id = request.args.get('before')
after_id = request.args.get('after')
# Check if the user is enrolled in the course # Verify user's enrollment
is_enrolled = db.session.query(Course).join(Enrollment).filter( enrollment = db.session.execute(
Enrollment.userID == current_user.id, select(Enrollment).where(
Enrollment.courseID == course_id and_(Enrollment.courseID == course_id, Enrollment.userID == current_user.id)
).first()
if not is_enrolled:
return jsonify({"error": "You are not enrolled in this course."}), 403
# Fetch the latest chat messages with limit and offset
chats = (
db.session.query(Chat)
.filter_by(courseID=course_id)
.order_by(desc(Chat.chatDate))
.offset(offset)
.limit(limit)
.all()
) )
).scalar()
# Format the chat messages if not enrollment:
chat_history = [ return jsonify({"error": "You are not enrolled in this course."}), 403
{ query = select(Chat).where(Chat.courseID == course_id)
"id": str(chat.id), if before_id:
"textContent": chat.textContent, try:
"userID": str(chat.userID), reference_message: Chat = db.session.execute(
"chatDate": chat.chatDate.isoformat(), select(Chat).where(Chat.id == uuid.UUID(before_id))
"is_mine": chat.userID == current_user.id, ).scalar()
} if not reference_message:
for chat in chats return jsonify({'message': 'Reference message not found'}), 404
] query = query.order_by(Chat.chatDate.desc()).where(Chat.chatDate < reference_message.chatDate)
except ValueError:
return jsonify({"chats": chat_history}), 200 return jsonify({'message': 'Invalid message ID format'}), 400
elif after_id:
try:
reference_message = db.session.execute(
select(Chat).where(Chat.id == uuid.UUID(after_id))
).scalar()
if not reference_message:
return jsonify({'message': 'Reference message not found'}), 404
query = query.order_by(Chat.chatDate.asc()).where(Chat.chatDate > reference_message.chatDate)
# For after queries, reverse the order later
query = query.order_by(Chat.chatDate)
except ValueError:
return jsonify({'message': 'Invalid message ID format'}), 400
else:
# Default ordering
query = query.order_by(desc(Chat.chatDate))
# Apply limit and execute query
query = query.limit(limit)
messages = list(db.session.execute(query).scalars())
# Reverse the order for 'after' queries to maintain consistency
if after_id:
messages.reverse()
# Format messages
chat_messages = [{
'id': str(msg.id),
'text': msg.textContent,
'userId': str(msg.userID),
'username': msg.user.username,
'timestamp': msg.chatDate.isoformat()
} for msg in messages]
return jsonify({
'messages': chat_messages,
'count': len(chat_messages),
'hasMore': len(chat_messages) == limit
}), 200
except Exception as e: except Exception as e:
return jsonify({"error": f"An error occurred: {str(e)}"}), 500 return jsonify({'message': f'An error occurred: {str(e)}'}), 500

6
backend/blueprints/course/__init__.py
Unescape View File

@ -110,9 +110,6 @@ def enroll_user():
return jsonify({'message': 'Already enrolled to this course'}) return jsonify({'message': 'Already enrolled to this course'})
return jsonify({'message': 'Enrollment successful'}), 200 return jsonify({'message': 'Enrollment successful'}), 200
@course.route('/createCourse', methods=['POST']) @course.route('/createCourse', methods=['POST'])
@auth_required() @auth_required()
def create_course(): def create_course():
@ -134,7 +131,7 @@ def create_course():
return jsonify({'message': 'Course name cannot be empty'}), 401 return jsonify({'message': 'Course name cannot be empty'}), 401
course_description: str = form_data.get('course_description', '') course_description: str = form_data.get('course_description', '')
category_id: uuid.UUID = uuid.UUID(form_data['category_uuid']) category_id: uuid.UUID = uuid.UUID(form_data['category_uuid'])
pages_required_for_community: int = int(form_data['community_unlock_at_pages']) # TODO: Add this field to model page_for_community: int = int(form_data.get('page_for_community', 1)) # TODO: Add this field to model
new_course: Course = Course( new_course: Course = Course(
name=course_name, name=course_name,
@ -142,6 +139,7 @@ def create_course():
authorID=g.current_user.id, authorID=g.current_user.id,
description=course_description, description=course_description,
isActive=True, isActive=True,
pageForCommunity=page_for_community,
publishedStatus=int(published_status), publishedStatus=int(published_status),
coverImage=cover_file_name, coverImage=cover_file_name,
serverFilename =pdf_file_name, serverFilename =pdf_file_name,

33
backend/blueprints/profile/__init__.py
Unescape View File

@ -15,7 +15,6 @@ from sqlalchemy.exc import IntegrityError
# from flask import url_for # from flask import url_for
profile = Blueprint('profile', __name__) profile = Blueprint('profile', __name__)
# Function to check allowed file extensions
def allowed_file(filename): def allowed_file(filename):
"""Check if the uploaded file has an allowed extension.""" """Check if the uploaded file has an allowed extension."""
return '.' in filename and filename.rsplit('.', 1)[1].lower() in current_app.config['ALLOWED_EXTENSIONS'] return '.' in filename and filename.rsplit('.', 1)[1].lower() in current_app.config['ALLOWED_EXTENSIONS']
@ -110,13 +109,8 @@ def manage_profile():
Handle GET and PUT requests for the logged-in user's profile. Handle GET and PUT requests for the logged-in user's profile.
""" """
current_user: User = g.current_user current_user: User = g.current_user
if request.method == 'GET': if request.method == 'GET':
try:
profile_picture = url_for('send_file', filename=current_user.pfpFilename, _external=True) profile_picture = url_for('send_file', filename=current_user.pfpFilename, _external=True)
except:
profile_picture = ""
try: try:
# Construct the user profile data # Construct the user profile data
profile_data = { profile_data = {
@ -133,24 +127,17 @@ def manage_profile():
"pfp_filename": current_user.pfpFilename, "pfp_filename": current_user.pfpFilename,
"profile_picture": profile_picture, "profile_picture": profile_picture,
} }
return jsonify({"profile": profile_data}), 200 return jsonify({"profile": profile_data}), 200
except Exception as e: except Exception as e:
return jsonify({"error": f"Failed to fetch profile. Error: {str(e)}"}), 500 return jsonify({"error": f"Failed to fetch profile. Error: {str(e)}"}), 500
elif request.method == 'PUT': elif request.method == 'PUT':
# Update the user's profile using form data # Update the user's profile using form data
try: try:
# email = request.form.get('email')
first_name = request.form.get('firstName') first_name = request.form.get('firstName')
last_name = request.form.get('lastName') last_name = request.form.get('lastName')
username = request.form.get('username') username = request.form.get('username')
dob = request.form.get('dob') dob = request.form.get('dob')
bio = request.form.get('bio') bio = request.form.get('bio')
# Update fields if provided
# if email:
# current_user.email = email
if first_name: if first_name:
current_user.firstName = first_name current_user.firstName = first_name
if last_name: if last_name:
@ -161,10 +148,7 @@ def manage_profile():
current_user.dob = dob # Ensure the date format is validated current_user.dob = dob # Ensure the date format is validated
if bio: if bio:
current_user.bio = bio current_user.bio = bio
# Commit changes to the database
db.session.commit() db.session.commit()
return jsonify({"message": "Profile updated successfully."}), 200 return jsonify({"message": "Profile updated successfully."}), 200
except IntegrityError: except IntegrityError:
db.session.rollback() db.session.rollback()
@ -174,7 +158,6 @@ def manage_profile():
return jsonify({"error": f"Failed to update profile. Error: {str(e)}"}), 500 return jsonify({"error": f"Failed to update profile. Error: {str(e)}"}), 500
@profile.route('/update-profile-picture', methods=['PATCH']) @profile.route('/update-profile-picture', methods=['PATCH'])
@auth_required() @auth_required()
def update_profile_picture(): def update_profile_picture():
@ -194,22 +177,18 @@ def update_profile_picture():
return jsonify({"error": "No selected file"}), 400 return jsonify({"error": "No selected file"}), 400
if not allowed_file(file.filename): if not allowed_file(file.filename):
return jsonify({"error": "Invalid file type"}), 400 return jsonify({"error": "Invalid file type"}), 400
# Secure the filename and save the new file # Secure the filename and save the new file
filename = secure_filename(f"user_{user.id}_{file.filename}") filename = secure_filename(f"user_{user.id}_{file.filename}")
new_filepath = os.path.join(USER_UPLOADS_DIR, filename) new_filepath = os.path.join(USER_UPLOADS_DIR, filename)
file.save(new_filepath) file.save(new_filepath)
# Delete the old profile picture (if it's not the default) # Delete the old profile picture (if it's not the default)
if user.pfpFilename != DEFAULT_PROFILE_FILE: if user.pfpFilename != DEFAULT_PROFILE_FILE:
old_filepath = os.path.join(USER_UPLOADS_DIR, user.pfpFilename) old_filepath = os.path.join(USER_UPLOADS_DIR, user.pfpFilename)
if os.path.exists(old_filepath): if os.path.exists(old_filepath):
os.remove(old_filepath) os.remove(old_filepath)
# Update the user's profile picture # Update the user's profile picture
user.pfpFilename = filename user.pfpFilename = filename
db.session.commit() db.session.commit()
# Generate the new profile URL # Generate the new profile URL
profile_url = url_for('send_file',filename=user.pfpFilename,_external=True) profile_url = url_for('send_file',filename=user.pfpFilename,_external=True)
@ -232,37 +211,25 @@ def change_password():
""" """
user = g.current_user user = g.current_user
data = request.form data = request.form
# Validate input data # Validate input data
current_password = data.get('current_password') current_password = data.get('current_password')
new_password = data.get('new_password') new_password = data.get('new_password')
confirm_password = data.get('confirm_password') confirm_password = data.get('confirm_password')
if not current_password or not new_password or not confirm_password: if not current_password or not new_password or not confirm_password:
return jsonify({"error": "All fields (current_password, new_password, confirm_password) are required"}), 400 return jsonify({"error": "All fields (current_password, new_password, confirm_password) are required"}), 400
# Check if current password matches the user's existing password # Check if current password matches the user's existing password
if not check_password_hash(user.hash_password, current_password): if not check_password_hash(user.hash_password, current_password):
return jsonify({"error": "Current password is incorrect"}), 400 return jsonify({"error": "Current password is incorrect"}), 400
# Check if new password and confirmation match # Check if new password and confirmation match
if new_password != confirm_password: if new_password != confirm_password:
return jsonify({"error": "New password and confirm password do not match"}), 400 return jsonify({"error": "New password and confirm password do not match"}), 400
# Check for password complexity (optional) # Check for password complexity (optional)
# Validate password # Validate password
try: try:
password_check_sanity(new_password) password_check_sanity(new_password)
except InsecurePasswordException as e: except InsecurePasswordException as e:
return jsonify({"error": str(e)}), 400 return jsonify({"error": str(e)}), 400
# Update the user's password # Update the user's password
user.hash_password = generate_password_hash(new_password) user.hash_password = generate_password_hash(new_password)
db.session.commit() db.session.commit()
return jsonify({"message": "Password updated successfully"}), 200 return jsonify({"message": "Password updated successfully"}), 200
# @profile.route('/hello')
# @auth_required()
# @requires_role([UserRole.ADMIN])

3
backend/config.py
Unescape View File

@ -20,6 +20,9 @@ DISABLE_PASSWORD_SANITY_CHECKS: bool = False
PROJECT_ROOT: os.path = os.path.dirname(os.path.abspath(__file__)) PROJECT_ROOT: os.path = os.path.dirname(os.path.abspath(__file__))
USER_UPLOADS_DIR: str = os.path.abspath(os.path.join(PROJECT_ROOT, "uploads")) USER_UPLOADS_DIR: str = os.path.abspath(os.path.join(PROJECT_ROOT, "uploads"))
SPAM_SCORE_THRESHOLD: int = 6
AI_SPAM_SERVICES_MICROSERVICE: str ='http://localhost:5000/test-spam'
DB_URI: str = f"{DB_ENGINE}://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}" DB_URI: str = f"{DB_ENGINE}://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
ACTIVATE_ACCOUNTS_ON_SIGNUP: bool = True ACTIVATE_ACCOUNTS_ON_SIGNUP: bool = True

2
backend/db/model.py
Unescape View File

@ -73,6 +73,8 @@ class Course(db.Model):
authorID: Mapped[uuid.UUID] = mapped_column(ForeignKey("user.id")) authorID: Mapped[uuid.UUID] = mapped_column(ForeignKey("user.id"))
author: Mapped["User"] = relationship(back_populates="publications") author: Mapped["User"] = relationship(back_populates="publications")
description: Mapped[str] = mapped_column(String(1024), nullable=False, default='') description: Mapped[str] = mapped_column(String(1024), nullable=False, default='')
pageForCommunity: Mapped[[int]] = mapped_column(Integer, nullable=False, default=1)
totalPages: Mapped[[int]] = mapped_column(Integer, nullable=False, default=1)
totalEnrolled: Mapped[int] = mapped_column(Integer, nullable=False, default='') totalEnrolled: Mapped[int] = mapped_column(Integer, nullable=False, default='')
isActive: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) isActive: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
publishedStatus: Mapped[int] = mapped_column(SmallInteger, nullable=False, default=PublishedStatus.DRAFT) publishedStatus: Mapped[int] = mapped_column(SmallInteger, nullable=False, default=PublishedStatus.DRAFT)

Write Preview
Loading…
Cancel
Save