313 lines
11 KiB
Python
313 lines
11 KiB
Python
"""
|
|
Complaints utility functions
|
|
|
|
Export and bulk operation utilities.
|
|
"""
|
|
import csv
|
|
import io
|
|
from datetime import datetime
|
|
from typing import List
|
|
|
|
from django.http import HttpResponse
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Font, PatternFill, Alignment
|
|
|
|
|
|
def export_complaints_csv(queryset, filters=None):
|
|
"""
|
|
Export complaints to CSV format.
|
|
|
|
Args:
|
|
queryset: Complaint queryset to export
|
|
filters: Optional dict of applied filters
|
|
|
|
Returns:
|
|
HttpResponse with CSV file
|
|
"""
|
|
response = HttpResponse(content_type='text/csv')
|
|
response['Content-Disposition'] = f'attachment; filename="complaints_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv"'
|
|
|
|
writer = csv.writer(response)
|
|
|
|
# Write header
|
|
writer.writerow([
|
|
'ID',
|
|
'Title',
|
|
'Patient Name',
|
|
'Patient MRN',
|
|
'Hospital',
|
|
'Department',
|
|
'Category',
|
|
'Severity',
|
|
'Priority',
|
|
'Status',
|
|
'Source',
|
|
'Assigned To',
|
|
'Created At',
|
|
'Due At',
|
|
'Is Overdue',
|
|
'Resolved At',
|
|
'Closed At',
|
|
'Description',
|
|
])
|
|
|
|
# Write data
|
|
for complaint in queryset:
|
|
writer.writerow([
|
|
str(complaint.id)[:8],
|
|
complaint.title,
|
|
complaint.patient.get_full_name(),
|
|
complaint.patient.mrn,
|
|
complaint.hospital.name_en,
|
|
complaint.department.name_en if complaint.department else '',
|
|
complaint.get_category_display(),
|
|
complaint.get_severity_display(),
|
|
complaint.get_priority_display(),
|
|
complaint.get_status_display(),
|
|
complaint.get_source_display(),
|
|
complaint.assigned_to.get_full_name() if complaint.assigned_to else '',
|
|
complaint.created_at.strftime('%Y-%m-%d %H:%M:%S'),
|
|
complaint.due_at.strftime('%Y-%m-%d %H:%M:%S'),
|
|
'Yes' if complaint.is_overdue else 'No',
|
|
complaint.resolved_at.strftime('%Y-%m-%d %H:%M:%S') if complaint.resolved_at else '',
|
|
complaint.closed_at.strftime('%Y-%m-%d %H:%M:%S') if complaint.closed_at else '',
|
|
complaint.description[:500],
|
|
])
|
|
|
|
return response
|
|
|
|
|
|
def export_complaints_excel(queryset, filters=None):
|
|
"""
|
|
Export complaints to Excel format with formatting.
|
|
|
|
Args:
|
|
queryset: Complaint queryset to export
|
|
filters: Optional dict of applied filters
|
|
|
|
Returns:
|
|
HttpResponse with Excel file
|
|
"""
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Complaints"
|
|
|
|
# Define styles
|
|
header_font = Font(bold=True, color="FFFFFF")
|
|
header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
|
|
header_alignment = Alignment(horizontal="center", vertical="center")
|
|
|
|
# Write header
|
|
headers = [
|
|
'ID', 'Title', 'Patient Name', 'Patient MRN', 'Hospital', 'Department',
|
|
'Category', 'Severity', 'Priority', 'Status', 'Source', 'Assigned To',
|
|
'Created At', 'Due At', 'Is Overdue', 'Resolved At', 'Closed At', 'Description'
|
|
]
|
|
|
|
for col_num, header in enumerate(headers, 1):
|
|
cell = ws.cell(row=1, column=col_num, value=header)
|
|
cell.font = header_font
|
|
cell.fill = header_fill
|
|
cell.alignment = header_alignment
|
|
|
|
# Write data
|
|
for row_num, complaint in enumerate(queryset, 2):
|
|
ws.cell(row=row_num, column=1, value=str(complaint.id)[:8])
|
|
ws.cell(row=row_num, column=2, value=complaint.title)
|
|
ws.cell(row=row_num, column=3, value=complaint.patient.get_full_name())
|
|
ws.cell(row=row_num, column=4, value=complaint.patient.mrn)
|
|
ws.cell(row=row_num, column=5, value=complaint.hospital.name_en)
|
|
ws.cell(row=row_num, column=6, value=complaint.department.name_en if complaint.department else '')
|
|
ws.cell(row=row_num, column=7, value=complaint.get_category_display())
|
|
ws.cell(row=row_num, column=8, value=complaint.get_severity_display())
|
|
ws.cell(row=row_num, column=9, value=complaint.get_priority_display())
|
|
ws.cell(row=row_num, column=10, value=complaint.get_status_display())
|
|
ws.cell(row=row_num, column=11, value=complaint.get_source_display())
|
|
ws.cell(row=row_num, column=12, value=complaint.assigned_to.get_full_name() if complaint.assigned_to else '')
|
|
ws.cell(row=row_num, column=13, value=complaint.created_at.strftime('%Y-%m-%d %H:%M:%S'))
|
|
ws.cell(row=row_num, column=14, value=complaint.due_at.strftime('%Y-%m-%d %H:%M:%S'))
|
|
ws.cell(row=row_num, column=15, value='Yes' if complaint.is_overdue else 'No')
|
|
ws.cell(row=row_num, column=16, value=complaint.resolved_at.strftime('%Y-%m-%d %H:%M:%S') if complaint.resolved_at else '')
|
|
ws.cell(row=row_num, column=17, value=complaint.closed_at.strftime('%Y-%m-%d %H:%M:%S') if complaint.closed_at else '')
|
|
ws.cell(row=row_num, column=18, value=complaint.description[:500])
|
|
|
|
# Auto-adjust column widths
|
|
for column in ws.columns:
|
|
max_length = 0
|
|
column_letter = column[0].column_letter
|
|
for cell in column:
|
|
try:
|
|
if len(str(cell.value)) > max_length:
|
|
max_length = len(cell.value)
|
|
except:
|
|
pass
|
|
adjusted_width = min(max_length + 2, 50)
|
|
ws.column_dimensions[column_letter].width = adjusted_width
|
|
|
|
# Save to response
|
|
response = HttpResponse(
|
|
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
|
|
)
|
|
response['Content-Disposition'] = f'attachment; filename="complaints_{datetime.now().strftime("%Y%m%d_%H%M%S")}.xlsx"'
|
|
wb.save(response)
|
|
|
|
return response
|
|
|
|
|
|
def bulk_assign_complaints(complaint_ids: List[str], user_id: str, current_user):
|
|
"""
|
|
Bulk assign complaints to a user.
|
|
|
|
Args:
|
|
complaint_ids: List of complaint IDs
|
|
user_id: ID of user to assign to
|
|
current_user: User performing the action
|
|
|
|
Returns:
|
|
dict: Result with success count and errors
|
|
"""
|
|
from apps.complaints.models import Complaint, ComplaintUpdate
|
|
from apps.accounts.models import User
|
|
from django.utils import timezone
|
|
|
|
try:
|
|
assignee = User.objects.get(id=user_id)
|
|
except User.DoesNotExist:
|
|
return {'success': False, 'error': 'User not found'}
|
|
|
|
success_count = 0
|
|
errors = []
|
|
|
|
for complaint_id in complaint_ids:
|
|
try:
|
|
complaint = Complaint.objects.get(id=complaint_id)
|
|
complaint.assigned_to = assignee
|
|
complaint.assigned_at = timezone.now()
|
|
complaint.save(update_fields=['assigned_to', 'assigned_at'])
|
|
|
|
# Create update
|
|
ComplaintUpdate.objects.create(
|
|
complaint=complaint,
|
|
update_type='assignment',
|
|
message=f"Bulk assigned to {assignee.get_full_name()}",
|
|
created_by=current_user
|
|
)
|
|
|
|
success_count += 1
|
|
except Complaint.DoesNotExist:
|
|
errors.append(f"Complaint {complaint_id} not found")
|
|
except Exception as e:
|
|
errors.append(f"Error assigning complaint {complaint_id}: {str(e)}")
|
|
|
|
return {
|
|
'success': True,
|
|
'success_count': success_count,
|
|
'total': len(complaint_ids),
|
|
'errors': errors
|
|
}
|
|
|
|
|
|
def bulk_change_status(complaint_ids: List[str], new_status: str, current_user, note: str = ''):
|
|
"""
|
|
Bulk change status of complaints.
|
|
|
|
Args:
|
|
complaint_ids: List of complaint IDs
|
|
new_status: New status to set
|
|
current_user: User performing the action
|
|
note: Optional note
|
|
|
|
Returns:
|
|
dict: Result with success count and errors
|
|
"""
|
|
from apps.complaints.models import Complaint, ComplaintUpdate
|
|
from django.utils import timezone
|
|
|
|
success_count = 0
|
|
errors = []
|
|
|
|
for complaint_id in complaint_ids:
|
|
try:
|
|
complaint = Complaint.objects.get(id=complaint_id)
|
|
old_status = complaint.status
|
|
complaint.status = new_status
|
|
|
|
# Handle status-specific logic
|
|
if new_status == 'resolved':
|
|
complaint.resolved_at = timezone.now()
|
|
complaint.resolved_by = current_user
|
|
elif new_status == 'closed':
|
|
complaint.closed_at = timezone.now()
|
|
complaint.closed_by = current_user
|
|
|
|
complaint.save()
|
|
|
|
# Create update
|
|
ComplaintUpdate.objects.create(
|
|
complaint=complaint,
|
|
update_type='status_change',
|
|
message=note or f"Bulk status change from {old_status} to {new_status}",
|
|
created_by=current_user,
|
|
old_status=old_status,
|
|
new_status=new_status
|
|
)
|
|
|
|
success_count += 1
|
|
except Complaint.DoesNotExist:
|
|
errors.append(f"Complaint {complaint_id} not found")
|
|
except Exception as e:
|
|
errors.append(f"Error changing status for complaint {complaint_id}: {str(e)}")
|
|
|
|
return {
|
|
'success': True,
|
|
'success_count': success_count,
|
|
'total': len(complaint_ids),
|
|
'errors': errors
|
|
}
|
|
|
|
|
|
def bulk_escalate_complaints(complaint_ids: List[str], current_user, reason: str = ''):
|
|
"""
|
|
Bulk escalate complaints.
|
|
|
|
Args:
|
|
complaint_ids: List of complaint IDs
|
|
current_user: User performing the action
|
|
reason: Escalation reason
|
|
|
|
Returns:
|
|
dict: Result with success count and errors
|
|
"""
|
|
from apps.complaints.models import Complaint, ComplaintUpdate
|
|
from django.utils import timezone
|
|
|
|
success_count = 0
|
|
errors = []
|
|
|
|
for complaint_id in complaint_ids:
|
|
try:
|
|
complaint = Complaint.objects.get(id=complaint_id)
|
|
complaint.escalated_at = timezone.now()
|
|
complaint.save(update_fields=['escalated_at'])
|
|
|
|
# Create update
|
|
ComplaintUpdate.objects.create(
|
|
complaint=complaint,
|
|
update_type='escalation',
|
|
message=f"Bulk escalation. Reason: {reason or 'No reason provided'}",
|
|
created_by=current_user
|
|
)
|
|
|
|
success_count += 1
|
|
except Complaint.DoesNotExist:
|
|
errors.append(f"Complaint {complaint_id} not found")
|
|
except Exception as e:
|
|
errors.append(f"Error escalating complaint {complaint_id}: {str(e)}")
|
|
|
|
return {
|
|
'success': True,
|
|
'success_count': success_count,
|
|
'total': len(complaint_ids),
|
|
'errors': errors
|
|
}
|