778 lines
34 KiB
Python
778 lines
34 KiB
Python
import logging
|
||
import os
|
||
import uuid
|
||
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
|
||
from flask_login import login_required, current_user
|
||
from werkzeug.utils import secure_filename
|
||
import bleach
|
||
from app import db
|
||
from app.models import (User, Ticket, Comment, ActivityLog, KnowledgeBase,
|
||
KBAttachment, UserRole, TicketStatus)
|
||
from app.services.log_service import log_action
|
||
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 = User.query.get_or_404(user_id)
|
||
if request.method == 'POST':
|
||
old_role = user.role
|
||
user.full_name = request.form.get('full_name', user.full_name).strip()
|
||
user.department= request.form.get('department', user.department).strip()
|
||
user.phone = request.form.get('phone', user.phone or '').strip()
|
||
user.role = request.form.get('role', user.role)
|
||
user.is_active = bool(request.form.get('is_active'))
|
||
new_pw = request.form.get('new_password', '')
|
||
if new_pw:
|
||
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 = User.query.get_or_404(user_id)
|
||
if user.id == current_user.id:
|
||
flash('You cannot delete your own account.', 'danger')
|
||
return redirect(url_for('admin.users'))
|
||
user.is_active = False
|
||
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', '')
|
||
|
||
q = Ticket.query
|
||
if status: q = q.filter_by(status=status)
|
||
if priority: q = q.filter_by(priority=priority)
|
||
if assigned == 'me': q = q.filter_by(assigned_to_id=current_user.id)
|
||
elif assigned == 'unassigned': q = q.filter_by(assigned_to_id=None)
|
||
|
||
tickets = q.order_by(Ticket.created_at.desc()).paginate(page=page, per_page=25)
|
||
return render_template('admin/tickets.html', tickets=tickets,
|
||
status=status, priority=priority, assigned=assigned)
|
||
|
||
|
||
# ─── Knowledge Base Management ────────────────────────────────────────────────
|
||
|
||
@admin_bp.route('/kb')
|
||
@login_required
|
||
@it_required
|
||
def kb_list():
|
||
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.
|
||
"""
|
||
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()
|
||
logger.info(f'[KB ATTACHMENT DELETE] att_id={att.id} article_id={article_id} completed')
|
||
# Return JSON so the edit page can remove the row without a full reload
|
||
return jsonify({'ok': True, 'att_id': att.id})
|
||
|
||
|
||
|
||
@admin_bp.route('/kb/<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 = KnowledgeBase.query.get_or_404(article_id)
|
||
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 = KnowledgeBase.query.get_or_404(article_id)
|
||
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 = KnowledgeBase.query.get_or_404(article_id)
|
||
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,
|
||
)
|
||
|
||
|
||
def _roles():
|
||
return [UserRole.EMPLOYEE, UserRole.IT_STAFF, UserRole.ADMIN]
|
||
|
||
|
||
@admin_bp.route('/settings', methods=['GET', 'POST'])
|
||
@admin_required
|
||
def settings():
|
||
from app.models import SystemSetting
|
||
|
||
if request.method == 'POST':
|
||
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} '
|
||
f'by admin_id={current_user.id} email={current_user.email}'
|
||
)
|
||
flash(f'User registration has been {state_label}.', 'success')
|
||
return redirect(url_for('admin.settings'))
|
||
|
||
registration_enabled = SystemSetting.get_bool('registration_enabled', default=True)
|
||
return render_template('admin/settings.html', registration_enabled=registration_enabled) |