129 lines
4.9 KiB
Python
129 lines
4.9 KiB
Python
"""
|
|
routes/admin_websites.py — Website management CRUD routes.
|
|
"""
|
|
|
|
import logging
|
|
from flask import Blueprint, render_template, request, redirect, url_for, flash, session, jsonify
|
|
from models import (
|
|
get_all_websites, get_website_by_id, get_website_credentials,
|
|
get_website_assigned_users, create_website, update_website, delete_website,
|
|
get_all_users, invalidate_admin_stats_cache,
|
|
)
|
|
from utils.decorators import admin_required
|
|
|
|
logger = logging.getLogger("routes.admin_websites")
|
|
admin_websites_bp = Blueprint("admin_websites", __name__, url_prefix="/admin/websites")
|
|
|
|
|
|
def _parse_credentials(form) -> list:
|
|
"""Extract credential rows from a multivalue form submission."""
|
|
creds = []
|
|
labels = form.getlist("cred_label[]")
|
|
usernames = form.getlist("cred_username[]")
|
|
passwords = form.getlist("cred_password[]")
|
|
for label, username, password in zip(labels, usernames, passwords):
|
|
if username.strip():
|
|
creds.append({
|
|
"label": label.strip(),
|
|
"username": username.strip(),
|
|
"password": password,
|
|
})
|
|
return creds
|
|
|
|
|
|
@admin_websites_bp.route("/")
|
|
@admin_required
|
|
def websites_list():
|
|
websites = get_all_websites()
|
|
all_users = get_all_users()
|
|
return render_template("admin/websites.html", websites=websites, all_users=all_users)
|
|
|
|
|
|
@admin_websites_bp.route("/<int:website_id>/detail")
|
|
@admin_required
|
|
def detail(website_id):
|
|
"""JSON endpoint: returns website + credentials + assigned users for the edit modal."""
|
|
site = get_website_by_id(website_id)
|
|
creds = get_website_credentials(website_id)
|
|
assigned = get_website_assigned_users(website_id)
|
|
if not site:
|
|
return jsonify({"error": "Not found"}), 404
|
|
return jsonify({
|
|
"id": site["id"],
|
|
"name": site["name"],
|
|
"url": site["url"],
|
|
"check_type": site["check_type"],
|
|
"visibility": site["visibility"],
|
|
"note": site["note"] or "",
|
|
"credentials": [{"label": c["label"], "username": c["username"], "password": c["password"]} for c in creds],
|
|
"assigned_user_ids": [u["id"] for u in assigned],
|
|
})
|
|
|
|
|
|
@admin_websites_bp.route("/create", methods=["POST"])
|
|
@admin_required
|
|
def create():
|
|
admin_id = session["user"]["id"]
|
|
name = request.form.get("name", "").strip()
|
|
url = request.form.get("url", "").strip()
|
|
check_type = request.form.get("check_type", "daily")
|
|
visibility = request.form.get("visibility", "all")
|
|
note = request.form.get("note", "").strip()
|
|
creds = _parse_credentials(request.form)
|
|
assigned = [int(x) for x in request.form.getlist("assigned_user_ids[]") if x]
|
|
|
|
if not name or not url:
|
|
flash("Name and URL are required.", "danger")
|
|
return redirect(url_for("admin_websites.websites_list"))
|
|
|
|
try:
|
|
create_website(admin_id, name, url, check_type, note, creds, visibility, assigned)
|
|
invalidate_admin_stats_cache()
|
|
flash(f"Website '{name}' created successfully.", "success")
|
|
logger.info(f"Website '{name}' created by admin_id={admin_id}.")
|
|
except Exception as e:
|
|
logger.error(f"create_website error: {e}")
|
|
flash(f"Error creating website: {e}", "danger")
|
|
|
|
return redirect(url_for("admin_websites.websites_list"))
|
|
|
|
|
|
@admin_websites_bp.route("/<int:website_id>/edit", methods=["POST"])
|
|
@admin_required
|
|
def edit(website_id):
|
|
admin_id = session["user"]["id"]
|
|
name = request.form.get("name", "").strip()
|
|
url = request.form.get("url", "").strip()
|
|
check_type = request.form.get("check_type", "daily")
|
|
visibility = request.form.get("visibility", "all")
|
|
note = request.form.get("note", "").strip()
|
|
creds = _parse_credentials(request.form)
|
|
assigned = [int(x) for x in request.form.getlist("assigned_user_ids[]") if x]
|
|
|
|
try:
|
|
update_website(admin_id, website_id, name, url, check_type, note, creds, visibility, assigned)
|
|
invalidate_admin_stats_cache()
|
|
flash(f"Website '{name}' updated successfully.", "success")
|
|
logger.info(f"Website id={website_id} updated by admin_id={admin_id}.")
|
|
except Exception as e:
|
|
logger.error(f"update_website error: {e}")
|
|
flash(f"Error updating website: {e}", "danger")
|
|
|
|
return redirect(url_for("admin_websites.websites_list"))
|
|
|
|
|
|
@admin_websites_bp.route("/<int:website_id>/delete", methods=["POST"])
|
|
@admin_required
|
|
def delete(website_id):
|
|
admin_id = session["user"]["id"]
|
|
try:
|
|
delete_website(admin_id, website_id)
|
|
invalidate_admin_stats_cache()
|
|
flash("Website deleted (soft) successfully.", "success")
|
|
logger.info(f"Website id={website_id} soft-deleted by admin_id={admin_id}.")
|
|
except Exception as e:
|
|
logger.error(f"delete_website error: {e}")
|
|
flash(f"Error deleting website: {e}", "danger")
|
|
|
|
return redirect(url_for("admin_websites.websites_list"))
|