122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
"""
|
|
routes/admin_users.py — User management CRUD routes.
|
|
"""
|
|
|
|
import logging
|
|
from flask import Blueprint, render_template, request, redirect, url_for, flash, session, jsonify
|
|
from models import (
|
|
get_all_users, create_user, update_user, delete_user,
|
|
get_user_by_id, create_password_reset_token, log_action,
|
|
invalidate_admin_stats_cache,
|
|
)
|
|
from utils.decorators import admin_required
|
|
from utils.email import send_email
|
|
|
|
logger = logging.getLogger("routes.admin_users")
|
|
admin_users_bp = Blueprint("admin_users", __name__, url_prefix="/admin/users")
|
|
|
|
|
|
@admin_users_bp.route("/")
|
|
@admin_required
|
|
def users_list():
|
|
users = get_all_users()
|
|
return render_template("admin/users.html", users=users)
|
|
|
|
|
|
@admin_users_bp.route("/create", methods=["POST"])
|
|
@admin_required
|
|
def create():
|
|
admin_id = session["user"]["id"]
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
role = request.form.get("role", "user")
|
|
full_name = request.form.get("full_name", "").strip()
|
|
email = request.form.get("email", "").strip() or None
|
|
|
|
if not username or not password:
|
|
flash("Username and password are required.", "danger")
|
|
return redirect(url_for("admin_users.users_list"))
|
|
|
|
try:
|
|
create_user(admin_id, username, password, role, full_name, email)
|
|
invalidate_admin_stats_cache()
|
|
flash(f"User '{username}' created successfully.", "success")
|
|
logger.info(f"User '{username}' created by admin_id={admin_id}.")
|
|
except Exception as e:
|
|
logger.error(f"create_user error: {e}")
|
|
flash(f"Error creating user: {e}", "danger")
|
|
|
|
return redirect(url_for("admin_users.users_list"))
|
|
|
|
|
|
@admin_users_bp.route("/<int:user_id>/edit", methods=["POST"])
|
|
@admin_required
|
|
def edit(user_id):
|
|
admin_id = session["user"]["id"]
|
|
username = request.form.get("username", "").strip()
|
|
role = request.form.get("role", "user")
|
|
full_name = request.form.get("full_name", "").strip()
|
|
email = request.form.get("email", "").strip() or None
|
|
is_active = int(request.form.get("is_active", 1))
|
|
password = request.form.get("password", "").strip() or None
|
|
|
|
try:
|
|
update_user(admin_id, user_id, username, role, full_name, is_active, password, email)
|
|
invalidate_admin_stats_cache()
|
|
flash(f"User '{username}' updated successfully.", "success")
|
|
logger.info(f"User id={user_id} updated by admin_id={admin_id}.")
|
|
except ValueError as e:
|
|
flash(str(e), "danger")
|
|
except Exception as e:
|
|
logger.error(f"update_user error: {e}")
|
|
flash(f"Error updating user: {e}", "danger")
|
|
|
|
return redirect(url_for("admin_users.users_list"))
|
|
|
|
|
|
@admin_users_bp.route("/<int:user_id>/delete", methods=["POST"])
|
|
@admin_required
|
|
def delete(user_id):
|
|
admin_id = session["user"]["id"]
|
|
try:
|
|
delete_user(admin_id, user_id)
|
|
invalidate_admin_stats_cache()
|
|
flash("User deleted successfully.", "success")
|
|
logger.info(f"User id={user_id} deleted by admin_id={admin_id}.")
|
|
except ValueError as e:
|
|
flash(str(e), "danger")
|
|
except Exception as e:
|
|
logger.error(f"delete_user error: {e}")
|
|
flash(f"Error deleting user: {e}", "danger")
|
|
|
|
return redirect(url_for("admin_users.users_list"))
|
|
|
|
|
|
@admin_users_bp.route("/<int:user_id>/send-reset", methods=["POST"])
|
|
@admin_required
|
|
def send_reset(user_id):
|
|
admin = session["user"]
|
|
user = get_user_by_id(user_id)
|
|
if not user:
|
|
return jsonify({"error": "User not found."}), 404
|
|
if not user.get("email"):
|
|
return jsonify({"error": f"User '{user['username']}' has no email address set."}), 400
|
|
try:
|
|
token = create_password_reset_token(user_id)
|
|
reset_url = url_for("auth.reset_password", token=token, _external=True)
|
|
body = (
|
|
f"Hi {user.get('full_name') or user['username']},\n\n"
|
|
f"An administrator has sent you a password reset link for your Bid Checker account.\n"
|
|
f"Click the link below to set a new password (valid for 1 hour):\n\n"
|
|
f"{reset_url}\n\n"
|
|
f"If you did not expect this, please contact your administrator.\n"
|
|
)
|
|
send_email(user["email"], "Bid Checker — Password Reset", body)
|
|
log_action(admin["id"], "SEND_RESET_LINK", "users", user_id,
|
|
f"Password reset link sent to '{user['username']}' ({user['email']}).")
|
|
logger.info(f"Reset email sent to user_id={user_id} by admin_id={admin['id']}.")
|
|
return jsonify({"ok": True, "to": user["email"]})
|
|
except Exception as e:
|
|
logger.error(f"send_reset error for user_id={user_id}: {e}")
|
|
return jsonify({"error": str(e)}), 500
|