saani
b58338db2f
- Increase max_length from 100 to 255 for first_name and last_name encrypted fields - Increase phone field max_length from 20 to 255 to accommodate encryption overhead - Add 'id' field to AppointmentRequest admin list_display for easier reference - Remove redundant docstring from _convert_to_datetime method The increased field lengths ensure adequate storage for encrypted data, which typically requires more space than plaintext values.
451 lines
15 KiB
Python
451 lines
15 KiB
Python
import uuid
|
|
from django.db import models
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
from cryptography.fernet import Fernet
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
|
import base64
|
|
import os
|
|
|
|
class EncryptionManager:
|
|
def __init__(self):
|
|
self.fernet = self._get_fernet()
|
|
|
|
def _get_fernet_key(self):
|
|
key = getattr(settings, 'ENCRYPTION_KEY', None) or os.environ.get('ENCRYPTION_KEY')
|
|
if not key:
|
|
key = Fernet.generate_key().decode()
|
|
key = key.encode()
|
|
return key
|
|
|
|
def _get_fernet(self):
|
|
key = self._get_fernet_key()
|
|
return Fernet(key)
|
|
|
|
def encrypt_value(self, value):
|
|
if value is None or value == "":
|
|
return value
|
|
encrypted_value = self.fernet.encrypt(value.encode())
|
|
return base64.urlsafe_b64encode(encrypted_value).decode()
|
|
|
|
def decrypt_value(self, encrypted_value):
|
|
if encrypted_value is None or encrypted_value == "":
|
|
return encrypted_value
|
|
try:
|
|
encrypted_bytes = base64.urlsafe_b64decode(encrypted_value.encode())
|
|
decrypted_value = self.fernet.decrypt(encrypted_bytes)
|
|
return decrypted_value.decode()
|
|
except Exception as e:
|
|
print(f"Decryption error: {e}")
|
|
return encrypted_value
|
|
|
|
encryption_manager = EncryptionManager()
|
|
|
|
class EncryptedCharField(models.CharField):
|
|
def from_db_value(self, value, expression, connection):
|
|
if value is None:
|
|
return value
|
|
return encryption_manager.decrypt_value(value)
|
|
|
|
def get_prep_value(self, value):
|
|
if value is None:
|
|
return value
|
|
return encryption_manager.encrypt_value(value)
|
|
|
|
class EncryptedEmailField(EncryptedCharField):
|
|
def __init__(self, *args, **kwargs):
|
|
kwargs['max_length'] = 254
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def from_db_value(self, value, expression, connection):
|
|
if value is None:
|
|
return value
|
|
return encryption_manager.decrypt_value(value)
|
|
|
|
def get_prep_value(self, value):
|
|
if value is None:
|
|
return value
|
|
return encryption_manager.encrypt_value(value)
|
|
|
|
class EncryptedTextField(models.TextField):
|
|
def from_db_value(self, value, expression, connection):
|
|
if value is None:
|
|
return value
|
|
return encryption_manager.decrypt_value(value)
|
|
|
|
def get_prep_value(self, value):
|
|
if value is None:
|
|
return value
|
|
return encryption_manager.encrypt_value(value)
|
|
|
|
class AdminWeeklyAvailability(models.Model):
|
|
DAYS_OF_WEEK = [
|
|
(0, 'Monday'),
|
|
(1, 'Tuesday'),
|
|
(2, 'Wednesday'),
|
|
(3, 'Thursday'),
|
|
(4, 'Friday'),
|
|
(5, 'Saturday'),
|
|
(6, 'Sunday'),
|
|
]
|
|
|
|
TIME_SLOT_CHOICES = [
|
|
('morning', 'Morning (9AM - 12PM)'),
|
|
('afternoon', 'Afternoon (1PM - 5PM)'),
|
|
('evening', 'Evening (6PM - 9PM)'),
|
|
]
|
|
|
|
availability_schedule = models.JSONField(
|
|
default=dict,
|
|
help_text="Dictionary with days as keys and lists of time slots as values"
|
|
)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
verbose_name = 'Admin Weekly Availability'
|
|
verbose_name_plural = 'Admin Weekly Availability'
|
|
|
|
def set_availability(self, day, time_slots):
|
|
if not self.availability_schedule:
|
|
self.availability_schedule = {}
|
|
|
|
if day not in [str(d[0]) for d in self.DAYS_OF_WEEK]:
|
|
raise ValueError(f"Invalid day: {day}")
|
|
|
|
valid_slots = [slot[0] for slot in self.TIME_SLOT_CHOICES]
|
|
for slot in time_slots:
|
|
if slot not in valid_slots:
|
|
raise ValueError(f"Invalid time slot: {slot}")
|
|
|
|
self.availability_schedule[str(day)] = time_slots
|
|
|
|
def get_availability_for_day(self, day):
|
|
return self.availability_schedule.get(str(day), [])
|
|
|
|
def is_available(self, day, time_slot):
|
|
return time_slot in self.get_availability_for_day(day)
|
|
|
|
def get_all_available_slots(self):
|
|
available_slots = []
|
|
for day_num, time_slots in self.availability_schedule.items():
|
|
day_name = dict(self.DAYS_OF_WEEK).get(int(day_num))
|
|
for time_slot in time_slots:
|
|
time_display = dict(self.TIME_SLOT_CHOICES).get(time_slot)
|
|
available_slots.append({
|
|
'day_num': int(day_num),
|
|
'day_name': day_name,
|
|
'time_slot': time_slot,
|
|
'time_display': time_display
|
|
})
|
|
return available_slots
|
|
|
|
def clear_availability(self, day=None):
|
|
if day is None:
|
|
self.availability_schedule = {}
|
|
else:
|
|
day_str = str(day)
|
|
if day_str in self.availability_schedule:
|
|
del self.availability_schedule[day_str]
|
|
|
|
def __str__(self):
|
|
if not self.availability_schedule:
|
|
return "No availability set"
|
|
|
|
display_strings = []
|
|
for day_num, time_slots in sorted(self.availability_schedule.items()):
|
|
day_name = dict(self.DAYS_OF_WEEK).get(int(day_num))
|
|
slot_displays = [dict(self.TIME_SLOT_CHOICES).get(slot) for slot in time_slots]
|
|
display_strings.append(f"{day_name}: {', '.join(slot_displays)}")
|
|
|
|
return " | ".join(display_strings)
|
|
|
|
class AppointmentRequest(models.Model):
|
|
STATUS_CHOICES = [
|
|
('pending_review', 'Pending Review'),
|
|
('scheduled', 'Scheduled'),
|
|
('rejected', 'Rejected'),
|
|
('completed', 'Completed'),
|
|
('cancelled', 'Cancelled'),
|
|
]
|
|
|
|
TIME_SLOT_CHOICES = [
|
|
('morning', 'Morning (9AM - 12PM)'),
|
|
('afternoon', 'Afternoon (1PM - 5PM)'),
|
|
('evening', 'Evening (6PM - 9PM)'),
|
|
]
|
|
|
|
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
first_name = EncryptedCharField(max_length=255)
|
|
last_name = EncryptedCharField(max_length=255)
|
|
email = EncryptedEmailField()
|
|
phone = EncryptedCharField(max_length=255, blank=True)
|
|
reason = EncryptedTextField(blank=True)
|
|
|
|
preferred_dates = models.JSONField(
|
|
help_text="List of preferred dates (YYYY-MM-DD format)"
|
|
)
|
|
preferred_time_slots = models.JSONField(
|
|
help_text="List of preferred time slots (morning/afternoon/evening)"
|
|
)
|
|
|
|
status = models.CharField(
|
|
max_length=20,
|
|
choices=STATUS_CHOICES,
|
|
default='pending_review'
|
|
)
|
|
|
|
scheduled_datetime = models.DateTimeField(null=True, blank=True)
|
|
scheduled_duration = models.PositiveIntegerField(
|
|
default=60,
|
|
help_text="Duration in minutes"
|
|
)
|
|
rejection_reason = EncryptedTextField(blank=True)
|
|
|
|
jitsi_meet_url = models.URLField(blank=True, help_text="Jitsi Meet URL for the video session")
|
|
jitsi_room_id = models.CharField(max_length=100, unique=True, blank=True, help_text="Jitsi room ID")
|
|
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
ordering = ['-created_at']
|
|
verbose_name = 'Appointment Request'
|
|
verbose_name_plural = 'Appointment Requests'
|
|
indexes = [
|
|
models.Index(fields=['status', 'scheduled_datetime']),
|
|
models.Index(fields=['email', 'created_at']),
|
|
]
|
|
|
|
@property
|
|
def full_name(self):
|
|
return f"{self.first_name} {self.last_name}"
|
|
|
|
@property
|
|
def formatted_created_at(self):
|
|
return self.created_at.strftime("%B %d, %Y at %I:%M %p")
|
|
|
|
@property
|
|
def formatted_scheduled_datetime(self):
|
|
if self.scheduled_datetime:
|
|
return self.scheduled_datetime.strftime("%B %d, %Y at %I:%M %p")
|
|
return None
|
|
|
|
@property
|
|
def has_jitsi_meeting(self):
|
|
return bool(self.jitsi_meet_url and self.jitsi_room_id)
|
|
|
|
@property
|
|
def meeting_in_future(self):
|
|
if not self.scheduled_datetime:
|
|
return False
|
|
return self.scheduled_datetime > timezone.now()
|
|
|
|
@property
|
|
def meeting_duration_display(self):
|
|
hours = self.scheduled_duration // 60
|
|
minutes = self.scheduled_duration % 60
|
|
if hours > 0:
|
|
return f"{hours}h {minutes}m"
|
|
return f"{minutes}m"
|
|
|
|
def get_preferred_dates_display(self):
|
|
try:
|
|
dates = [timezone.datetime.strptime(date, '%Y-%m-%d').strftime('%b %d, %Y')
|
|
for date in self.preferred_dates]
|
|
return ', '.join(dates)
|
|
except:
|
|
return ', '.join(self.preferred_dates)
|
|
|
|
def get_preferred_time_slots_display(self):
|
|
slot_display = {
|
|
'morning': 'Morning',
|
|
'afternoon': 'Afternoon',
|
|
'evening': 'Evening'
|
|
}
|
|
return ', '.join([slot_display.get(slot, slot) for slot in self.preferred_time_slots])
|
|
|
|
def generate_jitsi_room_id(self):
|
|
if not self.jitsi_room_id:
|
|
self.jitsi_room_id = f"therapy_session_{self.id.hex[:16]}"
|
|
return self.jitsi_room_id
|
|
|
|
def create_jitsi_meeting(self):
|
|
if not self.jitsi_room_id:
|
|
self.generate_jitsi_room_id()
|
|
|
|
jitsi_base_url = getattr(settings, 'JITSI_BASE_URL', 'https://meet.jit.si')
|
|
self.jitsi_meet_url = f"{jitsi_base_url}/{self.jitsi_room_id}"
|
|
return self.jitsi_meet_url
|
|
|
|
def get_jitsi_join_info(self):
|
|
if not self.has_jitsi_meeting:
|
|
return None
|
|
|
|
return {
|
|
'meeting_url': self.jitsi_meet_url,
|
|
'room_id': self.jitsi_room_id,
|
|
'scheduled_time': self.formatted_scheduled_datetime,
|
|
'duration': self.meeting_duration_display,
|
|
'join_instructions': 'Click the meeting URL to join the video session. No password required.'
|
|
}
|
|
|
|
def schedule_appointment(self, datetime_obj, duration=60, commit=True):
|
|
self.status = 'scheduled'
|
|
self.scheduled_datetime = datetime_obj
|
|
self.scheduled_duration = duration
|
|
self.rejection_reason = ''
|
|
|
|
self.create_jitsi_meeting()
|
|
|
|
if commit:
|
|
self.save()
|
|
|
|
def reject_appointment(self, reason='', commit=True):
|
|
self.status = 'rejected'
|
|
self.rejection_reason = reason
|
|
self.scheduled_datetime = None
|
|
self.jitsi_meet_url = ''
|
|
self.jitsi_room_id = ''
|
|
if commit:
|
|
self.save()
|
|
|
|
def cancel_appointment(self, reason='', commit=True):
|
|
self.status = 'cancelled'
|
|
self.rejection_reason = reason
|
|
if commit:
|
|
self.save()
|
|
|
|
def complete_appointment(self, commit=True):
|
|
self.status = 'completed'
|
|
if commit:
|
|
self.save()
|
|
|
|
def can_join_meeting(self):
|
|
if not self.scheduled_datetime or not self.has_jitsi_meeting:
|
|
return False
|
|
|
|
if self.status != 'scheduled':
|
|
return False
|
|
|
|
now = timezone.now()
|
|
meeting_start = self.scheduled_datetime
|
|
meeting_end = meeting_start + timezone.timedelta(minutes=self.scheduled_duration + 15) # 15 min buffer
|
|
|
|
return meeting_start - timezone.timedelta(minutes=10) <= now <= meeting_end
|
|
|
|
def get_meeting_status(self):
|
|
if not self.scheduled_datetime:
|
|
return "Not scheduled"
|
|
|
|
now = timezone.now()
|
|
meeting_start = self.scheduled_datetime
|
|
|
|
if now < meeting_start - timezone.timedelta(minutes=10):
|
|
return "Scheduled"
|
|
elif self.can_join_meeting():
|
|
return "Ready to join"
|
|
elif now > meeting_start + timezone.timedelta(minutes=self.scheduled_duration):
|
|
return "Completed"
|
|
else:
|
|
return "Ended"
|
|
|
|
def get_available_time_slots_for_date(self, date_str):
|
|
try:
|
|
from datetime import datetime
|
|
date_obj = datetime.strptime(date_str, '%Y-%m-%d').date()
|
|
day_of_week = date_obj.weekday()
|
|
|
|
availability = AdminWeeklyAvailability.objects.first()
|
|
if not availability:
|
|
return []
|
|
|
|
return availability.get_availability_for_day(day_of_week)
|
|
except Exception as e:
|
|
print(f"Error getting available slots: {e}")
|
|
return []
|
|
|
|
def are_preferences_available(self):
|
|
availability = AdminWeeklyAvailability.objects.first()
|
|
if not availability:
|
|
return False
|
|
|
|
for date_str in self.preferred_dates:
|
|
try:
|
|
from datetime import datetime
|
|
date_obj = datetime.strptime(date_str, '%Y-%m-%d').date()
|
|
day_of_week = date_obj.weekday()
|
|
|
|
available_slots = availability.get_availability_for_day(day_of_week)
|
|
if any(slot in available_slots for slot in self.preferred_time_slots):
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error checking availability for {date_str}: {e}")
|
|
continue
|
|
|
|
return False
|
|
|
|
def get_matching_availability(self):
|
|
availability = AdminWeeklyAvailability.objects.first()
|
|
if not availability:
|
|
return []
|
|
|
|
matching_slots = []
|
|
for date_str in self.preferred_dates:
|
|
try:
|
|
from datetime import datetime
|
|
date_obj = datetime.strptime(date_str, '%Y-%m-%d').date()
|
|
day_of_week = date_obj.weekday()
|
|
day_name = dict(AdminWeeklyAvailability.DAYS_OF_WEEK).get(day_of_week)
|
|
|
|
available_slots = availability.get_availability_for_day(day_of_week)
|
|
matching_time_slots = [slot for slot in self.preferred_time_slots if slot in available_slots]
|
|
|
|
if matching_time_slots:
|
|
matching_slots.append({
|
|
'date': date_str,
|
|
'day_name': day_name,
|
|
'available_slots': matching_time_slots,
|
|
'date_obj': date_obj
|
|
})
|
|
except Exception as e:
|
|
print(f"Error processing {date_str}: {e}")
|
|
continue
|
|
|
|
return matching_slots
|
|
|
|
def __str__(self):
|
|
return f"{self.full_name} - {self.get_status_display()} - {self.created_at.strftime('%Y-%m-%d')}"
|
|
|
|
|
|
def get_admin_availability():
|
|
availability, created = AdminWeeklyAvailability.objects.get_or_create(
|
|
id=1
|
|
)
|
|
return availability
|
|
|
|
def set_admin_availability(availability_dict):
|
|
availability = get_admin_availability()
|
|
|
|
for day, time_slots in availability_dict.items():
|
|
availability.set_availability(day, time_slots)
|
|
|
|
availability.save()
|
|
return availability
|
|
|
|
def get_available_slots_for_week():
|
|
availability = get_admin_availability()
|
|
return availability.get_all_available_slots()
|
|
|
|
def check_date_availability(date_str):
|
|
try:
|
|
from datetime import datetime
|
|
date_obj = datetime.strptime(date_str, '%Y-%m-%d').date()
|
|
day_of_week = date_obj.weekday()
|
|
|
|
availability = get_admin_availability()
|
|
return availability.get_availability_for_day(day_of_week)
|
|
except Exception as e:
|
|
print(f"Error checking date availability: {e}")
|
|
return [] |