358 lines
12 KiB
Python
358 lines
12 KiB
Python
"""
|
|
Appointment Confirmation Service.
|
|
|
|
This module handles patient appointment confirmation workflow including
|
|
token generation, validation, and confirmation processing.
|
|
"""
|
|
|
|
import logging
|
|
import secrets
|
|
from datetime import timedelta
|
|
from typing import Optional, Tuple
|
|
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from appointments.models import Appointment, AppointmentConfirmation
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConfirmationService:
|
|
"""Service class for appointment confirmation operations."""
|
|
|
|
# Default token expiration (7 days)
|
|
DEFAULT_EXPIRATION_DAYS = 7
|
|
|
|
# Maximum confirmation reminders to send
|
|
MAX_REMINDERS = 3
|
|
|
|
@staticmethod
|
|
def generate_token() -> str:
|
|
"""
|
|
Generate a secure random token for confirmation.
|
|
|
|
Returns:
|
|
str: 64-character hexadecimal token
|
|
"""
|
|
return secrets.token_hex(32)
|
|
|
|
@staticmethod
|
|
@transaction.atomic
|
|
def create_confirmation(
|
|
appointment: Appointment,
|
|
expiration_days: Optional[int] = None
|
|
) -> AppointmentConfirmation:
|
|
"""
|
|
Create a confirmation token for an appointment.
|
|
|
|
Args:
|
|
appointment: Appointment instance
|
|
expiration_days: Days until token expires (default: 7)
|
|
|
|
Returns:
|
|
AppointmentConfirmation: Created confirmation instance
|
|
"""
|
|
# Check if active confirmation already exists
|
|
existing = AppointmentConfirmation.objects.filter(
|
|
appointment=appointment,
|
|
status='PENDING'
|
|
).first()
|
|
|
|
if existing and existing.is_valid:
|
|
logger.info(f"Using existing valid confirmation for appointment {appointment.id}")
|
|
return existing
|
|
|
|
# Mark any existing confirmations as expired
|
|
AppointmentConfirmation.objects.filter(
|
|
appointment=appointment,
|
|
status='PENDING'
|
|
).update(status='EXPIRED')
|
|
|
|
# Calculate expiration
|
|
days = expiration_days or ConfirmationService.DEFAULT_EXPIRATION_DAYS
|
|
expires_at = timezone.now() + timedelta(days=days)
|
|
|
|
# Create new confirmation
|
|
confirmation = AppointmentConfirmation.objects.create(
|
|
appointment=appointment,
|
|
token=ConfirmationService.generate_token(),
|
|
status='PENDING',
|
|
expires_at=expires_at
|
|
)
|
|
|
|
logger.info(
|
|
f"Created confirmation token for appointment {appointment.id}, "
|
|
f"expires at {expires_at}"
|
|
)
|
|
|
|
return confirmation
|
|
|
|
@staticmethod
|
|
def get_confirmation_by_token(token: str) -> Optional[AppointmentConfirmation]:
|
|
"""
|
|
Retrieve confirmation by token.
|
|
|
|
Args:
|
|
token: Confirmation token
|
|
|
|
Returns:
|
|
AppointmentConfirmation or None
|
|
"""
|
|
try:
|
|
return AppointmentConfirmation.objects.select_related(
|
|
'appointment',
|
|
'appointment__patient',
|
|
'appointment__provider',
|
|
'appointment__clinic'
|
|
).get(token=token)
|
|
except AppointmentConfirmation.DoesNotExist:
|
|
return None
|
|
|
|
@staticmethod
|
|
@transaction.atomic
|
|
def confirm_appointment(
|
|
confirmation: AppointmentConfirmation,
|
|
method: str = 'LINK',
|
|
ip_address: Optional[str] = None,
|
|
user_agent: Optional[str] = None
|
|
) -> Tuple[bool, str]:
|
|
"""
|
|
Confirm an appointment using a confirmation token.
|
|
|
|
Args:
|
|
confirmation: AppointmentConfirmation instance
|
|
method: Confirmation method (LINK, SMS, PHONE, IN_PERSON)
|
|
ip_address: IP address of confirmer
|
|
user_agent: User agent of confirmer
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, message)
|
|
"""
|
|
# Check if already confirmed
|
|
if confirmation.status == 'CONFIRMED':
|
|
return False, "Appointment has already been confirmed"
|
|
|
|
# Check if expired
|
|
if confirmation.is_expired:
|
|
confirmation.status = 'EXPIRED'
|
|
confirmation.save()
|
|
return False, "Confirmation link has expired. Please contact us for assistance."
|
|
|
|
# Check if token is valid
|
|
if not confirmation.is_valid:
|
|
return False, f"Confirmation is not valid. Status: {confirmation.get_status_display()}"
|
|
|
|
# Check appointment status
|
|
appointment = confirmation.appointment
|
|
if appointment.status not in ['BOOKED', 'CONFIRMED']:
|
|
return False, f"Appointment cannot be confirmed. Current status: {appointment.get_status_display()}"
|
|
|
|
# Confirm the appointment
|
|
from appointments.services import AppointmentService
|
|
|
|
try:
|
|
# Update appointment status to CONFIRMED if not already
|
|
if appointment.status == 'BOOKED':
|
|
AppointmentService.confirm_appointment(appointment)
|
|
|
|
# Update confirmation record
|
|
confirmation.status = 'CONFIRMED'
|
|
confirmation.confirmation_method = method
|
|
confirmation.confirmed_at = timezone.now()
|
|
confirmation.confirmed_by_ip = ip_address
|
|
confirmation.confirmed_by_user_agent = user_agent
|
|
confirmation.save()
|
|
|
|
logger.info(
|
|
f"Appointment {appointment.id} confirmed via {method} "
|
|
f"using token {confirmation.token[:8]}..."
|
|
)
|
|
|
|
return True, "Appointment confirmed successfully!"
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error confirming appointment {appointment.id}: {e}")
|
|
return False, f"Error confirming appointment: {str(e)}"
|
|
|
|
@staticmethod
|
|
@transaction.atomic
|
|
def decline_appointment(
|
|
confirmation: AppointmentConfirmation,
|
|
reason: Optional[str] = None
|
|
) -> Tuple[bool, str]:
|
|
"""
|
|
Decline an appointment using a confirmation token.
|
|
|
|
Args:
|
|
confirmation: AppointmentConfirmation instance
|
|
reason: Optional reason for declining
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, message)
|
|
"""
|
|
# Check if already processed
|
|
if confirmation.status in ['CONFIRMED', 'DECLINED']:
|
|
return False, f"Confirmation already processed: {confirmation.get_status_display()}"
|
|
|
|
# Check if expired
|
|
if confirmation.is_expired:
|
|
confirmation.status = 'EXPIRED'
|
|
confirmation.save()
|
|
return False, "Confirmation link has expired"
|
|
|
|
# Mark as declined
|
|
confirmation.status = 'DECLINED'
|
|
confirmation.confirmed_at = timezone.now()
|
|
confirmation.save()
|
|
|
|
# Optionally cancel the appointment
|
|
appointment = confirmation.appointment
|
|
if appointment.status in ['BOOKED', 'CONFIRMED']:
|
|
from appointments.services import AppointmentService
|
|
|
|
cancel_reason = reason or "Patient declined via confirmation link"
|
|
AppointmentService.cancel_appointment(
|
|
appointment,
|
|
reason=cancel_reason
|
|
)
|
|
|
|
logger.info(f"Appointment {appointment.id} declined via confirmation token")
|
|
|
|
return True, "Appointment has been cancelled. Please contact us to reschedule."
|
|
|
|
@staticmethod
|
|
def send_confirmation_request(
|
|
confirmation: AppointmentConfirmation,
|
|
request=None
|
|
) -> bool:
|
|
"""
|
|
Send confirmation request to patient.
|
|
|
|
Args:
|
|
confirmation: AppointmentConfirmation instance
|
|
request: Optional request object for building absolute URL
|
|
|
|
Returns:
|
|
bool: True if sent successfully
|
|
"""
|
|
from datetime import datetime
|
|
from core.tasks import send_multi_channel_notification_task
|
|
|
|
appointment = confirmation.appointment
|
|
|
|
# Get confirmation URL
|
|
confirmation_url = confirmation.get_confirmation_url(request)
|
|
|
|
# Get scheduled datetime
|
|
scheduled_datetime = datetime.combine(
|
|
appointment.scheduled_date,
|
|
appointment.scheduled_time
|
|
)
|
|
|
|
# Prepare message
|
|
message = (
|
|
f"Please confirm your appointment:\n\n"
|
|
f"Date: {scheduled_datetime.strftime('%B %d, %Y at %I:%M %p')}\n"
|
|
f"Provider: {appointment.provider.user.get_full_name()}\n"
|
|
f"Clinic: {appointment.clinic.name_en}\n\n"
|
|
f"Click here to confirm: {confirmation_url}\n\n"
|
|
f"This link expires on {confirmation.expires_at.strftime('%B %d, %Y')}"
|
|
)
|
|
|
|
# Send via multiple channels
|
|
send_multi_channel_notification_task.delay(
|
|
user_id=str(appointment.patient.id) if hasattr(appointment.patient, 'user') else None,
|
|
title="Please Confirm Your Appointment",
|
|
message=message,
|
|
channels=['email', 'sms'],
|
|
email_subject="Appointment Confirmation Required",
|
|
)
|
|
|
|
# Update sent timestamp
|
|
confirmation.sent_at = timezone.now()
|
|
confirmation.save()
|
|
|
|
logger.info(f"Sent confirmation request for appointment {appointment.id}")
|
|
|
|
return True
|
|
|
|
@staticmethod
|
|
def send_confirmation_reminder(
|
|
confirmation: AppointmentConfirmation,
|
|
request=None
|
|
) -> Tuple[bool, str]:
|
|
"""
|
|
Send reminder to confirm appointment.
|
|
|
|
Args:
|
|
confirmation: AppointmentConfirmation instance
|
|
request: Optional request object for building absolute URL
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, message)
|
|
"""
|
|
# Check if max reminders reached
|
|
if confirmation.reminder_count >= ConfirmationService.MAX_REMINDERS:
|
|
return False, "Maximum reminders already sent"
|
|
|
|
# Check if still valid
|
|
if not confirmation.is_valid:
|
|
return False, "Confirmation is no longer valid"
|
|
|
|
# Send reminder
|
|
success = ConfirmationService.send_confirmation_request(confirmation, request)
|
|
|
|
if success:
|
|
# Update reminder count
|
|
confirmation.reminder_count += 1
|
|
confirmation.last_reminder_at = timezone.now()
|
|
confirmation.save()
|
|
|
|
logger.info(
|
|
f"Sent confirmation reminder #{confirmation.reminder_count} "
|
|
f"for appointment {confirmation.appointment.id}"
|
|
)
|
|
|
|
return True, f"Reminder sent (#{confirmation.reminder_count})"
|
|
|
|
return False, "Failed to send reminder"
|
|
|
|
@staticmethod
|
|
def get_pending_confirmations(days_ahead: int = 7):
|
|
"""
|
|
Get appointments needing confirmation.
|
|
|
|
Args:
|
|
days_ahead: Look ahead this many days
|
|
|
|
Returns:
|
|
QuerySet: Pending confirmations
|
|
"""
|
|
cutoff_date = timezone.now() + timedelta(days=days_ahead)
|
|
|
|
return AppointmentConfirmation.objects.filter(
|
|
status='PENDING',
|
|
expires_at__gt=timezone.now(),
|
|
appointment__scheduled_date__lte=cutoff_date.date(),
|
|
appointment__status__in=['BOOKED', 'CONFIRMED']
|
|
).select_related('appointment', 'appointment__patient')
|
|
|
|
@staticmethod
|
|
def expire_old_confirmations() -> int:
|
|
"""
|
|
Mark expired confirmations as EXPIRED.
|
|
|
|
Returns:
|
|
int: Number of confirmations expired
|
|
"""
|
|
count = AppointmentConfirmation.objects.filter(
|
|
status='PENDING',
|
|
expires_at__lt=timezone.now()
|
|
).update(status='EXPIRED')
|
|
|
|
if count > 0:
|
|
logger.info(f"Expired {count} old confirmation tokens")
|
|
|
|
return count
|