319 lines
8.5 KiB
Python
319 lines
8.5 KiB
Python
from functools import wraps
|
|
import sqlite3
|
|
from datetime import date
|
|
|
|
from flask import (
|
|
Blueprint,
|
|
flash,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
session,
|
|
url_for,
|
|
)
|
|
from werkzeug.security import check_password_hash, generate_password_hash
|
|
|
|
from .database import get_db
|
|
|
|
bp = Blueprint("admin", __name__, url_prefix="/admin")
|
|
|
|
|
|
def admin_required(view):
|
|
@wraps(view)
|
|
def wrapped_view(**kwargs):
|
|
if not session.get("admin"):
|
|
return redirect(url_for("admin.login"))
|
|
|
|
return view(**kwargs)
|
|
|
|
return wrapped_view
|
|
|
|
|
|
def validate_question_form(form, question_id=None):
|
|
question = form.get("question", "").strip()
|
|
option_a = form.get("option_a", "").strip()
|
|
option_b = form.get("option_b", "").strip()
|
|
question_date = form.get("date", "").strip()
|
|
|
|
data = {
|
|
"question": question,
|
|
"option_a": option_a,
|
|
"option_b": option_b,
|
|
"date": question_date,
|
|
}
|
|
|
|
if not all(data.values()):
|
|
return data, "All fields are required."
|
|
|
|
db = get_db()
|
|
params = [question_date]
|
|
duplicate_sql = "SELECT id FROM questions WHERE date = ?"
|
|
|
|
if question_id is not None:
|
|
duplicate_sql += " AND id != ?"
|
|
params.append(question_id)
|
|
|
|
duplicate = db.execute(duplicate_sql, params).fetchone()
|
|
|
|
if duplicate is not None:
|
|
return data, "Date already has a question."
|
|
|
|
return data, None
|
|
|
|
|
|
def validate_account_form(form, admin):
|
|
username = form.get("username", "").strip()
|
|
current_password = form.get("current_password", "")
|
|
new_password = form.get("new_password", "")
|
|
confirm_password = form.get("confirm_password", "")
|
|
|
|
data = {"username": username}
|
|
|
|
if not username:
|
|
return data, "Username is required."
|
|
|
|
if not check_password_hash(admin["password_hash"], current_password):
|
|
return data, "Current password is incorrect."
|
|
|
|
if new_password or confirm_password:
|
|
if len(new_password) < 8:
|
|
return data, "New password must be at least 8 characters."
|
|
|
|
if new_password != confirm_password:
|
|
return data, "New passwords do not match."
|
|
|
|
return data, None
|
|
|
|
|
|
def question_content_changed(question, data):
|
|
return any(
|
|
question[field] != data[field]
|
|
for field in ("question", "option_a", "option_b", "date")
|
|
)
|
|
|
|
|
|
@bp.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
|
|
db = get_db()
|
|
admin = db.execute(
|
|
"SELECT id, username, password_hash FROM admins WHERE username = ?",
|
|
(username,),
|
|
).fetchone()
|
|
|
|
if admin is not None and check_password_hash(admin["password_hash"], password):
|
|
session.clear()
|
|
session["admin"] = True
|
|
session["admin_username"] = admin["username"]
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
flash("Invalid username or password.")
|
|
|
|
return render_template("admin/login.html")
|
|
|
|
|
|
@bp.route("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("admin.login"))
|
|
|
|
|
|
@bp.route("/dashboard")
|
|
@admin_required
|
|
def dashboard():
|
|
db = get_db()
|
|
questions = db.execute(
|
|
"""
|
|
SELECT id, question, option_a, option_b, date
|
|
FROM questions
|
|
ORDER BY date DESC, id DESC
|
|
"""
|
|
).fetchall()
|
|
|
|
return render_template("admin/dashboard.html", questions=questions)
|
|
|
|
|
|
@bp.route("/account", methods=["GET", "POST"])
|
|
@admin_required
|
|
def account():
|
|
db = get_db()
|
|
admin = db.execute(
|
|
"""
|
|
SELECT id, username, password_hash
|
|
FROM admins
|
|
WHERE username = ?
|
|
""",
|
|
(session.get("admin_username"),),
|
|
).fetchone()
|
|
|
|
if admin is None:
|
|
session.clear()
|
|
return redirect(url_for("admin.login"))
|
|
|
|
error = None
|
|
form_data = {"username": admin["username"]}
|
|
|
|
if request.method == "POST":
|
|
form_data, error = validate_account_form(request.form, admin)
|
|
|
|
if error is None:
|
|
password_hash = admin["password_hash"]
|
|
new_password = request.form.get("new_password", "")
|
|
|
|
if new_password:
|
|
password_hash = generate_password_hash(new_password)
|
|
|
|
try:
|
|
db.execute(
|
|
"""
|
|
UPDATE admins
|
|
SET username = ?, password_hash = ?
|
|
WHERE id = ?
|
|
""",
|
|
(form_data["username"], password_hash, admin["id"]),
|
|
)
|
|
db.commit()
|
|
except sqlite3.IntegrityError:
|
|
error = "Username is already taken."
|
|
else:
|
|
session["admin_username"] = form_data["username"]
|
|
flash("Admin credentials updated.")
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
return render_template(
|
|
"admin/account.html",
|
|
admin=form_data,
|
|
error=error,
|
|
)
|
|
|
|
|
|
@bp.route("/question/new", methods=["GET", "POST"])
|
|
@admin_required
|
|
def new_question():
|
|
question = None
|
|
error = None
|
|
|
|
if request.method == "POST":
|
|
data, error = validate_question_form(request.form)
|
|
|
|
if error is None:
|
|
db = get_db()
|
|
|
|
try:
|
|
db.execute(
|
|
"""
|
|
INSERT INTO questions (question, option_a, option_b, date)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
(
|
|
data["question"],
|
|
data["option_a"],
|
|
data["option_b"],
|
|
data["date"],
|
|
),
|
|
)
|
|
db.commit()
|
|
return redirect(url_for("admin.dashboard"))
|
|
except sqlite3.IntegrityError:
|
|
error = "Date already has a question."
|
|
|
|
question = data
|
|
|
|
return render_template(
|
|
"admin/question_form.html",
|
|
question=question,
|
|
error=error,
|
|
title="Create Question",
|
|
)
|
|
|
|
|
|
@bp.route("/question/edit/<int:question_id>", methods=["GET", "POST"])
|
|
@admin_required
|
|
def edit_question(question_id):
|
|
db = get_db()
|
|
question = db.execute(
|
|
"""
|
|
SELECT id, question, option_a, option_b, date
|
|
FROM questions
|
|
WHERE id = ?
|
|
""",
|
|
(question_id,),
|
|
).fetchone()
|
|
|
|
if question is None:
|
|
flash("Question not found.")
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
error = None
|
|
|
|
if request.method == "POST":
|
|
data, error = validate_question_form(request.form, question_id=question_id)
|
|
|
|
if error is None:
|
|
try:
|
|
should_clear_votes = (
|
|
question["date"] == date.today().isoformat()
|
|
and question_content_changed(question, data)
|
|
)
|
|
|
|
db.execute(
|
|
"""
|
|
UPDATE questions
|
|
SET question = ?, option_a = ?, option_b = ?, date = ?
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
data["question"],
|
|
data["option_a"],
|
|
data["option_b"],
|
|
data["date"],
|
|
question_id,
|
|
),
|
|
)
|
|
|
|
if should_clear_votes:
|
|
db.execute("DELETE FROM votes WHERE question_id = ?", (question_id,))
|
|
|
|
db.commit()
|
|
return redirect(url_for("admin.dashboard"))
|
|
except sqlite3.IntegrityError:
|
|
error = "Date already has a question."
|
|
|
|
question = data
|
|
|
|
return render_template(
|
|
"admin/question_form.html",
|
|
question=question,
|
|
error=error,
|
|
title="Edit Question",
|
|
)
|
|
|
|
|
|
@bp.route("/question/delete/<int:question_id>", methods=["GET", "POST"])
|
|
@admin_required
|
|
def delete_question(question_id):
|
|
db = get_db()
|
|
question = db.execute(
|
|
"""
|
|
SELECT id, question, option_a, option_b, date
|
|
FROM questions
|
|
WHERE id = ?
|
|
""",
|
|
(question_id,),
|
|
).fetchone()
|
|
|
|
if question is None:
|
|
flash("Question not found.")
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
if request.method == "POST":
|
|
db.execute("DELETE FROM votes WHERE question_id = ?", (question_id,))
|
|
db.execute("DELETE FROM questions WHERE id = ?", (question_id,))
|
|
db.commit()
|
|
return redirect(url_for("admin.dashboard"))
|
|
|
|
return render_template("admin/delete_confirm.html", question=question)
|