630 lines
22 KiB
Python
630 lines
22 KiB
Python
"""
|
|
Appointments Django signals for automation.
|
|
|
|
This module contains signal handlers for appointment-related models to automate
|
|
notifications, reminders, and workflow triggers.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from django.db.models.signals import post_save, pre_save
|
|
from django.dispatch import receiver
|
|
from django.utils import timezone
|
|
|
|
from appointments.models import Appointment, AppointmentReminder
|
|
from appointments.tasks import (
|
|
cancel_appointment_reminders,
|
|
send_appointment_confirmation,
|
|
send_appointment_reminder,
|
|
)
|
|
from core.models import SubFile
|
|
from core.tasks import create_notification_task
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ============================================================================
|
|
# ID Generation Helpers
|
|
# ============================================================================
|
|
|
|
|
|
def generate_appointment_number(tenant):
|
|
"""
|
|
Generate unique Appointment Number.
|
|
Format: APT-YYYY-NNNNNN
|
|
"""
|
|
current_year = timezone.now().year
|
|
|
|
# Get last appointment for this tenant in current year
|
|
last_appointment = Appointment.objects.filter(
|
|
tenant=tenant,
|
|
created_at__year=current_year
|
|
).order_by('-created_at').first()
|
|
|
|
if last_appointment and last_appointment.appointment_number:
|
|
try:
|
|
last_number = int(last_appointment.appointment_number.split('-')[-1])
|
|
new_number = last_number + 1
|
|
except (ValueError, IndexError):
|
|
new_number = 1
|
|
else:
|
|
new_number = 1
|
|
|
|
return f"APT-{current_year}-{new_number:06d}"
|
|
|
|
|
|
# ============================================================================
|
|
# Appointment Signals
|
|
# ============================================================================
|
|
|
|
|
|
@receiver(pre_save, sender=Appointment)
|
|
def appointment_pre_save(sender, instance, **kwargs):
|
|
"""
|
|
Handle pre-save actions for Appointment model.
|
|
|
|
Actions:
|
|
- Auto-generate appointment number if not set
|
|
- Track status changes for post-save processing
|
|
- Validate time slot conflicts
|
|
"""
|
|
# Auto-generate appointment number
|
|
if not instance.appointment_number:
|
|
instance.appointment_number = generate_appointment_number(instance.tenant)
|
|
logger.info(f"Generated Appointment Number: {instance.appointment_number}")
|
|
|
|
if instance.pk:
|
|
try:
|
|
old_instance = Appointment.objects.get(pk=instance.pk)
|
|
instance._old_status = old_instance.status
|
|
instance._status_changed = old_instance.status != instance.status
|
|
except Appointment.DoesNotExist:
|
|
instance._old_status = None
|
|
instance._status_changed = False
|
|
else:
|
|
instance._old_status = None
|
|
instance._status_changed = True
|
|
|
|
|
|
@receiver(post_save, sender=Appointment)
|
|
def appointment_post_save(sender, instance, created, **kwargs):
|
|
"""
|
|
Handle post-save actions for Appointment model.
|
|
|
|
Actions:
|
|
- Auto-create sub-file if needed for new clinic
|
|
- Send confirmation when appointment is booked
|
|
- Schedule reminders when appointment is confirmed
|
|
- Trigger invoice generation when appointment is completed
|
|
- Cancel reminders when appointment is cancelled
|
|
- Notify relevant parties of status changes
|
|
"""
|
|
if created:
|
|
# New appointment created
|
|
logger.info(
|
|
f"Appointment created: {instance.id} for patient {instance.patient.mrn} "
|
|
f"on {instance.scheduled_date}"
|
|
)
|
|
|
|
# Auto-create sub-file if this is patient's first visit to this clinic
|
|
create_subfile_if_needed(instance)
|
|
|
|
# Send confirmation if appointment is booked
|
|
if instance.status == 'BOOKED':
|
|
send_appointment_confirmation.delay(str(instance.id))
|
|
|
|
# Create confirmation token for patient self-service (Phase 5)
|
|
create_patient_confirmation_token(instance)
|
|
|
|
# Notify provider about new appointment
|
|
notify_provider_new_appointment(instance)
|
|
|
|
# Schedule automatic reminders
|
|
schedule_appointment_reminders(instance)
|
|
elif hasattr(instance, '_status_changed') and instance._status_changed:
|
|
# Status changed
|
|
old_status = instance._old_status
|
|
new_status = instance.status
|
|
|
|
logger.info(
|
|
f"Appointment {instance.id} status changed: {old_status} -> {new_status}"
|
|
)
|
|
|
|
# Handle status-specific actions
|
|
if new_status == 'CONFIRMED':
|
|
handle_appointment_confirmed(instance)
|
|
elif new_status == 'RESCHEDULED':
|
|
handle_appointment_rescheduled(instance)
|
|
elif new_status == 'ARRIVED':
|
|
handle_appointment_arrived(instance)
|
|
elif new_status == 'IN_PROGRESS':
|
|
handle_appointment_in_progress(instance)
|
|
elif new_status == 'COMPLETED':
|
|
handle_appointment_completed(instance)
|
|
elif new_status == 'CANCELLED':
|
|
handle_appointment_cancelled(instance)
|
|
elif new_status == 'NO_SHOW':
|
|
handle_appointment_no_show(instance)
|
|
|
|
|
|
def create_subfile_if_needed(appointment: Appointment):
|
|
"""
|
|
Auto-create sub-file when patient books first appointment at a clinic.
|
|
|
|
Args:
|
|
appointment: Appointment instance
|
|
"""
|
|
try:
|
|
patient = appointment.patient
|
|
clinic = appointment.clinic
|
|
|
|
# Check if patient has a main file
|
|
if not hasattr(patient, 'file'):
|
|
logger.warning(
|
|
f"Patient {patient.mrn} has no main file. "
|
|
f"This should have been created automatically."
|
|
)
|
|
return
|
|
|
|
# Check if sub-file exists for this clinic
|
|
subfile_exists = SubFile.objects.filter(
|
|
file=patient.file,
|
|
clinic=clinic
|
|
).exists()
|
|
|
|
if not subfile_exists:
|
|
# Create sub-file
|
|
subfile = SubFile.objects.create(
|
|
file=patient.file,
|
|
clinic=clinic,
|
|
assigned_provider=appointment.provider.user if appointment.provider else None,
|
|
status='ACTIVE'
|
|
)
|
|
logger.info(
|
|
f"Created sub-file {subfile.sub_file_number} for patient {patient.mrn} "
|
|
f"at clinic {clinic.name_en}"
|
|
)
|
|
else:
|
|
logger.debug(
|
|
f"Sub-file already exists for patient {patient.mrn} at clinic {clinic.name_en}"
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error creating sub-file for appointment {appointment.id}: {e}")
|
|
|
|
|
|
def notify_provider_new_appointment(appointment: Appointment):
|
|
"""
|
|
Notify provider when new appointment is booked.
|
|
|
|
Args:
|
|
appointment: Appointment instance
|
|
"""
|
|
if appointment.provider:
|
|
try:
|
|
# Get scheduled datetime
|
|
from datetime import datetime, time
|
|
|
|
# Convert scheduled_time to time object if it's a string
|
|
if isinstance(appointment.scheduled_time, str):
|
|
time_obj = datetime.strptime(appointment.scheduled_time, '%H:%M').time()
|
|
else:
|
|
time_obj = appointment.scheduled_time
|
|
|
|
scheduled_datetime = datetime.combine(
|
|
appointment.scheduled_date,
|
|
time_obj
|
|
)
|
|
|
|
create_notification_task.delay(
|
|
user_id=str(appointment.provider.user.id),
|
|
title="New Appointment Booked",
|
|
message=f"New appointment with {appointment.patient.full_name_en} "
|
|
f"on {scheduled_datetime.strftime('%B %d, %Y at %I:%M %p')} "
|
|
f"for {appointment.service_type}",
|
|
notification_type='INFO',
|
|
related_object_type='appointment',
|
|
related_object_id=str(appointment.id),
|
|
)
|
|
logger.info(f"Notified provider {appointment.provider.user.username} of new appointment")
|
|
except Exception as e:
|
|
logger.error(f"Error notifying provider for appointment {appointment.id}: {e}")
|
|
|
|
|
|
def create_patient_confirmation_token(appointment: Appointment):
|
|
"""
|
|
Create confirmation token for patient self-service confirmation (Phase 5).
|
|
|
|
Args:
|
|
appointment: Appointment instance
|
|
"""
|
|
try:
|
|
from appointments.confirmation_service import ConfirmationService
|
|
from django.contrib.sites.shortcuts import get_current_site
|
|
|
|
# Create confirmation token
|
|
confirmation = ConfirmationService.create_confirmation(
|
|
appointment=appointment,
|
|
expiration_days=7
|
|
)
|
|
|
|
# Send confirmation request via email/SMS
|
|
# Note: request object not available in signal, so we pass None
|
|
# The service will handle sending without request context
|
|
ConfirmationService.send_confirmation_request(
|
|
confirmation=confirmation,
|
|
request=None
|
|
)
|
|
|
|
logger.info(
|
|
f"Created confirmation token for appointment {appointment.id}. "
|
|
f"Token expires: {confirmation.expires_at}"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error creating confirmation token for appointment {appointment.id}: {e}")
|
|
|
|
|
|
def schedule_appointment_reminders(appointment: Appointment):
|
|
"""
|
|
Schedule automatic reminders for appointment.
|
|
Creates reminder records and schedules Celery tasks.
|
|
|
|
Args:
|
|
appointment: Appointment instance
|
|
"""
|
|
try:
|
|
from datetime import datetime, time
|
|
|
|
# Convert scheduled_time to time object if it's a string
|
|
if isinstance(appointment.scheduled_time, str):
|
|
time_obj = datetime.strptime(appointment.scheduled_time, '%H:%M').time()
|
|
else:
|
|
time_obj = appointment.scheduled_time
|
|
|
|
# Get scheduled datetime
|
|
scheduled_datetime = timezone.make_aware(
|
|
datetime.combine(appointment.scheduled_date, time_obj)
|
|
)
|
|
|
|
# Determine reminder channel based on patient preferences
|
|
# Default to SMS if no preferences set
|
|
reminder_channel = 'SMS'
|
|
if hasattr(appointment.patient, 'notification_preferences'):
|
|
prefs = appointment.patient.notification_preferences
|
|
if prefs.whatsapp_enabled:
|
|
reminder_channel = 'WHATSAPP'
|
|
elif prefs.sms_enabled:
|
|
reminder_channel = 'SMS'
|
|
elif prefs.email_enabled:
|
|
reminder_channel = 'EMAIL'
|
|
|
|
# Schedule 24-hour reminder
|
|
reminder_24h_time = scheduled_datetime - timedelta(hours=24)
|
|
if reminder_24h_time > timezone.now():
|
|
reminder_24h = AppointmentReminder.objects.create(
|
|
appointment=appointment,
|
|
reminder_type=reminder_channel,
|
|
scheduled_for=reminder_24h_time,
|
|
status='SCHEDULED'
|
|
)
|
|
|
|
# Schedule Celery task
|
|
send_appointment_reminder.apply_async(
|
|
args=[str(appointment.id), 24],
|
|
eta=reminder_24h_time
|
|
)
|
|
logger.info(f"Scheduled 24-hour reminder for appointment {appointment.id}")
|
|
|
|
# Schedule 2-hour reminder
|
|
reminder_2h_time = scheduled_datetime - timedelta(hours=2)
|
|
if reminder_2h_time > timezone.now():
|
|
reminder_2h = AppointmentReminder.objects.create(
|
|
appointment=appointment,
|
|
reminder_type=reminder_channel,
|
|
scheduled_for=reminder_2h_time,
|
|
status='SCHEDULED'
|
|
)
|
|
|
|
# Schedule Celery task
|
|
send_appointment_reminder.apply_async(
|
|
args=[str(appointment.id), 2],
|
|
eta=reminder_2h_time
|
|
)
|
|
logger.info(f"Scheduled 2-hour reminder for appointment {appointment.id}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error scheduling reminders for appointment {appointment.id}: {e}")
|
|
|
|
|
|
def handle_appointment_confirmed(appointment: Appointment):
|
|
"""
|
|
Handle appointment confirmation.
|
|
|
|
Actions:
|
|
- Send confirmation to patient
|
|
- Reminders already scheduled on creation, no need to reschedule
|
|
"""
|
|
# Send confirmation
|
|
send_appointment_confirmation.delay(str(appointment.id))
|
|
|
|
logger.info(f"Appointment {appointment.id} confirmed")
|
|
|
|
|
|
def handle_appointment_rescheduled(appointment: Appointment):
|
|
"""
|
|
Handle appointment rescheduling.
|
|
|
|
Actions:
|
|
- Cancel old reminders
|
|
- Schedule new reminders for new time
|
|
- Notify patient of new time
|
|
- Notify provider of change
|
|
"""
|
|
from datetime import datetime, time
|
|
from core.tasks import send_multi_channel_notification_task
|
|
|
|
# Cancel old reminders
|
|
cancel_appointment_reminders.delay(str(appointment.id))
|
|
|
|
# Schedule new reminders
|
|
schedule_appointment_reminders(appointment)
|
|
|
|
# Convert scheduled_time to time object if it's a string
|
|
if isinstance(appointment.scheduled_time, str):
|
|
time_obj = datetime.strptime(appointment.scheduled_time, '%H:%M').time()
|
|
else:
|
|
time_obj = appointment.scheduled_time
|
|
|
|
# Get new scheduled datetime
|
|
new_datetime = datetime.combine(
|
|
appointment.scheduled_date,
|
|
time_obj
|
|
)
|
|
|
|
reschedule_reason = appointment.reschedule_reason or "Rescheduled by request"
|
|
|
|
# Notify patient of new time
|
|
patient_message = (
|
|
f"Your appointment has been rescheduled to {new_datetime.strftime('%B %d, %Y at %I:%M %p')} "
|
|
f"with {appointment.provider.get_full_name()}.\n\n"
|
|
f"Reason: {reschedule_reason}\n\n"
|
|
f"Please confirm your availability for the new time."
|
|
)
|
|
|
|
send_multi_channel_notification_task.delay(
|
|
user_id=str(appointment.patient.id) if hasattr(appointment.patient, 'user') else None,
|
|
title="Appointment Rescheduled",
|
|
message=patient_message,
|
|
channels=['email', 'sms', 'in_app'],
|
|
email_subject="Appointment Rescheduled",
|
|
)
|
|
|
|
# Notify provider
|
|
if appointment.provider:
|
|
create_notification_task.delay(
|
|
user_id=str(appointment.provider.user.id),
|
|
title="Appointment Rescheduled",
|
|
message=f"Appointment with {appointment.patient.full_name_en} "
|
|
f"has been rescheduled to {new_datetime.strftime('%B %d, %Y at %I:%M %p')}. "
|
|
f"Reason: {reschedule_reason}",
|
|
notification_type='INFO',
|
|
related_object_type='appointment',
|
|
related_object_id=str(appointment.id),
|
|
)
|
|
|
|
logger.info(f"Appointment {appointment.id} rescheduled - notifications sent")
|
|
|
|
|
|
def handle_appointment_arrived(appointment: Appointment):
|
|
"""
|
|
Handle patient arrival.
|
|
|
|
Actions:
|
|
- Notify provider that patient has arrived
|
|
- Log clearance status
|
|
|
|
Note: Financial and consent checks are performed in AppointmentService.mark_arrival()
|
|
before the status is changed to ARRIVED, so by the time this handler runs,
|
|
all prerequisites have been met.
|
|
"""
|
|
if appointment.provider:
|
|
create_notification_task.delay(
|
|
user_id=str(appointment.provider.user.id),
|
|
title="Patient Arrived",
|
|
message=f"Patient {appointment.patient.full_name_en} has arrived for their appointment. "
|
|
f"Financial clearance: {'✓' if appointment.finance_cleared else '✗'}, "
|
|
f"Consent verified: {'✓' if appointment.consent_verified else '✗'}",
|
|
notification_type='INFO',
|
|
related_object_type='appointment',
|
|
related_object_id=str(appointment.id),
|
|
)
|
|
|
|
logger.info(
|
|
f"Patient arrived for appointment {appointment.id}. "
|
|
f"Finance cleared: {appointment.finance_cleared}, "
|
|
f"Consent verified: {appointment.consent_verified}"
|
|
)
|
|
|
|
|
|
def handle_appointment_in_progress(appointment: Appointment):
|
|
"""
|
|
Handle appointment start.
|
|
|
|
Actions:
|
|
- Log appointment start time
|
|
"""
|
|
logger.info(f"Appointment {appointment.id} started")
|
|
|
|
|
|
def handle_appointment_completed(appointment: Appointment):
|
|
"""
|
|
Handle appointment completion.
|
|
|
|
Actions:
|
|
- Trigger invoice generation
|
|
- Send completion notification to patient
|
|
- Update appointment statistics
|
|
- Increment package usage if appointment is part of a package
|
|
"""
|
|
# Trigger invoice generation (if not already exists)
|
|
from finance.tasks import generate_invoice_from_appointment
|
|
|
|
generate_invoice_from_appointment.delay(str(appointment.id))
|
|
|
|
# Increment package usage if this appointment is part of a package
|
|
if appointment.package_purchase:
|
|
try:
|
|
from .package_integration_service import PackageIntegrationService
|
|
PackageIntegrationService.increment_package_usage(appointment)
|
|
logger.info(
|
|
f"Incremented package usage for appointment {appointment.id}. "
|
|
f"Package: {appointment.package_purchase.package.name_en}, "
|
|
f"Sessions used: {appointment.package_purchase.sessions_used}/{appointment.package_purchase.total_sessions}"
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error incrementing package usage for appointment {appointment.id}: {e}")
|
|
|
|
# Notify patient
|
|
if appointment.patient.email:
|
|
from core.tasks import send_email_task
|
|
|
|
send_email_task.delay(
|
|
subject="Appointment Completed",
|
|
message=f"Dear {appointment.patient.full_name_en},\n\n"
|
|
f"Your appointment with {appointment.provider.get_full_name()} "
|
|
f"has been completed.\n\n"
|
|
f"Thank you for visiting {appointment.clinic.name_en}.\n\n"
|
|
f"Best regards,\nTenhal Healthcare Team",
|
|
recipient_list=[appointment.patient.email],
|
|
)
|
|
|
|
logger.info(f"Appointment {appointment.id} completed")
|
|
|
|
|
|
def handle_appointment_cancelled(appointment: Appointment):
|
|
"""
|
|
Handle appointment cancellation.
|
|
|
|
Actions:
|
|
- Cancel scheduled reminders
|
|
- Notify patient via multiple channels
|
|
- Notify provider
|
|
- Free up the time slot
|
|
"""
|
|
from datetime import datetime, time
|
|
from core.tasks import send_email_task, send_multi_channel_notification_task
|
|
|
|
# Cancel reminders
|
|
cancel_appointment_reminders.delay(str(appointment.id))
|
|
|
|
# Convert scheduled_time to time object if it's a string
|
|
if isinstance(appointment.scheduled_time, str):
|
|
time_obj = datetime.strptime(appointment.scheduled_time, '%H:%M').time()
|
|
else:
|
|
time_obj = appointment.scheduled_time
|
|
|
|
# Get scheduled datetime
|
|
scheduled_datetime = datetime.combine(
|
|
appointment.scheduled_date,
|
|
time_obj
|
|
)
|
|
|
|
cancellation_reason = appointment.cancel_reason or "No reason provided"
|
|
|
|
# Notify patient via multiple channels
|
|
patient_message = (
|
|
f"Your appointment on {scheduled_datetime.strftime('%B %d, %Y at %I:%M %p')} "
|
|
f"with {appointment.provider.get_full_name()} has been cancelled.\n\n"
|
|
f"Reason: {cancellation_reason}\n\n"
|
|
f"Please contact us to reschedule if needed."
|
|
)
|
|
|
|
send_multi_channel_notification_task.delay(
|
|
user_id=str(appointment.patient.id) if hasattr(appointment.patient, 'user') else None,
|
|
title="Appointment Cancelled",
|
|
message=patient_message,
|
|
channels=['email', 'sms', 'in_app'],
|
|
email_subject="Appointment Cancelled",
|
|
)
|
|
|
|
# Notify provider
|
|
if appointment.provider:
|
|
create_notification_task.delay(
|
|
user_id=str(appointment.provider.user.id),
|
|
title="Appointment Cancelled",
|
|
message=f"Appointment with {appointment.patient.full_name_en} "
|
|
f"on {scheduled_datetime.strftime('%B %d, %Y at %I:%M %p')} has been cancelled. "
|
|
f"Reason: {cancellation_reason}",
|
|
notification_type='WARNING',
|
|
related_object_type='appointment',
|
|
related_object_id=str(appointment.id),
|
|
)
|
|
|
|
logger.info(f"Appointment {appointment.id} cancelled - notifications sent")
|
|
|
|
|
|
def handle_appointment_no_show(appointment: Appointment):
|
|
"""
|
|
Handle patient no-show.
|
|
|
|
Actions:
|
|
- Notify provider
|
|
- Update patient statistics
|
|
- Consider automatic rescheduling policy
|
|
"""
|
|
from datetime import datetime, time
|
|
from core.tasks import send_email_task
|
|
|
|
# Convert scheduled_time to time object if it's a string
|
|
if isinstance(appointment.scheduled_time, str):
|
|
time_obj = datetime.strptime(appointment.scheduled_time, '%H:%M').time()
|
|
else:
|
|
time_obj = appointment.scheduled_time
|
|
|
|
# Get scheduled datetime
|
|
scheduled_datetime = datetime.combine(
|
|
appointment.scheduled_date,
|
|
time_obj
|
|
)
|
|
|
|
# Notify provider
|
|
if appointment.provider:
|
|
create_notification_task.delay(
|
|
user_id=str(appointment.provider.user.id),
|
|
title="Patient No-Show",
|
|
message=f"Patient {appointment.patient.full_name_en} did not show up for "
|
|
f"appointment on {scheduled_datetime.strftime('%B %d, %Y at %I:%M %p')}",
|
|
notification_type='WARNING',
|
|
related_object_type='appointment',
|
|
related_object_id=str(appointment.id),
|
|
)
|
|
|
|
# Send notification to patient
|
|
if appointment.patient.email:
|
|
send_email_task.delay(
|
|
subject="Missed Appointment",
|
|
message=f"Dear {appointment.patient.full_name_en},\n\n"
|
|
f"We noticed you missed your appointment on "
|
|
f"{scheduled_datetime.strftime('%B %d, %Y at %I:%M %p')}.\n\n"
|
|
f"Please contact us to reschedule.\n\n"
|
|
f"Best regards,\nTenhal Healthcare Team",
|
|
recipient_list=[appointment.patient.email],
|
|
)
|
|
|
|
logger.warning(f"Patient no-show for appointment {appointment.id}")
|
|
|
|
|
|
# ============================================================================
|
|
# Signal Connection Helper
|
|
# ============================================================================
|
|
|
|
|
|
def connect_signals():
|
|
"""
|
|
Explicitly connect all signals.
|
|
|
|
This function can be called from apps.py to ensure signals are connected.
|
|
"""
|
|
logger.info("Appointments signals connected")
|