Kobe/docupulse
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Documentation in room files py

Browse Source
This commit is contained in:
Kobe
2025-05-30 12:57:25 +02:00
parent 986db28494
commit fee79c6ec7
2 changed files with 192 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
routes/room_files.py
View File

@@ -1,3 +1,24 @@
"""
Room Files Management Module
This module provides a Flask Blueprint for managing files within rooms in the DocuPulse application.
It handles file operations such as uploading, downloading, moving, deleting, and organizing files
within room contexts. The module implements permission-based access control and supports various
file operations including file organization, starring, and trash management.
Key Features:
- File upload and download with permission checks
- Folder creation and management
- File operations (move, rename, delete)
- File starring system
- Trash management with restore capability
- Bulk operations (zip download)
- File search functionality
The module uses a combination of database records (RoomFile model) and actual file system storage
to maintain file metadata and content.
"""
from flask import Blueprint, jsonify, request, abort, send_from_directory, send_file
from flask_login import login_required, current_user
import os
@@ -9,10 +30,13 @@ import io
import zipfile
from datetime import datetime
# Blueprint for room file operations
room_files_bp = Blueprint('room_files', __name__, url_prefix='/api/rooms')
# Root directory for storing room files
DATA_ROOT = '/data/rooms' # This should be a Docker volume
# Set of allowed file extensions for upload
ALLOWED_EXTENSIONS = {
# Documents
'pdf', 'docx', 'doc', 'txt', 'rtf', 'odt', 'md', 'csv',
@@ -37,18 +61,55 @@ ALLOWED_EXTENSIONS = {
}
def get_room_dir(room_id):
"""
Get the absolute path to a room's directory.
Args:
room_id (int): The ID of the room
Returns:
str: Absolute path to the room's directory
"""
return os.path.join(DATA_ROOT, str(room_id))
def user_has_permission(room, perm_name):
"""
Check if the current user has a specific permission in a room.
Args:
room (Room): The room object to check permissions for
perm_name (str): Name of the permission to check
Returns:
bool: True if user has permission, False otherwise
"""
if current_user.is_admin:
return True
perm = RoomMemberPermission.query.filter_by(room_id=room.id, user_id=current_user.id).first()
return getattr(perm, perm_name, False) if perm else False
def allowed_file(filename):
"""
Check if a file's extension is in the allowed extensions list.
Args:
filename (str): Name of the file to check
Returns:
bool: True if file extension is allowed, False otherwise
"""
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
def clean_path(path):
"""
Clean a path string by removing leading/trailing slashes.
Args:
path (str): Path string to clean
Returns:
str: Cleaned path string
"""
if not path:
return ''
return path.strip('/\\')
@@ -56,6 +117,15 @@ def clean_path(path):
@room_files_bp.route('/<int:room_id>/files', methods=['GET'])
@login_required
def list_room_files(room_id):
"""
List all files in a room's directory.
Args:
room_id (int): ID of the room to list files from
Returns:
JSON response containing list of files with their metadata
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_view'):
abort(403)
@@ -170,6 +240,16 @@ def upload_room_file(room_id):
@room_files_bp.route('/<int:room_id>/files/<filename>', methods=['GET'])
@login_required
def download_room_file(room_id, filename):
"""
Download a file from a room.
Args:
room_id (int): ID of the room containing the file
filename (str): Name of the file to download
Returns:
File download response or error message
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_download'):
abort(403)
@@ -187,6 +267,16 @@ def download_room_file(room_id, filename):
@room_files_bp.route('/<int:room_id>/files/<path:filename>', methods=['DELETE'])
@login_required
def delete_file(room_id, filename):
"""
Delete a file from a room (moves to trash).
Args:
room_id (int): ID of the room containing the file
filename (str): Name of the file to delete
Returns:
JSON response indicating success or error
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_delete'):
abort(403)
@@ -209,6 +299,15 @@ def delete_file(room_id, filename):
@room_files_bp.route('/<int:room_id>/folders', methods=['POST'])
@login_required
def create_room_folder(room_id):
"""
Create a new folder in a room.
Args:
room_id (int): ID of the room to create folder in
Returns:
JSON response indicating success or error
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_upload'):
abort(403)
@@ -271,6 +370,15 @@ def create_room_folder(room_id):
@room_files_bp.route('/<int:room_id>/rename', methods=['POST'])
@login_required
def rename_room_file(room_id):
"""
Rename a file or folder in a room.
Args:
room_id (int): ID of the room containing the file/folder
Returns:
JSON response indicating success or error
"""
room = Room.query.get_or_404(room_id)
# Allow rename if user can upload or delete
if not (user_has_permission(room, 'can_upload') or user_has_permission(room, 'can_delete')):
@@ -326,6 +434,15 @@ def rename_room_file(room_id):
@room_files_bp.route('/<int:room_id>/download-zip', methods=['POST'])
@login_required
def download_zip(room_id):
"""
Download multiple files as a zip archive.
Args:
room_id (int): ID of the room containing the files
Returns:
ZIP file download response
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_view'):
abort(403)
@@ -356,6 +473,15 @@ def download_zip(room_id):
@room_files_bp.route('/<int:room_id>/search', methods=['GET'])
@login_required
def search_room_files(room_id):
"""
Search for files in a room by name.
Args:
room_id (int): ID of the room to search in
Returns:
JSON response containing matching files
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_view'):
abort(403)
@@ -379,6 +505,15 @@ def search_room_files(room_id):
@room_files_bp.route('/<int:room_id>/move', methods=['POST'])
@login_required
def move_room_file(room_id):
"""
Move a file or folder to a different location within a room.
Args:
room_id (int): ID of the room containing the file/folder
Returns:
JSON response indicating success or error
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_move'):
abort(403)
@@ -424,6 +559,15 @@ def move_room_file(room_id):
@room_files_bp.route('/<int:room_id>/folders', methods=['GET'])
@login_required
def list_room_folders(room_id):
"""
List all folders in a room.
Args:
room_id (int): ID of the room to list folders from
Returns:
JSON response containing list of folder paths
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_view'):
abort(403)
@@ -446,6 +590,15 @@ def list_room_folders(room_id):
@room_files_bp.route('/<int:room_id>/star', methods=['POST'])
@login_required
def toggle_star(room_id):
"""
Toggle the starred status of a file.
Args:
room_id (int): ID of the room containing the file
Returns:
JSON response indicating success and new starred status
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_view'):
abort(403)
@@ -479,6 +632,15 @@ def toggle_star(room_id):
@room_files_bp.route('/<int:room_id>/starred', methods=['GET'])
@login_required
def get_starred_files(room_id):
"""
Get all starred files in a room.
Args:
room_id (int): ID of the room to get starred files from
Returns:
JSON response containing list of starred files
"""
room = Room.query.get_or_404(room_id)
if not user_has_permission(room, 'can_view'):
abort(403)
@@ -512,6 +674,12 @@ def get_starred_files(room_id):
@room_files_bp.route('/starred', methods=['GET'])
@login_required
def get_all_starred_files():
"""
Get all starred files across all accessible rooms.
Returns:
JSON response containing list of all starred files
"""
# Get all rooms the user has access to
if current_user.is_admin:
rooms = Room.query.all()
@@ -555,6 +723,12 @@ def get_all_starred_files():
@room_files_bp.route('/trash', methods=['GET'])
@login_required
def get_trash_files():
"""
Get all deleted files from accessible rooms.
Returns:
JSON response containing list of deleted files
"""
# Get all rooms the user has access to
if current_user.is_admin:
rooms = Room.query.all()
@@ -614,6 +788,15 @@ def get_trash_files():
@room_files_bp.route('/<int:room_id>/restore', methods=['POST'])
@login_required
def restore_file(room_id):
"""
Restore a deleted file from trash.
Args:
room_id (int): ID of the room containing the deleted file
Returns:
JSON response indicating success or error
"""
room = Room.query.get_or_404(room_id)
# Check for delete permission instead of view permission
if not user_has_permission(room, 'can_delete'):
@@ -644,6 +827,15 @@ def restore_file(room_id):
@room_files_bp.route('/<int:room_id>/delete-permanent', methods=['POST'])
@login_required
def delete_permanent(room_id):
"""
Permanently delete files from trash (admin only).
Args:
room_id (int): ID of the room containing the files to delete
Returns:
JSON response indicating success or error
"""
# Only allow admin users to permanently delete files
if not current_user.is_admin:
abort(403)