Mateus Tavares
2b1668206d
- padronizando o nome de get_db_connection e session para get_db_session, para não confundir com session do Flask ou sessoes web
- corrigindo potenciais erros
-- has_permission nao consegue com lazy load carregar permission depois de load_user fechar a conexao, entao joinedLoad com Permission antes de fechar
-- db.rollback não existe caso db = get_db_session() apareça muito depois dentro do try, padronizando antes de try
--- comparar role por nivel (Role.SECRETARIO_GERAL) e nao por nome ("Secretario Geral")
- unificacao de get_otp_qr_code
- mudança de nowutc() para now(UTC) conforme novo padrão
128 lines
4.2 KiB
Python
128 lines
4.2 KiB
Python
from flask import Blueprint, render_template, flash, redirect, url_for, request, jsonify
|
|
from functions.database import Usuario, get_db_session
|
|
from functions.decorators import require_login
|
|
from flask_login import login_required, current_user
|
|
from sqlalchemy.orm import joinedload
|
|
import pyotp
|
|
from werkzeug.security import generate_password_hash
|
|
import secrets
|
|
from functools import wraps
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
|
|
|
|
def admin_required(f):
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if not current_user.is_admin:
|
|
flash('Acesso não autorizado.', 'danger')
|
|
return redirect(url_for('home.index'))
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
@admin_bp.route('/')
|
|
@login_required
|
|
@admin_required
|
|
def dashboard():
|
|
"""Dashboard principal da área administrativa com lista de usuários"""
|
|
db = get_db_session()
|
|
try:
|
|
now = datetime.now()
|
|
|
|
# Carregar estatísticas relevantes
|
|
total_users = db.query(Usuario).count()
|
|
active_users = db.query(Usuario).filter(Usuario.is_active == True).count()
|
|
inactive_users = total_users - active_users
|
|
|
|
# Carregar lista de usuários
|
|
users = db.query(Usuario).options(
|
|
joinedload(Usuario.roles),
|
|
joinedload(Usuario.militante)
|
|
).all()
|
|
|
|
return render_template(
|
|
'admin/dashboard.html',
|
|
total_users=total_users,
|
|
active_users=active_users,
|
|
inactive_users=inactive_users,
|
|
users=users,
|
|
now=now
|
|
)
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Erro ao buscar dados do dashboard: {str(e)}")
|
|
flash('Erro ao carregar dados. Por favor, tente novamente.', 'danger')
|
|
return render_template('admin/dashboard.html',
|
|
total_users=0,
|
|
active_users=0,
|
|
inactive_users=0,
|
|
users=[])
|
|
finally:
|
|
db.close()
|
|
|
|
@admin_bp.route('/users/<int:user_id>/reset-otp', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def reset_user_otp(user_id):
|
|
"""Reseta o OTP de um usuário"""
|
|
db = get_db_session()
|
|
try:
|
|
user = db.query(Usuario).get(user_id)
|
|
if not user:
|
|
flash('Usuário não encontrado.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
# Gerar novo segredo OTP
|
|
user.otp_secret = pyotp.random_base32()
|
|
db.commit()
|
|
|
|
flash(f'OTP resetado com sucesso para {user.email}.', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
finally:
|
|
db.close()
|
|
|
|
@admin_bp.route('/users/<int:user_id>/reset-password', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def reset_user_password(user_id):
|
|
"""Reseta a senha de um usuário"""
|
|
db = get_db_session()
|
|
try:
|
|
user = db.query(Usuario).get(user_id)
|
|
if not user:
|
|
flash('Usuário não encontrado.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
# Gerar nova senha aleatória
|
|
new_password = secrets.token_urlsafe(8)
|
|
user.password = generate_password_hash(new_password)
|
|
db.commit()
|
|
|
|
flash(f'Senha resetada com sucesso. Nova senha: {new_password}', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
finally:
|
|
db.close()
|
|
|
|
@admin_bp.route('/users/<int:user_id>/toggle-status', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def toggle_user_status(user_id):
|
|
"""Ativa/desativa um usuário"""
|
|
db = get_db_session()
|
|
try:
|
|
user = db.query(Usuario).get(user_id)
|
|
if not user:
|
|
flash('Usuário não encontrado.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
user.is_active = not user.is_active
|
|
db.commit()
|
|
|
|
status = 'ativado' if user.is_active else 'desativado'
|
|
flash(f'Usuário {status} com sucesso.', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
finally:
|
|
db.close() |