179 lines
6.6 KiB
Python
179 lines
6.6 KiB
Python
import calendar
|
|
from collections import defaultdict
|
|
from datetime import date
|
|
|
|
from django.db.models import Count, Q
|
|
from django.db.models.functions import TruncDate
|
|
|
|
from apps.complaints.models import Inquiry
|
|
|
|
STATUS_MAP = {
|
|
"open": "جديد",
|
|
"in_progress": "تحت الإجراء",
|
|
"resolved": "تم التواصل",
|
|
"closed": "تم التواصل",
|
|
"contacted": "تم التواصل",
|
|
"contacted_no_response": "تم التواصل ولم يتم الرد",
|
|
}
|
|
|
|
STATUS_COLORS = {
|
|
"open": "#94A3B8",
|
|
"in_progress": "#F59E0B",
|
|
"contacted": "#10B981",
|
|
"contacted_no_response": "#EF4444",
|
|
"resolved": "#3B82F6",
|
|
"closed": "#6366F1",
|
|
}
|
|
|
|
SLA_ORDER = ["24_hours", "48_hours", "72_hours", "more_than_72_hours"]
|
|
SLA_LABELS = {
|
|
"24_hours": "24 Hours",
|
|
"48_hours": "48 Hours",
|
|
"72_hours": "72 Hours",
|
|
"more_than_72_hours": "> 72 Hours",
|
|
}
|
|
|
|
|
|
class InquiryReportService:
|
|
def __init__(self, hospital_id, year, month):
|
|
self.hospital_id = hospital_id
|
|
self.year = year
|
|
self.month = month
|
|
|
|
def _base_qs(self, is_outgoing=False):
|
|
return Inquiry.objects.filter(
|
|
hospital_id=self.hospital_id,
|
|
created_at__year=self.year,
|
|
created_at__month=self.month,
|
|
is_outgoing=is_outgoing,
|
|
)
|
|
|
|
def get_summary(self, is_outgoing=False):
|
|
qs = self._base_qs(is_outgoing)
|
|
total = qs.count()
|
|
by_status = dict(
|
|
qs.values_list("status").annotate(count=Count("id")).order_by()
|
|
)
|
|
by_sla = dict(
|
|
qs.values_list("timeline_sla").annotate(count=Count("id")).order_by()
|
|
)
|
|
|
|
first_half = qs.filter(created_at__day__lte=15).count()
|
|
second_half = total - first_half
|
|
|
|
contacted_count = by_status.get("contacted", 0) + by_status.get("resolved", 0) + by_status.get("closed", 0)
|
|
sla_compliance = (contacted_count / total * 100) if total > 0 else 0
|
|
|
|
return {
|
|
"total": total,
|
|
"by_status": by_status,
|
|
"by_sla": by_sla,
|
|
"first_half": first_half,
|
|
"second_half": second_half,
|
|
"sla_compliance": round(sla_compliance, 1),
|
|
}
|
|
|
|
def get_daily_trend(self, is_outgoing=False):
|
|
qs = self._base_qs(is_outgoing)
|
|
days_in_month = calendar.monthrange(self.year, self.month)[1]
|
|
daily_raw = (
|
|
qs.annotate(day=TruncDate("created_at"))
|
|
.values("day")
|
|
.annotate(count=Count("id"))
|
|
.order_by("day")
|
|
)
|
|
daily = {}
|
|
for entry in daily_raw:
|
|
if entry["day"]:
|
|
daily[entry["day"].day] = entry["count"]
|
|
result = []
|
|
for d in range(1, days_in_month + 1):
|
|
result.append({"day": d, "label": str(d), "count": daily.get(d, 0)})
|
|
return result
|
|
|
|
def get_employee_breakdown(self):
|
|
qs = self._base_qs(is_outgoing=False).select_related("assigned_to")
|
|
emp_data = defaultdict(lambda: {"contacted_no_response": 0, "in_progress": 0, "contacted": 0, "open": 0, "total": 0})
|
|
for inq in qs.iterator(chunk_size=2000):
|
|
name = inq.assigned_to.get_full_name() if inq.assigned_to else (inq.created_by.get_full_name() if inq.created_by else "Unassigned")
|
|
status = inq.status
|
|
if status in ("contacted", "resolved", "closed"):
|
|
status_key = "contacted"
|
|
elif status == "contacted_no_response":
|
|
status_key = "contacted_no_response"
|
|
elif status == "in_progress":
|
|
status_key = "in_progress"
|
|
else:
|
|
status_key = "open"
|
|
emp_data[name][status_key] += 1
|
|
emp_data[name]["total"] += 1
|
|
return sorted(
|
|
[{"name": name, **counts} for name, counts in emp_data.items()],
|
|
key=lambda x: x["total"],
|
|
reverse=True,
|
|
)
|
|
|
|
def get_department_breakdown(self):
|
|
qs = self._base_qs(is_outgoing=True).select_related("outgoing_department")
|
|
dept_data = defaultdict(lambda: {"contacted_no_response": 0, "in_progress": 0, "contacted": 0, "open": 0, "total": 0})
|
|
for inq in qs.iterator(chunk_size=2000):
|
|
name = inq.outgoing_department.name if inq.outgoing_department else "Unknown"
|
|
status = inq.status
|
|
if status in ("contacted", "resolved", "closed"):
|
|
status_key = "contacted"
|
|
elif status == "contacted_no_response":
|
|
status_key = "contacted_no_response"
|
|
elif status == "in_progress":
|
|
status_key = "in_progress"
|
|
else:
|
|
status_key = "open"
|
|
dept_data[name][status_key] += 1
|
|
dept_data[name]["total"] += 1
|
|
return sorted(
|
|
[{"name": name, **counts} for name, counts in dept_data.items()],
|
|
key=lambda x: x["total"],
|
|
reverse=True,
|
|
)
|
|
|
|
def get_chart_data(self):
|
|
inc_daily = self.get_daily_trend(is_outgoing=False)
|
|
out_daily = self.get_daily_trend(is_outgoing=True)
|
|
inc_summary = self.get_summary(is_outgoing=False)
|
|
out_summary = self.get_summary(is_outgoing=True)
|
|
|
|
return {
|
|
"incoming_daily": [{"x": d["label"], "y": d["count"]} for d in inc_daily],
|
|
"outgoing_daily": [{"x": d["label"], "y": d["count"]} for d in out_daily],
|
|
"incoming_by_status": [
|
|
{"x": STATUS_MAP.get(s, s), "y": c}
|
|
for s, c in inc_summary["by_status"].items()
|
|
],
|
|
"outgoing_by_status": [
|
|
{"x": STATUS_MAP.get(s, s), "y": c}
|
|
for s, c in out_summary["by_status"].items()
|
|
],
|
|
"incoming_by_sla": [
|
|
{"x": SLA_LABELS.get(s, s or "N/A"), "y": c}
|
|
for s, c in sorted(
|
|
inc_summary["by_sla"].items(),
|
|
key=lambda x: SLA_ORDER.index(x[0]) if x[0] in SLA_ORDER else 99,
|
|
)
|
|
],
|
|
"outgoing_by_sla": [
|
|
{"x": SLA_LABELS.get(s, s or "N/A"), "y": c}
|
|
for s, c in sorted(
|
|
out_summary["by_sla"].items(),
|
|
key=lambda x: SLA_ORDER.index(x[0]) if x[0] in SLA_ORDER else 99,
|
|
)
|
|
],
|
|
}
|
|
|
|
def get_available_months(self):
|
|
months = (
|
|
Inquiry.objects.filter(hospital_id=self.hospital_id)
|
|
.values_list("created_at__year", "created_at__month")
|
|
.distinct()
|
|
.order_by("-created_at__year", "-created_at__month")
|
|
)
|
|
return [{"year": y, "month": m, "label": f"{calendar.month_abbr[m]} {y}"} for y, m in months]
|