453 lines
16 KiB
Python
453 lines
16 KiB
Python
"""
|
|
PX Sources models - Manages origins of patient feedback
|
|
|
|
This module implements the PX Source management system that:
|
|
- Tracks sources of patient feedback (Complaints and Inquiries)
|
|
- Supports bilingual naming (English/Arabic)
|
|
- Enables status management
|
|
"""
|
|
|
|
from django.contrib.contenttypes.fields import GenericForeignKey
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.db import models
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from apps.core.models import UUIDModel, TimeStampedModel
|
|
|
|
|
|
class PXSource(UUIDModel, TimeStampedModel):
|
|
"""
|
|
PX Source model for managing feedback origins.
|
|
|
|
Simple model with bilingual naming and active status management.
|
|
"""
|
|
|
|
# Code for API references
|
|
code = models.CharField(
|
|
max_length=50, unique=True, help_text="Unique code for API references", blank=True, default=""
|
|
)
|
|
|
|
# Bilingual names
|
|
name_en = models.CharField(max_length=200, help_text="Source name in English")
|
|
name_ar = models.CharField(max_length=200, blank=True, help_text="Source name in Arabic")
|
|
|
|
# Description
|
|
description = models.TextField(blank=True, help_text="Detailed description")
|
|
|
|
# Source type
|
|
SOURCE_TYPE_CHOICES = [
|
|
("internal", _("Internal")),
|
|
("external", _("External")),
|
|
("partner", _("Partner")),
|
|
("government", _("Government")),
|
|
("other", _("Other")),
|
|
]
|
|
source_type = models.CharField(
|
|
max_length=50, choices=SOURCE_TYPE_CHOICES, default="internal", db_index=True, help_text="Type of source"
|
|
)
|
|
|
|
# Contact information for external sources
|
|
contact_email = models.EmailField(blank=True, help_text="Contact email for external sources")
|
|
contact_phone = models.CharField(max_length=20, blank=True, help_text="Contact phone for external sources")
|
|
|
|
# Status
|
|
is_active = models.BooleanField(
|
|
default=True, db_index=True, help_text="Whether this source is active for selection"
|
|
)
|
|
|
|
# Metadata
|
|
metadata = models.JSONField(default=dict, blank=True, help_text="Additional metadata")
|
|
|
|
# Cached usage stats
|
|
total_complaints = models.IntegerField(default=0, editable=False, help_text="Cached total complaints count")
|
|
total_inquiries = models.IntegerField(default=0, editable=False, help_text="Cached total inquiries count")
|
|
|
|
class Meta:
|
|
ordering = ["name_en"]
|
|
verbose_name = "PX Source"
|
|
verbose_name_plural = "PX Sources"
|
|
indexes = [
|
|
models.Index(fields=["is_active", "name_en"]),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.code} - {self.name_en}"
|
|
|
|
def save(self, *args, **kwargs):
|
|
# Auto-generate code if not provided
|
|
if not self.code and self.name_en:
|
|
# Create code from name (e.g., "Hospital A" -> "HOSP-A")
|
|
words = self.name_en.upper().split()
|
|
if len(words) >= 2:
|
|
self.code = "-".join(word[:4] for word in words[:3])
|
|
else:
|
|
self.code = self.name_en[:10].upper().replace(" ", "-")
|
|
# Ensure uniqueness
|
|
from django.db.models import Count
|
|
|
|
base_code = self.code
|
|
counter = 1
|
|
while PXSource.objects.filter(code=self.code).exclude(pk=self.pk).exists():
|
|
self.code = f"{base_code}-{counter}"
|
|
counter += 1
|
|
super().save(*args, **kwargs)
|
|
|
|
def get_localized_name(self, language=None):
|
|
from django.utils.translation import get_language
|
|
|
|
if language is None:
|
|
language = get_language()
|
|
if language == "ar" and self.name_ar:
|
|
return self.name_ar
|
|
return self.name_en
|
|
|
|
def get_localized_description(self):
|
|
"""Get localized description"""
|
|
return self.description
|
|
|
|
def activate(self):
|
|
"""Activate this source"""
|
|
if not self.is_active:
|
|
self.is_active = True
|
|
self.save(update_fields=["is_active"])
|
|
|
|
def deactivate(self):
|
|
"""Deactivate this source"""
|
|
if self.is_active:
|
|
self.is_active = False
|
|
self.save(update_fields=["is_active"])
|
|
|
|
@classmethod
|
|
def get_active_sources(cls):
|
|
"""
|
|
Get all active sources.
|
|
|
|
Returns:
|
|
QuerySet of active PXSource objects
|
|
"""
|
|
return cls.objects.filter(is_active=True).order_by("name_en")
|
|
|
|
def update_usage_stats(self):
|
|
"""Update cached usage statistics"""
|
|
from apps.complaints.models import Complaint, Inquiry
|
|
|
|
self.total_complaints = Complaint.objects.filter(source=self).count()
|
|
self.total_inquiries = Inquiry.objects.filter(source=self).count()
|
|
self.save(update_fields=["total_complaints", "total_inquiries"])
|
|
|
|
def get_usage_stats(self, days=30):
|
|
"""Get usage statistics for the last N days"""
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
|
|
cutoff = timezone.now() - timedelta(days=days)
|
|
return {
|
|
"total_usage": self.usage_records.filter(created_at__gte=cutoff).count(),
|
|
"complaints": self.usage_records.filter(created_at__gte=cutoff, content_type__model="complaint").count(),
|
|
"inquiries": self.usage_records.filter(created_at__gte=cutoff, content_type__model="inquiry").count(),
|
|
}
|
|
|
|
|
|
class SourceUser(UUIDModel, TimeStampedModel):
|
|
"""
|
|
Links users to PX Sources for management.
|
|
|
|
A user can be a source manager for a specific PX Source,
|
|
allowing them to create complaints and inquiries from that source.
|
|
"""
|
|
|
|
user = models.OneToOneField(
|
|
"accounts.User",
|
|
on_delete=models.CASCADE,
|
|
related_name="source_user_profile",
|
|
help_text="User who manages this source",
|
|
)
|
|
source = models.ForeignKey(
|
|
PXSource, on_delete=models.CASCADE, related_name="source_users", help_text="Source managed by this user"
|
|
)
|
|
hospital = models.ForeignKey(
|
|
"organizations.Hospital",
|
|
on_delete=models.CASCADE,
|
|
related_name="source_users",
|
|
help_text="Hospital this source user belongs to",
|
|
null=True,
|
|
blank=True,
|
|
)
|
|
|
|
# Status
|
|
is_active = models.BooleanField(default=True, db_index=True, help_text="Whether this source user is active")
|
|
|
|
# Permissions
|
|
can_create_complaints = models.BooleanField(default=True, help_text="User can create complaints from this source")
|
|
can_create_inquiries = models.BooleanField(default=True, help_text="User can create inquiries from this source")
|
|
can_create_observations = models.BooleanField(default=True, help_text="User can create observations from this source")
|
|
can_create_suggestions = models.BooleanField(default=True, help_text="User can create suggestions from this source")
|
|
|
|
class Meta:
|
|
ordering = ["source__name_en"]
|
|
verbose_name = "Source User"
|
|
verbose_name_plural = "Source Users"
|
|
indexes = [
|
|
models.Index(fields=["user", "is_active"]),
|
|
models.Index(fields=["source", "is_active"]),
|
|
]
|
|
unique_together = [["user", "source"]]
|
|
|
|
def __str__(self):
|
|
return f"{self.user.email} - {self.source.name_en}"
|
|
|
|
def activate(self):
|
|
"""Activate this source user"""
|
|
if not self.is_active:
|
|
self.is_active = True
|
|
self.save(update_fields=["is_active"])
|
|
|
|
def deactivate(self):
|
|
"""Deactivate this source user"""
|
|
if self.is_active:
|
|
self.is_active = False
|
|
self.save(update_fields=["is_active"])
|
|
|
|
@classmethod
|
|
def get_active_source_user(cls, user):
|
|
"""
|
|
Get active source user for a user.
|
|
|
|
Returns:
|
|
SourceUser object or None
|
|
"""
|
|
return cls.objects.filter(user=user, is_active=True).first()
|
|
|
|
|
|
class SourceUsage(UUIDModel, TimeStampedModel):
|
|
"""
|
|
Tracks usage of sources across the system.
|
|
|
|
This model can be used to analyze which sources are most commonly used,
|
|
track trends, and generate reports.
|
|
"""
|
|
|
|
source = models.ForeignKey(PXSource, on_delete=models.CASCADE, related_name="usage_records")
|
|
|
|
# Related object (could be Complaint, Inquiry, or other feedback types)
|
|
content_type = models.ForeignKey(
|
|
"contenttypes.ContentType", on_delete=models.CASCADE, help_text="Type of related object"
|
|
)
|
|
object_id = models.UUIDField(help_text="ID of related object")
|
|
|
|
# Hospital context (optional)
|
|
hospital = models.ForeignKey(
|
|
"organizations.Hospital",
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name="source_usage_records",
|
|
help_text="Hospital where this source was used",
|
|
)
|
|
|
|
# User who selected this source (optional)
|
|
user = models.ForeignKey(
|
|
"accounts.User",
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name="source_usage_records",
|
|
help_text="User who selected this source",
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
verbose_name = "Source Usage"
|
|
verbose_name_plural = "Source Usages"
|
|
indexes = [
|
|
models.Index(fields=["source", "-created_at"]),
|
|
models.Index(fields=["content_type", "object_id"]),
|
|
models.Index(fields=["hospital", "-created_at"]),
|
|
models.Index(fields=["created_at"]),
|
|
]
|
|
unique_together = [["content_type", "object_id"]]
|
|
|
|
def __str__(self):
|
|
return f"{self.source} - {self.created_at.strftime('%Y-%m-%d %H:%M')}"
|
|
|
|
|
|
class SourceComplaint(UUIDModel, TimeStampedModel):
|
|
class StatusChoices(models.TextChoices):
|
|
OPEN = "open", _("Open")
|
|
CONVERTED = "converted", _("Converted")
|
|
CLOSED = "closed", _("Closed")
|
|
|
|
px_source = models.ForeignKey(
|
|
PXSource, on_delete=models.CASCADE, related_name="source_complaints"
|
|
)
|
|
reference_number = models.CharField(max_length=50, unique=True, db_index=True)
|
|
subject = models.CharField(max_length=300)
|
|
description = models.TextField()
|
|
patient_name = models.CharField(max_length=200, blank=True)
|
|
contact_phone = models.CharField(max_length=20, blank=True)
|
|
contact_email = models.EmailField(blank=True)
|
|
metadata = models.JSONField(default=dict, blank=True)
|
|
status = models.CharField(
|
|
max_length=20, choices=StatusChoices.choices, default=StatusChoices.OPEN, db_index=True
|
|
)
|
|
system_complaint = models.ForeignKey(
|
|
"complaints.Complaint",
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name="source_complaint",
|
|
)
|
|
created_by = models.ForeignKey(
|
|
"accounts.User", on_delete=models.SET_NULL, null=True, related_name="created_source_complaints"
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
indexes = [
|
|
models.Index(fields=["px_source", "-created_at"]),
|
|
models.Index(fields=["status"]),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.reference_number} - {self.subject[:50]}"
|
|
|
|
def save(self, *args, **kwargs):
|
|
if not self.reference_number:
|
|
import random
|
|
import string
|
|
|
|
from django.utils import timezone
|
|
|
|
today = timezone.now().strftime("%Y%m%d")
|
|
suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=6))
|
|
self.reference_number = f"SRC-{today}-{suffix}"
|
|
super().save(*args, **kwargs)
|
|
|
|
|
|
class CommunicationRequest(UUIDModel, TimeStampedModel):
|
|
"""
|
|
Patient communication request from PX Source user to PX team.
|
|
|
|
When a patient asks the PX source user to communicate with the PX team,
|
|
a CommunicationRequest is created and PX staff are notified.
|
|
"""
|
|
|
|
class Reason(models.TextChoices):
|
|
COMPLAINT_FOLLOWUP = "complaint_followup", _("Complaint Follow-up")
|
|
GENERAL_INQUIRY = "general_inquiry", _("General Inquiry")
|
|
FEEDBACK = "feedback", _("Feedback Sharing")
|
|
URGENT = "urgent", _("Urgent Matter")
|
|
OTHER = "other", _("Other")
|
|
|
|
class Status(models.TextChoices):
|
|
PENDING = "pending", _("Pending")
|
|
CONTACTED = "contacted", _("Contacted")
|
|
RESOLVED = "resolved", _("Resolved")
|
|
CLOSED = "closed", _("Closed")
|
|
|
|
source_user = models.ForeignKey(
|
|
"SourceUser",
|
|
on_delete=models.CASCADE,
|
|
related_name="communication_requests",
|
|
)
|
|
hospital = models.ForeignKey(
|
|
"organizations.Hospital",
|
|
on_delete=models.CASCADE,
|
|
related_name="communication_requests",
|
|
)
|
|
|
|
patient_name = models.CharField(max_length=200, blank=True)
|
|
patient_phone = models.CharField(max_length=20, blank=True)
|
|
patient_mrn = models.CharField(max_length=50, blank=True, verbose_name="Patient MRN")
|
|
|
|
reason = models.CharField(max_length=30, choices=Reason.choices, default=Reason.GENERAL_INQUIRY)
|
|
message = models.TextField(help_text="Message from the patient/source user to PX team")
|
|
|
|
status = models.CharField(max_length=20, choices=Status.choices, default=Status.PENDING, db_index=True)
|
|
|
|
contacted_at = models.DateTimeField(null=True, blank=True, help_text="When PX team contacted the patient")
|
|
contacted_by = models.ForeignKey(
|
|
"accounts.User",
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name="contacted_communication_requests",
|
|
)
|
|
resolution_notes = models.TextField(blank=True, help_text="Notes from PX team about the resolution")
|
|
|
|
resolved_as_content_type = models.ForeignKey(
|
|
ContentType,
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name="resolved_communication_requests",
|
|
)
|
|
resolved_as_object_id = models.UUIDField(null=True, blank=True)
|
|
resolved_as = GenericForeignKey("resolved_as_content_type", "resolved_as_object_id")
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
verbose_name = "Communication Request"
|
|
verbose_name_plural = "Communication Requests"
|
|
indexes = [
|
|
models.Index(fields=["hospital", "status", "-created_at"]),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"CommReq {self.id} - {self.get_reason_display()} ({self.get_status_display()})"
|
|
|
|
@property
|
|
def is_resolved_as(self):
|
|
return self.resolved_as_content_type is not None
|
|
|
|
@property
|
|
def resolved_as_type_label(self):
|
|
if not self.is_resolved_as:
|
|
return None
|
|
model_name = self.resolved_as_content_type.model
|
|
labels = {
|
|
"complaint": "Complaint",
|
|
"inquiry": "Inquiry",
|
|
"observation": "Observation",
|
|
"feedback": "Suggestion",
|
|
}
|
|
return labels.get(model_name, model_name.title())
|
|
|
|
@property
|
|
def resolved_as_url_name(self):
|
|
if not self.is_resolved_as:
|
|
return None
|
|
model_name = self.resolved_as_content_type.model
|
|
urls = {
|
|
"complaint": "complaints:complaint_detail",
|
|
"inquiry": "inquiries:inquiry_detail",
|
|
"observation": "observations:observation_detail",
|
|
"feedback": "feedback:feedback_detail",
|
|
}
|
|
return urls.get(model_name)
|
|
|
|
def link_to_record(self, record):
|
|
from django.contrib.contenttypes.models import ContentType
|
|
ct = ContentType.objects.get_for_model(record)
|
|
self.resolved_as_content_type = ct
|
|
self.resolved_as_object_id = record.id
|
|
self.save(update_fields=["resolved_as_content_type", "resolved_as_object_id"])
|
|
|
|
@staticmethod
|
|
def get_initial_data(comm_req_id, hospital=None):
|
|
from django.shortcuts import get_object_or_404
|
|
comm_req = get_object_or_404(CommunicationRequest, pk=comm_req_id)
|
|
return {
|
|
"communication_request": comm_req,
|
|
"initial": {
|
|
"hospital": str(comm_req.hospital_id),
|
|
"contact_name": comm_req.patient_name or "",
|
|
"contact_phone": comm_req.patient_phone or "",
|
|
"patient_name": comm_req.patient_name or "",
|
|
"file_number": comm_req.patient_mrn or "",
|
|
"patient_file_number": comm_req.patient_mrn or "",
|
|
"description": comm_req.message or "",
|
|
"message": comm_req.message or "",
|
|
},
|
|
}
|