HH/apps/dashboard/services/comments_report_service.py
ismail c5f76b3855
Some checks are pending
Build and Push Docker Image / build (push) Waiting to run
updates
2026-05-11 14:45:30 +03:00

137 lines
4.7 KiB
Python

import calendar
from collections import defaultdict
from django.db.models import Count, Q
from apps.surveys.models import SurveyInstance, SurveyResponse
CATEGORIES = ["Hospital", "Medical", "Non-Medical", "Nursing", "ER", "Support Services"]
SENTIMENTS = ["negative", "positive", "neutral", "gratitude", "suggestion"]
class CommentsReportService:
def __init__(self, hospital_id, year, quarter):
self.hospital_id = hospital_id
self.year = year
self.quarter = quarter
def _base_qs(self):
return SurveyInstance.objects.filter(
survey_template__hospital_id=self.hospital_id,
completed_at__year=self.year,
completed_at__month__in=self._quarter_months(),
).exclude(comment="").exclude(comment__isnull=True)
def _quarter_months(self):
return {
1: [1, 2, 3],
2: [4, 5, 6],
3: [7, 8, 9],
4: [10, 11, 12],
}.get(self.quarter, [1, 2, 3])
def get_available_periods(self):
months = (
SurveyInstance.objects.filter(
survey_template__hospital_id=self.hospital_id,
)
.values_list("completed_at__year", "completed_at__month")
.distinct()
.order_by("-completed_at__year", "-completed_at__month")
)
periods = set()
for y, m in months:
if y is not None and m is not None:
q = (m - 1) // 3 + 1
periods.add((y, q))
return sorted(periods, reverse=True)
def get_summary(self):
qs = self._base_qs()
total = qs.count()
with_analysis = qs.filter(comment_analyzed=True).count()
negative = qs.filter(is_negative=True).count()
return {
"total": total,
"with_analysis": with_analysis,
"negative": negative,
"positive": total - negative,
}
def get_category_breakdown(self):
qs = self._base_qs().filter(comment_analyzed=True)
cat_data = defaultdict(int)
for s in qs.iterator(chunk_size=2000):
analysis = s.comment_analysis or {}
cat = analysis.get("category", "Uncategorized")
cat_data[cat] += 1
if not cat_data and qs.count() == 0:
return []
result = [{"category": c, "count": cat_data.get(c, 0)} for c in CATEGORIES]
return [r for r in result if r["count"] > 0]
def get_sentiment_breakdown(self):
qs = self._base_qs().filter(comment_analyzed=True)
sent_data = defaultdict(int)
for s in qs.iterator(chunk_size=2000):
analysis = s.comment_analysis or {}
sent = analysis.get("sentiment", "neutral")
sent_data[sent] += 1
return [{"sentiment": s, "count": sent_data.get(s, 0)} for s in SENTIMENTS if sent_data.get(s, 0) > 0]
def get_subcategory_breakdown(self):
qs = self._base_qs().filter(comment_analyzed=True)
subcat_data = defaultdict(lambda: defaultdict(int))
for s in qs.iterator(chunk_size=2000):
analysis = s.comment_analysis or {}
cat = analysis.get("category", "Uncategorized")
subcat = analysis.get("subcategory", "General")
subcat_data[cat][subcat] += 1
result = []
for cat, subcats in sorted(subcat_data.items()):
for subcat, count in sorted(subcats.items(), key=lambda x: -x[1]):
result.append({
"category": cat,
"subcategory": subcat,
"count": count,
})
return result
def get_action_plans(self):
qs = self._base_qs().filter(comment_analyzed=True)
plans = []
for s in qs.iterator(chunk_size=2000):
analysis = s.comment_analysis or {}
action = analysis.get("suggested_action_en", "") or analysis.get("suggested_action_ar", "")
if action:
dept = analysis.get("subcategory", "")
plans.append({
"comment": s.comment[:200] if s.comment else "",
"action_plan": action,
"responsible": dept,
"category": analysis.get("category", ""),
"status": "Pending",
})
return plans[:100]
def get_chart_data(self):
by_category = self.get_category_breakdown()
by_sentiment = self.get_sentiment_breakdown()
return {
"by_category": [
{"x": c["category"], "y": c["count"]}
for c in by_category
],
"by_sentiment": [
{"x": s["sentiment"].title(), "y": s["count"]}
for s in by_sentiment
],
}