clipboard
This commit is contained in:
+228
-2
@@ -1,20 +1,52 @@
|
||||
import fcntl
|
||||
import os
|
||||
import uuid
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
from flask import (
|
||||
Flask,
|
||||
flash,
|
||||
jsonify,
|
||||
redirect,
|
||||
render_template,
|
||||
request,
|
||||
send_file,
|
||||
url_for,
|
||||
)
|
||||
from flask_login import current_user, login_required, login_user, logout_user
|
||||
from sqlalchemy import func
|
||||
|
||||
from auth import init_login_manager
|
||||
from config import load_secret_key, sensors_api_url, users_database_uri
|
||||
from config import (
|
||||
clipboard_dir,
|
||||
clipboard_max_file_bytes,
|
||||
clipboard_max_total_bytes,
|
||||
load_secret_key,
|
||||
sensors_api_url,
|
||||
users_database_uri,
|
||||
)
|
||||
from sensors_client import SensorsAPIError, SensorsClient
|
||||
from user_models import Note, NoteItem, Notification, User, user_db
|
||||
from user_models import (
|
||||
ClipboardFile,
|
||||
Note,
|
||||
NoteItem,
|
||||
Notification,
|
||||
User,
|
||||
user_db,
|
||||
)
|
||||
|
||||
|
||||
def format_bytes(value: int) -> str:
|
||||
size = float(value)
|
||||
for unit in ("B", "KB", "MB", "GB", "TB"):
|
||||
if size < 1024 or unit == "TB":
|
||||
if unit == "B":
|
||||
return f"{int(size)} B"
|
||||
return f"{size:.1f} {unit}"
|
||||
size /= 1024
|
||||
return f"{int(size)} B"
|
||||
|
||||
|
||||
def create_app():
|
||||
@@ -24,6 +56,14 @@ def create_app():
|
||||
app.config["SQLALCHEMY_DATABASE_URI"] = users_database_uri()
|
||||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||
app.config["SENSORS_API_URL"] = sensors_api_url()
|
||||
app.config["CLIPBOARD_DIR"] = str(clipboard_dir())
|
||||
app.config["CLIPBOARD_MAX_FILE_BYTES"] = clipboard_max_file_bytes()
|
||||
app.config["CLIPBOARD_MAX_TOTAL_BYTES"] = clipboard_max_total_bytes()
|
||||
app.config["MAX_CONTENT_LENGTH"] = (
|
||||
app.config["CLIPBOARD_MAX_FILE_BYTES"] + 1024 * 1024
|
||||
)
|
||||
Path(app.config["CLIPBOARD_DIR"]).mkdir(parents=True, exist_ok=True)
|
||||
app.add_template_filter(format_bytes, "filesize")
|
||||
|
||||
user_db.init_app(app)
|
||||
sensors = SensorsClient(app.config["SENSORS_API_URL"])
|
||||
@@ -111,6 +151,181 @@ def create_app():
|
||||
def api_docs():
|
||||
return render_template("docs.html", user=current_user)
|
||||
|
||||
# --- Clipboard ---
|
||||
|
||||
def _clipboard_path(name):
|
||||
return Path(app.config["CLIPBOARD_DIR"]) / name
|
||||
|
||||
@contextmanager
|
||||
def _clipboard_lock():
|
||||
lock_path = Path(app.config["CLIPBOARD_DIR"]) / ".clipboard.lock"
|
||||
with open(lock_path, "a+") as lock_file:
|
||||
fcntl.flock(lock_file, fcntl.LOCK_EX)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
fcntl.flock(lock_file, fcntl.LOCK_UN)
|
||||
|
||||
def _clipboard_file_or_404(file_id):
|
||||
return ClipboardFile.query.filter_by(
|
||||
id=file_id, user_id=current_user.id
|
||||
).first_or_404()
|
||||
|
||||
@app.route("/clipboard")
|
||||
@login_required
|
||||
def clipboard_page():
|
||||
files = (
|
||||
ClipboardFile.query.filter_by(user_id=current_user.id)
|
||||
.order_by(ClipboardFile.created_at.desc(), ClipboardFile.id.desc())
|
||||
.all()
|
||||
)
|
||||
total_used = (
|
||||
user_db.session.query(
|
||||
func.coalesce(func.sum(ClipboardFile.size), 0)
|
||||
)
|
||||
.filter(ClipboardFile.user_id == current_user.id)
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
max_total = app.config["CLIPBOARD_MAX_TOTAL_BYTES"]
|
||||
return render_template(
|
||||
"clipboard.html",
|
||||
user=current_user,
|
||||
files=files,
|
||||
total_used=total_used,
|
||||
max_total=max_total,
|
||||
remaining=max(0, max_total - total_used),
|
||||
max_file=app.config["CLIPBOARD_MAX_FILE_BYTES"],
|
||||
)
|
||||
|
||||
@app.errorhandler(413)
|
||||
def request_too_large(_error):
|
||||
flash(
|
||||
"Upload too large. Maximum file size is "
|
||||
f"{format_bytes(app.config['CLIPBOARD_MAX_FILE_BYTES'])}.",
|
||||
"error",
|
||||
)
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
@app.route("/clipboard/upload", methods=["POST"])
|
||||
@login_required
|
||||
def clipboard_upload():
|
||||
upload = request.files.get("file")
|
||||
if not upload or not upload.filename:
|
||||
flash("Choose a file to upload.", "error")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
original_name = (
|
||||
os.path.basename(upload.filename.replace("\\", "/"))
|
||||
.strip()
|
||||
.replace("\x00", "")
|
||||
)
|
||||
if not original_name:
|
||||
flash("The file name is invalid.", "error")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
original_name = original_name[:255]
|
||||
|
||||
directory = Path(app.config["CLIPBOARD_DIR"])
|
||||
stored_name = uuid.uuid4().hex
|
||||
temp_path = directory / f".{stored_name}.tmp"
|
||||
final_path = directory / stored_name
|
||||
|
||||
try:
|
||||
upload.save(str(temp_path))
|
||||
size = temp_path.stat().st_size
|
||||
except OSError:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
flash("Could not save the uploaded file.", "error")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
if size <= 0:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
flash("The file is empty.", "error")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
max_file = app.config["CLIPBOARD_MAX_FILE_BYTES"]
|
||||
if size > max_file:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
flash(
|
||||
f"File is too large. Maximum size is {format_bytes(max_file)}.",
|
||||
"error",
|
||||
)
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
with _clipboard_lock():
|
||||
total_used = (
|
||||
user_db.session.query(
|
||||
func.coalesce(func.sum(ClipboardFile.size), 0)
|
||||
)
|
||||
.filter(ClipboardFile.user_id == current_user.id)
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
max_total = app.config["CLIPBOARD_MAX_TOTAL_BYTES"]
|
||||
if total_used + size > max_total:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
flash(
|
||||
f"Clipboard storage is full "
|
||||
f"({format_bytes(max_total)} limit).",
|
||||
"error",
|
||||
)
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
try:
|
||||
os.replace(temp_path, final_path)
|
||||
user_db.session.add(
|
||||
ClipboardFile(
|
||||
user_id=current_user.id,
|
||||
original_name=original_name,
|
||||
stored_name=stored_name,
|
||||
size=size,
|
||||
content_type=(
|
||||
upload.mimetype or "application/octet-stream"
|
||||
)[:255],
|
||||
)
|
||||
)
|
||||
user_db.session.commit()
|
||||
except Exception:
|
||||
user_db.session.rollback()
|
||||
final_path.unlink(missing_ok=True)
|
||||
temp_path.unlink(missing_ok=True)
|
||||
flash("Could not save the uploaded file.", "error")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
flash(f"Uploaded {original_name}.", "success")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
@app.route("/clipboard/<int:file_id>/download")
|
||||
@login_required
|
||||
def clipboard_download(file_id):
|
||||
entry = _clipboard_file_or_404(file_id)
|
||||
path = _clipboard_path(entry.stored_name)
|
||||
if not path.is_file():
|
||||
flash("The file is missing from the server.", "error")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
return send_file(
|
||||
path,
|
||||
as_attachment=True,
|
||||
download_name=entry.original_name,
|
||||
mimetype=entry.content_type or "application/octet-stream",
|
||||
)
|
||||
|
||||
@app.route("/clipboard/<int:file_id>/delete", methods=["POST"])
|
||||
@login_required
|
||||
def clipboard_delete(file_id):
|
||||
entry = _clipboard_file_or_404(file_id)
|
||||
name = entry.original_name
|
||||
path = _clipboard_path(entry.stored_name)
|
||||
try:
|
||||
path.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
flash("Could not delete the file from disk.", "error")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
user_db.session.delete(entry)
|
||||
user_db.session.commit()
|
||||
flash(f"Deleted {name}.", "success")
|
||||
return redirect(url_for("clipboard_page"))
|
||||
|
||||
# --- Notifications (stored in main_app.db) ---
|
||||
|
||||
@app.route("/notifications")
|
||||
@@ -145,6 +360,17 @@ def create_app():
|
||||
)
|
||||
)
|
||||
|
||||
@app.route("/notifications/<int:notification_id>/delete", methods=["POST"])
|
||||
@login_required
|
||||
def notification_delete(notification_id):
|
||||
notification = Notification.query.filter_by(
|
||||
id=notification_id, user_id=current_user.id
|
||||
).first_or_404()
|
||||
user_db.session.delete(notification)
|
||||
user_db.session.commit()
|
||||
flash("Deleted notification.", "success")
|
||||
return redirect(url_for("notifications_page"))
|
||||
|
||||
# --- Notes (stored in main_app.db) ---
|
||||
|
||||
def _note_or_404(note_id):
|
||||
|
||||
Reference in New Issue
Block a user