385 lines
15 KiB
Python
385 lines
15 KiB
Python
from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify
|
|
from flask_login import login_required, current_user
|
|
from models import db, Room, User, RoomMemberPermission, RoomFile, Notif
|
|
from forms import RoomForm
|
|
from routes.room_files import user_has_permission
|
|
from routes.auth import require_password_change
|
|
from utils import log_event, create_notification, get_unread_count
|
|
import os
|
|
import shutil
|
|
from datetime import datetime
|
|
|
|
rooms_bp = Blueprint('rooms', __name__, url_prefix='/rooms')
|
|
|
|
@rooms_bp.context_processor
|
|
def inject_unread_notifications():
|
|
if current_user.is_authenticated:
|
|
unread_count = get_unread_count(current_user.id)
|
|
return {'unread_notifications': unread_count}
|
|
return {'unread_notifications': 0}
|
|
|
|
@rooms_bp.route('/')
|
|
@login_required
|
|
@require_password_change
|
|
def rooms():
|
|
search = request.args.get('search', '').strip()
|
|
if current_user.is_admin:
|
|
query = Room.query
|
|
else:
|
|
# Get rooms where user is a member and has view permissions
|
|
query = Room.query.join(
|
|
RoomMemberPermission,
|
|
(Room.id == RoomMemberPermission.room_id) &
|
|
(RoomMemberPermission.user_id == current_user.id) &
|
|
(RoomMemberPermission.can_view == True)
|
|
)
|
|
if search:
|
|
query = query.filter(Room.name.ilike(f'%{search}%'))
|
|
rooms = query.order_by(Room.created_at.desc()).all()
|
|
return render_template('rooms/rooms.html', rooms=rooms, search=search)
|
|
|
|
@rooms_bp.route('/create', methods=['GET', 'POST'])
|
|
@login_required
|
|
@require_password_change
|
|
def create_room():
|
|
form = RoomForm()
|
|
if form.validate_on_submit():
|
|
room = Room(
|
|
name=form.name.data,
|
|
description=form.description.data,
|
|
created_by=current_user.id
|
|
)
|
|
|
|
# Add creator as a member with full permissions
|
|
room.members.append(current_user)
|
|
creator_permission = RoomMemberPermission(
|
|
room=room,
|
|
user=current_user,
|
|
can_view=True,
|
|
can_upload=True,
|
|
can_delete=True,
|
|
can_share=True
|
|
)
|
|
db.session.add(room)
|
|
db.session.add(creator_permission)
|
|
db.session.commit()
|
|
|
|
log_event(
|
|
event_type='room_create',
|
|
details={
|
|
'room_id': room.id,
|
|
'room_name': room.name,
|
|
'description': room.description,
|
|
'created_by': f"{current_user.username} {current_user.last_name}"
|
|
},
|
|
user_id=current_user.id
|
|
)
|
|
db.session.commit()
|
|
|
|
flash('Room created successfully!', 'success')
|
|
return redirect(url_for('rooms.rooms'))
|
|
return render_template('rooms/create_room.html', form=form)
|
|
|
|
@rooms_bp.route('/<int:room_id>')
|
|
@login_required
|
|
@require_password_change
|
|
def room(room_id):
|
|
room = Room.query.get_or_404(room_id)
|
|
# Admins always have access, managers need to be members
|
|
if not current_user.is_admin:
|
|
is_member = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=current_user.id).first() is not None
|
|
if not is_member:
|
|
flash('You do not have access to this room.', 'error')
|
|
return redirect(url_for('rooms.rooms'))
|
|
can_download = user_has_permission(room, 'can_download')
|
|
can_upload = user_has_permission(room, 'can_upload')
|
|
can_delete = user_has_permission(room, 'can_delete')
|
|
can_rename = user_has_permission(room, 'can_rename')
|
|
can_move = user_has_permission(room, 'can_move')
|
|
can_share = user_has_permission(room, 'can_share')
|
|
|
|
log_event(
|
|
event_type='room_open',
|
|
details={
|
|
'room_id': room_id,
|
|
'room_name': room.name,
|
|
'accessed_by': f"{current_user.username} {current_user.last_name}"
|
|
},
|
|
user_id=current_user.id
|
|
)
|
|
db.session.commit()
|
|
|
|
return render_template('rooms/room.html', room=room, can_download=can_download, can_upload=can_upload, can_delete=can_delete, can_rename=can_rename, can_move=can_move, can_share=can_share)
|
|
|
|
@rooms_bp.route('/<int:room_id>/members')
|
|
@login_required
|
|
@require_password_change
|
|
def room_members(room_id):
|
|
room = Room.query.get_or_404(room_id)
|
|
# Check if user is a member
|
|
if not current_user.is_admin:
|
|
is_member = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=current_user.id).first() is not None
|
|
if not is_member:
|
|
flash('You do not have access to this room.', 'error')
|
|
return redirect(url_for('rooms.rooms'))
|
|
# Only admins and managers who are members can manage room members
|
|
if not (current_user.is_admin or (current_user.is_manager and is_member)):
|
|
flash('Only administrators and managers can manage room members.', 'error')
|
|
return redirect(url_for('rooms.room', room_id=room_id))
|
|
member_permissions = {p.user_id: p for p in room.member_permissions}
|
|
available_users = User.query.filter(~User.id.in_(member_permissions.keys())).all()
|
|
return render_template('rooms/room_members.html', room=room, available_users=available_users, member_permissions=member_permissions)
|
|
|
|
@rooms_bp.route('/<int:room_id>/members/add', methods=['POST'])
|
|
@login_required
|
|
@require_password_change
|
|
def add_member(room_id):
|
|
room = Room.query.get_or_404(room_id)
|
|
# Membership check using RoomMemberPermission
|
|
is_member = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=current_user.id).first() is not None
|
|
if not is_member:
|
|
flash('You do not have access to this room.', 'error')
|
|
return redirect(url_for('rooms.rooms'))
|
|
# Only admins and managers who are members can manage room members
|
|
if not (current_user.is_admin or (current_user.is_manager and is_member)):
|
|
flash('Only administrators and managers can manage room members.', 'error')
|
|
return redirect(url_for('rooms.room', room_id=room_id))
|
|
user_id = request.form.get('user_id')
|
|
if not user_id:
|
|
flash('Please select a user to add.', 'error')
|
|
return redirect(url_for('rooms.room_members', room_id=room_id))
|
|
user = User.query.get_or_404(user_id)
|
|
# Check if already a member
|
|
if RoomMemberPermission.query.filter_by(room_id=room_id, user_id=user.id).first():
|
|
flash('User is already a member of this room.', 'error')
|
|
else:
|
|
perm = RoomMemberPermission(room_id=room_id, user_id=user.id, can_view=True)
|
|
db.session.add(perm)
|
|
# Ensure user is added to the room.members relationship
|
|
if user not in room.members:
|
|
room.members.append(user)
|
|
|
|
try:
|
|
# Create notification for the invited user
|
|
create_notification(
|
|
notif_type='room_invite',
|
|
user_id=user.id,
|
|
sender_id=current_user.id,
|
|
details={
|
|
'message': f'You have been invited to join room "{room.name}"',
|
|
'room_id': room_id,
|
|
'room_name': room.name,
|
|
'invited_by': f"{current_user.username} {current_user.last_name}",
|
|
'permissions': {
|
|
'can_view': True,
|
|
'can_download': False,
|
|
'can_upload': False,
|
|
'can_delete': False,
|
|
'can_rename': False,
|
|
'can_move': False,
|
|
'can_share': False
|
|
},
|
|
'timestamp': datetime.utcnow().isoformat()
|
|
}
|
|
)
|
|
|
|
log_event(
|
|
event_type='room_member_add',
|
|
details={
|
|
'room_id': room_id,
|
|
'room_name': room.name,
|
|
'added_user': f"{user.username} {user.last_name}",
|
|
'added_by': f"{current_user.username} {current_user.last_name}"
|
|
},
|
|
user_id=current_user.id
|
|
)
|
|
|
|
db.session.commit()
|
|
flash(f'{user.username} has been added to the room.', 'success')
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash('An error occurred while adding the member.', 'error')
|
|
print(f"Error adding member: {str(e)}")
|
|
|
|
return redirect(url_for('rooms.room_members', room_id=room_id))
|
|
|
|
@rooms_bp.route('/<int:room_id>/members/<int:user_id>/remove', methods=['POST'])
|
|
@login_required
|
|
@require_password_change
|
|
def remove_member(room_id, user_id):
|
|
room = Room.query.get_or_404(room_id)
|
|
# Membership check using RoomMemberPermission
|
|
is_member = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=current_user.id).first() is not None
|
|
if not is_member:
|
|
flash('You do not have access to this room.', 'error')
|
|
return redirect(url_for('rooms.rooms'))
|
|
# Only admins and managers who are members can manage room members
|
|
if not (current_user.is_admin or (current_user.is_manager and is_member)):
|
|
flash('Only administrators and managers can manage room members.', 'error')
|
|
return redirect(url_for('rooms.room', room_id=room_id))
|
|
if user_id == room.created_by:
|
|
flash('Cannot remove the room creator.', 'error')
|
|
else:
|
|
perm = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=user_id).first()
|
|
if perm:
|
|
db.session.delete(perm)
|
|
db.session.commit()
|
|
flash('Member has been removed from the room.', 'success')
|
|
else:
|
|
flash('Member not found.', 'error')
|
|
return redirect(url_for('rooms.room_members', room_id=room_id))
|
|
|
|
@rooms_bp.route('/<int:room_id>/members/<int:user_id>/permissions', methods=['POST'])
|
|
@login_required
|
|
def update_member_permissions(room_id, user_id):
|
|
room = Room.query.get_or_404(room_id)
|
|
# Check if user is a member
|
|
is_member = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=current_user.id).first() is not None
|
|
if not (current_user.is_admin or (current_user.is_manager and is_member)):
|
|
flash('Only administrators and managers can update permissions.', 'error')
|
|
return redirect(url_for('rooms.room_members', room_id=room_id))
|
|
perm = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=user_id).first()
|
|
if not perm:
|
|
flash('Member not found.', 'error')
|
|
return redirect(url_for('rooms.room_members', room_id=room_id))
|
|
|
|
# Get the user object
|
|
user = User.query.get(user_id)
|
|
if not user:
|
|
flash('User not found.', 'error')
|
|
return redirect(url_for('rooms.room_members', room_id=room_id))
|
|
|
|
perm.can_view = bool(request.form.get('can_view'))
|
|
perm.can_download = bool(request.form.get('can_download'))
|
|
perm.can_upload = bool(request.form.get('can_upload'))
|
|
perm.can_delete = bool(request.form.get('can_delete'))
|
|
perm.can_rename = bool(request.form.get('can_rename'))
|
|
perm.can_move = bool(request.form.get('can_move'))
|
|
perm.can_share = bool(request.form.get('can_share'))
|
|
db.session.commit()
|
|
|
|
log_event(
|
|
event_type='room_permission_update',
|
|
details={
|
|
'room_id': room_id,
|
|
'room_name': room.name,
|
|
'user': f"{user.username} {user.last_name}",
|
|
'updated_by': f"{current_user.username} {current_user.last_name}",
|
|
'permissions': {
|
|
'can_view': perm.can_view,
|
|
'can_download': perm.can_download,
|
|
'can_upload': perm.can_upload,
|
|
'can_delete': perm.can_delete,
|
|
'can_rename': perm.can_rename,
|
|
'can_move': perm.can_move,
|
|
'can_share': perm.can_share
|
|
}
|
|
},
|
|
user_id=current_user.id
|
|
)
|
|
db.session.commit()
|
|
|
|
flash('Permissions updated.', 'success')
|
|
return redirect(url_for('rooms.room_members', room_id=room_id))
|
|
|
|
@rooms_bp.route('/<int:room_id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def edit_room(room_id):
|
|
room = Room.query.get_or_404(room_id)
|
|
# Check if user is a member
|
|
is_member = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=current_user.id).first() is not None
|
|
if not (current_user.is_admin or (current_user.is_manager and is_member)):
|
|
flash('Only administrators and managers can edit rooms.', 'error')
|
|
return redirect(url_for('rooms.rooms'))
|
|
|
|
form = RoomForm()
|
|
|
|
if form.validate_on_submit():
|
|
old_name = room.name
|
|
old_description = room.description
|
|
|
|
room.name = form.name.data
|
|
room.description = form.description.data
|
|
db.session.commit()
|
|
|
|
log_event(
|
|
event_type='room_update',
|
|
details={
|
|
'room_id': room.id,
|
|
'old_name': old_name,
|
|
'new_name': room.name,
|
|
'old_description': old_description,
|
|
'new_description': room.description,
|
|
'updated_by': f"{current_user.username} {current_user.last_name}"
|
|
},
|
|
user_id=current_user.id
|
|
)
|
|
db.session.commit()
|
|
|
|
flash('Room updated successfully!', 'success')
|
|
return redirect(url_for('rooms.rooms'))
|
|
|
|
# Pre-populate form with existing room data
|
|
if request.method == 'GET':
|
|
form.name.data = room.name
|
|
form.description.data = room.description
|
|
|
|
return render_template('rooms/edit_room.html', form=form, room=room)
|
|
|
|
@rooms_bp.route('/<int:room_id>/delete', methods=['POST'])
|
|
@login_required
|
|
def delete_room(room_id):
|
|
room = Room.query.get_or_404(room_id)
|
|
# Check if user is a member
|
|
is_member = RoomMemberPermission.query.filter_by(room_id=room_id, user_id=current_user.id).first() is not None
|
|
if not (current_user.is_admin or (current_user.is_manager and is_member)):
|
|
flash('Only administrators and managers can delete rooms.', 'error')
|
|
return redirect(url_for('rooms.rooms'))
|
|
|
|
room_name = room.name
|
|
|
|
try:
|
|
print(f"Attempting to delete room {room_id} ({room_name})")
|
|
|
|
# Delete physical files
|
|
room_dir = os.path.join('/data/rooms', str(room_id))
|
|
if os.path.exists(room_dir):
|
|
shutil.rmtree(room_dir)
|
|
print(f"Deleted room directory: {room_dir}")
|
|
|
|
# Delete the room (cascade will handle the rest)
|
|
db.session.delete(room)
|
|
db.session.commit()
|
|
print("Room deleted successfully")
|
|
|
|
log_event(
|
|
event_type='room_delete',
|
|
details={
|
|
'room_id': room_id,
|
|
'room_name': room_name,
|
|
'deleted_by': f"{current_user.username} {current_user.last_name}"
|
|
},
|
|
user_id=current_user.id
|
|
)
|
|
db.session.commit()
|
|
|
|
flash(f'Room "{room_name}" has been deleted.', 'success')
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash('An error occurred while deleting the room. Please try again.', 'error')
|
|
print(f"Error deleting room: {str(e)}")
|
|
|
|
return redirect(url_for('rooms.rooms'))
|
|
|
|
@rooms_bp.route('/room/<int:room_id>/view/<path:file_path>')
|
|
@login_required
|
|
def view_file(room_id, file_path):
|
|
room = Room.query.get_or_404(room_id)
|
|
# Check if user has access to the room
|
|
if not current_user.is_admin and current_user not in room.members:
|
|
flash('You do not have access to this room.', 'error')
|
|
return redirect(url_for('rooms.rooms'))
|
|
|
|
file = RoomFile.query.filter_by(room_id=room_id, path=file_path).first_or_404()
|
|
|
|
# Continue with file viewing logic... |