"""Service layer that turns upstream signal payloads into M&EL writes. Each ``record_*`` function is idempotent by ``source_object_id`` and safe to call from inside the upstream transaction. Failures here log + swallow rather than propagate, so a misconfigured indicator cannot roll back an upstream award / enrolment / publish. """ from __future__ import annotations import logging from dataclasses import dataclass from decimal import Decimal from typing import Any from django.conf import settings from django.core.exceptions import ValidationError from django.db import transaction from django.utils import timezone from apps.mel.indicators.services import record_automated_point from apps.mel.tracking.models import ( ACTIVITY_STATUS_TRANSITIONS, Activity, ActivityStatus, BaselineDimension, BaselineRecord, ImpactScore, OutputDeliverable, OutputDeliverableStatus, Participant, ParticipantPersona, TrackingLifecycle, TrackingMilestone, TrackingRecord, UpstreamEvent, ) logger = logging.getLogger(__name__) @transaction.atomic def log_event( *, module: str, event_type: str, source_object_id: str | int, payload: dict[str, Any] | None = None, subject=None, ) -> UpstreamEvent: obj, created = UpstreamEvent.objects.get_or_create( module=module, event_type=event_type, source_object_id=str(source_object_id), defaults={ "payload": payload or {}, "subject": subject, }, ) if not created and (payload or subject): # Preserve the first-seen record but update payload if it grew. dirty = False if payload and obj.payload != payload: obj.payload = payload dirty = True if subject and obj.subject_id is None: obj.subject = subject dirty = True if dirty: obj.save(update_fields=["payload", "subject"]) return obj def _safe_record(indicator_code: str, source_module: str, event_type: str, source_id: str, value=1) -> None: try: record_automated_point( indicator_code=indicator_code, source_module=source_module, source_event=event_type, source_object_id=str(source_id), value=value, ) except Exception: # pragma: no cover - defensive logger.exception( "mel.datapoint failed indicator=%s source=%s/%s object=%s", indicator_code, source_module, event_type, source_id, ) # --------------------------------------------------------------------------- # Phase 4.5 — Participant timeline (cross-module) # --------------------------------------------------------------------------- _LIFECYCLE_COMPLETED_EVENTS = { "scholar_graduated", "course_completed", "programme_completed", "smehub_programme_complete", } _LIFECYCLE_EXITED_EVENTS = { "enrolment_withdrawn", "scholarship_terminated", } def ensure_participant(user, persona: str | None = None) -> Participant | None: """Idempotently create the :class:`Participant` row for ``user``. Returns ``None`` when no user is supplied — callers must tolerate this because some upstream events (e.g. anonymous repository document publish) don't carry a subject. """ if user is None or getattr(user, "pk", None) is None: return None participant, _ = Participant.objects.get_or_create(user=user) if persona: participant.add_persona(persona) return participant def _resolve_persona(source_module: str, override: str | None = None) -> str | None: if override: return override return { TrackingRecord.Module.RIMS: ParticipantPersona.SCHOLAR, TrackingRecord.Module.REP: ParticipantPersona.LEARNER, TrackingRecord.Module.SMEHUB: ParticipantPersona.ENTREPRENEUR, TrackingRecord.Module.ALUMNI: ParticipantPersona.ALUMNI, TrackingRecord.Module.REPOSITORY: ParticipantPersona.RESEARCHER, }.get(source_module) def _ensure_record(participant: Participant, source_module: str) -> TrackingRecord: record, _ = TrackingRecord.objects.get_or_create( participant=participant, source_module=source_module, ) return record def record_event( *, user, source_module: str, event_type: str, source_id: str | int, occurred_at=None, payload: dict | None = None, persona: str | None = None, label: str = "", ) -> TrackingMilestone | None: """Mirror an upstream event into the participant timeline. Idempotent on ``(record, event_type, source_id)``. Failures are logged but never re-raised so this can be called from inside upstream transactions. """ try: participant = ensure_participant(user, persona=_resolve_persona(source_module, persona)) if participant is None: return None record = _ensure_record(participant, source_module) ts = occurred_at or timezone.now() milestone, created = TrackingMilestone.objects.get_or_create( record=record, event_type=event_type, source_id=str(source_id), defaults={ "occurred_at": ts, "source_module": source_module, "payload": payload or {}, "label": label, }, ) if not created: return milestone # Lifecycle update. update_fields: list[str] = [] if record.last_event_at is None or ts > record.last_event_at: record.last_event_at = ts update_fields.append("last_event_at") if event_type in _LIFECYCLE_COMPLETED_EVENTS and record.status != TrackingLifecycle.COMPLETED: record.status = TrackingLifecycle.COMPLETED record.completed_at = ts update_fields.extend(["status", "completed_at"]) elif event_type in _LIFECYCLE_EXITED_EVENTS and record.status not in { TrackingLifecycle.COMPLETED, TrackingLifecycle.EXITED, }: record.status = TrackingLifecycle.EXITED record.exited_at = ts update_fields.extend(["status", "exited_at"]) elif record.status == TrackingLifecycle.STALLED: # Fresh activity un-stalls. record.status = TrackingLifecycle.ACTIVE record.stalled_at = None update_fields.extend(["status", "stalled_at"]) if update_fields: update_fields.append("updated_at") record.save(update_fields=update_fields) return milestone except Exception: # pragma: no cover — never break upstream logger.exception( "mel.timeline mirror failed module=%s event=%s src=%s", source_module, event_type, source_id, ) return None def capture_baseline( *, user, source_module: str, dimension: str, value_numeric=None, value_text: str = "", unit: str = "", captured_at=None, payload: dict | None = None, note: str = "", ) -> BaselineRecord | None: participant = ensure_participant(user, persona=_resolve_persona(source_module)) if participant is None: return None return BaselineRecord.objects.create( participant=participant, source_module=source_module, dimension=dimension, captured_at=captured_at or timezone.now(), value_numeric=value_numeric, value_text=value_text, unit=unit, payload=payload or {}, note=note, ) # --------------------------------------------------------------------------- # RIMS # --------------------------------------------------------------------------- def record_award(award) -> None: applicant = getattr(award.application, "applicant", None) if award.application_id else None call_slug = getattr(award.application.call, "slug", None) if award.application_id else None log_event( module=UpstreamEvent.Module.RIMS, event_type="award_created", source_object_id=award.pk, payload={ "application_id": award.application_id, "call_slug": call_slug, }, subject=applicant, ) _safe_record("rims-grants-awarded", "rims", "award_created", award.pk) record_event( user=applicant, source_module=TrackingRecord.Module.RIMS, event_type="award_created", source_id=award.pk, payload={"application_id": award.application_id, "call_slug": call_slug}, label=f"RIMS award #{award.pk}", ) def record_milestone_accepted(milestone) -> None: """PRD §4.3.4 / FRFA-AA015 — feed accepted milestones into the M&E pipeline so dashboards can count completion against indicators that listen for ``rims-projects-milestone-accepted``.""" project = getattr(milestone, "project", None) award = getattr(project, "award", None) if project else None applicant = ( getattr(award.application, "applicant", None) if award and getattr(award, "application_id", None) else None ) log_event( module=UpstreamEvent.Module.RIMS, event_type="milestone_accepted", source_object_id=milestone.pk, payload={ "project_id": getattr(project, "pk", None), "milestone_name": getattr(milestone, "name", ""), "completion_percent": str(getattr(milestone, "completion_percent", "")), }, subject=applicant, ) _safe_record( "rims-projects-milestone-accepted", "rims", "milestone_accepted", milestone.pk, ) record_event( user=applicant, source_module=TrackingRecord.Module.RIMS, event_type="milestone_accepted", source_id=milestone.pk, payload={ "project_id": getattr(project, "pk", None), "milestone_name": getattr(milestone, "name", ""), }, label=f"Milestone accepted — {getattr(milestone, 'name', '')[:80]}", ) def record_milestone_overdue(milestone) -> None: """Companion to record_milestone_accepted — emits when a milestone has been flagged as OVERDUE so off-track indicators can fire.""" project = getattr(milestone, "project", None) award = getattr(project, "award", None) if project else None applicant = ( getattr(award.application, "applicant", None) if award and getattr(award, "application_id", None) else None ) log_event( module=UpstreamEvent.Module.RIMS, event_type="milestone_overdue", source_object_id=milestone.pk, payload={ "project_id": getattr(project, "pk", None), "milestone_name": getattr(milestone, "name", ""), }, subject=applicant, ) _safe_record( "rims-projects-milestone-overdue", "rims", "milestone_overdue", milestone.pk, ) record_event( user=applicant, source_module=TrackingRecord.Module.RIMS, event_type="milestone_overdue", source_id=milestone.pk, payload={ "project_id": getattr(project, "pk", None), "milestone_name": getattr(milestone, "name", ""), }, label=f"Milestone overdue — {getattr(milestone, 'name', '')[:80]}", ) def record_graduation(graduation) -> None: scholar = getattr(graduation, "scholar", None) subject = getattr(scholar, "user", None) if scholar else None log_event( module=UpstreamEvent.Module.RIMS, event_type="scholar_graduated", source_object_id=getattr(graduation, "pk", graduation), payload={"scholar_id": getattr(graduation, "scholar_id", None)}, subject=subject, ) _safe_record("rims-scholars-graduated", "rims", "scholar_graduated", getattr(graduation, "pk", graduation)) record_event( user=subject, source_module=TrackingRecord.Module.RIMS, event_type="scholar_graduated", source_id=getattr(graduation, "pk", graduation), payload={"scholar_id": getattr(graduation, "scholar_id", None)}, label="Scholar graduated", ) # --------------------------------------------------------------------------- # Repository # --------------------------------------------------------------------------- def record_document_published(document) -> None: title = getattr(document, "title", "")[:200] log_event( module=UpstreamEvent.Module.REPOSITORY, event_type="document_published", source_object_id=document.pk, payload={"title": title}, ) _safe_record("repo-documents-published", "repository", "document_published", document.pk) # Best-effort author resolution: many Document models track an uploader. author = getattr(document, "uploaded_by", None) or getattr(document, "author", None) record_event( user=author, source_module=TrackingRecord.Module.REPOSITORY, event_type="document_published", source_id=document.pk, payload={"title": title}, label=f"Published: {title[:80]}", ) def record_document_access_granted(access) -> None: log_event( module=UpstreamEvent.Module.REPOSITORY, event_type="access_granted", source_object_id=access.pk, payload={"document_id": getattr(access, "document_id", None)}, subject=getattr(access, "requester", None), ) _safe_record("repo-documents-accessed", "repository", "access_granted", access.pk) record_event( user=getattr(access, "requester", None), source_module=TrackingRecord.Module.REPOSITORY, event_type="access_granted", source_id=access.pk, payload={"document_id": getattr(access, "document_id", None)}, label="Repository access granted", ) # --------------------------------------------------------------------------- # Activity tracking (FRMFL009–013) # --------------------------------------------------------------------------- @dataclass(frozen=True) class ActivityProgressResult: activity: Activity previous_status: str transitioned: bool @transaction.atomic def record_activity_progress( activity_id: int, *, new_status: str, user=None, actual_start=None, actual_end=None, deviation_reason: str = "", ) -> ActivityProgressResult: """Transition an Activity's status with an enforced transition matrix. Accepts a same-status write as a no-op so callers can idempotently "touch" an activity without triggering the signal. """ activity = Activity.objects.select_for_update().get(pk=activity_id) prev = activity.status if new_status == prev: return ActivityProgressResult(activity, prev, transitioned=False) allowed = ACTIVITY_STATUS_TRANSITIONS.get(prev, set()) if new_status not in allowed: raise ValidationError( f"Cannot transition Activity {activity.pk} from {prev} → {new_status}." ) activity.status = new_status update_fields = ["status", "updated_at"] if new_status == ActivityStatus.IN_PROGRESS and activity.actual_start is None: activity.actual_start = actual_start or timezone.now().date() update_fields.append("actual_start") if new_status == ActivityStatus.COMPLETED: activity.actual_end = actual_end or timezone.now().date() update_fields.append("actual_end") activity.deviation_flag = False update_fields.append("deviation_flag") if new_status in (ActivityStatus.DELAYED, ActivityStatus.INCOMPLETE): # M&E SRS Table 65 — an incomplete field collection is a deviation; the # reason is recorded and a follow-up monitoring activity is scheduled. activity.deviation_flag = True if deviation_reason: activity.deviation_reason = deviation_reason update_fields.append("deviation_reason") update_fields.append("deviation_flag") activity.save(update_fields=update_fields) logger.info( "mel.activity status=%s→%s pk=%s user=%s", prev, new_status, activity.pk, getattr(user, "pk", None), ) if new_status == ActivityStatus.DELAYED: transaction.on_commit(lambda: _emit_activity_delayed(activity)) if new_status == ActivityStatus.COMPLETED: transaction.on_commit(lambda: _maybe_notify_output_ready_to_close(activity)) return ActivityProgressResult(activity, prev, transitioned=True) @transaction.atomic def schedule_followup_monitoring( activity: Activity, *, scheduled_start, scheduled_end, name: str = "", user=None, ) -> Activity: """M&E SRS Table 65 — schedule follow-up monitoring for an incomplete activity. Creates a new PLANNED Activity linked to ``activity`` via ``parent`` (and the same Results-Framework row), so the incomplete collection is picked up again without losing the audit trail. Official indicator values stay untouched — they only move when a follow-up completes and verified data is recorded. """ followup = Activity.objects.create( logframe_row=activity.logframe_row, name=name.strip() or f"Follow-up monitoring — {activity.name}", description=( f"Follow-up for incomplete monitoring activity “{activity.name}”. " f"Reason: {activity.deviation_reason or 'not recorded'}." ), scheduled_start=scheduled_start, scheduled_end=scheduled_end, responsible=activity.responsible, parent=activity, status=ActivityStatus.PLANNED, ) logger.info( "mel.activity followup pk=%s parent=%s user=%s", followup.pk, activity.pk, getattr(user, "pk", None), ) return followup def _emit_activity_delayed(activity: Activity) -> None: from apps.mel.tracking.signals import mel_activity_delayed mel_activity_delayed.send(sender=Activity, activity=activity) def _maybe_notify_output_ready_to_close(activity: Activity) -> None: """G17 — when every sibling ACTIVITY under this row's OUTPUT parent has finished (COMPLETED or CANCELLED), emit a single MEL_OUTPUT_READY_TO_CLOSE notification to the logframe owner and fire :data:`mel_output_ready_to_close`. Visibility only: the OutputDeliverable status is NOT auto-mutated, since deliverables track their own ``achieved_quantity`` independently. """ from apps.mel.indicators.models import LogFrameLevel from apps.mel.tracking.signals import mel_output_ready_to_close activity_row = activity.logframe_row parent_row = getattr(activity_row, "parent", None) if parent_row is None or parent_row.level != LogFrameLevel.OUTPUT: return sibling_activity_rows = parent_row.children.filter(level=LogFrameLevel.ACTIVITY) open_count = Activity.objects.filter( logframe_row__in=sibling_activity_rows, ).exclude( status__in=(ActivityStatus.COMPLETED, ActivityStatus.CANCELLED), ).count() if open_count > 0: return deliverables = list(parent_row.tracked_outputs.all()) owner = getattr(parent_row.logframe, "owner", None) if owner is None or not deliverables: mel_output_ready_to_close.send( sender=Activity, output_row=parent_row, deliverables=deliverables, ) return from apps.core.notifications.models import Notification from apps.core.notifications.services import send_notification titles = ", ".join(d.title[:40] for d in deliverables[:3]) extra = f" (+{len(deliverables) - 3} more)" if len(deliverables) > 3 else "" message = ( f"All activities under output ‘{parent_row.title}’ are complete. " f"Ready to close: {titles}{extra}." ) try: send_notification( owner, message, verb=Notification.Verb.MEL_OUTPUT_READY_TO_CLOSE, ) except Exception: # pragma: no cover - never break the activity transition logger.exception( "mel.output ready_to_close notification failed row=%s", parent_row.pk ) mel_output_ready_to_close.send( sender=Activity, output_row=parent_row, deliverables=deliverables, ) def scan_delayed_activities(*, today=None) -> int: """Mark PLANNED/IN_PROGRESS activities past scheduled_end as DELAYED. Returns the number of activities flagged. Called by a Celery beat task. Uses the grace period from ``settings.MEL_ACTIVITY_DELAY_GRACE_DAYS``. """ from datetime import timedelta grace = int(getattr(settings, "MEL_ACTIVITY_DELAY_GRACE_DAYS", 0)) cutoff = (today or timezone.now().date()) - timedelta(days=grace) qs = Activity.objects.filter( status__in=(ActivityStatus.PLANNED, ActivityStatus.IN_PROGRESS), scheduled_end__lt=cutoff, ) flagged = 0 for activity in qs: try: record_activity_progress( activity.pk, new_status=ActivityStatus.DELAYED, deviation_reason="Automatically flagged — past scheduled end.", ) flagged += 1 except ValidationError: # Completed/Cancelled already, or illegal transition — skip. continue if flagged: logger.info("mel.activity.scan flagged=%s cutoff=%s", flagged, cutoff) return flagged # --------------------------------------------------------------------------- # Output monitoring (FRMFL014–019) # --------------------------------------------------------------------------- @transaction.atomic def record_output_progress( output_id: int, *, achieved: Decimal | float | int, evidence_url: str | None = None, ) -> OutputDeliverable: """Update ``achieved_quantity`` on an OutputDeliverable and re-derive status.""" deliverable = OutputDeliverable.objects.select_for_update().get(pk=output_id) deliverable.achieved_quantity = Decimal(str(achieved)) if evidence_url is not None: deliverable.evidence_url = evidence_url # Derive status from progress + due date. percent = deliverable.percent_achieved today = timezone.now().date() if percent is not None and percent >= 100: deliverable.status = OutputDeliverableStatus.COMPLETED elif deliverable.due_date and deliverable.due_date < today and ( deliverable.status != OutputDeliverableStatus.COMPLETED ): deliverable.status = OutputDeliverableStatus.OVERDUE elif deliverable.status == OutputDeliverableStatus.DRAFT and deliverable.achieved_quantity > 0: deliverable.status = OutputDeliverableStatus.IN_PROGRESS deliverable.save( update_fields=["achieved_quantity", "evidence_url", "status", "updated_at"], ) return deliverable def scan_overdue_outputs(*, today=None) -> int: """Flag past-due, still-open OutputDeliverables as OVERDUE (FRMFL016). ``record_output_progress`` only re-derives status when an officer posts progress, so a neglected deliverable that is never touched after its due date stays IN_PROGRESS/DRAFT forever. This nightly sweep closes that gap: any DRAFT/IN_PROGRESS deliverable whose ``due_date`` is in the past and that is not already at 100% is flipped to OVERDUE. Returns the number of deliverables newly flagged. """ today = today or timezone.now().date() qs = OutputDeliverable.objects.filter( status__in=( OutputDeliverableStatus.DRAFT, OutputDeliverableStatus.IN_PROGRESS, ), due_date__isnull=False, due_date__lt=today, ) flagged = 0 for deliverable in qs: percent = deliverable.percent_achieved if percent is not None and percent >= 100: # Fully achieved but never marked complete — leave for the progress # path to close; it is not "overdue" in any meaningful sense. continue OutputDeliverable.objects.filter(pk=deliverable.pk).update( status=OutputDeliverableStatus.OVERDUE, updated_at=timezone.now(), ) flagged += 1 if flagged: logger.info("mel.output.scan overdue_flagged=%s today=%s", flagged, today) return flagged def scan_overdue_corrective_actions(*, today=None) -> int: """M&E SRS Table 66 (E4) — remind + escalate overdue corrective actions. Any OPEN/IN_PROGRESS :class:`CorrectiveAction` whose ``due_date`` has passed is escalated to its owner and to programme managers/M&E officers, and stamped with ``escalated_at`` so it is not re-notified until it lapses again. Returns the number escalated this run. """ from datetime import timedelta from django.contrib.auth import get_user_model from django.db.models import Q from apps.core.notifications.models import Notification from apps.core.notifications.services import send_notification from apps.core.permissions.roles import UserRole from apps.mel.tracking.models import CorrectiveAction, CorrectiveActionStatus today = today or timezone.now().date() now = timezone.now() # Re-escalate at most once per day. recent = now - timedelta(days=1) qs = CorrectiveAction.objects.filter( status__in=(CorrectiveActionStatus.OPEN, CorrectiveActionStatus.IN_PROGRESS), due_date__isnull=False, due_date__lt=today, ).filter(Q(escalated_at__isnull=True) | Q(escalated_at__lt=recent)) escalated = 0 recipients_qs = None for action in qs: recipients = set() if action.owner_id: recipients.add(action.owner) if recipients_qs is None: recipients_qs = list( get_user_model().objects.filter( role__in=[ UserRole.MEL_OFFICER, UserRole.PROGRAM_MANAGER, UserRole.PROGRAM_DIRECTOR, ], is_active=True, ) ) recipients.update(recipients_qs) message = ( f"Overdue corrective action ‘{action.title}’ was due " f"{action.due_date:%d %b %Y} and is still {action.get_status_display().lower()}." ) for user in recipients: try: send_notification( user, message, verb=Notification.Verb.MEL_ACTIVITY_DELAYED, ) except Exception: # pragma: no cover - never break the sweep logger.exception( "mel.corrective escalation notify failed action=%s", action.pk ) CorrectiveAction.objects.filter(pk=action.pk).update(escalated_at=now) escalated += 1 if escalated: logger.info("mel.corrective.scan escalated=%s today=%s", escalated, today) return escalated @transaction.atomic def resolve_corrective_action(action_id: int, *, actor, new_status: str, note: str = ""): """M&E SRS Table 66 — advance a corrective action through its lifecycle. Setting status to VERIFIED records the follow-up assessment (who verified it was effective, when, and the effectiveness note) and stamps ``closed_at`` so the action only closes once implementation is verified. """ from apps.mel.tracking.models import CorrectiveAction, CorrectiveActionStatus action = CorrectiveAction.objects.select_for_update().get(pk=action_id) # M&E SRS Table 66 — a corrective action stays OPEN until implementation is # verified, and a closed (verified / cancelled) action does not silently # revert. Enforced at the service layer so non-UI callers obey the rule too. terminal = {CorrectiveActionStatus.VERIFIED, CorrectiveActionStatus.CANCELLED} if action.status in terminal and new_status != action.status: raise ValidationError( f"This corrective action is already {action.get_status_display().lower()} " "and cannot be reopened." ) # M&E SRS Table 66 — "Follow-up assessments shall be completed before # corrective actions are formally closed." VERIFIED (closed & effective) # requires a follow-up/effectiveness note. if new_status == CorrectiveActionStatus.VERIFIED and not (note or "").strip(): raise ValidationError( "Record a follow-up assessment before marking the action verified effective." ) action.status = new_status update_fields = ["status", "updated_at"] if note: if new_status == CorrectiveActionStatus.VERIFIED: action.effectiveness_note = note.strip() update_fields.append("effectiveness_note") else: action.outcome_note = note.strip() update_fields.append("outcome_note") if new_status == CorrectiveActionStatus.VERIFIED: action.verified_by = actor action.verified_at = timezone.now() action.closed_at = timezone.now() update_fields += ["verified_by", "verified_at", "closed_at"] elif new_status in (CorrectiveActionStatus.DONE, CorrectiveActionStatus.CANCELLED): action.closed_at = timezone.now() update_fields.append("closed_at") action.save(update_fields=update_fields) logger.info( "mel.corrective resolved pk=%s status=%s by=%s", action.pk, new_status, getattr(actor, "pk", None), ) return action # --------------------------------------------------------------------------- # Phase 4.5 — Impact scoring # --------------------------------------------------------------------------- _IMPACT_METHODOLOGY_VERSION = "v1" def _impact_weights() -> dict: """Read weight table from Django settings, falling back to v1 defaults. Override via ``settings.MEL_IMPACT_WEIGHTS`` (a dict). Each key is a component name; the value is the maximum points contributed by that component, which scales linearly between 0 and the cap. """ return getattr(settings, "MEL_IMPACT_WEIGHTS", None) or { "rims_milestones": 15, "rep_completions": 25, "smehub_progress": 25, "alumni_publications": 10, "employment_delta": 10, "revenue_delta": 15, } def _component_milestones(participant: Participant, source_module: str, terminal_event: str | None = None) -> int: qs = TrackingMilestone.objects.filter( record__participant=participant, source_module=source_module ) if terminal_event: qs = qs.filter(event_type=terminal_event) return qs.count() def _component_employment_delta(participant: Participant) -> tuple[int, int]: """Return (current_total, baseline_total) FTE+PTE for the participant.""" baseline = ( BaselineRecord.objects.filter( participant=participant, dimension__in=[BaselineDimension.EMPLOYMENT_FTE, BaselineDimension.EMPLOYMENT_PTE], ) .order_by("dimension", "captured_at") ) base_total = sum(int(b.value_numeric or 0) for b in baseline) # Current: latest EmploymentSnapshot on any SMEHubTrackingRecord for the # participant's entrepreneur profile. current_total = base_total if participant.entrepreneur_profile_id is not None: from apps.mel.tracking.models import EmploymentSnapshot latest = ( EmploymentSnapshot.objects.filter( tracking_record__entrepreneur_id=participant.entrepreneur_profile_id, ) .order_by("-captured_at") .first() ) if latest is not None: current_total = int(latest.fte_count or 0) + int(latest.pte_count or 0) return current_total, base_total def _component_revenue_delta(participant: Participant) -> tuple[float, float]: """Return (current_revenue, baseline_revenue) for the participant in USD-ish units.""" baseline = ( BaselineRecord.objects.filter( participant=participant, dimension=BaselineDimension.BUSINESS_REVENUE, ) .order_by("captured_at") .first() ) base = float(baseline.value_numeric or 0) if baseline else 0.0 # Current — sum the latest sp4 commercialisation entries that report revenue. current = base if participant.entrepreneur_profile_id is not None: from apps.mel.tracking.models import SMEHubTrackingRecord record = SMEHubTrackingRecord.objects.filter( entrepreneur_id=participant.entrepreneur_profile_id ).first() if record is not None: for entry in (record.sp4_commercialisation or []): payload = entry.get("payload") or {} value = payload.get("revenue") or payload.get("amount") if value: try: current = max(current, float(value)) except (TypeError, ValueError): continue return current, base def compute_impact_score( participant: Participant, *, period: str = ImpactScore.Period.Y1, methodology_version: str = _IMPACT_METHODOLOGY_VERSION, ) -> ImpactScore: """Compute (or refresh) the :class:`ImpactScore` for a participant + period. Idempotent on ``(participant, period, methodology_version)``. """ weights = _impact_weights() components: dict[str, float] = {} rims_count = _component_milestones( participant, TrackingRecord.Module.RIMS, terminal_event="milestone_accepted" ) components["rims_milestones"] = min(weights.get("rims_milestones", 0), rims_count * 3) rep_completions = _component_milestones( participant, TrackingRecord.Module.REP, terminal_event="course_completed" ) components["rep_completions"] = min(weights.get("rep_completions", 0), rep_completions * 5) smehub_count = _component_milestones(participant, TrackingRecord.Module.SMEHUB) components["smehub_progress"] = min(weights.get("smehub_progress", 0), smehub_count * 2) alumni_pubs = _component_milestones( participant, TrackingRecord.Module.ALUMNI, terminal_event="publication_added" ) components["alumni_publications"] = min( weights.get("alumni_publications", 0), alumni_pubs * 4 ) current_emp, baseline_emp = _component_employment_delta(participant) delta_emp = max(0, current_emp - baseline_emp) components["employment_delta"] = min(weights.get("employment_delta", 0), delta_emp * 2) current_rev, baseline_rev = _component_revenue_delta(participant) delta_rev_pct = ((current_rev - baseline_rev) / baseline_rev * 100) if baseline_rev else 0 components["revenue_delta"] = min( weights.get("revenue_delta", 0), max(0, delta_rev_pct) / 5 ) raw_total = sum(components.values()) cap = sum(weights.values()) or 100 score = round(min(100, raw_total / cap * 100), 2) obj, _ = ImpactScore.objects.update_or_create( participant=participant, period=period, methodology_version=methodology_version, defaults={ "score": score, "components": {k: round(v, 2) for k, v in components.items()}, }, ) return obj def generate_impact_report( *, participant: Participant | None = None, period: str = ImpactScore.Period.Y1, template_slug: str = "impact-summary", ) -> "Report | None": """Wrap :func:`apps.mel.reports.services.generate_report` with an impact-score payload. Falls back to building scores for a single participant when one is supplied — otherwise refreshes scores for every Participant. """ if participant is not None: compute_impact_score(participant, period=period) targets = [participant] else: targets = list(Participant.objects.all()) for p in targets: try: compute_impact_score(p, period=period) except Exception: # pragma: no cover — defensive logger.exception("compute_impact_score failed for participant=%s", p.pk) try: from apps.mel.reports.models import ReportTemplate from apps.mel.reports.services import generate_render_and_publish, quarter_window template = ReportTemplate.objects.filter(slug=template_slug, is_active=True).first() if template is None: logger.info("generate_impact_report: template %s missing — skipping", template_slug) return None label, start, end = quarter_window() return generate_render_and_publish( template=template, period_label=label, period_start=start, period_end=end, ) except Exception: # pragma: no cover - defensive logger.exception("generate_impact_report failed") return None # --------------------------------------------------------------------------- # Stall notification (FRSME-MEI012) # --------------------------------------------------------------------------- def notify_newly_stalled(records, *, threshold_days: int) -> int: """Alert MEL officers + SME admins about newly-stalled entrepreneurs. Best-effort: the stall-flagging sweep already persists the flag, and this helper exists only to surface it to operators. Returns the number of recipients notified. """ if not records: return 0 try: from django.contrib.auth import get_user_model from apps.smehub._shared.notifications import Verb, bulk_notify_smehub User = get_user_model() recipients = list( User.objects.filter( role__in=("mel_officer", "system_admin", "admin", "sme_admin"), is_active=True, ) ) if not recipients: return 0 count = len(records) message = ( f"{count} SME-Hub entrepreneur{'s' if count != 1 else ''} flagged stalled " f"after {threshold_days} days without a milestone — review the M&E tracking dashboard." ) from django.urls import reverse try: action_url = reverse("mel_tracking:smehub_stalled") except Exception: # pragma: no cover — URL resolution guard action_url = "/mel/events/smehub/stalled/" bulk_notify_smehub( recipients, message, verb=getattr(Verb, "SMEHUB_RECORDS_STALLED", None) or "smehub_records_stalled", action_url=action_url, ) return len(recipients) except Exception: # pragma: no cover — best-effort alerting logger.exception("notify_newly_stalled dispatch failed") return 0 # --------------------------------------------------------------------------- # G2 — QR-code attendance & registration # --------------------------------------------------------------------------- class AttendanceChannelClosedError(ValueError): """Raised when an attendance channel is inactive or outside its time window.""" def _attendance_dedupe_key( *, activity_id: int, email: str, phone: str, name: str, ) -> str: """SHA-256 of the attendee's strongest identifier scoped to the activity. Email > phone > name fallback. Same person scanning twice for the same activity collides on the unique constraint and is silently deduplicated. """ import hashlib canonical = (email or "").strip().lower() if not canonical: canonical = (phone or "").strip().replace(" ", "").lower() if not canonical: canonical = (name or "").strip().lower() if not canonical: canonical = f"anon:{timezone.now().isoformat()}" return hashlib.sha256(f"{activity_id}:{canonical}".encode("utf-8")).hexdigest() @transaction.atomic def ensure_attendance_channel(activity, *, created_by=None): """Idempotent get-or-create of the attendance channel for ``activity``.""" from apps.mel.tracking.models import ActivityAttendanceChannel channel, _created = ActivityAttendanceChannel.objects.get_or_create( activity=activity, defaults={"created_by": created_by, "is_active": True}, ) return channel def _channel_is_open(channel) -> bool: if not channel.is_active: return False now = timezone.now() if channel.opens_at and now < channel.opens_at: return False if channel.closes_at and now > channel.closes_at: return False return True @transaction.atomic def record_attendance( channel, *, attendee_user=None, attendee_name: str = "", attendee_email: str = "", attendee_phone: str = "", organisation: str = "", notes: str = "", ): """Record one attendance row. Idempotent on (channel, dedupe_key). Returns ``(attendance, created)``. Raises :class:`AttendanceChannelClosedError` when the channel is inactive or outside its time window — the public view should map that to HTTP 410. """ from apps.mel.tracking.models import BeneficiaryAttendance if not _channel_is_open(channel): raise AttendanceChannelClosedError( "This attendance channel is closed." ) if channel.require_email and not attendee_email: raise ValidationError("Email is required for this event.") name = (attendee_name or "").strip() if attendee_user is not None and not name: name = attendee_user.get_full_name() or attendee_user.email or "" dedupe_key = _attendance_dedupe_key( activity_id=channel.activity_id, email=attendee_email, phone=attendee_phone, name=name, ) existing = BeneficiaryAttendance.objects.filter( channel=channel, dedupe_key=dedupe_key, ).first() if existing is not None: return existing, False attendance = BeneficiaryAttendance.objects.create( channel=channel, attendee_user=attendee_user, attendee_name=name, attendee_email=attendee_email.strip().lower(), attendee_phone=attendee_phone.strip(), organisation=organisation.strip(), notes=notes.strip(), dedupe_key=dedupe_key, ) logger.info( "mel.attendance recorded activity=%s channel=%s dedupe=%s", channel.activity_id, channel.pk, dedupe_key[:8], ) return attendance, True def attendance_csv_rows(activity) -> list[list[str]]: """Return a list-of-rows CSV payload for ``activity``'s attendance. First row is the header. Subsequent rows are one per :class:`BeneficiaryAttendance`, ordered by ``attended_at``. """ channel = getattr(activity, "attendance_channel", None) rows = [["attended_at", "name", "email", "phone", "organisation", "notes"]] if channel is None: return rows for att in channel.attendances.order_by("attended_at"): rows.append([ att.attended_at.isoformat(), att.attendee_name, att.attendee_email, att.attendee_phone, att.organisation, att.notes.replace("\n", " "), ]) return rows