1225 lines
56 KiB
Python
1225 lines
56 KiB
Python
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)
|
||
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/<int:user_id>/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)
|
||
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/<int:user_id>/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'<[^>]+>', '', 'g')
|
||
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)
|
||
return render_template('admin/tickets.html', tickets=tickets,
|
||
status=status, priority=priority, assigned=assigned,
|
||
search=search)
|
||
|
||
|
||
# ─── 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": "<url>"} 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/<string:stored_name>')
|
||
@login_required
|
||
def kb_serve_file(stored_name):
|
||
"""Serve a KB attachment file. Login required — no public access.
|
||
|
||
Security note
|
||
-------------
|
||
The <string:> converter is used deliberately instead of <path:>.
|
||
The <path:> 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. <string:> 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/<int:article_id>/attachments/<int:att_id>/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()
|
||
# 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/<int:article_id>/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/<int:article_id>/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/<int:article_id>/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/<int:article_id>/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'<[^>]+>', '', 'g')
|
||
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/<int:item_id>/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/<int:item_id>/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')
|
||
if action not in valid_actions:
|
||
flash('Invalid action.', '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
|
||
|
||
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 == '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',
|
||
'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),
|
||
))
|
||
|
||
|
||
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 == '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
|
||
new_pw = request.form.get('email_ingestion_password', '').strip()
|
||
if new_pw:
|
||
fields['email_ingestion_password'] = 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)
|
||
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', ''),
|
||
'password': 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,
|
||
branding_settings=branding_settings,
|
||
email_settings=email_settings,
|
||
current_tz=current_tz,
|
||
common_timezones=common_timezones) |