222 lines
7.4 KiB
Python
222 lines
7.4 KiB
Python
import hmac
|
|
import re
|
|
from functools import wraps
|
|
|
|
from flask import (
|
|
Blueprint, current_app, flash, redirect, render_template,
|
|
request, session, url_for,
|
|
)
|
|
from werkzeug.security import check_password_hash
|
|
|
|
from app import db, log_action, Section, Topic
|
|
|
|
admin_bp = Blueprint("admin", __name__, url_prefix="/admin")
|
|
|
|
MEDIA_TYPES = ("none", "image", "video", "embed")
|
|
|
|
|
|
# ------------------------------------------------------------------ auth
|
|
def login_required(view):
|
|
@wraps(view)
|
|
def wrapped(*args, **kwargs):
|
|
if not session.get("admin"):
|
|
return redirect(url_for("admin.login", next=request.path))
|
|
return view(*args, **kwargs)
|
|
return wrapped
|
|
|
|
|
|
@admin_bp.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if session.get("admin"):
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "")
|
|
password = request.form.get("password", "")
|
|
cfg = current_app.config
|
|
expected_user = cfg.get("ADMIN_USERNAME", "")
|
|
pw_hash = cfg.get("ADMIN_PASSWORD_HASH", "")
|
|
|
|
user_ok = hmac.compare_digest(username, expected_user)
|
|
pass_ok = bool(pw_hash) and check_password_hash(pw_hash, password)
|
|
if user_ok and pass_ok:
|
|
session.clear()
|
|
session["admin"] = username
|
|
dest = request.args.get("next", "")
|
|
# only allow local admin redirects
|
|
if not dest.startswith("/admin"):
|
|
dest = url_for("admin.dashboard")
|
|
return redirect(dest)
|
|
flash("Incorrect username or password.", "error")
|
|
|
|
return render_template("admin/login.html")
|
|
|
|
|
|
@admin_bp.route("/logout", methods=["POST"])
|
|
def logout():
|
|
session.clear()
|
|
flash("Signed out.", "ok")
|
|
return redirect(url_for("admin.login"))
|
|
|
|
|
|
# ------------------------------------------------------------------ helpers
|
|
def _slugify(value):
|
|
value = (value or "").strip().lower()
|
|
value = re.sub(r"[^\w\s-]", "", value)
|
|
value = re.sub(r"[\s_]+", "-", value).strip("-")
|
|
return value or "topic"
|
|
|
|
|
|
def _unique_slug(base, exclude_id=None):
|
|
slug = base
|
|
n = 2
|
|
while True:
|
|
q = Topic.query.filter_by(slug=slug)
|
|
if exclude_id is not None:
|
|
q = q.filter(Topic.id != exclude_id)
|
|
if not q.first():
|
|
return slug
|
|
slug = f"{base}-{n}"
|
|
n += 1
|
|
|
|
|
|
def _int(value, default=0):
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
# ------------------------------------------------------------------ dashboard
|
|
@admin_bp.route("/")
|
|
@login_required
|
|
def dashboard():
|
|
sections = Section.query.order_by(Section.sort_order, Section.num).all()
|
|
return render_template("admin/dashboard.html", sections=sections)
|
|
|
|
|
|
# ------------------------------------------------------------------ topics
|
|
@admin_bp.route("/topic/new", methods=["GET", "POST"])
|
|
@admin_bp.route("/topic/<int:topic_id>", methods=["GET", "POST"])
|
|
@login_required
|
|
def topic_form(topic_id=None):
|
|
topic = Topic.query.get_or_404(topic_id) if topic_id else None
|
|
sections = Section.query.order_by(Section.sort_order, Section.num).all()
|
|
|
|
if request.method == "POST":
|
|
f = request.form
|
|
title = f.get("title", "").strip()
|
|
section_id = _int(f.get("section_id"))
|
|
if not title or not section_id:
|
|
flash("Title and section are required.", "error")
|
|
return render_template(
|
|
"admin/topic_form.html", topic=topic, sections=sections,
|
|
media_types=MEDIA_TYPES, form=f,
|
|
)
|
|
|
|
media_type = f.get("media_type", "none")
|
|
if media_type not in MEDIA_TYPES:
|
|
media_type = "none"
|
|
|
|
slug_input = f.get("slug", "").strip()
|
|
base_slug = _slugify(slug_input or title)
|
|
slug = _unique_slug(base_slug, exclude_id=topic.id if topic else None)
|
|
|
|
is_new = topic is None
|
|
if is_new:
|
|
topic = Topic()
|
|
|
|
topic.section_id = section_id
|
|
topic.slug = slug
|
|
topic.title = title
|
|
topic.body_html = f.get("body_html", "").strip() or None
|
|
topic.link_url = f.get("link_url", "").strip() or None
|
|
topic.link_label = f.get("link_label", "").strip() or None
|
|
topic.media_type = media_type
|
|
topic.media_url = f.get("media_url", "").strip() or None
|
|
topic.media_caption = f.get("media_caption", "").strip() or None
|
|
topic.sort_order = _int(f.get("sort_order"), 0)
|
|
|
|
if is_new:
|
|
db.session.add(topic)
|
|
db.session.commit()
|
|
log_action(
|
|
session.get("admin"), "create" if is_new else "update",
|
|
"topic", topic.id, topic.title,
|
|
)
|
|
flash(f"Topic '{topic.title}' saved.", "ok")
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
return render_template(
|
|
"admin/topic_form.html", topic=topic, sections=sections,
|
|
media_types=MEDIA_TYPES, form=None,
|
|
)
|
|
|
|
|
|
@admin_bp.route("/topic/<int:topic_id>/delete", methods=["POST"])
|
|
@login_required
|
|
def topic_delete(topic_id):
|
|
topic = Topic.query.get_or_404(topic_id)
|
|
title, tid = topic.title, topic.id
|
|
db.session.delete(topic)
|
|
db.session.commit()
|
|
log_action(session.get("admin"), "delete", "topic", tid, title)
|
|
flash(f"Topic '{title}' deleted.", "ok")
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
|
|
# ------------------------------------------------------------------ sections
|
|
@admin_bp.route("/section/new", methods=["GET", "POST"])
|
|
@admin_bp.route("/section/<int:section_id>", methods=["GET", "POST"])
|
|
@login_required
|
|
def section_form(section_id=None):
|
|
section = Section.query.get_or_404(section_id) if section_id else None
|
|
|
|
if request.method == "POST":
|
|
f = request.form
|
|
title = f.get("title", "").strip()
|
|
num = _int(f.get("num"))
|
|
if not title or not num:
|
|
flash("Number and title are required.", "error")
|
|
return render_template("admin/section_form.html", section=section, form=f)
|
|
|
|
# enforce unique num
|
|
clash = Section.query.filter(Section.num == num)
|
|
if section:
|
|
clash = clash.filter(Section.id != section.id)
|
|
if clash.first():
|
|
flash(f"Section number {num} is already in use.", "error")
|
|
return render_template("admin/section_form.html", section=section, form=f)
|
|
|
|
is_new = section is None
|
|
if is_new:
|
|
section = Section()
|
|
section.num = num
|
|
section.title = title
|
|
section.subtitle = f.get("subtitle", "").strip() or None
|
|
section.sort_order = _int(f.get("sort_order"), num * 10)
|
|
|
|
if is_new:
|
|
db.session.add(section)
|
|
db.session.commit()
|
|
log_action(
|
|
session.get("admin"), "create" if is_new else "update",
|
|
"section", section.id, section.title,
|
|
)
|
|
flash(f"Section '{section.title}' saved.", "ok")
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
return render_template("admin/section_form.html", section=section, form=None)
|
|
|
|
|
|
@admin_bp.route("/section/<int:section_id>/delete", methods=["POST"])
|
|
@login_required
|
|
def section_delete(section_id):
|
|
section = Section.query.get_or_404(section_id)
|
|
title, sid = section.title, section.id
|
|
db.session.delete(section) # cascades to its topics
|
|
db.session.commit()
|
|
log_action(session.get("admin"), "delete", "section", sid, title)
|
|
flash(f"Section '{title}' and its topics deleted.", "ok")
|
|
return redirect(url_for("admin.dashboard"))
|