bramval/Fossign
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

edited230126

Browse Source
This commit is contained in:
bramval
2026-01-23 18:07:29 +01:00
parent 32312fe4f2
commit 138136e835
18 changed files with 1354 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

288
app/routes/company.py
View File

@@ -1,7 +1,8 @@
import os
import uuid
from urllib.parse import urlparse, parse_qs
from flask import Blueprint, abort, current_app, flash, redirect, render_template, request, url_for
from flask import Blueprint, abort, current_app, flash, jsonify, redirect, render_template, request, url_for
from flask_login import current_user, login_required
from werkzeug.utils import secure_filename
@@ -11,6 +12,69 @@ from ..extensions import db
from ..models import Display, Playlist, PlaylistItem
ALLOWED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".tiff"}
ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".webm", ".ogg", ".mov", ".m4v"}
# Videos should have a maximum upload size of 250MB
MAX_VIDEO_BYTES = 250 * 1024 * 1024
def _normalize_youtube_embed_url(raw: str) -> str | None:
"""Normalize a user-provided YouTube URL into a privacy-friendly embed base URL.
Returns:
https://www.youtube-nocookie.com/embed/<VIDEO_ID>
or None if we cannot parse a valid video id.
"""
val = (raw or "").strip()
if not val:
return None
# Be forgiving for inputs like "youtu.be/<id>".
if not val.startswith("http://") and not val.startswith("https://"):
val = "https://" + val
try:
u = urlparse(val)
except Exception:
return None
host = (u.netloc or "").lower()
host = host[4:] if host.startswith("www.") else host
video_id: str | None = None
path = (u.path or "").strip("/")
if host in {"youtube.com", "m.youtube.com"}:
# /watch?v=<id>
if path == "watch":
v = (parse_qs(u.query).get("v") or [None])[0]
video_id = v
# /embed/<id>
elif path.startswith("embed/"):
video_id = path.split("/", 1)[1]
# /shorts/<id>
elif path.startswith("shorts/"):
video_id = path.split("/", 1)[1]
elif host == "youtu.be":
# /<id>
if path:
video_id = path.split("/", 1)[0]
# Basic validation: YouTube IDs are typically 11 chars (letters/digits/_/-)
if not video_id:
return None
video_id = video_id.strip()
if len(video_id) != 11:
return None
for ch in video_id:
if not (ch.isalnum() or ch in {"_", "-"}):
return None
return f"https://www.youtube-nocookie.com/embed/{video_id}"
def _save_compressed_image(uploaded_file, upload_folder: str) -> str:
"""Save an uploaded image as a compressed WEBP file.
@@ -129,7 +193,10 @@ def reorder_playlist_items(playlist_id: int):
if not playlist or playlist.company_id != current_user.company_id:
abort(404)
# Accept both form and JSON payloads.
order = (request.form.get("order") or "").strip()
if not order and request.is_json:
order = ((request.get_json(silent=True) or {}).get("order") or "").strip()
if not order:
abort(400)
@@ -150,6 +217,16 @@ def reorder_playlist_items(playlist_id: int):
id_to_item[item_id].position = pos
db.session.commit()
# Client currently doesn't require JSON, but returning JSON if requested
# helps debugging and future enhancements.
wants_json = (
(request.headers.get("X-Requested-With") == "XMLHttpRequest")
or ("application/json" in (request.headers.get("Accept") or ""))
or request.is_json
)
if wants_json:
return jsonify({"ok": True})
return ("", 204)
@@ -161,9 +238,25 @@ def add_playlist_item(playlist_id: int):
if not playlist or playlist.company_id != current_user.company_id:
abort(404)
item_type = request.form.get("item_type")
# Support AJAX/modal usage: return JSON when requested.
wants_json = (
(request.headers.get("X-Requested-With") == "XMLHttpRequest")
or ("application/json" in (request.headers.get("Accept") or ""))
or (request.form.get("response") == "json")
)
def _json_error(message: str, status: int = 400):
return jsonify({"ok": False, "error": message}), status
item_type = (request.form.get("item_type") or "").strip().lower()
title = request.form.get("title", "").strip() or None
duration = int(request.form.get("duration_seconds") or 10)
# Duration is only used for image/webpage. Video/YouTube plays until ended.
raw_duration = request.form.get("duration_seconds")
try:
duration = int(raw_duration) if raw_duration is not None else 10
except (TypeError, ValueError):
duration = 10
max_pos = (
db.session.query(db.func.max(PlaylistItem.position)).filter_by(playlist_id=playlist_id).scalar() or 0
@@ -181,37 +274,137 @@ def add_playlist_item(playlist_id: int):
if item_type in ("image", "video"):
f = request.files.get("file")
if not f or not f.filename:
if wants_json:
return _json_error("File required")
flash("File required", "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
filename = secure_filename(f.filename)
ext = os.path.splitext(filename)[1].lower()
if item_type == "image" and ext in (".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".tiff"):
if item_type == "image":
if ext not in ALLOWED_IMAGE_EXTENSIONS:
if wants_json:
return _json_error(
"Unsupported image type. Please upload one of: " + ", ".join(sorted(ALLOWED_IMAGE_EXTENSIONS))
)
flash("Unsupported image type", "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
try:
item.file_path = _save_compressed_image(f, current_app.config["UPLOAD_FOLDER"])
except Exception:
if wants_json:
return _json_error("Failed to process image upload", 500)
flash("Failed to process image upload", "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
else:
# Videos (and unknown image types): keep as-is but always rename to a UUID
if ext not in ALLOWED_VIDEO_EXTENSIONS:
if wants_json:
return _json_error(
"Unsupported video type. Please upload one of: " + ", ".join(sorted(ALLOWED_VIDEO_EXTENSIONS))
)
flash("Unsupported video type", "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
# Enforce video size limit (250MB) with a clear error message.
# This is separate from Flask's MAX_CONTENT_LENGTH, which caps the full request.
size = None
try:
size = getattr(f, "content_length", None)
# Werkzeug may report 0 for unknown per-part length.
if (size is None or size <= 0) and hasattr(f, "stream"):
# Measure by seeking in the file-like stream.
pos = f.stream.tell()
f.stream.seek(0, os.SEEK_END)
size = f.stream.tell()
f.stream.seek(pos, os.SEEK_SET)
except Exception:
size = None
if size is not None and size > MAX_VIDEO_BYTES:
msg = "Video file too large. Maximum allowed size is 250MB."
if wants_json:
return _json_error(msg, 413)
flash(msg, "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
# Keep as-is but always rename to a UUID.
unique = uuid.uuid4().hex + ext
save_path = os.path.join(current_app.config["UPLOAD_FOLDER"], unique)
f.save(save_path)
# Safety check: validate using the actual saved file size.
# (Some clients/framework layers don't reliably report per-part size.)
try:
saved_size = os.path.getsize(save_path)
except OSError:
saved_size = None
if saved_size is not None and saved_size > MAX_VIDEO_BYTES:
try:
os.remove(save_path)
except OSError:
pass
msg = "Video file too large. Maximum allowed size is 250MB."
if wants_json:
return _json_error(msg, 413)
flash(msg, "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
item.file_path = f"uploads/{unique}"
elif item_type == "webpage":
url = request.form.get("url", "").strip()
if not url:
if wants_json:
return _json_error("URL required")
flash("URL required", "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
item.url = url
elif item_type == "youtube":
raw = request.form.get("url", "").strip()
if not raw:
if wants_json:
return _json_error("YouTube URL required")
flash("YouTube URL required", "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
embed_url = _normalize_youtube_embed_url(raw)
if not embed_url:
if wants_json:
return _json_error("Invalid YouTube URL")
flash("Invalid YouTube URL", "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
item.url = embed_url
else:
if wants_json:
return _json_error("Invalid item type")
flash("Invalid item type", "danger")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
db.session.add(item)
db.session.commit()
if wants_json:
return jsonify(
{
"ok": True,
"item": {
"id": item.id,
"playlist_id": item.playlist_id,
"position": item.position,
"item_type": item.item_type,
"title": item.title,
"file_path": item.file_path,
"url": item.url,
"duration_seconds": item.duration_seconds,
},
}
)
flash("Item added", "success")
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
@@ -234,6 +427,52 @@ def delete_item(item_id: int):
return redirect(url_for("company.playlist_detail", playlist_id=playlist_id))
@bp.post("/items/<int:item_id>/duration")
@login_required
def update_item_duration(item_id: int):
"""Update duration_seconds for a playlist item.
Used from the playlist overview (inline edit).
"""
company_user_required()
item = db.session.get(PlaylistItem, item_id)
if not item or item.playlist.company_id != current_user.company_id:
abort(404)
# Duration only applies to images/webpages; videos play until ended.
if item.item_type == "video":
return jsonify({"ok": False, "error": "Duration cannot be set for video items"}), 400
wants_json = (
(request.headers.get("X-Requested-With") == "XMLHttpRequest")
or ("application/json" in (request.headers.get("Accept") or ""))
or request.is_json
)
def _json_error(message: str, status: int = 400):
return jsonify({"ok": False, "error": message}), status
raw = request.form.get("duration_seconds")
if raw is None and request.is_json:
raw = (request.get_json(silent=True) or {}).get("duration_seconds")
try:
duration = int(raw)
except (TypeError, ValueError):
if wants_json:
return _json_error("Invalid duration")
abort(400)
item.duration_seconds = max(1, duration)
db.session.commit()
if wants_json:
return jsonify({"ok": True, "duration_seconds": item.duration_seconds})
return ("", 204)
@bp.post("/displays/<int:display_id>/assign")
@login_required
def assign_playlist(display_id: int):
@@ -252,3 +491,42 @@ def assign_playlist(display_id: int):
db.session.commit()
flash("Display assignment updated", "success")
return redirect(url_for("company.dashboard"))
@bp.post("/displays/<int:display_id>")
@login_required
def update_display(display_id: int):
"""Update display metadata (description + assigned playlist).
Company users should be able to set a short description per display and assign a playlist.
"""
company_user_required()
display = db.session.get(Display, display_id)
if not display or display.company_id != current_user.company_id:
abort(404)
# Description (short, optional)
desc = (request.form.get("description") or "").strip() or None
if desc is not None:
desc = desc[:200]
display.description = desc
# Playlist assignment
playlist_id = (request.form.get("playlist_id") or "").strip()
if not playlist_id:
display.assigned_playlist_id = None
else:
try:
playlist_id_int = int(playlist_id)
except ValueError:
abort(400)
playlist = db.session.get(Playlist, playlist_id_int)
if not playlist or playlist.company_id != current_user.company_id:
abort(400)
display.assigned_playlist_id = playlist.id
db.session.commit()
flash("Display updated", "success")
return redirect(url_for("company.dashboard"))