153 lines
5.8 KiB
Python
153 lines
5.8 KiB
Python
"""
|
||
Management command to manually trigger PX digest email.
|
||
Usage: python manage.py send_px_digest [--period weekly|monthly] [--hospital HOSPITAL_ID]
|
||
"""
|
||
from django.core.management.base import BaseCommand
|
||
from django.utils import timezone
|
||
from datetime import timedelta
|
||
from django.template.loader import render_to_string
|
||
from django.core.mail import EmailMultiAlternatives
|
||
from django.utils.html import strip_tags
|
||
from django.conf import settings
|
||
|
||
from apps.organizations.models import Hospital
|
||
|
||
|
||
class Command(BaseCommand):
|
||
help = "Send PX digest email to admins (weekly or monthly)"
|
||
|
||
def add_arguments(self, parser):
|
||
parser.add_argument(
|
||
"--period",
|
||
type=str,
|
||
choices=["weekly", "monthly"],
|
||
default="weekly",
|
||
help="Digest period (default: weekly)",
|
||
)
|
||
parser.add_argument(
|
||
"--hospital",
|
||
type=str,
|
||
default=None,
|
||
help="Specific hospital ID to send digest for",
|
||
)
|
||
parser.add_argument(
|
||
"--dry-run",
|
||
action="store_true",
|
||
help="Generate digest but don't send emails — just print to stdout",
|
||
)
|
||
|
||
def handle(self, *args, **options):
|
||
from django.contrib.auth import get_user_model
|
||
from apps.analytics.services.ai_analytics import (
|
||
ExecutiveSummaryGenerator,
|
||
EarlyWarningSystem,
|
||
ActionRecommendationEngine,
|
||
)
|
||
|
||
User = get_user_model()
|
||
|
||
period = options["period"]
|
||
hospital_id = options["hospital"]
|
||
dry_run = options["dry_run"]
|
||
days = 7 if period == "weekly" else 30
|
||
|
||
now = timezone.now()
|
||
start_date = now - timedelta(days=days)
|
||
period_label = f"{start_date.strftime('%b %d')} – {now.strftime('%b %d, %Y')}"
|
||
|
||
if hospital_id:
|
||
hospitals = list(Hospital.objects.filter(id=hospital_id, status="active"))
|
||
else:
|
||
hospitals = list(Hospital.objects.filter(status="active"))
|
||
|
||
self.stdout.write(f"Sending {period} digest for {len(hospitals)} hospital(s) (dry_run={dry_run})")
|
||
|
||
for hospital in hospitals:
|
||
self.stdout.write(f"\n{'='*60}")
|
||
self.stdout.write(f" Hospital: {hospital.name}")
|
||
self.stdout.write(f"{'='*60}")
|
||
|
||
admins = list(
|
||
User.objects.filter(is_active=True, hospital=hospital, role="px_admin")
|
||
| User.objects.filter(is_active=True, is_superuser=True)
|
||
)
|
||
admins = list(set(admins))
|
||
admin_emails = [a.email for a in admins if a.email]
|
||
|
||
self.stdout.write(f" Recipients: {admin_emails}")
|
||
|
||
class _MockUser:
|
||
def __init__(self, u):
|
||
self.id = u.id
|
||
self.hospital = hospital
|
||
self.department = None
|
||
def is_px_admin(self):
|
||
return True
|
||
|
||
mock_admin = _MockUser(admins[0]) if admins else None
|
||
if not mock_admin:
|
||
self.stdout.write(self.style.WARNING(" No admins found, skipping"))
|
||
continue
|
||
|
||
# Gather metrics
|
||
from apps.analytics.tasks_digest import _gather_metrics
|
||
metrics = _gather_metrics(hospital, start_date, now)
|
||
|
||
# Generate AI summary
|
||
summary = ExecutiveSummaryGenerator.generate(
|
||
mock_admin, hospital_id=str(hospital.id), period=f"{days}d"
|
||
)
|
||
|
||
# Early warnings
|
||
early_warnings = EarlyWarningSystem.detect(
|
||
mock_admin, hospital_id=str(hospital.id), limit=5
|
||
)
|
||
|
||
# Recommendations
|
||
recommendations = ActionRecommendationEngine.generate_recommendations(
|
||
mock_admin, hospital_id=str(hospital.id), limit=3
|
||
)
|
||
|
||
self.stdout.write(f" AI Risk Level: {summary.get('risk_level', 'unknown')}")
|
||
self.stdout.write(f" Early Warnings: {len(early_warnings)} departments")
|
||
self.stdout.write(f" Recommendations: {len(recommendations)}")
|
||
|
||
# Render email
|
||
subject = f"PX360 {period.title()} Digest — {hospital.name} ({period_label})"
|
||
context = {
|
||
"period": period,
|
||
"period_label": period_label,
|
||
"hospital_name": hospital.name,
|
||
"summary": summary,
|
||
"metrics": metrics,
|
||
"early_warnings": early_warnings,
|
||
"recommendations": recommendations,
|
||
"dashboard_url": f"{getattr(settings, 'SITE_URL', '')}/analytics/dashboard/",
|
||
"command_center_url": f"{getattr(settings, 'SITE_URL', '')}/analytics/command-center/",
|
||
}
|
||
|
||
html_content = render_to_string("emails/px_digest_weekly.html", context)
|
||
text_content = strip_tags(html_content)
|
||
|
||
if dry_run:
|
||
# Print email content
|
||
self.stdout.write(f"\n{'='*60}")
|
||
self.stdout.write(f" Subject: {subject}")
|
||
self.stdout.write(f" To: {admin_emails}")
|
||
self.stdout.write(f"{'='*60}")
|
||
self.stdout.write(text_content[:500] + "...")
|
||
else:
|
||
# Send email
|
||
email = EmailMultiAlternatives(
|
||
subject=subject,
|
||
body=text_content,
|
||
from_email=getattr(settings, "DEFAULT_FROM_EMAIL", "noreply@px360.sa"),
|
||
to=admin_emails,
|
||
)
|
||
email.attach_alternative(html_content, "text/html")
|
||
email.send()
|
||
self.stdout.write(self.style.SUCCESS(f" ✓ Sent to {len(admin_emails)} recipients"))
|
||
|
||
self.stdout.write(f"\n{'='*60}")
|
||
self.stdout.write(self.style.SUCCESS(f"Digest complete: {len(hospitals)} hospital(s) processed"))
|