import logging import os import uuid import csv import io from datetime import datetime, timedelta from functools import wraps from flask import Blueprint, render_template, redirect, url_for, flash, request, abort, jsonify, current_app, send_from_directory, Response from flask_login import login_required, current_user from werkzeug.utils import secure_filename import bleach from app import db, limiter from app.models import (User, Ticket, Comment, ActivityLog, KnowledgeBase, KBAttachment, UserRole, TicketStatus, CannedResponse, TicketTemplate, TicketSatisfaction) from app.services.log_service import log_action, log_ticket_history from app.services.validation_service import validate_password, validate_file admin_bp = Blueprint('admin', __name__, url_prefix='/admin') logger = logging.getLogger(__name__) # ── KB body HTML sanitisation ───────────────────────────────────────────────── # TinyMCE produces rich HTML which must be sanitised server-side before # persistence to prevent stored XSS attacks. Only tags and attributes that # are safe to render are whitelisted; everything else is stripped. _KB_ALLOWED_TAGS = set(bleach.sanitizer.ALLOWED_TAGS) | { 'p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'pre', 'code', 'blockquote', 'hr', 'br', 'table', 'thead', 'tbody', 'tr', 'th', 'td', 'ul', 'ol', 'li', 'dl', 'dt', 'dd', 'img', 'figure', 'figcaption', 'div', 'span', 'section', 'strong', 'em', 'u', 's', 'sub', 'sup', } _KB_ALLOWED_ATTRS = { '*' : ['class', 'id', 'style'], 'a' : ['href', 'title', 'target', 'rel'], 'img': ['src', 'alt', 'width', 'height', 'title'], 'td' : ['colspan', 'rowspan'], 'th' : ['colspan', 'rowspan'], 'col': ['span'], } def _sanitize_kb_body(raw_html): """Strip disallowed tags/attributes from a TinyMCE-produced HTML body.""" cleaned = bleach.clean( raw_html or '', tags = _KB_ALLOWED_TAGS, attributes = _KB_ALLOWED_ATTRS, strip = True, ) logger.debug(f'[KB SANITIZE] input_len={len(raw_html or "")} output_len={len(cleaned)}') return cleaned 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(): from sqlalchemy import func, case now = datetime.utcnow() day7 = now - timedelta(days=7) day30 = now - timedelta(days=30) # ── Headline stats ──────────────────────────────────────────────────────── 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(), } # ── IT performance metrics ──────────────────────────────────────────────── # Avg resolution time (hours) — resolved tickets only resolved_tickets = Ticket.query.filter( Ticket.resolved_at.isnot(None), Ticket.created_at.isnot(None), ).all() if resolved_tickets: total_hours = sum( (t.resolved_at - t.created_at).total_seconds() / 3600 for t in resolved_tickets ) avg_resolution_hours = round(total_hours / len(resolved_tickets), 1) else: avg_resolution_hours = None # Resolved last 7 / 30 days resolved_7d = Ticket.query.filter(Ticket.resolved_at >= day7).count() resolved_30d = Ticket.query.filter(Ticket.resolved_at >= day30).count() # New tickets last 7 / 30 days new_7d = Ticket.query.filter(Ticket.created_at >= day7).count() new_30d = Ticket.query.filter(Ticket.created_at >= day30).count() # Unassigned open tickets unassigned = Ticket.query.filter( Ticket.assigned_to_id.is_(None), Ticket.status.in_([TicketStatus.OPEN, TicketStatus.IN_PROGRESS]), ).count() # SLA at-risk: open/in-progress tickets older than 3 days sla_breach = Ticket.query.filter( Ticket.status.in_([TicketStatus.OPEN, TicketStatus.IN_PROGRESS]), Ticket.created_at < now - timedelta(days=3), ).count() # Open tickets by priority priority_breakdown = {} for priority in ('critical', 'high', 'medium', 'low'): priority_breakdown[priority] = Ticket.query.filter( Ticket.priority == priority, Ticket.status.in_([TicketStatus.OPEN, TicketStatus.IN_PROGRESS, TicketStatus.PENDING]), ).count() # Open tickets by category (top 6) from app.models import TicketCategory category_breakdown = [] for cat in (TicketCategory.SOFTWARE, TicketCategory.HARDWARE, TicketCategory.NETWORK, TicketCategory.ACCESS, TicketCategory.EMAIL, TicketCategory.OTHER): count = Ticket.query.filter( Ticket.category == cat, Ticket.status.in_([TicketStatus.OPEN, TicketStatus.IN_PROGRESS]), ).count() category_breakdown.append((cat.replace('_', ' ').title(), count)) category_breakdown.sort(key=lambda x: x[1], reverse=True) # ── Staff performance ───────────────────────────────────────────────────── # Replaced N×5 per-member query loop with two aggregated SQL queries so # that the dashboard cost is constant regardless of IT team size. # # Query 1: grouped counts per assignee (open workload, totals, urgent). # Query 2: resolution timestamps for avg-hours calculation (one row per # resolved ticket, grouped in Python to avoid TIMESTAMPDIFF # dialect differences between MySQL and SQLite). it_members = User.query.filter( User.role.in_([UserRole.IT_STAFF, UserRole.ADMIN]), User.is_active == True, ).order_by(User.full_name).all() it_member_ids = [m.id for m in it_members] # ── Aggregated count query ──────────────────────────────────────────────── # Produces one row per assignee with all needed counts computed in SQL. count_rows = ( db.session.query( Ticket.assigned_to_id, func.sum(case( (Ticket.status.in_([TicketStatus.OPEN, TicketStatus.IN_PROGRESS, TicketStatus.PENDING]), 1), else_=0, )).label('assigned_open'), func.sum(case( (Ticket.status.in_([TicketStatus.RESOLVED, TicketStatus.CLOSED]), 1), else_=0, )).label('resolved_total'), func.sum(case( (Ticket.resolved_at >= day30, 1), else_=0, )).label('resolved_30d'), func.sum(case( ( (Ticket.priority.in_(['critical', 'high'])) & (Ticket.status.in_([TicketStatus.OPEN, TicketStatus.IN_PROGRESS])), 1, ), else_=0, )).label('urgent_open'), ) .filter(Ticket.assigned_to_id.in_(it_member_ids)) .group_by(Ticket.assigned_to_id) .all() ) counts_by_id = {row.assigned_to_id: row for row in count_rows} # ── Resolution-time query ───────────────────────────────────────────────── # Fetches only the two timestamp columns needed for avg-hours per assignee. # Using Python arithmetic avoids TIMESTAMPDIFF / strftime dialect issues. res_rows = ( db.session.query( Ticket.assigned_to_id, Ticket.created_at, Ticket.resolved_at, ) .filter( Ticket.assigned_to_id.in_(it_member_ids), Ticket.resolved_at.isnot(None), Ticket.created_at.isnot(None), ) .all() ) # Group timestamps by assignee in Python for the avg-hours calculation. res_by_id: dict[int, list[float]] = {} for row in res_rows: hours = (row.resolved_at - row.created_at).total_seconds() / 3600 res_by_id.setdefault(row.assigned_to_id, []).append(hours) # ── Assemble staff_stats from pre-fetched data ──────────────────────────── staff_stats = [] for member in it_members: row = counts_by_id.get(member.id) hours = res_by_id.get(member.id, []) staff_stats.append({ 'name' : member.full_name, 'assigned_open' : int(row.assigned_open) if row else 0, 'resolved_total': int(row.resolved_total) if row else 0, 'resolved_30d' : int(row.resolved_30d) if row else 0, 'avg_hours' : round(sum(hours) / len(hours), 1) if hours else None, 'urgent_open' : int(row.urgent_open) if row else 0, }) # Sort by open workload desc staff_stats.sort(key=lambda x: x['assigned_open'], reverse=True) # ── 5 most urgent open tickets ──────────────────────────────────────────── urgent_tickets = Ticket.query.filter( Ticket.status.in_([TicketStatus.OPEN, TicketStatus.IN_PROGRESS]), ).order_by( # critical first, then high, then by age case( (Ticket.priority == 'critical', 0), (Ticket.priority == 'high', 1), (Ticket.priority == 'medium', 2), else_=3, ), Ticket.created_at.asc(), ).limit(8).all() return render_template('admin/index.html', stats = stats, avg_resolution_hours= avg_resolution_hours, resolved_7d = resolved_7d, resolved_30d = resolved_30d, new_7d = new_7d, new_30d = new_30d, unassigned = unassigned, sla_breach = sla_breach, priority_breakdown = priority_breakdown, category_breakdown = category_breakdown, staff_stats = staff_stats, urgent_tickets = urgent_tickets, now = now, ) # ─── 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) pw_error = validate_password(password, confirm) if pw_error: flash(pw_error, '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) # Flush to obtain user.id from the DB sequence before logging. # Without this flush, user.id is None and the activity log entry # records entity_id=None, making the log entry unlinkable. db.session.flush() log_action(current_user.id, 'admin_user_create', 'user', user.id, f'email={email} role={role}') db.session.commit() 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 = db.session.get(User, user_id) or abort(404) 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: confirm_pw = request.form.get('confirm_password', '') pw_error = validate_password(new_pw, confirm_pw, user.password_hash) if pw_error: flash(pw_error, 'danger') return render_template('admin/edit_user.html', user=user, roles=_roles()) user.set_password(new_pw) logger.info(f'[ADMIN PASSWORD RESET] target_user_id={user.id} by admin_id={current_user.id}') log_action(current_user.id, 'admin_user_edit', 'user', user.id, f'role_change={old_role}->{user.role} active={user.is_active}') db.session.commit() 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 = db.session.get(User, user_id) or abort(404) if user.id == current_user.id: flash('You cannot delete your own account.', 'danger') return redirect(url_for('admin.users')) user.is_active = False log_action(current_user.id, 'admin_user_deactivate', 'user', user.id) db.session.commit() 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', '') search = request.args.get('q', '').strip() 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) if search: from app.models import Comment from sqlalchemy import func submitter_alias = db.aliased(User) assignee_alias = db.aliased(User) stripped_body = func.regexp_replace(Comment.body, r'<[^>]+>', '') q = ( q .outerjoin(submitter_alias, submitter_alias.id == Ticket.created_by_id) .outerjoin(assignee_alias, assignee_alias.id == Ticket.assigned_to_id) .outerjoin(Comment, Comment.ticket_id == Ticket.id) .filter( Ticket.title.ilike(f'%{search}%') | Ticket.ticket_number.ilike(f'%{search}%') | Ticket.description.ilike(f'%{search}%') | submitter_alias.full_name.ilike(f'%{search}%') | assignee_alias.full_name.ilike(f'%{search}%') | stripped_body.ilike(f'%{search}%') ) .distinct() ) tickets = q.order_by(Ticket.created_at.desc()).paginate(page=page, per_page=25) it_staff = User.query.filter( User.role.in_([UserRole.IT_STAFF, UserRole.ADMIN]), User.is_active == True ).order_by(User.full_name).all() return render_template('admin/tickets.html', tickets=tickets, status=status, priority=priority, assigned=assigned, search=search, it_staff=it_staff) @admin_bp.route('/tickets//delete', methods=['POST']) @login_required @admin_required def delete_ticket(ticket_id): """Permanently delete a single ticket and all its related data. Cascades handled by SQLAlchemy relationships (cascade='all, delete-orphan'): comments, attachments (rows), notifications, history, satisfaction, links. Physical attachment files on disk are deleted explicitly before the commit. """ ticket = db.session.get(Ticket, ticket_id) or abort(404) ticket_number = ticket.ticket_number ticket_title = ticket.title # Delete physical attachment files from disk before removing DB rows upload_dir = current_app.config.get('UPLOAD_FOLDER', '') for attachment in ticket.attachments.all(): if attachment.stored_name and upload_dir: filepath = os.path.join(upload_dir, attachment.stored_name) if os.path.isfile(filepath): try: os.remove(filepath) except OSError as exc: logger.warning(f'[TICKET DELETE] Could not remove file ' f'{filepath}: {exc}') log_action(current_user.id, 'admin_ticket_delete', 'ticket', ticket.id, f'ticket_number={ticket_number} title={ticket_title!r}') db.session.delete(ticket) db.session.commit() logger.info(f'[TICKET DELETE] ticket_number={ticket_number} ' f'ticket_id={ticket_id} by admin_id={current_user.id}') flash(f'Ticket {ticket_number} has been permanently deleted.', 'success') return redirect(url_for('admin.all_tickets')) # ─── Knowledge Base Management ──────────────────────────────────────────────── @admin_bp.route('/kb') @login_required @it_required def kb_list(): q = request.args.get('q', '').strip() category = request.args.get('category', '') author_id = request.args.get('author_id', '', type=str) published = request.args.get('published', '') # 'yes' | 'no' | '' query = KnowledgeBase.query if q: query = query.filter( KnowledgeBase.title.ilike(f'%{q}%') | KnowledgeBase.tags.ilike(f'%{q}%') ) if category: query = query.filter_by(category=category) if author_id: query = query.filter_by(author_id=int(author_id)) if published == 'yes': query = query.filter_by(is_published=True) elif published == 'no': query = query.filter_by(is_published=False) articles = query.order_by(KnowledgeBase.created_at.desc()).all() # Build filter option lists from existing data categories = sorted({a.category for a in KnowledgeBase.query.with_entities(KnowledgeBase.category).distinct() if a.category}) authors = User.query.filter( User.id.in_( db.session.query(KnowledgeBase.author_id).distinct() ), User.is_active == True, ).order_by(User.full_name).all() return render_template('admin/kb_list.html', articles = articles, categories = categories, authors = authors, q = q, sel_category = category, sel_author_id = author_id, sel_published = published, ) # ── 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 _image_exts = {'png', 'jpg', 'jpeg', 'gif', 'webp'} file_error = validate_file(f, _image_exts) if file_error: logger.warning(f'[KB IMAGE UPLOAD REJECTED] {file_error} filename="{f.filename}" user_id={current_user.id}') return jsonify({'error': file_error}), 400 ext = f.filename.rsplit('.', 1)[-1].lower() 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. Security note ------------- The converter is used deliberately instead of . The converter permits forward slashes in the captured segment, which would allow a crafted URL like /kb/files/../../etc/passwd to traverse outside the upload directory. disallows slashes, restricting the value to a flat filename — matching the UUID-based stored_name format (e.g. 'a1b2c3d4e5f6....png') used by all upload helpers. @login_required is applied here because the upload directory is shared across KB files, ticket attachments, comment images, and user avatars. Without authentication, an unauthenticated caller who knows or guesses any stored_name (UUID-based) could retrieve arbitrary files from the shared uploads folder — including confidential ticket attachments. """ 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) att_id_val, filename = att.id, att.filename # DB row is the source of truth — delete and commit it first, then remove # the physical file. If the process dies mid-operation this leaves at # worst an orphan file on disk (harmless, cleanable later) rather than a # DB row pointing at a file that's already gone (a broken download link). log_action(current_user.id, 'kb_attachment_delete', 'kb_attachment', att_id_val, f'article_id={article_id} filename={filename}') db.session.delete(att) db.session.commit() logger.info(f'[KB ATTACHMENT DELETE] att_id={att_id_val} article_id={article_id} by user_id={current_user.id}') if os.path.exists(filepath): try: os.remove(filepath) except OSError as exc: logger.warning(f'[KB ATTACHMENT DELETE] Could not remove file {filepath}: {exc}') # Return JSON so the edit page can remove the row without a full reload return jsonify({'ok': True, 'att_id': att_id_val}) @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 = _sanitize_kb_body(request.form.get('body', '')), 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: file_error = validate_file(f, _KB_ALLOWED_EXT) if file_error: logger.warning(f'[KB NEW UPLOAD REJECTED] {file_error} filename="{f.filename}" user_id={current_user.id}') continue att = _save_kb_file(f, article.id) db.session.add(att) log_action(current_user.id, 'kb_create', 'knowledge_base', article.id, f'title={article.title}') db.session.commit() 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 = db.session.get(KnowledgeBase, article_id) or abort(404) if request.method == 'POST': try: article.title = request.form.get('title', article.title).strip() article.body = _sanitize_kb_body(request.form.get('body', article.body)) 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: file_error = validate_file(f, _KB_ALLOWED_EXT) if file_error: logger.warning(f'[KB EDIT UPLOAD REJECTED] {file_error} filename="{f.filename}" user_id={current_user.id}') continue att = _save_kb_file(f, article.id) db.session.add(att) log_action(current_user.id, 'kb_edit', 'knowledge_base', article.id) db.session.commit() 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 = db.session.get(KnowledgeBase, article_id) or abort(404) article.is_published = not article.is_published state = 'published' if article.is_published else 'unpublished' log_action(current_user.id, f'kb_{state}', 'knowledge_base', article.id, f'title={article.title}') db.session.commit() 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 = db.session.get(KnowledgeBase, article_id) or abort(404) upload_dir = current_app.config['UPLOAD_FOLDER'] # Remove physical files before the cascade deletes the KBAttachment rows. # Without this step the DB records disappear but the files remain on disk # with no pointer to them — unrecoverable orphans. for att in article.attachments.all(): filepath = os.path.join(upload_dir, att.stored_name) if os.path.exists(filepath): os.remove(filepath) logger.info(f'[KB DELETE FILE] stored_name={att.stored_name} article_id={article_id} by user_id={current_user.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', methods=['GET', 'POST']) @login_required @admin_required def activity_logs(): # ── Cleanup action ──────────────────────────────────────────────────────── if request.method == 'POST': days = request.form.get('days', type=int) if days not in (7, 30, 60, 90): flash('Invalid retention period selected.', 'danger') return redirect(url_for('admin.activity_logs')) cutoff = datetime.utcnow() - timedelta(days=days) deleted = ActivityLog.query.filter(ActivityLog.created_at < cutoff).delete() db.session.commit() log_action(current_user.id, 'activity_log_cleanup', 'activity_log', None, f'deleted={deleted} older_than={days}_days cutoff={cutoff.strftime("%Y-%m-%d")}') db.session.commit() logger.info(f'[ACTIVITY LOG CLEANUP] deleted={deleted} days={days} by user_id={current_user.id}') flash(f'Deleted {deleted:,} log entr{"y" if deleted == 1 else "ies"} older than {days} days.', 'success') return redirect(url_for('admin.activity_logs')) # ── Stats for the summary bar ───────────────────────────────────────────── from sqlalchemy import func total = ActivityLog.query.count() oldest = db.session.query(func.min(ActivityLog.created_at)).scalar() counts_by_retention = {} for days in (7, 30, 60, 90): cutoff = datetime.utcnow() - timedelta(days=days) counts_by_retention[days] = ActivityLog.query.filter( ActivityLog.created_at < cutoff ).count() # ── Paginated log listing ───────────────────────────────────────────────── 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, total = total, oldest = oldest, counts_by_retention= counts_by_retention, ) @admin_bp.route('/tickets/export') @login_required @admin_required @limiter.limit('10 per hour') def export_tickets(): """Stream a CSV of tickets matching the current filter params.""" status = request.args.get('status', '') priority = request.args.get('priority', '') assigned = request.args.get('assigned', '') search = request.args.get('q', '').strip() 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) if search: from sqlalchemy import func submitter_alias = db.aliased(User) assignee_alias = db.aliased(User) stripped_body = func.regexp_replace(Comment.body, r'<[^>]+>', '') q = ( q .outerjoin(submitter_alias, submitter_alias.id == Ticket.created_by_id) .outerjoin(assignee_alias, assignee_alias.id == Ticket.assigned_to_id) .outerjoin(Comment, Comment.ticket_id == Ticket.id) .filter( Ticket.title.ilike(f'%{search}%') | Ticket.ticket_number.ilike(f'%{search}%') | Ticket.description.ilike(f'%{search}%') | submitter_alias.full_name.ilike(f'%{search}%') | assignee_alias.full_name.ilike(f'%{search}%') | stripped_body.ilike(f'%{search}%') ) .distinct() ) tickets = q.order_by(Ticket.created_at.desc()).all() log_action(current_user.id, 'ticket_export', 'ticket', None, f'status={status} priority={priority} assigned={assigned} q={search} count={len(tickets)}') logger.info(f'[ADMIN EXPORT] tickets count={len(tickets)} by admin_id={current_user.id}') # Build the entire CSV in memory within the request context so the # SQLAlchemy session remains active for all relationship accesses # (creator, assignee). Streaming generators execute outside the # request context and cause DetachedInstanceError on lazy-loaded attrs. buf = io.StringIO() writer = csv.writer(buf) writer.writerow([ 'Ticket #', 'Title', 'Category', 'Status', 'Priority', 'Submitted By', 'Assigned To', 'Created', 'Updated', 'Resolved', 'Due Date', 'AI Generated', 'Resolution Notes', ]) for t in tickets: writer.writerow([ t.ticket_number, t.title, t.category.replace('_', ' ').title(), t.status.replace('_', ' ').title(), t.priority.title(), t.creator.full_name, t.assignee.full_name if t.assignee else '', t.created_at.strftime('%Y-%m-%d %H:%M'), t.updated_at.strftime('%Y-%m-%d %H:%M') if t.updated_at else '', t.resolved_at.strftime('%Y-%m-%d %H:%M') if t.resolved_at else '', t.due_date.strftime('%Y-%m-%d') if t.due_date else '', 'Yes' if t.ai_generated else 'No', t.resolution_notes or '', ]) filename = f"tickets_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.csv" return Response( buf.getvalue(), mimetype = 'text/csv', headers = {'Content-Disposition': f'attachment; filename={filename}'}, ) # ─── Canned Responses Management ───────────────────────────────────────────── @admin_bp.route('/canned-responses') @login_required @it_required def canned_responses(): """List all canned responses.""" items = CannedResponse.query.order_by( CannedResponse.category, CannedResponse.title ).all() categories = sorted({r.category for r in items if r.category}) return render_template('admin/canned_responses.html', items=items, categories=categories) @admin_bp.route('/canned-responses/new', methods=['GET', 'POST']) @login_required @it_required def canned_response_new(): if request.method == 'POST': title = request.form.get('title', '').strip() body = request.form.get('body', '').strip() category = request.form.get('category', '').strip() if not title or not body: flash('Title and body are required.', 'danger') return render_template('admin/canned_response_edit.html', item=None) item = CannedResponse( title = title, body = body, category = category or None, created_by = current_user.id, ) db.session.add(item) db.session.flush() log_action(current_user.id, 'canned_response_create', 'canned_response', item.id, f'title={title}') db.session.commit() logger.info(f'[CANNED RESPONSE CREATE] id={item.id} by user_id={current_user.id}') flash('Canned response created.', 'success') return redirect(url_for('admin.canned_responses')) return render_template('admin/canned_response_edit.html', item=None) @admin_bp.route('/canned-responses//edit', methods=['GET', 'POST']) @login_required @it_required def canned_response_edit(item_id): item = db.session.get(CannedResponse, item_id) or abort(404) if request.method == 'POST': item.title = request.form.get('title', item.title).strip() item.body = request.form.get('body', item.body).strip() item.category = request.form.get('category', '').strip() or None item.is_active = bool(request.form.get('is_active')) if not item.title or not item.body: flash('Title and body are required.', 'danger') return render_template('admin/canned_response_edit.html', item=item) log_action(current_user.id, 'canned_response_edit', 'canned_response', item.id, f'title={item.title}') db.session.commit() logger.info(f'[CANNED RESPONSE EDIT] id={item.id} by user_id={current_user.id}') flash('Canned response updated.', 'success') return redirect(url_for('admin.canned_responses')) return render_template('admin/canned_response_edit.html', item=item) @admin_bp.route('/canned-responses//delete', methods=['POST']) @login_required @it_required def canned_response_delete(item_id): item = db.session.get(CannedResponse, item_id) or abort(404) log_action(current_user.id, 'canned_response_delete', 'canned_response', item.id, f'title={item.title}') logger.info(f'[CANNED RESPONSE DELETE] id={item.id} by user_id={current_user.id}') db.session.delete(item) db.session.commit() flash('Canned response deleted.', 'success') return redirect(url_for('admin.canned_responses')) # ─── Bulk Ticket Actions ───────────────────────────────────────────────────── @admin_bp.route('/tickets/bulk-action', methods=['POST']) @login_required @it_required def bulk_ticket_action(): """Apply a single action to multiple selected tickets at once. Accepted actions: - resolve → set status to Resolved, stamp resolved_at - close → set status to Closed, stamp closed_at - assign_me → assign all selected tickets to current user - unassign → clear assigned_to_id """ from app.services.notification_service import notify_status_change from app.services.sla_service import clear_sla_notification action = request.form.get('action', '') ticket_ids = request.form.getlist('ticket_ids', type=int) if not ticket_ids: flash('No tickets selected.', 'warning') return redirect(url_for('admin.all_tickets')) valid_actions = ('resolve', 'close', 'assign_me', 'unassign', 'assign_to', 'delete') if action not in valid_actions: flash('Invalid action.', 'danger') return redirect(url_for('admin.all_tickets')) assign_to_id = request.form.get('assign_to_id', type=int) if action == 'assign_to' and not assign_to_id: flash('Please select a staff member to assign to.', 'warning') return redirect(url_for('admin.all_tickets')) # Bulk delete is admin-only if action == 'delete' and not current_user.is_admin: flash('Only administrators may delete tickets.', 'danger') return redirect(url_for('admin.all_tickets')) tickets = Ticket.query.filter(Ticket.id.in_(ticket_ids)).all() now = datetime.utcnow() count = 0 notif_tickets = [] # collect for post-commit notifications # ── Bulk delete: handle separately so we can early-return cleanly ───────── if action == 'delete': upload_dir = current_app.config.get('UPLOAD_FOLDER', '') for ticket in tickets: for attachment in ticket.attachments.all(): if attachment.stored_name and upload_dir: filepath = os.path.join(upload_dir, attachment.stored_name) if os.path.isfile(filepath): try: os.remove(filepath) except OSError as exc: logger.warning(f'[BULK DELETE] Could not remove file ' f'{filepath}: {exc}') log_action(current_user.id, 'admin_ticket_delete', 'ticket', ticket.id, f'bulk=true ticket_number={ticket.ticket_number} title={ticket.title!r}') db.session.delete(ticket) count += 1 db.session.commit() logger.info(f'[BULK DELETE] deleted={count} ticket_ids={ticket_ids} ' f'by admin_id={current_user.id}') flash(f'{count} ticket{"s" if count != 1 else ""} permanently deleted.', 'success') return redirect(url_for('admin.all_tickets', status = request.form.get('status', ''), priority = request.form.get('priority', ''), assigned = request.form.get('assigned', ''), q = request.form.get('q', ''), page = request.form.get('page', 1), )) # ───────────────────────────────────────────────────────────────────────── for ticket in tickets: old_status = ticket.status old_assigned = ticket.assigned_to_id changed = False if action == 'resolve' and ticket.status not in ( TicketStatus.RESOLVED, TicketStatus.CLOSED): ticket.status = TicketStatus.RESOLVED ticket.resolved_at = now clear_sla_notification(ticket.id) log_ticket_history(ticket, 'status', old_status, TicketStatus.RESOLVED, current_user.id) notif_tickets.append((ticket, old_status)) changed = True elif action == 'close' and ticket.status != TicketStatus.CLOSED: ticket.status = TicketStatus.CLOSED ticket.closed_at = now clear_sla_notification(ticket.id) log_ticket_history(ticket, 'status', old_status, TicketStatus.CLOSED, current_user.id) notif_tickets.append((ticket, old_status)) changed = True elif action == 'assign_me' and ticket.assigned_to_id != current_user.id: ticket.assigned_to_id = current_user.id log_ticket_history(ticket, 'assigned_to', old_assigned or 'Unassigned', current_user.full_name, current_user.id) changed = True elif action == 'assign_to' and ticket.assigned_to_id != assign_to_id: target = db.session.get(User, assign_to_id) if target: ticket.assigned_to_id = assign_to_id log_ticket_history(ticket, 'assigned_to', old_assigned or 'Unassigned', target.full_name, current_user.id) changed = True elif action == 'unassign' and ticket.assigned_to_id is not None: ticket.assigned_to_id = None log_ticket_history(ticket, 'assigned_to', old_assigned or 'Unassigned', 'Unassigned', current_user.id) changed = True if changed: log_action(current_user.id, f'ticket_bulk_{action}', 'ticket', ticket.id, f'action={action}') count += 1 db.session.commit() logger.info(f'[BULK ACTION] action={action} affected={count} ' f'ticket_ids={ticket_ids} by user_id={current_user.id}') # Send status-change notifications after commit for ticket, old_status in notif_tickets: notify_status_change(ticket, old_status, current_user) action_labels = { 'resolve': 'resolved', 'close': 'closed', 'assign_me': 'assigned to you', 'assign_to': f'assigned to {db.session.get(User, assign_to_id).full_name if assign_to_id else "staff"}', 'unassign': 'unassigned', } flash(f'{count} ticket{"s" if count != 1 else ""} {action_labels[action]}.', 'success') # Preserve current filter params on redirect return redirect(url_for('admin.all_tickets', status = request.form.get('status', ''), priority = request.form.get('priority', ''), assigned = request.form.get('assigned', ''), q = request.form.get('q', ''), page = request.form.get('page', 1), )) # ─── Ticket Templates ──────────────────────────────────────────────────────── @admin_bp.route('/ticket-templates') @login_required @it_required def ticket_templates(): templates = TicketTemplate.query.order_by( TicketTemplate.sort_order, TicketTemplate.name ).all() return render_template('admin/ticket_templates.html', templates=templates) @admin_bp.route('/ticket-templates/new', methods=['GET', 'POST']) @login_required @it_required def ticket_template_new(): if request.method == 'POST': name = request.form.get('name', '').strip() category = request.form.get('category', 'other') priority = request.form.get('priority', 'medium') title_hint = request.form.get('title_hint', '').strip() description= request.form.get('description', '').strip() icon = request.form.get('icon', 'bi-file-text').strip() sort_order = request.form.get('sort_order', 0, type=int) is_active = bool(request.form.get('is_active')) if not name: flash('Template name is required.', 'danger') return render_template('admin/ticket_template_edit.html', t=None, categories=_categories_list(), priorities=_priorities_list()) t = TicketTemplate( name=name, category=category, priority=priority, title_hint=title_hint, description=description, icon=icon, sort_order=sort_order, is_active=is_active, created_by=current_user.id, ) db.session.add(t) db.session.flush() log_action(current_user.id, 'ticket_template_create', 'ticket_template', t.id, f'name={name}') db.session.commit() logger.info(f'[TEMPLATE CREATE] id={t.id} name={name} by user_id={current_user.id}') flash(f'Template "{name}" created.', 'success') return redirect(url_for('admin.ticket_templates')) return render_template('admin/ticket_template_edit.html', t=None, categories=_categories_list(), priorities=_priorities_list()) @admin_bp.route('/ticket-templates//edit', methods=['GET', 'POST']) @login_required @it_required def ticket_template_edit(tmpl_id): t = db.session.get(TicketTemplate, tmpl_id) or abort(404) if request.method == 'POST': t.name = request.form.get('name', t.name).strip() t.category = request.form.get('category', t.category) t.priority = request.form.get('priority', t.priority) t.title_hint = request.form.get('title_hint', '').strip() t.description = request.form.get('description', '').strip() t.icon = request.form.get('icon', 'bi-file-text').strip() t.sort_order = request.form.get('sort_order', 0, type=int) t.is_active = bool(request.form.get('is_active')) log_action(current_user.id, 'ticket_template_edit', 'ticket_template', t.id, f'name={t.name}') db.session.commit() logger.info(f'[TEMPLATE EDIT] id={t.id} by user_id={current_user.id}') flash(f'Template "{t.name}" updated.', 'success') return redirect(url_for('admin.ticket_templates')) return render_template('admin/ticket_template_edit.html', t=t, categories=_categories_list(), priorities=_priorities_list()) @admin_bp.route('/ticket-templates//delete', methods=['POST']) @login_required @it_required def ticket_template_delete(tmpl_id): t = db.session.get(TicketTemplate, tmpl_id) or abort(404) name = t.name log_action(current_user.id, 'ticket_template_delete', 'ticket_template', t.id, f'name={name}') logger.info(f'[TEMPLATE DELETE] id={t.id} name={name} by user_id={current_user.id}') db.session.delete(t) db.session.commit() flash(f'Template "{name}" deleted.', 'success') return redirect(url_for('admin.ticket_templates')) # ─── Satisfaction Survey Report ─────────────────────────────────────────────── @admin_bp.route('/satisfaction') @login_required @it_required def satisfaction_report(): surveys = TicketSatisfaction.query.filter( TicketSatisfaction.rating.isnot(None) ).order_by(TicketSatisfaction.submitted_at.desc()).all() total = len(surveys) avg_rating = round(sum(s.rating for s in surveys) / total, 2) if total else None dist = {i: sum(1 for s in surveys if s.rating == i) for i in range(1, 6)} pending = TicketSatisfaction.query.filter_by(rating=None).count() return render_template('admin/satisfaction_report.html', surveys=surveys, total=total, avg_rating=avg_rating, dist=dist, pending=pending) def _categories_list(): from app.models import TicketCategory return [TicketCategory.HARDWARE, TicketCategory.SOFTWARE, TicketCategory.NETWORK, TicketCategory.ACCESS, TicketCategory.EMAIL, TicketCategory.PRINTER, TicketCategory.PHONE, TicketCategory.SECURITY, TicketCategory.OTHER] def _priorities_list(): from app.models import TicketPriority return [TicketPriority.LOW, TicketPriority.MEDIUM, TicketPriority.HIGH, TicketPriority.CRITICAL] def _roles(): return [UserRole.EMPLOYEE, UserRole.IT_STAFF, UserRole.ADMIN] @admin_bp.route('/settings', methods=['GET', 'POST']) @login_required @admin_required def settings(): from app.models import SystemSetting if request.method == 'POST': form_type = request.form.get('form_type') # ── Registration toggle ─────────────────────────────────────────────── if form_type == 'registration': new_value = '1' if request.form.get('registration_enabled') == '1' else '0' old_value = SystemSetting.get('registration_enabled', 'true') SystemSetting.set('registration_enabled', new_value, 'Allow new users to self-register via /auth/register') db.session.commit() state_label = 'enabled' if new_value == '1' else 'disabled' log_action(current_user.id, 'setting_update', 'system_setting', None, f'registration_enabled changed from {old_value} to {new_value} by admin_id={current_user.id}') logger.info(f'[ADMIN SETTINGS] registration_enabled={new_value} by admin_id={current_user.id}') flash(f'User registration has been {state_label}.', 'success') return redirect(url_for('admin.settings')) # ── Timezone setting ────────────────────────────────────────────────── if form_type == 'survey': new_value = '1' if request.form.get('survey_enabled') == '1' else '0' old_value = SystemSetting.get('survey_enabled', '1') SystemSetting.set('survey_enabled', new_value, 'Send satisfaction survey email when a ticket is resolved') db.session.commit() state_label = 'enabled' if new_value == '1' else 'disabled' log_action(current_user.id, 'setting_update', 'system_setting', None, f'survey_enabled changed from {old_value} to {new_value}') logger.info(f'[ADMIN SETTINGS] survey_enabled={new_value} by admin_id={current_user.id}') flash(f'Satisfaction survey has been {state_label}.', 'success') return redirect(url_for('admin.settings')) if form_type == 'timezone': from zoneinfo import ZoneInfo, ZoneInfoNotFoundError new_tz = request.form.get('app_timezone', 'America/New_York').strip() try: ZoneInfo(new_tz) # validate before saving except (ZoneInfoNotFoundError, KeyError): flash(f'Unknown timezone "{new_tz}". Please select a valid timezone.', 'danger') return redirect(url_for('admin.settings')) old_tz = SystemSetting.get('app_timezone', 'America/New_York') SystemSetting.set('app_timezone', new_tz, 'Display timezone for all dates and times in the UI') db.session.commit() log_action(current_user.id, 'setting_update', 'system_setting', None, f'app_timezone changed from {old_tz!r} to {new_tz!r}') logger.info(f'[ADMIN SETTINGS] app_timezone={new_tz} by admin_id={current_user.id}') flash(f'Timezone updated to {new_tz}.', 'success') return redirect(url_for('admin.settings')) # ── Email ingestion settings ────────────────────────────────────────── if form_type == 'email_ingestion': fields = { 'email_ingestion_enabled' : request.form.get('email_ingestion_enabled', '0'), 'email_ingestion_host' : request.form.get('email_ingestion_host', '').strip(), 'email_ingestion_port' : request.form.get('email_ingestion_port', '993').strip(), 'email_ingestion_user' : request.form.get('email_ingestion_user', '').strip(), 'email_ingestion_folder' : request.form.get('email_ingestion_folder', 'INBOX').strip(), 'email_ingestion_move_to' : request.form.get('email_ingestion_move_to', 'Processed').strip(), 'email_ingestion_interval': request.form.get('email_ingestion_interval', '5').strip(), } # Password: only update if a new value was provided. Encrypted # at rest — see app/services/crypto_service.py. new_pw = request.form.get('email_ingestion_password', '').strip() if new_pw: from app.services.crypto_service import encrypt_secret fields['email_ingestion_password'] = encrypt_secret(new_pw) changes = [] for key, value in fields.items(): old_val = SystemSetting.get(key, '') if str(value) != str(old_val): SystemSetting.set(key, value) safe_key = key.replace('email_ingestion_', '') # Never log the password value changes.append(f'{safe_key}={"[updated]" if "password" in key else repr(value)}') db.session.commit() if changes: log_action(current_user.id, 'email_ingestion_settings_update', 'system_setting', None, ', '.join(changes)) logger.info(f'[ADMIN EMAIL INGEST] Settings updated: {", ".join(changes)} ' f'by admin_id={current_user.id}') flash('Email ingestion settings saved.', 'success') return redirect(url_for('admin.settings')) # ── Branding update ─────────────────────────────────────────────────── if form_type == 'branding': fields = { 'app_name' : ('Application name', request.form.get('app_name', '').strip()), 'app_subtitle' : ('App subtitle', request.form.get('app_subtitle', '').strip()), 'company_name' : ('Company name', request.form.get('company_name', '').strip()), 'logo_initials': ('Logo initials', request.form.get('logo_initials', '').strip()[:3]), 'primary_color': ('Primary colour', request.form.get('primary_color', '#2563eb').strip()), } changes = [] for key, (label, new_val) in fields.items(): old_val = SystemSetting.get(key, '') if new_val != old_val: SystemSetting.set(key, new_val) changes.append(f'{key}={new_val!r}') # ── Logo file upload ────────────────────────────────────────────── logo_file = request.files.get('logo_file') if logo_file and logo_file.filename: from app.services.validation_service import validate_file from werkzeug.utils import secure_filename allowed = {'png', 'jpg', 'jpeg', 'gif', 'webp'} file_error = validate_file(logo_file, allowed) if file_error: flash(f'Logo not saved: {file_error}', 'danger') else: logo_file.stream.seek(0, 2) logo_size = logo_file.stream.tell() logo_file.stream.seek(0) if logo_size > 2 * 1024 * 1024: flash('Logo image must be under 2 MB.', 'danger') else: ext = secure_filename(logo_file.filename).rsplit('.', 1)[-1].lower() stored_name = f"logo_{uuid.uuid4().hex}.{ext}" upload_dir = current_app.config['UPLOAD_FOLDER'] # Delete old logo from disk old_logo = SystemSetting.get('logo_stored_name', '') if old_logo: old_path = os.path.join(upload_dir, old_logo) if os.path.exists(old_path): os.remove(old_path) logo_file.save(os.path.join(upload_dir, stored_name)) SystemSetting.set('logo_stored_name', stored_name, 'Stored filename of the uploaded company logo image') changes.append(f'logo_stored_name={stored_name!r}') logger.info(f'[ADMIN BRANDING] logo uploaded file={stored_name} by admin_id={current_user.id}') # ── Logo removal ────────────────────────────────────────────────── if request.form.get('remove_logo') == '1': old_logo = SystemSetting.get('logo_stored_name', '') if old_logo: old_path = os.path.join(current_app.config['UPLOAD_FOLDER'], old_logo) if os.path.exists(old_path): os.remove(old_path) SystemSetting.set('logo_stored_name', '') changes.append('logo_stored_name removed') db.session.commit() if changes: log_action(current_user.id, 'branding_update', 'system_setting', None, f'changes: {", ".join(changes)} by admin_id={current_user.id}') logger.info(f'[ADMIN BRANDING] {", ".join(changes)} by admin_id={current_user.id}') flash('Branding settings saved.', 'success') return redirect(url_for('admin.settings')) registration_enabled = SystemSetting.get_bool('registration_enabled', default=True) survey_enabled = SystemSetting.get_bool('survey_enabled', default=True) email_settings = { 'enabled' : SystemSetting.get_bool('email_ingestion_enabled', default=False), 'host' : SystemSetting.get('email_ingestion_host', ''), 'port' : SystemSetting.get('email_ingestion_port', '993'), 'user' : SystemSetting.get('email_ingestion_user', ''), # The stored password is never sent back to the browser — the field # is always rendered blank with a placeholder indicating whether one # is already configured (see settings.html). 'has_password': bool(SystemSetting.get('email_ingestion_password', '')), 'folder' : SystemSetting.get('email_ingestion_folder', 'INBOX'), 'move_to' : SystemSetting.get('email_ingestion_move_to', 'Processed'), 'interval' : SystemSetting.get('email_ingestion_interval', '5'), } branding_settings = { 'app_name' : SystemSetting.get('app_name', 'TechDesk'), 'app_subtitle' : SystemSetting.get('app_subtitle', 'IT Helpdesk System'), 'company_name' : SystemSetting.get('company_name', ''), 'logo_stored_name': SystemSetting.get('logo_stored_name', ''), 'logo_initials' : SystemSetting.get('logo_initials', 'TD'), 'primary_color' : SystemSetting.get('primary_color', '#2563eb'), } current_tz = SystemSetting.get('app_timezone', 'America/New_York') # Common timezone list for the selector common_timezones = [ ('America/New_York', 'Eastern Time (ET) — New York'), ('America/Chicago', 'Central Time (CT) — Chicago'), ('America/Denver', 'Mountain Time (MT) — Denver'), ('America/Phoenix', 'Mountain Time, no DST — Phoenix'), ('America/Los_Angeles', 'Pacific Time (PT) — Los Angeles'), ('America/Anchorage', 'Alaska Time — Anchorage'), ('Pacific/Honolulu', 'Hawaii Time — Honolulu'), ('America/Puerto_Rico', 'Atlantic Time — Puerto Rico'), ('UTC', 'UTC — Coordinated Universal Time'), ('Europe/London', 'GMT/BST — London'), ('Europe/Paris', 'CET/CEST — Paris, Berlin'), ('Europe/Helsinki', 'EET/EEST — Helsinki, Athens'), ('Asia/Dubai', 'GST — Dubai'), ('Asia/Kolkata', 'IST — India'), ('Asia/Singapore', 'SGT — Singapore'), ('Asia/Tokyo', 'JST — Tokyo'), ('Australia/Sydney', 'AEST/AEDT — Sydney'), ('Pacific/Auckland', 'NZST/NZDT — Auckland'), ] return render_template('admin/settings.html', registration_enabled=registration_enabled, survey_enabled=survey_enabled, branding_settings=branding_settings, email_settings=email_settings, current_tz=current_tz, common_timezones=common_timezones) # ── Email Ingestion: Test Connection ───────────────────────────────────────── @admin_bp.route('/settings/email-ingestion/test', methods=['POST']) @login_required @admin_required def email_ingestion_test(): """AJAX endpoint — attempt an IMAP login with the supplied credentials and return a JSON result. The password field is optional: if left blank the stored password is used so the admin does not have to re-enter it just to run a connection test. """ from app.models import SystemSetting import imaplib host = request.form.get('host', '').strip() port_str = request.form.get('port', '993').strip() user = request.form.get('user', '').strip() password = request.form.get('password', '').strip() folder = request.form.get('folder', 'INBOX').strip() # Fall back to the stored password if the admin left the field blank if not password: from app.services.crypto_service import decrypt_secret password = decrypt_secret(SystemSetting.get('email_ingestion_password', '')) if not host or not user or not password: return jsonify(ok=False, message='Host, username, and password are required.') try: port = int(port_str) except ValueError: return jsonify(ok=False, message=f'Invalid port number: {port_str!r}') try: imap = imaplib.IMAP4_SSL(host, port) imap.login(user, password) # Verify the watch folder exists quoted_folder = f'"{folder}"' status, data = imap.select(quoted_folder, readonly=True) if status != 'OK': imap.logout() return jsonify( ok=False, message=( f'Login succeeded but folder {folder!r} was not found or ' f'is not accessible. Check the Watch Folder setting.' ) ) # Count unseen messages for informational feedback _, msg_data = imap.search(None, 'UNSEEN') unseen = len(msg_data[0].split()) if msg_data and msg_data[0] else 0 imap.logout() log_action( current_user.id, 'email_ingestion_test', 'system_setting', None, f'host={host} port={port} user={user} folder={folder} unseen={unseen}' ) logger.info( f'[ADMIN EMAIL INGEST] Connection test OK: host={host} user={user} ' f'folder={folder} unseen={unseen} by admin_id={current_user.id}' ) return jsonify( ok=True, message=( f'Connection successful. Folder {folder!r} is accessible ' f'({unseen} unseen message(s)).' ) ) except imaplib.IMAP4.error as exc: logger.warning( f'[ADMIN EMAIL INGEST] Connection test failed: host={host} user={user} ' f'error={exc} by admin_id={current_user.id}' ) return jsonify(ok=False, message=f'IMAP error: {exc}') except OSError as exc: logger.warning( f'[ADMIN EMAIL INGEST] Connection test network error: host={host} ' f'port={port} error={exc} by admin_id={current_user.id}' ) return jsonify(ok=False, message=f'Network error: {exc}') except Exception as exc: logger.error( f'[ADMIN EMAIL INGEST] Connection test unexpected error: {exc}', exc_info=True ) return jsonify(ok=False, message=f'Unexpected error: {exc}') # ── Email Ingestion: Run Now ────────────────────────────────────────────────── @admin_bp.route('/settings/email-ingestion/run-now', methods=['POST']) @login_required @admin_required def email_ingestion_run_now(): """Trigger an immediate ingestion run outside the scheduler cycle. Runs synchronously in the request (the job is fast for small mailboxes). Returns JSON so the UI can surface the outcome without a full page reload. """ from app.models import SystemSetting from app.services.email_ingestion_service import _run_ingestion if SystemSetting.get('email_ingestion_enabled', '0') != '1': return jsonify(ok=False, message='Email ingestion is currently disabled. Enable it first.') log_action( current_user.id, 'email_ingestion_run_now', 'system_setting', None, f'manual run triggered by admin_id={current_user.id}' ) logger.info(f'[ADMIN EMAIL INGEST] Manual run triggered by admin_id={current_user.id}') try: _run_ingestion(current_app._get_current_object()) return jsonify(ok=True, message='Ingestion run completed. Check the activity log for details.') except Exception as exc: logger.error(f'[ADMIN EMAIL INGEST] Manual run error: {exc}', exc_info=True) return jsonify(ok=False, message=f'Run failed: {exc}') @admin_bp.route('/license') @login_required @admin_required def license_page(): from app.services.license_service import get_status, days_until_expiry, TIER_FEATURES status = get_status() days_left = days_until_expiry() tier = status.get('tier', 'community') features = TIER_FEATURES.get(tier, set()) return render_template( 'admin/license.html', status = status, days_left = days_left, features = features, )