This commit is contained in:
pika 2025-03-24 20:34:42 +01:00
parent 34afc48816
commit 4e781ba819
8 changed files with 855 additions and 160 deletions

View file

@ -19,46 +19,43 @@ bp = Blueprint('files', __name__, url_prefix='/files')
@bp.route('/browser/<int:folder_id>')
@login_required
def browser(folder_id=None):
"""Display file browser interface"""
"""File browser page"""
# Get the current folder
current_folder = None
breadcrumbs = []
if folder_id:
current_folder = Folder.query.filter_by(id=folder_id, user_id=current_user.id).first_or_404()
# Generate breadcrumbs
breadcrumbs = []
current_folder = Folder.query.get_or_404(folder_id)
# Check if user has permission
if current_folder.user_id != current_user.id:
flash('You do not have permission to access this folder', 'error')
return redirect(url_for('files.browser'))
# Get breadcrumb navigation
breadcrumbs = []
if current_folder:
# Get parent folders for breadcrumb
parent = current_folder
while parent:
breadcrumbs.append(parent)
parent = parent.parent
breadcrumbs.reverse()
breadcrumbs.insert(0, parent)
if parent.parent_id:
parent = Folder.query.get(parent.parent_id)
else:
break
# For initial load - only get folders and files if it's not an AJAX request
if not request.headers.get('X-Requested-With') == 'XMLHttpRequest':
if current_folder:
folders = Folder.query.filter_by(parent_id=current_folder.id, user_id=current_user.id).all()
files = File.query.filter_by(folder_id=current_folder.id, user_id=current_user.id).all()
else:
folders = Folder.query.filter_by(parent_id=None, user_id=current_user.id).all()
files = File.query.filter_by(folder_id=None, user_id=current_user.id).all()
# Get the search query if provided
query = request.args.get('q', '')
if query:
# Filter folders and files by name containing the query
folders = [f for f in folders if query.lower() in f.name.lower()]
files = [f for f in files if query.lower() in f.name.lower()]
return render_template('files/browser.html',
current_folder=current_folder,
breadcrumbs=breadcrumbs,
folders=folders,
files=files)
# Get subfolders and files
if current_folder:
subfolders = Folder.query.filter_by(parent_id=current_folder.id, user_id=current_user.id).all()
files = File.query.filter_by(folder_id=current_folder.id, user_id=current_user.id).all()
else:
# For AJAX request, return just the folder contents
# Implement this if needed
pass
# Root level - show folders and files without parent folder
subfolders = Folder.query.filter_by(parent_id=None, user_id=current_user.id).all()
files = File.query.filter_by(folder_id=None, user_id=current_user.id).all()
# Always return a valid response
return render_template('files/browser.html',
current_folder=current_folder,
breadcrumbs=breadcrumbs,
subfolders=subfolders,
files=files)
@bp.route('/contents')
@login_required
@ -150,106 +147,14 @@ def folder_contents():
else:
return redirect(url_for('files.browser'))
@bp.route('/upload', methods=['GET', 'POST'])
@bp.route('/upload/<int:folder_id>', methods=['GET', 'POST'])
@bp.route('/upload')
@bp.route('/upload/<int:folder_id>')
@login_required
def upload(folder_id=None):
"""Page for uploading files"""
if request.method == 'POST':
# Handle file upload
if 'file' not in request.files:
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({'error': 'No file part'}), 400
flash('No file part', 'error')
return redirect(request.url)
file = request.files['file']
# Validate filename
if file.filename == '':
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({'error': 'No selected file'}), 400
flash('No selected file', 'error')
return redirect(request.url)
# Get the parent folder
parent_folder = None
if folder_id:
parent_folder = Folder.query.get_or_404(folder_id)
# Check if user has permission
if parent_folder.user_id != current_user.id:
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({'error': 'You do not have permission to upload to this folder'}), 403
flash('You do not have permission to upload to this folder', 'error')
return redirect(url_for('files.browser'))
# Process the file
if file and allowed_file(file.filename):
filename = secure_filename(file.filename)
# Generate a unique filename
file_uuid = str(uuid.uuid4())
_, file_extension = os.path.splitext(filename)
storage_name = f"{file_uuid}{file_extension}"
# Create storage path
upload_folder = current_app.config['UPLOAD_FOLDER']
user_folder = os.path.join(upload_folder, str(current_user.id))
os.makedirs(user_folder, exist_ok=True)
storage_path = os.path.join(user_folder, storage_name)
try:
# Save file to storage location
file.save(storage_path)
# Get file info
file_size = os.path.getsize(storage_path)
mime_type, _ = mimetypes.guess_type(filename)
if not mime_type:
mime_type = 'application/octet-stream'
# Create file record
db_file = File(
name=filename,
original_name=filename,
path=storage_path,
size=file_size,
type=mime_type,
user_id=current_user.id,
folder_id=parent_folder.id if parent_folder else None
)
db.session.add(db_file)
db.session.commit()
# Return success response with file info
return jsonify({
'success': True,
'file': {
'id': db_file.id,
'name': db_file.name,
'size': db_file.size,
'formatted_size': format_file_size(db_file.size),
'type': db_file.type,
'icon': db_file.icon_class
}
})
except Exception as e:
# Make sure to remove the file if there was an error
if os.path.exists(storage_path):
os.remove(storage_path)
current_app.logger.error(f"Upload error: {str(e)}")
return jsonify({'error': str(e)}), 500
else:
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({'error': 'File type not allowed'}), 400
flash('File type not allowed', 'error')
return redirect(request.url)
# GET request - show upload form
return render_template('files/upload.html', folder_id=folder_id)
"""Redirect to browser with upload modal visible"""
if folder_id:
return redirect(url_for('files.browser', folder_id=folder_id, show_upload=True))
return redirect(url_for('files.browser', show_upload=True))
@bp.route('/upload_folder', methods=['POST'])
@login_required
@ -414,33 +319,91 @@ def download(file_id):
@login_required
def create_folder():
"""Create a new folder"""
name = request.form.get('name')
parent_id = request.form.get('parent_id')
if not name:
flash('Folder name is required', 'danger')
return redirect(url_for('files.browser'))
# Create folder
folder = Folder(
name=name,
user_id=current_user.id,
parent_id=parent_id if parent_id else None
)
# Check if request is AJAX
is_ajax = request.headers.get('X-Requested-With') == 'XMLHttpRequest'
try:
# Get data from form or JSON depending on request type
if is_ajax and request.is_json:
data = request.get_json()
folder_name = data.get('folder_name')
parent_id = data.get('parent_id')
else:
folder_name = request.form.get('folder_name')
parent_id = request.form.get('parent_id')
if not folder_name:
if is_ajax:
return jsonify({'success': False, 'error': 'Folder name is required'}), 400
flash('Folder name is required', 'error')
return redirect(url_for('files.browser', folder_id=parent_id if parent_id else None))
# Validate folder name
if not is_valid_filename(folder_name):
if is_ajax:
return jsonify({'success': False, 'error': 'Invalid folder name'}), 400
flash('Invalid folder name. Please use only letters, numbers, spaces, and common punctuation.', 'error')
return redirect(url_for('files.browser', folder_id=parent_id if parent_id else None))
# Check if folder already exists
parent_folder = None
if parent_id:
parent_folder = Folder.query.get_or_404(parent_id)
# Check if user has permission
if parent_folder.user_id != current_user.id:
if is_ajax:
return jsonify({'success': False, 'error': 'Permission denied'}), 403
flash('You do not have permission to create a folder here', 'error')
return redirect(url_for('files.browser'))
# Check if folder with same name exists in parent
existing = Folder.query.filter_by(
name=folder_name,
parent_id=parent_id,
user_id=current_user.id
).first()
else:
# Check if folder with same name exists at root level
existing = Folder.query.filter_by(
name=folder_name,
parent_id=None,
user_id=current_user.id
).first()
if existing:
if is_ajax:
return jsonify({'success': False, 'error': f'A folder named "{folder_name}" already exists'}), 400
flash(f'A folder named "{folder_name}" already exists', 'error')
return redirect(url_for('files.browser', folder_id=parent_id if parent_id else None))
# Create folder
folder = Folder(
name=folder_name,
parent_id=parent_id if parent_id else None,
user_id=current_user.id
)
db.session.add(folder)
db.session.commit()
flash(f'Folder "{name}" created successfully', 'success')
if is_ajax:
return jsonify({
'success': True,
'folder': {
'id': folder.id,
'name': folder.name,
'created_at': folder.created_at.strftime('%Y-%m-%d %H:%M:%S')
}
})
flash(f'Folder "{folder_name}" created successfully', 'success')
return redirect(url_for('files.browser', folder_id=parent_id if parent_id else None))
except Exception as e:
db.session.rollback()
current_app.logger.error(f"Error creating folder: {str(e)}")
flash('Error creating folder', 'danger')
# Redirect to appropriate location
if parent_id:
return redirect(url_for('files.browser', folder_id=parent_id))
return redirect(url_for('files.browser'))
if is_ajax:
return jsonify({'success': False, 'error': f'Error creating folder: {str(e)}'}), 500
flash(f'Error creating folder: {str(e)}', 'error')
return redirect(url_for('files.browser', folder_id=parent_id if parent_id else None))
@bp.route('/rename', methods=['POST'])
@login_required
@ -742,5 +705,72 @@ def upload_xhr():
db.session.rollback()
return jsonify({'error': str(e)}), 500
@bp.route('/move', methods=['POST'])
@login_required
def move_item():
"""Move a file or folder to another folder"""
data = request.get_json()
item_id = data.get('item_id')
item_type = data.get('item_type')
target_folder_id = data.get('target_folder_id')
if not item_id or not item_type or not target_folder_id:
return jsonify({'success': False, 'error': 'Missing parameters'}), 400
# Get target folder
target_folder = Folder.query.get_or_404(target_folder_id)
# Check if user has permission
if target_folder.user_id != current_user.id:
return jsonify({'success': False, 'error': 'Permission denied'}), 403
# Move file or folder
if item_type == 'file':
file = File.query.get_or_404(item_id)
# Check if user has permission
if file.user_id != current_user.id:
return jsonify({'success': False, 'error': 'Permission denied'}), 403
# Check for circular reference
if file.id == target_folder.id:
return jsonify({'success': False, 'error': 'Cannot move a file to itself'}), 400
# Update file's folder_id
file.folder_id = target_folder.id
elif item_type == 'folder':
folder = Folder.query.get_or_404(item_id)
# Check if user has permission
if folder.user_id != current_user.id:
return jsonify({'success': False, 'error': 'Permission denied'}), 403
# Check for circular reference
if folder.id == target_folder.id or target_folder.id == folder.id:
return jsonify({'success': False, 'error': 'Cannot move a folder to itself'}), 400
# Check if target is a descendant of the folder
if is_descendant(target_folder, folder):
return jsonify({'success': False, 'error': 'Cannot move a folder to its descendant'}), 400
# Update folder's parent_id
folder.parent_id = target_folder.id
# Save changes
db.session.commit()
return jsonify({'success': True})
def is_descendant(folder, potential_ancestor):
"""Check if a folder is a descendant of another folder"""
current = folder
while current.parent_id is not None:
if current.parent_id == potential_ancestor.id:
return True
current = Folder.query.get(current.parent_id)
return False
# Import the helper functions from __init__.py
from app import get_file_icon, format_file_size