"""Reports domain for M&EL (FRMFL032–033). Consolidates data from indicators, tracking, and upstream apps (finance, grants, REP, repository) into versioned report artefacts distributed to stakeholders. """ from __future__ import annotations import hashlib from django.conf import settings from django.db import models from simple_history.models import HistoricalRecords from apps.core.storage.paths import upload_to_mel_reports # --------------------------------------------------------------------------- # Template # --------------------------------------------------------------------------- class ReportType(models.TextChoices): QUARTERLY = "quarterly", "Quarterly" ANNUAL = "annual", "Annual" PROJECT = "project", "Project" THEMATIC = "thematic", "Thematic" DONOR = "donor", "Donor" AD_HOC = "ad_hoc", "Ad hoc" class ReportTemplate(models.Model): """Blueprint describing which sections a Report renders and for which scope.""" name = models.CharField(max_length=255, unique=True) slug = models.SlugField(max_length=120, unique=True) description = models.TextField(blank=True) report_type = models.CharField( max_length=16, choices=ReportType.choices, default=ReportType.QUARTERLY, ) logframe = models.ForeignKey( "mel_indicators.LogFrame", on_delete=models.PROTECT, null=True, blank=True, related_name="report_templates", help_text="Scope — when set, reports consolidate this logframe only.", ) sections = models.JSONField( default=list, blank=True, help_text=( "Ordered list of section keys: indicator_rollup, activity_summary, " "variance_table, finance, compliance, narrative." ), ) schedule_cron = models.CharField( max_length=100, blank=True, help_text=( "Optional cron expression for scheduled generation — parsed by the " "generate_scheduled_reports Celery task." ), ) default_recipients = models.ManyToManyField( settings.AUTH_USER_MODEL, blank=True, related_name="mel_report_templates_subscribed", ) is_active = models.BooleanField(default=True, db_index=True) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_report_templates_created", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) history = HistoricalRecords() class Meta: ordering = ["name"] def __str__(self): return self.name # --------------------------------------------------------------------------- # Report # --------------------------------------------------------------------------- class ReportStatus(models.TextChoices): DRAFT = "draft", "Draft" IN_REVIEW = "in_review", "In review" APPROVED = "approved", "Approved" PUBLISHED = "published", "Published" ARCHIVED = "archived", "Archived" REPORT_STATUS_TRANSITIONS: dict[str, set[str]] = { ReportStatus.DRAFT: {ReportStatus.IN_REVIEW, ReportStatus.ARCHIVED}, ReportStatus.IN_REVIEW: {ReportStatus.DRAFT, ReportStatus.APPROVED, ReportStatus.ARCHIVED}, ReportStatus.APPROVED: {ReportStatus.PUBLISHED, ReportStatus.IN_REVIEW, ReportStatus.ARCHIVED}, # PUBLISHED → APPROVED is reserved for the explicit `unpublish_report` # service which records an audit-log reason. ARCHIVED is the only other # legal exit. The pre-save guard in signals.py blocks all other edits. ReportStatus.PUBLISHED: {ReportStatus.ARCHIVED, ReportStatus.APPROVED}, ReportStatus.ARCHIVED: set(), } class Report(models.Model): """A concrete generated report. ``consolidated_data`` captures the snapshot.""" template = models.ForeignKey( ReportTemplate, on_delete=models.PROTECT, related_name="reports", ) period_label = models.CharField(max_length=40, help_text="e.g. '2026-Q2'.") period_start = models.DateField() period_end = models.DateField() status = models.CharField( max_length=16, choices=ReportStatus.choices, default=ReportStatus.DRAFT, db_index=True, ) generated_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_reports_generated", ) consolidated_data = models.JSONField( default=dict, help_text="Snapshot of builder outputs at generation time.", ) narrative = models.TextField( blank=True, help_text="Human-written summary appended to the structured data.", ) generated_at = models.DateTimeField(auto_now_add=True) approved_at = models.DateTimeField(null=True, blank=True) published_at = models.DateTimeField(null=True, blank=True) share_token = models.CharField( max_length=64, blank=True, db_index=True, help_text="Public share token; generated on publish for donor/external read-only access.", ) history = HistoricalRecords() @staticmethod def generate_share_token() -> str: import secrets return secrets.token_urlsafe(32) class Meta: ordering = ["-generated_at"] indexes = [ models.Index(fields=["template", "period_label"]), models.Index(fields=["status", "generated_at"]), ] constraints = [ models.UniqueConstraint( fields=["template", "period_label"], name="uniq_report_template_period", ), models.CheckConstraint( check=models.Q(period_end__gte=models.F("period_start")), name="report_period_end_after_start", ), ] def __str__(self): return f"{self.template.name} — {self.period_label}" # --------------------------------------------------------------------------- # Artifact (rendered file) # --------------------------------------------------------------------------- class ArtifactFormat(models.TextChoices): PDF = "pdf", "PDF" XLSX = "xlsx", "Excel" WORD = "word", "Word" CSV = "csv", "CSV" HTML = "html", "HTML" class ReportArtifact(models.Model): report = models.ForeignKey( Report, on_delete=models.CASCADE, related_name="artifacts", ) format = models.CharField(max_length=8, choices=ArtifactFormat.choices) file = models.FileField(upload_to=upload_to_mel_reports, max_length=500) checksum = models.CharField( max_length=64, help_text="SHA-256 of the artefact bytes for tamper evidence.", ) size_bytes = models.PositiveIntegerField(default=0) generated_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ["-generated_at"] indexes = [ models.Index(fields=["report", "format"]), ] constraints = [ models.UniqueConstraint( fields=["report", "format"], name="uniq_report_artifact_format", ), ] def __str__(self): return f"{self.report} [{self.format}]" @staticmethod def checksum_for(data: bytes) -> str: return hashlib.sha256(data).hexdigest() # --------------------------------------------------------------------------- # Distribution # --------------------------------------------------------------------------- class DistributionChannel(models.TextChoices): EMAIL = "email", "Email" DOWNLOAD = "download", "Download link" PORTAL = "portal", "Portal notification" WEBHOOK = "webhook", "Webhook" # G18 class DistributionStatus(models.TextChoices): PENDING = "pending", "Pending" SENT = "sent", "Sent" FAILED = "failed", "Failed" class ReportDistribution(models.Model): report = models.ForeignKey( Report, on_delete=models.CASCADE, related_name="distributions", ) recipient_user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_report_distributions", ) recipient_email = models.EmailField( blank=True, help_text="Used when the recipient is not a registered user.", ) channel = models.CharField( max_length=16, choices=DistributionChannel.choices, default=DistributionChannel.EMAIL, ) status = models.CharField( max_length=16, choices=DistributionStatus.choices, default=DistributionStatus.PENDING, db_index=True, ) sent_at = models.DateTimeField(null=True, blank=True) error = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["report", "status"]), ] def __str__(self): who = self.recipient_user or self.recipient_email return f"{self.report} → {who} [{self.status}]" # --------------------------------------------------------------------------- # Compliance alerts # --------------------------------------------------------------------------- class ComplianceSeverity(models.TextChoices): LOW = "low", "Low" MEDIUM = "medium", "Medium" HIGH = "high", "High" CRITICAL = "critical", "Critical" class ComplianceRequirementType(models.TextChoices): """M&E SRS Table 66 — what obligation a compliance issue is measured against.""" DONOR_AGREEMENT = "donor_agreement", "Donor agreement" REGULATION = "regulation", "Government regulation" CONTRACT = "contract", "Contractual obligation" INTERNAL_POLICY = "internal_policy", "Internal policy / procedure" OTHER = "other", "Other" class ComplianceAlert(models.Model): """Flagged deviations (regulatory, donor, internal) surfaced by reports.""" report = models.ForeignKey( Report, on_delete=models.CASCADE, null=True, blank=True, related_name="compliance_alerts", ) indicator = models.ForeignKey( "mel_indicators.Indicator", on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_compliance_alerts", ) code = models.CharField(max_length=80, help_text="Machine-readable alert code.") severity = models.CharField( max_length=16, choices=ComplianceSeverity.choices, default=ComplianceSeverity.MEDIUM, db_index=True, ) message = models.TextField() # M&E SRS Table 66 — compliance review covers donor agreements, # regulations, contracts and internal procedures; manual reviews classify # the obligation the issue breaches. Blank on legacy/automated rows. requirement_type = models.CharField( max_length=24, choices=ComplianceRequirementType.choices, blank=True, db_index=True, ) raised_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_compliance_alerts_raised", help_text="Set when a Compliance Officer records the issue manually.", ) resolved = models.BooleanField(default=False, db_index=True) resolved_at = models.DateTimeField(null=True, blank=True) resolved_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_compliance_alerts_resolved", ) # M&E SRS Table 66 — the audit trail must not be modified: the resolution # reason lives in its own field instead of being appended to ``message``. resolution_note = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) history = HistoricalRecords() class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["severity", "resolved"]), models.Index(fields=["report"]), ] def __str__(self): return f"[{self.severity}] {self.code}" # --------------------------------------------------------------------------- # Phase 4.4 — Section schema + scheduling # --------------------------------------------------------------------------- class ReportSectionKind(models.TextChoices): """Section kinds the drag-drop builder can add. MEI015 — every value here MUST be a key that :func:`apps.mel.reports.builders.build_consolidated_data` knows how to build and that ``_report_body.html`` / ``_report_body_print.html`` know how to render, so the arrangement saved in the builder drives real report output. ``narrative`` is the sole exception: it surfaces the report's executive-summary prose (``Report.narrative``) at the top of the body and is deliberately excluded from the generated ``section_order``. """ NARRATIVE = "narrative", "Narrative" INDICATOR_ROLLUP = "indicator_rollup", "Indicator rollup" ACTIVITY_SUMMARY = "activity_summary", "Activity summary" VARIANCE_TABLE = "variance_table", "Variance table" OUTPUT_SNAPSHOT = "output_snapshot", "Output snapshot" OUTCOME_ASSESSMENT = "outcome_assessment", "Outcome assessment" IMPACT_EVALUATION = "impact_evaluation", "Impact evaluation" FINANCE = "finance", "Finance" BUDGET_VARIANCE = "budget_variance", "Budget vs expenditure" RIMS_FUNDING_SUMMARY = "rims_funding_summary", "RIMS funding summary" REP_COMPLETIONS_SUMMARY = "rep_completions_summary", "REP course completions" MOODLE_COMPLETIONS_SUMMARY = "moodle_completions_summary", "Moodle course completions" REPOSITORY_OUTPUTS_SUMMARY = "repository_outputs_summary", "Repository outputs" COMPLIANCE = "compliance", "Compliance alerts" class ReportSection(models.Model): template = models.ForeignKey( ReportTemplate, on_delete=models.CASCADE, related_name="section_rows", ) ordinal = models.PositiveIntegerField(default=0) kind = models.CharField(max_length=32, choices=ReportSectionKind.choices) title = models.CharField(max_length=255, blank=True) config = models.JSONField( default=dict, blank=True, help_text="Section-specific config: {'indicator_codes':[…], 'window':'last_quarter', …}.", ) htmx_lazy = models.BooleanField( default=False, help_text="When true, the section is loaded via an HTMX endpoint after the page renders.", ) class Meta: ordering = ["template_id", "ordinal", "pk"] indexes = [ models.Index(fields=["template", "ordinal"]), ] def __str__(self): return f"{self.template_id}/{self.ordinal} · {self.kind}" class ScheduleFrequency(models.TextChoices): DAILY = "daily", "Daily" WEEKLY = "weekly", "Weekly" MONTHLY = "monthly", "Monthly" QUARTERLY = "quarterly", "Quarterly" ANNUAL = "annual", "Annual" class ScheduledReport(models.Model): """Persistent schedule for a ReportTemplate (Phase 4.4). Each fire creates a Report + artefacts and ticks ``next_run_at`` forward. """ template = models.ForeignKey( ReportTemplate, on_delete=models.CASCADE, related_name="schedules", ) frequency = models.CharField( max_length=16, choices=ScheduleFrequency.choices, default=ScheduleFrequency.QUARTERLY, ) cron_expression = models.CharField( max_length=120, blank=True, help_text="Optional override — when set, takes precedence over ``frequency``.", ) formats = models.JSONField( default=list, blank=True, help_text="List of artefact formats to render: pdf / xlsx / html.", ) recipients = models.ManyToManyField( settings.AUTH_USER_MODEL, blank=True, related_name="mel_report_schedules", ) extra_emails = models.JSONField( default=list, blank=True, help_text="Free-text additional recipient emails.", ) is_active = models.BooleanField(default=True, db_index=True) last_run_at = models.DateTimeField(null=True, blank=True) next_run_at = models.DateTimeField(null=True, blank=True, db_index=True) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_report_schedules_created", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["next_run_at", "-created_at"] indexes = [ models.Index(fields=["is_active", "next_run_at"]), ] def __str__(self): return f"Schedule<{self.template_id} {self.frequency}>" # --------------------------------------------------------------------------- # G18 — Webhook distribution # --------------------------------------------------------------------------- def _default_webhook_secret() -> str: """Auto-generate a 32-byte URL-safe secret for new webhooks.""" import secrets as _secrets return _secrets.token_urlsafe(32) class ReportDistributionWebhook(models.Model): """G18 — POST-target endpoint that receives published-report payloads. When ``template`` is null, the webhook fires for every published report (global). When set, it only fires for that template. Each delivery is HMAC-SHA256-signed with ``secret`` so receivers can verify integrity. """ template = models.ForeignKey( ReportTemplate, on_delete=models.CASCADE, related_name="webhooks", null=True, blank=True, help_text="Bind to a template, or leave blank to fire on every published report.", ) name = models.CharField(max_length=120) url = models.URLField() secret = models.CharField( max_length=64, default=_default_webhook_secret, help_text="HMAC-SHA256 signing key; included as the X-IILMP-Signature header.", ) is_active = models.BooleanField(default=True, db_index=True) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_report_webhooks_created", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) history = HistoricalRecords() class Meta: ordering = ["-created_at"] def __str__(self): return f"{self.name} → {self.url}" class WebhookDeliveryStatus(models.TextChoices): PENDING = "pending", "Pending" SUCCESS = "success", "Success" FAILED = "failed", "Failed" class ReportWebhookDelivery(models.Model): """Per-(webhook, report) delivery row — idempotent on unique_together.""" webhook = models.ForeignKey( ReportDistributionWebhook, on_delete=models.CASCADE, related_name="deliveries", ) report = models.ForeignKey( Report, on_delete=models.CASCADE, related_name="webhook_deliveries", ) status = models.CharField( max_length=16, choices=WebhookDeliveryStatus.choices, default=WebhookDeliveryStatus.PENDING, db_index=True, ) attempts = models.PositiveSmallIntegerField(default=0) last_response_status = models.PositiveSmallIntegerField(null=True, blank=True) last_response_excerpt = models.CharField(max_length=2000, blank=True) last_attempted_at = models.DateTimeField(null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True) history = HistoricalRecords() class Meta: ordering = ["-created_at"] constraints = [ models.UniqueConstraint( fields=["webhook", "report"], name="uniq_webhook_delivery", ), ] def __str__(self): return f"{self.webhook_id}→{self.report_id} [{self.status}]"