234 lines
6.4 KiB
Python
234 lines
6.4 KiB
Python
import os
|
|
import random
|
|
|
|
from flask import Blueprint, flash, request, redirect, url_for
|
|
from flask import current_app, render_template, g
|
|
|
|
from werkzeug.utils import secure_filename
|
|
from werkzeug.exceptions import abort
|
|
from files.models import db, File, Category, FileType
|
|
|
|
from files.auth import admin_required
|
|
|
|
bp = Blueprint('files', __name__)
|
|
bp.add_url_rule('/uploads/<path:filename>', 'uploaded_file', build_only=True)
|
|
|
|
|
|
def _get_extension(filename):
|
|
return filename.rsplit('.', 1)[1].lower()
|
|
|
|
|
|
def _get_name(filename):
|
|
return filename.rsplit('.', 1)[0].lower()
|
|
|
|
|
|
def _get_path_in_upload(filename):
|
|
return os.path.join(current_app.config['UPLOAD_FOLDER'], filename)
|
|
|
|
|
|
def _get_unused_name(filename):
|
|
if os.path.exists(_get_path_in_upload(filename)):
|
|
filename = '{}{}.{}'.format(
|
|
_get_name(filename),
|
|
random.randint(0, 9),
|
|
_get_extension(filename))
|
|
return _get_unused_name(filename)
|
|
return filename
|
|
|
|
|
|
def _save_file(file, private, category, file_type):
|
|
if private is None:
|
|
is_private = 0
|
|
else:
|
|
is_private = 1
|
|
|
|
filename = secure_filename(_get_unused_name(file.filename))
|
|
|
|
if os.path.exists(_get_path_in_upload(filename)):
|
|
abort(500)
|
|
|
|
file.save(_get_path_in_upload(filename))
|
|
db_file = File(filename, is_private, category.id, file_type.id)
|
|
|
|
db.session.add(db_file)
|
|
db.session.commit()
|
|
|
|
return db_file
|
|
|
|
|
|
def _rename_file(file, new_name):
|
|
new_name = secure_filename(_get_unused_name(new_name))
|
|
|
|
os.rename(
|
|
_get_path_in_upload(file.filename),
|
|
_get_path_in_upload(new_name))
|
|
file.filename = new_name
|
|
|
|
db.session.commit()
|
|
|
|
|
|
def _delete_file(file):
|
|
os.remove(_get_path_in_upload(file.filename))
|
|
db.session.delete(file)
|
|
db.session.commit()
|
|
|
|
|
|
@bp.route('/', methods=['GET', 'POST'])
|
|
def index():
|
|
if g.user is None:
|
|
files = File.query.filter_by(private=0).all()
|
|
else:
|
|
files = File.query.all()
|
|
|
|
return render_template('files/index.html', files=files)
|
|
|
|
|
|
@bp.route('/upload', methods=['GET', 'POST'])
|
|
@admin_required
|
|
def upload_file():
|
|
if request.method == 'POST':
|
|
if 'file.upload.file' not in request.files:
|
|
flash('No file part')
|
|
return redirect(request.url)
|
|
|
|
file = request.files['file.upload.file']
|
|
|
|
if 'file.upload.private' not in request.form:
|
|
private = None
|
|
else:
|
|
private = request.form['file.upload.private']
|
|
|
|
if 'file.upload.category' not in request.form:
|
|
flash('No category selected')
|
|
return redirect(request.url)
|
|
|
|
category = Category.query.get(request.form['file.upload.category'])
|
|
if category is None:
|
|
flash('The category selected won\'t exists')
|
|
return redirect(request.url)
|
|
|
|
if 'file.upload.file_type' not in request.form:
|
|
flash('No file type selected')
|
|
return redirect(request.url)
|
|
|
|
file_type = FileType.query.get(request.form['file.upload.file_type'])
|
|
if file_type is None:
|
|
flash('The file type selected won\'t exists')
|
|
return redirect(request.url)
|
|
|
|
if file.filename == '':
|
|
flash('No seleted file')
|
|
return redirect(request.url)
|
|
|
|
if file:
|
|
_save_file(file, private, category, file_type)
|
|
return redirect(url_for('files.index'))
|
|
|
|
return render_template(
|
|
'files/upload.html',
|
|
categories=Category.query.all(),
|
|
file_types=FileType.query.all())
|
|
|
|
|
|
@bp.route('/preview/<int:file_id>')
|
|
def preview_file(file_id):
|
|
file = File.query.get(file_id)
|
|
categories = Category.query.all()
|
|
file_types = FileType.query.all()
|
|
|
|
if (file.private == 1 and g.user is not None) or (file.private == 0):
|
|
if file.type.name == 'Code':
|
|
try:
|
|
content = open(_get_path_in_upload(file.filename), 'r').read()
|
|
except UnicodeDecodeError:
|
|
flash('Error: file is binary, can\'t be displayed')
|
|
content = 'Error'
|
|
|
|
return render_template(
|
|
'files/preview/code.html',
|
|
file=file,
|
|
categories=categories,
|
|
file_types=file_types,
|
|
content=content)
|
|
|
|
if file.type.name == 'Image':
|
|
return render_template(
|
|
'files/preview/image.html',
|
|
file=file,
|
|
categories=categories,
|
|
file_types=file_types)
|
|
|
|
return render_template(
|
|
'files/preview/default.html',
|
|
file=file,
|
|
categories=categories,
|
|
file_types=file_types)
|
|
|
|
return abort(404)
|
|
|
|
|
|
@bp.route('/rename/<int:file_id>', methods=['POST'])
|
|
@admin_required
|
|
def rename_file(file_id):
|
|
file = File.query.get(file_id)
|
|
|
|
new_name = request.form['new_name'].lower()
|
|
extension = file.filename.rsplit('.', 1)[1].lower()
|
|
|
|
if "." in new_name and _get_extension(new_name):
|
|
new_name = new_name.rsplit('.', 1)[0] + '.' + extension
|
|
else:
|
|
new_name = new_name + '.' + extension
|
|
|
|
_rename_file(file, new_name)
|
|
|
|
return redirect(url_for('files.preview_file', file_id=file.id))
|
|
|
|
|
|
@bp.route('/recategorize/<int:file_id>', methods=['POST'])
|
|
@admin_required
|
|
def recategorize(file_id):
|
|
file = File.query.get(file_id)
|
|
|
|
if 'new_category' not in request.form:
|
|
flash('No category selected')
|
|
return redirect(request.url)
|
|
|
|
new_category = Category.query.get(request.form['new_category'])
|
|
if new_category is None:
|
|
flash('The category selected won\'t exists')
|
|
return redirect(request.url)
|
|
|
|
file.category_id = new_category.id
|
|
db.session.commit()
|
|
|
|
return redirect(url_for('files.preview_file', file_id=file.id))
|
|
|
|
|
|
@bp.route('/retype/<int:file_id>', methods=['POST'])
|
|
@admin_required
|
|
def retype(file_id):
|
|
file = File.query.get(file_id)
|
|
|
|
if 'new_type' not in request.form:
|
|
flash('No type selected')
|
|
return redirect(request.url)
|
|
|
|
new_type = FileType.query.get(request.form['new_type'])
|
|
if new_type is None:
|
|
flash('The file type selected won\'t exists')
|
|
return redirect(request.url)
|
|
|
|
file.file_type_id = new_type.id
|
|
db.session.commit()
|
|
|
|
return redirect(url_for('files.preview_file', file_id=file.id))
|
|
|
|
|
|
@bp.route('/delete/<int:file_id>', methods=['POST'])
|
|
@admin_required
|
|
def delete_file(file_id):
|
|
file = File.query.get(file_id)
|
|
_delete_file(file)
|
|
return redirect(url_for('index'))
|