from __future__ import annotations from django.contrib import messages from django.contrib.auth.decorators import login_required from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin from django.http import Http404 from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.views.decorators.http import require_POST from django.views.generic import CreateView, ListView, TemplateView from apps.alumni.profiles.models import AlumniProfile from apps.alumni.profiles.views import _user_is_staff from apps.core.permissions.roles import UserRole from apps.alumni.tracking.forms import ImpactRecordForm, ImpactSurveyLaunchForm, MilestoneForm from apps.alumni.tracking.models import ( CareerMilestone, ImpactRecord, MilestoneVisibility, ResearchOutput, ) from apps.alumni.tracking.services import ( current_period_label, generate_donor_impact_report, ) from apps.alumni.tracking.services_impact_survey import ( IMPACT_SURVEY_SLUG_PREFIX, alumni_cohort_queryset, launch_impact_survey, ) # Roles that may record a career milestone (i.e. are genuinely alumni-adjacent). # Anyone already holding an AlumniProfile is also allowed — see # ``_user_can_record_milestone``. This blocks investors / entrepreneurs / buyers # from silently becoming alumni via get_or_create. MILESTONE_ELIGIBLE_ROLES = { UserRole.ALUMNI.value, UserRole.SCHOLAR.value, UserRole.RESEARCHER.value, UserRole.LEARNER.value, } def _user_can_record_milestone(user) -> bool: if not user.is_authenticated: return False if _user_is_staff(user): return True if AlumniProfile.objects.filter(user=user).exists(): return True return getattr(user, "role", "") in MILESTONE_ELIGIBLE_ROLES class MilestoneListView(ListView): template_name = "alumni/tracking/milestone_list.html" context_object_name = "milestones" paginate_by = 25 def get_queryset(self): profile_id = self.request.GET.get("profile") qs = CareerMilestone.objects.select_related("profile", "profile__user").order_by("-occurred_on") if profile_id: qs = qs.filter(profile_id=profile_id) profile = get_object_or_404(AlumniProfile, pk=profile_id) if not profile.visibility_consent and not ( self.request.user.is_authenticated and (profile.user_id == self.request.user.pk or _user_is_staff(self.request.user)) ): raise Http404("Profile not visible.") else: qs = qs.filter( profile__visibility_consent=True, visibility=MilestoneVisibility.PUBLIC, ) return qs def get_context_data(self, **kwargs): from django.db.models import Count from apps.alumni.tracking.models import MilestoneType ctx = super().get_context_data(**kwargs) type_rows = ( self.get_queryset() .values("milestone_type") .annotate(total=Count("id")) .order_by("-total") ) labels = dict(MilestoneType.choices) ctx["type_breakdown"] = [ {"type": row["milestone_type"], "label": labels.get(row["milestone_type"], row["milestone_type"]), "total": row["total"]} for row in type_rows ] ctx["filtered_profile_id"] = self.request.GET.get("profile") ctx["can_record"] = _user_can_record_milestone(self.request.user) return ctx class MilestoneCreateView(LoginRequiredMixin, UserPassesTestMixin, CreateView): model = CareerMilestone form_class = MilestoneForm template_name = "alumni/tracking/milestone_create.html" def test_func(self): return _user_can_record_milestone(self.request.user) def handle_no_permission(self): if self.request.user.is_authenticated: messages.error( self.request, "Recording career milestones is limited to RUFORUM alumni and scholars. " "If you believe this is a mistake, contact the alumni office.", ) return redirect("alumni_tracking:milestone_list") return super().handle_no_permission() def form_valid(self, form): from apps.alumni.tracking.services import record_milestone profile, _ = AlumniProfile.objects.get_or_create(user=self.request.user) record_milestone( profile, milestone_type=form.cleaned_data["milestone_type"], title=form.cleaned_data["title"], description=form.cleaned_data.get("description", ""), occurred_on=form.cleaned_data["occurred_on"], url=form.cleaned_data.get("url", ""), visibility=form.cleaned_data.get("visibility", "public"), evidence_file=form.cleaned_data.get("evidence_file"), ) return redirect("alumni_tracking:milestone_list") class ImpactReportView(LoginRequiredMixin, UserPassesTestMixin, ListView): template_name = "alumni/tracking/impact_report.html" context_object_name = "snapshots" def test_func(self): return _user_is_staff(self.request.user) def get_queryset(self): from apps.alumni.tracking.models import ImpactSnapshot return ImpactSnapshot.objects.order_by("-period_label")[:24] def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) period = self.request.GET.get("period") or None report = generate_donor_impact_report(period) ctx["report"] = report # Selected period (falls back to the report's resolved current period) # so the picker and CSV export stay in sync. ctx["selected_period"] = self.request.GET.get("period") or report["period_label"] # Tracer/impact surveys — answers "was a study actually carried out?" # alongside the donor-facing metric tiles above (client complaint: # the impact dashboard shows findings but never the underlying study). ctx["impact_surveys"] = _impact_surveys_queryset()[:5] # Aggregated tracer insights (employment outcomes, income change, # training relevance, impact stories) pulled from actual responses. from apps.alumni.tracking.services_impact_survey import ( aggregate_tracer_insights, ) ctx["tracer_insights"] = aggregate_tracer_insights() return ctx ALUMNI_INDICATOR_CODES = ( "alumni-profiles-created", "alumni-milestones-recorded", "alumni-mentorships-active", "alumni-event-registrations", "alumni-orcid-synced", "alumni-publications-linked", "alumni-spotlights-published", ) class AlumniKPIDashboardView(LoginRequiredMixin, UserPassesTestMixin, TemplateView): """Officer-facing alumni KPI dashboard. Renders pre-wired M&EL indicator progress alongside in-module derived rates (profile completion, mentorship acceptance, event attendance, group activity). Distinct from the donor-facing impact report. """ template_name = "alumni/tracking/kpi_dashboard.html" def test_func(self): return _user_is_staff(self.request.user) def get_context_data(self, **kwargs): from datetime import timedelta from django.utils import timezone from apps.alumni.engagement.models import ( EventRegistration, GroupAnnouncement, GroupMembership, MentorshipPairing, MentorshipStatus, NetworkGroup, ) from apps.mel.indicators.models import Indicator from apps.mel.indicators.services import calculate_progress, flag_off_track ctx = super().get_context_data(**kwargs) period = self.request.GET.get("period") or current_period_label() ctx["period_label"] = period indicators = list( Indicator.objects.filter(code__in=ALUMNI_INDICATOR_CODES).select_related("logframe_row") ) rows = [] for ind in indicators: try: progress = calculate_progress(ind, period) except Exception: # pragma: no cover - defensive progress = {"actual": 0, "target": None, "rag": "grey", "percent": None} progress["indicator"] = ind rows.append(progress) ctx["indicator_rows"] = rows try: ctx["off_track"] = flag_off_track(indicators, period_label=period) except Exception: # pragma: no cover - defensive ctx["off_track"] = [] # Derived in-module rates. total_profiles = AlumniProfile.objects.count() opted_in = AlumniProfile.objects.filter(visibility_consent=True).count() completion_pct = round((opted_in / total_profiles) * 100, 1) if total_profiles else 0.0 total_requested = MentorshipPairing.objects.exclude( status=MentorshipStatus.CANCELLED ).count() accepted = MentorshipPairing.objects.filter( status__in=[MentorshipStatus.ACTIVE, MentorshipStatus.COMPLETED] ).count() acceptance_pct = round((accepted / total_requested) * 100, 1) if total_requested else 0.0 total_regs = EventRegistration.objects.count() attended = EventRegistration.objects.filter(attended=True).count() attendance_pct = round((attended / total_regs) * 100, 1) if total_regs else 0.0 thirty_days_ago = timezone.now() - timedelta(days=30) ctx["derived"] = { "total_profiles": total_profiles, "opted_in": opted_in, "completion_pct": completion_pct, "total_requested": total_requested, "accepted": accepted, "acceptance_pct": acceptance_pct, "attended": attended, "total_regs": total_regs, "attendance_pct": attendance_pct, "groups_total": NetworkGroup.objects.count(), "announcements_30d": GroupAnnouncement.objects.filter( created_at__gte=thirty_days_ago ).count(), "memberships_30d": GroupMembership.objects.filter( joined_at__gte=thirty_days_ago ).count(), } return ctx class ResearchOutputListView(LoginRequiredMixin, ListView): """The signed-in alumnus's own ORCID-synced research works. ORCID sync otherwise only runs on a weekly beat task, so without this surface an alumnus who just added their ORCID iD would see nothing for up to a week. The "Sync now" action (below) triggers it on demand. """ template_name = "alumni/tracking/research_outputs.html" context_object_name = "outputs" paginate_by = 25 def _profile(self): if not hasattr(self, "_cached_profile"): self._cached_profile = AlumniProfile.objects.filter( user=self.request.user ).first() return self._cached_profile def get_queryset(self): profile = self._profile() if profile is None: return ResearchOutput.objects.none() return profile.research_outputs.all() def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) profile = self._profile() ctx["profile"] = profile ctx["has_orcid"] = bool(profile and (profile.orcid_id or "").strip()) ctx["orcid_id"] = profile.orcid_id if profile else "" ctx["sync_log"] = getattr(profile, "orcid_sync_log", None) if profile else None return ctx @login_required @require_POST def research_outputs_sync(request): """Trigger an on-demand ORCID sync for the signed-in user's profile.""" from apps.alumni.tracking.services import sync_orcid profile = AlumniProfile.objects.filter(user=request.user).first() redirect_to = redirect("alumni_tracking:research_outputs") if profile is None: messages.error( request, "You don't have an alumni profile yet — create one before syncing ORCID.", ) return redirect_to if not (profile.orcid_id or "").strip(): messages.warning( request, "Add your ORCID iD to your profile first, then sync to import your publications.", ) return redirect_to before = profile.research_outputs.count() try: log = sync_orcid(profile) except Exception: # pragma: no cover - defensive; sync_orcid rarely raises messages.error( request, "We couldn't reach ORCID just now. Please try again in a few minutes.", ) return redirect_to if log.last_error: messages.error( request, "ORCID sync ran into a problem — please confirm your ORCID iD is correct.", ) return redirect_to added = profile.research_outputs.count() - before total = profile.research_outputs.count() if added > 0: messages.success( request, f"Synced {added} new research output{'' if added == 1 else 's'} from ORCID " f"({total} total).", ) else: messages.success( request, f"ORCID sync complete — no new works ({total} on file).", ) return redirect_to class ImpactRecordCreateView(LoginRequiredMixin, UserPassesTestMixin, CreateView): """Staff data-entry for a single donor-facing impact metric. Without this the impact report's citations / funding tiles only ever reflect seeded data. On save we rebuild the snapshot for the record's period so the donor report reflects the new number immediately. """ model = ImpactRecord form_class = ImpactRecordForm template_name = "alumni/tracking/impact_record_create.html" def test_func(self): return _user_is_staff(self.request.user) def form_valid(self, form): from apps.alumni.tracking.services import build_impact_snapshot record = form.save() # Refresh the donor snapshot for this period so the tile updates now. try: build_impact_snapshot(record.period_label) except Exception: # pragma: no cover - defensive pass messages.success( self.request, f"Impact record saved: {record.get_metric_display()} = {record.value} " f"for {record.period_label}.", ) url = reverse("alumni_tracking:impact_report") return redirect(f"{url}?period={record.period_label}") @login_required def impact_report_export(request): """Simple CSV export of the latest impact report — staff only.""" if not _user_is_staff(request.user): raise Http404("Not available.") from csv import writer from io import StringIO from django.http import HttpResponse report = generate_donor_impact_report(request.GET.get("period")) buf = StringIO() w = writer(buf) w.writerow(["period", report["period_label"]]) w.writerow(["total_alumni", report["total_alumni"]]) w.writerow(["active_mentorships", report["active_mentorships"]]) w.writerow(["total_publications", report["total_publications"]]) w.writerow(["total_citations", report["total_citations"]]) w.writerow(["funding_secured_usd", report["funding_secured_usd"]]) resp = HttpResponse(buf.getvalue(), content_type="text/csv") resp["Content-Disposition"] = ( f'attachment; filename="alumni_impact_{report["period_label"]}.csv"' ) return resp # --------------------------------------------------------------------------- # Impact / tracer surveys — thin wrapper UI over apps.mel.feedback's Survey # engine. See apps.alumni.tracking.services_impact_survey.launch_impact_survey. # --------------------------------------------------------------------------- def _impact_surveys_queryset(): """Surveys launched via the alumni impact-survey wrapper, with counts. Tagged by slug prefix (see IMPACT_SURVEY_SLUG_PREFIX) — the shared Survey model has no field of its own to distinguish "launched from alumni tracking" from any other M&EL survey. """ from django.db.models import Count, Q from apps.mel.feedback.models import Survey, SurveyResponse return ( Survey.objects.filter(slug__startswith=IMPACT_SURVEY_SLUG_PREFIX) .annotate( dispatched_count=Count("dispatches", distinct=True), completed_count=Count( "responses", filter=Q(responses__status=SurveyResponse.Status.COMPLETED), distinct=True, ), ) .order_by("-created_at") ) class ImpactSurveyListView(LoginRequiredMixin, UserPassesTestMixin, ListView): """Officer view of alumni-launched impact/tracer surveys. Shows dispatch/completion counts per survey and links out to the existing mel_feedback results dashboard and take-form preview — this view does not duplicate any of that machinery. """ template_name = "alumni/tracking/impact_survey_list.html" context_object_name = "surveys" paginate_by = 25 def test_func(self): return _user_is_staff(self.request.user) def get_queryset(self): return _impact_surveys_queryset() class ImpactSurveyLaunchView(LoginRequiredMixin, UserPassesTestMixin, TemplateView): """Officer form to launch a new alumni impact/tracer survey.""" template_name = "alumni/tracking/impact_survey_launch.html" def test_func(self): return _user_is_staff(self.request.user) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx.setdefault("form", ImpactSurveyLaunchForm()) return ctx def post(self, request, *args, **kwargs): form = ImpactSurveyLaunchForm(request.POST) if not form.is_valid(): return self.render_to_response(self.get_context_data(form=form)) cohort_filter = form.cohort_filter() # v1: only the standard tracer question set is offered from this UI # (launch_impact_survey defaults to DEFAULT_TRACER_QUESTIONS when # questions=None). Custom question builders live in mel_feedback's # own survey builder for surveys that need bespoke instruments. questions = None if not form.cleaned_data["use_standard_questions"]: messages.info( request, "Custom question sets aren't available from this form yet — " "the standard tracer questions were used. To customise " "questions, edit the survey afterwards from the M&EL survey " "builder.", ) survey = launch_impact_survey( title=form.cleaned_data["title"], description=form.cleaned_data.get("description", ""), cohort_filter=cohort_filter, questions=questions, created_by=request.user, ) invited = survey.dispatches.count() if invited: messages.success( request, f"“{survey.title}” launched and invited to {invited} " f"alumn{'us' if invited == 1 else 'i'}.", ) else: messages.warning( request, f"“{survey.title}” was launched, but no alumni matched that " "cohort — nobody was invited. Adjust the filters and try again.", ) return redirect("alumni_tracking:impact_survey_list")