2974 lines
108 KiB
Python
2974 lines
108 KiB
Python
"""
|
|
Core views for the Tenhal Multidisciplinary Healthcare Platform.
|
|
|
|
This module contains views for dashboard, patients, consents, and file management.
|
|
"""
|
|
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.views.generic import TemplateView as BaseTemplateView
|
|
from django.contrib import messages
|
|
from django.db.models import Q, Count
|
|
from django.http import JsonResponse
|
|
from django.shortcuts import redirect
|
|
from django.utils import timezone
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django.views.generic import TemplateView, ListView, DetailView, CreateView, UpdateView, DeleteView, View
|
|
from django.urls import reverse_lazy
|
|
|
|
from .mixins import (
|
|
TenantFilterMixin,
|
|
RolePermissionMixin,
|
|
AuditLogMixin,
|
|
HTMXResponseMixin,
|
|
SuccessMessageMixin,
|
|
PaginationMixin,
|
|
)
|
|
from .models import Patient, Consent, ConsentTemplate, File, AuditLog, User
|
|
from .forms import PatientForm, ConsentForm, ConsentTemplateForm, PatientSearchForm, UserSignupForm
|
|
|
|
|
|
# ============================================================================
|
|
# USER SIGNUP VIEW
|
|
# ============================================================================
|
|
|
|
class SignupView(SuccessMessageMixin, CreateView):
|
|
"""
|
|
User signup view.
|
|
Allows new users to create an account.
|
|
"""
|
|
model = User
|
|
form_class = UserSignupForm
|
|
template_name = 'registration/signup.html'
|
|
success_url = reverse_lazy('login')
|
|
success_message = _("Account created successfully! You can now log in.")
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
"""Redirect authenticated users to dashboard."""
|
|
if request.user.is_authenticated:
|
|
return redirect('core:dashboard')
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
def form_valid(self, form):
|
|
"""Save user and optionally log them in."""
|
|
response = super().form_valid(form)
|
|
|
|
# Optionally auto-login after signup
|
|
# Uncomment the following lines to enable auto-login:
|
|
# from django.contrib.auth import login
|
|
# login(self.request, self.object)
|
|
# return redirect('core:dashboard')
|
|
|
|
return response
|
|
|
|
|
|
# ============================================================================
|
|
# CONSENT TEMPLATE MANAGEMENT VIEWS
|
|
# ============================================================================
|
|
|
|
class ConsentTemplateListView(LoginRequiredMixin, RolePermissionMixin,
|
|
TenantFilterMixin, PaginationMixin, ListView):
|
|
"""
|
|
List all consent templates with filtering and search.
|
|
|
|
Features:
|
|
- Filter by consent type
|
|
- Filter by active/inactive status
|
|
- Search by title
|
|
- Sort by type, version, date
|
|
"""
|
|
model = ConsentTemplate
|
|
template_name = 'clinic/consent_template_list.html'
|
|
context_object_name = 'templates'
|
|
paginate_by = 25
|
|
allowed_roles = [User.Role.ADMIN, User.Role.DOCTOR, User.Role.NURSE]
|
|
|
|
def get_queryset(self):
|
|
"""Get filtered and sorted queryset."""
|
|
queryset = super().get_queryset()
|
|
|
|
# Apply search
|
|
search_query = self.request.GET.get('search', '').strip()
|
|
if search_query:
|
|
queryset = queryset.filter(
|
|
Q(title_en__icontains=search_query) |
|
|
Q(title_ar__icontains=search_query) |
|
|
Q(content_en__icontains=search_query) |
|
|
Q(content_ar__icontains=search_query)
|
|
)
|
|
|
|
# Apply filters
|
|
consent_type = self.request.GET.get('consent_type')
|
|
if consent_type:
|
|
queryset = queryset.filter(consent_type=consent_type)
|
|
|
|
status = self.request.GET.get('status')
|
|
if status == 'active':
|
|
queryset = queryset.filter(is_active=True)
|
|
elif status == 'inactive':
|
|
queryset = queryset.filter(is_active=False)
|
|
|
|
# Apply sorting
|
|
sort_by = self.request.GET.get('sort', '-created_at')
|
|
valid_sort_fields = [
|
|
'consent_type', '-consent_type',
|
|
'title_en', '-title_en',
|
|
'version', '-version',
|
|
'created_at', '-created_at',
|
|
]
|
|
if sort_by in valid_sort_fields:
|
|
queryset = queryset.order_by(sort_by)
|
|
|
|
return queryset
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add filter options to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
context['consent_types'] = ConsentTemplate.ConsentType.choices
|
|
context['current_filters'] = {
|
|
'search': self.request.GET.get('search', ''),
|
|
'consent_type': self.request.GET.get('consent_type', ''),
|
|
'status': self.request.GET.get('status', ''),
|
|
'sort': self.request.GET.get('sort', '-created_at'),
|
|
}
|
|
|
|
# Add statistics
|
|
all_templates = ConsentTemplate.objects.filter(tenant=self.request.user.tenant)
|
|
context['stats'] = {
|
|
'total': all_templates.count(),
|
|
'active': all_templates.filter(is_active=True).count(),
|
|
'inactive': all_templates.filter(is_active=False).count(),
|
|
}
|
|
|
|
return context
|
|
|
|
|
|
class ConsentTemplateDetailView(LoginRequiredMixin, RolePermissionMixin,
|
|
TenantFilterMixin, DetailView):
|
|
"""
|
|
View consent template details with preview functionality.
|
|
"""
|
|
model = ConsentTemplate
|
|
template_name = 'clinic/consent_template_detail.html'
|
|
context_object_name = 'template'
|
|
allowed_roles = [User.Role.ADMIN, User.Role.DOCTOR, User.Role.NURSE]
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add preview data to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
template = self.object
|
|
|
|
# Get a sample patient for preview (if any exists)
|
|
sample_patient = Patient.objects.filter(
|
|
tenant=self.request.user.tenant
|
|
).first()
|
|
|
|
if sample_patient:
|
|
context['preview_en'] = template.get_populated_content(sample_patient, 'en')
|
|
context['preview_ar'] = template.get_populated_content(sample_patient, 'ar')
|
|
context['sample_patient'] = sample_patient
|
|
|
|
# Get version history
|
|
context['version_history'] = ConsentTemplate.objects.filter(
|
|
tenant=template.tenant,
|
|
consent_type=template.consent_type
|
|
).order_by('-version')[:10]
|
|
|
|
return context
|
|
|
|
|
|
class ConsentTemplateCreateView(LoginRequiredMixin, RolePermissionMixin,
|
|
AuditLogMixin, SuccessMessageMixin, CreateView):
|
|
"""
|
|
Create a new consent template.
|
|
"""
|
|
model = ConsentTemplate
|
|
form_class = ConsentTemplateForm
|
|
template_name = 'clinic/consent_template_form.html'
|
|
success_message = _("Consent template created successfully!")
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_success_url(self):
|
|
"""Redirect to template detail page."""
|
|
return reverse_lazy('core:consent_template_detail', kwargs={'pk': self.object.pk})
|
|
|
|
def form_valid(self, form):
|
|
"""Set tenant before saving."""
|
|
form.instance.tenant = self.request.user.tenant
|
|
|
|
# Auto-set version to 1 for new templates
|
|
if not form.instance.version:
|
|
form.instance.version = 1
|
|
|
|
return super().form_valid(form)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add form title to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
context['form_title'] = _('Create New Consent Template')
|
|
context['submit_text'] = _('Create Template')
|
|
return context
|
|
|
|
|
|
class ConsentTemplateUpdateView(LoginRequiredMixin, RolePermissionMixin,
|
|
TenantFilterMixin, AuditLogMixin,
|
|
SuccessMessageMixin, UpdateView):
|
|
"""
|
|
Update an existing consent template.
|
|
Creates a new version instead of modifying the existing one.
|
|
"""
|
|
model = ConsentTemplate
|
|
form_class = ConsentTemplateForm
|
|
template_name = 'clinic/consent_template_form.html'
|
|
success_message = _("Consent template updated successfully!")
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_success_url(self):
|
|
"""Redirect to template detail page."""
|
|
return reverse_lazy('core:consent_template_detail', kwargs={'pk': self.object.pk})
|
|
|
|
def form_valid(self, form):
|
|
"""Create new version instead of updating."""
|
|
# Deactivate old version
|
|
old_template = self.get_object()
|
|
old_template.is_active = False
|
|
old_template.save(update_fields=['is_active'])
|
|
|
|
# Create new version
|
|
form.instance.pk = None # Force creation of new object
|
|
form.instance.tenant = self.request.user.tenant
|
|
form.instance.version = old_template.version + 1
|
|
form.instance.is_active = True
|
|
|
|
return super().form_valid(form)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add form title to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
context['form_title'] = _(f'Update Consent Template: {self.object.title_en}')
|
|
context['submit_text'] = _('Update Template (Creates New Version)')
|
|
context['is_update'] = True
|
|
return context
|
|
|
|
|
|
class ConsentTemplateDeleteView(LoginRequiredMixin, RolePermissionMixin,
|
|
TenantFilterMixin, AuditLogMixin, DeleteView):
|
|
"""
|
|
Deactivate a consent template (soft delete).
|
|
"""
|
|
model = ConsentTemplate
|
|
template_name = 'clinic/consent_template_confirm_delete.html'
|
|
success_url = reverse_lazy('core:consent_template_list')
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def delete(self, request, *args, **kwargs):
|
|
"""Soft delete by deactivating instead of deleting."""
|
|
self.object = self.get_object()
|
|
|
|
# Deactivate instead of delete
|
|
self.object.is_active = False
|
|
self.object.save(update_fields=['is_active'])
|
|
|
|
messages.success(
|
|
request,
|
|
_(f'Consent template "{self.object.title_en}" has been deactivated.')
|
|
)
|
|
|
|
return redirect(self.success_url)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add additional context."""
|
|
context = super().get_context_data(**kwargs)
|
|
context['form_title'] = _('Create Account')
|
|
return context
|
|
|
|
|
|
class DashboardView(LoginRequiredMixin, TemplateView):
|
|
"""
|
|
Role-based dashboard view.
|
|
|
|
Shows different widgets based on user role:
|
|
- Admin/FrontDesk: All appointments, all patients, financial summary
|
|
- Clinical staff (Doctor/Nurse/OT/SLP/ABA): Their appointments, their patients
|
|
- Finance: Financial metrics, pending invoices
|
|
"""
|
|
|
|
template_name = 'core/dashboard.html'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
user = self.request.user
|
|
today = timezone.now().date()
|
|
|
|
# Common context
|
|
context['user_role'] = user.role
|
|
context['today'] = today
|
|
|
|
# Role-specific widgets
|
|
if user.role in [User.Role.ADMIN, User.Role.FRONT_DESK]:
|
|
context.update(self._get_admin_widgets())
|
|
elif user.role in [User.Role.DOCTOR, User.Role.NURSE, User.Role.OT,
|
|
User.Role.SLP, User.Role.ABA]:
|
|
context.update(self._get_clinical_widgets())
|
|
elif user.role == User.Role.FINANCE:
|
|
context.update(self._get_finance_widgets())
|
|
|
|
return context
|
|
|
|
def _get_admin_widgets(self):
|
|
"""Get widgets for Admin and FrontDesk roles."""
|
|
from appointments.models import Appointment
|
|
from finance.models import Invoice, Payment
|
|
from hr.models import Attendance, LeaveRequest
|
|
from django.db.models import Sum, Count, Q
|
|
from datetime import timedelta
|
|
|
|
today = timezone.now().date()
|
|
tenant = self.request.user.tenant
|
|
|
|
# Today's appointments
|
|
todays_appointments = Appointment.objects.filter(
|
|
tenant=tenant,
|
|
scheduled_date=today
|
|
).select_related('patient', 'provider', 'clinic')
|
|
|
|
# Recent patients (last 10)
|
|
recent_patients = Patient.objects.filter(
|
|
tenant=tenant
|
|
).order_by('-created_at')[:10]
|
|
|
|
# Enhanced appointment stats
|
|
appointment_stats = {
|
|
'total_today': todays_appointments.count(),
|
|
'booked': todays_appointments.filter(status=Appointment.Status.BOOKED).count(),
|
|
'confirmed': todays_appointments.filter(status=Appointment.Status.CONFIRMED).count(),
|
|
'arrived': todays_appointments.filter(status=Appointment.Status.ARRIVED).count(),
|
|
'in_progress': todays_appointments.filter(status=Appointment.Status.IN_PROGRESS).count(),
|
|
'completed': todays_appointments.filter(status=Appointment.Status.COMPLETED).count(),
|
|
'cancelled': todays_appointments.filter(status=Appointment.Status.CANCELLED).count(),
|
|
'no_show': todays_appointments.filter(status=Appointment.Status.NO_SHOW).count(),
|
|
}
|
|
|
|
# This week's appointments
|
|
week_start = today - timedelta(days=today.weekday())
|
|
week_end = week_start + timedelta(days=6)
|
|
week_appointments = Appointment.objects.filter(
|
|
tenant=tenant,
|
|
scheduled_date__range=[week_start, week_end]
|
|
)
|
|
|
|
# Enhanced financial summary
|
|
pending_invoices = Invoice.objects.filter(
|
|
tenant=tenant,
|
|
status__in=[Invoice.Status.DRAFT, Invoice.Status.ISSUED]
|
|
)
|
|
|
|
overdue_invoices = Invoice.objects.filter(
|
|
tenant=tenant,
|
|
status=Invoice.Status.ISSUED,
|
|
due_date__lt=today
|
|
)
|
|
|
|
# Today's revenue
|
|
todays_payments = Payment.objects.filter(
|
|
invoice__tenant=tenant,
|
|
payment_date__date=today,
|
|
status=Payment.Status.COMPLETED
|
|
)
|
|
|
|
# This month's revenue
|
|
month_start = today.replace(day=1)
|
|
month_payments = Payment.objects.filter(
|
|
invoice__tenant=tenant,
|
|
payment_date__date__gte=month_start,
|
|
status=Payment.Status.COMPLETED
|
|
)
|
|
|
|
financial_summary = {
|
|
'pending_invoices_count': pending_invoices.count(),
|
|
'pending_amount': sum(inv.amount_due for inv in pending_invoices),
|
|
'overdue_count': overdue_invoices.count(),
|
|
'overdue_amount': sum(inv.amount_due for inv in overdue_invoices),
|
|
'revenue_today': todays_payments.aggregate(total=Sum('amount'))['total'] or 0,
|
|
'revenue_month': month_payments.aggregate(total=Sum('amount'))['total'] or 0,
|
|
}
|
|
|
|
# HR summary
|
|
todays_attendance = Attendance.objects.filter(
|
|
tenant=tenant,
|
|
date=today
|
|
)
|
|
|
|
pending_leave_requests = LeaveRequest.objects.filter(
|
|
tenant=tenant,
|
|
status=LeaveRequest.Status.PENDING
|
|
)
|
|
|
|
staff_on_leave_today = LeaveRequest.objects.filter(
|
|
tenant=tenant,
|
|
status=LeaveRequest.Status.APPROVED,
|
|
start_date__lte=today,
|
|
end_date__gte=today
|
|
)
|
|
|
|
hr_summary = {
|
|
'total_staff': User.objects.filter(tenant=tenant, is_active=True).count(),
|
|
'present_today': todays_attendance.filter(status=Attendance.Status.PRESENT).count(),
|
|
'late_today': todays_attendance.filter(status=Attendance.Status.LATE).count(),
|
|
'absent_today': todays_attendance.filter(status=Attendance.Status.ABSENT).count(),
|
|
'on_leave_today': staff_on_leave_today.count(),
|
|
'pending_leave_requests': pending_leave_requests.count(),
|
|
}
|
|
|
|
# Clinical summary
|
|
clinical_summary = self._get_clinical_summary(tenant, today)
|
|
|
|
# System health
|
|
system_health = self._get_system_health(tenant)
|
|
|
|
# Quick stats
|
|
quick_stats = {
|
|
'total_patients': Patient.objects.filter(tenant=tenant).count(),
|
|
'new_patients_month': Patient.objects.filter(
|
|
tenant=tenant,
|
|
created_at__date__gte=month_start
|
|
).count(),
|
|
'total_appointments_today': todays_appointments.count(),
|
|
'total_appointments_week': week_appointments.count(),
|
|
'active_providers': User.objects.filter(
|
|
tenant=tenant,
|
|
role__in=[User.Role.DOCTOR, User.Role.NURSE, User.Role.OT,
|
|
User.Role.SLP, User.Role.ABA],
|
|
is_active=True
|
|
).count(),
|
|
}
|
|
|
|
return {
|
|
'todays_appointments': todays_appointments[:10],
|
|
'recent_patients': recent_patients,
|
|
'appointment_stats': appointment_stats,
|
|
'financial_summary': financial_summary,
|
|
'hr_summary': hr_summary,
|
|
'clinical_summary': clinical_summary,
|
|
'system_health': system_health,
|
|
'quick_stats': quick_stats,
|
|
}
|
|
|
|
def _get_clinical_summary(self, tenant, today):
|
|
"""Get summary of clinical documents across all disciplines."""
|
|
from nursing.models import NursingEncounter
|
|
from medical.models import MedicalConsultation, MedicalFollowUp
|
|
from aba.models import ABAConsult
|
|
from ot.models import OTConsult, OTSession
|
|
from slp.models import SLPConsult, SLPAssessment, SLPIntervention
|
|
|
|
# Count documents by discipline
|
|
nursing_count = NursingEncounter.objects.filter(
|
|
patient__tenant=tenant
|
|
).count()
|
|
|
|
medical_count = (
|
|
MedicalConsultation.objects.filter(patient__tenant=tenant).count() +
|
|
MedicalFollowUp.objects.filter(patient__tenant=tenant).count()
|
|
)
|
|
|
|
aba_count = ABAConsult.objects.filter(patient__tenant=tenant).count()
|
|
|
|
ot_count = (
|
|
OTConsult.objects.filter(patient__tenant=tenant).count() +
|
|
OTSession.objects.filter(patient__tenant=tenant).count()
|
|
)
|
|
|
|
slp_count = (
|
|
SLPConsult.objects.filter(patient__tenant=tenant).count() +
|
|
SLPAssessment.objects.filter(patient__tenant=tenant).count() +
|
|
SLPIntervention.objects.filter(patient__tenant=tenant).count()
|
|
)
|
|
|
|
# Count unsigned documents (if they have signed_by field)
|
|
unsigned_count = 0
|
|
try:
|
|
unsigned_count += MedicalConsultation.objects.filter(
|
|
patient__tenant=tenant,
|
|
signed_by__isnull=True
|
|
).count()
|
|
unsigned_count += MedicalFollowUp.objects.filter(
|
|
patient__tenant=tenant,
|
|
signed_by__isnull=True
|
|
).count()
|
|
except:
|
|
pass
|
|
|
|
return {
|
|
'total_documents': nursing_count + medical_count + aba_count + ot_count + slp_count,
|
|
'nursing_count': nursing_count,
|
|
'medical_count': medical_count,
|
|
'aba_count': aba_count,
|
|
'ot_count': ot_count,
|
|
'slp_count': slp_count,
|
|
'unsigned_count': unsigned_count,
|
|
}
|
|
|
|
def _get_system_health(self, tenant):
|
|
"""Get system health indicators."""
|
|
from finance.models import CSID
|
|
from datetime import timedelta
|
|
|
|
system_health = {
|
|
'zatca_compliant': True,
|
|
'csid_status': 'unknown',
|
|
'csid_expiry_days': None,
|
|
'alerts': [],
|
|
}
|
|
|
|
# Check CSID status
|
|
try:
|
|
active_csid = CSID.objects.filter(
|
|
tenant=tenant,
|
|
status=CSID.Status.ACTIVE
|
|
).order_by('-expiry_date').first()
|
|
|
|
if active_csid:
|
|
days_until_expiry = active_csid.days_until_expiry
|
|
system_health['csid_expiry_days'] = days_until_expiry
|
|
|
|
if days_until_expiry <= 0:
|
|
system_health['csid_status'] = 'expired'
|
|
system_health['zatca_compliant'] = False
|
|
system_health['alerts'].append({
|
|
'level': 'danger',
|
|
'message': 'CSID has expired! E-invoicing is not possible.'
|
|
})
|
|
elif days_until_expiry <= 7:
|
|
system_health['csid_status'] = 'expiring_soon'
|
|
system_health['alerts'].append({
|
|
'level': 'danger',
|
|
'message': f'CSID expires in {days_until_expiry} days! Renew immediately.'
|
|
})
|
|
elif days_until_expiry <= 30:
|
|
system_health['csid_status'] = 'needs_renewal'
|
|
system_health['alerts'].append({
|
|
'level': 'warning',
|
|
'message': f'CSID expires in {days_until_expiry} days. Plan renewal.'
|
|
})
|
|
else:
|
|
system_health['csid_status'] = 'active'
|
|
else:
|
|
system_health['csid_status'] = 'none'
|
|
system_health['zatca_compliant'] = False
|
|
system_health['alerts'].append({
|
|
'level': 'danger',
|
|
'message': 'No active CSID found. E-invoicing is not configured.'
|
|
})
|
|
except Exception as e:
|
|
system_health['alerts'].append({
|
|
'level': 'warning',
|
|
'message': f'Unable to check CSID status: {str(e)}'
|
|
})
|
|
|
|
return system_health
|
|
|
|
def _get_revenue_trends(self, tenant, today):
|
|
"""Get revenue trends for charts."""
|
|
from finance.models import Payment
|
|
from django.db.models import Sum
|
|
from datetime import timedelta
|
|
|
|
# Last 7 days revenue
|
|
daily_revenue = []
|
|
for i in range(6, -1, -1):
|
|
day = today - timedelta(days=i)
|
|
revenue = Payment.objects.filter(
|
|
invoice__tenant=tenant,
|
|
payment_date__date=day,
|
|
status=Payment.Status.COMPLETED
|
|
).aggregate(total=Sum('amount'))['total'] or 0
|
|
|
|
daily_revenue.append({
|
|
'date': day,
|
|
'revenue': float(revenue)
|
|
})
|
|
|
|
# Last 12 months revenue
|
|
monthly_revenue = []
|
|
for i in range(11, -1, -1):
|
|
month_date = today.replace(day=1) - timedelta(days=i*30)
|
|
month_start = month_date.replace(day=1)
|
|
|
|
if i == 0:
|
|
month_end = today
|
|
else:
|
|
next_month = month_start.replace(day=28) + timedelta(days=4)
|
|
month_end = next_month.replace(day=1) - timedelta(days=1)
|
|
|
|
revenue = Payment.objects.filter(
|
|
invoice__tenant=tenant,
|
|
payment_date__date__gte=month_start,
|
|
payment_date__date__lte=month_end,
|
|
status=Payment.Status.COMPLETED
|
|
).aggregate(total=Sum('amount'))['total'] or 0
|
|
|
|
monthly_revenue.append({
|
|
'month': month_start.strftime('%b %Y'),
|
|
'revenue': float(revenue)
|
|
})
|
|
|
|
return {
|
|
'daily': daily_revenue,
|
|
'monthly': monthly_revenue
|
|
}
|
|
|
|
def _get_patient_analytics(self, tenant, today):
|
|
"""Get patient analytics and demographics."""
|
|
from datetime import timedelta
|
|
|
|
# Patient growth over last 12 months
|
|
patient_growth = []
|
|
for i in range(11, -1, -1):
|
|
month_date = today.replace(day=1) - timedelta(days=i*30)
|
|
month_start = month_date.replace(day=1)
|
|
|
|
if i == 0:
|
|
month_end = today
|
|
else:
|
|
next_month = month_start.replace(day=28) + timedelta(days=4)
|
|
month_end = next_month.replace(day=1) - timedelta(days=1)
|
|
|
|
count = Patient.objects.filter(
|
|
tenant=tenant,
|
|
created_at__date__gte=month_start,
|
|
created_at__date__lte=month_end
|
|
).count()
|
|
|
|
patient_growth.append({
|
|
'month': month_start.strftime('%b %Y'),
|
|
'count': count
|
|
})
|
|
|
|
# Gender distribution
|
|
male_count = Patient.objects.filter(tenant=tenant, sex='M').count()
|
|
female_count = Patient.objects.filter(tenant=tenant, sex='F').count()
|
|
total_patients = male_count + female_count
|
|
|
|
# Age distribution
|
|
from datetime import date
|
|
age_groups = {
|
|
'0-5': 0,
|
|
'6-12': 0,
|
|
'13-18': 0,
|
|
'19-30': 0,
|
|
'31-50': 0,
|
|
'51+': 0
|
|
}
|
|
|
|
for patient in Patient.objects.filter(tenant=tenant):
|
|
age = (date.today() - patient.date_of_birth).days // 365
|
|
if age <= 5:
|
|
age_groups['0-5'] += 1
|
|
elif age <= 12:
|
|
age_groups['6-12'] += 1
|
|
elif age <= 18:
|
|
age_groups['13-18'] += 1
|
|
elif age <= 30:
|
|
age_groups['19-30'] += 1
|
|
elif age <= 50:
|
|
age_groups['31-50'] += 1
|
|
else:
|
|
age_groups['51+'] += 1
|
|
|
|
return {
|
|
'growth': patient_growth,
|
|
'gender': {
|
|
'male': male_count,
|
|
'female': female_count,
|
|
'male_percent': (male_count / total_patients * 100) if total_patients > 0 else 0,
|
|
'female_percent': (female_count / total_patients * 100) if total_patients > 0 else 0,
|
|
},
|
|
'age_groups': age_groups,
|
|
'total': total_patients
|
|
}
|
|
|
|
def _get_appointment_analytics(self, tenant, today):
|
|
"""Get detailed appointment analytics."""
|
|
from appointments.models import Appointment
|
|
from datetime import timedelta
|
|
|
|
# This month's appointments
|
|
month_start = today.replace(day=1)
|
|
month_appointments = Appointment.objects.filter(
|
|
tenant=tenant,
|
|
scheduled_date__gte=month_start,
|
|
scheduled_date__lte=today
|
|
)
|
|
|
|
# Completion rate
|
|
total_completed = month_appointments.filter(status=Appointment.Status.COMPLETED).count()
|
|
total_scheduled = month_appointments.count()
|
|
completion_rate = (total_completed / total_scheduled * 100) if total_scheduled > 0 else 0
|
|
|
|
# No-show rate
|
|
no_shows = month_appointments.filter(status=Appointment.Status.NO_SHOW).count()
|
|
no_show_rate = (no_shows / total_scheduled * 100) if total_scheduled > 0 else 0
|
|
|
|
# Cancellation rate
|
|
cancelled = month_appointments.filter(status=Appointment.Status.CANCELLED).count()
|
|
cancellation_rate = (cancelled / total_scheduled * 100) if total_scheduled > 0 else 0
|
|
|
|
# Appointments by clinic
|
|
from core.models import Clinic
|
|
by_clinic = []
|
|
for clinic in Clinic.objects.filter(tenant=tenant, is_active=True):
|
|
count = month_appointments.filter(clinic=clinic).count()
|
|
if count > 0:
|
|
by_clinic.append({
|
|
'clinic': clinic.name_en,
|
|
'count': count
|
|
})
|
|
|
|
# Provider workload
|
|
from appointments.models import Provider
|
|
provider_workload = []
|
|
for provider in Provider.objects.filter(tenant=tenant, is_available=True)[:10]:
|
|
count = month_appointments.filter(provider=provider).count()
|
|
completed = month_appointments.filter(
|
|
provider=provider,
|
|
status=Appointment.Status.COMPLETED
|
|
).count()
|
|
|
|
provider_workload.append({
|
|
'provider': provider.user.get_full_name(),
|
|
'total': count,
|
|
'completed': completed
|
|
})
|
|
|
|
return {
|
|
'completion_rate': round(completion_rate, 1),
|
|
'no_show_rate': round(no_show_rate, 1),
|
|
'cancellation_rate': round(cancellation_rate, 1),
|
|
'by_clinic': by_clinic,
|
|
'provider_workload': provider_workload,
|
|
'total_month': total_scheduled,
|
|
'completed_month': total_completed,
|
|
'no_shows_month': no_shows,
|
|
'cancelled_month': cancelled
|
|
}
|
|
|
|
def _get_clinical_widgets(self):
|
|
"""Get widgets for clinical staff (Doctor, Nurse, OT, SLP, ABA)."""
|
|
from appointments.models import Appointment
|
|
|
|
today = timezone.now().date()
|
|
user = self.request.user
|
|
tenant = user.tenant
|
|
|
|
# Get provider profile for this user (if exists)
|
|
try:
|
|
provider = user.provider_profile
|
|
except:
|
|
provider = None
|
|
|
|
# My appointments today
|
|
if provider:
|
|
my_appointments = Appointment.objects.filter(
|
|
tenant=tenant,
|
|
provider=provider,
|
|
scheduled_date=today
|
|
).select_related('patient', 'clinic').order_by('scheduled_time')
|
|
else:
|
|
my_appointments = Appointment.objects.none()
|
|
|
|
# My upcoming appointments (next 7 days)
|
|
if provider:
|
|
upcoming_appointments = Appointment.objects.filter(
|
|
tenant=tenant,
|
|
provider=provider,
|
|
scheduled_date__gt=today,
|
|
scheduled_date__lte=today + timezone.timedelta(days=7),
|
|
status__in=[Appointment.Status.BOOKED, Appointment.Status.CONFIRMED]
|
|
).select_related('patient', 'clinic').order_by('scheduled_date', 'scheduled_time')[:10]
|
|
else:
|
|
upcoming_appointments = Appointment.objects.none()
|
|
|
|
# My recent patients
|
|
if provider:
|
|
my_recent_patients = Patient.objects.filter(
|
|
tenant=tenant,
|
|
appointments__provider=provider
|
|
).distinct().order_by('-appointments__scheduled_date')[:10]
|
|
else:
|
|
my_recent_patients = Patient.objects.none()
|
|
|
|
# My stats
|
|
my_stats = {
|
|
'appointments_today': my_appointments.count(),
|
|
'completed_today': my_appointments.filter(status=Appointment.Status.COMPLETED).count(),
|
|
'pending_today': my_appointments.filter(
|
|
status__in=[Appointment.Status.CONFIRMED, Appointment.Status.ARRIVED]
|
|
).count(),
|
|
'total_patients': my_recent_patients.count(),
|
|
}
|
|
|
|
# Pending tasks (appointments needing attention)
|
|
pending_tasks = my_appointments.filter(
|
|
status__in=[Appointment.Status.ARRIVED, Appointment.Status.IN_PROGRESS]
|
|
)
|
|
|
|
return {
|
|
'my_appointments': my_appointments,
|
|
'upcoming_appointments': upcoming_appointments,
|
|
'my_recent_patients': my_recent_patients,
|
|
'my_stats': my_stats,
|
|
'pending_tasks': pending_tasks,
|
|
}
|
|
|
|
def _get_finance_widgets(self):
|
|
"""Get widgets for Finance role."""
|
|
from finance.models import Invoice, Payment
|
|
|
|
today = timezone.now().date()
|
|
tenant = self.request.user.tenant
|
|
|
|
# Pending invoices
|
|
pending_invoices = Invoice.objects.filter(
|
|
tenant=tenant,
|
|
status__in=[Invoice.Status.DRAFT, Invoice.Status.ISSUED]
|
|
).select_related('patient', 'payer').order_by('-issue_date')[:10]
|
|
|
|
# Overdue invoices
|
|
overdue_invoices = Invoice.objects.filter(
|
|
tenant=tenant,
|
|
status=Invoice.Status.ISSUED,
|
|
due_date__lt=today
|
|
).select_related('patient', 'payer').order_by('due_date')[:10]
|
|
|
|
# Recent payments
|
|
recent_payments = Payment.objects.filter(
|
|
invoice__tenant=tenant
|
|
).select_related('invoice', 'invoice__patient').order_by('-payment_date')[:10]
|
|
|
|
# Payment methods breakdown
|
|
all_payments = Payment.objects.filter(
|
|
invoice__tenant=tenant,
|
|
status=Payment.Status.COMPLETED
|
|
)
|
|
|
|
cash_total = sum(
|
|
p.amount for p in all_payments.filter(method=Payment.PaymentMethod.CASH)
|
|
)
|
|
card_total = sum(
|
|
p.amount for p in all_payments.filter(method=Payment.PaymentMethod.CARD)
|
|
)
|
|
insurance_total = sum(
|
|
p.amount for p in all_payments.filter(method=Payment.PaymentMethod.INSURANCE)
|
|
)
|
|
transfer_total = sum(
|
|
p.amount for p in all_payments.filter(method=Payment.PaymentMethod.BANK_TRANSFER)
|
|
)
|
|
|
|
total_payments = cash_total + card_total + insurance_total + transfer_total
|
|
|
|
payment_methods = {
|
|
'cash': cash_total,
|
|
'card': card_total,
|
|
'insurance': insurance_total,
|
|
'transfer': transfer_total,
|
|
'cash_percent': (cash_total / total_payments * 100) if total_payments > 0 else 0,
|
|
'card_percent': (card_total / total_payments * 100) if total_payments > 0 else 0,
|
|
'insurance_percent': (insurance_total / total_payments * 100) if total_payments > 0 else 0,
|
|
'transfer_percent': (transfer_total / total_payments * 100) if total_payments > 0 else 0,
|
|
}
|
|
|
|
# Financial stats
|
|
financial_stats = {
|
|
'pending_invoices_count': pending_invoices.count(),
|
|
'pending_amount': sum(inv.total - inv.amount_paid for inv in pending_invoices),
|
|
'overdue_count': overdue_invoices.count(),
|
|
'overdue_amount': sum(inv.total - inv.amount_paid for inv in overdue_invoices),
|
|
'payments_today': Payment.objects.filter(
|
|
invoice__tenant=tenant,
|
|
payment_date=today
|
|
).count(),
|
|
'revenue_today': sum(
|
|
p.amount for p in Payment.objects.filter(
|
|
invoice__tenant=tenant,
|
|
payment_date=today,
|
|
status=Payment.Status.COMPLETED
|
|
)
|
|
),
|
|
}
|
|
|
|
return {
|
|
'pending_invoices': pending_invoices,
|
|
'overdue_invoices': overdue_invoices,
|
|
'recent_payments': recent_payments,
|
|
'financial_stats': financial_stats,
|
|
'payment_methods': payment_methods,
|
|
}
|
|
|
|
|
|
class PatientListView(LoginRequiredMixin, TenantFilterMixin, PaginationMixin,
|
|
HTMXResponseMixin, ListView):
|
|
"""
|
|
Full-featured patient list view.
|
|
|
|
Features:
|
|
- Live search (name, MRN, phone, national ID)
|
|
- Filtering (status, date range, gender)
|
|
- Sorting (name, MRN, registration date)
|
|
- Quick actions (view, edit)
|
|
- Export to CSV/Excel
|
|
- HTMX support for live search and infinite scroll
|
|
"""
|
|
model = Patient
|
|
template_name = 'core/patient_list.html'
|
|
htmx_template_name = 'core/partials/patient_list_partial.html'
|
|
context_object_name = 'patients'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
"""Get filtered and sorted queryset."""
|
|
queryset = super().get_queryset()
|
|
|
|
# Apply search
|
|
search_query = self.request.GET.get('search', '').strip()
|
|
if search_query:
|
|
queryset = queryset.filter(
|
|
Q(first_name_en__icontains=search_query) |
|
|
Q(last_name_en__icontains=search_query) |
|
|
Q(first_name_ar__icontains=search_query) |
|
|
Q(last_name_ar__icontains=search_query) |
|
|
Q(mrn__icontains=search_query) |
|
|
Q(national_id__icontains=search_query) |
|
|
Q(phone__icontains=search_query) |
|
|
Q(email__icontains=search_query)
|
|
)
|
|
|
|
# Apply filters
|
|
status = self.request.GET.get('status')
|
|
if status:
|
|
queryset = queryset.filter(is_active=(status == 'active'))
|
|
|
|
gender = self.request.GET.get('gender')
|
|
if gender:
|
|
queryset = queryset.filter(gender=gender)
|
|
|
|
date_from = self.request.GET.get('date_from')
|
|
if date_from:
|
|
queryset = queryset.filter(date_of_birth__gte=date_from)
|
|
|
|
date_to = self.request.GET.get('date_to')
|
|
if date_to:
|
|
queryset = queryset.filter(date_of_birth__lte=date_to)
|
|
|
|
# Apply sorting
|
|
sort_by = self.request.GET.get('sort', '-created_at')
|
|
valid_sort_fields = [
|
|
'mrn', '-mrn',
|
|
'first_name_en', '-first_name_en',
|
|
'last_name_en', '-last_name_en',
|
|
'date_of_birth', '-date_of_birth',
|
|
'created_at', '-created_at',
|
|
]
|
|
if sort_by in valid_sort_fields:
|
|
queryset = queryset.order_by(sort_by)
|
|
|
|
return queryset.select_related('tenant')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add search form and export options to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
# Add search form
|
|
context['search_form'] = PatientSearchForm(self.request.GET)
|
|
|
|
# Add current filters for display
|
|
context['current_filters'] = {
|
|
'search': self.request.GET.get('search', ''),
|
|
'status': self.request.GET.get('status', ''),
|
|
'gender': self.request.GET.get('gender', ''),
|
|
'sort': self.request.GET.get('sort', '-created_at'),
|
|
}
|
|
|
|
# Add export flag
|
|
context['can_export'] = self.request.user.role in [
|
|
User.Role.ADMIN, User.Role.FRONT_DESK
|
|
]
|
|
|
|
return context
|
|
|
|
def render_to_response(self, context, **response_kwargs):
|
|
"""Handle CSV/Excel export."""
|
|
export_format = self.request.GET.get('export')
|
|
|
|
if export_format in ['csv', 'excel']:
|
|
return self._export_patients(export_format, context['patients'])
|
|
|
|
return super().render_to_response(context, **response_kwargs)
|
|
|
|
def _export_patients(self, format_type, patients):
|
|
"""Export patients to CSV or Excel."""
|
|
import csv
|
|
from django.http import HttpResponse
|
|
|
|
if format_type == 'csv':
|
|
response = HttpResponse(content_type='text/csv')
|
|
response['Content-Disposition'] = f'attachment; filename="patients_{timezone.now().date()}.csv"'
|
|
|
|
writer = csv.writer(response)
|
|
writer.writerow([
|
|
'MRN', 'First Name (EN)', 'Last Name (EN)',
|
|
'First Name (AR)', 'Last Name (AR)',
|
|
'National ID', 'Date of Birth', 'Gender',
|
|
'Phone', 'Email', 'Status', 'Registration Date'
|
|
])
|
|
|
|
for patient in patients:
|
|
writer.writerow([
|
|
patient.mrn,
|
|
patient.first_name_en,
|
|
patient.last_name_en,
|
|
patient.first_name_ar,
|
|
patient.last_name_ar,
|
|
patient.national_id,
|
|
patient.date_of_birth,
|
|
patient.get_gender_display(),
|
|
patient.phone,
|
|
patient.email,
|
|
'Active' if patient.is_active else 'Inactive',
|
|
patient.created_at.date(),
|
|
])
|
|
|
|
return response
|
|
|
|
|
|
# ============================================================================
|
|
# USER PROFILE VIEWS
|
|
# ============================================================================
|
|
|
|
class UserProfileView(LoginRequiredMixin, DetailView):
|
|
"""
|
|
User's own profile view.
|
|
Shows profile information, statistics, and recent activity.
|
|
"""
|
|
model = User
|
|
template_name = 'core/user_profile.html'
|
|
context_object_name = 'profile_user'
|
|
|
|
def get_object(self, queryset=None):
|
|
"""Return the current logged-in user."""
|
|
return self.request.user
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add user statistics and activity to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
user = self.object
|
|
|
|
# Profile completion
|
|
context['profile_completion'] = user.get_profile_completion()
|
|
|
|
# User statistics (for clinical staff)
|
|
if user.role in [User.Role.DOCTOR, User.Role.NURSE, User.Role.OT,
|
|
User.Role.SLP, User.Role.ABA]:
|
|
context['user_stats'] = self._get_clinical_stats(user)
|
|
|
|
# Recent activity
|
|
context['recent_activity'] = AuditLog.objects.filter(
|
|
user=user
|
|
).select_related('content_type').order_by('-timestamp')[:10]
|
|
|
|
# Account info
|
|
context['account_age_days'] = (timezone.now().date() - user.date_joined.date()).days
|
|
context['last_login_formatted'] = user.last_login
|
|
|
|
return context
|
|
|
|
def _get_clinical_stats(self, user):
|
|
"""Get statistics for clinical staff."""
|
|
from appointments.models import Appointment
|
|
|
|
today = timezone.now().date()
|
|
this_month_start = today.replace(day=1)
|
|
|
|
try:
|
|
provider = user.provider_profile
|
|
|
|
# Appointments stats
|
|
total_appointments = Appointment.objects.filter(provider=provider).count()
|
|
this_month_appointments = Appointment.objects.filter(
|
|
provider=provider,
|
|
scheduled_date__gte=this_month_start
|
|
).count()
|
|
completed_appointments = Appointment.objects.filter(
|
|
provider=provider,
|
|
status=Appointment.Status.COMPLETED
|
|
).count()
|
|
|
|
# Patient stats
|
|
unique_patients = Appointment.objects.filter(
|
|
provider=provider
|
|
).values('patient').distinct().count()
|
|
|
|
return {
|
|
'total_appointments': total_appointments,
|
|
'this_month_appointments': this_month_appointments,
|
|
'completed_appointments': completed_appointments,
|
|
'unique_patients': unique_patients,
|
|
}
|
|
except:
|
|
return {}
|
|
|
|
|
|
class UserProfileUpdateView(LoginRequiredMixin, SuccessMessageMixin, UpdateView):
|
|
"""
|
|
User profile edit view.
|
|
Allows users to update their own profile information.
|
|
"""
|
|
model = User
|
|
form_class = None # Will be set in get_form_class
|
|
template_name = 'core/user_profile_edit.html'
|
|
success_message = _("Profile updated successfully!")
|
|
|
|
def get_object(self, queryset=None):
|
|
"""Return the current logged-in user."""
|
|
return self.request.user
|
|
|
|
def get_form_class(self):
|
|
"""Return appropriate form based on context."""
|
|
from .forms import UserProfileForm, UserPreferencesForm
|
|
|
|
# Check if we're editing preferences
|
|
if self.request.GET.get('section') == 'preferences':
|
|
return UserPreferencesForm
|
|
|
|
return UserProfileForm
|
|
|
|
def get_form_kwargs(self):
|
|
"""Add user to form kwargs for preferences form."""
|
|
kwargs = super().get_form_kwargs()
|
|
|
|
if self.request.GET.get('section') == 'preferences':
|
|
# For preferences form, remove instance and add user
|
|
kwargs.pop('instance', None)
|
|
kwargs['user'] = self.request.user
|
|
|
|
return kwargs
|
|
|
|
def get_success_url(self):
|
|
"""Redirect back to profile."""
|
|
return reverse_lazy('core:user_profile')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add section info to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
context['section'] = self.request.GET.get('section', 'profile')
|
|
return context
|
|
|
|
|
|
class UserPasswordChangeView(LoginRequiredMixin, SuccessMessageMixin, TemplateView):
|
|
"""
|
|
Password change view for users.
|
|
"""
|
|
template_name = 'core/user_password_change.html'
|
|
success_message = _("Password changed successfully!")
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add password change form to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
from .forms import UserPasswordChangeForm
|
|
|
|
if self.request.method == 'POST':
|
|
context['form'] = UserPasswordChangeForm(
|
|
user=self.request.user,
|
|
data=self.request.POST
|
|
)
|
|
else:
|
|
context['form'] = UserPasswordChangeForm(user=self.request.user)
|
|
|
|
return context
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
"""Handle password change form submission."""
|
|
from django.contrib.auth import update_session_auth_hash
|
|
from .forms import UserPasswordChangeForm
|
|
|
|
form = UserPasswordChangeForm(user=request.user, data=request.POST)
|
|
|
|
if form.is_valid():
|
|
user = form.save()
|
|
# Keep user logged in after password change
|
|
update_session_auth_hash(request, user)
|
|
|
|
messages.success(request, self.success_message)
|
|
return redirect('core:user_profile')
|
|
|
|
# If form is invalid, re-render with errors
|
|
context = self.get_context_data(**kwargs)
|
|
context['form'] = form
|
|
return self.render_to_response(context)
|
|
|
|
|
|
# ============================================================================
|
|
# STAFF MANAGEMENT VIEWS (Admin Only)
|
|
# ============================================================================
|
|
|
|
class UserListView(LoginRequiredMixin, RolePermissionMixin, PaginationMixin, ListView):
|
|
"""
|
|
Staff list view (Admin only).
|
|
Shows all staff members with search and filtering.
|
|
"""
|
|
model = User
|
|
template_name = 'core/user_list.html'
|
|
context_object_name = 'staff_members'
|
|
paginate_by = 25
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_queryset(self):
|
|
"""Get filtered and sorted queryset."""
|
|
queryset = User.objects.filter(tenant=self.request.user.tenant)
|
|
|
|
# Apply search
|
|
search_query = self.request.GET.get('search', '').strip()
|
|
if search_query:
|
|
queryset = queryset.filter(
|
|
Q(first_name__icontains=search_query) |
|
|
Q(last_name__icontains=search_query) |
|
|
Q(email__icontains=search_query) |
|
|
Q(username__icontains=search_query) |
|
|
Q(employee_id__icontains=search_query)
|
|
)
|
|
|
|
# Apply filters
|
|
role = self.request.GET.get('role')
|
|
if role:
|
|
queryset = queryset.filter(role=role)
|
|
|
|
status = self.request.GET.get('status')
|
|
if status == 'active':
|
|
queryset = queryset.filter(is_active=True)
|
|
elif status == 'inactive':
|
|
queryset = queryset.filter(is_active=False)
|
|
|
|
# Apply sorting
|
|
sort_by = self.request.GET.get('sort', '-date_joined')
|
|
valid_sort_fields = [
|
|
'first_name', '-first_name',
|
|
'last_name', '-last_name',
|
|
'email', '-email',
|
|
'role', '-role',
|
|
'date_joined', '-date_joined',
|
|
'last_login', '-last_login',
|
|
]
|
|
if sort_by in valid_sort_fields:
|
|
queryset = queryset.order_by(sort_by)
|
|
|
|
return queryset
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add search form and statistics to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
from .forms import UserSearchForm
|
|
|
|
# Add search form
|
|
context['search_form'] = UserSearchForm(self.request.GET)
|
|
|
|
# Add current filters
|
|
context['current_filters'] = {
|
|
'search': self.request.GET.get('search', ''),
|
|
'role': self.request.GET.get('role', ''),
|
|
'status': self.request.GET.get('status', ''),
|
|
'sort': self.request.GET.get('sort', '-date_joined'),
|
|
}
|
|
|
|
# Add statistics
|
|
all_staff = User.objects.filter(tenant=self.request.user.tenant)
|
|
context['stats'] = {
|
|
'total_staff': all_staff.count(),
|
|
'active_staff': all_staff.filter(is_active=True).count(),
|
|
'inactive_staff': all_staff.filter(is_active=False).count(),
|
|
}
|
|
|
|
# Role breakdown
|
|
context['role_breakdown'] = {}
|
|
for role_value, role_label in User.Role.choices:
|
|
count = all_staff.filter(role=role_value).count()
|
|
if count > 0:
|
|
context['role_breakdown'][role_label] = count
|
|
|
|
return context
|
|
|
|
|
|
class UserDetailView(LoginRequiredMixin, RolePermissionMixin, DetailView):
|
|
"""
|
|
Staff detail view (Admin only).
|
|
Shows detailed information about a staff member.
|
|
"""
|
|
model = User
|
|
template_name = 'core/user_detail.html'
|
|
context_object_name = 'staff_member'
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_queryset(self):
|
|
"""Filter by tenant."""
|
|
return User.objects.filter(tenant=self.request.user.tenant)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add statistics and activity to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
user = self.object
|
|
|
|
# Profile completion
|
|
context['profile_completion'] = user.get_profile_completion()
|
|
|
|
# User statistics (for clinical staff)
|
|
if user.role in [User.Role.DOCTOR, User.Role.NURSE, User.Role.OT,
|
|
User.Role.SLP, User.Role.ABA]:
|
|
context['user_stats'] = self._get_clinical_stats(user)
|
|
|
|
# Recent activity
|
|
context['recent_activity'] = AuditLog.objects.filter(
|
|
user=user
|
|
).select_related('content_type').order_by('-timestamp')[:20]
|
|
|
|
# Account info
|
|
context['account_age_days'] = (timezone.now().date() - user.date_joined.date()).days
|
|
|
|
# Login history
|
|
context['login_history'] = user.history.filter(
|
|
history_type='~'
|
|
).order_by('-history_date')[:10] if hasattr(user, 'history') else []
|
|
|
|
return context
|
|
|
|
def _get_clinical_stats(self, user):
|
|
"""Get statistics for clinical staff."""
|
|
from appointments.models import Appointment
|
|
|
|
today = timezone.now().date()
|
|
this_month_start = today.replace(day=1)
|
|
|
|
try:
|
|
provider = user.provider_profile
|
|
|
|
# Appointments stats
|
|
total_appointments = Appointment.objects.filter(provider=provider).count()
|
|
this_month_appointments = Appointment.objects.filter(
|
|
provider=provider,
|
|
scheduled_date__gte=this_month_start
|
|
).count()
|
|
completed_appointments = Appointment.objects.filter(
|
|
provider=provider,
|
|
status=Appointment.Status.COMPLETED
|
|
).count()
|
|
|
|
# Patient stats
|
|
unique_patients = Appointment.objects.filter(
|
|
provider=provider
|
|
).values('patient').distinct().count()
|
|
|
|
return {
|
|
'total_appointments': total_appointments,
|
|
'this_month_appointments': this_month_appointments,
|
|
'completed_appointments': completed_appointments,
|
|
'unique_patients': unique_patients,
|
|
}
|
|
except:
|
|
return {}
|
|
|
|
|
|
class UserCreateView(LoginRequiredMixin, RolePermissionMixin, AuditLogMixin,
|
|
SuccessMessageMixin, CreateView):
|
|
"""
|
|
Staff creation view (Admin only).
|
|
"""
|
|
model = User
|
|
template_name = 'core/user_form.html'
|
|
success_message = _("Staff member created successfully!")
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_form_class(self):
|
|
"""Return admin form."""
|
|
from .forms import UserAdminForm
|
|
return UserAdminForm
|
|
|
|
def form_valid(self, form):
|
|
"""Set tenant before saving."""
|
|
form.instance.tenant = self.request.user.tenant
|
|
return super().form_valid(form)
|
|
|
|
def get_success_url(self):
|
|
"""Redirect to staff detail page."""
|
|
return reverse_lazy('core:user_detail', kwargs={'pk': self.object.pk})
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add form title to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
context['form_title'] = _('Create New Staff Member')
|
|
context['submit_text'] = _('Create Staff Member')
|
|
return context
|
|
|
|
|
|
class UserUpdateView(LoginRequiredMixin, RolePermissionMixin, AuditLogMixin,
|
|
SuccessMessageMixin, UpdateView):
|
|
"""
|
|
Staff update view (Admin only).
|
|
"""
|
|
model = User
|
|
template_name = 'core/user_form.html'
|
|
success_message = _("Staff member updated successfully!")
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_form_class(self):
|
|
"""Return admin form."""
|
|
from .forms import UserAdminForm
|
|
return UserAdminForm
|
|
|
|
def get_queryset(self):
|
|
"""Filter by tenant."""
|
|
return User.objects.filter(tenant=self.request.user.tenant)
|
|
|
|
def get_success_url(self):
|
|
"""Redirect to staff detail page."""
|
|
return reverse_lazy('core:user_detail', kwargs={'pk': self.object.pk})
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add form title to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
context['form_title'] = _(f'Update Staff Member: {self.object.get_full_name()}')
|
|
context['submit_text'] = _('Update Staff Member')
|
|
return context
|
|
|
|
|
|
class UserDeactivateView(LoginRequiredMixin, RolePermissionMixin, AuditLogMixin, View):
|
|
"""
|
|
Staff deactivation/activation view (Admin only).
|
|
Toggles is_active status.
|
|
"""
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
"""Toggle user active status."""
|
|
user_id = kwargs.get('pk')
|
|
|
|
try:
|
|
user = User.objects.get(pk=user_id, tenant=request.user.tenant)
|
|
|
|
# Prevent self-deactivation
|
|
if user.pk == request.user.pk:
|
|
messages.error(request, _("You cannot deactivate your own account"))
|
|
return redirect('core:user_detail', pk=user_id)
|
|
|
|
# Toggle status
|
|
user.is_active = not user.is_active
|
|
user.save(update_fields=['is_active'])
|
|
|
|
# Success message
|
|
if user.is_active:
|
|
messages.success(request, _(f"{user.get_full_name()} has been activated"))
|
|
else:
|
|
messages.success(request, _(f"{user.get_full_name()} has been deactivated"))
|
|
|
|
return redirect('core:user_detail', pk=user_id)
|
|
|
|
except User.DoesNotExist:
|
|
messages.error(request, _("Staff member not found"))
|
|
return redirect('core:user_list')
|
|
|
|
|
|
# ============================================================================
|
|
# EMPLOYEE HR SELF-SERVICE VIEWS
|
|
# ============================================================================
|
|
|
|
class EmployeeHRDashboardView(LoginRequiredMixin, TemplateView):
|
|
"""
|
|
Employee HR dashboard showing attendance, schedule, and leave information.
|
|
"""
|
|
template_name = 'core/employee_hr_dashboard.html'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add HR data to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
user = self.request.user
|
|
today = timezone.now().date()
|
|
|
|
from hr.models import Attendance, Schedule, LeaveRequest, LeaveBalance, Holiday
|
|
|
|
# Today's attendance
|
|
try:
|
|
today_attendance = Attendance.objects.get(
|
|
employee=user,
|
|
date=today,
|
|
tenant=user.tenant
|
|
)
|
|
context['today_attendance'] = today_attendance
|
|
except Attendance.DoesNotExist:
|
|
context['today_attendance'] = None
|
|
|
|
# This week's schedule
|
|
from datetime import timedelta
|
|
weekday_map = {
|
|
0: 'MON', 1: 'TUE', 2: 'WED', 3: 'THU',
|
|
4: 'FRI', 5: 'SAT', 6: 'SUN'
|
|
}
|
|
|
|
week_schedule = []
|
|
for i in range(7):
|
|
day = today + timedelta(days=i)
|
|
day_code = weekday_map[day.weekday()]
|
|
|
|
try:
|
|
schedule = Schedule.objects.get(
|
|
employee=user,
|
|
day_of_week=day_code,
|
|
is_active=True,
|
|
tenant=user.tenant
|
|
)
|
|
week_schedule.append({
|
|
'date': day,
|
|
'schedule': schedule,
|
|
'is_today': day == today
|
|
})
|
|
except Schedule.DoesNotExist:
|
|
week_schedule.append({
|
|
'date': day,
|
|
'schedule': None,
|
|
'is_today': day == today
|
|
})
|
|
|
|
context['week_schedule'] = week_schedule
|
|
|
|
# Leave requests summary
|
|
pending_leaves = LeaveRequest.objects.filter(
|
|
employee=user,
|
|
status=LeaveRequest.Status.PENDING,
|
|
tenant=user.tenant
|
|
).count()
|
|
|
|
approved_leaves = LeaveRequest.objects.filter(
|
|
employee=user,
|
|
status=LeaveRequest.Status.APPROVED,
|
|
start_date__gte=today,
|
|
tenant=user.tenant
|
|
).order_by('start_date')[:3]
|
|
|
|
context['pending_leaves_count'] = pending_leaves
|
|
context['upcoming_leaves'] = approved_leaves
|
|
|
|
# Leave balance for current year
|
|
current_year = today.year
|
|
leave_balances = LeaveBalance.objects.filter(
|
|
employee=user,
|
|
year=current_year,
|
|
tenant=user.tenant
|
|
)
|
|
context['leave_balances'] = leave_balances
|
|
|
|
# Upcoming holidays
|
|
upcoming_holidays = Holiday.objects.filter(
|
|
date__gte=today,
|
|
tenant=user.tenant
|
|
).order_by('date')[:5]
|
|
context['upcoming_holidays'] = upcoming_holidays
|
|
|
|
# Attendance statistics (last 30 days)
|
|
thirty_days_ago = today - timedelta(days=30)
|
|
recent_attendance = Attendance.objects.filter(
|
|
employee=user,
|
|
date__gte=thirty_days_ago,
|
|
tenant=user.tenant
|
|
)
|
|
|
|
context['attendance_stats'] = {
|
|
'total_days': recent_attendance.count(),
|
|
'present_days': recent_attendance.filter(status=Attendance.Status.PRESENT).count(),
|
|
'late_days': recent_attendance.filter(status=Attendance.Status.LATE).count(),
|
|
'absent_days': recent_attendance.filter(status=Attendance.Status.ABSENT).count(),
|
|
'total_hours': sum(a.hours_worked or 0 for a in recent_attendance),
|
|
}
|
|
|
|
return context
|
|
|
|
|
|
class EmployeeAttendanceView(LoginRequiredMixin, ListView):
|
|
"""
|
|
Employee's own attendance records.
|
|
"""
|
|
model = None # Will be set in get_queryset
|
|
template_name = 'core/employee_attendance.html'
|
|
context_object_name = 'attendances'
|
|
paginate_by = 31 # One month
|
|
|
|
def get_queryset(self):
|
|
"""Get employee's own attendance records."""
|
|
from hr.models import Attendance
|
|
|
|
queryset = Attendance.objects.filter(
|
|
employee=self.request.user,
|
|
tenant=self.request.user.tenant
|
|
)
|
|
|
|
# Filter by month/year if provided
|
|
month = self.request.GET.get('month')
|
|
year = self.request.GET.get('year')
|
|
|
|
if month and year:
|
|
queryset = queryset.filter(date__month=month, date__year=year)
|
|
elif not month and not year:
|
|
# Default to current month
|
|
today = timezone.now().date()
|
|
queryset = queryset.filter(date__month=today.month, date__year=today.year)
|
|
|
|
return queryset.order_by('-date')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add month selector and statistics."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
# Current month/year or selected
|
|
today = timezone.now().date()
|
|
selected_month = int(self.request.GET.get('month', today.month))
|
|
selected_year = int(self.request.GET.get('year', today.year))
|
|
|
|
context['selected_month'] = selected_month
|
|
context['selected_year'] = selected_year
|
|
|
|
# Month options
|
|
context['months'] = [
|
|
(1, _('January')), (2, _('February')), (3, _('March')),
|
|
(4, _('April')), (5, _('May')), (6, _('June')),
|
|
(7, _('July')), (8, _('August')), (9, _('September')),
|
|
(10, _('October')), (11, _('November')), (12, _('December'))
|
|
]
|
|
|
|
# Year options (current year and 2 years back)
|
|
context['years'] = range(today.year, today.year - 3, -1)
|
|
|
|
# Statistics for selected month
|
|
attendances = self.get_queryset()
|
|
context['month_stats'] = {
|
|
'total_days': attendances.count(),
|
|
'present': attendances.filter(status='PRESENT').count(),
|
|
'late': attendances.filter(status='LATE').count(),
|
|
'absent': attendances.filter(status='ABSENT').count(),
|
|
'on_leave': attendances.filter(status='LEAVE').count(),
|
|
'total_hours': sum(a.hours_worked or 0 for a in attendances),
|
|
}
|
|
|
|
return context
|
|
|
|
|
|
class EmployeeScheduleView(LoginRequiredMixin, TemplateView):
|
|
"""
|
|
Employee's work schedule view.
|
|
"""
|
|
template_name = 'core/employee_schedule.html'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add schedule data."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
from hr.models import Schedule
|
|
|
|
# Get employee's schedules
|
|
schedules = Schedule.objects.filter(
|
|
employee=self.request.user,
|
|
tenant=self.request.user.tenant,
|
|
is_active=True
|
|
).order_by('day_of_week')
|
|
|
|
# Organize by day
|
|
schedule_by_day = {}
|
|
for schedule in schedules:
|
|
schedule_by_day[schedule.day_of_week] = schedule
|
|
|
|
context['schedule_by_day'] = schedule_by_day
|
|
context['days'] = Schedule.DayOfWeek.choices
|
|
|
|
# Calculate total weekly hours
|
|
total_hours = sum(s.duration_hours for s in schedules)
|
|
context['total_weekly_hours'] = total_hours
|
|
|
|
return context
|
|
|
|
|
|
class EmployeeLeaveRequestListView(LoginRequiredMixin, ListView):
|
|
"""
|
|
Employee's leave requests list.
|
|
"""
|
|
model = None
|
|
template_name = 'core/employee_leave_requests.html'
|
|
context_object_name = 'leave_requests'
|
|
paginate_by = 20
|
|
|
|
def get_queryset(self):
|
|
"""Get employee's own leave requests."""
|
|
from hr.models import LeaveRequest
|
|
|
|
queryset = LeaveRequest.objects.filter(
|
|
employee=self.request.user,
|
|
tenant=self.request.user.tenant
|
|
).select_related('reviewed_by')
|
|
|
|
# Filter by status if provided
|
|
status = self.request.GET.get('status')
|
|
if status:
|
|
queryset = queryset.filter(status=status)
|
|
|
|
return queryset.order_by('-created_at')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add leave balance and statistics."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
from hr.models import LeaveBalance, LeaveRequest
|
|
|
|
# Current year leave balances
|
|
current_year = timezone.now().year
|
|
leave_balances = LeaveBalance.objects.filter(
|
|
employee=self.request.user,
|
|
year=current_year,
|
|
tenant=self.request.user.tenant
|
|
)
|
|
context['leave_balances'] = leave_balances
|
|
|
|
# Leave request statistics
|
|
all_requests = LeaveRequest.objects.filter(
|
|
employee=self.request.user,
|
|
tenant=self.request.user.tenant
|
|
)
|
|
|
|
context['leave_stats'] = {
|
|
'total': all_requests.count(),
|
|
'pending': all_requests.filter(status=LeaveRequest.Status.PENDING).count(),
|
|
'approved': all_requests.filter(status=LeaveRequest.Status.APPROVED).count(),
|
|
'rejected': all_requests.filter(status=LeaveRequest.Status.REJECTED).count(),
|
|
}
|
|
|
|
# Status filter options
|
|
context['status_filter'] = self.request.GET.get('status', '')
|
|
context['status_choices'] = LeaveRequest.Status.choices
|
|
|
|
return context
|
|
|
|
|
|
class EmployeeLeaveRequestCreateView(LoginRequiredMixin, SuccessMessageMixin, CreateView):
|
|
"""
|
|
Create a new leave request.
|
|
"""
|
|
template_name = 'core/employee_leave_request_form.html'
|
|
success_message = _("Leave request submitted successfully!")
|
|
|
|
def get_queryset(self):
|
|
"""Get LeaveRequest model."""
|
|
from hr.models import LeaveRequest
|
|
return LeaveRequest.objects.filter(
|
|
employee=self.request.user,
|
|
tenant=self.request.user.tenant
|
|
)
|
|
|
|
def get_form_class(self):
|
|
"""Return form class."""
|
|
from hr.forms import LeaveRequestForm
|
|
return LeaveRequestForm
|
|
|
|
def form_valid(self, form):
|
|
"""Set employee and tenant."""
|
|
from hr.models import LeaveRequest
|
|
|
|
form.instance.employee = self.request.user
|
|
form.instance.tenant = self.request.user.tenant
|
|
|
|
# Calculate days
|
|
if form.instance.start_date and form.instance.end_date:
|
|
delta = form.instance.end_date - form.instance.start_date
|
|
form.instance.days_requested = delta.days + 1
|
|
|
|
return super().form_valid(form)
|
|
|
|
def get_success_url(self):
|
|
"""Redirect to leave requests list."""
|
|
return reverse_lazy('core:employee_leave_requests')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add leave balances to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
from hr.models import LeaveBalance
|
|
|
|
# Current year leave balances
|
|
current_year = timezone.now().year
|
|
leave_balances = LeaveBalance.objects.filter(
|
|
employee=self.request.user,
|
|
year=current_year,
|
|
tenant=self.request.user.tenant
|
|
)
|
|
context['leave_balances'] = leave_balances
|
|
|
|
return context
|
|
|
|
|
|
class EmployeeLeaveRequestDetailView(LoginRequiredMixin, DetailView):
|
|
"""
|
|
View leave request details.
|
|
"""
|
|
model = None
|
|
template_name = 'core/employee_leave_request_detail.html'
|
|
context_object_name = 'leave_request'
|
|
|
|
def get_queryset(self):
|
|
"""Get employee's own leave requests only."""
|
|
from hr.models import LeaveRequest
|
|
|
|
return LeaveRequest.objects.filter(
|
|
employee=self.request.user,
|
|
tenant=self.request.user.tenant
|
|
).select_related('reviewed_by')
|
|
|
|
|
|
class EmployeeHolidaysView(LoginRequiredMixin, ListView):
|
|
"""
|
|
View company holidays.
|
|
"""
|
|
model = None
|
|
template_name = 'core/employee_holidays.html'
|
|
context_object_name = 'holidays'
|
|
paginate_by = 50
|
|
|
|
def get_queryset(self):
|
|
"""Get holidays for current and next year."""
|
|
from hr.models import Holiday
|
|
|
|
current_year = timezone.now().year
|
|
|
|
queryset = Holiday.objects.filter(
|
|
tenant=self.request.user.tenant
|
|
).filter(
|
|
Q(date__year=current_year) |
|
|
Q(date__year=current_year + 1) |
|
|
Q(is_recurring=True)
|
|
)
|
|
|
|
# Filter by year if provided
|
|
year = self.request.GET.get('year')
|
|
if year:
|
|
queryset = queryset.filter(date__year=year)
|
|
|
|
return queryset.order_by('date')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add year selector."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
current_year = timezone.now().year
|
|
context['current_year'] = current_year
|
|
context['years'] = range(current_year, current_year + 3)
|
|
context['selected_year'] = self.request.GET.get('year', current_year)
|
|
|
|
return context
|
|
|
|
|
|
class EmployeeClockInOutView(LoginRequiredMixin, View):
|
|
"""
|
|
Quick clock in/out endpoint.
|
|
"""
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
"""Handle clock in/out."""
|
|
from hr.models import Attendance
|
|
|
|
today = timezone.now().date()
|
|
now = timezone.now().time()
|
|
|
|
attendance, created = Attendance.objects.get_or_create(
|
|
employee=request.user,
|
|
date=today,
|
|
tenant=request.user.tenant,
|
|
defaults={'check_in': now, 'status': Attendance.Status.PRESENT}
|
|
)
|
|
|
|
if not created and not attendance.check_out:
|
|
# Clock out
|
|
attendance.check_out = now
|
|
attendance.save()
|
|
messages.success(request, _('Clocked out successfully at {}').format(now.strftime('%H:%M')))
|
|
elif created:
|
|
messages.success(request, _('Clocked in successfully at {}').format(now.strftime('%H:%M')))
|
|
else:
|
|
messages.warning(request, _('You have already clocked out for today'))
|
|
|
|
# Redirect back to referring page or HR dashboard
|
|
return redirect(request.META.get('HTTP_REFERER', 'core:employee_hr_dashboard'))
|
|
|
|
|
|
# ============================================================================
|
|
# TENANT SETTINGS VIEWS
|
|
# ============================================================================
|
|
|
|
class TenantSettingsView(LoginRequiredMixin, RolePermissionMixin, TemplateView):
|
|
"""
|
|
Main view for tenant settings management.
|
|
Shows all setting categories with completion status.
|
|
"""
|
|
template_name = 'core/settings/settings_dashboard.html'
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
# Get user's tenant
|
|
tenant = self.request.user.tenant
|
|
if not tenant:
|
|
context['error'] = _('No tenant associated with your account')
|
|
return context
|
|
|
|
# Get settings service
|
|
from .settings_service import get_tenant_settings_service
|
|
service = get_tenant_settings_service(tenant)
|
|
|
|
# Get all categories
|
|
from .models import SettingTemplate
|
|
categories = SettingTemplate.Category.choices
|
|
|
|
# Get settings status for each category
|
|
category_status = []
|
|
for category_value, category_label in categories:
|
|
templates = SettingTemplate.objects.filter(
|
|
category=category_value,
|
|
is_active=True
|
|
)
|
|
|
|
required_templates = templates.filter(is_required=True)
|
|
missing = service.get_missing_required_settings()
|
|
missing_in_category = [t for t in missing if t.category == category_value]
|
|
|
|
total_count = templates.count()
|
|
required_count = required_templates.count()
|
|
missing_count = len(missing_in_category)
|
|
|
|
category_status.append({
|
|
'value': category_value,
|
|
'label': category_label,
|
|
'total_count': total_count,
|
|
'required_count': required_count,
|
|
'missing_count': missing_count,
|
|
'is_complete': missing_count == 0,
|
|
})
|
|
|
|
context['tenant'] = tenant
|
|
context['categories'] = category_status
|
|
context['overall_complete'] = service.validate_required_settings()
|
|
|
|
return context
|
|
|
|
|
|
class CategorySettingsView(LoginRequiredMixin, RolePermissionMixin, TemplateView):
|
|
"""
|
|
View for editing settings in a specific category.
|
|
"""
|
|
template_name = 'core/settings/category_settings.html'
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
category = kwargs.get('category')
|
|
tenant = self.request.user.tenant
|
|
|
|
if not tenant:
|
|
context['error'] = _('No tenant associated with your account')
|
|
return context
|
|
|
|
# Validate category
|
|
from .models import SettingTemplate
|
|
valid_categories = [c[0] for c in SettingTemplate.Category.choices]
|
|
if category not in valid_categories:
|
|
context['error'] = _('Invalid category')
|
|
return context
|
|
|
|
# Get category label
|
|
category_label = dict(SettingTemplate.Category.choices).get(category)
|
|
|
|
# Get form
|
|
from .forms import TenantSettingsForm
|
|
if self.request.method == 'POST':
|
|
form = TenantSettingsForm(
|
|
self.request.POST,
|
|
self.request.FILES,
|
|
tenant=tenant,
|
|
category=category,
|
|
user=self.request.user
|
|
)
|
|
else:
|
|
form = TenantSettingsForm(
|
|
tenant=tenant,
|
|
category=category,
|
|
user=self.request.user
|
|
)
|
|
|
|
context['form'] = form
|
|
context['category'] = category
|
|
context['category_label'] = category_label
|
|
context['tenant'] = tenant
|
|
|
|
return context
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
"""Handle form submission."""
|
|
category = kwargs.get('category')
|
|
tenant = request.user.tenant
|
|
|
|
from .forms import TenantSettingsForm
|
|
form = TenantSettingsForm(
|
|
request.POST,
|
|
request.FILES,
|
|
tenant=tenant,
|
|
category=category,
|
|
user=request.user
|
|
)
|
|
|
|
if form.is_valid():
|
|
# Save settings
|
|
saved_count = form.save()
|
|
|
|
messages.success(
|
|
request,
|
|
_(f'Successfully saved {saved_count} settings')
|
|
)
|
|
|
|
# Redirect back to settings dashboard
|
|
return redirect('core:tenant_settings')
|
|
|
|
# If form is invalid, re-render with errors
|
|
context = self.get_context_data(**kwargs)
|
|
context['form'] = form
|
|
return self.render_to_response(context)
|
|
|
|
|
|
class SettingsExportView(LoginRequiredMixin, RolePermissionMixin, TemplateView):
|
|
"""
|
|
Export tenant settings as JSON.
|
|
"""
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
"""Export settings to JSON."""
|
|
tenant = request.user.tenant
|
|
|
|
if not tenant:
|
|
return JsonResponse({'error': 'No tenant associated'}, status=400)
|
|
|
|
from .settings_service import get_tenant_settings_service
|
|
service = get_tenant_settings_service(tenant)
|
|
|
|
# Export settings
|
|
settings_dict = service.export_settings()
|
|
|
|
# Return as JSON download
|
|
from django.http import JsonResponse
|
|
import json
|
|
|
|
response = JsonResponse(settings_dict)
|
|
response['Content-Disposition'] = f'attachment; filename="tenant_settings_{tenant.code}_{timezone.now().date()}.json"'
|
|
|
|
return response
|
|
|
|
|
|
class SettingsImportView(LoginRequiredMixin, RolePermissionMixin, TemplateView):
|
|
"""
|
|
Import tenant settings from JSON.
|
|
"""
|
|
template_name = 'core/settings/settings_import.html'
|
|
allowed_roles = [User.Role.ADMIN]
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context['tenant'] = self.request.user.tenant
|
|
return context
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
"""Handle settings import."""
|
|
tenant = request.user.tenant
|
|
|
|
if not tenant:
|
|
messages.error(request, _('No tenant associated with your account'))
|
|
return redirect('core:tenant_settings')
|
|
|
|
# Get uploaded file
|
|
import_file = request.FILES.get('settings_file')
|
|
|
|
if not import_file:
|
|
messages.error(request, _('Please select a file to import'))
|
|
return redirect('core:settings_import')
|
|
|
|
try:
|
|
# Parse JSON
|
|
import json
|
|
settings_data = json.load(import_file)
|
|
|
|
# Import settings
|
|
from .settings_service import get_tenant_settings_service
|
|
service = get_tenant_settings_service(tenant)
|
|
|
|
count = service.import_settings(settings_data, request.user)
|
|
|
|
messages.success(
|
|
request,
|
|
_(f'Successfully imported {count} settings')
|
|
)
|
|
|
|
return redirect('core:tenant_settings')
|
|
|
|
except json.JSONDecodeError:
|
|
messages.error(request, _('Invalid JSON file'))
|
|
return redirect('core:settings_import')
|
|
except Exception as e:
|
|
messages.error(request, _(f'Error importing settings: {str(e)}'))
|
|
return redirect('core:settings_import')
|
|
|
|
|
|
class PatientDetailView(LoginRequiredMixin, TenantFilterMixin, HTMXResponseMixin, DetailView):
|
|
"""
|
|
Comprehensive patient detail view with tabs.
|
|
|
|
Tabs:
|
|
- Overview: Demographics, contact info, caregiver
|
|
- Appointments: All appointments (past, upcoming, cancelled)
|
|
- Clinical History: All clinical documents across disciplines
|
|
- Files & Consents: Patient files, sub-files, signed consents
|
|
- Audit Trail: All actions related to this patient
|
|
|
|
HTMX support for tab switching without page reload.
|
|
"""
|
|
model = Patient
|
|
template_name = 'core/patient_detail.html'
|
|
htmx_template_name = 'core/partials/patient_detail_tab.html'
|
|
context_object_name = 'patient'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add comprehensive patient data to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
patient = self.object
|
|
|
|
# Get active tab (default: overview)
|
|
active_tab = self.request.GET.get('tab', 'overview')
|
|
context['active_tab'] = active_tab
|
|
|
|
# Load data based on active tab (for HTMX requests)
|
|
if active_tab == 'overview':
|
|
context.update(self._get_overview_data(patient))
|
|
elif active_tab == 'appointments':
|
|
context.update(self._get_appointments_data(patient))
|
|
elif active_tab == 'clinical':
|
|
context.update(self._get_clinical_data(patient))
|
|
elif active_tab == 'files':
|
|
context.update(self._get_files_data(patient))
|
|
elif active_tab == 'audit':
|
|
context.update(self._get_audit_data(patient))
|
|
|
|
# For non-HTMX requests, load all data
|
|
if not self.request.headers.get('HX-Request'):
|
|
context.update(self._get_overview_data(patient))
|
|
context.update(self._get_appointments_data(patient))
|
|
context.update(self._get_clinical_data(patient))
|
|
context.update(self._get_files_data(patient))
|
|
|
|
return context
|
|
|
|
def _get_overview_data(self, patient):
|
|
"""Get patient overview data."""
|
|
return {
|
|
'age': patient.get_age() if hasattr(patient, 'get_age') else None,
|
|
'full_name_en': f"{patient.first_name_en} {patient.last_name_en}",
|
|
'full_name_ar': f"{patient.first_name_ar} {patient.last_name_ar}",
|
|
}
|
|
|
|
def _get_appointments_data(self, patient):
|
|
"""Get patient appointments data."""
|
|
from appointments.models import Appointment
|
|
|
|
appointments = Appointment.objects.filter(
|
|
patient=patient
|
|
).select_related('provider', 'clinic').order_by('-scheduled_date', '-scheduled_time')
|
|
|
|
# Categorize appointments
|
|
today = timezone.now().date()
|
|
|
|
return {
|
|
'appointments': appointments,
|
|
'appointments_upcoming': appointments.filter(
|
|
scheduled_date__gte=today,
|
|
status__in=[Appointment.Status.BOOKED, Appointment.Status.CONFIRMED]
|
|
)[:5],
|
|
'appointments_past': appointments.filter(
|
|
scheduled_date__lt=today
|
|
)[:10],
|
|
'appointments_cancelled': appointments.filter(
|
|
status__in=[Appointment.Status.CANCELLED, Appointment.Status.NO_SHOW]
|
|
)[:5],
|
|
'appointments_count': appointments.count(),
|
|
}
|
|
|
|
def _get_clinical_data(self, patient):
|
|
"""Get all clinical documents for patient."""
|
|
from nursing.models import NursingEncounter
|
|
from medical.models import MedicalConsultation, MedicalFollowUp
|
|
from aba.models import ABAConsult
|
|
from ot.models import OTConsult, OTSession
|
|
from slp.models import SLPConsult, SLPAssessment, SLPIntervention, SLPProgressReport
|
|
|
|
clinical_data = {
|
|
# Nursing
|
|
'nursing_encounters': NursingEncounter.objects.filter(
|
|
patient=patient
|
|
).order_by('-encounter_date')[:5],
|
|
'nursing_count': NursingEncounter.objects.filter(patient=patient).count(),
|
|
|
|
# Medical
|
|
'medical_consultations': MedicalConsultation.objects.filter(
|
|
patient=patient
|
|
).select_related('provider').order_by('-consultation_date')[:5],
|
|
'medical_followups': MedicalFollowUp.objects.filter(
|
|
patient=patient
|
|
).select_related('provider').order_by('-followup_date')[:5],
|
|
'medical_count': MedicalConsultation.objects.filter(patient=patient).count() +
|
|
MedicalFollowUp.objects.filter(patient=patient).count(),
|
|
|
|
# ABA
|
|
'aba_consults': ABAConsult.objects.filter(
|
|
patient=patient
|
|
).select_related('provider').order_by('-consultation_date')[:5],
|
|
'aba_count': ABAConsult.objects.filter(patient=patient).count(),
|
|
|
|
# OT
|
|
'ot_consults': OTConsult.objects.filter(
|
|
patient=patient
|
|
).select_related('provider').order_by('-consultation_date')[:5],
|
|
'ot_sessions': OTSession.objects.filter(
|
|
patient=patient
|
|
).select_related('provider').order_by('-session_date')[:5],
|
|
'ot_count': OTConsult.objects.filter(patient=patient).count() +
|
|
OTSession.objects.filter(patient=patient).count(),
|
|
|
|
# SLP
|
|
'slp_consults': SLPConsult.objects.filter(
|
|
patient=patient
|
|
).select_related('provider').order_by('-consultation_date')[:5],
|
|
'slp_assessments': SLPAssessment.objects.filter(
|
|
patient=patient
|
|
).select_related('provider').order_by('-assessment_date')[:5],
|
|
'slp_interventions': SLPIntervention.objects.filter(
|
|
patient=patient
|
|
).select_related('provider').order_by('-session_date')[:5],
|
|
'slp_count': SLPConsult.objects.filter(patient=patient).count() +
|
|
SLPAssessment.objects.filter(patient=patient).count() +
|
|
SLPIntervention.objects.filter(patient=patient).count(),
|
|
}
|
|
|
|
# Calculate total clinical documents
|
|
clinical_data['total_clinical_docs'] = sum([
|
|
clinical_data['nursing_count'],
|
|
clinical_data['medical_count'],
|
|
clinical_data['aba_count'],
|
|
clinical_data['ot_count'],
|
|
clinical_data['slp_count'],
|
|
])
|
|
|
|
return clinical_data
|
|
|
|
def _get_files_data(self, patient):
|
|
"""Get patient files and consents."""
|
|
files = File.objects.filter(patient=patient).prefetch_related('subfiles')
|
|
consents = Consent.objects.filter(patient=patient).order_by('-signed_at')
|
|
|
|
return {
|
|
'files': files,
|
|
'files_count': files.count(),
|
|
'consents': consents,
|
|
'consents_count': consents.count(),
|
|
'consents_signed': consents.filter(signed_at__isnull=False).count(),
|
|
'consents_pending': consents.filter(signed_at__isnull=True).count(),
|
|
}
|
|
|
|
def _get_audit_data(self, patient):
|
|
"""Get audit trail for patient."""
|
|
from django.contrib.contenttypes.models import ContentType
|
|
patient_ct = ContentType.objects.get_for_model(Patient)
|
|
|
|
audit_logs = AuditLog.objects.filter(
|
|
Q(content_type=patient_ct, object_id=str(patient.pk)) |
|
|
Q(changes__contains={'patient_id': str(patient.pk)})
|
|
).select_related('user').order_by('-timestamp')[:50]
|
|
|
|
return {
|
|
'audit_logs': audit_logs,
|
|
'audit_count': audit_logs.count(),
|
|
}
|
|
|
|
|
|
class PatientCreateView(LoginRequiredMixin, RolePermissionMixin, AuditLogMixin,
|
|
SuccessMessageMixin, CreateView):
|
|
"""
|
|
Patient creation view.
|
|
|
|
Features:
|
|
- Auto-generate MRN on save
|
|
- Create main File automatically via signal
|
|
- Set tenant from current user
|
|
- Role-based access (Admin, FrontDesk only)
|
|
- Bilingual form support
|
|
"""
|
|
model = Patient
|
|
form_class = PatientForm
|
|
template_name = 'core/patient_form.html'
|
|
success_message = "Patient created successfully! MRN: {mrn}"
|
|
allowed_roles = [User.Role.ADMIN, User.Role.FRONT_DESK]
|
|
|
|
def get_success_url(self):
|
|
"""Redirect to patient detail page."""
|
|
return reverse_lazy('core:patient_detail', kwargs={'pk': self.object.pk})
|
|
|
|
def form_valid(self, form):
|
|
"""Set tenant and generate MRN before saving."""
|
|
# Set tenant from current user
|
|
form.instance.tenant = self.request.user.tenant
|
|
|
|
# Generate MRN if not provided
|
|
if not form.instance.mrn:
|
|
form.instance.mrn = self._generate_mrn()
|
|
|
|
# Save the patient
|
|
response = super().form_valid(form)
|
|
|
|
# Update success message with MRN
|
|
self.success_message = self.success_message.format(mrn=self.object.mrn)
|
|
|
|
return response
|
|
|
|
def _generate_mrn(self):
|
|
"""Generate unique MRN for patient."""
|
|
import random
|
|
from django.db import IntegrityError
|
|
|
|
tenant = self.request.user.tenant
|
|
max_attempts = 10
|
|
|
|
for _ in range(max_attempts):
|
|
# Generate MRN: TENANT_CODE + YEAR + 6-digit random number
|
|
year = timezone.now().year
|
|
random_num = random.randint(100000, 999999)
|
|
mrn = f"{tenant.code}-{year}-{random_num}"
|
|
|
|
# Check if MRN already exists
|
|
if not Patient.objects.filter(mrn=mrn).exists():
|
|
return mrn
|
|
|
|
# Fallback: use timestamp-based MRN
|
|
timestamp = int(timezone.now().timestamp())
|
|
return f"{tenant.code}-{year}-{timestamp}"
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add form title to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
context['form_title'] = _('Create New Patient')
|
|
context['submit_text'] = _('Create Patient')
|
|
return context
|
|
|
|
|
|
class PatientUpdateView(LoginRequiredMixin, RolePermissionMixin, TenantFilterMixin,
|
|
AuditLogMixin, SuccessMessageMixin, UpdateView):
|
|
"""
|
|
Patient update view.
|
|
|
|
Features:
|
|
- Role-based access (Admin, FrontDesk, or assigned provider)
|
|
- Audit trail of changes
|
|
- Version history via simple_history
|
|
- Cannot change MRN or tenant
|
|
"""
|
|
model = Patient
|
|
form_class = PatientForm
|
|
template_name = 'core/patient_form.html'
|
|
success_message = "Patient updated successfully!"
|
|
allowed_roles = [User.Role.ADMIN, User.Role.FRONT_DESK, User.Role.DOCTOR,
|
|
User.Role.NURSE, User.Role.OT, User.Role.SLP, User.Role.ABA]
|
|
|
|
def get_success_url(self):
|
|
"""Redirect to patient detail page."""
|
|
return reverse_lazy('core:patient_detail', kwargs={'pk': self.object.pk})
|
|
|
|
def get_form(self, form_class=None):
|
|
"""Disable MRN field for updates."""
|
|
form = super().get_form(form_class)
|
|
|
|
# Make MRN read-only
|
|
if 'mrn' in form.fields:
|
|
form.fields['mrn'].disabled = True
|
|
form.fields['mrn'].help_text = 'MRN cannot be changed after creation'
|
|
|
|
return form
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add form title and history to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
context['form_title'] = _('Update Patient: %(mrn)s') % {'mrn': self.object.mrn}
|
|
context['submit_text'] = _('Update Patient')
|
|
|
|
# Add version history if simple_history is available
|
|
if hasattr(self.object, 'history'):
|
|
context['history'] = self.object.history.all()[:10]
|
|
|
|
return context
|
|
|
|
|
|
class ConsentCreateView(LoginRequiredMixin, RolePermissionMixin, AuditLogMixin,
|
|
SuccessMessageMixin, CreateView):
|
|
"""
|
|
Consent creation view with e-signature support.
|
|
|
|
Features:
|
|
- E-signature canvas integration
|
|
- Capture signature metadata (IP, user agent, timestamp)
|
|
- Generate immutable PDF artifact
|
|
- Link to patient
|
|
- Consent type selection
|
|
- Bilingual consent text
|
|
"""
|
|
model = Consent
|
|
form_class = ConsentForm
|
|
template_name = 'clinic/consent_form.html'
|
|
success_message = "Consent signed successfully!"
|
|
allowed_roles = [User.Role.ADMIN, User.Role.FRONT_DESK, User.Role.DOCTOR,
|
|
User.Role.NURSE, User.Role.OT, User.Role.SLP, User.Role.ABA]
|
|
|
|
def get_success_url(self):
|
|
"""Redirect to patient detail page."""
|
|
return reverse_lazy('core:patient_detail', kwargs={'pk': self.object.patient.pk})
|
|
|
|
def form_valid(self, form):
|
|
"""Capture signature metadata before saving."""
|
|
# Set tenant
|
|
form.instance.tenant = self.request.user.tenant
|
|
|
|
# Capture signature metadata
|
|
form.instance.signed_by_name = form.cleaned_data.get('signed_by_name', '')
|
|
form.instance.signed_at = timezone.now()
|
|
form.instance.signed_ip = self._get_client_ip()
|
|
form.instance.signed_user_agent = self.request.META.get('HTTP_USER_AGENT', '')[:255]
|
|
|
|
# Calculate signature hash
|
|
if form.cleaned_data.get('signature_image'):
|
|
form.instance.signature_hash = self._calculate_signature_hash(
|
|
form.cleaned_data['signature_image']
|
|
)
|
|
|
|
# Save consent
|
|
response = super().form_valid(form)
|
|
|
|
# Generate PDF artifact (async task in production)
|
|
self._generate_consent_pdf()
|
|
|
|
return response
|
|
|
|
def _get_client_ip(self):
|
|
"""Get client IP address."""
|
|
x_forwarded_for = self.request.META.get('HTTP_X_FORWARDED_FOR')
|
|
if x_forwarded_for:
|
|
ip = x_forwarded_for.split(',')[0]
|
|
else:
|
|
ip = self.request.META.get('REMOTE_ADDR')
|
|
return ip
|
|
|
|
def _calculate_signature_hash(self, signature_image):
|
|
"""Calculate hash of signature for verification."""
|
|
import hashlib
|
|
|
|
if hasattr(signature_image, 'read'):
|
|
content = signature_image.read()
|
|
signature_image.seek(0) # Reset file pointer
|
|
else:
|
|
content = str(signature_image).encode()
|
|
|
|
return hashlib.sha256(content).hexdigest()
|
|
|
|
def _generate_consent_pdf(self):
|
|
"""Generate PDF artifact of signed consent."""
|
|
# TODO: Implement PDF generation
|
|
# This would use reportlab or weasyprint to generate a PDF
|
|
# containing the consent text, signature, and metadata
|
|
pass
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add consent text and patient info to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
# Get patient if provided in URL
|
|
patient_id = self.request.GET.get('patient')
|
|
if patient_id:
|
|
try:
|
|
context['patient'] = Patient.objects.get(pk=patient_id, tenant=self.request.user.tenant)
|
|
except Patient.DoesNotExist:
|
|
pass
|
|
|
|
context['form_title'] = _('Sign Consent Form')
|
|
context['submit_text'] = _('Sign & Submit')
|
|
|
|
return context
|
|
|
|
|
|
class ConsentListView(LoginRequiredMixin, TenantFilterMixin, PaginationMixin, ListView):
|
|
"""
|
|
Consent list view.
|
|
|
|
Features:
|
|
- List all consents
|
|
- Filter by patient, consent type, status
|
|
- Search by patient name
|
|
- Sort by date
|
|
"""
|
|
model = Consent
|
|
template_name = 'clinic/consent_list.html'
|
|
context_object_name = 'consents'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
"""Get filtered and sorted queryset."""
|
|
queryset = super().get_queryset()
|
|
|
|
# Apply search
|
|
search_query = self.request.GET.get('search', '').strip()
|
|
if search_query:
|
|
queryset = queryset.filter(
|
|
Q(patient__first_name_en__icontains=search_query) |
|
|
Q(patient__last_name_en__icontains=search_query) |
|
|
Q(patient__mrn__icontains=search_query) |
|
|
Q(signed_by_name__icontains=search_query)
|
|
)
|
|
|
|
# Apply filters
|
|
patient_id = self.request.GET.get('patient')
|
|
if patient_id:
|
|
queryset = queryset.filter(patient_id=patient_id)
|
|
|
|
consent_type = self.request.GET.get('consent_type')
|
|
if consent_type:
|
|
queryset = queryset.filter(consent_type=consent_type)
|
|
|
|
# Filter by signed/unsigned
|
|
status = self.request.GET.get('status')
|
|
if status == 'signed':
|
|
queryset = queryset.filter(signed_at__isnull=False)
|
|
elif status == 'unsigned':
|
|
queryset = queryset.filter(signed_at__isnull=True)
|
|
|
|
return queryset.select_related('patient', 'tenant').order_by('-created_at')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add filter options to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
context['consent_types'] = Consent.ConsentType.choices
|
|
context['current_filters'] = {
|
|
'search': self.request.GET.get('search', ''),
|
|
'patient': self.request.GET.get('patient', ''),
|
|
'consent_type': self.request.GET.get('consent_type', ''),
|
|
'status': self.request.GET.get('status', ''),
|
|
}
|
|
|
|
return context
|
|
|
|
|
|
class ConsentDetailView(LoginRequiredMixin, TenantFilterMixin, DetailView):
|
|
"""
|
|
Consent detail view - immutable display.
|
|
|
|
Features:
|
|
- Display signed consent with signature
|
|
- Show signature metadata (who, when, where)
|
|
- Download PDF artifact
|
|
- Audit trail
|
|
- Verification of signature hash
|
|
"""
|
|
model = Consent
|
|
template_name = 'clinic/consent_detail.html'
|
|
context_object_name = 'consent'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add verification and audit data to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
consent = self.object
|
|
|
|
# Verify signature hash if available
|
|
if consent.signature_hash and consent.signature_image:
|
|
context['signature_verified'] = self._verify_signature_hash(consent)
|
|
else:
|
|
context['signature_verified'] = None
|
|
|
|
# Get audit trail for this consent
|
|
from django.contrib.contenttypes.models import ContentType
|
|
consent_ct = ContentType.objects.get_for_model(Consent)
|
|
context['audit_logs'] = AuditLog.objects.filter(
|
|
content_type=consent_ct,
|
|
object_id=str(consent.pk)
|
|
).select_related('user').order_by('-timestamp')[:10]
|
|
|
|
# Check if PDF is available
|
|
context['pdf_available'] = hasattr(consent, 'pdf_file') and consent.pdf_file
|
|
|
|
return context
|
|
|
|
def _verify_signature_hash(self, consent):
|
|
"""Verify signature hash matches current signature image."""
|
|
import hashlib
|
|
|
|
try:
|
|
if hasattr(consent.signature_image, 'read'):
|
|
content = consent.signature_image.read()
|
|
consent.signature_image.seek(0)
|
|
current_hash = hashlib.sha256(content).hexdigest()
|
|
return current_hash == consent.signature_hash
|
|
except Exception:
|
|
return False
|
|
|
|
return False
|
|
|
|
|
|
class FileHistoryView(LoginRequiredMixin, TenantFilterMixin, PaginationMixin, ListView):
|
|
"""
|
|
File access history view.
|
|
|
|
Features:
|
|
- Shows all file access/modifications
|
|
- Filter by patient, user, action type
|
|
- Date range filtering
|
|
- Audit compliance reporting
|
|
- Export to CSV for compliance
|
|
"""
|
|
model = AuditLog
|
|
template_name = 'core/file_history.html'
|
|
context_object_name = 'audit_logs'
|
|
paginate_by = 50
|
|
|
|
def get_queryset(self):
|
|
"""Get filtered audit logs."""
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
queryset = super().get_queryset()
|
|
|
|
# Filter by file-related models using ContentType
|
|
try:
|
|
file_content_types = ContentType.objects.filter(
|
|
model__in=['file', 'subfile', 'patient', 'consent', 'attachment']
|
|
)
|
|
queryset = queryset.filter(content_type__in=file_content_types)
|
|
except Exception:
|
|
# If filtering fails, just return all audit logs
|
|
pass
|
|
|
|
# Apply filters
|
|
patient_id = self.request.GET.get('patient')
|
|
if patient_id:
|
|
queryset = queryset.filter(
|
|
Q(object_id=patient_id) |
|
|
Q(changes__contains={'patient_id': patient_id})
|
|
)
|
|
|
|
user_id = self.request.GET.get('user')
|
|
if user_id:
|
|
queryset = queryset.filter(user_id=user_id)
|
|
|
|
action = self.request.GET.get('action')
|
|
if action:
|
|
queryset = queryset.filter(action=action)
|
|
|
|
date_from = self.request.GET.get('date_from')
|
|
if date_from:
|
|
queryset = queryset.filter(timestamp__date__gte=date_from)
|
|
|
|
date_to = self.request.GET.get('date_to')
|
|
if date_to:
|
|
queryset = queryset.filter(timestamp__date__lte=date_to)
|
|
|
|
return queryset.select_related('user', 'content_type').order_by('-timestamp')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add filter options and stats to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
# Add filter options
|
|
context['patients'] = Patient.objects.filter(
|
|
tenant=self.request.user.tenant
|
|
).order_by('first_name_en')[:100]
|
|
|
|
context['users'] = User.objects.filter(
|
|
tenant=self.request.user.tenant
|
|
).order_by('first_name')
|
|
|
|
context['action_choices'] = [
|
|
('CREATE', 'Create'),
|
|
('UPDATE', 'Update'),
|
|
('DELETE', 'Delete'),
|
|
('VIEW', 'View'),
|
|
]
|
|
|
|
# Add current filters
|
|
context['current_filters'] = {
|
|
'patient': self.request.GET.get('patient', ''),
|
|
'user': self.request.GET.get('user', ''),
|
|
'action': self.request.GET.get('action', ''),
|
|
'date_from': self.request.GET.get('date_from', ''),
|
|
'date_to': self.request.GET.get('date_to', ''),
|
|
}
|
|
|
|
# Add stats
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
queryset = self.get_queryset()
|
|
|
|
# Get Patient ContentType for filtering
|
|
try:
|
|
patient_ct = ContentType.objects.get(model='patient')
|
|
unique_patients_count = queryset.filter(
|
|
content_type=patient_ct
|
|
).values('object_id').distinct().count()
|
|
except ContentType.DoesNotExist:
|
|
unique_patients_count = 0
|
|
|
|
context['stats'] = {
|
|
'total_logs': queryset.count(),
|
|
'unique_users': queryset.values('user').distinct().count(),
|
|
'unique_patients': unique_patients_count,
|
|
}
|
|
|
|
# Export capability
|
|
context['can_export'] = self.request.user.role in [
|
|
User.Role.ADMIN, User.Role.FRONT_DESK
|
|
]
|
|
|
|
return context
|
|
|
|
def render_to_response(self, context, **response_kwargs):
|
|
"""Handle CSV export for compliance."""
|
|
if self.request.GET.get('export') == 'csv':
|
|
return self._export_to_csv(context['audit_logs'])
|
|
|
|
return super().render_to_response(context, **response_kwargs)
|
|
|
|
def _export_to_csv(self, audit_logs):
|
|
"""Export audit logs to CSV."""
|
|
import csv
|
|
from django.http import HttpResponse
|
|
|
|
response = HttpResponse(content_type='text/csv')
|
|
response['Content-Disposition'] = f'attachment; filename="file_history_{timezone.now().date()}.csv"'
|
|
|
|
writer = csv.writer(response)
|
|
writer.writerow([
|
|
'Timestamp', 'User', 'Action', 'Model', 'Object ID',
|
|
'IP Address', 'Changes'
|
|
])
|
|
|
|
for log in audit_logs:
|
|
writer.writerow([
|
|
log.timestamp.strftime('%Y-%m-%d %H:%M:%S'),
|
|
log.user.get_full_name() if log.user else 'System',
|
|
log.action,
|
|
log.content_type.model if log.content_type else 'Unknown',
|
|
log.object_id,
|
|
log.ip_address,
|
|
str(log.changes) if log.changes else '',
|
|
])
|
|
|
|
return response
|
|
|
|
|
|
# ============================================================================
|
|
# PUBLIC CONSENT SIGNING VIEWS (No Authentication Required)
|
|
# ============================================================================
|
|
|
|
class ConsentSignPublicView(TemplateView):
|
|
"""
|
|
Public view for signing consent via email link.
|
|
|
|
No authentication required - accessible via secure token.
|
|
"""
|
|
template_name = 'core/consent_sign_public.html'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add token and consent to context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
token_string = self.kwargs.get('token')
|
|
|
|
# Verify token
|
|
from core.services import ConsentEmailService
|
|
is_valid, message, token = ConsentEmailService.verify_token(token_string)
|
|
|
|
context['is_valid'] = is_valid
|
|
context['message'] = message
|
|
context['token'] = token
|
|
|
|
if token:
|
|
context['consent'] = token.consent
|
|
context['patient'] = token.consent.patient
|
|
|
|
return context
|
|
|
|
|
|
class ConsentSignPublicSubmitView(View):
|
|
"""
|
|
Handle consent signing submission from public form.
|
|
|
|
No authentication required.
|
|
"""
|
|
|
|
def post(self, request, token):
|
|
"""Process consent signing."""
|
|
from core.services import ConsentEmailService
|
|
from django.contrib import messages
|
|
from django.shortcuts import redirect, render
|
|
import base64
|
|
from django.core.files.base import ContentFile
|
|
|
|
# Verify token
|
|
is_valid, message, token_obj = ConsentEmailService.verify_token(token)
|
|
|
|
if not is_valid:
|
|
messages.error(request, message)
|
|
return redirect('core:consent_sign_public', token=token)
|
|
|
|
# Get form data
|
|
signed_by_name = request.POST.get('signed_by_name', '').strip()
|
|
signed_by_relationship = request.POST.get('signed_by_relationship', '').strip()
|
|
signature_method = request.POST.get('signature_method', 'TYPED')
|
|
signature_data = request.POST.get('signature_data', '')
|
|
|
|
# Validate required fields
|
|
if not signed_by_name or not signed_by_relationship:
|
|
messages.error(request, _("Please provide your name and relationship."))
|
|
return redirect('core:consent_sign_public', token=token)
|
|
|
|
# Handle signature image if drawn
|
|
signature_image = None
|
|
if signature_method == 'DRAWN' and signature_data:
|
|
try:
|
|
# Convert base64 to image file
|
|
format_str, imgstr = signature_data.split(';base64,')
|
|
ext = format_str.split('/')[-1]
|
|
signature_image = ContentFile(
|
|
base64.b64decode(imgstr),
|
|
name=f'signature_{token_obj.consent.id}.{ext}'
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error processing signature image: {e}")
|
|
messages.error(request, _("Error processing signature. Please try again."))
|
|
return redirect('core:consent_sign_public', token=token)
|
|
|
|
# Get IP and user agent
|
|
signed_ip = self._get_client_ip(request)
|
|
signed_user_agent = request.META.get('HTTP_USER_AGENT', '')[:255]
|
|
|
|
try:
|
|
# Sign consent
|
|
consent = ConsentEmailService.sign_consent_via_token(
|
|
token=token_obj,
|
|
signed_by_name=signed_by_name,
|
|
signed_by_relationship=signed_by_relationship,
|
|
signature_method=signature_method,
|
|
signature_image=signature_image,
|
|
signed_ip=signed_ip,
|
|
signed_user_agent=signed_user_agent
|
|
)
|
|
|
|
# Success
|
|
return render(request, 'core/consent_sign_success.html', {
|
|
'consent': consent,
|
|
'patient': consent.patient,
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error signing consent via token: {e}")
|
|
messages.error(request, _("An error occurred while signing the consent. Please try again."))
|
|
return redirect('core:consent_sign_public', token=token)
|
|
|
|
def _get_client_ip(self, request):
|
|
"""Get client IP address."""
|
|
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
|
|
if x_forwarded_for:
|
|
ip = x_forwarded_for.split(',')[0]
|
|
else:
|
|
ip = request.META.get('REMOTE_ADDR')
|
|
return ip
|
|
|
|
|
|
class ConsentSendEmailView(LoginRequiredMixin, RolePermissionMixin, View):
|
|
"""
|
|
Staff view to send consent form via email.
|
|
|
|
Requires authentication.
|
|
"""
|
|
allowed_roles = [User.Role.ADMIN, User.Role.DOCTOR, User.Role.NURSE,
|
|
User.Role.FRONT_DESK]
|
|
|
|
def post(self, request, consent_id):
|
|
"""Send consent form via email."""
|
|
from core.services import ConsentEmailService
|
|
from core.models import Consent
|
|
from django.contrib import messages
|
|
from django.shortcuts import redirect
|
|
|
|
try:
|
|
consent = Consent.objects.get(
|
|
id=consent_id,
|
|
tenant=request.user.tenant
|
|
)
|
|
|
|
# Get email from form
|
|
email = request.POST.get('email', '').strip()
|
|
if not email:
|
|
messages.error(request, _("Please provide an email address."))
|
|
return redirect('core:consent_detail', pk=consent_id)
|
|
|
|
# Send consent
|
|
token = ConsentEmailService.send_consent_for_signing(
|
|
consent=consent,
|
|
email=email,
|
|
sent_by=request.user,
|
|
expiry_hours=72 # 3 days
|
|
)
|
|
|
|
messages.success(
|
|
request,
|
|
f"Consent form sent to {email}. Link expires in 72 hours."
|
|
)
|
|
|
|
return redirect('core:consent_detail', pk=consent_id)
|
|
|
|
except Consent.DoesNotExist:
|
|
messages.error(request, _("Consent not found."))
|
|
return redirect('core:consent_list')
|
|
except Exception as e:
|
|
logger.error(f"Error sending consent email: {e}")
|
|
messages.error(request, _("Failed to send email: %(error)s") % {'error': str(e)})
|
|
return redirect('core:consent_detail', pk=consent_id)
|
|
|
|
|
|
# ============================================================================
|
|
# PUBLIC LANDING PAGE VIEW
|
|
# ============================================================================
|
|
|
|
class LandingPageView(TemplateView):
|
|
"""
|
|
Public landing page for Tenhal platform.
|
|
No authentication required.
|
|
Supports bilingual content (Arabic/English).
|
|
Uses ColorAdmin one-page parallax template.
|
|
"""
|
|
template_name = 'core/landing_page_enhanced.html'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add any additional context if needed."""
|
|
context = super().get_context_data(**kwargs)
|
|
# Add any dynamic content here if needed
|
|
return context
|