371 lines
16 KiB
Python
371 lines
16 KiB
Python
import os
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime
|
|
from flask import (Blueprint, render_template, redirect, url_for,
|
|
flash, request, current_app, send_from_directory, abort)
|
|
from flask_login import login_required, current_user
|
|
from werkzeug.utils import secure_filename
|
|
from app import db
|
|
from app.models import (Ticket, Comment, Attachment, Notification,
|
|
TicketStatus, TicketPriority, TicketCategory,
|
|
User, UserRole, KnowledgeBase)
|
|
from app.services.notification_service import (
|
|
notify_new_ticket, notify_status_change,
|
|
notify_comment_added, notify_assignment,
|
|
)
|
|
from app.services.log_service import log_action, log_ticket_history
|
|
|
|
tickets_bp = Blueprint('tickets', __name__)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ALLOWED_EXT = {'png', 'jpg', 'jpeg', 'gif', 'pdf', 'doc', 'docx', 'txt', 'zip', 'log'}
|
|
|
|
|
|
def allowed_file(filename):
|
|
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXT
|
|
|
|
|
|
def save_attachment(file, ticket_id=None, comment_id=None, uploader_id=None):
|
|
filename = secure_filename(file.filename)
|
|
ext = filename.rsplit('.', 1)[1].lower() if '.' in filename else ''
|
|
stored_name = f"{uuid.uuid4().hex}.{ext}"
|
|
upload_dir = current_app.config['UPLOAD_FOLDER']
|
|
file.save(os.path.join(upload_dir, stored_name))
|
|
att = Attachment(
|
|
ticket_id = ticket_id,
|
|
comment_id = comment_id,
|
|
filename = filename,
|
|
stored_name= stored_name,
|
|
file_size = os.path.getsize(os.path.join(upload_dir, stored_name)),
|
|
mime_type = file.content_type,
|
|
uploaded_by= uploader_id,
|
|
)
|
|
db.session.add(att)
|
|
return att
|
|
|
|
|
|
# ─── Dashboard ────────────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/')
|
|
@tickets_bp.route('/dashboard')
|
|
@login_required
|
|
def dashboard():
|
|
if current_user.is_it_staff:
|
|
open_count = Ticket.query.filter_by(status=TicketStatus.OPEN).count()
|
|
in_progress_count= Ticket.query.filter_by(status=TicketStatus.IN_PROGRESS).count()
|
|
pending_count = Ticket.query.filter_by(status=TicketStatus.PENDING).count()
|
|
resolved_count = Ticket.query.filter_by(status=TicketStatus.RESOLVED).count()
|
|
my_tickets = Ticket.query.filter_by(assigned_to_id=current_user.id).filter(
|
|
Ticket.status.notin_([TicketStatus.CLOSED])
|
|
).order_by(Ticket.created_at.desc()).limit(10).all()
|
|
recent_tickets = Ticket.query.order_by(Ticket.created_at.desc()).limit(15).all()
|
|
return render_template('tickets/dashboard_it.html',
|
|
open_count=open_count, in_progress_count=in_progress_count,
|
|
pending_count=pending_count, resolved_count=resolved_count,
|
|
my_tickets=my_tickets, recent_tickets=recent_tickets,
|
|
)
|
|
else:
|
|
my_tickets = Ticket.query.filter_by(created_by_id=current_user.id).order_by(
|
|
Ticket.created_at.desc()).limit(20).all()
|
|
open_count = sum(1 for t in my_tickets if t.status == TicketStatus.OPEN)
|
|
active_count = sum(1 for t in my_tickets if t.status == TicketStatus.IN_PROGRESS)
|
|
resolved_count = sum(1 for t in my_tickets if t.status == TicketStatus.RESOLVED)
|
|
articles = KnowledgeBase.query.filter_by(is_published=True).order_by(
|
|
KnowledgeBase.view_count.desc()).limit(5).all()
|
|
return render_template('tickets/dashboard_employee.html',
|
|
my_tickets=my_tickets, open_count=open_count,
|
|
active_count=active_count, resolved_count=resolved_count,
|
|
articles=articles,
|
|
)
|
|
|
|
|
|
# ─── Create Ticket ────────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/tickets/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
def create_ticket():
|
|
if request.method == 'POST':
|
|
title = request.form.get('title', '').strip()
|
|
description = request.form.get('description', '').strip()
|
|
category = request.form.get('category', TicketCategory.OTHER)
|
|
priority = request.form.get('priority', TicketPriority.MEDIUM)
|
|
location = request.form.get('location', '').strip()
|
|
asset_tag = request.form.get('asset_tag', '').strip()
|
|
|
|
if not title or not description:
|
|
flash('Title and description are required.', 'danger')
|
|
return render_template('tickets/create.html',
|
|
categories=_categories(), priorities=_priorities())
|
|
|
|
ticket = Ticket(
|
|
title = title,
|
|
description = description,
|
|
category = category,
|
|
priority = priority,
|
|
location = location,
|
|
asset_tag = asset_tag,
|
|
created_by_id = current_user.id,
|
|
status = TicketStatus.OPEN,
|
|
)
|
|
ticket.ticket_number = ticket.generate_ticket_number()
|
|
db.session.add(ticket)
|
|
db.session.flush() # get ticket.id before attachments
|
|
|
|
# Handle file uploads
|
|
for f in request.files.getlist('attachments'):
|
|
if f and f.filename and allowed_file(f.filename):
|
|
save_attachment(f, ticket_id=ticket.id, uploader_id=current_user.id)
|
|
|
|
log_action(current_user.id, 'ticket_create', 'ticket', ticket.id,
|
|
f'ticket_number={ticket.ticket_number} priority={priority} category={category}')
|
|
db.session.commit()
|
|
logger.info(f'[TICKET CREATE] ticket_id={ticket.id} number={ticket.ticket_number} by user_id={current_user.id}')
|
|
notify_new_ticket(ticket)
|
|
flash(f'Ticket {ticket.ticket_number} created successfully!', 'success')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket.id))
|
|
|
|
return render_template('tickets/create.html',
|
|
categories=_categories(), priorities=_priorities())
|
|
|
|
|
|
# ─── Ticket List ──────────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/tickets')
|
|
@login_required
|
|
def ticket_list():
|
|
page = request.args.get('page', 1, type=int)
|
|
status = request.args.get('status', '')
|
|
priority = request.args.get('priority', '')
|
|
category = request.args.get('category', '')
|
|
search = request.args.get('q', '')
|
|
|
|
query = Ticket.query
|
|
if not current_user.is_it_staff:
|
|
query = query.filter_by(created_by_id=current_user.id)
|
|
|
|
if status:
|
|
query = query.filter_by(status=status)
|
|
if priority:
|
|
query = query.filter_by(priority=priority)
|
|
if category:
|
|
query = query.filter_by(category=category)
|
|
if search:
|
|
query = query.filter(
|
|
Ticket.title.ilike(f'%{search}%') |
|
|
Ticket.ticket_number.ilike(f'%{search}%') |
|
|
Ticket.description.ilike(f'%{search}%')
|
|
)
|
|
|
|
tickets = query.order_by(Ticket.created_at.desc()).paginate(page=page, per_page=20)
|
|
return render_template('tickets/list.html',
|
|
tickets=tickets, status=status, priority=priority,
|
|
category=category, search=search,
|
|
statuses=_statuses(), priorities=_priorities(), categories=_categories(),
|
|
)
|
|
|
|
|
|
# ─── Ticket Detail ────────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/tickets/<int:ticket_id>', methods=['GET', 'POST'])
|
|
@login_required
|
|
def ticket_detail(ticket_id):
|
|
ticket = Ticket.query.get_or_404(ticket_id)
|
|
|
|
# Employees can only view their own tickets
|
|
if not current_user.is_it_staff and ticket.created_by_id != current_user.id:
|
|
abort(403)
|
|
|
|
if request.method == 'POST':
|
|
body = request.form.get('body', '').strip()
|
|
is_internal = bool(request.form.get('is_internal')) and current_user.is_it_staff
|
|
|
|
if not body:
|
|
flash('Comment cannot be empty.', 'danger')
|
|
else:
|
|
comment = Comment(
|
|
ticket_id = ticket.id,
|
|
author_id = current_user.id,
|
|
body = body,
|
|
is_internal= is_internal,
|
|
)
|
|
db.session.add(comment)
|
|
db.session.flush()
|
|
|
|
for f in request.files.getlist('attachments'):
|
|
if f and f.filename and allowed_file(f.filename):
|
|
save_attachment(f, ticket_id=ticket.id,
|
|
comment_id=comment.id, uploader_id=current_user.id)
|
|
|
|
log_action(current_user.id, 'comment_create', 'comment', comment.id,
|
|
f'ticket_id={ticket.id} internal={is_internal}')
|
|
db.session.commit()
|
|
logger.info(f'[COMMENT CREATE] comment_id={comment.id} ticket_id={ticket.id} by user_id={current_user.id}')
|
|
notify_comment_added(comment)
|
|
flash('Comment added.', 'success')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket.id))
|
|
|
|
comments = Comment.query.filter_by(ticket_id=ticket.id)
|
|
if not current_user.is_it_staff:
|
|
comments = comments.filter_by(is_internal=False)
|
|
comments = comments.order_by(Comment.created_at.asc()).all()
|
|
|
|
it_staff = User.query.filter(
|
|
User.role.in_([UserRole.IT_STAFF, UserRole.ADMIN]),
|
|
User.is_active == True,
|
|
).all() if current_user.is_it_staff else []
|
|
|
|
history = ticket.history.order_by('changed_at').all()
|
|
|
|
return render_template('tickets/detail.html',
|
|
ticket=ticket, comments=comments,
|
|
it_staff=it_staff, history=history,
|
|
statuses=_statuses(), priorities=_priorities(),
|
|
)
|
|
|
|
|
|
# ─── Update Ticket (IT Only) ──────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/tickets/<int:ticket_id>/update', methods=['POST'])
|
|
@login_required
|
|
def update_ticket(ticket_id):
|
|
if not current_user.is_it_staff:
|
|
abort(403)
|
|
|
|
ticket = Ticket.query.get_or_404(ticket_id)
|
|
old_status = ticket.status
|
|
old_priority = ticket.priority
|
|
old_assigned = ticket.assigned_to_id
|
|
|
|
new_status = request.form.get('status', ticket.status)
|
|
new_priority = request.form.get('priority', ticket.priority)
|
|
new_assigned = request.form.get('assigned_to_id', type=int)
|
|
internal_notes= request.form.get('internal_notes', ticket.internal_notes)
|
|
resolution = request.form.get('resolution_notes', ticket.resolution_notes)
|
|
due_date_str = request.form.get('due_date', '')
|
|
|
|
changes = []
|
|
|
|
if new_status != old_status:
|
|
ticket.status = new_status
|
|
log_ticket_history(ticket, 'status', old_status, new_status, current_user.id)
|
|
changes.append(f'status: {old_status} → {new_status}')
|
|
if new_status == TicketStatus.RESOLVED:
|
|
ticket.resolved_at = datetime.utcnow()
|
|
elif new_status == TicketStatus.CLOSED:
|
|
ticket.closed_at = datetime.utcnow()
|
|
|
|
if new_priority != old_priority:
|
|
ticket.priority = new_priority
|
|
log_ticket_history(ticket, 'priority', old_priority, new_priority, current_user.id)
|
|
changes.append(f'priority: {old_priority} → {new_priority}')
|
|
|
|
if new_assigned != old_assigned:
|
|
ticket.assigned_to_id = new_assigned
|
|
log_ticket_history(ticket, 'assigned_to', str(old_assigned), str(new_assigned), current_user.id)
|
|
changes.append(f'assigned_to: {old_assigned} → {new_assigned}')
|
|
notify_assignment(ticket, current_user)
|
|
|
|
ticket.internal_notes = internal_notes
|
|
ticket.resolution_notes = resolution
|
|
|
|
if due_date_str:
|
|
try:
|
|
ticket.due_date = datetime.strptime(due_date_str, '%Y-%m-%d')
|
|
except ValueError:
|
|
pass
|
|
|
|
log_action(current_user.id, 'ticket_update', 'ticket', ticket.id,
|
|
f'changes=[{"; ".join(changes)}]')
|
|
db.session.commit()
|
|
logger.info(f'[TICKET UPDATE] ticket_id={ticket.id} changes={changes} by user_id={current_user.id}')
|
|
|
|
if new_status != old_status:
|
|
notify_status_change(ticket, old_status, current_user)
|
|
|
|
flash('Ticket updated successfully.', 'success')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket.id))
|
|
|
|
|
|
# ─── Delete Comment (IT Only) ─────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/comments/<int:comment_id>/delete', methods=['POST'])
|
|
@login_required
|
|
def delete_comment(comment_id):
|
|
comment = Comment.query.get_or_404(comment_id)
|
|
if not current_user.is_it_staff and comment.author_id != current_user.id:
|
|
abort(403)
|
|
ticket_id = comment.ticket_id
|
|
log_action(current_user.id, 'comment_delete', 'comment', comment.id,
|
|
f'ticket_id={ticket_id}')
|
|
logger.info(f'[COMMENT DELETE] comment_id={comment.id} ticket_id={ticket_id} by user_id={current_user.id}')
|
|
db.session.delete(comment)
|
|
db.session.commit()
|
|
flash('Comment deleted.', 'success')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
|
|
# ─── Attachment Download ──────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/attachments/<int:att_id>')
|
|
@login_required
|
|
def download_attachment(att_id):
|
|
att = Attachment.query.get_or_404(att_id)
|
|
upload_dir = current_app.config['UPLOAD_FOLDER']
|
|
return send_from_directory(upload_dir, att.stored_name, as_attachment=True,
|
|
download_name=att.filename)
|
|
|
|
|
|
# ─── Notifications ────────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/notifications')
|
|
@login_required
|
|
def notifications():
|
|
notifs = Notification.query.filter_by(user_id=current_user.id).order_by(
|
|
Notification.created_at.desc()).paginate(page=request.args.get('page', 1, type=int), per_page=30)
|
|
return render_template('tickets/notifications.html', notifs=notifs)
|
|
|
|
|
|
@tickets_bp.route('/notifications/mark-read', methods=['POST'])
|
|
@login_required
|
|
def mark_notifications_read():
|
|
Notification.query.filter_by(user_id=current_user.id, is_read=False).update({'is_read': True})
|
|
db.session.commit()
|
|
return redirect(request.referrer or url_for('tickets.notifications'))
|
|
|
|
|
|
# ─── Knowledge Base ───────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/kb')
|
|
@login_required
|
|
def knowledge_base():
|
|
articles = KnowledgeBase.query.filter_by(is_published=True).order_by(
|
|
KnowledgeBase.view_count.desc()).all()
|
|
return render_template('tickets/knowledge_base.html', articles=articles)
|
|
|
|
|
|
@tickets_bp.route('/kb/<int:article_id>')
|
|
@login_required
|
|
def kb_article(article_id):
|
|
article = KnowledgeBase.query.get_or_404(article_id)
|
|
article.view_count += 1
|
|
db.session.commit()
|
|
return render_template('tickets/kb_article.html', article=article)
|
|
|
|
|
|
# ─── Helpers ─────────────────────────────────────────────────────────────────
|
|
|
|
def _statuses():
|
|
return [TicketStatus.OPEN, TicketStatus.IN_PROGRESS,
|
|
TicketStatus.PENDING, TicketStatus.RESOLVED, TicketStatus.CLOSED]
|
|
|
|
def _priorities():
|
|
return [TicketPriority.LOW, TicketPriority.MEDIUM,
|
|
TicketPriority.HIGH, TicketPriority.CRITICAL]
|
|
|
|
def _categories():
|
|
return [TicketCategory.HARDWARE, TicketCategory.SOFTWARE,
|
|
TicketCategory.NETWORK, TicketCategory.ACCESS,
|
|
TicketCategory.EMAIL, TicketCategory.PRINTER,
|
|
TicketCategory.PHONE, TicketCategory.SECURITY, TicketCategory.OTHER]
|