Marwan Alwali 84c1fb798e update
2025-09-08 19:52:52 +03:00

660 lines
27 KiB
Python

from django.db import models
from django.contrib.auth import get_user_model
from django.core.validators import MinValueValidator, MaxValueValidator
from decimal import Decimal
User = get_user_model()
class QualityIndicator(models.Model):
"""Quality metrics and performance indicators for monitoring and improvement"""
CATEGORY_CHOICES = [
('clinical', 'Clinical'),
('safety', 'Safety'),
('operational', 'Operational'),
('financial', 'Financial'),
('patient_satisfaction', 'Patient Satisfaction'),
('staff_satisfaction', 'Staff Satisfaction'),
('compliance', 'Compliance'),
('efficiency', 'Efficiency'),
('quality', 'Quality'),
('outcome', 'Outcome'),
]
TYPE_CHOICES = [
('structure', 'Structure'),
('process', 'Process'),
('outcome', 'Outcome'),
('balancing', 'Balancing'),
]
FREQUENCY_CHOICES = [
('daily', 'Daily'),
('weekly', 'Weekly'),
('monthly', 'Monthly'),
('quarterly', 'Quarterly'),
('annually', 'Annually'),
]
tenant = models.ForeignKey('core.Tenant', on_delete=models.CASCADE, related_name='quality_indicators')
name = models.CharField(max_length=200)
description = models.TextField()
category = models.CharField(max_length=50, choices=CATEGORY_CHOICES)
type = models.CharField(max_length=20, choices=TYPE_CHOICES)
measurement_unit = models.CharField(max_length=50)
target_value = models.DecimalField(max_digits=10, decimal_places=2)
current_value = models.DecimalField(max_digits=10, decimal_places=2)
threshold_warning = models.DecimalField(max_digits=10, decimal_places=2)
threshold_critical = models.DecimalField(max_digits=10, decimal_places=2)
calculation_method = models.TextField()
data_source = models.CharField(max_length=200)
frequency = models.CharField(max_length=20, choices=FREQUENCY_CHOICES)
responsible_department = models.ForeignKey('hr.Department', on_delete=models.SET_NULL, null=True, blank=True)
responsible_user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='responsible_quality_indicators')
is_active = models.BooleanField(default=True)
regulatory_requirement = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'quality_indicator'
indexes = [
models.Index(fields=['tenant', 'category', 'type']),
models.Index(fields=['is_active', 'regulatory_requirement']),
models.Index(fields=['responsible_department', 'responsible_user']),
]
ordering = ['name']
def __str__(self):
return f"{self.name} ({self.category})"
@property
def latest_measurement(self):
"""Get the most recent measurement for this indicator"""
return self.measurements.order_by('-measurement_date').first()
@property
def current_status(self):
"""Get current status based on latest measurement"""
latest = self.latest_measurement
if not latest:
return 'no_data'
return latest.status
class QualityMeasurement(models.Model):
"""Individual measurements and values for quality indicators"""
STATUS_CHOICES = [
('within_target', 'Within Target'),
('warning', 'Warning'),
('critical', 'Critical'),
('improving', 'Improving'),
('declining', 'Declining'),
]
tenant = models.ForeignKey('core.Tenant', on_delete=models.CASCADE, related_name='quality_measurements')
indicator = models.ForeignKey(QualityIndicator, on_delete=models.CASCADE, related_name='measurements')
measurement_date = models.DateField()
value = models.DecimalField(max_digits=10, decimal_places=2)
numerator = models.IntegerField(null=True, blank=True)
denominator = models.IntegerField(null=True, blank=True)
status = models.CharField(max_length=20, choices=STATUS_CHOICES)
measurement_method = models.TextField(blank=True)
sample_size = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
data_source = models.CharField(max_length=200, null=True, blank=True)
notes = models.TextField(blank=True)
data_source_reference = models.CharField(max_length=200, blank=True)
verified_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='verified_measurements')
verified_at = models.DateTimeField(null=True, blank=True)
created_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='created_measurements')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'quality_measurement'
indexes = [
models.Index(fields=['tenant', 'indicator', 'measurement_date']),
models.Index(fields=['status', 'verified_by']),
models.Index(fields=['created_by', 'created_at']),
]
ordering = ['-measurement_date']
unique_together = ['indicator', 'measurement_date']
def __str__(self):
return f"{self.indicator.name} - {self.measurement_date}: {self.value}"
def save(self, *args, **kwargs):
"""Auto-determine status based on thresholds"""
if not self.status:
if self.value >= self.indicator.threshold_critical:
self.status = 'critical'
elif self.value >= self.indicator.threshold_warning:
self.status = 'warning'
else:
self.status = 'within_target'
super().save(*args, **kwargs)
class IncidentReport(models.Model):
"""Patient safety incidents, adverse events, and near-miss reporting"""
INCIDENT_TYPE_CHOICES = [
('medication_error', 'Medication Error'),
('fall', 'Fall'),
('infection', 'Infection'),
('equipment_failure', 'Equipment Failure'),
('documentation_error', 'Documentation Error'),
('communication_failure', 'Communication Failure'),
('surgical_complication', 'Surgical Complication'),
('diagnostic_error', 'Diagnostic Error'),
('treatment_delay', 'Treatment Delay'),
('other', 'Other'),
]
SEVERITY_CHOICES = [
('no_harm', 'No Harm'),
('minor_harm', 'Minor Harm'),
('moderate_harm', 'Moderate Harm'),
('severe_harm', 'Severe Harm'),
('death', 'Death'),
]
CATEGORY_CHOICES = [
('patient_safety', 'Patient Safety'),
('medication', 'Medication'),
('infection_control', 'Infection Control'),
('equipment', 'Equipment'),
('documentation', 'Documentation'),
('communication', 'Communication'),
('surgical', 'Surgical'),
('diagnostic', 'Diagnostic'),
('treatment', 'Treatment'),
('environmental', 'Environmental'),
]
STATUS_CHOICES = [
('reported', 'Reported'),
('under_investigation', 'Under Investigation'),
('investigated', 'Investigated'),
('closed', 'Closed'),
('cancelled', 'Cancelled'),
]
PRIORITY_CHOICES = [
('low', 'Low'),
('medium', 'Medium'),
('high', 'High'),
('urgent', 'Urgent'),
]
tenant = models.ForeignKey('core.Tenant', on_delete=models.CASCADE, related_name='incident_reports')
incident_number = models.CharField(max_length=50, unique=True)
title = models.CharField(max_length=200)
description = models.TextField()
incident_type = models.CharField(max_length=30, choices=INCIDENT_TYPE_CHOICES)
severity = models.CharField(max_length=20, choices=SEVERITY_CHOICES)
category = models.CharField(max_length=30, choices=CATEGORY_CHOICES)
location = models.CharField(max_length=200)
incident_date = models.DateField()
incident_time = models.TimeField()
discovered_date = models.DateTimeField()
patient = models.ForeignKey('patients.PatientProfile', on_delete=models.SET_NULL, null=True, blank=True, related_name='incident_reports')
witness_information = models.TextField(null=True, blank=True)
reported_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='reported_incidents')
status = models.CharField(max_length=30, choices=STATUS_CHOICES, default='reported')
priority = models.CharField(max_length=10, choices=PRIORITY_CHOICES, default='medium')
root_cause = models.TextField(blank=True)
contributing_factors = models.TextField(blank=True)
corrective_actions = models.TextField(blank=True)
preventive_actions = models.TextField(blank=True)
assigned_to = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='assigned_incidents')
due_date = models.DateField(null=True, blank=True)
closed_date = models.DateTimeField(null=True, blank=True)
is_confidential = models.BooleanField(default=False)
regulatory_notification = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'incident_report'
indexes = [
models.Index(fields=['tenant', 'incident_type', 'severity']),
models.Index(fields=['status', 'priority', 'assigned_to']),
models.Index(fields=['incident_date', 'patient']),
]
ordering = ['-incident_date']
def __str__(self):
return f"{self.incident_number}: {self.title}"
def save(self, *args, **kwargs):
if not self.incident_number:
# Generate incident number
from django.utils import timezone
year = timezone.now().year
count = IncidentReport.objects.filter(
tenant=self.tenant,
created_at__year=year
).count() + 1
self.incident_number = f"INC-{year}-{count:04d}"
super().save(*args, **kwargs)
class RiskAssessment(models.Model):
"""Risk assessments and risk management activities"""
RISK_CATEGORY_CHOICES = [
('clinical', 'Clinical'),
('operational', 'Operational'),
('financial', 'Financial'),
('regulatory', 'Regulatory'),
('reputational', 'Reputational'),
('strategic', 'Strategic'),
('technology', 'Technology'),
('environmental', 'Environmental'),
('security', 'Security'),
('legal', 'Legal'),
]
RISK_TYPE_CHOICES = [
('patient_safety', 'Patient Safety'),
('quality', 'Quality'),
('compliance', 'Compliance'),
('financial', 'Financial'),
('operational', 'Operational'),
('strategic', 'Strategic'),
('technology', 'Technology'),
('environmental', 'Environmental'),
('security', 'Security'),
('legal', 'Legal'),
]
LIKELIHOOD_CHOICES = [
(1, 'Very Low'),
(2, 'Low'),
(3, 'Medium'),
(4, 'High'),
(5, 'Very High'),
]
IMPACT_CHOICES = [
(1, 'Very Low'),
(2, 'Low'),
(3, 'Medium'),
(4, 'High'),
(5, 'Very High'),
]
RISK_LEVEL_CHOICES = [
('low', 'Low'),
('medium', 'Medium'),
('high', 'High'),
('critical', 'Critical'),
]
CONTROL_EFFECTIVENESS_CHOICES = [
('poor', 'Poor'),
('fair', 'Fair'),
('good', 'Good'),
('excellent', 'Excellent'),
]
STATUS_CHOICES = [
('draft', 'Draft'),
('active', 'Active'),
('under_review', 'Under Review'),
('closed', 'Closed'),
('cancelled', 'Cancelled'),
]
tenant = models.ForeignKey('core.Tenant', on_delete=models.CASCADE, related_name='risk_assessments')
title = models.CharField(max_length=200)
description = models.TextField()
risk_category = models.CharField(max_length=20, choices=RISK_CATEGORY_CHOICES)
risk_type = models.CharField(max_length=20, choices=RISK_TYPE_CHOICES)
likelihood = models.IntegerField(choices=LIKELIHOOD_CHOICES)
impact = models.IntegerField(choices=IMPACT_CHOICES)
risk_score = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(25)])
risk_level = models.CharField(max_length=20, choices=RISK_LEVEL_CHOICES)
current_controls = models.TextField()
control_effectiveness = models.CharField(max_length=20, choices=CONTROL_EFFECTIVENESS_CHOICES)
# residual_likelihood = models.CharField(max_length=20, choices=LIKELIHOOD_CHOICES)
# residual_impact = models.CharField(max_length=20, choices=IMPACT_CHOICES)
# residual_risk_score = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(25)])
# residual_risk_level = models.CharField(max_length=20, choices=RISK_LEVEL_CHOICES)
mitigation_plan = models.TextField()
target_completion_date = models.DateTimeField()
responsible_person = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='responsible_risks')
review_date = models.DateField()
review_notes = models.TextField(null=True, blank=True)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='draft')
incident_report = models.ForeignKey(IncidentReport, on_delete=models.SET_NULL, null=True, blank=True, related_name='risk_assessments')
created_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='created_risk_assessments')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'risk_assessment'
indexes = [
models.Index(fields=['tenant', 'risk_category', 'risk_level']),
models.Index(fields=['status', 'responsible_person']),
models.Index(fields=['review_date', 'created_by']),
]
ordering = ['-risk_score', '-created_at']
def __str__(self):
return f"{self.title} ({self.risk_level})"
def save(self, *args, **kwargs):
"""Auto-calculate risk scores and levels"""
# Risk score calculation (1-25 scale)
likelihood_values = {'very_low': 1, 'low': 2, 'medium': 3, 'high': 4, 'very_high': 5}
impact_values = {'very_low': 1, 'low': 2, 'medium': 3, 'high': 4, 'very_high': 5}
self.risk_score = likelihood_values[self.likelihood] * impact_values[self.impact]
self.residual_risk_score = likelihood_values[self.residual_likelihood] * impact_values[self.residual_impact]
# Risk level determination
if self.risk_score >= 20:
self.risk_level = 'critical'
elif self.risk_score >= 12:
self.risk_level = 'high'
elif self.risk_score >= 6:
self.risk_level = 'medium'
else:
self.risk_level = 'low'
if self.residual_risk_score >= 20:
self.residual_risk_level = 'critical'
elif self.residual_risk_score >= 12:
self.residual_risk_level = 'high'
elif self.residual_risk_score >= 6:
self.residual_risk_level = 'medium'
else:
self.residual_risk_level = 'low'
super().save(*args, **kwargs)
class AuditPlan(models.Model):
"""Quality audits and compliance monitoring plans"""
AUDIT_TYPE_CHOICES = [
('internal', 'Internal'),
('external', 'External'),
('regulatory', 'Regulatory'),
('accreditation', 'Accreditation'),
('quality', 'Quality'),
('compliance', 'Compliance'),
('safety', 'Safety'),
('operational', 'Operational'),
('financial', 'Financial'),
('clinical', 'Clinical'),
]
STATUS_CHOICES = [
('planned', 'Planned'),
('in_progress', 'In Progress'),
('completed', 'Completed'),
('cancelled', 'Cancelled'),
('postponed', 'Postponed'),
]
PRIORITY_CHOICES = [
('low', 'Low'),
('medium', 'Medium'),
('high', 'High'),
('urgent', 'Urgent'),
]
tenant = models.ForeignKey('core.Tenant', on_delete=models.CASCADE, related_name='audit_plans')
title = models.CharField(max_length=200)
description = models.TextField()
audit_type = models.CharField(max_length=20, choices=AUDIT_TYPE_CHOICES)
scope = models.TextField()
criteria = models.TextField()
department = models.ForeignKey('hr.Department', on_delete=models.SET_NULL, null=True, blank=True, related_name='audit_plans')
auditor = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='led_audits')
audit_team = models.ManyToManyField(User, related_name='audit_team_memberships', blank=True)
planned_start_date = models.DateField()
planned_end_date = models.DateField()
actual_start_date = models.DateField(null=True, blank=True)
actual_end_date = models.DateField(null=True, blank=True)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='planned')
priority = models.CharField(max_length=10, choices=PRIORITY_CHOICES, default='medium')
regulatory_requirement = models.BooleanField(default=False)
accreditation_body = models.CharField(max_length=200, blank=True)
objectives = models.TextField(blank=True)
notes = models.TextField(blank=True)
created_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='created_audit_plans')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'audit_plan'
indexes = [
models.Index(fields=['tenant', 'audit_type', 'status']),
models.Index(fields=['department', 'auditor']),
models.Index(fields=['planned_start_date', 'planned_end_date']),
]
ordering = ['planned_start_date']
def __str__(self):
return f"{self.title} ({self.audit_type})"
@property
def findings_count(self):
"""Get count of findings for this audit"""
return self.findings.count()
@property
def open_findings_count(self):
"""Get count of open findings"""
return self.findings.exclude(status='closed').count()
class AuditFinding(models.Model):
"""Audit findings, non-conformities, and observations"""
FINDING_TYPE_CHOICES = [
('non_conformity', 'Non-Conformity'),
('observation', 'Observation'),
('opportunity_for_improvement', 'Opportunity for Improvement'),
('positive_finding', 'Positive Finding'),
('recommendation', 'Recommendation'),
]
SEVERITY_CHOICES = [
('minor', 'Minor'),
('major', 'Major'),
('critical', 'Critical'),
]
CATEGORY_CHOICES = [
('documentation', 'Documentation'),
('process', 'Process'),
('training', 'Training'),
('equipment', 'Equipment'),
('environment', 'Environment'),
('management', 'Management'),
('communication', 'Communication'),
('safety', 'Safety'),
('quality', 'Quality'),
('compliance', 'Compliance'),
]
STATUS_CHOICES = [
('open', 'Open'),
('in_progress', 'In Progress'),
('completed', 'Completed'),
('verified', 'Verified'),
('closed', 'Closed'),
]
tenant = models.ForeignKey('core.Tenant', on_delete=models.CASCADE, related_name='audit_findings')
audit_plan = models.ForeignKey(AuditPlan, on_delete=models.CASCADE, related_name='findings')
finding_number = models.CharField(max_length=50)
title = models.CharField(max_length=200)
description = models.TextField()
finding_type = models.CharField(max_length=30, choices=FINDING_TYPE_CHOICES)
finding_date = models.DateField()
severity = models.CharField(max_length=20, choices=SEVERITY_CHOICES)
category = models.CharField(max_length=20, choices=CATEGORY_CHOICES)
criteria_reference = models.CharField(max_length=200)
evidence = models.TextField()
root_cause = models.TextField(blank=True)
corrective_action_required = models.BooleanField(default=True)
corrective_actions = models.TextField(blank=True)
responsible_person = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='responsible_findings')
target_completion_date = models.DateField(null=True, blank=True)
actual_completion_date = models.DateField(null=True, blank=True)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='open')
verification_method = models.CharField(max_length=200, blank=True)
verified_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='verified_findings')
verified_date = models.DateField(null=True, blank=True)
created_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='created_findings')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'audit_finding'
indexes = [
models.Index(fields=['tenant', 'audit_plan', 'finding_type']),
models.Index(fields=['severity', 'status', 'responsible_person']),
models.Index(fields=['target_completion_date', 'verified_by']),
]
ordering = ['-severity', 'target_completion_date']
unique_together = ['audit_plan', 'finding_number']
def __str__(self):
return f"{self.finding_number}: {self.title}"
def save(self, *args, **kwargs):
if not self.finding_number:
# Generate finding number
count = AuditFinding.objects.filter(audit_plan=self.audit_plan).count() + 1
self.finding_number = f"{self.audit_plan.id}-F{count:03d}"
super().save(*args, **kwargs)
class ImprovementProject(models.Model):
"""Quality improvement projects and initiatives"""
PROJECT_TYPE_CHOICES = [
('quality_improvement', 'Quality Improvement'),
('process_improvement', 'Process Improvement'),
('safety_initiative', 'Safety Initiative'),
('compliance_project', 'Compliance Project'),
('cost_reduction', 'Cost Reduction'),
('efficiency_improvement', 'Efficiency Improvement'),
('patient_satisfaction', 'Patient Satisfaction'),
('staff_satisfaction', 'Staff Satisfaction'),
('technology_implementation', 'Technology Implementation'),
('training_program', 'Training Program'),
]
METHODOLOGY_CHOICES = [
('pdsa', 'PDSA (Plan-Do-Study-Act)'),
('lean', 'Lean'),
('six_sigma', 'Six Sigma'),
('kaizen', 'Kaizen'),
('root_cause_analysis', 'Root Cause Analysis'),
('failure_mode_analysis', 'Failure Mode Analysis'),
('other', 'Other'),
]
STATUS_CHOICES = [
('planned', 'Planned'),
('active', 'Active'),
('on_hold', 'On Hold'),
('completed', 'Completed'),
('cancelled', 'Cancelled'),
]
PHASE_CHOICES = [
('define', 'Define'),
('measure', 'Measure'),
('analyze', 'Analyze'),
('improve', 'Improve'),
('control', 'Control'),
('plan', 'Plan'),
('do', 'Do'),
('study', 'Study'),
('act', 'Act'),
]
tenant = models.ForeignKey('core.Tenant', on_delete=models.CASCADE, related_name='improvement_projects')
project_number = models.CharField(max_length=50, unique=True)
title = models.CharField(max_length=200)
description = models.TextField()
project_type = models.CharField(max_length=30, choices=PROJECT_TYPE_CHOICES)
methodology = models.CharField(max_length=30, choices=METHODOLOGY_CHOICES)
problem_statement = models.TextField()
goal_statement = models.TextField()
success_metrics = models.TextField()
baseline_data = models.TextField(blank=True)
target_metrics = models.TextField()
scope = models.TextField()
project_manager = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='managed_projects')
project_team = models.ManyToManyField(User, related_name='project_team_memberships', blank=True)
sponsor = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='sponsored_projects')
# stakeholders = models.ManyToManyField(User, related_name='stakeholder_memberships', blank=True)
department = models.ForeignKey('hr.Department', on_delete=models.SET_NULL, null=True, blank=True, related_name='improvement_projects')
planned_start_date = models.DateField()
planned_end_date = models.DateField()
actual_start_date = models.DateField(null=True, blank=True)
actual_end_date = models.DateField(null=True, blank=True)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='planned')
phase = models.CharField(max_length=20, choices=PHASE_CHOICES, default='define')
estimated_cost = models.DecimalField(max_digits=12, decimal_places=2, null=True, blank=True)
resources_required = models.TextField(blank=True)
actual_cost = models.DecimalField(max_digits=12, decimal_places=2, null=True, blank=True)
roi_expected = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
roi_actual = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
lessons_learned = models.TextField(blank=True)
notes = models.TextField(blank=True)
sustainability_plan = models.TextField(blank=True)
created_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='created_improvement_projects')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'improvement_project'
indexes = [
models.Index(fields=['tenant', 'project_type', 'status']),
models.Index(fields=['project_manager', 'sponsor', 'department']),
models.Index(fields=['planned_start_date', 'planned_end_date']),
]
ordering = ['planned_start_date']
def __str__(self):
return f"{self.project_number}: {self.title}"
def save(self, *args, **kwargs):
if not self.project_number:
# Generate project number
from django.utils import timezone
year = timezone.now().year
count = ImprovementProject.objects.filter(
tenant=self.tenant,
created_at__year=year
).count() + 1
self.project_number = f"QI-{year}-{count:04d}"
super().save(*args, **kwargs)
@property
def duration_planned(self):
"""Get planned project duration in days"""
if self.planned_start_date and self.planned_end_date:
return (self.planned_end_date - self.planned_start_date).days
return None
@property
def duration_actual(self):
"""Get actual project duration in days"""
if self.actual_start_date and self.actual_end_date:
return (self.actual_end_date - self.actual_start_date).days
return None