391 lines
14 KiB
Python
391 lines
14 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
|
|
|
|
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 = 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(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)
|
|
)
|
|
|
|
# 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', '')
|
|
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})
|
|
|
|
|
|
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(candidate=candidate)
|
|
|
|
# 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,
|
|
})
|
|
|
|
def candidate_update_stage(request, slug):
|
|
"""Handle HTMX stage update requests"""
|
|
try:
|
|
if not request.user.is_staff:
|
|
return render(request, 'recruitment/partials/error.html', {'error': 'Permission denied'}, status=403)
|
|
|
|
candidate = get_object_or_404(models.Candidate, slug=slug)
|
|
|
|
if request.method != 'POST':
|
|
return render(request, 'recruitment/partials/error.html', {'error': 'Only POST method is allowed'}, status=405)
|
|
|
|
# Handle form data
|
|
form = forms.CandidateStageForm(request.POST, candidate=candidate)
|
|
if form.is_valid():
|
|
stage_value = form.cleaned_data['stage']
|
|
|
|
# Validate stage value
|
|
valid_stages = [choice[0] for choice in models.Candidate.Stage.choices]
|
|
if stage_value not in valid_stages:
|
|
return render(request, 'recruitment/partials/error.html', {'error': f'Invalid stage value. Must be one of: {", ".join(valid_stages)}'}, status=400)
|
|
|
|
# Check transition rules
|
|
if candidate.pk and stage_value != candidate.stage:
|
|
old_stage = candidate.stage
|
|
if not candidate.can_transition_to(stage_value):
|
|
return render(request, 'recruitment/partials/error.html', {'error': f'Cannot transition from "{old_stage}" to "{stage_value}". Transition not allowed.'}, status=400)
|
|
|
|
# Update the stage
|
|
old_stage = candidate.stage
|
|
candidate.stage = stage_value
|
|
candidate.save()
|
|
|
|
# Return success template
|
|
context = {
|
|
'form': form,
|
|
'success': True,
|
|
'message': f'Stage updated from "{old_stage}" to "{candidate.stage}"',
|
|
'new_stage': candidate.stage,
|
|
'new_stage_display': candidate.get_stage_display(),
|
|
'candidate': candidate
|
|
}
|
|
messages.success(request,"Candidate Stage Updated")
|
|
return redirect("candidate_detail",slug=candidate.slug)
|
|
def response():
|
|
stage_form = forms.CandidateStageForm(candidate=candidate)
|
|
context['stage_form'] = stage_form
|
|
stage_form_partial = render_to_string('recruitment/partials/stage_update_modal.html#id-stage', context)
|
|
success_html = render_to_string('recruitment/partials/stage_update_success.html', context)
|
|
yield SSE.patch_elements(stage_form_partial,"#id_stage")
|
|
yield SSE.patch_elements(success_html,"#availableStagesInfo")
|
|
yield SSE.patch_signals({'stage':candidate.stage})
|
|
|
|
return DatastarResponse(response())
|
|
# return render(request, 'recruitment/partials/stage_update_success.html', context)
|
|
else:
|
|
# Return form with errors
|
|
context = {
|
|
'form': form,
|
|
'candidate': candidate,
|
|
'stage_form': forms.CandidateStageForm(candidate=candidate)
|
|
}
|
|
return render(request, 'recruitment/partials/stage_update_form.html', context)
|
|
|
|
except Exception as e:
|
|
# Log the error for debugging
|
|
import traceback
|
|
error_details = traceback.format_exc()
|
|
print(f"Error in candidate_update_stage: {error_details}")
|
|
|
|
return render(request, 'partials/error.html', {'error': f'Internal server error: {str(e)}'}, status=500)
|
|
|
|
|
|
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.'
|
|
|
|
|
|
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)
|