1099 lines
49 KiB
Python
1099 lines
49 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, TicketLink, CannedResponse,
|
|
KBFeedback, TicketTemplate, TicketSatisfaction,
|
|
TicketWatcher, TimeEntry)
|
|
from app.services.notification_service import (
|
|
notify_new_ticket, notify_status_change,
|
|
notify_comment_added, notify_assignment,
|
|
send_satisfaction_survey, notify_watchers,
|
|
)
|
|
from app.services.log_service import log_action, log_ticket_history
|
|
from app.services.sla_service import clear_sla_notification
|
|
from app.services.validation_service import validate_file, render_comment_body
|
|
|
|
tickets_bp = Blueprint('tickets', __name__)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Allowed extensions for ticket and comment attachments.
|
|
# validate_file() uses this set for both extension and magic-byte checks.
|
|
ALLOWED_EXT = {'png', 'jpg', 'jpeg', 'gif', 'pdf', 'doc', 'docx', 'txt', 'zip', 'log'}
|
|
|
|
|
|
def _strip_html(text: str) -> str:
|
|
"""Remove HTML tags from *text* for plain-text search matching.
|
|
|
|
Comment bodies are stored as sanitized HTML (rendered at write time via
|
|
render_comment_body). Searching with ilike('%term%') against raw HTML
|
|
produces two problems:
|
|
1. A search for 'bold' misses '<strong>bold</strong>' in the stored body.
|
|
2. HTML tag names ('strong', 'pre') can accidentally match search terms.
|
|
Stripping tags before comparison gives consistent, tag-agnostic results.
|
|
|
|
Uses a simple regex rather than a full HTML parser — sufficient for the
|
|
sanitized subset of HTML that bleach allows in comment bodies.
|
|
"""
|
|
import re
|
|
return re.sub(r'<[^>]+>', '', text)
|
|
|
|
|
|
def _resolve_mime_type(att):
|
|
"""Return a reliable MIME type for an attachment.
|
|
|
|
Browsers sometimes send 'application/octet-stream' for images on upload,
|
|
and older attachments may have NULL mime_type. Fall back to an
|
|
extension-based lookup so images are always served inline correctly.
|
|
"""
|
|
stored = (att.mime_type or '').lower().strip()
|
|
if stored.startswith('image/'):
|
|
return stored
|
|
ext_map = {
|
|
'png': 'image/png', 'jpg': 'image/jpeg', 'jpeg': 'image/jpeg',
|
|
'gif': 'image/gif', 'webp': 'image/webp', 'svg': 'image/svg+xml',
|
|
'pdf': 'application/pdf',
|
|
'doc': 'application/msword',
|
|
'docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
|
'txt': 'text/plain', 'log': 'text/plain',
|
|
'zip': 'application/zip',
|
|
}
|
|
ext = att.filename.rsplit('.', 1)[-1].lower() if '.' in att.filename else ''
|
|
return ext_map.get(ext, stored or 'application/octet-stream')
|
|
|
|
|
|
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
|
|
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, now=datetime.utcnow(),
|
|
)
|
|
|
|
|
|
# ─── 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,
|
|
due_date = _sla_due_date(priority), # auto-set SLA deadline
|
|
)
|
|
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:
|
|
file_error = validate_file(f, ALLOWED_EXT)
|
|
if file_error:
|
|
logger.warning(f'[TICKET UPLOAD REJECTED] {file_error} filename="{f.filename}" user_id={current_user.id}')
|
|
continue
|
|
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))
|
|
|
|
templates = TicketTemplate.query.filter_by(is_active=True).order_by(
|
|
TicketTemplate.sort_order, TicketTemplate.name
|
|
).all()
|
|
return render_template('tickets/create.html',
|
|
categories=_categories(), priorities=_priorities(),
|
|
templates=templates)
|
|
|
|
|
|
# ─── Create Ticket on Behalf of Employee (IT Staff Only) ─────────────────────
|
|
|
|
@tickets_bp.route('/tickets/behalf', methods=['GET', 'POST'])
|
|
@login_required
|
|
def create_ticket_behalf():
|
|
"""Allow IT staff to create a ticket on behalf of an employee.
|
|
|
|
Use case: an employee calls or emails to report an issue and cannot or
|
|
does not create a ticket themselves. IT staff completes the form,
|
|
selecting the employee from a searchable dropdown.
|
|
|
|
Data model
|
|
----------
|
|
ticket.created_by_id = employee's user ID (ticket shows as theirs)
|
|
ticket.created_by_staff_id = IT staff's user ID (audit trail)
|
|
|
|
The employee sees this ticket in their own dashboard and receives the
|
|
same new-ticket confirmation notification they would if self-submitted.
|
|
"""
|
|
if not current_user.is_it_staff:
|
|
abort(403)
|
|
|
|
employees = User.query.filter(
|
|
User.is_active == True,
|
|
).order_by(User.full_name).all()
|
|
|
|
if request.method == 'POST':
|
|
employee_id = request.form.get('employee_id', type=int)
|
|
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()
|
|
|
|
# Validate employee selection
|
|
employee = db.session.get(User, employee_id) if employee_id else None
|
|
if not employee or not employee.is_active:
|
|
flash('Please select a valid active employee.', 'danger')
|
|
return render_template('tickets/create_behalf.html',
|
|
employees=employees,
|
|
categories=_categories(),
|
|
priorities=_priorities())
|
|
|
|
if not title or not description:
|
|
flash('Title and description are required.', 'danger')
|
|
return render_template('tickets/create_behalf.html',
|
|
employees=employees,
|
|
categories=_categories(),
|
|
priorities=_priorities(),
|
|
selected_employee_id=employee_id)
|
|
|
|
ticket = Ticket(
|
|
title = title,
|
|
description = description,
|
|
category = category,
|
|
priority = priority,
|
|
location = location,
|
|
asset_tag = asset_tag,
|
|
created_by_id = employee.id, # ticket belongs to the employee
|
|
created_by_staff_id = current_user.id, # IT staff who filed it
|
|
status = TicketStatus.OPEN,
|
|
due_date = _sla_due_date(priority), # auto-set SLA deadline
|
|
)
|
|
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:
|
|
file_error = validate_file(f, ALLOWED_EXT)
|
|
if file_error:
|
|
logger.warning(
|
|
f'[BEHALF UPLOAD REJECTED] {file_error} '
|
|
f'filename="{f.filename}" staff_id={current_user.id}'
|
|
)
|
|
continue
|
|
save_attachment(f, ticket_id=ticket.id, uploader_id=current_user.id)
|
|
|
|
log_action(
|
|
current_user.id, 'ticket_create_behalf', 'ticket', ticket.id,
|
|
f'ticket_number={ticket.ticket_number} on_behalf_of=user_id:{employee.id} '
|
|
f'({employee.full_name}) priority={priority} category={category}'
|
|
)
|
|
db.session.commit()
|
|
logger.info(
|
|
f'[TICKET CREATE BEHALF] ticket_id={ticket.id} '
|
|
f'number={ticket.ticket_number} '
|
|
f'employee_id={employee.id} staff_id={current_user.id}'
|
|
)
|
|
notify_new_ticket(ticket)
|
|
flash(
|
|
f'Ticket {ticket.ticket_number} created on behalf of {employee.full_name}.',
|
|
'success'
|
|
)
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket.id))
|
|
|
|
return render_template('tickets/create_behalf.html',
|
|
employees=employees,
|
|
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:
|
|
# Extend search to cover comments and assignee name via outer joins.
|
|
# distinct() prevents duplicate ticket rows when multiple comments match.
|
|
# Comment bodies are stored as sanitized HTML — use REGEXP_REPLACE to
|
|
# strip tags at the SQL level before matching so 'bold' finds
|
|
# '<strong>bold</strong>' and HTML tag names don't pollute results.
|
|
from app.models import Comment
|
|
from sqlalchemy import func
|
|
assignee_alias = db.aliased(User)
|
|
stripped_body = func.regexp_replace(Comment.body, r'<[^>]+>', '')
|
|
query = (
|
|
query
|
|
.outerjoin(Comment, Comment.ticket_id == Ticket.id)
|
|
.outerjoin(assignee_alias, assignee_alias.id == Ticket.assigned_to_id)
|
|
.filter(
|
|
Ticket.title.ilike(f'%{search}%') |
|
|
Ticket.ticket_number.ilike(f'%{search}%') |
|
|
Ticket.description.ilike(f'%{search}%') |
|
|
stripped_body.ilike(f'%{search}%') |
|
|
assignee_alias.full_name.ilike(f'%{search}%')
|
|
)
|
|
.distinct()
|
|
)
|
|
|
|
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 = db.session.get(Ticket, ticket_id) or abort(404)
|
|
|
|
# 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 = render_comment_body(body),
|
|
is_internal= is_internal,
|
|
)
|
|
db.session.add(comment)
|
|
db.session.flush()
|
|
|
|
# ── Attachment limits (mirrors client-side constants in detail.html)
|
|
MAX_COMMENT_FILES = 5
|
|
MAX_COMMENT_BYTES = 25 * 1024 * 1024 # 25 MB total per comment
|
|
|
|
uploaded_files = [f for f in request.files.getlist('attachments') if f and f.filename]
|
|
if len(uploaded_files) > MAX_COMMENT_FILES:
|
|
db.session.rollback()
|
|
logger.warning(f'[COMMENT UPLOAD REJECTED] Too many files ({len(uploaded_files)}) ticket_id={ticket.id} user_id={current_user.id}')
|
|
flash(f'Too many attachments — maximum {MAX_COMMENT_FILES} files per comment.', 'danger')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket.id))
|
|
|
|
total_bytes = 0
|
|
for f in uploaded_files:
|
|
f.stream.seek(0, 2)
|
|
total_bytes += f.stream.tell()
|
|
f.stream.seek(0)
|
|
if total_bytes > MAX_COMMENT_BYTES:
|
|
db.session.rollback()
|
|
logger.warning(f'[COMMENT UPLOAD REJECTED] Total size {total_bytes} exceeds limit ticket_id={ticket.id} user_id={current_user.id}')
|
|
flash(f'Total attachment size exceeds the 25 MB limit per comment.', 'danger')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket.id))
|
|
|
|
for f in uploaded_files:
|
|
file_error = validate_file(f, ALLOWED_EXT)
|
|
if file_error:
|
|
logger.warning(f'[COMMENT UPLOAD REJECTED] {file_error} filename="{f.filename}" user_id={current_user.id}')
|
|
continue
|
|
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()
|
|
|
|
# Collect all links for this ticket from both directions
|
|
links_src = ticket.links_as_source.all()
|
|
links_tgt = ticket.links_as_target.all()
|
|
# Build a unified list of (link_obj, other_ticket) tuples for the template
|
|
linked_tickets = (
|
|
[(lnk, lnk.linked_ticket) for lnk in links_src] +
|
|
[(lnk, lnk.ticket) for lnk in links_tgt]
|
|
)
|
|
|
|
# Canned responses for IT staff comment form
|
|
canned_responses = []
|
|
if current_user.is_it_staff:
|
|
canned_responses = CannedResponse.query.filter_by(is_active=True).order_by(
|
|
CannedResponse.category, CannedResponse.title
|
|
).all()
|
|
|
|
is_watching = TicketWatcher.query.filter_by(
|
|
ticket_id=ticket_id, user_id=current_user.id
|
|
).first() is not None
|
|
total_minutes = sum(e.minutes for e in ticket.time_entries.all())
|
|
|
|
return render_template('tickets/detail.html',
|
|
ticket=ticket, comments=comments,
|
|
it_staff=it_staff, history=history,
|
|
statuses=_statuses(), priorities=_priorities(),
|
|
linked_tickets=linked_tickets,
|
|
canned_responses=canned_responses,
|
|
now=datetime.utcnow(),
|
|
is_watching=is_watching,
|
|
total_minutes=total_minutes,
|
|
time_entries=ticket.time_entries.order_by(TimeEntry.logged_at.desc()).limit(10).all(),
|
|
)
|
|
|
|
|
|
# ─── 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 = db.session.get(Ticket, ticket_id) or abort(404)
|
|
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()
|
|
clear_sla_notification(ticket.id) # remove from breach suppression list
|
|
elif new_status == TicketStatus.CLOSED:
|
|
ticket.closed_at = datetime.utcnow()
|
|
clear_sla_notification(ticket.id) # remove from breach suppression list
|
|
elif new_status == TicketStatus.OPEN:
|
|
clear_sla_notification(ticket.id) # re-opened — allow fresh breach alerts
|
|
|
|
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:
|
|
# Resolve user IDs to full names for human-readable history entries.
|
|
# None means unassigned.
|
|
def _user_label(uid):
|
|
if uid is None:
|
|
return 'Unassigned'
|
|
u = db.session.get(User, uid)
|
|
return u.full_name if u else f'User #{uid}'
|
|
|
|
ticket.assigned_to_id = new_assigned
|
|
log_ticket_history(ticket, 'assigned_to',
|
|
_user_label(old_assigned),
|
|
_user_label(new_assigned),
|
|
current_user.id)
|
|
changes.append(f'assigned_to: {old_assigned} → {new_assigned}')
|
|
# notify_assignment is called AFTER commit below — see Fix #13.
|
|
|
|
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}')
|
|
|
|
# Both notification calls are placed after commit so that create_notification's
|
|
# independent commit never races against an uncommitted ticket state. If the
|
|
# parent commit above had failed, neither notification would be sent — which
|
|
# is the correct behaviour (no notification for a change that did not persist).
|
|
if new_assigned != old_assigned:
|
|
notify_assignment(ticket, current_user)
|
|
|
|
if new_status != old_status:
|
|
notify_status_change(ticket, old_status, current_user)
|
|
if new_status == TicketStatus.RESOLVED:
|
|
send_satisfaction_survey(ticket)
|
|
|
|
if changes:
|
|
notify_watchers(
|
|
ticket,
|
|
event_title = f'Ticket {ticket.ticket_number} updated',
|
|
event_message = '; '.join(changes),
|
|
exclude_user_id = current_user.id,
|
|
)
|
|
|
|
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
|
from flask import jsonify
|
|
return jsonify({'ok': True, 'changes': changes,
|
|
'status': ticket.status, 'priority': ticket.priority,
|
|
'assigned_to_id': ticket.assigned_to_id})
|
|
|
|
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 = db.session.get(Comment, comment_id) or abort(404)
|
|
ticket_id = comment.ticket_id
|
|
|
|
# Authorization: IT staff may always delete any comment.
|
|
# Employees may only delete their own comments, and only while the
|
|
# ticket is still open or in-progress. Allowing deletion on resolved
|
|
# or closed tickets would silently alter the historical record of a
|
|
# completed support interaction.
|
|
if not current_user.is_it_staff:
|
|
if comment.author_id != current_user.id:
|
|
abort(403)
|
|
ticket = db.session.get(Ticket, ticket_id) or abort(404)
|
|
if ticket.status in (TicketStatus.RESOLVED, TicketStatus.CLOSED):
|
|
logger.warning(
|
|
f'[COMMENT DELETE BLOCKED] comment_id={comment.id} '
|
|
f'ticket_id={ticket_id} status={ticket.status} '
|
|
f'user_id={current_user.id} — ticket is {ticket.status}'
|
|
)
|
|
flash('Comments cannot be deleted on resolved or closed tickets.', 'warning')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
# Include a truncated snapshot of the body in the audit log so the
|
|
# content is recoverable from logs even after the DB row is gone.
|
|
body_snapshot = comment.body[:200].replace('\n', ' ')
|
|
log_action(current_user.id, 'comment_delete', 'comment', comment.id,
|
|
f'ticket_id={ticket_id} body_snapshot="{body_snapshot}"')
|
|
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 = db.session.get(Attachment, att_id) or abort(404)
|
|
# Authorization: employees may only download attachments belonging to
|
|
# their own tickets. IT staff have unrestricted access across all tickets.
|
|
# att.ticket_id is the authoritative link — comment attachments also carry
|
|
# the parent ticket_id, so this check covers both ticket and comment files.
|
|
if not current_user.is_it_staff:
|
|
ticket = db.session.get(Ticket, att.ticket_id) or abort(404)
|
|
if ticket.created_by_id != current_user.id:
|
|
logger.warning(
|
|
f'[ATTACHMENT ACCESS DENIED] att_id={att_id} ticket_id={att.ticket_id} '
|
|
f'user_id={current_user.id}'
|
|
)
|
|
abort(403)
|
|
upload_dir = current_app.config['UPLOAD_FOLDER']
|
|
mime = _resolve_mime_type(att)
|
|
is_image = mime.startswith('image/')
|
|
return send_from_directory(
|
|
upload_dir,
|
|
att.stored_name,
|
|
as_attachment = not is_image,
|
|
download_name = att.filename,
|
|
mimetype = mime,
|
|
)
|
|
|
|
|
|
# ─── 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():
|
|
q = request.args.get('q', '').strip()
|
|
category = request.args.get('category', '')
|
|
sort = request.args.get('sort', 'popular') # 'popular' | 'newest'
|
|
|
|
query = KnowledgeBase.query.filter_by(is_published=True)
|
|
|
|
if q:
|
|
query = query.filter(
|
|
KnowledgeBase.title.ilike(f'%{q}%') |
|
|
KnowledgeBase.tags.ilike(f'%{q}%')
|
|
)
|
|
if category:
|
|
query = query.filter_by(category=category)
|
|
|
|
if sort == 'newest':
|
|
query = query.order_by(KnowledgeBase.updated_at.desc())
|
|
else:
|
|
query = query.order_by(KnowledgeBase.view_count.desc())
|
|
|
|
articles = query.all()
|
|
categories = sorted({a.category for a in KnowledgeBase.query.filter_by(is_published=True).with_entities(KnowledgeBase.category).distinct() if a.category})
|
|
|
|
return render_template('tickets/knowledge_base.html',
|
|
articles = articles,
|
|
categories = categories,
|
|
q = q,
|
|
sel_category = category,
|
|
sort = sort,
|
|
)
|
|
|
|
|
|
@tickets_bp.route('/kb/<int:article_id>')
|
|
@login_required
|
|
def kb_article(article_id):
|
|
article = db.session.get(KnowledgeBase, article_id) or abort(404)
|
|
# Increment view_count atomically at the SQL level. A Python-level
|
|
# read-modify-write (article.view_count += 1) is not safe under concurrent
|
|
# requests: two simultaneous reads both see the same value and one
|
|
# increment is silently lost. The SQL expression KnowledgeBase.view_count + 1
|
|
# delegates the addition to the database, which serialises it correctly.
|
|
from sqlalchemy import update as sa_update
|
|
db.session.execute(
|
|
sa_update(KnowledgeBase)
|
|
.where(KnowledgeBase.id == article_id)
|
|
.values(view_count=KnowledgeBase.view_count + 1)
|
|
)
|
|
db.session.commit()
|
|
# Re-fetch so the template receives the post-increment value.
|
|
db.session.refresh(article)
|
|
# Load the current user's vote (if any) so the feedback widget shows state
|
|
user_feedback = KBFeedback.query.filter_by(
|
|
article_id=article.id, user_id=current_user.id
|
|
).first()
|
|
# Aggregate counts for the widget
|
|
helpful_count = KBFeedback.query.filter_by(article_id=article.id, is_helpful=True).count()
|
|
not_helpful_count = KBFeedback.query.filter_by(article_id=article.id, is_helpful=False).count()
|
|
return render_template('tickets/kb_article.html',
|
|
article=article,
|
|
user_feedback=user_feedback,
|
|
helpful_count=helpful_count,
|
|
not_helpful_count=not_helpful_count,
|
|
)
|
|
|
|
|
|
# ─── Ticket Link / Unlink ────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/tickets/<int:ticket_id>/link', methods=['POST'])
|
|
@login_required
|
|
def link_ticket(ticket_id):
|
|
"""Create a bidirectional link between two tickets (IT staff only)."""
|
|
if not current_user.is_it_staff:
|
|
abort(403)
|
|
|
|
ticket = db.session.get(Ticket, ticket_id) or abort(404)
|
|
raw_input = request.form.get('linked_ticket_id', '').strip()
|
|
link_type = request.form.get('link_type', 'related')
|
|
|
|
if not raw_input:
|
|
flash('Please search for and select a ticket to link.', 'danger')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
# Resolve by ticket_number (e.g. TKT-20260406-0001) or numeric DB id
|
|
if raw_input.upper().startswith('TKT-'):
|
|
other = Ticket.query.filter_by(ticket_number=raw_input.upper()).first()
|
|
else:
|
|
try:
|
|
other = db.session.get(Ticket, int(raw_input))
|
|
except ValueError:
|
|
other = Ticket.query.filter_by(ticket_number=raw_input.upper()).first()
|
|
|
|
if not other:
|
|
flash(f'Ticket "{raw_input}" not found.', 'danger')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
other_id = other.id
|
|
|
|
if other_id == ticket_id:
|
|
flash('A ticket cannot be linked to itself.', 'danger')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
# Enforce canonical ordering (smaller id first) for the uniqueness constraint
|
|
a, b = sorted([ticket_id, other_id])
|
|
existing = TicketLink.query.filter_by(ticket_id=a, linked_ticket_id=b).first()
|
|
if existing:
|
|
flash(f'These tickets are already linked.', 'warning')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
if link_type not in ('related', 'duplicate', 'follow_up'):
|
|
link_type = 'related'
|
|
|
|
link = TicketLink(
|
|
ticket_id = a,
|
|
linked_ticket_id = b,
|
|
link_type = link_type,
|
|
created_by = current_user.id,
|
|
)
|
|
db.session.add(link)
|
|
log_action(current_user.id, 'ticket_link_create', 'ticket', ticket_id,
|
|
f'linked_to={other_id} type={link_type}')
|
|
db.session.commit()
|
|
logger.info(f'[TICKET LINK] ticket_id={a} linked_ticket_id={b} '
|
|
f'type={link_type} by user_id={current_user.id}')
|
|
flash(f'Linked to {other.ticket_number} ({link_type.replace("_", " ")}).', 'success')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
|
|
@tickets_bp.route('/tickets/<int:ticket_id>/unlink/<int:link_id>', methods=['POST'])
|
|
@login_required
|
|
def unlink_ticket(ticket_id, link_id):
|
|
"""Remove a ticket link (IT staff only)."""
|
|
if not current_user.is_it_staff:
|
|
abort(403)
|
|
link = db.session.get(TicketLink, link_id) or abort(404)
|
|
if link.ticket_id != ticket_id and link.linked_ticket_id != ticket_id:
|
|
abort(403)
|
|
log_action(current_user.id, 'ticket_link_delete', 'ticket', ticket_id,
|
|
f'link_id={link_id} removed')
|
|
logger.info(f'[TICKET UNLINK] link_id={link_id} ticket_id={ticket_id} '
|
|
f'by user_id={current_user.id}')
|
|
db.session.delete(link)
|
|
db.session.commit()
|
|
flash('Link removed.', 'success')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
|
|
# ─── Ticket Re-open ───────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/tickets/<int:ticket_id>/reopen', methods=['POST'])
|
|
@login_required
|
|
def reopen_ticket(ticket_id):
|
|
"""Allow the ticket creator to re-open a resolved or closed ticket.
|
|
|
|
A re-open creates a comment with the employee's explanation, resets the
|
|
ticket status to Open, and notifies all IT staff so the ticket resurfaces
|
|
in the queue without being lost.
|
|
"""
|
|
ticket = db.session.get(Ticket, ticket_id) or abort(404)
|
|
|
|
# Only the original creator (or IT staff) can re-open
|
|
if not current_user.is_it_staff and ticket.created_by_id != current_user.id:
|
|
abort(403)
|
|
|
|
if ticket.status not in (TicketStatus.RESOLVED, TicketStatus.CLOSED):
|
|
flash('Only resolved or closed tickets can be re-opened.', 'warning')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
reason = request.form.get('reopen_reason', '').strip()
|
|
if not reason:
|
|
flash('Please describe why the issue has returned.', 'danger')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
old_status = ticket.status
|
|
ticket.status = TicketStatus.OPEN
|
|
ticket.resolved_at = None # clear resolved timestamp
|
|
|
|
# Clear SLA suppression so the re-opened ticket can breach again if neglected
|
|
from app.services.sla_service import clear_sla_notification
|
|
clear_sla_notification(ticket.id)
|
|
|
|
# Post a system comment documenting the re-open
|
|
reopen_body = render_comment_body(
|
|
f'**Issue has returned — ticket re-opened**\n\n{reason}'
|
|
)
|
|
comment = Comment(
|
|
ticket_id = ticket.id,
|
|
author_id = current_user.id,
|
|
body = reopen_body,
|
|
is_internal= False,
|
|
)
|
|
db.session.add(comment)
|
|
db.session.flush()
|
|
|
|
log_ticket_history(ticket, 'status', old_status, TicketStatus.OPEN, current_user.id)
|
|
log_action(current_user.id, 'ticket_reopen', 'ticket', ticket.id,
|
|
f'previous_status={old_status} reason_len={len(reason)}')
|
|
db.session.commit()
|
|
logger.info(f'[TICKET REOPEN] ticket_id={ticket.id} '
|
|
f'previous_status={old_status} by user_id={current_user.id}')
|
|
|
|
# Notify IT staff that the ticket has been re-opened
|
|
notify_status_change(ticket, old_status, current_user)
|
|
flash('Ticket re-opened. IT staff have been notified.', 'success')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
|
|
# ─── Canned Responses API (IT only) ──────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/canned-responses')
|
|
@login_required
|
|
def get_canned_responses():
|
|
"""Return active canned responses as JSON for the comment form picker."""
|
|
if not current_user.is_it_staff:
|
|
abort(403)
|
|
responses = CannedResponse.query.filter_by(is_active=True).order_by(
|
|
CannedResponse.category, CannedResponse.title
|
|
).all()
|
|
return __import__('flask').jsonify({'responses': [
|
|
{
|
|
'id' : r.id,
|
|
'title' : r.title,
|
|
'body' : r.body,
|
|
'category': r.category or '',
|
|
}
|
|
for r in responses
|
|
]})
|
|
|
|
|
|
# ─── KB Article Feedback ─────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/kb/<int:article_id>/feedback', methods=['POST'])
|
|
@login_required
|
|
def kb_feedback(article_id):
|
|
"""Record or update a thumbs-up / thumbs-down vote for a KB article.
|
|
|
|
One vote per user per article — subsequent submissions update the existing
|
|
row. Submitting the same vote a second time toggles it off (removes it),
|
|
giving users an undo path.
|
|
"""
|
|
article = db.session.get(KnowledgeBase, article_id) or abort(404)
|
|
value = request.form.get('helpful') # '1' = helpful, '0' = not helpful
|
|
|
|
if value not in ('0', '1'):
|
|
abort(400)
|
|
|
|
is_helpful = value == '1'
|
|
existing = KBFeedback.query.filter_by(
|
|
article_id=article.id, user_id=current_user.id
|
|
).first()
|
|
|
|
if existing:
|
|
if existing.is_helpful == is_helpful:
|
|
# Same vote again → toggle off (remove)
|
|
db.session.delete(existing)
|
|
log_action(current_user.id, 'kb_feedback_remove', 'knowledge_base',
|
|
article.id, f'was_helpful={is_helpful}')
|
|
logger.info(f'[KB FEEDBACK REMOVE] article_id={article.id} '
|
|
f'user_id={current_user.id} was_helpful={is_helpful}')
|
|
else:
|
|
# Changed vote → update
|
|
existing.is_helpful = is_helpful
|
|
log_action(current_user.id, 'kb_feedback_update', 'knowledge_base',
|
|
article.id, f'is_helpful={is_helpful}')
|
|
logger.info(f'[KB FEEDBACK UPDATE] article_id={article.id} '
|
|
f'user_id={current_user.id} is_helpful={is_helpful}')
|
|
else:
|
|
fb = KBFeedback(
|
|
article_id = article.id,
|
|
user_id = current_user.id,
|
|
is_helpful = is_helpful,
|
|
)
|
|
db.session.add(fb)
|
|
log_action(current_user.id, 'kb_feedback_create', 'knowledge_base',
|
|
article.id, f'is_helpful={is_helpful}')
|
|
logger.info(f'[KB FEEDBACK CREATE] article_id={article.id} '
|
|
f'user_id={current_user.id} is_helpful={is_helpful}')
|
|
|
|
db.session.commit()
|
|
|
|
# Return JSON so the widget can update without a full reload
|
|
helpful_count = KBFeedback.query.filter_by(article_id=article.id, is_helpful=True).count()
|
|
not_helpful_count = KBFeedback.query.filter_by(article_id=article.id, is_helpful=False).count()
|
|
from flask import jsonify
|
|
return jsonify({
|
|
'ok' : True,
|
|
'helpful_count' : helpful_count,
|
|
'not_helpful_count': not_helpful_count,
|
|
'user_vote' : None if (existing and existing.is_helpful == is_helpful)
|
|
else ('helpful' if is_helpful else 'not_helpful'),
|
|
})
|
|
|
|
|
|
# ─── Ticket Survey ────────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/survey/<token>', methods=['GET', 'POST'])
|
|
def ticket_survey(token):
|
|
"""Public survey endpoint — no login required.
|
|
|
|
The employee clicks a star in the resolution email which hits this route
|
|
with ?rating=N. A GET with rating pre-selects the star; POST submits
|
|
the full form (rating + optional comment).
|
|
"""
|
|
survey = TicketSatisfaction.query.filter_by(survey_token=token).first()
|
|
if not survey:
|
|
return render_template('tickets/survey_invalid.html'), 404
|
|
|
|
if survey.submitted:
|
|
return render_template('tickets/survey_done.html', survey=survey)
|
|
|
|
# Star-click from email: rating is in query string → auto-submit
|
|
quick_rating = request.args.get('rating', type=int)
|
|
|
|
if request.method == 'POST' or quick_rating:
|
|
rating = quick_rating or request.form.get('rating', type=int)
|
|
comment = request.form.get('comment', '').strip()
|
|
|
|
if not rating or not (1 <= rating <= 5):
|
|
flash('Please select a rating between 1 and 5 stars.', 'danger')
|
|
return render_template('tickets/survey.html', survey=survey)
|
|
|
|
survey.rating = rating
|
|
survey.comment = comment or None
|
|
survey.submitted_at = datetime.utcnow()
|
|
log_action(
|
|
survey.user_id, 'survey_submit', 'ticket', survey.ticket_id,
|
|
f'rating={rating}'
|
|
)
|
|
db.session.commit()
|
|
logger.info(
|
|
f'[SURVEY SUBMIT] ticket_id={survey.ticket_id} '
|
|
f'user_id={survey.user_id} rating={rating}'
|
|
)
|
|
return render_template('tickets/survey_done.html', survey=survey)
|
|
|
|
return render_template('tickets/survey.html', survey=survey,
|
|
quick_rating=quick_rating)
|
|
|
|
|
|
# ─── Ticket Watchers ─────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/tickets/<int:ticket_id>/watch', methods=['POST'])
|
|
@login_required
|
|
def watch_ticket(ticket_id):
|
|
from app.services.license_service import feature_enabled
|
|
from flask import jsonify
|
|
if not feature_enabled('watchers'):
|
|
return jsonify({'error': 'Ticket watchers require a Business or Enterprise license.'}), 403
|
|
ticket = db.session.get(Ticket, ticket_id) or abort(404)
|
|
if not current_user.is_it_staff and ticket.created_by_id != current_user.id:
|
|
abort(403)
|
|
existing = TicketWatcher.query.filter_by(ticket_id=ticket_id, user_id=current_user.id).first()
|
|
if not existing:
|
|
db.session.add(TicketWatcher(ticket_id=ticket_id, user_id=current_user.id))
|
|
db.session.commit()
|
|
return jsonify({'watching': True})
|
|
|
|
|
|
@tickets_bp.route('/tickets/<int:ticket_id>/unwatch', methods=['POST'])
|
|
@login_required
|
|
def unwatch_ticket(ticket_id):
|
|
from app.services.license_service import feature_enabled
|
|
from flask import jsonify
|
|
if not feature_enabled('watchers'):
|
|
return jsonify({'error': 'Ticket watchers require a Business or Enterprise license.'}), 403
|
|
ticket = db.session.get(Ticket, ticket_id) or abort(404)
|
|
if not current_user.is_it_staff and ticket.created_by_id != current_user.id:
|
|
abort(403)
|
|
TicketWatcher.query.filter_by(ticket_id=ticket_id, user_id=current_user.id).delete()
|
|
db.session.commit()
|
|
return jsonify({'watching': False})
|
|
|
|
|
|
# ─── Time Tracking ────────────────────────────────────────────────────────────
|
|
|
|
@tickets_bp.route('/tickets/<int:ticket_id>/log-time', methods=['POST'])
|
|
@login_required
|
|
def log_time(ticket_id):
|
|
from app.services.license_service import feature_enabled
|
|
if not feature_enabled('time_tracking'):
|
|
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
|
from flask import jsonify
|
|
return jsonify({'error': 'Time tracking requires a Business or Enterprise license.'}), 403
|
|
flash('Time tracking requires a Business or Enterprise license.', 'warning')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
if not current_user.is_it_staff:
|
|
abort(403)
|
|
ticket = db.session.get(Ticket, ticket_id) or abort(404)
|
|
try:
|
|
minutes = int(request.form.get('minutes', 0))
|
|
except (ValueError, TypeError):
|
|
minutes = 0
|
|
if minutes <= 0:
|
|
flash('Please enter a valid number of minutes.', 'warning')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
note = request.form.get('note', '').strip()[:200]
|
|
entry = TimeEntry(ticket_id=ticket_id, user_id=current_user.id, minutes=minutes, note=note)
|
|
db.session.add(entry)
|
|
log_action(current_user.id, 'time_log', 'ticket', ticket_id,
|
|
f'minutes={minutes} note={note!r}')
|
|
db.session.commit()
|
|
from flask import jsonify
|
|
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
|
total = sum(e.minutes for e in ticket.time_entries.all())
|
|
return jsonify({'ok': True, 'minutes': minutes, 'total_minutes': total})
|
|
flash(f'{minutes} minutes logged.', 'success')
|
|
return redirect(url_for('tickets.ticket_detail', ticket_id=ticket_id))
|
|
|
|
|
|
# ─── Helpers ─────────────────────────────────────────────────────────────────
|
|
|
|
def _sla_due_date(priority: str) -> 'datetime':
|
|
"""Return the SLA due datetime for a ticket based on its priority.
|
|
|
|
Thresholds are read from SystemSetting so admins can tune them in-app
|
|
without a code deploy. Falls back to config values if settings are absent.
|
|
"""
|
|
from app.models import SystemSetting
|
|
hours_map = {
|
|
TicketPriority.CRITICAL: int(SystemSetting.get('sla_critical_hours', current_app.config.get('SLA_CRITICAL_HOURS', 4))),
|
|
TicketPriority.HIGH: int(SystemSetting.get('sla_high_hours', current_app.config.get('SLA_HIGH_HOURS', 8))),
|
|
TicketPriority.MEDIUM: int(SystemSetting.get('sla_medium_hours', current_app.config.get('SLA_MEDIUM_HOURS', 48))),
|
|
TicketPriority.LOW: int(SystemSetting.get('sla_low_hours', current_app.config.get('SLA_LOW_HOURS', 120))),
|
|
}
|
|
from datetime import timedelta
|
|
hours = hours_map.get(priority, 48)
|
|
return datetime.utcnow() + timedelta(hours=hours)
|
|
|
|
|
|
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] |