522 lines
17 KiB
Python
522 lines
17 KiB
Python
"""
|
|
Consent Management Service for comprehensive consent handling.
|
|
|
|
This module provides services for:
|
|
- Consent validity checking
|
|
- Auto consent status checks
|
|
- Consent version control
|
|
- Consent expiry alerts
|
|
"""
|
|
|
|
import logging
|
|
from datetime import date, timedelta
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
from django.db.models import Q
|
|
from django.utils import timezone
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from core.models import Consent, ConsentTemplate, Patient, Tenant
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConsentManagementService:
|
|
"""Service for managing consent lifecycle and validation."""
|
|
|
|
@staticmethod
|
|
def check_patient_consent_status(
|
|
patient: Patient,
|
|
consent_type: str = None
|
|
) -> Dict:
|
|
"""
|
|
Check consent status for a patient.
|
|
|
|
Args:
|
|
patient: Patient instance
|
|
consent_type: Optional specific consent type to check
|
|
|
|
Returns:
|
|
Dictionary with consent status information
|
|
"""
|
|
query = Consent.objects.filter(
|
|
patient=patient,
|
|
is_active=True
|
|
)
|
|
|
|
if consent_type:
|
|
query = query.filter(consent_type=consent_type)
|
|
|
|
consents = query.order_by('-created_at')
|
|
|
|
status = {
|
|
'has_valid_consent': False,
|
|
'expired_consents': [],
|
|
'expiring_soon': [],
|
|
'active_consents': [],
|
|
'missing_types': [],
|
|
}
|
|
|
|
# Check each consent type
|
|
required_types = [
|
|
Consent.ConsentType.GENERAL_TREATMENT,
|
|
Consent.ConsentType.SERVICE_SPECIFIC,
|
|
]
|
|
|
|
for req_type in required_types:
|
|
type_consents = consents.filter(consent_type=req_type)
|
|
|
|
if not type_consents.exists():
|
|
status['missing_types'].append(req_type)
|
|
continue
|
|
|
|
latest = type_consents.first()
|
|
|
|
if latest.is_expired:
|
|
status['expired_consents'].append({
|
|
'id': str(latest.id),
|
|
'type': latest.get_consent_type_display(),
|
|
'expiry_date': latest.expiry_date,
|
|
'days_expired': abs(latest.days_until_expiry) if latest.days_until_expiry else 0
|
|
})
|
|
elif latest.needs_renewal:
|
|
status['expiring_soon'].append({
|
|
'id': str(latest.id),
|
|
'type': latest.get_consent_type_display(),
|
|
'expiry_date': latest.expiry_date,
|
|
'days_remaining': latest.days_until_expiry
|
|
})
|
|
else:
|
|
status['active_consents'].append({
|
|
'id': str(latest.id),
|
|
'type': latest.get_consent_type_display(),
|
|
'expiry_date': latest.expiry_date,
|
|
'days_remaining': latest.days_until_expiry
|
|
})
|
|
|
|
# Patient has valid consent if all required types are active
|
|
status['has_valid_consent'] = (
|
|
len(status['missing_types']) == 0 and
|
|
len(status['expired_consents']) == 0
|
|
)
|
|
|
|
return status
|
|
|
|
@staticmethod
|
|
def get_expiring_consents(
|
|
tenant: Tenant,
|
|
days_threshold: int = 30
|
|
) -> List[Dict]:
|
|
"""
|
|
Get all consents expiring within threshold days.
|
|
|
|
Args:
|
|
tenant: Tenant instance
|
|
days_threshold: Number of days to look ahead
|
|
|
|
Returns:
|
|
List of expiring consent information
|
|
"""
|
|
threshold_date = date.today() + timedelta(days=days_threshold)
|
|
|
|
expiring_consents = Consent.objects.filter(
|
|
tenant=tenant,
|
|
is_active=True,
|
|
expiry_date__lte=threshold_date,
|
|
expiry_date__gte=date.today()
|
|
).select_related('patient').order_by('expiry_date')
|
|
|
|
result = []
|
|
for consent in expiring_consents:
|
|
result.append({
|
|
'consent_id': str(consent.id),
|
|
'patient_id': str(consent.patient.id),
|
|
'patient_name': consent.patient.full_name_en,
|
|
'patient_mrn': consent.patient.mrn,
|
|
'consent_type': consent.get_consent_type_display(),
|
|
'expiry_date': consent.expiry_date,
|
|
'days_remaining': consent.days_until_expiry,
|
|
'caregiver_phone': consent.patient.caregiver_phone,
|
|
'caregiver_email': consent.patient.email or consent.patient.caregiver_name,
|
|
})
|
|
|
|
return result
|
|
|
|
@staticmethod
|
|
def get_expired_consents(tenant: Tenant) -> List[Dict]:
|
|
"""
|
|
Get all expired consents.
|
|
|
|
Args:
|
|
tenant: Tenant instance
|
|
|
|
Returns:
|
|
List of expired consent information
|
|
"""
|
|
expired_consents = Consent.objects.filter(
|
|
tenant=tenant,
|
|
is_active=True,
|
|
expiry_date__lt=date.today()
|
|
).select_related('patient').order_by('expiry_date')
|
|
|
|
result = []
|
|
for consent in expired_consents:
|
|
result.append({
|
|
'consent_id': str(consent.id),
|
|
'patient_id': str(consent.patient.id),
|
|
'patient_name': consent.patient.full_name_en,
|
|
'patient_mrn': consent.patient.mrn,
|
|
'consent_type': consent.get_consent_type_display(),
|
|
'expiry_date': consent.expiry_date,
|
|
'days_expired': abs(consent.days_until_expiry) if consent.days_until_expiry else 0,
|
|
'caregiver_phone': consent.patient.caregiver_phone,
|
|
})
|
|
|
|
return result
|
|
|
|
@staticmethod
|
|
def create_consent_from_template(
|
|
patient: Patient,
|
|
template: ConsentTemplate,
|
|
expiry_days: int = 365,
|
|
language: str = 'en'
|
|
) -> Consent:
|
|
"""
|
|
Create a new consent from a template.
|
|
|
|
Args:
|
|
patient: Patient instance
|
|
template: ConsentTemplate instance
|
|
expiry_days: Number of days until expiry
|
|
language: Language for content ('en' or 'ar')
|
|
|
|
Returns:
|
|
Created Consent instance
|
|
"""
|
|
# Get populated content
|
|
content = template.get_populated_content(patient, language)
|
|
|
|
# Calculate expiry date
|
|
expiry_date = date.today() + timedelta(days=expiry_days)
|
|
|
|
# Create consent
|
|
consent = Consent.objects.create(
|
|
tenant=patient.tenant,
|
|
patient=patient,
|
|
consent_type=template.consent_type,
|
|
content_text=content,
|
|
version=template.version,
|
|
expiry_date=expiry_date,
|
|
is_active=True
|
|
)
|
|
|
|
logger.info(f"Created consent {consent.id} for patient {patient.mrn} from template {template.id}")
|
|
|
|
return consent
|
|
|
|
@staticmethod
|
|
def renew_consent(
|
|
old_consent: Consent,
|
|
expiry_days: int = 365
|
|
) -> Consent:
|
|
"""
|
|
Renew an expired or expiring consent.
|
|
|
|
Args:
|
|
old_consent: Existing consent to renew
|
|
expiry_days: Number of days until new expiry
|
|
|
|
Returns:
|
|
New Consent instance
|
|
"""
|
|
# Deactivate old consent
|
|
old_consent.is_active = False
|
|
old_consent.save()
|
|
|
|
# Create new consent with incremented version
|
|
new_consent = Consent.objects.create(
|
|
tenant=old_consent.tenant,
|
|
patient=old_consent.patient,
|
|
consent_type=old_consent.consent_type,
|
|
content_text=old_consent.content_text,
|
|
version=old_consent.version + 1,
|
|
expiry_date=date.today() + timedelta(days=expiry_days),
|
|
is_active=True
|
|
)
|
|
|
|
logger.info(f"Renewed consent {old_consent.id} with new consent {new_consent.id}")
|
|
|
|
return new_consent
|
|
|
|
@staticmethod
|
|
def get_active_template_version(
|
|
tenant: Tenant,
|
|
consent_type: str
|
|
) -> Optional[ConsentTemplate]:
|
|
"""
|
|
Get the active (latest) version of a consent template.
|
|
|
|
Args:
|
|
tenant: Tenant instance
|
|
consent_type: Type of consent
|
|
|
|
Returns:
|
|
Latest active ConsentTemplate or None
|
|
"""
|
|
return ConsentTemplate.objects.filter(
|
|
tenant=tenant,
|
|
consent_type=consent_type,
|
|
is_active=True
|
|
).order_by('-version').first()
|
|
|
|
@staticmethod
|
|
def create_new_template_version(
|
|
old_template: ConsentTemplate,
|
|
content_en: str,
|
|
content_ar: str = None
|
|
) -> ConsentTemplate:
|
|
"""
|
|
Create a new version of a consent template.
|
|
|
|
Args:
|
|
old_template: Existing template
|
|
content_en: New English content
|
|
content_ar: New Arabic content (optional)
|
|
|
|
Returns:
|
|
New ConsentTemplate instance
|
|
"""
|
|
# Deactivate old template
|
|
old_template.is_active = False
|
|
old_template.save()
|
|
|
|
# Create new version
|
|
new_template = ConsentTemplate.objects.create(
|
|
tenant=old_template.tenant,
|
|
consent_type=old_template.consent_type,
|
|
title_en=old_template.title_en,
|
|
title_ar=old_template.title_ar,
|
|
content_en=content_en,
|
|
content_ar=content_ar or old_template.content_ar,
|
|
version=old_template.version + 1,
|
|
is_active=True
|
|
)
|
|
|
|
logger.info(f"Created new template version {new_template.id} (v{new_template.version})")
|
|
|
|
return new_template
|
|
|
|
@staticmethod
|
|
def validate_consent_before_booking(
|
|
patient: Patient,
|
|
required_types: List[str] = None
|
|
) -> Tuple[bool, List[str]]:
|
|
"""
|
|
Validate that patient has all required active consents before booking.
|
|
|
|
Args:
|
|
patient: Patient instance
|
|
required_types: List of required consent types (defaults to GENERAL_TREATMENT)
|
|
|
|
Returns:
|
|
Tuple of (is_valid, list_of_missing_or_expired_types)
|
|
"""
|
|
if required_types is None:
|
|
required_types = [Consent.ConsentType.GENERAL_TREATMENT]
|
|
|
|
missing_or_expired = []
|
|
|
|
for consent_type in required_types:
|
|
# Get latest consent of this type
|
|
latest_consent = Consent.objects.filter(
|
|
patient=patient,
|
|
consent_type=consent_type,
|
|
is_active=True
|
|
).order_by('-created_at').first()
|
|
|
|
if not latest_consent:
|
|
missing_or_expired.append(f"{consent_type} (Missing)")
|
|
elif latest_consent.is_expired:
|
|
missing_or_expired.append(f"{consent_type} (Expired)")
|
|
|
|
is_valid = len(missing_or_expired) == 0
|
|
|
|
return is_valid, missing_or_expired
|
|
|
|
@staticmethod
|
|
def get_consent_statistics(tenant: Tenant) -> Dict:
|
|
"""
|
|
Get comprehensive consent statistics for a tenant.
|
|
|
|
Args:
|
|
tenant: Tenant instance
|
|
|
|
Returns:
|
|
Dictionary with consent statistics
|
|
"""
|
|
all_consents = Consent.objects.filter(tenant=tenant, is_active=True)
|
|
|
|
stats = {
|
|
'total_active': all_consents.count(),
|
|
'expiring_30_days': all_consents.filter(
|
|
expiry_date__lte=date.today() + timedelta(days=30),
|
|
expiry_date__gte=date.today()
|
|
).count(),
|
|
'expiring_7_days': all_consents.filter(
|
|
expiry_date__lte=date.today() + timedelta(days=7),
|
|
expiry_date__gte=date.today()
|
|
).count(),
|
|
'expired': all_consents.filter(
|
|
expiry_date__lt=date.today()
|
|
).count(),
|
|
'by_type': {},
|
|
'patients_without_consent': 0,
|
|
}
|
|
|
|
# Count by type
|
|
for consent_type, display_name in Consent.ConsentType.choices:
|
|
stats['by_type'][consent_type] = all_consents.filter(
|
|
consent_type=consent_type
|
|
).count()
|
|
|
|
# Count patients without any active consent
|
|
from core.models import Patient
|
|
total_patients = Patient.objects.filter(tenant=tenant).count()
|
|
patients_with_consent = all_consents.values('patient').distinct().count()
|
|
stats['patients_without_consent'] = total_patients - patients_with_consent
|
|
|
|
return stats
|
|
|
|
|
|
class ConsentNotificationService:
|
|
"""Service for sending consent-related notifications."""
|
|
|
|
@staticmethod
|
|
def send_expiry_reminder(consent: Consent) -> bool:
|
|
"""
|
|
Send expiry reminder to patient/caregiver.
|
|
|
|
Args:
|
|
consent: Consent instance
|
|
|
|
Returns:
|
|
True if notification sent successfully
|
|
"""
|
|
from core.tasks import send_email_task, create_notification_task
|
|
|
|
patient = consent.patient
|
|
days_remaining = consent.days_until_expiry
|
|
|
|
# Prepare message
|
|
message = _(
|
|
"Dear {caregiver_name},\n\n"
|
|
"The {consent_type} consent for {patient_name} (MRN: {mrn}) "
|
|
"will expire in {days} days on {expiry_date}.\n\n"
|
|
"Please contact the clinic to renew the consent.\n\n"
|
|
"Best regards,\nAgdar Centre"
|
|
).format(
|
|
caregiver_name=patient.caregiver_name or "Guardian",
|
|
consent_type=consent.get_consent_type_display(),
|
|
patient_name=patient.full_name_en,
|
|
mrn=patient.mrn,
|
|
days=days_remaining,
|
|
expiry_date=consent.expiry_date.strftime('%Y-%m-%d')
|
|
)
|
|
|
|
# Send email if available
|
|
if patient.email:
|
|
send_email_task.delay(
|
|
subject=f"Consent Expiry Reminder - {patient.mrn}",
|
|
message=message,
|
|
recipient_list=[patient.email]
|
|
)
|
|
|
|
# Send SMS if phone available
|
|
if patient.caregiver_phone:
|
|
# TODO: Integrate with SMS service
|
|
pass
|
|
|
|
logger.info(f"Sent expiry reminder for consent {consent.id}")
|
|
return True
|
|
|
|
@staticmethod
|
|
def send_expired_notification(consent: Consent) -> bool:
|
|
"""
|
|
Send notification that consent has expired.
|
|
|
|
Args:
|
|
consent: Consent instance
|
|
|
|
Returns:
|
|
True if notification sent successfully
|
|
"""
|
|
from core.tasks import send_email_task
|
|
|
|
patient = consent.patient
|
|
|
|
message = _(
|
|
"Dear {caregiver_name},\n\n"
|
|
"The {consent_type} consent for {patient_name} (MRN: {mrn}) "
|
|
"has expired as of {expiry_date}.\n\n"
|
|
"Please contact the clinic immediately to renew the consent. "
|
|
"No appointments can be booked until the consent is renewed.\n\n"
|
|
"Best regards,\nAgdar Centre"
|
|
).format(
|
|
caregiver_name=patient.caregiver_name or "Guardian",
|
|
consent_type=consent.get_consent_type_display(),
|
|
patient_name=patient.full_name_en,
|
|
mrn=patient.mrn,
|
|
expiry_date=consent.expiry_date.strftime('%Y-%m-%d')
|
|
)
|
|
|
|
# Send email if available
|
|
if patient.email:
|
|
send_email_task.delay(
|
|
subject=f"URGENT: Consent Expired - {patient.mrn}",
|
|
message=message,
|
|
recipient_list=[patient.email]
|
|
)
|
|
|
|
logger.info(f"Sent expired notification for consent {consent.id}")
|
|
return True
|
|
|
|
@staticmethod
|
|
def notify_reception_expired_consents(tenant: Tenant, expired_list: List[Dict]) -> bool:
|
|
"""
|
|
Notify reception staff about expired consents.
|
|
|
|
Args:
|
|
tenant: Tenant instance
|
|
expired_list: List of expired consent information
|
|
|
|
Returns:
|
|
True if notification sent successfully
|
|
"""
|
|
from core.tasks import create_notification_task
|
|
from core.models import User
|
|
|
|
# Get reception staff
|
|
reception_users = User.objects.filter(
|
|
tenant=tenant,
|
|
role=User.Role.FRONT_DESK,
|
|
is_active=True
|
|
)
|
|
|
|
if not expired_list:
|
|
return True
|
|
|
|
message = f"{len(expired_list)} patient consent(s) have expired and require renewal."
|
|
|
|
for user in reception_users:
|
|
create_notification_task.delay(
|
|
user_id=str(user.id),
|
|
title="Expired Consents Alert",
|
|
message=message,
|
|
notification_type='WARNING',
|
|
related_object_type='consent',
|
|
related_object_id=None
|
|
)
|
|
|
|
logger.info(f"Notified {reception_users.count()} reception staff about {len(expired_list)} expired consents")
|
|
return True
|