314 lines
12 KiB
Python
314 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from django.contrib import messages
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.db.models import (
|
|
CharField,
|
|
DateTimeField,
|
|
DecimalField,
|
|
F,
|
|
JSONField,
|
|
OuterRef,
|
|
Q,
|
|
Subquery,
|
|
)
|
|
from django.http import Http404, HttpResponse
|
|
from django.shortcuts import get_object_or_404, redirect
|
|
from django.urls import reverse
|
|
from django.utils.http import url_has_allowed_host_and_scheme
|
|
from django.views.decorators.http import require_POST
|
|
from django.views.generic import DetailView, ListView, TemplateView, UpdateView
|
|
|
|
from apps.profiles.models import SearchProfile
|
|
|
|
from .forms import ApplicationForm
|
|
from .models import Application, Feedback, JobPosting, ScoreRun
|
|
from .services.applications import (
|
|
build_application_export,
|
|
build_print_html,
|
|
delete_application_dossier,
|
|
track_application_changes,
|
|
)
|
|
from .services.feedback import record_feedback
|
|
from .services.relevance import it_relevance_query
|
|
from .services.skill_demand import build_skill_demand_report
|
|
|
|
|
|
class SkillInsightsView(LoginRequiredMixin, TemplateView):
|
|
template_name = "jobs/skill_insights.html"
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
profile = SearchProfile.objects.filter(user=self.request.user, is_active=True).first()
|
|
context["active_profile"] = profile
|
|
context["report"] = (
|
|
build_skill_demand_report(
|
|
profile,
|
|
category=self.request.GET.get("category", "all"),
|
|
coverage=self.request.GET.get("coverage", "all"),
|
|
)
|
|
if profile
|
|
else None
|
|
)
|
|
return context
|
|
|
|
|
|
class JobListView(LoginRequiredMixin, ListView):
|
|
model = JobPosting
|
|
template_name = "jobs/list.html"
|
|
context_object_name = "jobs"
|
|
paginate_by = 30
|
|
|
|
VALID_RECOMMENDATIONS = {
|
|
ScoreRun.Recommendation.STRONG,
|
|
ScoreRun.Recommendation.POSSIBLE,
|
|
ScoreRun.Recommendation.WEAK,
|
|
ScoreRun.Recommendation.HIDDEN,
|
|
}
|
|
VALID_MATCH_FILTERS = {*VALID_RECOMMENDATIONS, "radar", "all"}
|
|
VALID_SORTS = {"match", "newest", "closing"}
|
|
|
|
def get_queryset(self):
|
|
queryset = JobPosting.objects.select_related("employer").all()
|
|
profile = SearchProfile.objects.filter(user=self.request.user, is_active=True).first()
|
|
if profile:
|
|
latest_score = ScoreRun.objects.filter(job=OuterRef("pk"), profile=profile).order_by(
|
|
"-created_at"
|
|
)
|
|
queryset = queryset.annotate(
|
|
match_score=Subquery(
|
|
latest_score.values("score")[:1],
|
|
output_field=DecimalField(max_digits=5, decimal_places=2),
|
|
),
|
|
match_confidence=Subquery(
|
|
latest_score.values("confidence")[:1],
|
|
output_field=DecimalField(max_digits=4, decimal_places=3),
|
|
),
|
|
match_recommendation=Subquery(
|
|
latest_score.values("recommendation")[:1], output_field=CharField()
|
|
),
|
|
match_positives=Subquery(
|
|
latest_score.values("positives")[:1], output_field=JSONField()
|
|
),
|
|
match_concerns=Subquery(
|
|
latest_score.values("concerns")[:1], output_field=JSONField()
|
|
),
|
|
match_hard_exclusions=Subquery(
|
|
latest_score.values("hard_exclusions")[:1], output_field=JSONField()
|
|
),
|
|
match_scored_at=Subquery(
|
|
latest_score.values("created_at")[:1], output_field=DateTimeField()
|
|
),
|
|
)
|
|
query = self.request.GET.get("q", "").strip()
|
|
status = self.request.GET.get("status", "active").strip()
|
|
workplace = self.request.GET.get("workplace", "").strip()
|
|
focus = self.request.GET.get("focus", "it").strip()
|
|
recommendation = self.request.GET.get("match", "radar").strip()
|
|
sort = self.request.GET.get("sort", "match").strip()
|
|
if recommendation not in self.VALID_MATCH_FILTERS:
|
|
recommendation = "radar"
|
|
if query:
|
|
queryset = queryset.filter(
|
|
Q(original_title__icontains=query)
|
|
| Q(employer__name__icontains=query)
|
|
| Q(description_text__icontains=query)
|
|
| Q(raw_location__icontains=query)
|
|
)
|
|
if status:
|
|
queryset = queryset.filter(status=status)
|
|
if workplace:
|
|
queryset = queryset.filter(workplace_type=workplace)
|
|
if focus != "all":
|
|
queryset = queryset.filter(it_relevance_query())
|
|
if profile and recommendation == "radar":
|
|
queryset = queryset.filter(
|
|
Q(
|
|
match_recommendation__in=[
|
|
ScoreRun.Recommendation.STRONG,
|
|
ScoreRun.Recommendation.POSSIBLE,
|
|
ScoreRun.Recommendation.WEAK,
|
|
]
|
|
)
|
|
| Q(match_recommendation__isnull=True)
|
|
)
|
|
elif profile and recommendation in self.VALID_RECOMMENDATIONS:
|
|
queryset = queryset.filter(match_recommendation=recommendation)
|
|
if sort not in self.VALID_SORTS:
|
|
sort = "match"
|
|
if sort == "newest":
|
|
return queryset.order_by("-first_seen", F("match_score").desc(nulls_last=True))
|
|
if sort == "closing":
|
|
return queryset.order_by(F("valid_through").asc(nulls_last=True), "-first_seen")
|
|
return queryset.order_by(F("match_score").desc(nulls_last=True), "-first_seen")
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
active_jobs = JobPosting.objects.filter(status=JobPosting.Status.ACTIVE)
|
|
it_jobs = active_jobs.filter(it_relevance_query())
|
|
context.update(
|
|
{
|
|
"active_profile": SearchProfile.objects.filter(
|
|
user=self.request.user, is_active=True
|
|
).first(),
|
|
"active_job_count": active_jobs.count(),
|
|
"it_job_count": it_jobs.count(),
|
|
"filtered_non_it_count": active_jobs.exclude(it_relevance_query()).count(),
|
|
"current_focus": self.request.GET.get("focus", "it"),
|
|
"current_match": (
|
|
self.request.GET.get("match", "radar")
|
|
if self.request.GET.get("match", "radar") in self.VALID_MATCH_FILTERS
|
|
else "radar"
|
|
),
|
|
"current_sort": self.request.GET.get("sort", "match"),
|
|
}
|
|
)
|
|
return context
|
|
|
|
|
|
class JobDetailView(LoginRequiredMixin, DetailView):
|
|
model = JobPosting
|
|
template_name = "jobs/detail.html"
|
|
context_object_name = "job"
|
|
|
|
def get_queryset(self):
|
|
return JobPosting.objects.select_related("employer", "duplicate_of").prefetch_related(
|
|
"source_aliases__source", "provenance", "versions", "feedback"
|
|
)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
profile = SearchProfile.objects.filter(user=self.request.user, is_active=True).first()
|
|
context["score"] = (
|
|
ScoreRun.objects.filter(job=self.object, profile=profile)
|
|
.order_by("-created_at")
|
|
.first()
|
|
if profile
|
|
else None
|
|
)
|
|
context["application"] = Application.objects.filter(
|
|
user=self.request.user, job=self.object
|
|
).first()
|
|
context["latest_feedback"] = Feedback.objects.filter(
|
|
user=self.request.user, job=self.object
|
|
).first()
|
|
return context
|
|
|
|
|
|
@login_required
|
|
def job_feedback(request, pk):
|
|
job = get_object_or_404(JobPosting, pk=pk)
|
|
if request.method != "POST":
|
|
return redirect("jobs:detail", pk=pk)
|
|
action = request.POST.get("action", "")
|
|
valid_actions = {choice for choice, _ in Feedback.Action.choices}
|
|
if action not in valid_actions:
|
|
messages.error(request, "Ongeldige actie.")
|
|
return redirect("jobs:detail", pk=pk)
|
|
record_feedback(
|
|
user=request.user,
|
|
job=job,
|
|
action=action,
|
|
reason=request.POST.get("reason", ""),
|
|
)
|
|
messages.success(request, "Actie opgeslagen.")
|
|
next_url = request.POST.get("next")
|
|
if next_url and url_has_allowed_host_and_scheme(
|
|
next_url,
|
|
allowed_hosts={request.get_host()},
|
|
require_https=request.is_secure(),
|
|
):
|
|
return redirect(next_url)
|
|
return redirect(reverse("jobs:detail", kwargs={"pk": pk}))
|
|
|
|
|
|
class ApplicationListView(LoginRequiredMixin, ListView):
|
|
model = Application
|
|
template_name = "applications/list.html"
|
|
context_object_name = "applications"
|
|
|
|
def get_queryset(self):
|
|
return Application.objects.filter(user=self.request.user).select_related(
|
|
"job", "job__employer"
|
|
)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
applications = list(context["applications"])
|
|
context["applications"] = applications
|
|
context["pipeline_columns"] = [
|
|
{
|
|
"status": status,
|
|
"label": label,
|
|
"applications": [item for item in applications if item.status == status],
|
|
}
|
|
for status, label in Application.Status.choices
|
|
]
|
|
return context
|
|
|
|
|
|
class ApplicationUpdateView(LoginRequiredMixin, UpdateView):
|
|
model = Application
|
|
form_class = ApplicationForm
|
|
template_name = "applications/edit.html"
|
|
|
|
def get_queryset(self):
|
|
return Application.objects.filter(user=self.request.user).select_related("job")
|
|
|
|
def form_valid(self, form):
|
|
previous = (
|
|
Application.objects.filter(pk=self.object.pk)
|
|
.values("status", "notes", "contact_name", "contact_email")
|
|
.get()
|
|
)
|
|
response = super().form_valid(form)
|
|
track_application_changes(
|
|
application=self.object,
|
|
user=self.request.user,
|
|
previous=previous,
|
|
current={
|
|
"status": self.object.status,
|
|
"notes": self.object.notes,
|
|
"contact_name": self.object.contact_name,
|
|
"contact_email": self.object.contact_email,
|
|
},
|
|
)
|
|
return response
|
|
|
|
def get_success_url(self):
|
|
messages.success(self.request, "Sollicitatiedossier opgeslagen.")
|
|
return reverse("jobs:applications")
|
|
|
|
|
|
@login_required
|
|
def application_export(request, pk: int):
|
|
application = get_object_or_404(
|
|
Application.objects.filter(user=request.user).select_related("job"), pk=pk
|
|
)
|
|
zip_payload, filename = build_application_export(application)
|
|
response = HttpResponse(zip_payload, content_type="application/zip")
|
|
response["Content-Disposition"] = f'attachment; filename="{filename}"'
|
|
return response
|
|
|
|
|
|
@login_required
|
|
def application_print(request, pk: int):
|
|
application = get_object_or_404(
|
|
Application.objects.filter(user=request.user).select_related("job"), pk=pk
|
|
)
|
|
return HttpResponse(build_print_html(application), content_type="text/html; charset=utf-8")
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def application_delete(request, pk: int):
|
|
if Application.objects.filter(pk=pk).exclude(user=request.user).exists():
|
|
raise Http404
|
|
deleted = delete_application_dossier(application_id=pk, user=request.user)
|
|
if deleted:
|
|
messages.success(request, "Sollicitatiedossier verwijderd.")
|
|
else:
|
|
messages.info(request, "Geen dossier verwijderd.")
|
|
return redirect("jobs:applications")
|