Initial commit: файловое хранилище с веб-интерфейсом
This commit is contained in:
@@ -0,0 +1,274 @@
|
||||
import os
|
||||
import shutil
|
||||
import mimetypes
|
||||
from flask import Flask, render_template, request, redirect, url_for, send_from_directory, flash, abort, jsonify
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
|
||||
from werkzeug.security import generate_password_hash, check_password_hash
|
||||
from werkzeug.utils import secure_filename
|
||||
from datetime import datetime
|
||||
from config import Config
|
||||
from models import db, User, Log
|
||||
from utils import get_disk_usage, allowed_file, format_size
|
||||
from PIL import Image
|
||||
import io
|
||||
import base64
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config.from_object(Config)
|
||||
db.init_app(app)
|
||||
|
||||
login_manager = LoginManager()
|
||||
login_manager.init_app(app)
|
||||
login_manager.login_view = 'login'
|
||||
|
||||
@login_manager.user_loader
|
||||
def load_user(user_id):
|
||||
return User.query.get(int(user_id))
|
||||
|
||||
with app.app_context():
|
||||
db.create_all()
|
||||
if not User.query.filter_by(username=os.environ.get('ADMIN_USER', 'admin')).first():
|
||||
admin = User(
|
||||
username=os.environ.get('ADMIN_USER', 'admin'),
|
||||
password=generate_password_hash(os.environ.get('ADMIN_PASS', 'password'))
|
||||
)
|
||||
db.session.add(admin)
|
||||
db.session.commit()
|
||||
|
||||
# ---------- Вспомогательные функции ----------
|
||||
def safe_join(base, path):
|
||||
base = os.path.abspath(base)
|
||||
target = os.path.abspath(os.path.join(base, path.lstrip('/')))
|
||||
if not target.startswith(base):
|
||||
return None
|
||||
return target
|
||||
|
||||
def get_item_info(full_path, rel_path):
|
||||
if os.path.isfile(full_path):
|
||||
stat = os.stat(full_path)
|
||||
is_image = False
|
||||
thumb = None
|
||||
mime = mimetypes.guess_type(full_path)[0] or 'application/octet-stream'
|
||||
if mime and mime.startswith('image/'):
|
||||
is_image = True
|
||||
try:
|
||||
with Image.open(full_path) as img:
|
||||
img.thumbnail((150, 150))
|
||||
buffered = io.BytesIO()
|
||||
img.save(buffered, format="JPEG" if img.format != 'PNG' else 'PNG')
|
||||
thumb = base64.b64encode(buffered.getvalue()).decode('utf-8')
|
||||
except:
|
||||
pass
|
||||
return {
|
||||
'type': 'file',
|
||||
'name': os.path.basename(rel_path),
|
||||
'path': rel_path,
|
||||
'size': stat.st_size,
|
||||
'size_formatted': format_size(stat.st_size),
|
||||
'modified': datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
'is_image': is_image,
|
||||
'thumb': thumb,
|
||||
'mime': mime
|
||||
}
|
||||
elif os.path.isdir(full_path):
|
||||
return {
|
||||
'type': 'dir',
|
||||
'name': os.path.basename(rel_path),
|
||||
'path': rel_path,
|
||||
'modified': datetime.fromtimestamp(os.path.getmtime(full_path)).strftime('%Y-%m-%d %H:%M:%S')
|
||||
}
|
||||
return None
|
||||
|
||||
# ---------- Маршруты ----------
|
||||
@app.route('/login', methods=['GET', 'POST'])
|
||||
def login():
|
||||
if request.method == 'POST':
|
||||
username = request.form['username']
|
||||
password = request.form['password']
|
||||
user = User.query.filter_by(username=username).first()
|
||||
if user and check_password_hash(user.password, password):
|
||||
login_user(user)
|
||||
return redirect(url_for('index'))
|
||||
flash('Неверный логин или пароль', 'danger')
|
||||
return render_template('login.html')
|
||||
|
||||
@app.route('/logout')
|
||||
@login_required
|
||||
def logout():
|
||||
logout_user()
|
||||
return redirect(url_for('login'))
|
||||
|
||||
@app.route('/')
|
||||
@login_required
|
||||
def index():
|
||||
current_path = request.args.get('path', '')
|
||||
storage = app.config['STORAGE_FOLDER']
|
||||
full_path = safe_join(storage, current_path)
|
||||
if not full_path or not os.path.exists(full_path):
|
||||
return redirect(url_for('index', path=''))
|
||||
|
||||
items = []
|
||||
try:
|
||||
for name in os.listdir(full_path):
|
||||
item_full = os.path.join(full_path, name)
|
||||
rel = os.path.join(current_path, name).replace('\\', '/')
|
||||
info = get_item_info(item_full, rel)
|
||||
if info:
|
||||
items.append(info)
|
||||
items.sort(key=lambda x: (x['type'] != 'dir', x['name'].lower()))
|
||||
except Exception as e:
|
||||
flash(f'Ошибка чтения папки: {e}', 'danger')
|
||||
|
||||
breadcrumbs = []
|
||||
if current_path:
|
||||
parts = current_path.split('/')
|
||||
path_accum = ''
|
||||
for part in parts:
|
||||
path_accum = os.path.join(path_accum, part).replace('\\', '/')
|
||||
breadcrumbs.append({'name': part, 'path': path_accum})
|
||||
else:
|
||||
breadcrumbs.append({'name': 'Корень', 'path': ''})
|
||||
|
||||
disk = get_disk_usage(storage)
|
||||
return render_template('index.html',
|
||||
items=items,
|
||||
current_path=current_path,
|
||||
breadcrumbs=breadcrumbs,
|
||||
disk=disk,
|
||||
format_size=format_size,
|
||||
username=current_user.username)
|
||||
|
||||
@app.route('/upload', methods=['POST'])
|
||||
@login_required
|
||||
def upload():
|
||||
current_path = request.form.get('path', '')
|
||||
storage = app.config['STORAGE_FOLDER']
|
||||
target_dir = safe_join(storage, current_path)
|
||||
if not target_dir or not os.path.isdir(target_dir):
|
||||
return jsonify({'error': 'Неверный путь'}), 400
|
||||
|
||||
if 'file' not in request.files:
|
||||
return jsonify({'error': 'Файл не выбран'}), 400
|
||||
file = request.files['file']
|
||||
if file.filename == '':
|
||||
return jsonify({'error': 'Пустое имя'}), 400
|
||||
if not allowed_file(file.filename):
|
||||
return jsonify({'error': 'Тип файла не разрешён'}), 400
|
||||
|
||||
filename = secure_filename(file.filename)
|
||||
save_path = os.path.join(target_dir, filename)
|
||||
if os.path.exists(save_path):
|
||||
name, ext = os.path.splitext(filename)
|
||||
filename = f"{name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}{ext}"
|
||||
save_path = os.path.join(target_dir, filename)
|
||||
|
||||
file.save(save_path)
|
||||
size = os.path.getsize(save_path)
|
||||
log_entry = Log(username=current_user.username, action='upload', filename=os.path.join(current_path, filename), file_size=size)
|
||||
db.session.add(log_entry)
|
||||
db.session.commit()
|
||||
return jsonify({'success': True, 'filename': filename})
|
||||
|
||||
@app.route('/files/<path:filename>')
|
||||
@login_required
|
||||
def download_file(filename):
|
||||
storage = app.config['STORAGE_FOLDER']
|
||||
full_path = safe_join(storage, filename)
|
||||
if not full_path or not os.path.isfile(full_path):
|
||||
abort(404)
|
||||
size = os.path.getsize(full_path)
|
||||
log_entry = Log(username=current_user.username, action='download', filename=filename, file_size=size)
|
||||
db.session.add(log_entry)
|
||||
db.session.commit()
|
||||
if request.args.get('download'):
|
||||
return send_from_directory(os.path.dirname(full_path), os.path.basename(full_path), as_attachment=True)
|
||||
return send_from_directory(os.path.dirname(full_path), os.path.basename(full_path), as_attachment=False)
|
||||
|
||||
@app.route('/delete/<path:filename>', methods=['POST'])
|
||||
@login_required
|
||||
def delete_file(filename):
|
||||
storage = app.config['STORAGE_FOLDER']
|
||||
full_path = safe_join(storage, filename)
|
||||
if not full_path:
|
||||
flash('Неверный путь', 'danger')
|
||||
return redirect(request.referrer or url_for('index'))
|
||||
if os.path.isfile(full_path):
|
||||
os.remove(full_path)
|
||||
log_entry = Log(username=current_user.username, action='delete', filename=filename, file_size=0)
|
||||
db.session.add(log_entry)
|
||||
db.session.commit()
|
||||
flash(f'Файл "{os.path.basename(filename)}" удалён', 'info')
|
||||
elif os.path.isdir(full_path):
|
||||
shutil.rmtree(full_path)
|
||||
flash(f'Папка "{os.path.basename(filename)}" удалена', 'info')
|
||||
else:
|
||||
flash('Объект не найден', 'warning')
|
||||
return redirect(request.referrer or url_for('index'))
|
||||
|
||||
@app.route('/create_folder', methods=['POST'])
|
||||
@login_required
|
||||
def create_folder():
|
||||
current_path = request.form.get('path', '')
|
||||
folder_name = request.form.get('folder_name', '').strip()
|
||||
if not folder_name:
|
||||
return jsonify({'error': 'Имя папки не может быть пустым'}), 400
|
||||
folder_name = secure_filename(folder_name)
|
||||
if not folder_name:
|
||||
return jsonify({'error': 'Недопустимое имя папки'}), 400
|
||||
|
||||
storage = app.config['STORAGE_FOLDER']
|
||||
target_dir = safe_join(storage, current_path)
|
||||
if not target_dir or not os.path.isdir(target_dir):
|
||||
return jsonify({'error': 'Неверный путь'}), 400
|
||||
|
||||
new_folder = os.path.join(target_dir, folder_name)
|
||||
try:
|
||||
os.mkdir(new_folder)
|
||||
return jsonify({'success': True})
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@app.route('/move', methods=['POST'])
|
||||
@login_required
|
||||
def move_item():
|
||||
source = request.form.get('source_path')
|
||||
dest_dir = request.form.get('dest_path', '')
|
||||
if not source:
|
||||
return jsonify({'error': 'Не указан источник'}), 400
|
||||
|
||||
storage = app.config['STORAGE_FOLDER']
|
||||
source_full = safe_join(storage, source)
|
||||
dest_full = safe_join(storage, dest_dir)
|
||||
if not source_full or not dest_full:
|
||||
return jsonify({'error': 'Неверный путь'}), 400
|
||||
if not os.path.exists(source_full):
|
||||
return jsonify({'error': 'Источник не существует'}), 404
|
||||
if not os.path.isdir(dest_full):
|
||||
return jsonify({'error': 'Папка назначения не существует'}), 400
|
||||
if os.path.isdir(source_full) and dest_full.startswith(source_full):
|
||||
return jsonify({'error': 'Нельзя переместить папку в саму себя'}), 400
|
||||
|
||||
try:
|
||||
target = os.path.join(dest_full, os.path.basename(source_full))
|
||||
if os.path.exists(target):
|
||||
name, ext = os.path.splitext(os.path.basename(source))
|
||||
new_name = f"{name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}{ext}"
|
||||
target = os.path.join(dest_full, new_name)
|
||||
shutil.move(source_full, target)
|
||||
log_entry = Log(username=current_user.username, action='move',
|
||||
filename=f"{source} -> {os.path.join(dest_dir, os.path.basename(target))}", file_size=0)
|
||||
db.session.add(log_entry)
|
||||
db.session.commit()
|
||||
return jsonify({'success': True, 'new_path': os.path.join(dest_dir, os.path.basename(target)).replace('\\', '/')})
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@app.route('/logs')
|
||||
@login_required
|
||||
def view_logs():
|
||||
logs = Log.query.order_by(Log.timestamp.desc()).limit(100).all()
|
||||
return render_template('logs.html', logs=logs, format_size=format_size)
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(host=os.environ.get('FLASK_HOST', '0.0.0.0'), port=int(os.environ.get('FLASK_PORT', 5000)), debug=False)
|
||||
Reference in New Issue
Block a user