import json from django.shortcuts import render, get_object_or_404,redirect from django.contrib import messages from django.http import JsonResponse from django.db.models.fields.json import KeyTextTransform from recruitment.utils import json_to_markdown_table from django.db.models import Count, Avg, F, FloatField from django.db.models.functions import Cast 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_resume_template_view(request, slug): """Display formatted resume template for a candidate""" candidate = get_object_or_404(models.Candidate, slug=slug) if not request.user.is_staff: messages.error(request, _("You don't have permission to view this page.")) return redirect('candidate_list') return render(request, 'recruitment/candidate_resume_template.html', { 'candidate': candidate }) @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): # --- Performance Optimization: Aggregate Data in ONE Query --- # 1. Base Job Query: Get all jobs and annotate with candidate count jobs_with_counts = models.JobPosting.objects.annotate( candidate_count=Count('candidates') ).order_by('-candidate_count') total_jobs = jobs_with_counts.count() total_candidates = models.Candidate.objects.count() job_titles = [job.title for job in jobs_with_counts] job_app_counts = [job.candidate_count for job in jobs_with_counts] average_applications = round(jobs_with_counts.aggregate( avg_apps=Avg('candidate_count') )['avg_apps'] or 0, 2) # 5. New: Candidate Quality & Funnel Metrics # Assuming 'match_score' is a direct IntegerField/FloatField on the Candidate model # (based on the final, optimized version of handle_reume_parsing_and_scoring) # Average Match Score (Overall Quality) candidates_with_score = models.Candidate.objects.filter( # Filter only candidates that have been parsed/scored is_resume_parsed=True ).annotate( score_as_text=KeyTextTransform( 'match_score', KeyTextTransform('scoring_data', F('ai_analysis_data')) ) ).annotate( # Cast the extracted text score to a FloatField so AVG() can operate on it. sortable_score=Cast('score_as_text', output_field=FloatField()) ) # 2b. AGGREGATE using the newly created 'sortable_score' field avg_match_score_result = candidates_with_score.aggregate( avg_score=Avg('sortable_score') )['avg_score'] avg_match_score = round(avg_match_score_result or 0, 1) # 2c. Use the annotated QuerySet for other metrics # Scored Candidates Ratio (Now simpler, as we filtered the QuerySet) total_scored = candidates_with_score.count() scored_ratio = round((total_scored / total_candidates) * 100, 1) if total_candidates > 0 else 0 # High Potential Candidates (Filter the annotated QuerySet) high_potential_count = candidates_with_score.filter( sortable_score__gte=75 ).count() high_potential_ratio = round((high_potential_count / total_candidates) * 100, 1) if total_candidates > 0 else 0 jobs=models.JobPosting.objects.all().order_by('internal_job_id') selected_job_id=request.GET.get('selected_job_id','') candidate_stage=['APPLIED','EXAM','INTERVIEW','OFFER'] apply_count,exam_count,interview_count,offer_count=[0]*4 if selected_job_id: job=jobs.get(internal_job_id=selected_job_id) apply_count=job.screening_candidates_count exam_count=job.exam_candidates_count interview_count=job.interview_candidates_count offer_count=job.offer_candidates_count all_candidates_count=job.all_candidates_count else: #default job job=jobs.first() apply_count=job.screening_candidates_count exam_count=job.exam_candidates_count interview_count=job.interview_candidates_count offer_count=job.offer_candidates_count all_candidates_count=job.all_candidates_count candidates_count=[ apply_count,exam_count,interview_count,offer_count ] context = { 'total_jobs': total_jobs, 'total_candidates': total_candidates, 'average_applications': average_applications, # Chart Data 'job_titles': json.dumps(job_titles), 'job_app_counts': json.dumps(job_app_counts), # New Analytical Metrics (FIXED) 'avg_match_score': avg_match_score, 'high_potential_count': high_potential_count, 'high_potential_ratio': high_potential_ratio, 'scored_ratio': scored_ratio, 'current_job_id':selected_job_id, 'jobs':jobs, 'all_candidates_count':all_candidates_count, 'candidate_stage':json.dumps(candidate_stage), 'candidates_count':json.dumps(candidates_count) ,'my_job':job } 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