"""Tracking domain for M&EL (PRD §5.3). Includes: * :class:`UpstreamEvent` — append-only log of signals consumed from other apps. * :class:`Activity` / :class:`ActivityDependency` — scheduled work items aligned with LogFrame ACTIVITY rows (FRMFL009–013). * :class:`OutputDeliverable` — tangible deliverables tracked against LogFrame OUTPUT rows (FRMFL014–019). * :class:`CorrectiveAction` — remediation tied to any subject via GenericFK (FRMFL031). * :class:`Participant` / :class:`TrackingRecord` / :class:`TrackingMilestone` / :class:`BaselineRecord` / :class:`ImpactScore` — Phase 4.5 cross-module longitudinal tracking spanning RIMS scholars, REP learners, SME-Hub entrepreneurs, and Alumni. """ from __future__ import annotations from django.conf import settings from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ValidationError from django.db import models from simple_history.models import HistoricalRecords class UpstreamEvent(models.Model): """Append-only log of upstream signals consumed by M&EL. Populated by ``apps.mel.tracking.receivers``. Provides a cheap, auditable backstop: even if the receiver cannot find an Indicator to update, the event itself is still recorded for later reconciliation. """ class Module(models.TextChoices): RIMS = "rims", "RIMS" REP = "rep", "REP" REPOSITORY = "repository", "Repository" SMEHUB = "smehub", "SME-Hub" ALUMNI = "alumni", "Alumni" module = models.CharField(max_length=16, choices=Module.choices, db_index=True) event_type = models.CharField(max_length=64, db_index=True) source_object_id = models.CharField(max_length=64, db_index=True) payload = models.JSONField(default=dict, blank=True) subject = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_upstream_events", help_text="Person the event is about, when identifiable.", ) recorded_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ["-recorded_at"] indexes = [ models.Index(fields=["module", "event_type", "recorded_at"]), ] constraints = [ models.UniqueConstraint( fields=["module", "event_type", "source_object_id"], name="uniq_upstream_event", ), ] def __str__(self): return f"{self.module}/{self.event_type} #{self.source_object_id}" # --------------------------------------------------------------------------- # Activities (FRMFL009–013) # --------------------------------------------------------------------------- class ActivityStatus(models.TextChoices): PLANNED = "planned", "Planned" IN_PROGRESS = "in_progress", "In progress" DELAYED = "delayed", "Delayed" # M&E SRS Table 65 — field data collection could not be completed # (logistics / inaccessible locations). A follow-up monitoring activity is # scheduled and official indicator values are NOT updated until completed. INCOMPLETE = "incomplete", "Incomplete" COMPLETED = "completed", "Completed" CANCELLED = "cancelled", "Cancelled" # Allowed forward transitions — enforced by services.record_activity_progress. ACTIVITY_STATUS_TRANSITIONS: dict[str, set[str]] = { ActivityStatus.PLANNED: { ActivityStatus.IN_PROGRESS, ActivityStatus.DELAYED, ActivityStatus.INCOMPLETE, ActivityStatus.CANCELLED, }, ActivityStatus.IN_PROGRESS: { ActivityStatus.DELAYED, ActivityStatus.INCOMPLETE, ActivityStatus.COMPLETED, ActivityStatus.CANCELLED, }, ActivityStatus.DELAYED: { ActivityStatus.IN_PROGRESS, ActivityStatus.INCOMPLETE, ActivityStatus.COMPLETED, ActivityStatus.CANCELLED, }, # M&E SRS Table 65 — an incomplete collection is resumed or abandoned; a # follow-up monitoring activity is scheduled separately. ActivityStatus.INCOMPLETE: { ActivityStatus.IN_PROGRESS, ActivityStatus.COMPLETED, ActivityStatus.CANCELLED, }, ActivityStatus.COMPLETED: set(), ActivityStatus.CANCELLED: set(), } class Activity(models.Model): """A scheduled unit of work contributing to a LogFrame ACTIVITY row.""" logframe_row = models.ForeignKey( "mel_indicators.LogFrameRow", on_delete=models.PROTECT, related_name="tracked_activities", help_text="Must reference a row at ACTIVITY level.", ) name = models.CharField(max_length=255) description = models.TextField(blank=True) scheduled_start = models.DateField() scheduled_end = models.DateField() actual_start = models.DateField(null=True, blank=True) actual_end = models.DateField(null=True, blank=True) status = models.CharField( max_length=16, choices=ActivityStatus.choices, default=ActivityStatus.PLANNED, db_index=True, ) responsible = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_activities_responsible", ) parent = models.ForeignKey( "self", on_delete=models.PROTECT, null=True, blank=True, related_name="children", ) deviation_flag = models.BooleanField(default=False, db_index=True) deviation_reason = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) history = HistoricalRecords() class Meta: ordering = ["scheduled_start", "name"] indexes = [ models.Index(fields=["status", "scheduled_end"]), models.Index(fields=["deviation_flag"]), models.Index(fields=["logframe_row"]), ] constraints = [ models.CheckConstraint( check=models.Q(scheduled_end__gte=models.F("scheduled_start")), name="activity_schedule_ordering", ), ] verbose_name_plural = "activities" def __str__(self): return self.name def clean(self): if self.logframe_row_id: # Lazy import to avoid cycle at module load. from apps.mel.indicators.models import LogFrameLevel if self.logframe_row.level != LogFrameLevel.ACTIVITY: raise ValidationError( {"logframe_row": "Tracked activities must reference a row at ACTIVITY level."}, ) if self.parent_id and self.parent_id == self.pk: raise ValidationError({"parent": "An activity cannot be its own parent."}) class ActivityDependency(models.Model): """Predecessor / successor edge for Gantt-style scheduling.""" class Type(models.TextChoices): FS = "fs", "Finish-to-start" SS = "ss", "Start-to-start" FF = "ff", "Finish-to-finish" from_activity = models.ForeignKey( Activity, on_delete=models.CASCADE, related_name="outgoing_dependencies", ) to_activity = models.ForeignKey( Activity, on_delete=models.CASCADE, related_name="incoming_dependencies", ) dependency_type = models.CharField( max_length=2, choices=Type.choices, default=Type.FS, ) created_at = models.DateTimeField(auto_now_add=True) class Meta: constraints = [ models.UniqueConstraint( fields=["from_activity", "to_activity"], name="uniq_activity_dependency", ), models.CheckConstraint( check=~models.Q(from_activity=models.F("to_activity")), name="activity_dependency_not_self", ), ] verbose_name_plural = "activity dependencies" def __str__(self): return f"{self.from_activity_id} → {self.to_activity_id} ({self.dependency_type})" # --------------------------------------------------------------------------- # Output deliverables (FRMFL014–019) # --------------------------------------------------------------------------- class OutputDeliverableStatus(models.TextChoices): DRAFT = "draft", "Draft" IN_PROGRESS = "in_progress", "In progress" COMPLETED = "completed", "Completed" OVERDUE = "overdue", "Overdue" class OutputDeliverable(models.Model): """A tangible deliverable tracked against a LogFrame OUTPUT row.""" logframe_row = models.ForeignKey( "mel_indicators.LogFrameRow", on_delete=models.PROTECT, related_name="tracked_outputs", ) title = models.CharField(max_length=255) description = models.TextField(blank=True) target_quantity = models.DecimalField(max_digits=14, decimal_places=4) achieved_quantity = models.DecimalField(max_digits=14, decimal_places=4, default=0) unit = models.CharField(max_length=60) due_date = models.DateField(null=True, blank=True) status = models.CharField( max_length=16, choices=OutputDeliverableStatus.choices, default=OutputDeliverableStatus.DRAFT, ) evidence_url = models.URLField(blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) history = HistoricalRecords() class Meta: ordering = ["due_date", "title"] indexes = [ models.Index(fields=["status", "due_date"]), models.Index(fields=["logframe_row"]), ] def __str__(self): return self.title @property def percent_achieved(self) -> float | None: """Return achieved / target * 100 or ``None`` when target is zero/null.""" if self.target_quantity in (None, 0): return None return float(self.achieved_quantity) / float(self.target_quantity) * 100 def clean(self): if self.logframe_row_id: from apps.mel.indicators.models import LogFrameLevel if self.logframe_row.level != LogFrameLevel.OUTPUT: raise ValidationError( {"logframe_row": "Deliverables must reference a LogFrame row at OUTPUT level."}, ) # --------------------------------------------------------------------------- # Corrective actions (FRMFL031) # --------------------------------------------------------------------------- class CorrectiveActionStatus(models.TextChoices): OPEN = "open", "Open" IN_PROGRESS = "in_progress", "In progress" DONE = "done", "Done" # M&E SRS Table 66 — remains open until implementation is verified effective # (follow-up assessment); VERIFIED is the terminal "closed & effective" state. VERIFIED = "verified", "Verified effective" CANCELLED = "cancelled", "Cancelled" class CorrectiveSeverity(models.TextChoices): """M&E SRS Table 66 — issue classification driving the planning rule: high-risk / critical issues must carry a responsible officer and due date.""" LOW = "low", "Low" MEDIUM = "medium", "Medium" HIGH = "high", "High" CRITICAL = "critical", "Critical" class CorrectiveAction(models.Model): """Remediation tied to any subject via GenericFK. Subject can be an :class:`Activity`, :class:`OutputDeliverable`, or an Indicator — any content type may be referenced, but concrete callers use those three today. """ subject_type = models.ForeignKey( ContentType, on_delete=models.PROTECT, related_name="mel_corrective_actions", ) subject_id = models.PositiveIntegerField() subject = GenericForeignKey("subject_type", "subject_id") title = models.CharField(max_length=255) severity = models.CharField( max_length=16, choices=CorrectiveSeverity.choices, default=CorrectiveSeverity.MEDIUM, db_index=True, help_text=( "M&E SRS Table 66 — high/critical issues require a responsible " "officer and a due date before the action can be saved." ), ) root_cause = models.TextField() action_plan = models.TextField() owner = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_corrective_actions_owned", ) due_date = models.DateField(null=True, blank=True) status = models.CharField( max_length=16, choices=CorrectiveActionStatus.choices, default=CorrectiveActionStatus.OPEN, db_index=True, ) outcome_note = models.TextField(blank=True) # M&E SRS Table 66 — Follow-up Assessment: did the action restore performance? effectiveness_note = models.TextField( blank=True, help_text="Follow-up assessment — evidence the corrective action worked.", ) verified_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_corrective_actions_verified", ) verified_at = models.DateTimeField(null=True, blank=True) # M&E SRS Table 66 (E4) — overdue-action escalation bookkeeping. escalated_at = models.DateTimeField(null=True, blank=True) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_corrective_actions_created", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) closed_at = models.DateTimeField(null=True, blank=True) history = HistoricalRecords() class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["subject_type", "subject_id"]), models.Index(fields=["status", "due_date"]), ] def __str__(self): return f"[{self.status}] {self.title}" @property def is_open(self) -> bool: return self.status in { CorrectiveActionStatus.OPEN, CorrectiveActionStatus.IN_PROGRESS, } @property def is_overdue(self) -> bool: from django.utils import timezone return bool( self.is_open and self.due_date and self.due_date < timezone.now().date() ) @property def awaiting_verification(self) -> bool: """M&E SRS Table 66 — DONE means implemented, not effective. The action only counts as closed once a follow-up assessment moves it to VERIFIED (or it is cancelled); until then monitoring surfaces keep it visible as awaiting verification. """ return self.status == CorrectiveActionStatus.DONE # --------------------------------------------------------------------------- # SME-Hub longitudinal tracking (FRSME-MEI001 → MEI007, MEI012) # --------------------------------------------------------------------------- class SMEHubTrackingRecord(models.Model): """One row per SME-Hub entrepreneur — longitudinal trail across SP1–SP5. Created by ``apps.mel.tracking.receivers`` when SP1 emits ``smehub_baseline_initialised``. Subsequent SP2/SP3/SP4/SP5 milestones append to the matching JSON list. Stall detection uses ``last_milestone_at`` together with ``settings.SMEHUB_STALL_THRESHOLD_DAYS``. """ entrepreneur = models.OneToOneField( "smehub_onboarding.EntrepreneurProfile", on_delete=models.CASCADE, related_name="mel_tracking_record", ) business = models.ForeignKey( "smehub_onboarding.Business", on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_tracking_records", help_text="Primary business at SP1 baseline lock-in (FRSME-MEI001).", ) baseline_snapshot = models.ForeignKey( "smehub_onboarding.BusinessBaseline", on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_tracking_records", ) sp1_baseline_at = models.DateTimeField(null=True, blank=True) sp2_milestones = models.JSONField(default=list, blank=True) sp3_partnerships = models.JSONField(default=list, blank=True) sp4_commercialisation = models.JSONField(default=list, blank=True) sp5_investment = models.JSONField(default=list, blank=True) funding_secured_total = models.DecimalField( max_digits=16, decimal_places=2, default=0, help_text="Running total of funding_secured payloads across SP5 + manual entries.", ) last_milestone_at = models.DateTimeField(null=True, blank=True, db_index=True) is_stalled = models.BooleanField(default=False, db_index=True) stall_flagged_at = models.DateTimeField(null=True, blank=True) # FRSME-MEI019 — non-response data quality indicator. Bumped by the # ``record_feedback_non_responses`` Celery task when a dispatched # beneficiary survey passes its ``due_at`` without a submission. feedback_non_response_count = models.PositiveIntegerField(default=0) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-last_milestone_at", "-created_at"] indexes = [ models.Index(fields=["is_stalled", "last_milestone_at"]), ] verbose_name = "SME-Hub tracking record" verbose_name_plural = "SME-Hub tracking records" def __str__(self): return f"SMEHubTracking #{self.pk} — entrepreneur={self.entrepreneur_id}" # FRSME-MEI007 — the SP → bucket map an officer picks from when manually # entering a data point that a failed feed would otherwise have written. MANUAL_ENTRY_BUCKETS = ( ("sp2_milestones", "Incubation (SP2)"), ("sp3_partnerships", "Partnerships (SP3)"), ("sp4_commercialisation", "Commercialisation (SP4)"), ("sp5_investment", "Investment (SP5)"), ) def record_manual_entry( self, *, bucket: str, key: str, note: str, reason: str, by_user, amount=None ) -> None: """FRSME-MEI007 — append a MANUALLY-ENTERED milestone to a bucket. The entry is flagged ``manual`` and carries the officer's reason so the JSON trail itself distinguishes automated feeds from manual repairs. ``append_milestone`` also mirrors it to the participant timeline. ``amount`` (optional) lets a repaired *funding_secured* feed carry its value so the ``funding_secured_total`` KPI (FRSME-MEI006/MEI013) reflects it — mirroring the automated ``smehub_funding_secured`` receiver. It is applied only when this is a funding entry (``key == "funding_secured"``). """ from decimal import Decimal, InvalidOperation payload = { "manual": True, "source": "manual", "note": note or "", "reason": reason or "", "entered_by_id": getattr(by_user, "pk", None), } delta = None if amount is not None and key == "funding_secured": try: delta = Decimal(str(amount)) except (InvalidOperation, ValueError, TypeError): delta = None if delta is not None and delta > 0: payload["amount"] = str(delta) else: delta = None self.append_milestone(bucket=bucket, key=key, payload=payload) if delta is not None: # append_milestone re-saved the record; add the funding delta on top. self.funding_secured_total = (self.funding_secured_total or Decimal("0")) + delta self.save(update_fields=["funding_secured_total", "updated_at"]) def append_milestone(self, *, bucket: str, key: str, payload: dict | None = None) -> None: """Append a milestone to the named JSON bucket and bump last_milestone_at. ``bucket`` is one of ``sp2_milestones``, ``sp3_partnerships``, ``sp4_commercialisation``, ``sp5_investment``. Idempotent on ``(key, payload.id)`` when the payload contains an ``id`` field — replays of the same upstream signal will not duplicate entries. """ from django.utils import timezone if bucket not in { "sp2_milestones", "sp3_partnerships", "sp4_commercialisation", "sp5_investment", }: raise ValueError(f"Unknown milestone bucket: {bucket}") ts = timezone.now() entries = list(getattr(self, bucket) or []) payload = payload or {} new_entry_id = payload.get("id") if new_entry_id is not None: for existing in entries: if existing.get("event") == key and existing.get("payload", {}).get("id") == new_entry_id: return entries.append({"event": key, "ts": ts.isoformat(), "payload": payload}) setattr(self, bucket, entries) self.last_milestone_at = ts # Re-flag previously stalled records as active when fresh activity arrives. if self.is_stalled: self.is_stalled = False self.stall_flagged_at = None self.save( update_fields=[ bucket, "last_milestone_at", "is_stalled", "stall_flagged_at", "updated_at", ], ) # Mirror to the cross-module participant timeline (Phase 4.5). Best # effort — failures are swallowed by record_event itself. user = getattr(getattr(self.entrepreneur, "user", None), "pk", None) if user is not None: from apps.mel.tracking.services import record_event entrepreneur_user = self.entrepreneur.user payload_id = payload.get("id") if payload else None source_id = f"{bucket}:{key}:{payload_id or ts.isoformat()}" record_event( user=entrepreneur_user, source_module=TrackingRecord.Module.SMEHUB, event_type=f"smehub_{key}", source_id=source_id, occurred_at=ts, payload={"bucket": bucket, **(payload or {})}, persona=ParticipantPersona.ENTREPRENEUR, label=key.replace("_", " ").title(), ) class SMEHubFeedHandlerLog(models.Model): """Log row for any failed signal-to-tracking translation (FRSME-MEI007). Receivers wrap their bodies in try/except and write a row here on failure — never blocking the upstream service. """ signal_name = models.CharField(max_length=120, db_index=True) payload = models.JSONField(default=dict, blank=True) error = models.TextField() entrepreneur = models.ForeignKey( "smehub_onboarding.EntrepreneurProfile", on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_feed_handler_logs", ) created_at = models.DateTimeField(auto_now_add=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_feed_handler_logs_resolved", ) # FRSME-MEI007 — mandatory reason captured when an officer resolves a # failed feed (whether by acknowledging it or by manually entering the # missing data point). Preserved for the audit trail. resolution_note = models.TextField( blank=True, default="", help_text="Officer's mandatory reason recorded on resolution (FRSME-MEI007).", ) # True when resolution included a manual data-point entry against the # entrepreneur's tracking record (as opposed to a plain acknowledgement). manual_entry = models.BooleanField(default=False) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["resolved_at", "signal_name"]), ] def __str__(self): flag = "open" if self.resolved_at is None else "resolved" return f"[{flag}] {self.signal_name} #{self.pk}" @property def is_resolved(self) -> bool: return self.resolved_at is not None class SMEHubBeneficiaryFeedback(models.Model): """Links a feedback submission to an SMEHubTrackingRecord + trigger event. Created by SP6 receivers when a post-incubation / post-connection / post-deal survey is dispatched. The feedback submission itself is the canonical response — this row makes it easy to roll up the survey completion rate per stage in M&EL reports (FRSME-MEI017–019). """ class Stage(models.TextChoices): POST_INCUBATION = "post_incubation", "Post-incubation" POST_CONNECTION = "post_connection", "Post-connection" POST_DEAL = "post_deal", "Post-deal" class Status(models.TextChoices): SENT = "sent", "Sent — awaiting response" COMPLETED = "completed", "Completed" NON_RESPONDED = "non_responded", "Non-responded (window expired)" tracking_record = models.ForeignKey( "SMEHubTrackingRecord", on_delete=models.CASCADE, related_name="beneficiary_feedback", ) stage = models.CharField(max_length=24, choices=Stage.choices, db_index=True) trigger_event = models.CharField( max_length=120, help_text="Upstream signal name + source object id (e.g. 'cohort_member_completed:42').", ) feedback_channel = models.ForeignKey( "mel_feedback.FeedbackChannel", on_delete=models.SET_NULL, null=True, blank=True, related_name="smehub_beneficiary_feedback", ) feedback_submission = models.ForeignKey( "mel_feedback.FeedbackSubmission", on_delete=models.SET_NULL, null=True, blank=True, related_name="smehub_beneficiary_feedback", ) status = models.CharField( max_length=16, choices=Status.choices, default=Status.SENT, db_index=True, ) dispatched_at = models.DateTimeField(auto_now_add=True) due_at = models.DateTimeField( null=True, blank=True, help_text="Response window deadline; SMEHUB_FEEDBACK_RESPONSE_WINDOW_DAYS from dispatch.", ) reminded_at = models.DateTimeField( null=True, blank=True, help_text="Timestamp of the most recent reminder dispatched (FRSME-MEI019).", ) completed_at = models.DateTimeField(null=True, blank=True) class Meta: ordering = ["-dispatched_at"] indexes = [ models.Index(fields=["tracking_record", "stage"]), ] constraints = [ models.UniqueConstraint( fields=["tracking_record", "stage", "trigger_event"], name="uniq_smehub_beneficiary_feedback_trigger", ), ] verbose_name = "SME-Hub beneficiary feedback" verbose_name_plural = "SME-Hub beneficiary feedback" def __str__(self): return f"{self.get_stage_display()} — record #{self.tracking_record_id}" class EmploymentSnapshot(models.Model): """Phase 8 — post-baseline employment update (FRSME-MEI013 ``jobs_created``). ``BusinessBaseline`` is a one-shot snapshot at SP1 lock-in; this model captures subsequent FTE/PTE counts so the ``smehub-jobs-created`` indicator can report a real delta instead of a placeholder. Recorded manually by an M&EL officer / SME admin from the tracking detail page. Multiple snapshots per record form a time-series — the indicator builder uses the latest row per record. """ tracking_record = models.ForeignKey( "SMEHubTrackingRecord", on_delete=models.CASCADE, related_name="employment_snapshots", ) captured_at = models.DateTimeField(auto_now_add=True, db_index=True) fte_count = models.PositiveIntegerField() pte_count = models.PositiveIntegerField() note = models.CharField(max_length=255, blank=True) captured_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="+", ) class Meta: ordering = ["-captured_at"] indexes = [ models.Index(fields=["tracking_record", "-captured_at"]), ] verbose_name = "SME-Hub employment snapshot" verbose_name_plural = "SME-Hub employment snapshots" def __str__(self): return ( f"Employment snapshot record={self.tracking_record_id} " f"FTE={self.fte_count} PTE={self.pte_count}" ) @property def total(self) -> int: return self.fte_count + self.pte_count # --------------------------------------------------------------------------- # Phase 4.5 — Cross-module per-participant longitudinal tracking # --------------------------------------------------------------------------- # # These models give M&EL one canonical timeline per User that spans every # upstream module (RIMS scholarships, REP courses, SME-Hub entrepreneurship, # Alumni). Existing per-domain trackers (SMEHubTrackingRecord, AlumniProfile, # Enrolment) keep their own state — Phase 4.5 only mirrors event-grade rows # here so dashboards and impact scoring have a single source of truth. class ParticipantPersona(models.TextChoices): SCHOLAR = "scholar", "RIMS scholar" LEARNER = "learner", "REP learner" ENTREPRENEUR = "entrepreneur", "SME-Hub entrepreneur" ALUMNI = "alumni", "Alumni" RESEARCHER = "researcher", "Repository contributor" class Participant(models.Model): """One row per User who has at least one MEL-relevant persona.""" user = models.OneToOneField( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="mel_participant", ) personas = models.JSONField( default=list, blank=True, help_text="List of ParticipantPersona codes the user qualifies for.", ) primary_persona = models.CharField( max_length=24, choices=ParticipantPersona.choices, blank=True, help_text="Persona used as the default tab on the participant page.", ) scholarship = models.ForeignKey( "rims_scholarships.Scholarship", on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_participants", ) entrepreneur_profile = models.OneToOneField( "smehub_onboarding.EntrepreneurProfile", on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_participant", ) alumni_profile = models.OneToOneField( "alumni_profiles.AlumniProfile", on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_participant", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-updated_at"] indexes = [ models.Index(fields=["primary_persona"]), ] verbose_name = "MEL participant" verbose_name_plural = "MEL participants" def __str__(self): return f"Participant<{self.user_id}>" def has_persona(self, persona: str) -> bool: return persona in (self.personas or []) def add_persona(self, persona: str, *, make_primary: bool = False) -> bool: """Add a persona to the list. Returns ``True`` when changed.""" personas = list(self.personas or []) changed = False if persona not in personas: personas.append(persona) self.personas = personas changed = True if make_primary and self.primary_persona != persona: self.primary_persona = persona changed = True elif not self.primary_persona: self.primary_persona = persona changed = True if changed: self.save(update_fields=["personas", "primary_persona", "updated_at"]) return changed class TrackingLifecycle(models.TextChoices): ACTIVE = "active", "Active" COMPLETED = "completed", "Completed" STALLED = "stalled", "Stalled" EXITED = "exited", "Exited" class TrackingRecord(models.Model): """One row per (participant, source_module). Holds lifecycle + last activity. A participant may have several rows — e.g. a single User who is both a RIMS scholar and an SME-Hub entrepreneur has two records. This keeps each module's lifecycle independent while still rolling up to one :class:`Participant` for impact scoring. """ class Module(models.TextChoices): RIMS = "rims", "RIMS" REP = "rep", "REP" REPOSITORY = "repository", "Repository" SMEHUB = "smehub", "SME-Hub" ALUMNI = "alumni", "Alumni" participant = models.ForeignKey( Participant, on_delete=models.CASCADE, related_name="tracking_records", ) source_module = models.CharField(max_length=16, choices=Module.choices, db_index=True) status = models.CharField( max_length=16, choices=TrackingLifecycle.choices, default=TrackingLifecycle.ACTIVE, db_index=True, ) last_event_at = models.DateTimeField(null=True, blank=True, db_index=True) stalled_at = models.DateTimeField(null=True, blank=True) completed_at = models.DateTimeField(null=True, blank=True) exited_at = models.DateTimeField(null=True, blank=True) note = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-last_event_at", "-created_at"] constraints = [ models.UniqueConstraint( fields=["participant", "source_module"], name="uniq_tracking_record_participant_module", ), ] indexes = [ models.Index(fields=["status", "source_module"]), ] def __str__(self): return f"TrackingRecord" class TrackingMilestone(models.Model): """Canonical timeline row — one per upstream event. Idempotent on ``(record, event_type, source_id)``. Existing legacy ``UpstreamEvent`` rows continue to be written; this model is the *participant-shaped* mirror used by dashboards and impact scoring. """ record = models.ForeignKey( TrackingRecord, on_delete=models.CASCADE, related_name="milestones", ) event_type = models.CharField(max_length=64, db_index=True) occurred_at = models.DateTimeField(db_index=True) source_module = models.CharField( max_length=16, choices=TrackingRecord.Module.choices, db_index=True, ) source_id = models.CharField(max_length=64, db_index=True) payload = models.JSONField(default=dict, blank=True) label = models.CharField( max_length=255, blank=True, help_text="Human-readable summary; defaults to event_type if blank.", ) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ["-occurred_at", "-pk"] constraints = [ models.UniqueConstraint( fields=["record", "event_type", "source_id"], name="uniq_tracking_milestone", ), ] indexes = [ models.Index(fields=["record", "-occurred_at"]), models.Index(fields=["source_module", "event_type"]), ] def __str__(self): return f"{self.event_type} record={self.record_id} src={self.source_module}/{self.source_id}" @property def display_label(self) -> str: return self.label or self.event_type.replace("_", " ").title() class BaselineDimension(models.TextChoices): INCOME = "income", "Personal income" EMPLOYMENT_FTE = "employment_fte", "Employment (FTE)" EMPLOYMENT_PTE = "employment_pte", "Employment (PTE)" BUSINESS_REVENUE = "business_revenue", "Business revenue" PUBLICATIONS = "publications", "Publications" DEGREE_LEVEL = "degree_level", "Degree level" HOUSEHOLD_SIZE = "household_size", "Household size" OTHER = "other", "Other" class BaselineRecord(models.Model): """Snapshot of a measurable dimension at intake. Compared against :class:`ImpactScore` inputs to compute deltas. One row per (participant, dimension); subsequent updates create a new row so history is preserved. """ participant = models.ForeignKey( Participant, on_delete=models.CASCADE, related_name="baselines", ) source_module = models.CharField( max_length=16, choices=TrackingRecord.Module.choices, db_index=True, ) dimension = models.CharField(max_length=32, choices=BaselineDimension.choices, db_index=True) captured_at = models.DateTimeField(db_index=True) value_numeric = models.DecimalField(max_digits=18, decimal_places=4, null=True, blank=True) value_text = models.CharField(max_length=255, blank=True) unit = models.CharField(max_length=32, blank=True) payload = models.JSONField(default=dict, blank=True) note = models.CharField(max_length=255, blank=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ["participant_id", "dimension", "-captured_at"] indexes = [ models.Index(fields=["participant", "dimension", "-captured_at"]), ] def __str__(self): return f"Baseline<{self.participant_id}/{self.dimension}@{self.captured_at:%Y-%m-%d}>" class ImpactScore(models.Model): """Periodic composite score per participant. ``score`` is a 0–100 weighted sum derived from milestone counts + baseline-vs-current deltas. Components carry the per-dimension breakdown so the UI can show the contribution of each input. Methodology version captures the weight set used so historical scores stay reproducible. """ class Period(models.TextChoices): BASELINE = "baseline", "Baseline" Y1 = "y1", "Year 1" Y2 = "y2", "Year 2" Y3 = "y3", "Year 3" Y5 = "y5", "Year 5" participant = models.ForeignKey( Participant, on_delete=models.CASCADE, related_name="impact_scores", ) period = models.CharField(max_length=16, choices=Period.choices, db_index=True) score = models.DecimalField(max_digits=5, decimal_places=2) components = models.JSONField(default=dict, blank=True) methodology_version = models.CharField(max_length=32, default="v1") computed_at = models.DateTimeField(auto_now_add=True, db_index=True) class Meta: ordering = ["participant_id", "period", "-computed_at"] constraints = [ models.UniqueConstraint( fields=["participant", "period", "methodology_version"], name="uniq_impact_score_period_method", ), ] indexes = [ models.Index(fields=["period", "-score"]), ] def __str__(self): return f"Impact<{self.participant_id}/{self.period}={self.score}>" # --------------------------------------------------------------------------- # MEL Configuration (FRSME-MEI012) # --------------------------------------------------------------------------- class MELConfiguration(models.Model): """Singleton (pk=1) holding tunable MEL knobs. Currently exposes the SME-Hub stall threshold (in days). Falls back to ``settings.SMEHUB_STALL_THRESHOLD_DAYS`` if no row exists yet. """ smehub_stall_threshold_days = models.PositiveSmallIntegerField( default=90, help_text="Entrepreneurs with no milestone in this many days are flagged stalled.", ) updated_at = models.DateTimeField(auto_now=True) updated_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="+", ) class Meta: verbose_name = "MEL configuration" verbose_name_plural = "MEL configuration" def __str__(self): return f"MELConfiguration(stall_threshold_days={self.smehub_stall_threshold_days})" @classmethod def get(cls) -> "MELConfiguration": obj, _ = cls.objects.get_or_create(pk=1) return obj def save(self, *args, **kwargs): # Enforce singleton — always pk=1. self.pk = 1 super().save(*args, **kwargs) # --------------------------------------------------------------------------- # G2 — QR-code attendance & registration # --------------------------------------------------------------------------- def _generate_attendance_token() -> str: """Mirror of FeedbackChannel.generate_token — 32-char URL-safe token.""" import secrets return secrets.token_urlsafe(24) class ActivityAttendanceChannel(models.Model): """Public scan target for an Activity. Mirrors the FeedbackChannel pattern. One channel per Activity. The token-encoded URL is encoded into a QR code that officers print/display at the event venue. Anyone with the URL can submit attendance — minimal friction is the point. """ activity = models.OneToOneField( "mel_tracking.Activity", on_delete=models.CASCADE, related_name="attendance_channel", ) token = models.CharField( max_length=48, unique=True, default=_generate_attendance_token, help_text="Opaque token used in the public scan URL.", ) is_active = models.BooleanField(default=True, db_index=True) opens_at = models.DateTimeField( null=True, blank=True, help_text="Optional gate — channel returns 410 before this time.", ) closes_at = models.DateTimeField( null=True, blank=True, help_text="Optional gate — channel returns 410 after this time.", ) require_email = models.BooleanField( default=False, help_text="When True, the public form requires an email for self-identification.", ) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="created_attendance_channels", ) 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"attendance[{self.activity_id}]" class BeneficiaryAttendance(models.Model): """One row per attendee per activity. Idempotent on (channel, dedupe_key). The ``dedupe_key`` is a SHA-256 of (lowercased email or phone or name+activity_id) so accidental double-submissions from the same attendee don't inflate the count. """ channel = models.ForeignKey( ActivityAttendanceChannel, on_delete=models.CASCADE, related_name="attendances", ) attendee_user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="mel_event_attendances", ) attendee_name = models.CharField(max_length=255, blank=True) attendee_email = models.EmailField(blank=True) attendee_phone = models.CharField(max_length=40, blank=True) organisation = models.CharField(max_length=255, blank=True) notes = models.TextField(blank=True) dedupe_key = models.CharField(max_length=64, db_index=True) attended_at = models.DateTimeField(auto_now_add=True, db_index=True) history = HistoricalRecords() class Meta: ordering = ["-attended_at"] constraints = [ models.UniqueConstraint( fields=["channel", "dedupe_key"], name="uniq_attendance_per_channel", ), ] indexes = [ models.Index(fields=["channel", "attended_at"]), ] def __str__(self): return f"{self.attendee_name or self.attendee_email or 'anon'} @ activity {self.channel.activity_id}"