"""Alumni-facing job board. This is a *proxy* surface over the ``apps.alumni.careers.JobListing`` model — there is no Alumni-owned job model. The alumni view scopes to listings whose ``audience`` includes alumni and adds a match-aware sort that ranks by the overlap between the alumnus's expertise tags and the listing's sector + title tokens. """ from __future__ import annotations import logging from django.contrib import messages from django.contrib.auth.decorators import login_required from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin from django.core.exceptions import ValidationError from django.db.models import Q from django.http import Http404, HttpResponse 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, DetailView, ListView, View from apps.alumni.engagement.matcher import _token_overlap_score from apps.alumni.jobs.forms import PartnerJobSubmissionForm from apps.alumni.profiles.models import AlumniProfile from apps.alumni.profiles.views import _user_is_staff from apps.alumni.careers.forms import JobApplicationForm from apps.alumni.careers.models import JobApplication, JobListing from apps.alumni.careers.services import apply_for_job from apps.core.audit.rep_helpers import action_create, audit_rep from apps.core.permissions.roles import UserRole from apps.smehub.linkage.models import PartnerDirectoryEntry logger = logging.getLogger(__name__) def _user_partner_entry(user) -> PartnerDirectoryEntry | None: """Return the verified PartnerDirectoryEntry the user manages, if any.""" if not user.is_authenticated: return None return ( PartnerDirectoryEntry.objects.filter( managed_by=user, verification_status=PartnerDirectoryEntry.Status.VERIFIED, ) .first() ) def _can_submit_job(user) -> bool: """Eligibility to submit a partner job — the single source of truth shared by the board button gate and ``PartnerJobSubmitView.test_func`` so the button never shows to a user who would then hit a 403.""" if not user.is_authenticated: return False if _user_is_staff(user): return True if user.has_perm("rep_career.post_job_listing"): return True if getattr(user, "role", "") == UserRole.PARTNER.value: return True return _user_partner_entry(user) is not None def _alumni_audience_filter(): return Q( audience__in=[ JobListing.Audience.ALL, JobListing.Audience.ALUMNI, ] ) def _viewer_expertise_tags(user) -> list[str]: profile = AlumniProfile.objects.filter(user=user).only("expertise_tags").first() if not profile: return [] return [t.lower() for t in (profile.expertise_tags or [])] def _listing_tokens(listing: JobListing) -> set[str]: return { t.strip().lower() for raw in (listing.sector or "", listing.title or "") for t in raw.replace(",", " ").replace("/", " ").split() if t.strip() } def _score_listing(listing: JobListing, mentee_tags: set[str]) -> float: if not mentee_tags: return 0.0 return _token_overlap_score(mentee_tags, _listing_tokens(listing)) def _match_reason(listing: JobListing, mentee_tags: set[str]) -> str: """Return the first overlapping tag for a 'matches: X' badge.""" if not mentee_tags: return "" overlap = mentee_tags & _listing_tokens(listing) return sorted(overlap)[0] if overlap else "" class AlumniJobBoardView(LoginRequiredMixin, ListView): template_name = "alumni/jobs/board.html" context_object_name = "listings" paginate_by = 20 def get_queryset(self): qs = JobListing.objects.filter( _alumni_audience_filter(), is_active=True, listing_status=JobListing.ListingStatus.PUBLISHED, ) q = (self.request.GET.get("q") or "").strip() if q: qs = qs.filter( Q(title__icontains=q) | Q(company__icontains=q) | Q(location__icontains=q) | Q(description__icontains=q) ) jt = (self.request.GET.get("job_type") or "").strip() if jt: qs = qs.filter(job_type=jt) sector = (self.request.GET.get("sector") or "").strip() if sector: qs = qs.filter(sector__iexact=sector) sort = (self.request.GET.get("sort") or "").strip() if sort == "closing": return qs.exclude(closes_at__isnull=True).order_by("closes_at") if sort == "recent": return qs.order_by("-created_at") # Default: match-aware if the alumnus has expertise tags. tags = set(_viewer_expertise_tags(self.request.user)) if tags: scored = [] for L in qs: score = _score_listing(L, tags) # Annotate the in-memory object so the template can read it without # another DB hit. We don't persist these — they're per-render only. L.match_score = score L.match_reason = _match_reason(L, tags) scored.append((L, score)) scored.sort(key=lambda pair: (pair[1], pair[0].created_at), reverse=True) return [L for L, _score in scored] return qs.order_by("-created_at") def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["job_type_choices"] = JobListing.JobType.choices active_sectors = set( JobListing.objects.filter( _alumni_audience_filter(), is_active=True, listing_status=JobListing.ListingStatus.PUBLISHED, ) .exclude(sector="") .values_list("sector", flat=True) .distinct() ) ctx["distinct_sectors"] = [ s for s, _ in JobListing.SECTOR_CHOICES if s in active_sectors ] ctx["my_applications"] = set( JobApplication.objects.filter(applicant=self.request.user).values_list( "listing_id", flat=True ) ) ctx["sort"] = self.request.GET.get("sort") or "match" ctx["q"] = self.request.GET.get("q") or "" ctx["job_type"] = self.request.GET.get("job_type") or "" ctx["sector"] = self.request.GET.get("sector") or "" ctx["can_submit_job"] = _can_submit_job(self.request.user) return ctx class AlumniJobDetailView(LoginRequiredMixin, DetailView): template_name = "alumni/jobs/detail.html" context_object_name = "listing" def get_queryset(self): # The public board only exposes published, alumni-audience listings — # but the submitter (posted_by) must be able to open their OWN listing # in any state (pending/rejected/expired) from "My submissions", and # staff can open any listing. Otherwise those links 404. public = ( _alumni_audience_filter() & Q(is_active=True) & Q(listing_status=JobListing.ListingStatus.PUBLISHED) ) viewer = self.request.user if not viewer.is_authenticated: return JobListing.objects.filter(public) if _user_is_staff(viewer): return JobListing.objects.all() # Owner (submitter) and anyone who has already applied can always open # the listing — so "My submissions" and "My applications" links never # 404 just because a listing was later unpublished or is still pending. return JobListing.objects.filter( public | Q(posted_by=viewer) | Q(applications__applicant=viewer) ).distinct() def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) listing = self.object ctx["already_applied"] = JobApplication.objects.filter( listing=listing, applicant=self.request.user ).exists() # Only show the apply form when the listing is genuinely open to # applications and the viewer isn't the person who posted it. ctx["is_owner"] = listing.posted_by_id == self.request.user.pk ctx["can_apply"] = ( listing.is_active and listing.listing_status == JobListing.ListingStatus.PUBLISHED and not ctx["is_owner"] ) ctx["form"] = JobApplicationForm() return ctx def post(self, request, *args, **kwargs): # DetailView.get_context_data() reads self.object; set it so the # re-render path (invalid form, or a ValidationError like "already # applied") doesn't crash with AttributeError. self.object = listing = self.get_object() form = JobApplicationForm(request.POST, request.FILES) if form.is_valid(): try: apply_for_job( request.user, listing, form.cleaned_data.get("cover_letter") or "", cover_letter_file=form.cleaned_data.get("cover_letter_file"), ) audit_rep( actor=request.user, action=action_create(), target_model="JobApplication", object_id=f"{listing.pk}:{request.user.pk}", object_repr=f"{request.user.email} -> {listing.title}", changes={"listing_id": listing.pk, "via": "alumni"}, request=request, ) messages.success(request, "Application submitted.") return redirect("alumni_jobs:detail", pk=listing.pk) except ValidationError as exc: messages.error(request, " ".join(exc.messages)) ctx = self.get_context_data(object=listing) ctx["form"] = form return render(request, self.template_name, ctx) class AlumniMyApplicationsView(LoginRequiredMixin, ListView): template_name = "alumni/jobs/my_applications.html" context_object_name = "applications" paginate_by = 25 def get_queryset(self): return ( JobApplication.objects.filter( applicant=self.request.user, listing__audience__in=[ JobListing.Audience.ALL, JobListing.Audience.ALUMNI, ], ) .select_related("listing") .order_by("-applied_at") ) # --------------------------------------------------------------------------- # Partner submission flow # --------------------------------------------------------------------------- _PARTNER_JOB_WIZARD_STEPS = [ {"n": 1, "title": "Role basics"}, {"n": 2, "title": "Description"}, ] class PartnerJobSubmitView(LoginRequiredMixin, UserPassesTestMixin, CreateView): """Verified partners submit alumni-only listings; staff approve before publication. Staff (alumni officer / admin) can use this view too — useful for posting on behalf of a partner over the phone. """ template_name = "alumni/jobs/submit.html" form_class = PartnerJobSubmissionForm def test_func(self): return _can_submit_job(self.request.user) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["partner_entry"] = _user_partner_entry(self.request.user) ctx["wizard_steps"] = _PARTNER_JOB_WIZARD_STEPS ctx["cancel_url"] = reverse("alumni_jobs:board") return ctx def form_valid(self, form): listing: JobListing = form.save(commit=False) listing.audience = JobListing.Audience.ALUMNI listing.listing_status = JobListing.ListingStatus.PENDING listing.is_active = True listing.posted_by = self.request.user partner = _user_partner_entry(self.request.user) if partner is not None: listing.submitted_by_partner = partner if not listing.company: listing.company = partner.organisation_name listing.save() _notify_officers_of_submission(listing) messages.success( self.request, "Listing submitted and now pending admin approval — an alumni officer " "will review it shortly, and it appears on the board once approved.", ) return redirect("alumni_jobs:board") class PartnerMySubmissionsView(LoginRequiredMixin, UserPassesTestMixin, ListView): """Status tracker for listings a partner-org account has submitted — pending / published / rejected — so submitters aren't left guessing after ``PartnerJobSubmitView`` hands them back to the board.""" template_name = "alumni/jobs/my_submissions.html" context_object_name = "listings" paginate_by = 25 def test_func(self): return _can_submit_job(self.request.user) def get_queryset(self): return ( JobListing.objects.filter(posted_by=self.request.user) .select_related("submitted_by_partner") .order_by("-created_at") ) def _notify_officers_of_submission(listing: JobListing) -> None: try: from django.contrib.auth import get_user_model from apps.core.notifications.models import Notification from apps.core.notifications.services import bulk_notify User = get_user_model() officers = User.objects.filter(role__in=["alumni_officer", "admin", "system_admin"]) bulk_notify( officers, f"New partner job submitted for review: {listing.title} — {listing.company}", verb=Notification.Verb.ALUMNI_JOB_SUBMITTED, ) except Exception: # pragma: no cover - defensive logger.exception("alumni.jobs submit notify failed listing=%s", listing.pk) class PartnerJobApprovalQueueView(LoginRequiredMixin, UserPassesTestMixin, ListView): template_name = "alumni/jobs/pending.html" context_object_name = "listings" paginate_by = 25 def test_func(self): return _user_is_staff(self.request.user) def get_queryset(self): return ( JobListing.objects.filter( audience=JobListing.Audience.ALUMNI, listing_status=JobListing.ListingStatus.PENDING, ) .select_related("submitted_by_partner", "posted_by") .order_by("created_at") ) @login_required @require_POST def partner_job_approve_view(request, pk: int) -> HttpResponse: if not _user_is_staff(request.user): raise Http404("Not available.") listing = get_object_or_404(JobListing, pk=pk) listing.listing_status = JobListing.ListingStatus.PUBLISHED listing.is_active = True listing.save(update_fields=["listing_status", "is_active"]) _notify_partner_of_decision(listing, approved=True) messages.success(request, f"Approved: {listing.title}") return redirect("alumni_jobs:pending") @login_required @require_POST def partner_job_reject_view(request, pk: int) -> HttpResponse: if not _user_is_staff(request.user): raise Http404("Not available.") listing = get_object_or_404(JobListing, pk=pk) reason = (request.POST.get("reason") or "").strip() listing.listing_status = JobListing.ListingStatus.REJECTED listing.is_active = False if reason: listing.employer_notes = reason listing.save(update_fields=["listing_status", "is_active", "employer_notes"]) _notify_partner_of_decision(listing, approved=False, reason=reason) messages.success(request, f"Rejected: {listing.title}") return redirect("alumni_jobs:pending") def _notify_partner_of_decision(listing: JobListing, *, approved: bool, reason: str = "") -> None: try: from apps.core.notifications.models import Notification from apps.core.notifications.services import send_notification if listing.posted_by is None: return if approved: message = f"Your alumni job listing '{listing.title}' has been approved and is live." verb = Notification.Verb.ALUMNI_JOB_APPROVED else: # The rejection message carries the full, correct semantics for the # reader. A dedicated ALUMNI_JOB_REJECTED verb belongs in the # notifications app (apps/core/notifications) — adding one there needs # a migration outside this module, so it is tracked as a follow-up. # Until then we deliberately use ALUMNI_JOB_APPROVED's sibling verb # only for grouping; the human-readable text below is authoritative. message = ( f"Your alumni job listing '{listing.title}' was not approved and " f"has been returned. Reason: {reason or 'see the notes on the listing'}." ) verb = Notification.Verb.ALUMNI_JOB_SUBMITTED send_notification(listing.posted_by, message, verb=verb) except Exception: # pragma: no cover - defensive logger.exception("alumni.jobs decision notify failed listing=%s", listing.pk)