import logging import os import uuid from functools import wraps from flask import Blueprint, render_template, redirect, url_for, flash, request, abort, jsonify, current_app, send_from_directory from flask_login import login_required, current_user from werkzeug.utils import secure_filename from app import db from app.models import (User, Ticket, Comment, ActivityLog, KnowledgeBase, KBAttachment, 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) # ── Helpers ────────────────────────────────────────────────────────────────── _KB_ALLOWED_EXT = { 'png', 'jpg', 'jpeg', 'gif', 'webp', # images 'pdf', 'doc', 'docx', 'xls', 'xlsx', # documents 'ppt', 'pptx', 'txt', 'csv', 'zip', 'log', # misc } def _kb_allowed(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in _KB_ALLOWED_EXT def _save_kb_file(file, article_id): """Save an uploaded file and return a KBAttachment (not yet committed).""" filename = secure_filename(file.filename) ext = filename.rsplit('.', 1)[1].lower() if '.' in filename else 'bin' stored_name = f"{uuid.uuid4().hex}.{ext}" upload_dir = current_app.config['UPLOAD_FOLDER'] os.makedirs(upload_dir, exist_ok=True) filepath = os.path.join(upload_dir, stored_name) file.save(filepath) return KBAttachment( article_id = article_id, filename = filename, stored_name = stored_name, mime_type = file.content_type, file_size = os.path.getsize(filepath), uploaded_by = current_user.id, ) # ── TinyMCE image upload endpoint ───────────────────────────────────────────── @admin_bp.route('/kb/upload-image', methods=['POST']) @login_required @it_required def kb_upload_image(): """ TinyMCE image upload handler. Returns JSON: {"location": ""} on success. """ f = request.files.get('file') if not f or not f.filename: return jsonify({'error': 'No file provided'}), 400 ext = f.filename.rsplit('.', 1)[-1].lower() if '.' in f.filename else '' if ext not in {'png', 'jpg', 'jpeg', 'gif', 'webp'}: return jsonify({'error': 'Only image files are accepted (PNG, JPG, GIF, WEBP)'}), 400 stored_name = f"{uuid.uuid4().hex}.{ext}" upload_dir = current_app.config['UPLOAD_FOLDER'] os.makedirs(upload_dir, exist_ok=True) f.save(os.path.join(upload_dir, stored_name)) logger.info(f'[KB IMAGE UPLOAD] stored_name={stored_name} by user_id={current_user.id}') location = url_for('admin.kb_serve_file', stored_name=stored_name) return jsonify({'location': location}) # ── File-serve route (images embedded in articles + attachment downloads) ───── @admin_bp.route('/kb/files/') @login_required def kb_serve_file(stored_name): """Serve a KB attachment file. Login required — no public access.""" upload_dir = current_app.config['UPLOAD_FOLDER'] return send_from_directory(upload_dir, stored_name) # ── Delete a single KB attachment ───────────────────────────────────────────── @admin_bp.route('/kb//attachments//delete', methods=['POST']) @login_required @it_required def kb_delete_attachment(article_id, att_id): att = KBAttachment.query.filter_by(id=att_id, article_id=article_id).first_or_404() upload_dir = current_app.config['UPLOAD_FOLDER'] filepath = os.path.join(upload_dir, att.stored_name) if os.path.exists(filepath): os.remove(filepath) log_action(current_user.id, 'kb_attachment_delete', 'kb_attachment', att.id, f'article_id={article_id} filename={att.filename}') logger.info(f'[KB ATTACHMENT DELETE] att_id={att.id} article_id={article_id} by user_id={current_user.id}') db.session.delete(att) db.session.commit() logger.info(f'[KB ATTACHMENT DELETE] att_id={att.id} article_id={article_id} completed') # Return JSON so the edit page can remove the row without a full reload return jsonify({'ok': True, 'att_id': att.id}) @admin_bp.route('/kb//attachments', methods=['GET']) @login_required @it_required def kb_get_attachments(article_id): """Return current attachment list as JSON for dynamic UI refresh.""" from app.models import KBAttachment atts = KBAttachment.query.filter_by(article_id=article_id).all() return jsonify({'attachments': [ { 'id': a.id, 'filename': a.filename, 'stored_name': a.stored_name, 'file_size': a.file_size, 'mime_type': a.mime_type or '', 'url': url_for('admin.kb_serve_file', stored_name=a.stored_name), 'delete_url': url_for('admin.kb_delete_attachment', article_id=article_id, att_id=a.id), } for a in atts ]}) # ── Create article ───────────────────────────────────────────────────────────── @admin_bp.route('/kb/new', methods=['GET', 'POST']) @login_required @it_required def kb_new(): if request.method == 'POST': try: 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')) and not bool(request.form.get('_save_as_draft')), ) db.session.add(article) db.session.flush() for f in request.files.getlist('attachments'): if f and f.filename and _kb_allowed(f.filename): att = _save_kb_file(f, article.id) db.session.add(att) 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 successfully.', 'success') return jsonify({'ok': True, 'article_id': article.id, 'redirect': url_for('admin.kb_list')}) except Exception as exc: db.session.rollback() logger.error(f'[KB CREATE ERROR] {exc}') return jsonify({'error': str(exc)}), 500 return render_template('admin/kb_edit.html', article=None) # ── Edit article ─────────────────────────────────────────────────────────────── @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': try: 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')) and not bool(request.form.get('_save_as_draft')) for f in request.files.getlist('attachments'): if f and f.filename and _kb_allowed(f.filename): att = _save_kb_file(f, article.id) db.session.add(att) 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 successfully.', 'success') return jsonify({'ok': True, 'article_id': article.id, 'redirect': url_for('admin.kb_list')}) except Exception as exc: db.session.rollback() logger.error(f'[KB EDIT ERROR] {exc}') return jsonify({'error': str(exc)}), 500 return render_template('admin/kb_edit.html', article=article) @admin_bp.route('/kb//publish', methods=['POST']) @login_required @it_required def kb_toggle_publish(article_id): """Quick publish/unpublish toggle — callable from the article list.""" article = KnowledgeBase.query.get_or_404(article_id) article.is_published = not article.is_published db.session.commit() state = 'published' if article.is_published else 'unpublished' log_action(current_user.id, f'kb_{state}', 'knowledge_base', article.id, f'title={article.title}') logger.info(f'[KB TOGGLE PUBLISH] article_id={article.id} is_published={article.is_published} by user_id={current_user.id}') flash(f'Article "{article.title}" has been {state}.', 'success') return redirect(url_for('admin.kb_list')) @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]