from collections import Counter from datetime import timedelta from decimal import Decimal from django import forms from django.contrib import messages from django.core.exceptions import PermissionDenied from django.db.models import Count, Q, Sum from django.shortcuts import get_object_or_404, redirect from django.utils import timezone from django.views import View from django.views.generic import CreateView, DetailView, ListView, TemplateView, UpdateView from apps.core.lists import paginate_qs from apps.core.permissions.mixins import RoleRequiredMixin from apps.core.permissions.roles import UserRole from apps.mel.tracking.models import ( ACTIVITY_STATUS_TRANSITIONS, Activity, ActivityStatus, BaselineRecord, EmploymentSnapshot, ImpactScore, MELConfiguration, OutputDeliverable, OutputDeliverableStatus, Participant, ParticipantPersona, SMEHubBeneficiaryFeedback, SMEHubFeedHandlerLog, SMEHubTrackingRecord, TrackingLifecycle, TrackingMilestone, TrackingRecord, UpstreamEvent, ) class EventLogView(RoleRequiredMixin, ListView): allowed_roles = (UserRole.MEL_OFFICER, UserRole.ADMIN, UserRole.SYSTEM_ADMIN) model = UpstreamEvent context_object_name = "events" paginate_by = 10 template_name = "tracking/event_log.html" def get_queryset(self): qs = UpstreamEvent.objects.select_related("subject").order_by("-recorded_at") module = self.request.GET.get("module", "").strip() if module: qs = qs.filter(module=module) event_type = self.request.GET.get("event_type", "").strip() if event_type: qs = qs.filter(event_type__icontains=event_type) return qs def get_context_data(self, **kwargs): from datetime import timedelta from django.db.models import Count, Q from django.utils import timezone ctx = super().get_context_data(**kwargs) ctx["module_choices"] = UpstreamEvent.Module.choices # Module-volume strip — counts events from the last 7 days so officers # see flow rate at a glance, not lifetime totals. since = timezone.now() - timedelta(days=7) ctx["module_stats"] = UpstreamEvent.objects.filter( recorded_at__gte=since ).aggregate( rims=Count("pk", filter=Q(module=UpstreamEvent.Module.RIMS)), rep=Count("pk", filter=Q(module=UpstreamEvent.Module.REP)), smehub=Count("pk", filter=Q(module=UpstreamEvent.Module.SMEHUB)), alumni=Count("pk", filter=Q(module=UpstreamEvent.Module.ALUMNI)), repository=Count("pk", filter=Q(module=UpstreamEvent.Module.REPOSITORY)), ) ctx["selected_module"] = self.request.GET.get("module", "") return ctx class MilestoneListView(RoleRequiredMixin, ListView): """Full cross-module activity feed — the "View all" target for the dashboard's "Recent cross-module activity" preview. One row per TrackingMilestone.""" allowed_roles = (UserRole.MEL_OFFICER, UserRole.ADMIN, UserRole.SYSTEM_ADMIN) model = TrackingMilestone context_object_name = "milestones" paginate_by = 10 template_name = "tracking/milestone_list.html" def get_queryset(self): qs = TrackingMilestone.objects.select_related( "record__participant__user" ).order_by("-occurred_at") module = self.request.GET.get("module", "").strip() if module: qs = qs.filter(source_module=module) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["module_choices"] = TrackingRecord.Module.choices ctx["selected_module"] = self.request.GET.get("module", "") return ctx # --------------------------------------------------------------------------- # SME-Hub × M&EL portfolio (FRSME-MEI011, MEI013) # --------------------------------------------------------------------------- _SMEHUB_ROLES = ( UserRole.MEL_OFFICER, UserRole.SME_ADMIN, UserRole.ADMIN, UserRole.SYSTEM_ADMIN, ) class SMEHubPortfolioView(RoleRequiredMixin, ListView): """Aggregate dashboard with drill-down filters (cohort / sector / AIH / country / period) — FRSME-MEI011. """ allowed_roles = _SMEHUB_ROLES model = SMEHubTrackingRecord context_object_name = "records" paginate_by = 10 template_name = "tracking/smehub_portfolio.html" def get_queryset(self): qs = ( SMEHubTrackingRecord.objects.select_related( "entrepreneur__user", "business", "baseline_snapshot", ) .order_by("-last_milestone_at", "-created_at") ) gp = self.request.GET cohort_id = gp.get("cohort", "").strip() sector = gp.get("sector", "").strip() country = gp.get("country", "").strip() aih = gp.get("aih", "").strip() only_stalled = gp.get("stalled") == "1" if sector: qs = qs.filter(business__sector=sector) if country: qs = qs.filter(business__country=country) if cohort_id: qs = qs.filter( entrepreneur__cohort_memberships__cohort_id=cohort_id, ).distinct() if aih: qs = qs.filter(business__affiliations__institution_id=aih).distinct() if only_stalled: qs = qs.filter(is_stalled=True) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["stats"] = _portfolio_stats() ctx["filters"] = { "cohort": self.request.GET.get("cohort", ""), "sector": self.request.GET.get("sector", ""), "country": self.request.GET.get("country", ""), "aih": self.request.GET.get("aih", ""), "stalled": self.request.GET.get("stalled", ""), } ctx["sector_choices"] = _distinct_sector_choices() ctx["country_choices"] = _distinct_country_choices() ctx["cohort_choices"] = _cohort_choices() ctx["aih_choices"] = _aih_choices() return ctx class EmploymentSnapshotForm(forms.Form): """FRSME-MEI013 — manual employment update entered by an M&EL officer.""" fte_count = forms.IntegerField(min_value=0, label="Full-time employees") pte_count = forms.IntegerField(min_value=0, label="Part-time employees") note = forms.CharField(required=False, max_length=255, widget=forms.TextInput, label="Note (optional)") class SMEHubTrackingDetailView(RoleRequiredMixin, DetailView): """Drill into a single tracking record showing the full longitudinal trail.""" allowed_roles = _SMEHUB_ROLES model = SMEHubTrackingRecord context_object_name = "record" template_name = "tracking/smehub_tracking_detail.html" def get_context_data(self, **kwargs): from apps.mel.tracking.smehub_employment import compute_jobs_created ctx = super().get_context_data(**kwargs) record: SMEHubTrackingRecord = self.object # type: ignore[assignment] ctx["surveys"] = paginate_qs( self.request, SMEHubBeneficiaryFeedback.objects.filter(tracking_record=record) .select_related("feedback_channel", "feedback_submission") .order_by("-dispatched_at"), per_page=25, param="survey_page", ) ctx["employment_snapshots"] = paginate_qs( self.request, EmploymentSnapshot.objects.filter(tracking_record=record) .select_related("captured_by") .order_by("-captured_at"), per_page=25, param="snapshot_page", ) ctx["employment_form"] = EmploymentSnapshotForm() ctx["jobs_created"] = compute_jobs_created(record) baseline = record.baseline_snapshot or getattr(record.business, "baseline", None) ctx["baseline_total"] = ( (int(baseline.fte_count or 0) + int(baseline.pte_count or 0)) if baseline else 0 ) return ctx class RecordEmploymentSnapshotView(RoleRequiredMixin, View): """FRSME-MEI013 — POST handler creating a new employment snapshot row.""" allowed_roles = _SMEHUB_ROLES def post(self, request, pk: int): from apps.mel.tracking.smehub_employment import record_employment_snapshot record = get_object_or_404(SMEHubTrackingRecord, pk=pk) form = EmploymentSnapshotForm(request.POST) if not form.is_valid(): messages.error(request, "Provide whole-number FTE and PTE counts.") return redirect("mel_tracking:smehub_tracking_detail", pk=record.pk) record_employment_snapshot( record, fte_count=form.cleaned_data["fte_count"], pte_count=form.cleaned_data["pte_count"], by_user=request.user, note=form.cleaned_data.get("note", ""), ) messages.success(request, "Employment snapshot recorded.") return redirect("mel_tracking:smehub_tracking_detail", pk=record.pk) class SMEHubStalledView(SMEHubPortfolioView): """Convenience filter — pre-applies ``is_stalled=True`` (FRSME-MEI012).""" template_name = "tracking/smehub_portfolio.html" def get_queryset(self): return super().get_queryset().filter(is_stalled=True) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["stalled_only"] = True return ctx class SMEHubIndicatorDashboardView(RoleRequiredMixin, TemplateView): """FRSME-MEI011 — real-time SME-Hub indicator dashboard. Headline KPIs + a drill-down panel selectable by cohort / sector / AIH / country / month. Filters compose (sector + country narrows the AIH drill-down). JSON response (``?format=json``) supports incremental drill-down without a full page reload. """ allowed_roles = _SMEHUB_ROLES template_name = "tracking/smehub_indicator_dashboard.html" def _parse_int(self, value: str | None) -> int | None: if not value: return None try: return int(value) except (TypeError, ValueError): return None def _parse_date(self, value: str | None): if not value: return None from datetime import date try: return date.fromisoformat(value) except ValueError: return None def _filters_from_request(self) -> dict: gp = self.request.GET return { "cohort": self._parse_int(gp.get("cohort")), "sector": (gp.get("sector") or "").strip() or None, "aih": self._parse_int(gp.get("aih")), "country": (gp.get("country") or "").strip() or None, "period_start": self._parse_date(gp.get("period_start")), "period_end": self._parse_date(gp.get("period_end")), } def get(self, request, *args, **kwargs): from django.http import JsonResponse from apps.mel.tracking.services_dashboard import ( DIMENSIONS, filter_options, smehub_metrics_by_dimension, smehub_metrics_overview, ) filters = self._filters_from_request() dimension = (request.GET.get("dimension") or "cohort").strip() if dimension not in DIMENSIONS: dimension = "cohort" overview = smehub_metrics_overview(**filters) drilldown = smehub_metrics_by_dimension(dimension, **filters) if request.GET.get("format") == "json": return JsonResponse( { "overview": { **overview, # Decimals → strings for JSON safety. "total_funding_secured": str(overview["total_funding_secured"]), }, "dimension": dimension, "drilldown": [ {**row, "funding": str(row["funding"])} for row in drilldown ], } ) options = filter_options() context = self.get_context_data(**kwargs) context.update( { "overview": overview, "dimension": dimension, "drilldown": drilldown, "filters_raw": { "cohort": request.GET.get("cohort", ""), "sector": request.GET.get("sector", ""), "aih": request.GET.get("aih", ""), "country": request.GET.get("country", ""), "period_start": request.GET.get("period_start", ""), "period_end": request.GET.get("period_end", ""), }, "filter_options": options, "dimensions": [ ("cohort", "Cohort"), ("sector", "Sector"), ("aih", "AIH"), ("country", "Country"), ("month", "Month"), ], } ) return self.render_to_response(context) class MELConfigurationForm(forms.Form): """FRSME-MEI012 — tune the stall threshold without a deploy.""" smehub_stall_threshold_days = forms.IntegerField( min_value=1, max_value=365, label="Stall threshold (days)", help_text=( "Entrepreneurs with no milestone in this many days are flagged stalled " "on the next nightly sweep." ), ) class MELConfigurationUpdateView(RoleRequiredMixin, View): """FRSME-MEI012 — admin UI for tuning MEL configuration (stall threshold).""" allowed_roles = ( UserRole.SME_ADMIN, UserRole.ADMIN, UserRole.SYSTEM_ADMIN, UserRole.MEL_OFFICER, ) template_name = "tracking/mel_configuration_form.html" def get(self, request): from django.shortcuts import render config = MELConfiguration.get() form = MELConfigurationForm( initial={"smehub_stall_threshold_days": config.smehub_stall_threshold_days} ) return render(request, self.template_name, {"form": form, "config": config}) def post(self, request): from django.shortcuts import render config = MELConfiguration.get() form = MELConfigurationForm(request.POST) if not form.is_valid(): return render(request, self.template_name, {"form": form, "config": config}) config.smehub_stall_threshold_days = form.cleaned_data["smehub_stall_threshold_days"] config.updated_by = request.user config.save() messages.success( request, f"Stall threshold set to {config.smehub_stall_threshold_days} days.", ) return redirect("mel_tracking:mel_configuration") class SMEHubFeedbackOverviewView(RoleRequiredMixin, TemplateView): """FRSME-MEI018/019 admin overview — beneficiary survey completion rates, overdue counts, and the entrepreneurs with the most non-responses.""" allowed_roles = _SMEHUB_ROLES template_name = "tracking/smehub_feedback_overview.html" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) Stage = SMEHubBeneficiaryFeedback.Stage Status = SMEHubBeneficiaryFeedback.Status per_stage = [] for stage_value, stage_label in Stage.choices: stage_qs = SMEHubBeneficiaryFeedback.objects.filter(stage=stage_value) total = stage_qs.count() sent = stage_qs.filter(status=Status.SENT).count() completed = stage_qs.filter(status=Status.COMPLETED).count() non_responded = stage_qs.filter(status=Status.NON_RESPONDED).count() now = timezone.now() overdue = stage_qs.filter( status=Status.SENT, due_at__isnull=False, due_at__lt=now, ).count() completion_pct = (completed / total * 100) if total else 0 per_stage.append({ "stage": stage_value, "label": stage_label, "total": total, "sent": sent, "completed": completed, "non_responded": non_responded, "overdue": overdue, "completion_pct": round(completion_pct, 1), }) ctx["per_stage"] = per_stage ctx["total_dispatched"] = SMEHubBeneficiaryFeedback.objects.count() ctx["total_completed"] = SMEHubBeneficiaryFeedback.objects.filter( status=Status.COMPLETED, ).count() ctx["total_overdue"] = SMEHubBeneficiaryFeedback.objects.filter( status=Status.SENT, due_at__isnull=False, due_at__lt=timezone.now(), ).count() # Top 5 entrepreneurs with the highest non-response counts. ctx["top_non_responders"] = ( SMEHubTrackingRecord.objects.filter(feedback_non_response_count__gt=0) .select_related("entrepreneur__user", "business") .order_by("-feedback_non_response_count")[:5] ) # Latest 5 dispatched rows for context. ctx["recent_dispatches"] = ( SMEHubBeneficiaryFeedback.objects .select_related("tracking_record__entrepreneur__user", "feedback_channel") .order_by("-dispatched_at")[:5] ) return ctx # --------------------------------------------------------------------------- # FRSME-MEI007 — failed-feed review / resolve + reason-gated manual entry # --------------------------------------------------------------------------- class FeedFailureResolveForm(forms.Form): """Resolve a failed SME-Hub feed. A reason is always mandatory; when ``manual_entry`` is set the officer must also supply the bucket + label of the missing data point so it can be recorded on the tracking record.""" reason = forms.CharField( widget=forms.Textarea(attrs={"rows": 3}), label="Reason (recorded in the audit trail)", ) manual_entry = forms.BooleanField( required=False, label="I am manually entering the missing data point", ) bucket = forms.ChoiceField( required=False, choices=[("", "—")] + list(SMEHubTrackingRecord.MANUAL_ENTRY_BUCKETS), label="Milestone stage", ) entry_key = forms.CharField( required=False, max_length=80, label="Data point", help_text="Short machine-friendly label, e.g. funding_secured.", ) note = forms.CharField( required=False, max_length=255, widget=forms.TextInput, label="Detail (optional)", ) amount = forms.DecimalField( required=False, min_value=0, max_digits=16, decimal_places=2, label="Amount (funding only)", help_text="For a funding_secured repair — the value to add to the funding KPI.", ) def clean_reason(self): reason = (self.cleaned_data.get("reason") or "").strip() if not reason: raise forms.ValidationError("A reason is required to resolve a failed feed.") return reason def clean(self): cleaned = super().clean() if cleaned.get("manual_entry"): if not cleaned.get("bucket"): self.add_error("bucket", "Pick the milestone stage for the manual entry.") if not (cleaned.get("entry_key") or "").strip(): self.add_error("entry_key", "Name the data point you are entering.") return cleaned class SMEHubFeedFailureListView(RoleRequiredMixin, ListView): """FRSME-MEI007 — unresolved (or all) failed SME-Hub → M&EL data feeds, flagged for officer/admin review and manual repair.""" allowed_roles = _SMEHUB_ROLES model = SMEHubFeedHandlerLog context_object_name = "failures" paginate_by = 20 template_name = "tracking/smehub_feed_failures.html" def get_queryset(self): qs = SMEHubFeedHandlerLog.objects.select_related( "entrepreneur__user", "resolved_by" ) status = self.request.GET.get("status", "open").strip() if status == "resolved": qs = qs.filter(resolved_at__isnull=False) elif status == "all": pass else: # default: open status = "open" qs = qs.filter(resolved_at__isnull=True) signal = self.request.GET.get("signal", "").strip() if signal: qs = qs.filter(signal_name=signal) self._status = status return qs.order_by("-created_at") def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["status_filter"] = getattr(self, "_status", "open") ctx["selected_signal"] = self.request.GET.get("signal", "") ctx["open_count"] = SMEHubFeedHandlerLog.objects.filter( resolved_at__isnull=True ).count() ctx["resolved_count"] = SMEHubFeedHandlerLog.objects.filter( resolved_at__isnull=False ).count() ctx["manual_count"] = SMEHubFeedHandlerLog.objects.filter( manual_entry=True ).count() ctx["signal_choices"] = list( SMEHubFeedHandlerLog.objects.values_list("signal_name", flat=True) .distinct() .order_by("signal_name") ) ctx["resolve_form"] = FeedFailureResolveForm() return ctx class SMEHubFeedFailureResolveView(RoleRequiredMixin, View): """FRSME-MEI007 — POST: mark a failed feed resolved with a mandatory reason, optionally recording the missing data point as a MANUAL entry against the entrepreneur's tracking record. Both paths write to the audit trail.""" allowed_roles = _SMEHUB_ROLES def post(self, request, pk: int): from apps.core.audit.models import AuditLog from apps.core.audit.mixins import log_audit log = get_object_or_404(SMEHubFeedHandlerLog, pk=pk) redirect_to = request.POST.get("next") or reverse( "mel_tracking:smehub_feed_failures" ) if log.resolved_at is not None: messages.info(request, "This feed failure is already resolved.") return HttpResponseRedirect(redirect_to) form = FeedFailureResolveForm(request.POST) if not form.is_valid(): first_error = next(iter(form.errors.values()))[0] messages.error(request, first_error) return HttpResponseRedirect(redirect_to) cd = form.cleaned_data reason = cd["reason"] do_manual = bool(cd.get("manual_entry")) manual_recorded = False if do_manual: if log.entrepreneur_id is None: messages.error( request, "This failure has no linked entrepreneur — the data point cannot " "be entered manually. Resolve it as an acknowledgement instead.", ) return HttpResponseRedirect(redirect_to) record, _ = SMEHubTrackingRecord.objects.get_or_create( entrepreneur=log.entrepreneur ) record.record_manual_entry( bucket=cd["bucket"], key=cd["entry_key"].strip(), note=cd.get("note", ""), reason=reason, by_user=request.user, amount=cd.get("amount"), ) manual_recorded = True log.resolved_at = timezone.now() log.resolved_by = request.user log.resolution_note = reason log.manual_entry = manual_recorded log.save( update_fields=[ "resolved_at", "resolved_by", "resolution_note", "manual_entry", ] ) # Audit trail — mirror the indicators manual-entry / target-revision # pattern. A manual data entry is a CREATE; a plain acknowledgement is # an UPDATE of the failure row. log_audit( actor=request.user, action=AuditLog.Action.CREATE if manual_recorded else AuditLog.Action.UPDATE, target_app="mel_tracking", target_model="SMEHubFeedHandlerLog", object_id=log.pk, object_repr=str(log), changes={ "feed_failure_resolved": True, "signal_name": log.signal_name, "manual_entry": manual_recorded, "bucket": cd.get("bucket") if manual_recorded else None, "data_point": cd.get("entry_key") if manual_recorded else None, "amount": str(cd["amount"]) if (manual_recorded and cd.get("amount") is not None) else None, "entrepreneur_id": log.entrepreneur_id, "reason": reason, }, ) applied_funding = bool( manual_recorded and (cd.get("entry_key") or "").strip() == "funding_secured" and cd.get("amount") ) if applied_funding: messages.success( request, "Failed feed resolved, the missing data point recorded manually, and " f"the funding total credited with {cd['amount']}.", ) elif manual_recorded: messages.success( request, "Failed feed resolved and the missing data point recorded manually " "against the tracking record.", ) else: messages.success(request, "Failed feed resolved.") return HttpResponseRedirect(redirect_to) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _portfolio_stats() -> dict: """Compute the 6 SME-Hub × M&EL stat-strip values (FRSME-MEI013).""" from apps.smehub.onboarding.models import Business, EntrepreneurProfile verified_business_count = Business.objects.filter( verification_status=Business.Status.VERIFIED, ).count() incubated_count = ( EntrepreneurProfile.objects.filter( verification_status=EntrepreneurProfile.Status.VERIFIED, cohort_memberships__status="active", ) .distinct() .count() ) verified_entrepreneur_count = EntrepreneurProfile.objects.filter( verification_status=EntrepreneurProfile.Status.VERIFIED, ).count() incubated_pct = ( round((incubated_count / verified_entrepreneur_count) * 100, 1) if verified_entrepreneur_count else 0 ) partnerships = 0 deals_closed = 0 funding_total = SMEHubTrackingRecord.objects.aggregate( total=Sum("funding_secured_total") )["total"] or Decimal("0") try: from apps.smehub.linkage.models import MoU from apps.smehub.marketplace.models import DemandResponse partnerships = ( MoU.objects.filter(status__in=[MoU.Status.FORMALISED, MoU.Status.AMENDED]).count() + DemandResponse.objects.filter(status=DemandResponse.Status.AWARDED).count() ) except ImportError: # pragma: no cover pass try: from apps.smehub.showcasing.models import DealAgreement deals_closed = DealAgreement.objects.count() except ImportError: # pragma: no cover pass stalled_count = SMEHubTrackingRecord.objects.filter(is_stalled=True).count() total_records = SMEHubTrackingRecord.objects.count() from apps.mel.tracking.smehub_employment import compute_total_jobs_created jobs_created = compute_total_jobs_created() return { "businesses_registered": verified_business_count, "percentage_incubated": incubated_pct, "partnerships_formed": partnerships, "deals_closed": deals_closed, "funding_secured_total": funding_total, "jobs_created": jobs_created, "stalled_count": stalled_count, "total_records": total_records, } def _distinct_sector_choices() -> list[tuple[str, str]]: from apps.smehub.onboarding.models import Business qs = ( Business.objects.filter(verification_status=Business.Status.VERIFIED) .exclude(sector="") .values_list("sector", flat=True) .distinct() .order_by("sector") ) return [(s, s) for s in qs] def _distinct_country_choices() -> list[tuple[str, str]]: from apps.smehub.onboarding.models import Business qs = ( Business.objects.filter(verification_status=Business.Status.VERIFIED) .exclude(country="") .values_list("country", flat=True) .distinct() .order_by("country") ) return [(c, c) for c in qs] def _cohort_choices() -> list[tuple[int, str]]: try: from apps.smehub.incubation.models import Cohort except ImportError: return [] return [ (c.pk, f"{c.programme.title} — {c.name}") for c in Cohort.objects.select_related("programme").order_by("-created_at")[:50] ] def _aih_choices() -> list[tuple[str, str]]: try: from apps.core.authentication.models import Institution except ImportError: return [] return [ (str(inst.pk), inst.name) for inst in Institution.objects.order_by("name")[:50] ] def _distinct_period_choices() -> list[tuple[str, str]]: """G19 — distinct DataPoint.period_label values, newest first.""" from apps.mel.indicators.models import DataPoint labels = ( DataPoint.objects.exclude(period_label="") .values_list("period_label", flat=True) .distinct() .order_by("-period_label")[:24] ) return [(label, label) for label in labels] # --------------------------------------------------------------------------- # Activity & OutputDeliverable CRUD (FRMFL009-019) # --------------------------------------------------------------------------- from django.core.exceptions import ValidationError # noqa: E402 from django.http import HttpResponseRedirect # noqa: E402 from django.urls import reverse, reverse_lazy # noqa: E402 from apps.mel.indicators.models import LogFrame # noqa: E402 from apps.mel.tracking.forms import ( # noqa: E402 ActivityForm, ActivityTransitionForm, CorrectiveActionCreateForm, OutputDeliverableForm, RecordOutputProgressForm, ) from apps.mel.tracking.services import ( # noqa: E402 record_activity_progress, record_output_progress, ) OFFICER_ROLES = (UserRole.MEL_OFFICER, UserRole.ADMIN, UserRole.SYSTEM_ADMIN) VIEWER_ROLES = OFFICER_ROLES + ( UserRole.PROGRAM_MANAGER, UserRole.PROGRAM_DIRECTOR, UserRole.GRANTS_MANAGER, ) class ActivityListView(RoleRequiredMixin, ListView): allowed_roles = VIEWER_ROLES model = Activity context_object_name = "activities" paginate_by = 10 template_name = "tracking/activity_list.html" def get_queryset(self): qs = ( Activity.objects.select_related("logframe_row__logframe", "responsible") .order_by("scheduled_start", "name") ) status = self.request.GET.get("status", "").strip() if status: qs = qs.filter(status=status) logframe = self.request.GET.get("logframe", "").strip() if logframe: qs = qs.filter(logframe_row__logframe__slug=logframe) if self.request.GET.get("delayed") == "1": qs = qs.filter(deviation_flag=True) return qs def get_context_data(self, **kwargs): from django.db.models import Count, Q ctx = super().get_context_data(**kwargs) ctx["status_choices"] = ActivityStatus.choices ctx["logframe_options"] = LogFrame.objects.order_by("name") ctx["selected_status"] = self.request.GET.get("status", "") ctx["selected_logframe"] = self.request.GET.get("logframe", "") ctx["selected_delayed"] = self.request.GET.get("delayed", "") # Triage strip — counts active activities by status. Cancelled and # completed don't get a card; "delayed" is the prominent one. ctx["status_stats"] = Activity.objects.aggregate( planned=Count("pk", filter=Q(status=ActivityStatus.PLANNED)), in_progress=Count("pk", filter=Q(status=ActivityStatus.IN_PROGRESS)), delayed=Count("pk", filter=Q(status=ActivityStatus.DELAYED)), completed=Count("pk", filter=Q(status=ActivityStatus.COMPLETED)), ) return ctx class ActivityCreateView(RoleRequiredMixin, CreateView): allowed_roles = OFFICER_ROLES model = Activity form_class = ActivityForm template_name = "tracking/activity_form.html" def get_initial(self): initial = super().get_initial() row_id = self.request.GET.get("logframe_row") if row_id: initial["logframe_row"] = row_id # M&E SRS Table 65 (A1/E1) — "Schedule field verification" / follow-up # monitoring hand-offs arrive with the activity pre-described. name = self.request.GET.get("name", "").strip() if name: initial["name"] = name[:255] description = self.request.GET.get("description", "").strip() if description: initial["description"] = description return initial def get_success_url(self): return reverse_lazy("mel_tracking:activity_detail", kwargs={"pk": self.object.pk}) class ActivityUpdateView(RoleRequiredMixin, UpdateView): allowed_roles = OFFICER_ROLES model = Activity form_class = ActivityForm template_name = "tracking/activity_form.html" def get_success_url(self): return reverse_lazy("mel_tracking:activity_detail", kwargs={"pk": self.object.pk}) class ActivityDetailView(RoleRequiredMixin, DetailView): allowed_roles = VIEWER_ROLES model = Activity context_object_name = "activity" template_name = "tracking/activity_detail.html" def get_queryset(self): return ( Activity.objects.select_related("logframe_row__logframe", "responsible", "parent") ) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) activity: Activity = ctx["activity"] labels = dict(ActivityStatus.choices) legal = sorted(ACTIVITY_STATUS_TRANSITIONS.get(activity.status, set())) ctx["legal_transitions"] = [ {"value": s, "label": labels.get(s, s)} for s in legal ] ctx["dependencies_in"] = activity.incoming_dependencies.select_related("from_activity") ctx["dependencies_out"] = activity.outgoing_dependencies.select_related("to_activity") ctx["history"] = activity.history.select_related("history_user")[:5] # G21 — polymorphic evidence gallery context. from django.contrib.contenttypes.models import ContentType from apps.mel.indicators.models import EvidenceAttachment ct = ContentType.objects.get_for_model(Activity) ctx["attachment_ct_id"] = ct.pk ctx["attachments"] = list( EvidenceAttachment.objects.filter(content_type=ct, object_id=activity.pk) ) # M&E SRS Table 65 — follow-up monitoring for incomplete collection. from apps.mel.tracking.forms import FollowupMonitoringForm ctx["is_incomplete"] = activity.status == ActivityStatus.INCOMPLETE ctx["followup_form"] = FollowupMonitoringForm() ctx["followups"] = list( activity.children.select_related("responsible").order_by("scheduled_start") ) return ctx class ActivityTransitionView(RoleRequiredMixin, View): """POST-only — transition Activity status via the FSM service.""" allowed_roles = OFFICER_ROLES def post(self, request, pk: int): activity = get_object_or_404(Activity, pk=pk) form = ActivityTransitionForm(request.POST) if not form.is_valid(): messages.error(request, "Pick a valid transition.") return HttpResponseRedirect( reverse("mel_tracking:activity_detail", kwargs={"pk": activity.pk}) ) cd = form.cleaned_data try: record_activity_progress( activity.pk, new_status=cd["new_status"], user=request.user, actual_start=cd.get("actual_start"), actual_end=cd.get("actual_end"), deviation_reason=cd.get("deviation_reason", ""), ) except ValidationError as exc: messages.error(request, str(exc.message if hasattr(exc, "message") else exc)) except Activity.DoesNotExist: raise else: messages.success( request, f"Activity transitioned to {dict(ActivityStatus.choices)[cd['new_status']]}.", ) return HttpResponseRedirect( reverse("mel_tracking:activity_detail", kwargs={"pk": activity.pk}) ) class ActivityScheduleFollowupView(RoleRequiredMixin, View): """M&E SRS Table 65 — schedule follow-up monitoring for an incomplete activity.""" allowed_roles = OFFICER_ROLES def post(self, request, pk: int): from apps.mel.tracking.forms import FollowupMonitoringForm from apps.mel.tracking.services import schedule_followup_monitoring activity = get_object_or_404(Activity, pk=pk) if activity.status != ActivityStatus.INCOMPLETE: messages.error( request, "Follow-up monitoring can only be scheduled for an incomplete activity.", ) return HttpResponseRedirect( reverse("mel_tracking:activity_detail", kwargs={"pk": activity.pk}) ) form = FollowupMonitoringForm(request.POST) if not form.is_valid(): messages.error(request, "Enter valid follow-up dates.") return HttpResponseRedirect( reverse("mel_tracking:activity_detail", kwargs={"pk": activity.pk}) ) cd = form.cleaned_data followup = schedule_followup_monitoring( activity, scheduled_start=cd["scheduled_start"], scheduled_end=cd["scheduled_end"], name=cd.get("name", ""), user=request.user, ) messages.success( request, f"Follow-up monitoring “{followup.name}” scheduled for " f"{followup.scheduled_start:%d %b %Y}.", ) return HttpResponseRedirect( reverse("mel_tracking:activity_detail", kwargs={"pk": followup.pk}) ) class OutputDeliverableListView(RoleRequiredMixin, ListView): allowed_roles = VIEWER_ROLES model = OutputDeliverable context_object_name = "deliverables" paginate_by = 10 template_name = "tracking/output_deliverable_list.html" def get_queryset(self): qs = ( OutputDeliverable.objects.select_related("logframe_row__logframe") .order_by("due_date", "title") ) status = self.request.GET.get("status", "").strip() if status: qs = qs.filter(status=status) logframe = self.request.GET.get("logframe", "").strip() if logframe: qs = qs.filter(logframe_row__logframe__slug=logframe) if self.request.GET.get("overdue") == "1": qs = qs.filter(status=OutputDeliverableStatus.OVERDUE) q = self.request.GET.get("q", "").strip() if q: qs = qs.filter(title__icontains=q) return qs def get_context_data(self, **kwargs): from django.db.models import Count, Q ctx = super().get_context_data(**kwargs) ctx["status_choices"] = OutputDeliverableStatus.choices ctx["logframe_options"] = LogFrame.objects.order_by("name") ctx["selected_status"] = self.request.GET.get("status", "") ctx["selected_logframe"] = self.request.GET.get("logframe", "") ctx["selected_overdue"] = self.request.GET.get("overdue", "") ctx["q"] = self.request.GET.get("q", "") ctx["status_stats"] = OutputDeliverable.objects.aggregate( draft=Count("pk", filter=Q(status=OutputDeliverableStatus.DRAFT)), in_progress=Count("pk", filter=Q(status=OutputDeliverableStatus.IN_PROGRESS)), completed=Count("pk", filter=Q(status=OutputDeliverableStatus.COMPLETED)), overdue=Count("pk", filter=Q(status=OutputDeliverableStatus.OVERDUE)), ) return ctx class OutputDeliverableCreateView(RoleRequiredMixin, CreateView): allowed_roles = OFFICER_ROLES model = OutputDeliverable form_class = OutputDeliverableForm template_name = "tracking/output_deliverable_form.html" def get_initial(self): initial = super().get_initial() row_id = self.request.GET.get("logframe_row") if row_id: initial["logframe_row"] = row_id return initial def get_success_url(self): return reverse_lazy("mel_tracking:output_detail", kwargs={"pk": self.object.pk}) class OutputDeliverableUpdateView(RoleRequiredMixin, UpdateView): allowed_roles = OFFICER_ROLES model = OutputDeliverable form_class = OutputDeliverableForm template_name = "tracking/output_deliverable_form.html" def get_success_url(self): return reverse_lazy("mel_tracking:output_detail", kwargs={"pk": self.object.pk}) class OutputDeliverableDetailView(RoleRequiredMixin, DetailView): allowed_roles = VIEWER_ROLES model = OutputDeliverable context_object_name = "deliverable" template_name = "tracking/output_deliverable_detail.html" def get_queryset(self): return OutputDeliverable.objects.select_related("logframe_row__logframe") def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) deliverable: OutputDeliverable = ctx["deliverable"] ctx["progress_form"] = RecordOutputProgressForm( initial={ "achieved": deliverable.achieved_quantity, "evidence_url": deliverable.evidence_url, } ) ctx["percent"] = deliverable.percent_achieved ctx["status_labels"] = dict(OutputDeliverableStatus.choices) ctx["history"] = deliverable.history.select_related("history_user")[:5] # Linked activities under sibling ACTIVITY rows whose parent is this output. ctx["linked_activities"] = paginate_qs( self.request, Activity.objects.filter( logframe_row__parent_id=deliverable.logframe_row_id, ) .select_related("logframe_row") .order_by("scheduled_end"), per_page=25, param="act_page", ) # G21 — polymorphic evidence gallery context. from django.contrib.contenttypes.models import ContentType from apps.mel.indicators.models import EvidenceAttachment ct = ContentType.objects.get_for_model(OutputDeliverable) ctx["attachment_ct_id"] = ct.pk ctx["attachments"] = list( EvidenceAttachment.objects.filter(content_type=ct, object_id=deliverable.pk) ) return ctx class _RevisionHistoryView(RoleRequiredMixin, View): """Read-only full revision trail with per-field diffs (defect #41). Renders the tracking-owned ``tracking/revision_history.html`` template. Each UPDATE revision carries a simple_history ``diff_against`` delta so the trail shows *what* changed (field: old → new), not just who/when. Subclasses supply the subject + its detail URL/label. """ template_name = "tracking/revision_history.html" subject_kind: str = "" _DIFF_EXCLUDED_FIELDS = ("updated_at", "created_at") def get_subject(self): raise NotImplementedError def get_subject_url(self): raise NotImplementedError def get_subject_label(self) -> str: raise NotImplementedError @staticmethod def _fmt_value(val) -> str: if val is None or val == "": return "—" text = str(val) return text if len(text) <= 80 else text[:77] + "…" def _changes_for(self, entry) -> list[dict]: """[{field, old, new}] for an UPDATE via simple_history's diff_against.""" prev = getattr(entry, "prev_record", None) if prev is None: return [] try: delta = entry.diff_against(prev, excluded_fields=self._DIFF_EXCLUDED_FIELDS) except Exception: # pragma: no cover — defensive against schema drift return [] return [ { "field": change.field, "old": self._fmt_value(change.old), "new": self._fmt_value(change.new), } for change in delta.changes ] def get(self, request, *args, **kwargs): from django.shortcuts import render subject = self.get_subject() history_qs = subject.history.select_related("history_user").order_by( "-history_date" ) rows = [ { "h": h, "changes": self._changes_for(h) if h.history_type == "~" else [], } for h in history_qs ] return render( request, self.template_name, { "subject_kind": self.subject_kind, "subject_label": self.get_subject_label(), "subject_url": self.get_subject_url(), # Keep the queryset (not a list) so callers/tests can .count() # and the template still iterates it directly if needed. "history_records": history_qs, "history_rows": rows, }, ) class ActivityHistoryView(_RevisionHistoryView): allowed_roles = VIEWER_ROLES subject_kind = "Activity" def get_subject(self): self._obj = get_object_or_404(Activity, pk=self.kwargs["pk"]) return self._obj def get_subject_url(self): return reverse("mel_tracking:activity_detail", kwargs={"pk": self._obj.pk}) def get_subject_label(self) -> str: return self._obj.name class OutputDeliverableHistoryView(_RevisionHistoryView): allowed_roles = VIEWER_ROLES subject_kind = "Output deliverable" def get_subject(self): self._obj = get_object_or_404(OutputDeliverable, pk=self.kwargs["pk"]) return self._obj def get_subject_url(self): return reverse("mel_tracking:output_detail", kwargs={"pk": self._obj.pk}) def get_subject_label(self) -> str: return self._obj.title class RecordOutputProgressView(RoleRequiredMixin, View): """POST-only — call record_output_progress and re-derive status.""" allowed_roles = OFFICER_ROLES def post(self, request, pk: int): deliverable = get_object_or_404(OutputDeliverable, pk=pk) form = RecordOutputProgressForm(request.POST) if not form.is_valid(): messages.error(request, "Provide a non-negative achieved quantity.") return HttpResponseRedirect( reverse("mel_tracking:output_detail", kwargs={"pk": deliverable.pk}) ) cd = form.cleaned_data record_output_progress( deliverable.pk, achieved=cd["achieved"], evidence_url=(cd.get("evidence_url") or None), ) messages.success(request, "Output progress recorded.") return HttpResponseRedirect( reverse("mel_tracking:output_detail", kwargs={"pk": deliverable.pk}) ) # --------------------------------------------------------------------------- # Phase 4.5 — Cross-module participant tracking dashboard # --------------------------------------------------------------------------- _PARTICIPANT_ROLES = ( UserRole.MEL_OFFICER, UserRole.ADMIN, UserRole.SYSTEM_ADMIN, ) # Read-only viewer tier for the cross-module M&EL dashboard (defect #16 / E-1). # The topnav "Dashboard" entry is shown to the programme-leadership roles, so the # dashboard VIEW must admit them too. They get the read-only dashboard only — # write surfaces (data entry, config, corrective actions) stay gated to the # officer/admin tiers. ``VIEWER_ROLES`` (defined above, used by Activity views) # already encodes officer + PM/PD/GM, so reuse it. Matches the MEI011 actor set. _DASHBOARD_VIEWER_ROLES = VIEWER_ROLES def _participant_directory_context(request): """Build context for the participant-directory table + filter form. Extracted so both the dashboard "preview" (top 10) and the dedicated directory page can share the same filter semantics. """ gp = request.GET persona = (gp.get("persona") or "").strip() module = (gp.get("module") or "").strip() status = (gp.get("status") or "").strip() search = (gp.get("q") or "").strip() participants = ( Participant.objects.select_related( "user", "scholarship", "entrepreneur_profile", "alumni_profile", ) .prefetch_related("tracking_records") ) if persona: participants = participants.filter(personas__contains=[persona]) if search: participants = participants.filter( Q(user__email__icontains=search) | Q(user__first_name__icontains=search) | Q(user__last_name__icontains=search) ) # Module filter cuts via the related TrackingRecord — distinct() because the join multiplies rows. if module: participants = participants.filter(tracking_records__source_module=module).distinct() if status: participants = participants.filter(tracking_records__status=status).distinct() return { "participants": participants.order_by("-updated_at"), "filters": {"persona": persona, "module": module, "status": status, "q": search}, "persona_choices": ParticipantPersona.choices, "module_choices": TrackingRecord.Module.choices, "status_choices": TrackingLifecycle.choices, } class CrossModuleTrackingDashboardView(RoleRequiredMixin, TemplateView): """M&EL Dashboard — monitoring health, evaluation queue & cross-module reach. Leads with RAG status, action queue (corrective actions / pending reports / late activities), and attention items (urgent feedback / recent data points). Cross-module reach (Participants × Module) and the participant preview live further down the page. Full participant directory moved to :class:`ParticipantDirectoryView` at /mel/events/dashboard/participants/. """ allowed_roles = _DASHBOARD_VIEWER_ROLES template_name = "tracking/tracking_dashboard.html" def _parse_filters(self) -> dict: """G19 — read filter params from query string. ``cohort``, ``aih`` are integer FKs (Cohort, Institution). ``sector``, ``country`` are CharField string matches against Business. ``period`` is a DataPoint.period_label (e.g. "2026Q2", "2026-04"). Empty / unset values fall through (no filter applied). """ gp = self.request.GET def _int_or_none(value: str) -> int | None: try: return int(value) except (TypeError, ValueError): return None return { # M&E SRS Tables 65–66 — the framework's primary navigation axis is # Results Framework (programme) + reporting period. "rf": (gp.get("rf") or "").strip(), "cohort_id": _int_or_none(gp.get("cohort", "")), "sector": (gp.get("sector") or "").strip(), "country": (gp.get("country") or "").strip(), "aih_id": _int_or_none(gp.get("aih", "")), "period": (gp.get("period") or "").strip(), } def _filtered_indicators(self, filters: dict): """Filter active Indicators via their logframe's programme/cohort. v1: cohort/AIH filters narrow via SMEHubTrackingRecord linkage when present; otherwise the indicator passes through. Sector/country likewise rely on SMEHub Business denormalisation. period is applied at the DataPoint join inside calculate_progress, not here. """ from apps.mel.indicators.models import Indicator qs = Indicator.objects.filter(is_active=True).select_related("logframe_row__logframe") # M&E SRS — scope to a single Results Framework (programme) when chosen. if filters.get("rf"): qs = qs.filter(logframe_row__logframe__slug=filters["rf"]) # No direct FK from Indicator to cohort/business. The most common path # is via SMEHubTrackingRecord — narrowed when any of those filters set. # To avoid silently zeroing the dashboard, scope the join only when at # least one of the participant-rooted filters is set. if any([filters["cohort_id"], filters["sector"], filters["country"], filters["aih_id"]]): from apps.mel.tracking.models import SMEHubTrackingRecord recs = SMEHubTrackingRecord.objects.all() if filters["sector"]: recs = recs.filter(business__sector=filters["sector"]) if filters["country"]: recs = recs.filter(business__country=filters["country"]) if filters["cohort_id"]: recs = recs.filter(entrepreneur__cohort_memberships__cohort_id=filters["cohort_id"]).distinct() if filters["aih_id"]: recs = recs.filter(business__affiliations__institution_id=filters["aih_id"]).distinct() programme_titles = set() for r in recs.values_list("entrepreneur__cohort_memberships__cohort__programme__title", flat=True): if r: programme_titles.add(r) if programme_titles: qs = qs.filter(logframe_row__logframe__programme__in=programme_titles) else: qs = qs.none() return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) # Officer/admin tier may act on the dashboard (record data, open corrective # actions, edit config); the widened viewer tier (PM/PD/GM, defect #16) sees # a read-only view, so write CTAs are hidden for them. user = self.request.user ctx["can_manage_mel"] = bool( getattr(user, "is_superuser", False) or getattr(user, "role", "") in _PARTICIPANT_ROLES ) filters = self._parse_filters() ctx["filters"] = { "rf": self.request.GET.get("rf", ""), "cohort": self.request.GET.get("cohort", ""), "sector": self.request.GET.get("sector", ""), "country": self.request.GET.get("country", ""), "aih": self.request.GET.get("aih", ""), "period": self.request.GET.get("period", ""), } # M&E SRS — Results Framework (programme) scope selector. Active frames # first, then closed, so the primary axis leads the filter bar. from apps.mel.indicators.models import ( LogFrame as _LFchoice, LogFrameStatus as _LFSchoice, ) ctx["rf_choices"] = list( _LFchoice.objects.order_by("-status", "name").values_list("slug", "name") ) selected_rf = None if filters["rf"]: selected_rf = _LFchoice.objects.filter(slug=filters["rf"]).first() ctx["selected_rf"] = selected_rf ctx["sector_choices"] = _distinct_sector_choices() ctx["country_choices"] = _distinct_country_choices() ctx["cohort_choices"] = _cohort_choices() ctx["aih_choices"] = _aih_choices() ctx["period_choices"] = _distinct_period_choices() any_filter_active = any([ filters["rf"], filters["cohort_id"], filters["sector"], filters["country"], filters["aih_id"], filters["period"], ]) ctx["any_filter_active"] = any_filter_active # ── RAG health ────────────────────────────────────────────── from apps.mel.indicators.models import RagStatus from apps.mel.indicators.services import calculate_progress, current_period_label rag_counts = {"green": 0, "amber": 0, "red": 0, "unknown": 0} off_track_red: list[dict] = [] off_track_amber: list[dict] = [] # M&E framework — aggregate indicator performance up the results chain # into the three reportable levels: Output → Outcome → Impact. Activity / # Input indicators roll into Output, so every indicator lands in exactly # one reporting band. _LEVEL_BAND = { "output": "output", "activity": "output", "input": "output", "outcome": "outcome", "impact": "impact", } level_band = { band: {"count": 0, "green": 0, "amber": 0, "red": 0, "unknown": 0, "pct_sum": 0.0, "pct_n": 0} for band in ("output", "outcome", "impact") } active_indicators = self._filtered_indicators(filters) # ── Per-Strategic-Objective scorecards ────────────────────── # After the results-framework refactor the Strategic Objective (LogFrame) # IS the unit of planning, so the dashboard leads with a card per active # SO. Seed a blank card for every active SO (so objectives with no data # still appear), then accumulate each indicator's RAG + attainment into # its SO, bucketed by reporting band (Output/Outcome/Impact). from apps.mel.indicators.models import ( LogFrame as _LFcard, LogFrameStatus as _LFScard, ) _so_active_lfs = list( _LFcard.objects.filter(status=_LFScard.ACTIVE) .select_related("finance_budget") .order_by("name") ) if selected_rf is not None: _so_active_lfs = [lf for lf in _so_active_lfs if lf.pk == selected_rf.pk] def _blank_so_band(): return {"count": 0, "green": 0, "amber": 0, "red": 0, "unknown": 0, "pct_sum": 0.0, "pct_n": 0} so_cards = { lf.pk: { "logframe": lf, "bands": {b: _blank_so_band() for b in ("output", "outcome", "impact")}, "indicator_count": 0, "rag": {"green": 0, "amber": 0, "red": 0, "unknown": 0}, "pct_sum": 0.0, "pct_n": 0, } for lf in _so_active_lfs } for ind in active_indicators: period_label = filters["period"] or current_period_label(ind.frequency) progress = calculate_progress(ind, period_label) rag_counts[progress["rag"]] = rag_counts.get(progress["rag"], 0) + 1 band = _LEVEL_BAND.get(getattr(ind.logframe_row, "level", ""), None) if band: b = level_band[band] b["count"] += 1 b[progress["rag"]] = b.get(progress["rag"], 0) + 1 if progress["percent"] is not None: b["pct_sum"] += min(float(progress["percent"]), 150.0) b["pct_n"] += 1 # Accumulate the same indicator into its Strategic Objective card. _card = so_cards.get(getattr(ind.logframe_row, "logframe_id", None)) if _card is not None: _card["indicator_count"] += 1 _card["rag"][progress["rag"]] = _card["rag"].get(progress["rag"], 0) + 1 if progress["percent"] is not None: _card["pct_sum"] += min(float(progress["percent"]), 150.0) _card["pct_n"] += 1 if band: cb = _card["bands"][band] cb["count"] += 1 cb[progress["rag"]] = cb.get(progress["rag"], 0) + 1 if progress["percent"] is not None: cb["pct_sum"] += min(float(progress["percent"]), 150.0) cb["pct_n"] += 1 if progress["rag"] == RagStatus.RED: progress["indicator"] = ind off_track_red.append(progress) elif progress["rag"] == RagStatus.AMBER: progress["indicator"] = ind off_track_amber.append(progress) ctx["rag_counts"] = rag_counts ctx["rag_total"] = sum(rag_counts.values()) ctx["off_track_indicators"] = (off_track_red + off_track_amber)[:5] # Finalise the results-chain band: mean attainment + a band-level rating. results_chain = [] for key, title in (("output", "Outputs"), ("outcome", "Outcomes"), ("impact", "Impact")): b = level_band[key] avg = round(b["pct_sum"] / b["pct_n"], 1) if b["pct_n"] else None if avg is None: band_rag = "unknown" elif avg >= 85: band_rag = "green" elif avg >= 60: band_rag = "amber" else: band_rag = "red" results_chain.append({ "key": key, "title": title, "count": b["count"], "avg_attainment": avg, "rag": band_rag, "green": b["green"], "amber": b["amber"], "red": b["red"], "unknown": b["unknown"], }) ctx["results_chain"] = results_chain # ── G13 — In-flight alert escalations chip ────────────────── from apps.mel.indicators.models import AlertEscalation ctx["active_escalations_count"] = AlertEscalation.objects.filter( acknowledged_at__isnull=True, ).count() ctx["tier3_escalations_count"] = AlertEscalation.objects.filter( acknowledged_at__isnull=True, tier_reached=3, ).count() # ── G4 — Programme health hero card + top programmes widget ─ from apps.mel.indicators.models import ( LogFrame as _LF, LogFrameStatus as _LFS, ProgrammeHealthScore, ) active_lf_ids = list( _LF.objects.filter(status=_LFS.ACTIVE).values_list("pk", flat=True) ) latest_per_lf: dict[int, ProgrammeHealthScore] = {} if active_lf_ids: for row in ( ProgrammeHealthScore.objects.filter(logframe_id__in=active_lf_ids) .select_related("logframe") .order_by("logframe_id", "-period_start", "-computed_at") ): if row.logframe_id not in latest_per_lf: latest_per_lf[row.logframe_id] = row latest_scores = list(latest_per_lf.values()) if latest_scores: avg = sum(float(r.score) for r in latest_scores) / len(latest_scores) ctx["programme_health_avg"] = round(avg, 1) ctx["programme_health_count"] = len(latest_scores) else: ctx["programme_health_avg"] = None ctx["programme_health_count"] = 0 # Top 3 programmes by current score, with last-6-period sparkline # rendered as ready-to-paint SVG points (y is inverted server-side so # high scores draw HIGH on the chart — template-side widthratio cannot # express the inversion cleanly). top = sorted(latest_scores, key=lambda r: float(r.score), reverse=True)[:3] SPARK_W, SPARK_H, SPARK_PAD = 120, 36, 4 top_payload = [] for row in top: history = list( ProgrammeHealthScore.objects.filter(logframe_id=row.logframe_id) .order_by("-period_start", "-computed_at") .values_list("score", "period_label")[:6] ) history.reverse() values = [float(s) for s, _ in history] polyline_points = "" if len(values) >= 2: inner_w = SPARK_W - 2 * SPARK_PAD inner_h = SPARK_H - 2 * SPARK_PAD steps = len(values) - 1 pts = [] for i, v in enumerate(values): x = SPARK_PAD + (i / steps) * inner_w # Clamp 0..100 then invert for SVG y (top = 0). clamped = max(0.0, min(100.0, v)) y = SPARK_PAD + (1 - clamped / 100.0) * inner_h pts.append(f"{x:.1f},{y:.1f}") polyline_points = " ".join(pts) top_payload.append({ "logframe": row.logframe, "score": float(row.score), "period_label": row.period_label, "trend": [ {"value": float(s), "label": label} for s, label in history ], "polyline_points": polyline_points, "spark_w": SPARK_W, "spark_h": SPARK_H, "trend_count": len(values), }) ctx["top_programme_health"] = top_payload # ── M&E SRS Table 66 — Compliance status ──────────────────── from apps.mel.reports.models import ComplianceAlert, ComplianceSeverity compliance_qs = ComplianceAlert.objects.filter(resolved=False) if selected_rf is not None: compliance_qs = compliance_qs.filter( Q(indicator__logframe_row__logframe=selected_rf) | Q(report__template__logframe=selected_rf) ).distinct() sev_counts = {s: 0 for s, _ in ComplianceSeverity.choices} for row in compliance_qs.values("severity").annotate(n=Count("id")): sev_counts[row["severity"]] = row["n"] ctx["compliance_open_count"] = sum(sev_counts.values()) ctx["compliance_critical_count"] = ( sev_counts.get("critical", 0) + sev_counts.get("high", 0) ) ctx["compliance_by_severity"] = sev_counts ctx["recent_compliance_alerts"] = list( compliance_qs.select_related("indicator").order_by("-created_at")[:4] ) # ── M&E SRS Table 66 — Budget utilisation (G3 finance binding) ── from apps.mel.reports.builders import build_budget_variance budget_frames = _LFchoice.objects.filter( status=_LFSchoice.ACTIVE, finance_budget__isnull=False, ).select_related("finance_budget") if selected_rf is not None: budget_frames = budget_frames.filter(pk=selected_rf.pk) total_budgeted = total_expended = 0.0 bound_count = flagged_categories = 0 for lf in budget_frames: payload = build_budget_variance(logframe=lf) if not payload.get("installed"): continue bound_count += 1 flagged_categories += payload.get("flagged_categories", 0) for cat in payload.get("categories", []): total_budgeted += cat.get("budgeted") or 0.0 total_expended += cat.get("expended") or 0.0 if bound_count and total_budgeted > 0: utilisation = round(total_expended / total_budgeted * 100, 1) ctx["budget_summary"] = { "bound_count": bound_count, "budgeted": total_budgeted, "expended": total_expended, "utilisation": utilisation, "flagged_categories": flagged_categories, } else: ctx["budget_summary"] = None # ── Finalise per-Strategic-Objective scorecards ───────────── # Attach governance flags (open compliance, budget pacing, health) to # each SO card and derive an overall RAG from mean attainment. so_compliance: dict[int, int] = {} if so_cards: for _row in ( ComplianceAlert.objects.filter( resolved=False, indicator__logframe_row__logframe_id__in=list(so_cards.keys()), ) .values("indicator__logframe_row__logframe_id") .annotate(n=Count("id")) ): so_compliance[_row["indicator__logframe_row__logframe_id"]] = _row["n"] so_budget: dict[int, dict] = {} for lf in _so_active_lfs: if not lf.finance_budget_id: continue payload = build_budget_variance(logframe=lf) if not payload.get("installed"): continue tot_b = sum(c.get("budgeted") or 0.0 for c in payload.get("categories", [])) tot_e = sum(c.get("expended") or 0.0 for c in payload.get("categories", [])) so_budget[lf.pk] = { "utilisation": round(tot_e / tot_b * 100, 1) if tot_b else None, "flagged": payload.get("flagged_categories", 0), } def _rag_from_pct(avg): if avg is None: return "unknown" if avg >= 85: return "green" if avg >= 60: return "amber" return "red" _rag_rank = {"red": 0, "amber": 1, "unknown": 2, "green": 3} strategic_objectives = [] for lf in _so_active_lfs: card = so_cards[lf.pk] avg = round(card["pct_sum"] / card["pct_n"], 1) if card["pct_n"] else None bands_out = [] # Display top-down (goal → means): Impact, Outcome, Output. for key, title in (("impact", "Impact"), ("outcome", "Outcomes"), ("output", "Outputs")): cb = card["bands"][key] b_avg = round(cb["pct_sum"] / cb["pct_n"], 1) if cb["pct_n"] else None bands_out.append({ "key": key, "title": title, "count": cb["count"], "attainment": b_avg, "rag": _rag_from_pct(b_avg), "green": cb["green"], "amber": cb["amber"], "red": cb["red"], "unknown": cb["unknown"], }) health = latest_per_lf.get(lf.pk) strategic_objectives.append({ "logframe": lf, "indicator_count": card["indicator_count"], "attainment": avg, "rag": _rag_from_pct(avg), "no_data": card["rag"]["unknown"], "bands": bands_out, "compliance_open": so_compliance.get(lf.pk, 0), "budget": so_budget.get(lf.pk), "health": round(float(health.score), 1) if health else None, }) # Needs-attention first: red, then amber, then no-data, then green. strategic_objectives.sort( key=lambda c: (_rag_rank.get(c["rag"], 4), c["logframe"].name) ) ctx["strategic_objectives"] = strategic_objectives ctx["strategic_objectives_count"] = len(strategic_objectives) # ── Action queue ──────────────────────────────────────────── from apps.mel.tracking.models import CorrectiveAction, CorrectiveActionStatus today = timezone.now().date() ctx["today_date"] = today ca_open = CorrectiveAction.objects.filter( status__in=[CorrectiveActionStatus.OPEN, CorrectiveActionStatus.IN_PROGRESS], ) ctx["open_corrective_actions_count"] = ca_open.count() ctx["overdue_corrective_actions_count"] = ca_open.filter( due_date__isnull=False, due_date__lt=today, ).count() ctx["open_corrective_actions"] = list( ca_open.select_related("subject_type", "owner").order_by( "-status", # OPEN before IN_PROGRESS (alpha order) "due_date", )[:5] ) from apps.mel.reports.models import Report, ReportStatus pending_reports_qs = Report.objects.filter( status__in=[ReportStatus.DRAFT, ReportStatus.IN_REVIEW], ).select_related("template") if selected_rf is not None: pending_reports_qs = pending_reports_qs.filter(template__logframe=selected_rf) if filters["period"]: pending_reports_qs = pending_reports_qs.filter(period_label=filters["period"]) ctx["pending_reports_count"] = pending_reports_qs.count() ctx["pending_reports"] = list(pending_reports_qs.order_by("-generated_at")[:5]) late_activities_qs = Activity.objects.filter( Q(status=ActivityStatus.DELAYED) | ( Q(scheduled_end__lt=today) & ~Q(status__in=[ActivityStatus.COMPLETED, ActivityStatus.CANCELLED]) ), ).select_related("logframe_row__logframe", "responsible") if selected_rf is not None: late_activities_qs = late_activities_qs.filter( logframe_row__logframe=selected_rf ) ctx["late_activities_count"] = late_activities_qs.count() ctx["late_activities"] = list(late_activities_qs.order_by("scheduled_end")[:5]) # ── Attention items ───────────────────────────────────────── from apps.mel.feedback.models import ( FeedbackSeverity, FeedbackStatus, FeedbackSubmission, ) from apps.mel.indicators.models import DataPoint urgent_feedback_qs = FeedbackSubmission.objects.filter( severity__in=[FeedbackSeverity.HIGH, FeedbackSeverity.CRITICAL], status__in=[FeedbackStatus.NEW, FeedbackStatus.TRIAGED], ).select_related("channel") ctx["urgent_feedback_count"] = urgent_feedback_qs.count() ctx["urgent_feedback"] = list(urgent_feedback_qs.order_by("-created_at")[:5]) seven_days = timezone.now() - timedelta(days=7) dp_qs = DataPoint.objects.filter(created_at__gte=seven_days) if filters["period"]: dp_qs = dp_qs.filter(period_label=filters["period"]) dp_rows = dp_qs.values("status").annotate(n=Count("id")) dp_summary = {"verified": 0, "pending": 0, "rejected": 0, "total": 0} for row in dp_rows: key = row["status"] dp_summary[key] = row["n"] dp_summary["total"] += row["n"] ctx["recent_datapoints"] = dp_summary # ── Cross-module reach (filtered by participant attributes) ── participants_qs = Participant.objects.all() if filters["cohort_id"]: participants_qs = participants_qs.filter( entrepreneur_profile__cohort_memberships__cohort_id=filters["cohort_id"] ).distinct() # Sector / country / AIH are SME-Hub Business attributes reached from a # Participant via its EntrepreneurProfile's SMEHubTrackingRecord (the # OneToOne reverse ``mel_tracking_record`` → ``business``). EntrepreneurProfile # has no direct ``business`` relation, so the naive # ``entrepreneur_profile__business__…`` lookup raises FieldError (defect #2). if filters["sector"]: participants_qs = participants_qs.filter( entrepreneur_profile__mel_tracking_record__business__sector=filters["sector"] ) if filters["country"]: participants_qs = participants_qs.filter( entrepreneur_profile__mel_tracking_record__business__country=filters["country"] ) if filters["aih_id"]: participants_qs = participants_qs.filter( entrepreneur_profile__mel_tracking_record__business__affiliations__institution_id=filters["aih_id"] ).distinct() records = TrackingRecord.objects.filter(participant__in=participants_qs) if any_filter_active else TrackingRecord.objects.all() ctx["totals"] = { "participants": participants_qs.count() if any_filter_active else Participant.objects.count(), "active": records.filter(status=TrackingLifecycle.ACTIVE).count(), "completed": records.filter(status=TrackingLifecycle.COMPLETED).count(), "stalled": records.filter(status=TrackingLifecycle.STALLED).count(), } per_module = [] for value, label in TrackingRecord.Module.choices: qs = records.filter(source_module=value) per_module.append({ "value": value, "label": label, "total": qs.count(), "active": qs.filter(status=TrackingLifecycle.ACTIVE).count(), "completed": qs.filter(status=TrackingLifecycle.COMPLETED).count(), "stalled": qs.filter(status=TrackingLifecycle.STALLED).count(), }) ctx["per_module"] = per_module milestones_qs = TrackingMilestone.objects.select_related("record__participant__user") if any_filter_active: milestones_qs = milestones_qs.filter(record__participant__in=participants_qs) ctx["recent_milestones"] = list(milestones_qs.order_by("-occurred_at")[:5]) impact_qs = ImpactScore.objects.select_related("participant__user") if any_filter_active: impact_qs = impact_qs.filter(participant__in=participants_qs) ctx["top_impact_scores"] = list(impact_qs.order_by("-score")[:5]) # ── Participants preview (top 10) ─────────────────────────── ctx["participants_preview"] = list( participants_qs.select_related("user") .prefetch_related("tracking_records") .order_by("-updated_at")[:5] ) ctx["participants_total"] = participants_qs.count() if any_filter_active else Participant.objects.count() # ── SME-Hub stall flag (kept; rendered inside cross-module section) ─ config = MELConfiguration.get() ctx["stall_threshold_days"] = config.smehub_stall_threshold_days stalled_qs = ( SMEHubTrackingRecord.objects.filter(is_stalled=True) .select_related("entrepreneur__user", "business") .order_by("-stall_flagged_at") ) if filters["sector"]: stalled_qs = stalled_qs.filter(business__sector=filters["sector"]) if filters["country"]: stalled_qs = stalled_qs.filter(business__country=filters["country"]) if filters["cohort_id"]: stalled_qs = stalled_qs.filter(entrepreneur__cohort_memberships__cohort_id=filters["cohort_id"]).distinct() if filters["aih_id"]: stalled_qs = stalled_qs.filter(business__affiliations__institution_id=filters["aih_id"]).distinct() ctx["smehub_stalled_count"] = stalled_qs.count() ctx["smehub_stalled_top"] = list(stalled_qs[:5]) return ctx class ParticipantDirectoryView(RoleRequiredMixin, TemplateView): """Full participant directory — filterable table moved off the dashboard. Reachable from the dashboard's "Open the full directory →" link. """ allowed_roles = _PARTICIPANT_ROLES template_name = "tracking/participant_directory.html" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx.update(_participant_directory_context(self.request)) ctx["participants"] = paginate_qs( self.request, ctx["participants"], per_page=50 ) ctx["participants_total"] = Participant.objects.count() return ctx class ParticipantRecordView(RoleRequiredMixin, DetailView): """Per-participant timeline + baseline + impact scorecard.""" allowed_roles = _PARTICIPANT_ROLES model = Participant context_object_name = "participant" template_name = "tracking/participant_record.html" def get_queryset(self): return Participant.objects.select_related( "user", "scholarship", "entrepreneur_profile", "alumni_profile", ) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) participant: Participant = self.object # type: ignore[assignment] ctx["records"] = list( participant.tracking_records.order_by("source_module") ) active_module = (self.request.GET.get("module") or "").strip() milestones = TrackingMilestone.objects.filter( record__participant=participant, ) if active_module: milestones = milestones.filter(source_module=active_module) ctx["milestones"] = paginate_qs( self.request, milestones.order_by("-occurred_at"), per_page=50, param="tl_page", ) ctx["active_module"] = active_module ctx["baselines"] = ( BaselineRecord.objects.filter(participant=participant) .order_by("dimension", "-captured_at") ) ctx["impact_scores"] = ( ImpactScore.objects.filter(participant=participant) .order_by("-period", "-computed_at") ) ctx["latest_impact"] = ctx["impact_scores"].first() return ctx class TimelineFragmentView(RoleRequiredMixin, View): """HTMX endpoint that swaps the timeline list when the module tab changes.""" allowed_roles = _PARTICIPANT_ROLES def get(self, request, pk: int): participant = get_object_or_404(Participant, pk=pk) module = (request.GET.get("module") or "").strip() milestones = TrackingMilestone.objects.filter(record__participant=participant) if module: milestones = milestones.filter(source_module=module) milestones = paginate_qs( request, milestones.order_by("-occurred_at"), per_page=50, param="tl_page", ) from django.shortcuts import render return render( request, "tracking/partials/timeline_list.html", {"milestones": milestones, "active_module": module, "participant": participant}, ) class CorrectiveActionCreateView(RoleRequiredMixin, View): """Open a new CorrectiveAction against a subject (indicator / activity / output). Subject is identified by GET/POST params: ``subject_app`` — Django app label (e.g., ``mel_indicators``) ``subject_model`` — model name lower-cased (e.g., ``indicator``) ``subject_id`` — primary key of the subject ``next`` — optional return URL after save """ allowed_roles = OFFICER_ROLES template_name = "tracking/corrective_action_form.html" def _resolve_subject(self, request): from django.contrib.contenttypes.models import ContentType app_label = (request.GET.get("subject_app") or request.POST.get("subject_app") or "").strip() model_name = (request.GET.get("subject_model") or request.POST.get("subject_model") or "").strip() try: subject_id = int(request.GET.get("subject_id") or request.POST.get("subject_id") or 0) except (TypeError, ValueError): subject_id = 0 if not (app_label and model_name and subject_id): return None, None, None try: ct = ContentType.objects.get(app_label=app_label, model=model_name) except ContentType.DoesNotExist: return None, None, None subject = ct.model_class().objects.filter(pk=subject_id).first() return ct, subject_id, subject def get(self, request): from django.shortcuts import render ct, subject_id, subject = self._resolve_subject(request) if not subject: messages.error(request, "Missing or invalid subject for corrective action.") return HttpResponseRedirect(reverse("mel_tracking:event_log")) initial = {} prefill_title = request.GET.get("title") if prefill_title: initial["title"] = prefill_title[:255] prefill_cause = request.GET.get("root_cause") if prefill_cause: initial["root_cause"] = prefill_cause[:2000] return render( request, self.template_name, { "form": CorrectiveActionCreateForm(initial=initial), "subject": subject, "subject_app": ct.app_label, "subject_model": ct.model, "subject_id": subject_id, "next_url": request.GET.get("next") or "", }, ) def post(self, request): ct, subject_id, subject = self._resolve_subject(request) if not subject: messages.error(request, "Missing or invalid subject for corrective action.") return HttpResponseRedirect(reverse("mel_tracking:event_log")) form = CorrectiveActionCreateForm(request.POST) if not form.is_valid(): from django.shortcuts import render return render( request, self.template_name, { "form": form, "subject": subject, "subject_app": ct.app_label, "subject_model": ct.model, "subject_id": subject_id, "next_url": request.POST.get("next") or "", }, status=400, ) action = form.save(commit=False) action.subject_type = ct action.subject_id = subject_id action.created_by = request.user action.save() messages.success(request, "Corrective action opened.") next_url = (request.POST.get("next") or "").strip() return HttpResponseRedirect(next_url or reverse("mel_tracking:event_log")) class CorrectiveActionResolveView(RoleRequiredMixin, View): """M&E SRS Table 66 — advance a corrective action (in-progress / done / verified-effective / cancelled). VERIFIED records the follow-up assessment.""" allowed_roles = OFFICER_ROLES + ( UserRole.PROGRAM_MANAGER, UserRole.PROGRAM_DIRECTOR, ) def post(self, request, pk: int): from apps.mel.tracking.models import CorrectiveAction, CorrectiveActionStatus from apps.mel.tracking.services import resolve_corrective_action action = get_object_or_404(CorrectiveAction, pk=pk) new_status = request.POST.get("status", "") valid = {s.value for s in CorrectiveActionStatus} if new_status not in valid: messages.error(request, "Invalid corrective-action status.") else: note = (request.POST.get("note") or "").strip() if new_status == CorrectiveActionStatus.VERIFIED and not note: messages.error( request, "Record a follow-up assessment before marking the action verified effective.", ) else: try: resolve_corrective_action( action.pk, actor=request.user, new_status=new_status, note=note, ) except ValidationError as exc: messages.error(request, "; ".join(exc.messages)) else: messages.success(request, "Corrective action updated.") next_url = (request.POST.get("next") or "").strip() return HttpResponseRedirect( next_url or reverse("mel_tracking:tracking_dashboard") ) class CorrectiveActionListView(RoleRequiredMixin, ListView): """M&E SRS Table 66 — the corrective-action register. Corrective actions were previously create-and-resolve only (write-only); this register lists every action with its severity, owner, due date and lifecycle state so monitoring "shall track corrective actions … until implementation is verified" is an actual browsable surface. """ allowed_roles = OFFICER_ROLES + ( UserRole.PROGRAM_MANAGER, UserRole.PROGRAM_DIRECTOR, ) model = None # set in get_queryset via local import context_object_name = "actions" paginate_by = 20 template_name = "tracking/corrective_action_list.html" def get_queryset(self): from apps.mel.tracking.models import CorrectiveAction, CorrectiveActionStatus qs = ( CorrectiveAction.objects.select_related("owner", "created_by", "subject_type") .prefetch_related("subject") .order_by("-created_at") ) scope = self.request.GET.get("scope", "open").strip() if scope == "open": qs = qs.filter( status__in=[ CorrectiveActionStatus.OPEN, CorrectiveActionStatus.IN_PROGRESS, ] ) elif scope == "awaiting": qs = qs.filter(status=CorrectiveActionStatus.DONE) elif scope == "closed": qs = qs.filter( status__in=[ CorrectiveActionStatus.VERIFIED, CorrectiveActionStatus.CANCELLED, ] ) severity = self.request.GET.get("severity", "").strip() if severity: qs = qs.filter(severity=severity) if self.request.GET.get("overdue") == "1": qs = qs.filter( status__in=[ CorrectiveActionStatus.OPEN, CorrectiveActionStatus.IN_PROGRESS, ], due_date__lt=timezone.now().date(), ) return qs def get_context_data(self, **kwargs): from django.db.models import Count, Q from apps.mel.tracking.models import ( CorrectiveAction, CorrectiveActionStatus, CorrectiveSeverity, ) ctx = super().get_context_data(**kwargs) today = timezone.now().date() ctx["stats"] = CorrectiveAction.objects.aggregate( open_total=Count( "pk", filter=Q( status__in=[ CorrectiveActionStatus.OPEN, CorrectiveActionStatus.IN_PROGRESS, ] ), ), overdue=Count( "pk", filter=Q( status__in=[ CorrectiveActionStatus.OPEN, CorrectiveActionStatus.IN_PROGRESS, ], due_date__lt=today, ), ), awaiting=Count("pk", filter=Q(status=CorrectiveActionStatus.DONE)), verified=Count("pk", filter=Q(status=CorrectiveActionStatus.VERIFIED)), ) ctx["scope"] = self.request.GET.get("scope", "open").strip() ctx["severity_filter"] = self.request.GET.get("severity", "").strip() ctx["overdue_filter"] = self.request.GET.get("overdue", "").strip() ctx["severity_choices"] = CorrectiveSeverity.choices ctx["register_url"] = self.request.get_full_path() ctx["can_resolve"] = ( self.request.user.is_superuser or getattr(self.request.user, "role", "") in self.allowed_roles ) return ctx # --------------------------------------------------------------------------- # G2 — QR-code attendance & registration # --------------------------------------------------------------------------- class PublicAttendanceView(View): """Token-gated attendance form. No auth required. Mirrors :class:`apps.mel.feedback.views.PublicFeedbackView`. POSTs route through :func:`apps.mel.tracking.services.record_attendance` so the dedupe-key uniqueness logic stays consistent. """ template_name = "tracking/attendance_scan.html" def _get_channel(self, token: str): from apps.mel.tracking.models import ActivityAttendanceChannel return get_object_or_404(ActivityAttendanceChannel, token=token) def get(self, request, token: str): from django.shortcuts import render from apps.mel.tracking.forms import PublicAttendanceForm from apps.mel.tracking.services import _channel_is_open channel = self._get_channel(token) if not _channel_is_open(channel): return render(request, self.template_name, { "channel": channel, "form": None, "closed": True, }, status=410) return render(request, self.template_name, { "channel": channel, "form": PublicAttendanceForm(), "closed": False, }) def post(self, request, token: str): from django.shortcuts import render from apps.mel.tracking.forms import PublicAttendanceForm from apps.mel.tracking.services import ( AttendanceChannelClosedError, record_attendance, ) channel = self._get_channel(token) form = PublicAttendanceForm(request.POST) if not form.is_valid(): return render(request, self.template_name, { "channel": channel, "form": form, "closed": False, }) try: attendance, created = record_attendance( channel, attendee_user=request.user if request.user.is_authenticated else None, attendee_name=form.cleaned_data["attendee_name"], attendee_email=form.cleaned_data["attendee_email"], attendee_phone=form.cleaned_data["attendee_phone"], organisation=form.cleaned_data["organisation"], notes=form.cleaned_data["notes"], ) except AttendanceChannelClosedError: return render(request, self.template_name, { "channel": channel, "form": None, "closed": True, }, status=410) return render(request, self.template_name, { "channel": channel, "form": None, "closed": False, "attendance": attendance, "created": created, }) class ActivityAttendancePageView(RoleRequiredMixin, View): """Officer-facing attendance page: QR card, counts, attendees list.""" allowed_roles = (UserRole.MEL_OFFICER, UserRole.ADMIN, UserRole.SYSTEM_ADMIN) def get(self, request, pk: int): from django.shortcuts import render from apps.mel.tracking.services import ensure_attendance_channel activity = get_object_or_404(Activity, pk=pk) channel = ensure_attendance_channel(activity, created_by=request.user) attendees = list(channel.attendances.order_by("-attended_at")) scan_url = request.build_absolute_uri( reverse("mel_tracking:attendance_public", kwargs={"token": channel.token}) ) return render(request, "tracking/activity_attendance.html", { "activity": activity, "channel": channel, "attendees": attendees, "attendee_count": len(attendees), "scan_url": scan_url, }) class ActivityAttendanceQrView(RoleRequiredMixin, View): """PNG endpoint serving the QR encoding the public scan URL.""" allowed_roles = (UserRole.MEL_OFFICER, UserRole.ADMIN, UserRole.SYSTEM_ADMIN) def get(self, request, pk: int): from django.http import HttpResponse from apps.mel.tracking.services import ensure_attendance_channel try: import qrcode except ImportError: return HttpResponse( "QR generator not installed on server. Run pip install qrcode.", content_type="text/plain", status=503, ) activity = get_object_or_404(Activity, pk=pk) channel = ensure_attendance_channel(activity, created_by=request.user) scan_url = request.build_absolute_uri( reverse("mel_tracking:attendance_public", kwargs={"token": channel.token}) ) img = qrcode.make(scan_url) response = HttpResponse(content_type="image/png") img.save(response, format="PNG") response["Cache-Control"] = "private, max-age=600" return response class ActivityAttendanceCsvView(RoleRequiredMixin, View): """CSV export of attendance for one activity.""" allowed_roles = (UserRole.MEL_OFFICER, UserRole.ADMIN, UserRole.SYSTEM_ADMIN) def get(self, request, pk: int): import csv from django.http import HttpResponse from apps.mel.tracking.services import attendance_csv_rows activity = get_object_or_404(Activity, pk=pk) rows = attendance_csv_rows(activity) response = HttpResponse(content_type="text/csv") response["Content-Disposition"] = ( f'attachment; filename="attendance-activity-{activity.pk}.csv"' ) writer = csv.writer(response) writer.writerows(rows) return response class ActivityAttendanceRotateTokenView(RoleRequiredMixin, View): """POST — rotate the attendance channel token (e.g., printed QR leaked).""" allowed_roles = (UserRole.MEL_OFFICER, UserRole.ADMIN, UserRole.SYSTEM_ADMIN) def post(self, request, pk: int): from apps.mel.tracking.models import _generate_attendance_token from apps.mel.tracking.services import ensure_attendance_channel activity = get_object_or_404(Activity, pk=pk) channel = ensure_attendance_channel(activity, created_by=request.user) channel.token = _generate_attendance_token() channel.save(update_fields=["token", "updated_at"]) messages.success(request, "Attendance link rotated. Print a fresh QR code.") return HttpResponseRedirect( reverse("mel_tracking:activity_attendance", kwargs={"pk": activity.pk}) ) # --------------------------------------------------------------------------- # AR010 — indicator-mapping reference (read-only) # --------------------------------------------------------------------------- class IndicatorReferenceView(RoleRequiredMixin, TemplateView): """Read-only crosswalk of REP event → indicator + AR010 derived rules. Surfaces the live mapping consumed by ``apply_rep_outbox_row`` so M&EL officers can audit what fires when, without reading Python. The data is sourced from ``apps/mel/tracking/indicators.py`` — change a code or threshold there and this page updates automatically. """ template_name = "tracking/indicator_reference.html" allowed_roles = ( UserRole.MEL_OFFICER, UserRole.PROGRAM_MANAGER, UserRole.PROGRAM_DIRECTOR, UserRole.ADMIN, UserRole.SYSTEM_ADMIN, ) def get_context_data(self, **kwargs): from apps.mel.tracking.indicators import ( AR010_DIGITAL_COMPETENCY_PASS_THRESHOLD, iter_canonical_rows, iter_derived_rows, ) ctx = super().get_context_data(**kwargs) ctx["canonical_rows"] = list(iter_canonical_rows()) ctx["derived_rows"] = list(iter_derived_rows()) ctx["digital_competency_threshold"] = AR010_DIGITAL_COMPETENCY_PASS_THRESHOLD return ctx