418 lines
15 KiB
Python
418 lines
15 KiB
Python
import json
|
|
from django.shortcuts import render, get_object_or_404,redirect
|
|
from django.contrib import messages
|
|
from django.http import JsonResponse
|
|
|
|
from recruitment.utils import json_to_markdown_table
|
|
from . import models
|
|
from django.utils.translation import get_language
|
|
from . import forms
|
|
from django.contrib.auth.decorators import login_required
|
|
import ast
|
|
from django.template.loader import render_to_string
|
|
# from .dashboard import get_dashboard_data
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.contrib.messages.views import SuccessMessageMixin
|
|
from django.views.generic import ListView, CreateView, UpdateView, DeleteView, DetailView
|
|
# JobForm removed - using JobPostingForm instead
|
|
from django.urls import reverse_lazy
|
|
from django.db.models import Q, Count, Avg
|
|
from django.db.models import FloatField
|
|
|
|
from datastar_py.django import (
|
|
DatastarResponse,
|
|
ServerSentEventGenerator as SSE,
|
|
read_signals,
|
|
)
|
|
# from rich import print
|
|
from rich.markdown import CodeBlock
|
|
|
|
class JobListView(LoginRequiredMixin, ListView):
|
|
model = models.JobPosting
|
|
template_name = 'jobs/job_list.html'
|
|
context_object_name = 'jobs'
|
|
paginate_by = 10
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset().order_by('-created_at')
|
|
|
|
# Handle search
|
|
search_query = self.request.GET.get('search', '')
|
|
if search_query:
|
|
queryset = queryset.filter(
|
|
Q(title__icontains=search_query) |
|
|
Q(description__icontains=search_query) |
|
|
Q(department__icontains=search_query)
|
|
)
|
|
|
|
# Filter for non-staff users
|
|
if not self.request.user.is_staff:
|
|
queryset = queryset.filter(status='Published')
|
|
|
|
status=self.request.GET.get('status')
|
|
if status:
|
|
queryset=queryset.filter(status=status)
|
|
|
|
return queryset
|
|
|
|
def get_context_data(self, **kwargs):
|
|
|
|
context = super().get_context_data(**kwargs)
|
|
context['search_query'] = self.request.GET.get('search', '')
|
|
context['lang'] = get_language()
|
|
return context
|
|
|
|
|
|
class JobCreateView(LoginRequiredMixin, SuccessMessageMixin, CreateView):
|
|
model = models.JobPosting
|
|
form_class = forms.JobPostingForm
|
|
template_name = 'jobs/create_job.html'
|
|
success_url = reverse_lazy('job_list')
|
|
success_message = 'Job created successfully.'
|
|
|
|
|
|
class JobUpdateView(LoginRequiredMixin, SuccessMessageMixin, UpdateView):
|
|
model = models.JobPosting
|
|
form_class = forms.JobPostingForm
|
|
template_name = 'jobs/edit_job.html'
|
|
success_url = reverse_lazy('job_list')
|
|
success_message = 'Job updated successfully.'
|
|
slug_url_kwarg = 'slug'
|
|
|
|
|
|
class JobDeleteView(LoginRequiredMixin, SuccessMessageMixin, DeleteView):
|
|
model = models.JobPosting
|
|
template_name = 'jobs/partials/delete_modal.html'
|
|
success_url = reverse_lazy('job_list')
|
|
success_message = 'Job deleted successfully.'
|
|
slug_url_kwarg = 'slug'
|
|
|
|
class JobCandidatesListView(LoginRequiredMixin, ListView):
|
|
model = models.Candidate
|
|
template_name = 'jobs/job_candidates_list.html'
|
|
context_object_name = 'candidates'
|
|
paginate_by = 10
|
|
|
|
|
|
|
|
def get_queryset(self):
|
|
# Get the job by slug
|
|
self.job = get_object_or_404(models.JobPosting, slug=self.kwargs['slug'])
|
|
|
|
# Filter candidates for this specific job
|
|
queryset = models.Candidate.objects.filter(job=self.job)
|
|
|
|
if self.request.GET.get('stage'):
|
|
stage=self.request.GET.get('stage')
|
|
queryset=queryset.filter(stage=stage)
|
|
|
|
|
|
# Handle search
|
|
search_query = self.request.GET.get('search', '')
|
|
if search_query:
|
|
queryset = queryset.filter(
|
|
Q(first_name__icontains=search_query) |
|
|
Q(last_name__icontains=search_query) |
|
|
Q(email__icontains=search_query) |
|
|
Q(phone__icontains=search_query) |
|
|
Q(stage__icontains=search_query)
|
|
)
|
|
|
|
# Filter for non-staff users
|
|
if not self.request.user.is_staff:
|
|
return models.Candidate.objects.none() # Restrict for non-staff
|
|
|
|
return queryset.order_by('-created_at')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context['search_query'] = self.request.GET.get('search', '')
|
|
context['job'] = getattr(self, 'job', None)
|
|
return context
|
|
|
|
|
|
class CandidateListView(LoginRequiredMixin, ListView):
|
|
model = models.Candidate
|
|
template_name = 'recruitment/candidate_list.html'
|
|
context_object_name = 'candidates'
|
|
paginate_by = 100
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset()
|
|
|
|
# Handle search
|
|
search_query = self.request.GET.get('search', '')
|
|
job = self.request.GET.get('job', '')
|
|
stage = self.request.GET.get('stage', '')
|
|
if search_query:
|
|
queryset = queryset.filter(
|
|
Q(first_name__icontains=search_query) |
|
|
Q(last_name__icontains=search_query) |
|
|
Q(email__icontains=search_query) |
|
|
Q(phone__icontains=search_query) |
|
|
Q(stage__icontains=search_query) |
|
|
Q(job__title__icontains=search_query)
|
|
)
|
|
if job:
|
|
queryset = queryset.filter(job__slug=job)
|
|
if stage:
|
|
queryset = queryset.filter(stage=stage)
|
|
# Filter for non-staff users
|
|
if not self.request.user.is_staff:
|
|
return models.Candidate.objects.none() # Restrict for non-staff
|
|
|
|
return queryset.order_by('-created_at')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context['search_query'] = self.request.GET.get('search', '')
|
|
context['job_filter'] = self.request.GET.get('job', '')
|
|
context['stage_filter'] = self.request.GET.get('stage', '')
|
|
context['available_jobs'] = models.JobPosting.objects.all().order_by('created_at').distinct()
|
|
return context
|
|
|
|
|
|
class CandidateCreateView(LoginRequiredMixin, SuccessMessageMixin, CreateView):
|
|
model = models.Candidate
|
|
form_class = forms.CandidateForm
|
|
template_name = 'recruitment/candidate_create.html'
|
|
success_url = reverse_lazy('candidate_list')
|
|
success_message = 'Candidate created successfully.'
|
|
|
|
def get_initial(self):
|
|
initial = super().get_initial()
|
|
if 'slug' in self.kwargs:
|
|
job = get_object_or_404(models.JobPosting, slug=self.kwargs['slug'])
|
|
initial['job'] = job
|
|
return initial
|
|
|
|
def form_valid(self, form):
|
|
if 'slug' in self.kwargs:
|
|
job = get_object_or_404(models.JobPosting, slug=self.kwargs['slug'])
|
|
form.instance.job = job
|
|
return super().form_valid(form)
|
|
|
|
|
|
class CandidateUpdateView(LoginRequiredMixin, SuccessMessageMixin, UpdateView):
|
|
model = models.Candidate
|
|
form_class = forms.CandidateForm
|
|
template_name = 'recruitment/candidate_update.html'
|
|
success_url = reverse_lazy('candidate_list')
|
|
success_message = 'Candidate updated successfully.'
|
|
slug_url_kwarg = 'slug'
|
|
|
|
|
|
class CandidateDeleteView(LoginRequiredMixin, SuccessMessageMixin, DeleteView):
|
|
model = models.Candidate
|
|
template_name = 'recruitment/candidate_delete.html'
|
|
success_url = reverse_lazy('candidate_list')
|
|
success_message = 'Candidate deleted successfully.'
|
|
slug_url_kwarg = 'slug'
|
|
|
|
|
|
# def job_detail(request, slug):
|
|
# job = get_object_or_404(models.JobPosting, slug=slug, status='Published')
|
|
# form = forms.CandidateForm()
|
|
|
|
# return render(request, 'jobs/job_detail.html', {'job': job, 'form': form})
|
|
|
|
|
|
|
|
|
|
@login_required
|
|
def training_list(request):
|
|
materials = models.TrainingMaterial.objects.all().order_by('-created_at')
|
|
return render(request, 'recruitment/training_list.html', {'materials': materials})
|
|
|
|
|
|
@login_required
|
|
def candidate_detail(request, slug):
|
|
from rich.json import JSON
|
|
candidate = get_object_or_404(models.Candidate, slug=slug)
|
|
try:
|
|
parsed = ast.literal_eval(candidate.parsed_summary)
|
|
except:
|
|
parsed = {}
|
|
|
|
# Create stage update form for staff users
|
|
stage_form = None
|
|
if request.user.is_staff:
|
|
stage_form = forms.CandidateStageForm()
|
|
|
|
# parsed = JSON(json.dumps(parsed), indent=2, highlight=True, skip_keys=False, ensure_ascii=False, check_circular=True, allow_nan=True, default=None, sort_keys=False)
|
|
# parsed = json_to_markdown_table([parsed])
|
|
return render(request, 'recruitment/candidate_detail.html', {
|
|
'candidate': candidate,
|
|
'parsed': parsed,
|
|
'stage_form': stage_form,
|
|
})
|
|
|
|
@login_required
|
|
def candidate_update_stage(request, slug):
|
|
"""Handle HTMX stage update requests"""
|
|
candidate = get_object_or_404(models.Candidate, slug=slug)
|
|
form = forms.CandidateStageForm(request.POST, instance=candidate)
|
|
if form.is_valid():
|
|
stage_value = form.cleaned_data['stage']
|
|
candidate.stage = stage_value
|
|
candidate.save(update_fields=['stage'])
|
|
messages.success(request,"Candidate Stage Updated")
|
|
return redirect("candidate_detail",slug=candidate.slug)
|
|
|
|
class TrainingListView(LoginRequiredMixin, ListView):
|
|
model = models.TrainingMaterial
|
|
template_name = 'recruitment/training_list.html'
|
|
context_object_name = 'materials'
|
|
paginate_by = 10
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset()
|
|
|
|
# Handle search
|
|
search_query = self.request.GET.get('search', '')
|
|
if search_query:
|
|
queryset = queryset.filter(
|
|
Q(title__icontains=search_query)
|
|
)
|
|
|
|
# Filter for non-staff users
|
|
if not self.request.user.is_staff:
|
|
return models.TrainingMaterial.objects.none() # Restrict for non-staff
|
|
|
|
return queryset.filter(created_by=self.request.user).order_by('-created_at')
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context['search_query'] = self.request.GET.get('search', '')
|
|
return context
|
|
|
|
|
|
class TrainingCreateView(LoginRequiredMixin, SuccessMessageMixin, CreateView):
|
|
model = models.TrainingMaterial
|
|
form_class = forms.TrainingMaterialForm
|
|
template_name = 'recruitment/training_create.html'
|
|
success_url = reverse_lazy('training_list')
|
|
success_message = 'Training material created successfully.'
|
|
|
|
def form_valid(self, form):
|
|
form.instance.created_by = self.request.user
|
|
return super().form_valid(form)
|
|
|
|
|
|
class TrainingUpdateView(LoginRequiredMixin, SuccessMessageMixin, UpdateView):
|
|
model = models.TrainingMaterial
|
|
form_class = forms.TrainingMaterialForm
|
|
template_name = 'recruitment/training_update.html'
|
|
success_url = reverse_lazy('training_list')
|
|
success_message = 'Training material updated successfully.'
|
|
slug_url_kwarg = 'slug'
|
|
|
|
|
|
class TrainingDetailView(LoginRequiredMixin, DetailView):
|
|
model = models.TrainingMaterial
|
|
template_name = 'recruitment/training_detail.html'
|
|
context_object_name = 'material'
|
|
slug_url_kwarg = 'slug'
|
|
|
|
class TrainingDeleteView(LoginRequiredMixin, SuccessMessageMixin, DeleteView):
|
|
model = models.TrainingMaterial
|
|
template_name = 'recruitment/training_delete.html'
|
|
success_url = reverse_lazy('training_list')
|
|
success_message = 'Training material deleted successfully.'
|
|
|
|
|
|
@login_required
|
|
def dashboard_view(request):
|
|
total_jobs = models.JobPosting.objects.count()
|
|
total_candidates = models.Candidate.objects.count()
|
|
jobs = models.JobPosting.objects.all()
|
|
|
|
job_titles = [job.title for job in jobs]
|
|
job_app_counts = [job.candidates.count() for job in jobs]
|
|
average_applications = round(sum(job_app_counts) / total_jobs, 2) if total_jobs > 0 else 0
|
|
|
|
context = {
|
|
'total_jobs': total_jobs,
|
|
'total_candidates': total_candidates,
|
|
'job_titles': job_titles,
|
|
'job_app_counts': job_app_counts,
|
|
'average_applications': average_applications,
|
|
}
|
|
return render(request, 'recruitment/dashboard.html', context)
|
|
|
|
|
|
@login_required
|
|
def candidate_offer_view(request, slug):
|
|
"""View for candidates in the Offer stage"""
|
|
job = get_object_or_404(models.JobPosting, slug=slug)
|
|
|
|
# Filter candidates for this specific job and stage
|
|
candidates = job.offer_candidates
|
|
|
|
# Handle search
|
|
search_query = request.GET.get('search', '')
|
|
if search_query:
|
|
candidates = candidates.filter(
|
|
Q(first_name__icontains=search_query) |
|
|
Q(last_name__icontains=search_query) |
|
|
Q(email__icontains=search_query) |
|
|
Q(phone__icontains=search_query)
|
|
)
|
|
|
|
candidates = candidates.order_by('-created_at')
|
|
|
|
context = {
|
|
'job': job,
|
|
'candidates': candidates,
|
|
'search_query': search_query,
|
|
'current_stage': 'Offer',
|
|
}
|
|
return render(request, 'recruitment/candidate_offer_view.html', context)
|
|
|
|
|
|
@login_required
|
|
def update_candidate_status(request, job_slug, candidate_slug, stage_type, status):
|
|
"""Handle exam/interview/offer status updates"""
|
|
from django.utils import timezone
|
|
|
|
job = get_object_or_404(models.JobPosting, slug=job_slug)
|
|
candidate = get_object_or_404(models.Candidate, slug=candidate_slug, job=job)
|
|
print(stage_type,status)
|
|
|
|
if request.method == "POST":
|
|
if stage_type == 'exam':
|
|
candidate.exam_status = status
|
|
candidate.exam_date = timezone.now()
|
|
candidate.save(update_fields=['exam_status', 'exam_date'])
|
|
elif stage_type == 'interview':
|
|
candidate.interview_status = status
|
|
candidate.interview_date = timezone.now()
|
|
candidate.save(update_fields=['interview_status', 'interview_date'])
|
|
elif stage_type == 'offer':
|
|
candidate.offer_status = status
|
|
candidate.offer_date = timezone.now()
|
|
candidate.save(update_fields=['offer_status', 'offer_date'])
|
|
messages.success(request, f"Candidate {status} successfully!")
|
|
else:
|
|
messages.error(request, "No changes made.")
|
|
|
|
if stage_type == 'exam':
|
|
return redirect('candidate_exam_view', job.slug)
|
|
elif stage_type == 'interview':
|
|
return redirect('candidate_interview_view', job.slug)
|
|
elif stage_type == 'offer':
|
|
return redirect('candidate_offer_view', job.slug)
|
|
|
|
return redirect('candidate_detail', candidate.slug)
|
|
else:
|
|
if stage_type == 'exam':
|
|
return render(request,"includes/candidate_update_exam_form.html",{'candidate':candidate,'job':job})
|
|
elif stage_type == 'interview':
|
|
return render(request,"includes/candidate_update_interview_form.html",{'candidate':candidate,'job':job})
|
|
elif stage_type == 'offer':
|
|
return render(request,"includes/candidate_update_offer_form.html",{'candidate':candidate,'job':job})
|
|
|
|
|
|
# Removed incorrect JobDetailView class.
|
|
# The job_detail view is handled by function-based view in recruitment.views
|