520 lines
22 KiB
Python
520 lines
22 KiB
Python
"""
|
|
Seed KPI Report Data
|
|
|
|
Creates all prerequisite data (hospitals, departments, sources, complaints,
|
|
surveys) and generates all 7 KPI report types across 12 months
|
|
so that /analytics/kpi-reports/ renders with rich data.
|
|
|
|
AI analysis is skipped for speed. Works offline.
|
|
|
|
Usage:
|
|
python manage.py seed_kpi_data
|
|
python manage.py seed_kpi_data --year 2025
|
|
python manage.py seed_kpi_data --skip-reports
|
|
"""
|
|
|
|
import logging
|
|
import random
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
from django.contrib.auth.models import Group
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from apps.accounts.models import User
|
|
from apps.analytics.kpi_models import KPIReport, KPIReportType
|
|
from apps.analytics.kpi_service import KPICalculationService
|
|
from apps.complaints.models import Complaint, ComplaintStatus, ComplaintUpdate
|
|
from apps.dashboard.models import ComplaintRequest
|
|
from apps.organizations.models import Department, Hospital, Location, MainSection
|
|
from apps.organizations.models import Patient
|
|
from apps.px_sources.models import PXSource
|
|
from apps.surveys.models import (
|
|
SurveyInstance,
|
|
SurveyQuestion,
|
|
SurveyResponse,
|
|
SurveyStatus,
|
|
SurveyTemplate,
|
|
)
|
|
|
|
for _ln in [
|
|
"apps.notifications",
|
|
"apps.complaints.signals",
|
|
"apps.analytics.kpi_service",
|
|
"apps.core.ai_service",
|
|
"apps.surveys.signals",
|
|
"mshastra",
|
|
"urllib3",
|
|
"openai",
|
|
"httpx",
|
|
"httpcore",
|
|
"django.request",
|
|
"django.db.backends",
|
|
]:
|
|
logging.getLogger(_ln).setLevel(logging.CRITICAL)
|
|
logging.disable(logging.CRITICAL)
|
|
DEPARTMENTS = [
|
|
{"name": "Surgery", "code": "SURG"},
|
|
{"name": "Cardiology", "code": "CARD"},
|
|
{"name": "Orthopedics", "code": "ORTH"},
|
|
{"name": "Pediatrics", "code": "PEDS"},
|
|
{"name": "Emergency", "code": "ER"},
|
|
{"name": "Laboratory", "code": "LAB"},
|
|
{"name": "ICU", "code": "ICU"},
|
|
{"name": "Pharmacy", "code": "PHARM"},
|
|
{"name": "Nursing", "code": "NURS"},
|
|
{"name": "Administration", "code": "ADMIN"},
|
|
{"name": "Reception", "code": "RECEP"},
|
|
{"name": "Medical Reports", "code": "MEDREP"},
|
|
{"name": "Housekeeping", "code": "HOUSE"},
|
|
{"name": "Maintenance", "code": "MAINT"},
|
|
{"name": "Security", "code": "SEC"},
|
|
]
|
|
LOCATIONS = [
|
|
{"id": 48, "name_en": "Inpatient", "name_ar": "التنويم"},
|
|
{"id": 49, "name_en": "Outpatient Clinics", "name_ar": "العيادات الخارجية"},
|
|
{"id": 82, "name_en": "Emergency", "name_ar": "الطوارئ"},
|
|
]
|
|
MAIN_SECTIONS = [
|
|
{"id": 101, "name_en": "Inpatient Services", "name_ar": "خدمات التنويم"},
|
|
{"id": 102, "name_en": "Outpatient Clinics", "name_ar": "العيادات الخارجية"},
|
|
{"id": 103, "name_en": "Emergency Department", "name_ar": "قسم الطوارئ"},
|
|
]
|
|
SOURCES = [
|
|
{"code": "PATIENT", "name_en": "Patient", "name_ar": "مريض"},
|
|
{"code": "FAMILY", "name_en": "Family Member", "name_ar": "عضو العائلة"},
|
|
{"code": "STAFF", "name_en": "Staff", "name_ar": "موظف"},
|
|
{"code": "CALL-CENTER", "name_en": "Call Center", "name_ar": "مركز الاتصال"},
|
|
{"code": "MOH", "name_en": "Ministry of Health", "name_ar": "وزارة الصحة"},
|
|
{"code": "CHI", "name_en": "Council of Health Insurance", "name_ar": "مجلس الضمان الصحي"},
|
|
{"code": "SOCIAL-MEDIA", "name_en": "Social Media", "name_ar": "وسائل التواصل الاجتماعي"},
|
|
{"code": "SURVEY", "name_en": "Survey", "name_ar": "استبيان"},
|
|
]
|
|
COMPLAINT_TITLES = [
|
|
"Long wait time for consultation",
|
|
"Rude staff behavior at reception",
|
|
"Incorrect billing charges",
|
|
"Delayed medication administration",
|
|
"Room cleanliness issues",
|
|
"Poor communication from doctor",
|
|
"Difficulty scheduling appointment",
|
|
"Food quality below standard",
|
|
"Noise levels in ward disturbing rest",
|
|
"Incomplete discharge instructions",
|
|
"Pain not managed adequately",
|
|
"Lab results delayed",
|
|
"Parking facility inadequate",
|
|
"Air conditioning malfunction",
|
|
"Disrespectful treatment by nurse",
|
|
"Wrong medication dosage given",
|
|
"Inadequate wheelchair access",
|
|
"Long queue at pharmacy",
|
|
"Missing personal belongings",
|
|
"Unprofessional phone etiquette",
|
|
"Slow emergency response",
|
|
"Lack of privacy during examination",
|
|
"Overcharged for procedures",
|
|
"Unclear insurance coverage explanation",
|
|
"Delay in surgical scheduling",
|
|
"Insufficient visitor seating",
|
|
"Poor hygiene in restroom",
|
|
"Unhelpful customer service",
|
|
"Equipment malfunction during procedure",
|
|
"Inadequate pain management post-surgery",
|
|
]
|
|
COMPLAINT_DESCRIPTIONS = [
|
|
"Patient reported experiencing significant delays and poor communication throughout the visit.",
|
|
"The complainant described multiple service failures affecting their overall experience.",
|
|
"A detailed account of systematic issues encountered during the hospital stay.",
|
|
"Patient expressed frustration with the lack of timely response to their concerns.",
|
|
"Complaint highlights areas requiring immediate management attention and improvement.",
|
|
]
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Seed KPI data: complaints, surveys, and generate all 7 KPI report types"
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument("--year", type=int, default=2025)
|
|
parser.add_argument("--hospital-code", type=str, default="HH-KPI")
|
|
parser.add_argument("--complaints-per-month", type=int, default=30)
|
|
parser.add_argument("--surveys-per-month", type=int, default=20)
|
|
parser.add_argument("--skip-reports", action="store_true")
|
|
parser.add_argument("--clear", action="store_true")
|
|
parser.add_argument("--force", action="store_true")
|
|
|
|
def handle(self, *args, **options):
|
|
self.year = options["year"]
|
|
self.hospital_code = options["hospital_code"]
|
|
self.complaints_per_month = options["complaints_per_month"]
|
|
self.surveys_per_month = options["surveys_per_month"]
|
|
self.skip_reports = options["skip_reports"]
|
|
self.do_clear = options["clear"]
|
|
self.force = options["force"]
|
|
|
|
self.stdout.write(f"\n{'=' * 60}")
|
|
self.stdout.write("KPI Report Data Seeding")
|
|
self.stdout.write(f"{'=' * 60}\n")
|
|
self.stdout.write(f" Year: {self.year}")
|
|
self.stdout.write(f" Hospital code: {self.hospital_code}")
|
|
self.stdout.write(f" Complaints/month: {self.complaints_per_month}")
|
|
self.stdout.write(f" Surveys/month: {self.surveys_per_month}")
|
|
self.stdout.write(f" Skip reports: {self.skip_reports}")
|
|
self.stdout.write(f" Clear existing: {self.do_clear}")
|
|
self.stdout.write("")
|
|
|
|
KPICalculationService.generate_ai_analysis = classmethod(lambda cls, report: {"note": "skipped during seed"})
|
|
self.stdout.write(" AI analysis: DISABLED\n")
|
|
|
|
if self.do_clear:
|
|
self._clear_existing()
|
|
|
|
hospital = self._seed_infrastructure()
|
|
admin_user, staff_users = self._seed_users(hospital)
|
|
patients = self._seed_patients(hospital)
|
|
sources = self._seed_sources()
|
|
survey_templates = self._seed_survey_templates(hospital)
|
|
|
|
for month in range(1, 13):
|
|
self.stdout.write(f"\n--- Month {month:02d} ---")
|
|
with transaction.atomic():
|
|
complaints = self._seed_complaints(hospital, admin_user, staff_users, sources, month)
|
|
self._seed_complaint_updates(complaints, staff_users)
|
|
self._seed_complaint_requests(complaints, staff_users)
|
|
self._seed_surveys(hospital, patients, survey_templates, month)
|
|
self.stdout.write(f" {len(complaints)} complaints, {self.surveys_per_month} surveys")
|
|
|
|
if not self.skip_reports:
|
|
self._generate_reports(hospital, admin_user)
|
|
|
|
self.stdout.write(self.style.SUCCESS("\nKPI data seeding completed!\n"))
|
|
|
|
def _clear_existing(self):
|
|
self.stdout.write("Clearing existing data...")
|
|
KPIReport.objects.filter(hospital__code=self.hospital_code).delete()
|
|
Complaint.objects.filter(hospital__code=self.hospital_code).delete()
|
|
SurveyInstance.objects.filter(survey_template__hospital__code=self.hospital_code).delete()
|
|
ComplaintRequest.objects.filter(hospital__code=self.hospital_code).delete()
|
|
self.stdout.write(" Cleared.\n")
|
|
|
|
def _seed_infrastructure(self):
|
|
self.stdout.write("Infrastructure...")
|
|
hospital, created = Hospital.objects.get_or_create(
|
|
code=self.hospital_code,
|
|
defaults={
|
|
"name": "Al-Hammadi Hospital (KPI Test)",
|
|
"name_ar": "مستشفى الحمادي (اختبار مؤشرات الأداء)",
|
|
"city": "Riyadh",
|
|
"status": "active",
|
|
},
|
|
)
|
|
self.stdout.write(f" Hospital: {hospital.name} {'(new)' if created else '(exists)'}")
|
|
|
|
for d in DEPARTMENTS:
|
|
Department.objects.get_or_create(
|
|
hospital=hospital,
|
|
code=d["code"],
|
|
defaults={"name": d["name"], "status": "active"},
|
|
)
|
|
for loc in LOCATIONS:
|
|
Location.objects.get_or_create(
|
|
id=loc["id"],
|
|
defaults={"name_en": loc["name_en"], "name_ar": loc["name_ar"]},
|
|
)
|
|
for ms in MAIN_SECTIONS:
|
|
MainSection.objects.get_or_create(
|
|
id=ms["id"],
|
|
defaults={"name_en": ms["name_en"], "name_ar": ms["name_ar"]},
|
|
)
|
|
|
|
self.stdout.write(
|
|
f" Depts: {Department.objects.filter(hospital=hospital).count()}, "
|
|
f"Locs: {Location.objects.filter(id__in=[l['id'] for l in LOCATIONS]).count()}"
|
|
)
|
|
return hospital
|
|
|
|
def _seed_users(self, hospital):
|
|
admin_group, _ = Group.objects.get_or_create(name="Hospital Admin")
|
|
px_admin_group, _ = Group.objects.get_or_create(name="PX Admin")
|
|
admin_user, created = User.objects.get_or_create(
|
|
email="kpi-admin@hospital.test",
|
|
defaults={
|
|
"first_name": "KPI",
|
|
"last_name": "Admin",
|
|
"hospital": hospital,
|
|
"is_active": True,
|
|
},
|
|
)
|
|
if created:
|
|
admin_user.set_password("testpass123")
|
|
admin_user.save()
|
|
admin_user.groups.add(admin_group, px_admin_group)
|
|
|
|
staff_users = []
|
|
for i in range(5):
|
|
u, c = User.objects.get_or_create(
|
|
email=f"kpi-staff{i + 1}@hospital.test",
|
|
defaults={
|
|
"first_name": "Staff",
|
|
"last_name": f"Member{i + 1}",
|
|
"hospital": hospital,
|
|
"is_active": True,
|
|
},
|
|
)
|
|
if c:
|
|
u.set_password("testpass123")
|
|
u.save()
|
|
staff_users.append(u)
|
|
|
|
self.stdout.write(f" Users: 1 admin + {len(staff_users)} staff")
|
|
return admin_user, staff_users
|
|
|
|
def _seed_patients(self, hospital):
|
|
patients = []
|
|
for i in range(20):
|
|
p, _ = Patient.objects.get_or_create(
|
|
mrn=f"PTN-KPI-{i + 1:04d}",
|
|
defaults={
|
|
"first_name": "Patient",
|
|
"last_name": f"KPI{i + 1}",
|
|
"primary_hospital": hospital,
|
|
"status": "active",
|
|
},
|
|
)
|
|
patients.append(p)
|
|
self.stdout.write(f" Patients: {len(patients)}")
|
|
return patients
|
|
|
|
def _seed_sources(self):
|
|
sources = {}
|
|
for src in SOURCES:
|
|
s, _ = PXSource.objects.get_or_create(
|
|
code=src["code"],
|
|
defaults={"name_en": src["name_en"], "name_ar": src["name_ar"], "is_active": True},
|
|
)
|
|
sources[src["code"]] = s
|
|
self.stdout.write(f" Sources: {len(sources)}")
|
|
return sources
|
|
|
|
def _seed_survey_templates(self, hospital):
|
|
templates = {}
|
|
for survey_type, name in [
|
|
("stage", "Patient Experience Survey"),
|
|
("general", "General Feedback Survey"),
|
|
("complaint_resolution", "Complaint Resolution Survey"),
|
|
]:
|
|
t, created = SurveyTemplate.objects.get_or_create(
|
|
name=name,
|
|
hospital=hospital,
|
|
survey_type=survey_type,
|
|
defaults={
|
|
"scoring_method": "average",
|
|
"negative_threshold": Decimal("3.0"),
|
|
"is_active": True,
|
|
},
|
|
)
|
|
if created:
|
|
SurveyQuestion.objects.create(
|
|
survey_template=t,
|
|
text="How would you rate your overall experience?",
|
|
question_type="rating",
|
|
order=1,
|
|
is_required=True,
|
|
)
|
|
templates[survey_type] = t
|
|
self.stdout.write(f" Survey templates: {len(templates)}")
|
|
return templates
|
|
|
|
def _random_dt(self, month):
|
|
day = random.randint(1, 28)
|
|
hour = random.randint(7, 18)
|
|
minute = random.randint(0, 59)
|
|
return timezone.make_aware(datetime(self.year, month, day, hour, minute))
|
|
|
|
def _seed_complaints(self, hospital, admin_user, staff_users, sources, month):
|
|
complaints = []
|
|
source_list = list(sources.values())
|
|
departments = list(Department.objects.filter(hospital=hospital))
|
|
locations = list(Location.objects.filter(id__in=[l["id"] for l in LOCATIONS]))
|
|
main_sections = list(MainSection.objects.filter(id__in=[m["id"] for m in MAIN_SECTIONS]))
|
|
|
|
for i in range(self.complaints_per_month):
|
|
created_at = self._random_dt(month)
|
|
source = random.choice(source_list)
|
|
department = random.choice(departments) if departments else None
|
|
location = random.choice(locations) if locations else None
|
|
main_section = random.choice(main_sections) if main_sections else None
|
|
roll = random.random()
|
|
if roll < 0.70:
|
|
status = ComplaintStatus.CLOSED
|
|
resolved_at = created_at + timedelta(hours=random.uniform(12, 60))
|
|
assigned_at = created_at + timedelta(hours=random.uniform(0.1, 1.5))
|
|
assigned_to = random.choice(staff_users)
|
|
elif roll < 0.85:
|
|
status = ComplaintStatus.RESOLVED
|
|
resolved_at = created_at + timedelta(hours=random.uniform(24, 71))
|
|
assigned_at = created_at + timedelta(hours=random.uniform(0.1, 1.8))
|
|
assigned_to = random.choice(staff_users)
|
|
elif roll < 0.90:
|
|
status = ComplaintStatus.IN_PROGRESS
|
|
resolved_at = None
|
|
assigned_at = created_at + timedelta(hours=random.uniform(0.2, 2.5))
|
|
assigned_to = random.choice(staff_users)
|
|
elif roll < 0.93:
|
|
status = ComplaintStatus.OPEN
|
|
resolved_at = None
|
|
assigned_at = None
|
|
assigned_to = None
|
|
elif roll < 0.96:
|
|
status = ComplaintStatus.CLOSED
|
|
resolved_at = created_at + timedelta(hours=random.uniform(80, 120))
|
|
assigned_at = created_at + timedelta(hours=random.uniform(3, 8))
|
|
assigned_to = random.choice(staff_users)
|
|
else:
|
|
status = ComplaintStatus.CANCELLED
|
|
resolved_at = None
|
|
assigned_at = None
|
|
assigned_to = None
|
|
complaint = Complaint(
|
|
hospital=hospital,
|
|
department=department,
|
|
location=location,
|
|
main_section=main_section,
|
|
source=source,
|
|
title=f"{random.choice(COMPLAINT_TITLES)} ({month}/{i + 1})",
|
|
description=random.choice(COMPLAINT_DESCRIPTIONS),
|
|
complaint_type="complaint",
|
|
status=status,
|
|
severity=random.choice(["low", "medium", "high", "critical"]),
|
|
priority=random.choice(["low", "medium", "high", "critical"]),
|
|
assigned_to=assigned_to,
|
|
assigned_at=assigned_at,
|
|
resolved_at=resolved_at,
|
|
resolved_by=admin_user if resolved_at else None,
|
|
created_by=admin_user,
|
|
due_at=created_at + timedelta(hours=random.choice([72, 96, 120])),
|
|
contact_name=f"Patient {month}-{i + 1}",
|
|
contact_phone=f"+9665{random.randint(10000000, 99999999)}",
|
|
)
|
|
complaint.save()
|
|
complaints.append(complaint)
|
|
|
|
return complaints
|
|
|
|
def _seed_complaint_updates(self, complaints, staff_users):
|
|
for c in complaints:
|
|
if c.status in (ComplaintStatus.CLOSED, ComplaintStatus.RESOLVED):
|
|
ComplaintUpdate.objects.create(
|
|
complaint=c,
|
|
update_type="communication",
|
|
message="We acknowledge your complaint and are looking into it.",
|
|
created_by=random.choice(staff_users),
|
|
created_at=c.created_at + timedelta(hours=random.uniform(1, 40)),
|
|
)
|
|
if c.resolved_at:
|
|
ComplaintUpdate.objects.create(
|
|
complaint=c,
|
|
update_type="resolution",
|
|
message="Your complaint has been addressed.",
|
|
created_by=random.choice(staff_users),
|
|
created_at=c.resolved_at,
|
|
)
|
|
elif c.status == ComplaintStatus.IN_PROGRESS:
|
|
ComplaintUpdate.objects.create(
|
|
complaint=c,
|
|
update_type="communication",
|
|
message="Your complaint is being investigated.",
|
|
created_by=random.choice(staff_users),
|
|
created_at=c.created_at + timedelta(hours=random.uniform(2, 50)),
|
|
)
|
|
|
|
def _seed_complaint_requests(self, complaints, staff_users):
|
|
for c in complaints:
|
|
if random.random() < 0.15:
|
|
ComplaintRequest.objects.create(
|
|
staff=random.choice(staff_users),
|
|
complaint=c,
|
|
hospital=c.hospital,
|
|
patient_name=c.contact_name,
|
|
request_date=c.created_at.date(),
|
|
filled=c.assigned_to is None,
|
|
not_filled=False,
|
|
on_hold=False,
|
|
from_barcode=random.random() < 0.3,
|
|
filling_time_category=random.choice(
|
|
["same_time", "within_6h", "6_to_24h", "after_1_day", "not_mentioned"]
|
|
),
|
|
)
|
|
|
|
def _seed_surveys(self, hospital, patients, templates, month):
|
|
for i in range(self.surveys_per_month):
|
|
if i < self.surveys_per_month * 0.6:
|
|
survey_type = "stage"
|
|
elif i < self.surveys_per_month * 0.8:
|
|
survey_type = "general"
|
|
else:
|
|
survey_type = "complaint_resolution"
|
|
template = templates[survey_type]
|
|
patient = random.choice(patients)
|
|
completed_at = self._random_dt(month)
|
|
|
|
score_roll = random.random()
|
|
if score_roll < 0.65:
|
|
total_score = Decimal(str(random.choice(["4.00", "4.50", "5.00"])))
|
|
elif score_roll < 0.85:
|
|
total_score = Decimal(str(random.choice(["3.00", "3.50"])))
|
|
else:
|
|
total_score = Decimal(str(random.choice(["1.00", "2.00", "2.50"])))
|
|
is_negative = total_score < template.negative_threshold
|
|
survey = SurveyInstance(
|
|
survey_template=template,
|
|
patient=patient,
|
|
hospital=hospital,
|
|
status=SurveyStatus.COMPLETED,
|
|
completed_at=completed_at,
|
|
sent_at=completed_at - timedelta(hours=random.randint(1, 24)),
|
|
total_score=total_score,
|
|
is_negative=is_negative,
|
|
delivery_channel=random.choice(["sms", "whatsapp", "email"]),
|
|
)
|
|
survey.save()
|
|
for question in template.questions.all():
|
|
if question.question_type in ("rating", "likert"):
|
|
SurveyResponse.objects.create(survey_instance=survey, question=question, numeric_value=total_score)
|
|
|
|
def _generate_reports(self, hospital, admin_user):
|
|
self.stdout.write("\nGenerating KPI reports...")
|
|
report_types = [rt[0] for rt in KPIReportType.choices]
|
|
created = 0
|
|
skipped = 0
|
|
failed = 0
|
|
for month in range(1, 13):
|
|
for report_type in report_types:
|
|
existing = KPIReport.objects.filter(
|
|
report_type=report_type, hospital=hospital, year=self.year, month=month
|
|
).first()
|
|
if existing and existing.status == "completed" and not self.force:
|
|
skipped += 1
|
|
continue
|
|
if existing:
|
|
existing.monthly_data.all().delete()
|
|
existing.source_breakdowns.all().delete()
|
|
existing.department_breakdowns.all().delete()
|
|
existing.location_breakdowns.all().delete()
|
|
existing.delete()
|
|
try:
|
|
with transaction.atomic():
|
|
KPICalculationService.generate_monthly_report(
|
|
report_type=report_type,
|
|
hospital=hospital,
|
|
year=self.year,
|
|
month=month,
|
|
generated_by=admin_user,
|
|
)
|
|
created += 1
|
|
self.stdout.write(f" {report_type} {self.year}-{month:02d}: OK")
|
|
except Exception as e:
|
|
failed += 1
|
|
self.stdout.write(self.style.ERROR(f" {report_type} {self.year}-{month:02d}: FAILED - {e}"))
|
|
self.stdout.write(f"\n Reports: {created} created, {skipped} skipped, {failed} failed")
|