import json import csv from datetime import datetime from django.shortcuts import render, get_object_or_404,redirect from django.contrib import messages from django.http import JsonResponse, HttpResponse 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 django.db.models.functions import Coalesce, Cast, Replace, NullIf 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 django.utils.text import slugify # 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 FloatField from django.db.models import F, IntegerField, Count, Avg, Sum, Q, ExpressionWrapper, fields, Value,CharField from django.db.models.functions import Cast, Coalesce, TruncDate from django.contrib.auth.decorators import login_required from django.shortcuts import render from django.utils import timezone from datetime import timedelta import json # Add imports for user type restrictions from recruitment.decorators import StaffRequiredMixin, staff_user_required from datastar_py.django import ( DatastarResponse, ServerSentEventGenerator as SSE, read_signals, ) # from rich import print from rich.markdown import CodeBlock class JobListView(LoginRequiredMixin, StaffRequiredMixin, 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, StaffRequiredMixin, 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, StaffRequiredMixin, 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, StaffRequiredMixin, 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 JobApplicationListView(LoginRequiredMixin, StaffRequiredMixin, ListView): model = models.Application template_name = 'jobs/job_candidates_list.html' context_object_name = 'applications' 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.Application.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.Application.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 ApplicationListView(LoginRequiredMixin, StaffRequiredMixin, ListView): model = models.Application template_name = 'recruitment/candidate_list.html' context_object_name = 'applications' 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.Application.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 ApplicationCreateView(LoginRequiredMixin, StaffRequiredMixin, SuccessMessageMixin, CreateView): model = models.Application form_class = forms.ApplicationForm 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) def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) if self.request.method == 'GET': context['person_form'] = forms.PersonForm() return context class ApplicationUpdateView(LoginRequiredMixin, StaffRequiredMixin, SuccessMessageMixin, UpdateView): model = models.Application form_class = forms.ApplicationForm template_name = 'recruitment/candidate_update.html' success_url = reverse_lazy('candidate_list') success_message = 'Candidate updated successfully.' slug_url_kwarg = 'slug' class ApplicationDeleteView(LoginRequiredMixin, StaffRequiredMixin, SuccessMessageMixin, DeleteView): model = models.Application template_name = 'recruitment/candidate_delete.html' success_url = reverse_lazy('candidate_list') success_message = 'Candidate deleted successfully.' slug_url_kwarg = 'slug' def retry_scoring_view(request,slug): from django_q.tasks import async_task application = get_object_or_404(models.Application, slug=slug) async_task( 'recruitment.tasks.handle_reume_parsing_and_scoring', application.pk, hook='recruitment.hooks.callback_ai_parsing', sync=True, ) return redirect('candidate_detail', slug=application.slug) @login_required @staff_user_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 @staff_user_required def candidate_detail(request, slug): from rich.json import JSON candidate = get_object_or_404(models.Application, 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.ApplicationStageForm() # 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 @staff_user_required def candidate_resume_template_view(request, slug): """Display formatted resume template for a candidate""" application = get_object_or_404(models.Application, 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', { 'application': application }) @login_required @staff_user_required def candidate_update_stage(request, slug): """Handle HTMX stage update requests""" application = get_object_or_404(models.Application, slug=slug) form = forms.ApplicationStageForm(request.POST, instance=application) if form.is_valid(): stage_value = form.cleaned_data['stage'] application.stage = stage_value application.save(update_fields=['stage']) messages.success(request,"application Stage Updated") return redirect("candidate_detail",slug=application.slug) class TrainingListView(LoginRequiredMixin, StaffRequiredMixin, 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, StaffRequiredMixin, 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, StaffRequiredMixin, 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, StaffRequiredMixin, DetailView): model = models.TrainingMaterial template_name = 'recruitment/training_detail.html' context_object_name = 'material' slug_url_kwarg = 'slug' class TrainingDeleteView(LoginRequiredMixin, StaffRequiredMixin, SuccessMessageMixin, DeleteView): model = models.TrainingMaterial template_name = 'recruitment/training_delete.html' success_url = reverse_lazy('training_list') success_message = 'Training material deleted successfully.' # IMPORTANT: Ensure 'models' correctly refers to your Django models file # Example: from . import models # --- Constants --- SCORE_PATH = 'ai_analysis_data__analysis_data__match_score' HIGH_POTENTIAL_THRESHOLD = 75 MAX_TIME_TO_HIRE_DAYS = 90 TARGET_TIME_TO_HIRE_DAYS = 45 # Used for the template visualization @login_required @staff_user_required def dashboard_view(request): selected_job_pk = request.GET.get('selected_job_pk') today = timezone.now().date() # --- 1. BASE QUERYSETS & GLOBAL METRICS (UNFILTERED) --- all_jobs_queryset = models.JobPosting.objects.all().order_by('-created_at') all_candidates_queryset = models.Application.objects.all() # Global KPI Card Metrics total_jobs_global = all_jobs_queryset.count() total_participants = models.Participants.objects.count() total_jobs_posted_linkedin = all_jobs_queryset.filter(linkedin_post_id__isnull=False).count() # Data for Job App Count Chart (always for ALL jobs) job_titles = [job.title for job in all_jobs_queryset] job_app_counts = [job.applications.count() for job in all_jobs_queryset] # --- 2. TIME SERIES: GLOBAL DAILY APPLICANTS --- # Group ALL candidates by creation date global_daily_applications_qs = all_candidates_queryset.annotate( date=TruncDate('created_at') ).values('date').annotate( count=Count('pk') ).order_by('date') global_dates = [item['date'].strftime('%Y-%m-%d') for item in global_daily_applications_qs] global_counts = [item['count'] for item in global_daily_applications_qs] # --- 3. FILTERING LOGIC: Determine the scope for scoped metrics --- candidate_queryset = all_candidates_queryset job_scope_queryset = all_jobs_queryset interview_queryset = models.ScheduledInterview.objects.all() current_job = None if selected_job_pk: # Filter all base querysets candidate_queryset = candidate_queryset.filter(job__pk=selected_job_pk) interview_queryset = interview_queryset.filter(job__pk=selected_job_pk) try: current_job = all_jobs_queryset.get(pk=selected_job_pk) job_scope_queryset = models.JobPosting.objects.filter(pk=selected_job_pk) except models.JobPosting.DoesNotExist: pass # --- 4. TIME SERIES: SCOPED DAILY APPLICANTS --- # Only run if a specific job is selected scoped_dates = [] scoped_counts = [] if selected_job_pk: scoped_daily_applications_qs = candidate_queryset.annotate( date=TruncDate('created_at') ).values('date').annotate( count=Count('pk') ).order_by('date') scoped_dates = [item['date'].strftime('%Y-%m-%d') for item in scoped_daily_applications_qs] scoped_counts = [item['count'] for item in scoped_daily_applications_qs] # --- 5. SCOPED CORE AGGREGATIONS (FILTERED OR ALL) --- total_candidates = candidate_queryset.count() candidates_with_score_query = candidate_queryset.filter( is_resume_parsed=True ).annotate( annotated_match_score=Coalesce( Cast(SCORE_PATH, output_field=IntegerField()), 0 ) ) # safe_match_score_cast = Cast( # # 3. If the result after stripping quotes is an empty string (''), convert it to NULL. # NullIf( # # 2. Use Replace to remove the literal double quotes (") that might be present. # Replace( # # 1. Use the double-underscore path (which uses the ->> operator for the final value) # # and cast to CharField for text-based cleanup functions. # Cast(SCORE_PATH, output_field=CharField()), # Value('"'), Value('') # Replace the double quote character with an empty string # ), # Value('') # Value to check for (empty string) # ), # output_field=IntegerField() # 4. Cast the clean, non-empty string (or NULL) to an integer. # ) # candidates_with_score_query= candidate_queryset.filter(is_resume_parsed=True).annotate( # # The Coalesce handles NULL values (from missing data, non-numeric data, or NullIf) and sets them to 0. # annotated_match_score=Coalesce(safe_match_score_cast, Value(0)) # ) # A. Pipeline & Volume Metrics (Scoped) total_active_jobs = job_scope_queryset.filter(status="ACTIVE").count() last_week = timezone.now() - timedelta(days=7) new_candidates_7days = candidate_queryset.filter(created_at__gte=last_week).count() open_positions_agg = job_scope_queryset.filter(status="ACTIVE").aggregate(total_open=Sum('open_positions')) total_open_positions = open_positions_agg['total_open'] or 0 average_applications_result = job_scope_queryset.annotate( candidate_count=Count('applications', distinct=True) ).aggregate(avg_apps=Avg('candidate_count'))['avg_apps'] average_applications = round(average_applications_result or 0, 2) # B. Efficiency & Conversion Metrics (Scoped) hired_candidates = candidate_queryset.filter( stage='Hired' ) lst=[c.time_to_hire_days for c in hired_candidates] time_to_hire_query = hired_candidates.annotate( time_diff=ExpressionWrapper( F('join_date') - F('created_at__date'), output_field=fields.DurationField() ) ).aggregate(avg_time_to_hire=Avg('time_diff')) print(time_to_hire_query) avg_time_to_hire_days = ( time_to_hire_query.get('avg_time_to_hire').days if time_to_hire_query.get('avg_time_to_hire') else 0 ) print(avg_time_to_hire_days) applied_count = candidate_queryset.filter(stage='Applied').count() advanced_count = candidate_queryset.filter(stage__in=['Exam', 'Interview', 'Offer']).count() screening_pass_rate = round( (advanced_count / applied_count) * 100, 1 ) if applied_count > 0 else 0 offers_extended_count = candidate_queryset.filter(stage='Offer').count() offers_accepted_count = candidate_queryset.filter(offer_status='Accepted').count() offers_accepted_rate = round( (offers_accepted_count / offers_extended_count) * 100, 1 ) if offers_extended_count > 0 else 0 filled_positions = offers_accepted_count vacancy_fill_rate = round( (filled_positions / total_open_positions) * 100, 1 ) if total_open_positions > 0 else 0 # C. Activity & Quality Metrics (Scoped) current_year, current_week, _ = today.isocalendar() meetings_scheduled_this_week = interview_queryset.filter( interview_date__week=current_week, interview_date__year=current_year ).count() avg_match_score_result = candidates_with_score_query.aggregate(avg_score=Avg('annotated_match_score'))['avg_score'] avg_match_score = round(avg_match_score_result or 0, 1) high_potential_count = candidates_with_score_query.filter(annotated_match_score__gte=HIGH_POTENTIAL_THRESHOLD).count() high_potential_ratio = round( (high_potential_count / total_candidates) * 100, 1 ) if total_candidates > 0 else 0 total_scored_candidates = candidates_with_score_query.count() scored_ratio = round( (total_scored_candidates / total_candidates) * 100, 1 ) if total_candidates > 0 else 0 # --- 6. CHART DATA PREPARATION --- # A. Pipeline Funnel (Scoped) stage_counts = candidate_queryset.values('stage').annotate(count=Count('stage')) stage_map = {item['stage']: item['count'] for item in stage_counts} candidate_stage = ['Applied', 'Exam', 'Interview', 'Offer', 'Hired'] candidates_count = [ stage_map.get('Applied', 0), stage_map.get('Exam', 0), stage_map.get('Interview', 0), stage_map.get('Offer', 0), stage_map.get('Hired',0) ] # --- 7. GAUGE CHART CALCULATION (Time-to-Hire) --- current_days = avg_time_to_hire_days rotation_percent = current_days / MAX_TIME_TO_HIRE_DAYS if MAX_TIME_TO_HIRE_DAYS > 0 else 0 rotation_degrees = rotation_percent * 180 rotation_degrees_final = round(min(rotation_degrees, 180), 1) # Ensure max 180 degrees # hiring_source_counts = candidate_queryset.values('hiring_source').annotate(count=Count('stage')) source_map= {item['hiring_source']: item['count'] for item in hiring_source_counts} candidates_count_in_each_source = [ source_map.get('Public', 0), source_map.get('Internal', 0), source_map.get('Agency', 0), ] all_hiring_sources=["Public", "Internal", "Agency"] # --- 8. CONTEXT RETURN --- context = { # Global KPIs 'total_jobs_global': total_jobs_global, 'total_participants': total_participants, 'total_jobs_posted_linkedin': total_jobs_posted_linkedin, # Scoped KPIs 'total_active_jobs': total_active_jobs, 'total_candidates': total_candidates, 'new_candidates_7days': new_candidates_7days, 'total_open_positions': total_open_positions, 'average_applications': average_applications, 'avg_time_to_hire_days': avg_time_to_hire_days, 'screening_pass_rate': screening_pass_rate, 'offers_accepted_rate': offers_accepted_rate, 'vacancy_fill_rate': vacancy_fill_rate, 'meetings_scheduled_this_week': meetings_scheduled_this_week, 'avg_match_score': avg_match_score, 'high_potential_count': high_potential_count, 'high_potential_ratio': high_potential_ratio, 'scored_ratio': scored_ratio, # Chart Data 'candidate_stage': json.dumps(candidate_stage), 'candidates_count': json.dumps(candidates_count), 'job_titles': json.dumps(job_titles), 'job_app_counts': json.dumps(job_app_counts), # 'source_volume_chart_data' is intentionally REMOVED # Time Series Data 'global_dates': json.dumps(global_dates), 'global_counts': json.dumps(global_counts), 'scoped_dates': json.dumps(scoped_dates), 'scoped_counts': json.dumps(scoped_counts), 'is_job_scoped': bool(selected_job_pk), # Gauge Data 'gauge_max_days': MAX_TIME_TO_HIRE_DAYS, 'gauge_target_days': TARGET_TIME_TO_HIRE_DAYS, 'gauge_rotation_degrees': rotation_degrees_final, # UI Control 'jobs': all_jobs_queryset, 'current_job_id': selected_job_pk, 'current_job': current_job, 'candidates_count_in_each_source': json.dumps(candidates_count_in_each_source), 'all_hiring_sources': json.dumps(all_hiring_sources), } return render(request, 'recruitment/dashboard.html', context) @login_required @staff_user_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 @staff_user_required def candidate_hired_view(request, slug): """View for hired candidates""" job = get_object_or_404(models.JobPosting, slug=slug) # Filter candidates with offer_status = 'Accepted' candidates = job.hired_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': 'Hired', } return render(request, 'recruitment/candidate_hired_view.html', context) @login_required @staff_user_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.Application, slug=candidate_slug, job=job) print(stage_type) print(status) print(request.method) 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']) return render(request,'recruitment/partials/exam-results.html',{'candidate':candidate,'job':job}) elif stage_type == 'interview': candidate.interview_status = status candidate.interview_date = timezone.now() candidate.save(update_fields=['interview_status', 'interview_date']) return render(request,'recruitment/partials/interview-results.html',{'candidate':candidate,'job':job}) elif stage_type == 'offer': candidate.offer_status = status candidate.offer_date = timezone.now() candidate.save(update_fields=['offer_status', 'offer_date']) return render(request,'recruitment/partials/offer-results.html',{'candidate':candidate,'job':job}) 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}) # Stage configuration for CSV export STAGE_CONFIG = { 'screening': { 'filter': {'stage': 'Applied'}, 'fields': ['name', 'email', 'phone', 'created_at', 'stage', 'ai_score', 'years_experience', 'screening_rating', 'professional_category', 'top_skills', 'strengths', 'weaknesses'], 'headers': ['Name', 'Email', 'Phone', 'Application Date', 'Screening Status', 'Match Score', 'Years Experience', 'Screening Rating', 'Professional Category', 'Top 3 Skills', 'Strengths', 'Weaknesses'] }, 'exam': { 'filter': {'stage': 'Exam'}, 'fields': ['name', 'email', 'phone', 'created_at', 'exam_status', 'exam_date', 'ai_score', 'years_experience', 'screening_rating'], 'headers': ['Name', 'Email', 'Phone', 'Application Date', 'Exam Status', 'Exam Date', 'Match Score', 'Years Experience', 'Screening Rating'] }, 'interview': { 'filter': {'stage': 'Interview'}, 'fields': ['name', 'email', 'phone', 'created_at', 'interview_status', 'interview_date', 'ai_score', 'years_experience', 'professional_category', 'top_skills'], 'headers': ['Name', 'Email', 'Phone', 'Application Date', 'Interview Status', 'Interview Date', 'Match Score', 'Years Experience', 'Professional Category', 'Top 3 Skills'] }, 'offer': { 'filter': {'stage': 'Offer'}, 'fields': ['name', 'email', 'phone', 'created_at', 'offer_status', 'offer_date', 'ai_score', 'years_experience', 'professional_category'], 'headers': ['Name', 'Email', 'Phone', 'Application Date', 'Offer Status', 'Offer Date', 'Match Score', 'Years Experience', 'Professional Category'] }, 'hired': { 'filter': {'offer_status': 'Accepted'}, 'fields': ['name', 'email', 'phone', 'created_at', 'offer_date', 'ai_score', 'years_experience', 'professional_category', 'join_date'], 'headers': ['Name', 'Email', 'Phone', 'Application Date', 'Hire Date', 'Match Score', 'Years Experience', 'Professional Category', 'Join Date'] } } @login_required @staff_user_required def export_candidates_csv(request, job_slug, stage): """Export candidates for a specific stage as CSV""" job = get_object_or_404(models.JobPosting, slug=job_slug) # Validate stage if stage not in STAGE_CONFIG: messages.error(request, "Invalid stage specified for export.") return redirect('job_detail', job.slug) config = STAGE_CONFIG[stage] # Filter candidates based on stage if stage == 'hired': candidates = job.applications.filter(**config['filter']) else: candidates = job.applications.filter(**config['filter']) # Handle search if provided 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') # Create CSV response response = HttpResponse(content_type='text/csv') filename = f"{slugify(job.title)}_{stage}_{datetime.now().strftime('%Y-%m-%d')}.csv" response['Content-Disposition'] = f'attachment; filename="{filename}"' # Write UTF-8 BOM for Excel compatibility response.write('\ufeff') writer = csv.writer(response) # Write headers headers = config['headers'].copy() headers.extend(['Job Title', 'Department']) writer.writerow(headers) # Write candidate data for candidate in candidates: row = [] # Extract data based on stage configuration for field in config['fields']: if field == 'name': row.append(candidate.name) elif field == 'email': row.append(candidate.email) elif field == 'phone': row.append(candidate.phone) elif field == 'created_at': row.append(candidate.created_at.strftime('%Y-%m-%d %H:%M') if candidate.created_at else '') elif field == 'stage': row.append(candidate.stage or '') elif field == 'exam_status': row.append(candidate.exam_status or '') elif field == 'exam_date': row.append(candidate.exam_date.strftime('%Y-%m-%d %H:%M') if candidate.exam_date else '') elif field == 'interview_status': row.append(candidate.interview_status or '') elif field == 'interview_date': row.append(candidate.interview_date.strftime('%Y-%m-%d %H:%M') if candidate.interview_date else '') elif field == 'offer_status': row.append(candidate.offer_status or '') elif field == 'offer_date': row.append(candidate.offer_date.strftime('%Y-%m-%d %H:%M') if candidate.offer_date else '') elif field == 'ai_score': # Extract AI score using model property try: score = candidate.match_score row.append(f"{score}%" if score else '') except: row.append('') elif field == 'years_experience': # Extract years of experience using model property try: years = candidate.years_of_experience row.append(f"{years}" if years else '') except: row.append('') elif field == 'screening_rating': # Extract screening rating using model property try: rating = candidate.screening_stage_rating row.append(rating if rating else '') except: row.append('') elif field == 'professional_category': # Extract professional category using model property try: category = candidate.professional_category row.append(category if category else '') except: row.append('') elif field == 'top_skills': # Extract top 3 skills using model property try: skills = candidate.top_3_keywords row.append(', '.join(skills) if skills else '') except: row.append('') elif field == 'strengths': # Extract strengths using model property try: strengths = candidate.strengths row.append(strengths if strengths else '') except: row.append('') elif field == 'weaknesses': # Extract weaknesses using model property try: weaknesses = candidate.weaknesses row.append(weaknesses if weaknesses else '') except: row.append('') elif field == 'join_date': row.append(candidate.join_date.strftime('%Y-%m-%d') if candidate.join_date else '') else: row.append(getattr(candidate, field, '')) # Add job information row.extend([job.title, job.department or '']) writer.writerow(row) return response # Removed incorrect # The job_detail view is handled by function-based view in recruitment.views @login_required @staff_user_required def sync_hired_candidates(request, job_slug): """Sync hired candidates to external sources using Django-Q""" from django_q.tasks import async_task from .tasks import sync_hired_candidates_task if request.method == 'POST': job = get_object_or_404(models.JobPosting, slug=job_slug) try: # Enqueue sync task to Django-Q for background processing task_id = async_task( sync_hired_candidates_task, job_slug, group=f"sync_job_{job_slug}", timeout=300 # 5 minutes timeout ) print("task_id",task_id) # Return immediate response with task ID for tracking return JsonResponse({ 'status': 'queued', 'message': 'Sync task has been queued for background processing', 'task_id': task_id }) except Exception as e: return JsonResponse({ 'status': 'error', 'message': f'Failed to queue sync task: {str(e)}' }, status=500) # For GET requests, return error return JsonResponse({ 'status': 'error', 'message': 'Only POST requests are allowed' }, status=405) @login_required @staff_user_required def test_source_connection(request, source_id): """Test connection to an external source""" from .candidate_sync_service import CandidateSyncService if request.method == 'POST': source = get_object_or_404(models.Source, id=source_id) try: # Initialize sync service sync_service = CandidateSyncService() # Test connection result = sync_service.test_source_connection(source) # Return JSON response return JsonResponse({ 'status': 'success', 'result': result }) except Exception as e: return JsonResponse({ 'status': 'error', 'message': f'Connection test failed: {str(e)}' }, status=500) # For GET requests, return error return JsonResponse({ 'status': 'error', 'message': 'Only POST requests are allowed' }, status=405) @login_required @staff_user_required def sync_task_status(request, task_id): """Check the status of a sync task""" from django_q.models import Task try: # Get the task from Django-Q task = Task.objects.get(pk=task_id) print("task",task) # Determine status based on task state if task.success: status = 'completed' message = 'Sync completed successfully' result = task.result elif task.stopped: status = 'failed' message = 'Sync task failed or was stopped' result = task.result elif task.started: status = 'running' message = 'Sync is currently running' result = None else: status = 'pending' message = 'Sync task is queued and waiting to start' result = None print("result",result) return JsonResponse({ 'status': status, 'message': message, 'result': result, 'task_id': task_id, 'started': task.started, 'stopped': task.stopped, 'success': task.success }) except Task.DoesNotExist: return JsonResponse({ 'status': 'error', 'message': 'Task not found' }, status=404) except Exception as e: return JsonResponse({ 'status': 'error', 'message': f'Failed to check task status: {str(e)}' }, status=500) @login_required @staff_user_required def sync_history(request, job_slug=None): """View sync history and logs""" from .models import IntegrationLog from django_q.models import Task # Get sync logs if job_slug: # Filter for specific job job = get_object_or_404(models.JobPosting, slug=job_slug) logs = IntegrationLog.objects.filter( action=IntegrationLog.ActionChoices.SYNC, request_data__job_slug=job_slug ).order_by('-created_at') else: # Get all sync logs logs = IntegrationLog.objects.filter( action=IntegrationLog.ActionChoices.SYNC ).order_by('-created_at') # Get recent sync tasks recent_tasks = Task.objects.filter( group__startswith='sync_job_' ).order_by('-started')[:20] context = { 'logs': logs, 'recent_tasks': recent_tasks, 'job': job if job_slug else None, } return render(request, 'recruitment/sync_history.html', context) #participants views class ParticipantsListView(LoginRequiredMixin, StaffRequiredMixin, ListView): model = models.Participants template_name = 'participants/participants_list.html' context_object_name = 'participants' 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(name__icontains=search_query) | Q(email__icontains=search_query) | Q(phone__icontains=search_query) | Q(designation__icontains=search_query) ) # Filter for non-staff users if not self.request.user.is_staff: return models.Participants.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 ParticipantsDetailView(LoginRequiredMixin, StaffRequiredMixin, DetailView): model = models.Participants template_name = 'participants/participants_detail.html' context_object_name = 'participant' slug_url_kwarg = 'slug' class ParticipantsCreateView(LoginRequiredMixin, StaffRequiredMixin, SuccessMessageMixin, CreateView): model = models.Participants form_class = forms.ParticipantsForm template_name = 'participants/participants_create.html' success_url = reverse_lazy('job_list') success_message = 'Participant 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['jobs'] = [job] # return initial class ParticipantsUpdateView(LoginRequiredMixin, StaffRequiredMixin, SuccessMessageMixin, UpdateView): model = models.Participants form_class = forms.ParticipantsForm template_name = 'participants/participants_create.html' success_url = reverse_lazy('job_list') success_message = 'Participant updated successfully.' slug_url_kwarg = 'slug' class ParticipantsDeleteView(LoginRequiredMixin, StaffRequiredMixin, SuccessMessageMixin, DeleteView): model = models.Participants success_url = reverse_lazy('participants_list') # Redirect to the participants list after success success_message = 'Participant deleted successfully.' slug_url_kwarg = 'slug'