144 lines
5.2 KiB
Python
144 lines
5.2 KiB
Python
import logging
|
|
|
|
from celery import shared_task
|
|
from django.utils import timezone
|
|
|
|
from apps.organizations.models import Hospital
|
|
|
|
from .kpi_models import KPIReportType
|
|
from .kpi_service import KPICalculationService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@shared_task(bind=True, ignore_result=True)
|
|
def calculate_daily_kpis(self):
|
|
today = timezone.now()
|
|
year = today.year
|
|
month = today.month
|
|
|
|
hospitals = Hospital.objects.filter(status="active")
|
|
report_types = [rt[0] for rt in KPIReportType.choices]
|
|
|
|
for hospital in hospitals:
|
|
for report_type in report_types:
|
|
try:
|
|
KPICalculationService.generate_monthly_report(
|
|
report_type=report_type,
|
|
hospital=hospital,
|
|
year=year,
|
|
month=month,
|
|
)
|
|
logger.info(f"Daily KPI calculated: {report_type} for {hospital.name} ({year}-{month:02d})")
|
|
except Exception as e:
|
|
logger.exception(f"Failed daily KPI: {report_type} for {hospital.name}: {e}")
|
|
|
|
|
|
# =============================================================================
|
|
# AI-Powered Analytics — Async Celery Tasks
|
|
# =============================================================================
|
|
|
|
|
|
@shared_task(bind=True, ignore_result=True, max_retries=2, default_retry_delay=60)
|
|
def generate_executive_summary_task(
|
|
self, user_id=None, hospital_id=None, department_id=None, period="30d", force_refresh=False
|
|
):
|
|
"""
|
|
Async task: Generate AI executive summary and cache it.
|
|
Called by Celery beat daily, or manually when user forces refresh.
|
|
"""
|
|
from apps.analytics.services.ai_analytics import ExecutiveSummaryGenerator
|
|
|
|
# Build a mock user-like object for role scoping (admin sees all)
|
|
class _User:
|
|
def __init__(self, uid):
|
|
self.id = uid
|
|
self.hospital = None
|
|
self.department = None
|
|
|
|
def is_px_admin(self):
|
|
return True
|
|
|
|
user = _User(user_id) if user_id else _User("system")
|
|
|
|
try:
|
|
result = ExecutiveSummaryGenerator.generate(
|
|
user,
|
|
hospital_id=hospital_id,
|
|
department_id=department_id,
|
|
period=period,
|
|
force_refresh=force_refresh,
|
|
)
|
|
logger.info(
|
|
f"Executive summary generated: {period} for hospital={hospital_id}, "
|
|
f"risk={result.get('risk_level', 'unknown')}"
|
|
)
|
|
except Exception as e:
|
|
logger.exception(f"Failed to generate executive summary: {e}")
|
|
self.retry(exc=e)
|
|
|
|
|
|
@shared_task(bind=True, ignore_result=True, max_retries=2, default_retry_delay=60)
|
|
def generate_action_recommendations_task(self, user_id=None, hospital_id=None, department_id=None):
|
|
"""
|
|
Async task: Generate AI action recommendations from complaint clusters.
|
|
"""
|
|
from apps.analytics.services.ai_analytics import ActionRecommendationEngine
|
|
|
|
class _User:
|
|
def __init__(self, uid):
|
|
self.id = uid
|
|
self.hospital = None
|
|
self.department = None
|
|
|
|
def is_px_admin(self):
|
|
return True
|
|
|
|
user = _User(user_id) if user_id else _User("system")
|
|
|
|
try:
|
|
results = ActionRecommendationEngine.generate_recommendations(
|
|
user, hospital_id=hospital_id, department_id=department_id, limit=5
|
|
)
|
|
logger.info(f"Action recommendations generated: {len(results)} recommendations")
|
|
except Exception as e:
|
|
logger.exception(f"Failed to generate action recommendations: {e}")
|
|
self.retry(exc=e)
|
|
|
|
|
|
@shared_task(bind=True, ignore_result=True)
|
|
def precompute_dashboard_cache_task(self):
|
|
"""
|
|
Async task: Pre-compute all cacheable analytics data for all active hospitals.
|
|
Run every 5 minutes so the dashboard is always fast.
|
|
"""
|
|
from apps.analytics.services.ai_analytics import (
|
|
ExecutiveSummaryGenerator,
|
|
EarlyWarningSystem,
|
|
ComplaintVolumeForecaster,
|
|
SLABreachPredictor,
|
|
ActionRecommendationEngine,
|
|
)
|
|
from django.contrib.auth import get_user_model
|
|
|
|
User = get_user_model()
|
|
admin_users = User.objects.filter(is_active=True, role="px_admin")
|
|
|
|
if not admin_users.exists():
|
|
# Fallback: use first superuser
|
|
admin_users = User.objects.filter(is_active=True, is_superuser=True)
|
|
|
|
hospitals = list(Hospital.objects.filter(status="active"))
|
|
|
|
for admin in admin_users[:3]: # limit to first 3 admins
|
|
for hospital in hospitals:
|
|
try:
|
|
ExecutiveSummaryGenerator.generate(admin, hospital_id=str(hospital.id), period="30d")
|
|
EarlyWarningSystem.detect(admin, hospital_id=str(hospital.id), limit=5)
|
|
ComplaintVolumeForecaster.forecast(admin, hospital_id=str(hospital.id), forecast_days=30)
|
|
SLABreachPredictor.predict(admin, hospital_id=str(hospital.id), limit=10)
|
|
ActionRecommendationEngine.generate_recommendations(admin, hospital_id=str(hospital.id), limit=5)
|
|
logger.info(f"Precomputed analytics for admin={admin.id}, hospital={hospital.id}")
|
|
except Exception as e:
|
|
logger.exception(f"Failed to precompute for admin={admin.id}, hospital={hospital.id}: {e}")
|