import logging from functools import wraps from flask import Blueprint, render_template, redirect, url_for, flash, request, abort from flask_login import login_required, current_user from app import db from app.models import (User, Ticket, Comment, ActivityLog, KnowledgeBase, UserRole, TicketStatus) from app.services.log_service import log_action admin_bp = Blueprint('admin', __name__, url_prefix='/admin') logger = logging.getLogger(__name__) def admin_required(f): @wraps(f) def decorated(*args, **kwargs): if not current_user.is_authenticated or not current_user.is_admin: abort(403) return f(*args, **kwargs) return decorated def it_required(f): @wraps(f) def decorated(*args, **kwargs): if not current_user.is_authenticated or not current_user.is_it_staff: abort(403) return f(*args, **kwargs) return decorated # ─── Admin Dashboard ────────────────────────────────────────────────────────── @admin_bp.route('/') @login_required @it_required def index(): stats = { 'total_tickets' : Ticket.query.count(), 'open' : Ticket.query.filter_by(status=TicketStatus.OPEN).count(), 'in_progress' : Ticket.query.filter_by(status=TicketStatus.IN_PROGRESS).count(), 'resolved' : Ticket.query.filter_by(status=TicketStatus.RESOLVED).count(), 'closed' : Ticket.query.filter_by(status=TicketStatus.CLOSED).count(), 'total_users' : User.query.filter_by(is_active=True).count(), 'employees' : User.query.filter_by(role=UserRole.EMPLOYEE, is_active=True).count(), 'it_staff' : User.query.filter( User.role.in_([UserRole.IT_STAFF, UserRole.ADMIN]), User.is_active == True).count(), } recent_logs = ActivityLog.query.order_by(ActivityLog.created_at.desc()).limit(20).all() return render_template('admin/index.html', stats=stats, recent_logs=recent_logs) # ─── User Management ───────────────────────────────────────────────────────── @admin_bp.route('/users') @login_required @admin_required def users(): all_users = User.query.order_by(User.created_at.desc()).all() return render_template('admin/users.html', users=all_users) @admin_bp.route('/users/new', methods=['GET', 'POST']) @login_required @admin_required def create_user(): if request.method == 'POST': email = request.form.get('email', '').strip().lower() username = request.form.get('username', '').strip() full_name = request.form.get('full_name', '').strip() department = request.form.get('department', '').strip() phone = request.form.get('phone', '').strip() role = request.form.get('role', UserRole.EMPLOYEE) password = request.form.get('password', '') confirm = request.form.get('confirm_password', '') is_active = bool(request.form.get('is_active')) # ── Validation ──────────────────────────────────────────────────────── if not email or not username or not full_name or not password: flash('Email, username, full name and password are all required.', 'danger') return render_template('admin/create_user.html', roles=_roles(), form=request.form) if User.query.filter_by(email=email).first(): flash('That email address is already registered.', 'danger') return render_template('admin/create_user.html', roles=_roles(), form=request.form) if User.query.filter_by(username=username).first(): flash('That username is already taken.', 'danger') return render_template('admin/create_user.html', roles=_roles(), form=request.form) if password != confirm: flash('Passwords do not match.', 'danger') return render_template('admin/create_user.html', roles=_roles(), form=request.form) if len(password) < 8: flash('Password must be at least 8 characters.', 'danger') return render_template('admin/create_user.html', roles=_roles(), form=request.form) # ── Create ──────────────────────────────────────────────────────────── user = User( email = email, username = username, full_name = full_name, department = department, phone = phone, role = role, is_active = is_active, ) user.set_password(password) db.session.add(user) db.session.commit() log_action(current_user.id, 'admin_user_create', 'user', user.id, f'email={email} role={role}') logger.info(f'[ADMIN USER CREATE] user_id={user.id} email={email} role={role} by admin_id={current_user.id}') flash(f'User {full_name} ({email}) created successfully.', 'success') return redirect(url_for('admin.users')) return render_template('admin/create_user.html', roles=_roles(), form={}) @admin_bp.route('/users//edit', methods=['GET', 'POST']) @login_required @admin_required def edit_user(user_id): user = User.query.get_or_404(user_id) if request.method == 'POST': old_role = user.role user.full_name = request.form.get('full_name', user.full_name).strip() user.department= request.form.get('department', user.department).strip() user.phone = request.form.get('phone', user.phone or '').strip() user.role = request.form.get('role', user.role) user.is_active = bool(request.form.get('is_active')) new_pw = request.form.get('new_password', '') if new_pw: user.set_password(new_pw) logger.info(f'[ADMIN PASSWORD RESET] target_user_id={user.id} by admin_id={current_user.id}') db.session.commit() log_action(current_user.id, 'admin_user_edit', 'user', user.id, f'role_change={old_role}->{user.role} active={user.is_active}') logger.info(f'[ADMIN USER EDIT] user_id={user.id} by admin_id={current_user.id}') flash('User updated.', 'success') return redirect(url_for('admin.users')) return render_template('admin/edit_user.html', user=user, roles=_roles()) @admin_bp.route('/users//delete', methods=['POST']) @login_required @admin_required def delete_user(user_id): user = User.query.get_or_404(user_id) if user.id == current_user.id: flash('You cannot delete your own account.', 'danger') return redirect(url_for('admin.users')) user.is_active = False db.session.commit() log_action(current_user.id, 'admin_user_deactivate', 'user', user.id) logger.info(f'[ADMIN USER DEACTIVATE] user_id={user.id} by admin_id={current_user.id}') flash('User deactivated.', 'success') return redirect(url_for('admin.users')) # ─── Ticket Management (IT) ─────────────────────────────────────────────────── @admin_bp.route('/tickets') @login_required @it_required def all_tickets(): page = request.args.get('page', 1, type=int) status = request.args.get('status', '') priority = request.args.get('priority', '') assigned = request.args.get('assigned', '') q = Ticket.query if status: q = q.filter_by(status=status) if priority: q = q.filter_by(priority=priority) if assigned == 'me': q = q.filter_by(assigned_to_id=current_user.id) elif assigned == 'unassigned': q = q.filter_by(assigned_to_id=None) tickets = q.order_by(Ticket.created_at.desc()).paginate(page=page, per_page=25) return render_template('admin/tickets.html', tickets=tickets, status=status, priority=priority, assigned=assigned) # ─── Knowledge Base Management ──────────────────────────────────────────────── @admin_bp.route('/kb') @login_required @it_required def kb_list(): articles = KnowledgeBase.query.order_by(KnowledgeBase.created_at.desc()).all() return render_template('admin/kb_list.html', articles=articles) @admin_bp.route('/kb/new', methods=['GET', 'POST']) @login_required @it_required def kb_new(): if request.method == 'POST': article = KnowledgeBase( title = request.form.get('title', '').strip(), body = request.form.get('body', '').strip(), category = request.form.get('category', ''), tags = request.form.get('tags', ''), author_id = current_user.id, is_published= bool(request.form.get('is_published')), ) db.session.add(article) db.session.commit() log_action(current_user.id, 'kb_create', 'knowledge_base', article.id, f'title={article.title}') logger.info(f'[KB CREATE] article_id={article.id} by user_id={current_user.id}') flash('Article created.', 'success') return redirect(url_for('admin.kb_list')) return render_template('admin/kb_edit.html', article=None) @admin_bp.route('/kb//edit', methods=['GET', 'POST']) @login_required @it_required def kb_edit(article_id): article = KnowledgeBase.query.get_or_404(article_id) if request.method == 'POST': article.title = request.form.get('title', article.title).strip() article.body = request.form.get('body', article.body).strip() article.category = request.form.get('category', article.category) article.tags = request.form.get('tags', article.tags) article.is_published= bool(request.form.get('is_published')) db.session.commit() log_action(current_user.id, 'kb_edit', 'knowledge_base', article.id) logger.info(f'[KB EDIT] article_id={article.id} by user_id={current_user.id}') flash('Article updated.', 'success') return redirect(url_for('admin.kb_list')) return render_template('admin/kb_edit.html', article=article) @admin_bp.route('/kb//delete', methods=['POST']) @login_required @it_required def kb_delete(article_id): article = KnowledgeBase.query.get_or_404(article_id) log_action(current_user.id, 'kb_delete', 'knowledge_base', article.id, f'title={article.title}') logger.info(f'[KB DELETE] article_id={article.id} by user_id={current_user.id}') db.session.delete(article) db.session.commit() flash('Article deleted.', 'success') return redirect(url_for('admin.kb_list')) # ─── Activity Log ───────────────────────────────────────────────────────────── @admin_bp.route('/logs') @login_required @admin_required def activity_logs(): page = request.args.get('page', 1, type=int) logs = ActivityLog.query.order_by(ActivityLog.created_at.desc()).paginate( page=page, per_page=50) return render_template('admin/activity_logs.html', logs=logs) def _roles(): return [UserRole.EMPLOYEE, UserRole.IT_STAFF, UserRole.ADMIN]