513 lines
16 KiB
Python
513 lines
16 KiB
Python
"""
|
|
Utility functions for appointments app.
|
|
Provides helper functions for common operations.
|
|
"""
|
|
|
|
from django.core.exceptions import ValidationError
|
|
from django.utils import timezone
|
|
from django.conf import settings
|
|
from datetime import datetime, timedelta
|
|
import uuid
|
|
|
|
|
|
def get_tenant_from_request(request):
|
|
"""
|
|
Extract tenant from request object.
|
|
|
|
Args:
|
|
request: Django request object
|
|
|
|
Returns:
|
|
Tenant: Tenant object or None
|
|
|
|
Raises:
|
|
AttributeError: If request doesn't have user or tenant
|
|
"""
|
|
if not hasattr(request, 'user'):
|
|
return None
|
|
|
|
if not hasattr(request.user, 'tenant'):
|
|
return None
|
|
|
|
return request.user.tenant
|
|
|
|
|
|
def check_appointment_conflicts(provider, start_datetime, end_datetime, exclude_appointment_id=None):
|
|
"""
|
|
Check for appointment scheduling conflicts for a provider.
|
|
|
|
Args:
|
|
provider: User object (healthcare provider)
|
|
start_datetime: Proposed appointment start datetime
|
|
end_datetime: Proposed appointment end datetime
|
|
exclude_appointment_id: Optional appointment ID to exclude from conflict check
|
|
|
|
Returns:
|
|
dict: {
|
|
'has_conflict': bool,
|
|
'conflicting_appointments': QuerySet,
|
|
'message': str
|
|
}
|
|
"""
|
|
from .models import AppointmentRequest
|
|
|
|
# Query for overlapping appointments
|
|
conflicts = AppointmentRequest.objects.filter(
|
|
provider=provider,
|
|
status__in=['SCHEDULED', 'CONFIRMED', 'CHECKED_IN', 'IN_PROGRESS']
|
|
).filter(
|
|
scheduled_datetime__lt=end_datetime,
|
|
scheduled_end_datetime__gt=start_datetime
|
|
)
|
|
|
|
# Exclude specific appointment if provided (for rescheduling)
|
|
if exclude_appointment_id:
|
|
conflicts = conflicts.exclude(pk=exclude_appointment_id)
|
|
|
|
has_conflict = conflicts.exists()
|
|
|
|
result = {
|
|
'has_conflict': has_conflict,
|
|
'conflicting_appointments': conflicts,
|
|
'message': ''
|
|
}
|
|
|
|
if has_conflict:
|
|
conflict_count = conflicts.count()
|
|
result['message'] = (
|
|
f"Provider has {conflict_count} conflicting appointment(s) "
|
|
f"between {start_datetime.strftime('%Y-%m-%d %H:%M')} and "
|
|
f"{end_datetime.strftime('%Y-%m-%d %H:%M')}"
|
|
)
|
|
else:
|
|
result['message'] = "No scheduling conflicts found"
|
|
|
|
return result
|
|
|
|
|
|
def send_appointment_notification(appointment, notification_type, recipient=None):
|
|
"""
|
|
Send appointment notification to patient or provider.
|
|
|
|
Args:
|
|
appointment: AppointmentRequest object
|
|
notification_type: Type of notification ('confirmation', 'reminder', 'cancellation', 'reschedule')
|
|
recipient: Optional recipient ('patient' or 'provider'), defaults to 'patient'
|
|
|
|
Returns:
|
|
dict: {
|
|
'success': bool,
|
|
'method': str (email, sms, etc.),
|
|
'message': str
|
|
}
|
|
"""
|
|
if recipient is None:
|
|
recipient = 'patient'
|
|
|
|
result = {
|
|
'success': False,
|
|
'method': None,
|
|
'message': ''
|
|
}
|
|
|
|
# Determine recipient contact info
|
|
if recipient == 'patient':
|
|
contact_email = appointment.patient.email if hasattr(appointment.patient, 'email') else None
|
|
contact_phone = appointment.patient.phone_number if hasattr(appointment.patient, 'phone_number') else None
|
|
recipient_name = appointment.patient.get_full_name()
|
|
else: # provider
|
|
contact_email = appointment.provider.email
|
|
contact_phone = getattr(appointment.provider, 'phone_number', None)
|
|
recipient_name = appointment.provider.get_full_name()
|
|
|
|
# Prepare notification content based on type
|
|
subject, message = _prepare_notification_content(appointment, notification_type, recipient_name)
|
|
|
|
# Try to send via email first
|
|
if contact_email:
|
|
try:
|
|
from django.core.mail import send_mail
|
|
|
|
send_mail(
|
|
subject=subject,
|
|
message=message,
|
|
from_email=settings.DEFAULT_FROM_EMAIL,
|
|
recipient_list=[contact_email],
|
|
fail_silently=False
|
|
)
|
|
|
|
result['success'] = True
|
|
result['method'] = 'email'
|
|
result['message'] = f"Email notification sent to {contact_email}"
|
|
return result
|
|
|
|
except Exception as e:
|
|
result['message'] = f"Email sending failed: {str(e)}"
|
|
|
|
# Fallback to SMS if email fails and phone is available
|
|
if contact_phone:
|
|
# SMS implementation would go here
|
|
# For now, just log that SMS would be sent
|
|
result['success'] = True
|
|
result['method'] = 'sms'
|
|
result['message'] = f"SMS notification would be sent to {contact_phone}"
|
|
return result
|
|
|
|
result['message'] = "No valid contact method available"
|
|
return result
|
|
|
|
|
|
def _prepare_notification_content(appointment, notification_type, recipient_name):
|
|
"""
|
|
Prepare notification subject and message content.
|
|
|
|
Args:
|
|
appointment: AppointmentRequest object
|
|
notification_type: Type of notification
|
|
recipient_name: Name of recipient
|
|
|
|
Returns:
|
|
tuple: (subject, message)
|
|
"""
|
|
tenant_name = appointment.tenant.name if appointment.tenant else "Hospital"
|
|
|
|
if notification_type == 'confirmation':
|
|
subject = f"Appointment Confirmation - {tenant_name}"
|
|
message = f"""
|
|
Dear {recipient_name},
|
|
|
|
Your appointment has been confirmed:
|
|
|
|
Date: {appointment.scheduled_datetime.strftime('%B %d, %Y')}
|
|
Time: {appointment.scheduled_datetime.strftime('%I:%M %p')}
|
|
Provider: {appointment.provider.get_full_name()}
|
|
Type: {appointment.get_appointment_type_display()}
|
|
Location: {appointment.location or 'To be determined'}
|
|
|
|
Please arrive 15 minutes early for check-in.
|
|
|
|
If you need to reschedule or cancel, please contact us at least 24 hours in advance.
|
|
|
|
Thank you,
|
|
{tenant_name}
|
|
"""
|
|
|
|
elif notification_type == 'reminder':
|
|
subject = f"Appointment Reminder - {tenant_name}"
|
|
message = f"""
|
|
Dear {recipient_name},
|
|
|
|
This is a reminder of your upcoming appointment:
|
|
|
|
Date: {appointment.scheduled_datetime.strftime('%B %d, %Y')}
|
|
Time: {appointment.scheduled_datetime.strftime('%I:%M %p')}
|
|
Provider: {appointment.provider.get_full_name()}
|
|
Location: {appointment.location or 'To be determined'}
|
|
|
|
Please arrive 15 minutes early for check-in.
|
|
|
|
Thank you,
|
|
{tenant_name}
|
|
"""
|
|
|
|
elif notification_type == 'cancellation':
|
|
subject = f"Appointment Cancelled - {tenant_name}"
|
|
message = f"""
|
|
Dear {recipient_name},
|
|
|
|
Your appointment has been cancelled:
|
|
|
|
Original Date: {appointment.scheduled_datetime.strftime('%B %d, %Y')}
|
|
Original Time: {appointment.scheduled_datetime.strftime('%I:%M %p')}
|
|
Reason: {appointment.cancellation_reason or 'Not specified'}
|
|
|
|
If you would like to reschedule, please contact us.
|
|
|
|
Thank you,
|
|
{tenant_name}
|
|
"""
|
|
|
|
elif notification_type == 'reschedule':
|
|
subject = f"Appointment Rescheduled - {tenant_name}"
|
|
message = f"""
|
|
Dear {recipient_name},
|
|
|
|
Your appointment has been rescheduled:
|
|
|
|
New Date: {appointment.scheduled_datetime.strftime('%B %d, %Y')}
|
|
New Time: {appointment.scheduled_datetime.strftime('%I:%M %p')}
|
|
Provider: {appointment.provider.get_full_name()}
|
|
Location: {appointment.location or 'To be determined'}
|
|
|
|
Please arrive 15 minutes early for check-in.
|
|
|
|
Thank you,
|
|
{tenant_name}
|
|
"""
|
|
|
|
else:
|
|
subject = f"Appointment Update - {tenant_name}"
|
|
message = f"""
|
|
Dear {recipient_name},
|
|
|
|
There has been an update to your appointment. Please contact us for details.
|
|
|
|
Thank you,
|
|
{tenant_name}
|
|
"""
|
|
|
|
return subject, message.strip()
|
|
|
|
|
|
def generate_meeting_url(appointment, platform=None):
|
|
"""
|
|
Generate telemedicine meeting URL for an appointment.
|
|
|
|
Args:
|
|
appointment: AppointmentRequest object
|
|
platform: Optional platform override ('ZOOM', 'TEAMS', 'WEBEX', 'DOXY', 'CUSTOM')
|
|
|
|
Returns:
|
|
dict: {
|
|
'success': bool,
|
|
'meeting_url': str,
|
|
'meeting_id': str,
|
|
'meeting_password': str,
|
|
'platform': str,
|
|
'message': str
|
|
}
|
|
"""
|
|
if platform is None:
|
|
platform = appointment.telemedicine_platform or 'CUSTOM'
|
|
|
|
result = {
|
|
'success': False,
|
|
'meeting_url': None,
|
|
'meeting_id': None,
|
|
'meeting_password': None,
|
|
'platform': platform,
|
|
'message': ''
|
|
}
|
|
|
|
# Generate unique meeting ID
|
|
meeting_id = f"{appointment.request_id.hex[:8]}-{uuid.uuid4().hex[:8]}"
|
|
|
|
# Generate meeting password
|
|
meeting_password = uuid.uuid4().hex[:12].upper()
|
|
|
|
# Generate platform-specific URLs
|
|
if platform == 'ZOOM':
|
|
# In production, this would integrate with Zoom API
|
|
meeting_url = f"https://zoom.us/j/{meeting_id}"
|
|
result['message'] = "Zoom meeting URL generated (mock)"
|
|
|
|
elif platform == 'TEAMS':
|
|
# In production, this would integrate with Microsoft Teams API
|
|
meeting_url = f"https://teams.microsoft.com/l/meetup-join/{meeting_id}"
|
|
result['message'] = "Microsoft Teams meeting URL generated (mock)"
|
|
|
|
elif platform == 'WEBEX':
|
|
# In production, this would integrate with Webex API
|
|
meeting_url = f"https://webex.com/meet/{meeting_id}"
|
|
result['message'] = "Webex meeting URL generated (mock)"
|
|
|
|
elif platform == 'DOXY':
|
|
# Doxy.me uses simple room URLs
|
|
room_name = f"{appointment.provider.last_name.lower()}-{meeting_id[:8]}"
|
|
meeting_url = f"https://doxy.me/{room_name}"
|
|
meeting_password = None # Doxy.me doesn't use passwords by default
|
|
result['message'] = "Doxy.me meeting URL generated"
|
|
|
|
else: # CUSTOM or OTHER
|
|
# Generic meeting URL
|
|
meeting_url = f"https://telehealth.{appointment.tenant.domain if hasattr(appointment.tenant, 'domain') else 'hospital.com'}/meet/{meeting_id}"
|
|
result['message'] = "Custom meeting URL generated"
|
|
|
|
result.update({
|
|
'success': True,
|
|
'meeting_url': meeting_url,
|
|
'meeting_id': meeting_id,
|
|
'meeting_password': meeting_password
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
def calculate_appointment_duration(appointment_type, specialty=None):
|
|
"""
|
|
Calculate recommended appointment duration based on type and specialty.
|
|
|
|
Args:
|
|
appointment_type: Type of appointment
|
|
specialty: Optional medical specialty
|
|
|
|
Returns:
|
|
int: Recommended duration in minutes
|
|
"""
|
|
# Default durations by appointment type
|
|
duration_map = {
|
|
'CONSULTATION': 30,
|
|
'FOLLOW_UP': 15,
|
|
'PROCEDURE': 60,
|
|
'SURGERY': 120,
|
|
'DIAGNOSTIC': 45,
|
|
'THERAPY': 60,
|
|
'VACCINATION': 15,
|
|
'SCREENING': 30,
|
|
'EMERGENCY': 30,
|
|
'TELEMEDICINE': 20,
|
|
'OTHER': 30
|
|
}
|
|
|
|
# Specialty-specific adjustments
|
|
specialty_adjustments = {
|
|
'SURGERY': 30,
|
|
'PSYCHIATRY': 15,
|
|
'CARDIOLOGY': 15,
|
|
'ONCOLOGY': 15,
|
|
}
|
|
|
|
base_duration = duration_map.get(appointment_type, 30)
|
|
|
|
if specialty and specialty in specialty_adjustments:
|
|
base_duration += specialty_adjustments[specialty]
|
|
|
|
return base_duration
|
|
|
|
|
|
def validate_appointment_time(scheduled_datetime, tenant):
|
|
"""
|
|
Validate if appointment time is within operating hours.
|
|
|
|
Args:
|
|
scheduled_datetime: Proposed appointment datetime
|
|
tenant: Tenant object
|
|
|
|
Returns:
|
|
dict: {
|
|
'valid': bool,
|
|
'message': str
|
|
}
|
|
"""
|
|
# Default operating hours (8 AM to 6 PM)
|
|
operating_start = 8
|
|
operating_end = 18
|
|
|
|
# Check if tenant has custom operating hours
|
|
if hasattr(tenant, 'operating_hours'):
|
|
day_of_week = scheduled_datetime.weekday()
|
|
if str(day_of_week) in tenant.operating_hours:
|
|
hours = tenant.operating_hours[str(day_of_week)]
|
|
if hours.get('enabled'):
|
|
operating_start = int(hours.get('start_time', '08:00').split(':')[0])
|
|
operating_end = int(hours.get('end_time', '18:00').split(':')[0])
|
|
|
|
appointment_hour = scheduled_datetime.hour
|
|
|
|
if appointment_hour < operating_start or appointment_hour >= operating_end:
|
|
return {
|
|
'valid': False,
|
|
'message': f"Appointment time must be between {operating_start}:00 and {operating_end}:00"
|
|
}
|
|
|
|
# Check if appointment is in the past
|
|
if scheduled_datetime < timezone.now():
|
|
return {
|
|
'valid': False,
|
|
'message': "Appointment cannot be scheduled in the past"
|
|
}
|
|
|
|
# Check if appointment is on a weekend (optional)
|
|
if scheduled_datetime.weekday() in [5, 6]: # Saturday, Sunday
|
|
return {
|
|
'valid': False,
|
|
'message': "Appointments cannot be scheduled on weekends"
|
|
}
|
|
|
|
return {
|
|
'valid': True,
|
|
'message': "Appointment time is valid"
|
|
}
|
|
|
|
|
|
def get_available_time_slots(provider, date, duration_minutes=30):
|
|
"""
|
|
Get available time slots for a provider on a specific date.
|
|
|
|
Args:
|
|
provider: User object (healthcare provider)
|
|
date: Date to check availability
|
|
duration_minutes: Duration of appointment in minutes
|
|
|
|
Returns:
|
|
list: List of available time slots as datetime objects
|
|
"""
|
|
from .models import AppointmentRequest, SlotAvailability
|
|
|
|
available_slots = []
|
|
|
|
# Get provider's availability slots for the date
|
|
slots = SlotAvailability.objects.filter(
|
|
provider=provider,
|
|
date=date,
|
|
is_active=True,
|
|
is_blocked=False
|
|
).order_by('start_time')
|
|
|
|
for slot in slots:
|
|
if slot.available_capacity > 0:
|
|
# Create datetime from date and time
|
|
slot_datetime = datetime.combine(date, slot.start_time)
|
|
slot_datetime = timezone.make_aware(slot_datetime)
|
|
|
|
# Check if there's a conflict
|
|
end_datetime = slot_datetime + timedelta(minutes=duration_minutes)
|
|
conflict_check = check_appointment_conflicts(
|
|
provider,
|
|
slot_datetime,
|
|
end_datetime
|
|
)
|
|
|
|
if not conflict_check['has_conflict']:
|
|
available_slots.append(slot_datetime)
|
|
|
|
return available_slots
|
|
|
|
|
|
def format_appointment_summary(appointment):
|
|
"""
|
|
Format appointment details as a summary string.
|
|
|
|
Args:
|
|
appointment: AppointmentRequest object
|
|
|
|
Returns:
|
|
str: Formatted appointment summary
|
|
"""
|
|
summary = f"""
|
|
Appointment Summary
|
|
==================
|
|
Patient: {appointment.patient.get_full_name()}
|
|
MRN: {appointment.patient.mrn if hasattr(appointment.patient, 'mrn') else 'N/A'}
|
|
Provider: {appointment.provider.get_full_name()}
|
|
Type: {appointment.get_appointment_type_display()}
|
|
Specialty: {appointment.get_specialty_display()}
|
|
Date: {appointment.scheduled_datetime.strftime('%B %d, %Y') if appointment.scheduled_datetime else 'Not scheduled'}
|
|
Time: {appointment.scheduled_datetime.strftime('%I:%M %p') if appointment.scheduled_datetime else 'Not scheduled'}
|
|
Duration: {appointment.duration_minutes} minutes
|
|
Location: {appointment.location or 'Not specified'}
|
|
Status: {appointment.get_status_display()}
|
|
Priority: {appointment.get_priority_display()}
|
|
"""
|
|
|
|
if appointment.is_telemedicine:
|
|
summary += f"\nTelemedicine: Yes"
|
|
summary += f"\nPlatform: {appointment.get_telemedicine_platform_display() if appointment.telemedicine_platform else 'Not specified'}"
|
|
if appointment.meeting_url:
|
|
summary += f"\nMeeting URL: {appointment.meeting_url}"
|
|
|
|
if appointment.chief_complaint:
|
|
summary += f"\nChief Complaint: {appointment.chief_complaint}"
|
|
|
|
return summary.strip()
|