409 lines
14 KiB
Python
409 lines
14 KiB
Python
"""
|
|
Appointment availability service for the Tenhal Multidisciplinary Healthcare Platform.
|
|
|
|
This module provides functionality to check provider availability and generate
|
|
available time slots based on provider schedules and existing appointments.
|
|
"""
|
|
|
|
from datetime import datetime, date, time, timedelta
|
|
from typing import List, Dict, Tuple
|
|
from django.db.models import Q
|
|
from django.utils import timezone
|
|
|
|
from .models import Schedule, Appointment, Provider
|
|
from core.models import User
|
|
|
|
|
|
class AvailabilityService:
|
|
"""Service for checking provider availability and generating time slots."""
|
|
|
|
@staticmethod
|
|
def get_available_slots(
|
|
provider_id: str,
|
|
date: date,
|
|
duration: int = 30
|
|
) -> Dict[str, any]:
|
|
"""
|
|
Get available time slots for a provider on a specific date.
|
|
|
|
Args:
|
|
provider_id: UUID of the provider (User ID)
|
|
date: Date to check availability
|
|
duration: Appointment duration in minutes (default: 30)
|
|
|
|
Returns:
|
|
Dictionary with 'slots' list and additional metadata including 'reason' if no slots
|
|
"""
|
|
# Try to get User first (primary method)
|
|
try:
|
|
user = User.objects.get(id=provider_id)
|
|
# Check if user has a provider profile
|
|
if hasattr(user, 'provider_profile'):
|
|
provider = user.provider_profile
|
|
else:
|
|
# User doesn't have provider profile, try to find schedules by user
|
|
provider = None
|
|
except User.DoesNotExist:
|
|
# Fallback: try Provider model directly
|
|
try:
|
|
provider = Provider.objects.get(id=provider_id, is_available=True)
|
|
user = provider.user
|
|
except Provider.DoesNotExist:
|
|
return {
|
|
'slots': [],
|
|
'reason': 'provider_not_found',
|
|
'message': 'Provider not found or not available'
|
|
}
|
|
|
|
# Get day of week (0=Sunday, 6=Saturday)
|
|
day_of_week = (date.weekday() + 1) % 7 # Convert Python's Monday=0 to Sunday=0
|
|
|
|
# Get provider's schedule for this day
|
|
# Try both provider and user-based schedules
|
|
if provider:
|
|
schedules = Schedule.objects.filter(
|
|
provider=provider,
|
|
day_of_week=day_of_week,
|
|
is_active=True
|
|
)
|
|
else:
|
|
# No schedules if no provider profile
|
|
return {
|
|
'slots': [],
|
|
'reason': 'no_provider_profile',
|
|
'message': 'Provider profile not configured'
|
|
}
|
|
|
|
# Get all schedules for this provider to show working days
|
|
all_schedules = Schedule.objects.filter(
|
|
provider=provider,
|
|
is_active=True
|
|
).order_by('day_of_week')
|
|
|
|
working_days = []
|
|
day_names = ['Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday']
|
|
for schedule in all_schedules:
|
|
day_name = day_names[schedule.day_of_week]
|
|
if day_name not in working_days:
|
|
working_days.append(day_name)
|
|
|
|
if not schedules.exists():
|
|
return {
|
|
'slots': [],
|
|
'reason': 'no_schedule',
|
|
'message': f'Provider does not work on {day_names[day_of_week]}',
|
|
'working_days': ', '.join(working_days) if working_days else 'Not configured',
|
|
'provider_name': provider.user.get_full_name()
|
|
}
|
|
|
|
# Get all booked appointments for this provider on this date
|
|
# Check appointments by both provider and user
|
|
booked_appointments = Appointment.objects.filter(
|
|
Q(provider=provider) if provider else Q(provider__user=user),
|
|
scheduled_date=date,
|
|
status__in=[
|
|
Appointment.Status.BOOKED,
|
|
Appointment.Status.CONFIRMED,
|
|
Appointment.Status.ARRIVED,
|
|
Appointment.Status.IN_PROGRESS
|
|
]
|
|
).values_list('scheduled_time', 'duration')
|
|
|
|
# Convert booked appointments to time ranges
|
|
booked_ranges = []
|
|
for appt_time, appt_duration in booked_appointments:
|
|
start = datetime.combine(date, appt_time)
|
|
end = start + timedelta(minutes=appt_duration)
|
|
booked_ranges.append((start.time(), end.time()))
|
|
|
|
# Generate available slots from all schedules
|
|
available_slots = []
|
|
|
|
for schedule in schedules:
|
|
slots = AvailabilityService._generate_slots_for_schedule(
|
|
schedule=schedule,
|
|
date=date,
|
|
duration=duration,
|
|
booked_ranges=booked_ranges
|
|
)
|
|
available_slots.extend(slots)
|
|
|
|
# Sort slots by time and remove duplicates
|
|
available_slots = sorted(set(available_slots))
|
|
|
|
# Filter out past slots if the date is today
|
|
now = timezone.localtime(timezone.now()) # Ensure we get local time
|
|
today = now.date()
|
|
|
|
if date == today:
|
|
# Get current time in the local timezone
|
|
current_time = now.time()
|
|
available_slots = [
|
|
slot for slot in available_slots
|
|
if slot > current_time
|
|
]
|
|
|
|
# Format slots for display
|
|
formatted_slots = [
|
|
{
|
|
'time': slot.strftime('%H:%M'),
|
|
'display': slot.strftime('%I:%M %p')
|
|
}
|
|
for slot in available_slots
|
|
]
|
|
|
|
# If no slots available, determine the reason
|
|
if not formatted_slots:
|
|
# Check if all slots are booked
|
|
total_possible_slots = 0
|
|
for schedule in schedules:
|
|
current = datetime.combine(date, schedule.start_time)
|
|
end = datetime.combine(date, schedule.end_time)
|
|
while current + timedelta(minutes=duration) <= end:
|
|
total_possible_slots += 1
|
|
current += timedelta(minutes=schedule.slot_duration)
|
|
|
|
if total_possible_slots > 0:
|
|
# Slots exist but all are booked
|
|
return {
|
|
'slots': [],
|
|
'reason': 'all_booked',
|
|
'message': 'All time slots are fully booked for this date',
|
|
'total_slots': total_possible_slots,
|
|
'booked_slots': len(booked_ranges),
|
|
'provider_name': provider.user.get_full_name()
|
|
}
|
|
else:
|
|
# No slots could be generated (shouldn't happen if schedule exists)
|
|
return {
|
|
'slots': [],
|
|
'reason': 'no_slots_generated',
|
|
'message': 'No time slots could be generated for this schedule',
|
|
'provider_name': provider.user.get_full_name()
|
|
}
|
|
|
|
# Return successful result with slots
|
|
return {
|
|
'slots': formatted_slots,
|
|
'reason': None,
|
|
'message': f'{len(formatted_slots)} slot(s) available',
|
|
'provider_name': provider.user.get_full_name(),
|
|
'working_days': ', '.join(working_days) if working_days else None
|
|
}
|
|
|
|
@staticmethod
|
|
def _generate_slots_for_schedule(
|
|
schedule: Schedule,
|
|
date: date,
|
|
duration: int,
|
|
booked_ranges: List[Tuple[time, time]]
|
|
) -> List[time]:
|
|
"""
|
|
Generate time slots for a specific schedule.
|
|
|
|
Args:
|
|
schedule: Schedule object
|
|
date: Date for the slots
|
|
duration: Appointment duration in minutes
|
|
booked_ranges: List of (start_time, end_time) tuples for booked slots
|
|
|
|
Returns:
|
|
List of available time objects
|
|
"""
|
|
slots = []
|
|
|
|
# Start from schedule start time
|
|
current = datetime.combine(date, schedule.start_time)
|
|
end = datetime.combine(date, schedule.end_time)
|
|
|
|
# Use schedule's slot duration or provided duration
|
|
slot_duration = schedule.slot_duration
|
|
|
|
while current + timedelta(minutes=duration) <= end:
|
|
slot_time = current.time()
|
|
slot_end_time = (current + timedelta(minutes=duration)).time()
|
|
|
|
# Check if this slot conflicts with any booked appointment
|
|
is_available = True
|
|
for booked_start, booked_end in booked_ranges:
|
|
# Check for overlap
|
|
if AvailabilityService._times_overlap(
|
|
slot_time, slot_end_time,
|
|
booked_start, booked_end
|
|
):
|
|
is_available = False
|
|
break
|
|
|
|
if is_available:
|
|
slots.append(slot_time)
|
|
|
|
# Move to next slot
|
|
current += timedelta(minutes=slot_duration)
|
|
|
|
return slots
|
|
|
|
@staticmethod
|
|
def _times_overlap(
|
|
start1: time,
|
|
end1: time,
|
|
start2: time,
|
|
end2: time
|
|
) -> bool:
|
|
"""
|
|
Check if two time ranges overlap.
|
|
|
|
Args:
|
|
start1: Start time of first range
|
|
end1: End time of first range
|
|
start2: Start time of second range
|
|
end2: End time of second range
|
|
|
|
Returns:
|
|
True if ranges overlap, False otherwise
|
|
"""
|
|
# Convert times to minutes for easier comparison
|
|
start1_mins = start1.hour * 60 + start1.minute
|
|
end1_mins = end1.hour * 60 + end1.minute
|
|
start2_mins = start2.hour * 60 + start2.minute
|
|
end2_mins = end2.hour * 60 + end2.minute
|
|
|
|
# Check for overlap
|
|
return not (end1_mins <= start2_mins or end2_mins <= start1_mins)
|
|
|
|
@staticmethod
|
|
def is_slot_available(
|
|
provider_id: str,
|
|
date: date,
|
|
time: time,
|
|
duration: int = 30,
|
|
exclude_appointment_id: str = None
|
|
) -> bool:
|
|
"""
|
|
Check if a specific time slot is available for a provider.
|
|
|
|
Args:
|
|
provider_id: UUID of the provider
|
|
date: Date to check
|
|
time: Time to check
|
|
duration: Appointment duration in minutes
|
|
exclude_appointment_id: Optional appointment ID to exclude (for rescheduling)
|
|
|
|
Returns:
|
|
True if slot is available, False otherwise
|
|
"""
|
|
try:
|
|
provider = Provider.objects.get(id=provider_id, is_available=True)
|
|
except Provider.DoesNotExist:
|
|
return False
|
|
|
|
# Get day of week
|
|
day_of_week = (date.weekday() + 1) % 7
|
|
|
|
# Check if provider has a schedule for this day/time
|
|
schedule_exists = Schedule.objects.filter(
|
|
provider=provider,
|
|
day_of_week=day_of_week,
|
|
is_active=True,
|
|
start_time__lte=time,
|
|
end_time__gte=time
|
|
).exists()
|
|
|
|
if not schedule_exists:
|
|
return False
|
|
|
|
# Check for conflicting appointments
|
|
slot_start = datetime.combine(date, time)
|
|
slot_end = slot_start + timedelta(minutes=duration)
|
|
|
|
conflicting_query = Q(
|
|
provider=provider,
|
|
scheduled_date=date,
|
|
status__in=[
|
|
Appointment.Status.BOOKED,
|
|
Appointment.Status.CONFIRMED,
|
|
Appointment.Status.ARRIVED,
|
|
Appointment.Status.IN_PROGRESS
|
|
]
|
|
)
|
|
|
|
# Exclude specific appointment if provided (for rescheduling)
|
|
if exclude_appointment_id:
|
|
conflicting_query &= ~Q(id=exclude_appointment_id)
|
|
|
|
conflicting_appointments = Appointment.objects.filter(conflicting_query)
|
|
|
|
for appt in conflicting_appointments:
|
|
appt_start = datetime.combine(date, appt.scheduled_time)
|
|
appt_end = appt_start + timedelta(minutes=appt.duration)
|
|
|
|
# Check for overlap
|
|
if not (slot_end <= appt_start or appt_end <= slot_start):
|
|
return False
|
|
|
|
return True
|
|
|
|
@staticmethod
|
|
def get_provider_schedule_summary(
|
|
provider_id: str,
|
|
start_date: date,
|
|
end_date: date
|
|
) -> Dict[str, any]:
|
|
"""
|
|
Get a summary of provider's schedule and availability.
|
|
|
|
Args:
|
|
provider_id: UUID of the provider
|
|
start_date: Start date of the period
|
|
end_date: End date of the period
|
|
|
|
Returns:
|
|
Dictionary with schedule summary information
|
|
"""
|
|
try:
|
|
provider = Provider.objects.get(id=provider_id)
|
|
except Provider.DoesNotExist:
|
|
return {}
|
|
|
|
# Get all schedules
|
|
schedules = Schedule.objects.filter(
|
|
provider=provider,
|
|
is_active=True
|
|
).order_by('day_of_week', 'start_time')
|
|
|
|
# Get appointments in date range
|
|
appointments = Appointment.objects.filter(
|
|
provider=provider,
|
|
scheduled_date__gte=start_date,
|
|
scheduled_date__lte=end_date,
|
|
status__in=[
|
|
Appointment.Status.BOOKED,
|
|
Appointment.Status.CONFIRMED,
|
|
Appointment.Status.ARRIVED,
|
|
Appointment.Status.IN_PROGRESS,
|
|
Appointment.Status.COMPLETED
|
|
]
|
|
)
|
|
|
|
return {
|
|
'provider': {
|
|
'id': str(provider.id),
|
|
'name': provider.user.get_full_name(),
|
|
'is_available': provider.is_available,
|
|
'max_daily_appointments': provider.max_daily_appointments
|
|
},
|
|
'schedules': [
|
|
{
|
|
'day': schedule.get_day_of_week_display(),
|
|
'start_time': schedule.start_time.strftime('%H:%M'),
|
|
'end_time': schedule.end_time.strftime('%H:%M'),
|
|
'slot_duration': schedule.slot_duration
|
|
}
|
|
for schedule in schedules
|
|
],
|
|
'appointments_count': appointments.count(),
|
|
'date_range': {
|
|
'start': start_date.isoformat(),
|
|
'end': end_date.isoformat()
|
|
}
|
|
}
|