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,KeyTransform from recruitment.utils import json_to_markdown_table from django.db.models import Count, Avg, F, FloatField 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.shortcuts import render from django.utils import timezone from datetime import timedelta import json from django.utils.translation import gettext_lazy as _ # Add imports for user type restrictions from recruitment.decorators import StaffRequiredMixin, staff_user_required,candidate_user_required,staff_or_candidate_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_filter = self.request.GET.get('status') if status_filter: queryset = queryset.filter(status=status_filter) 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() context['status_filter']=self.request.GET.get('status') 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_applications_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/applications_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(person__first_name__icontains=search_query) | Q(person__last_name__icontains=search_query) | Q(person__email__icontains=search_query) | Q(person__phone__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/application_create.html' success_url = reverse_lazy('application_list') success_message = _('Application 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 form_invalid(self, form): messages.error(self.request, f"{form.errors.as_text()}") return super().form_invalid(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/application_update.html' success_url = reverse_lazy('application_list') success_message = _('Application updated successfully.') slug_url_kwarg = 'slug' class ApplicationDeleteView(LoginRequiredMixin, StaffRequiredMixin, SuccessMessageMixin, DeleteView): model = models.Application template_name = 'recruitment/application_delete.html' success_url = reverse_lazy('application_list') success_message = _('Application 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_resume_parsing_and_scoring', application.pk, hook='recruitment.hooks.callback_ai_parsing', sync=True, ) return redirect('application_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 application_detail(request, slug): from rich.json import JSON application = get_object_or_404(models.Application, slug=slug) try: parsed = ast.literal_eval(application.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/application_detail.html', { 'application': application, 'parsed': parsed, 'stage_form': stage_form, }) @login_required @staff_user_required def application_resume_template_view(request, slug): """Display formatted resume template for a application""" 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('application_list') return render(request, 'recruitment/application_resume_template.html', { 'application': application }) @login_required @staff_user_required def application_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("application_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_applications_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 applications by creation date global_daily_applications_qs = all_applications_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 --- application_queryset = all_applications_queryset job_scope_queryset = all_jobs_queryset interview_queryset = models.ScheduledInterview.objects.all() current_job = None if selected_job_pk: # Filter all base querysets application_queryset = application_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 = application_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_applications = application_queryset.count() score_expression = Cast( Coalesce( KeyTextTransform( 'match_score', KeyTransform('analysis_data_en', 'ai_analysis_data') ), Value('0'), ), output_field=IntegerField() ) # 2. ANNOTATE the queryset with the new field applications_with_score_query = application_queryset.annotate( annotated_match_score=score_expression ) # 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. # ) # applications_with_score_query= application_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_applications_7days = application_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( applications_count=Count('applications', distinct=True) ).aggregate(avg_apps=Avg('applications_count'))['avg_apps'] average_applications = round(average_applications_result or 0, 2) # B. Efficiency & Conversion Metrics (Scoped) hired_applications = application_queryset.filter( stage='Hired' ) lst=[c.time_to_hire_days for c in hired_applications] time_to_hire_query = hired_applications.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 = application_queryset.filter(stage='Applied').count() advanced_count = application_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 = application_queryset.filter(stage='Offer').count() offers_accepted_count = application_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 = applications_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 = applications_with_score_query.filter(annotated_match_score__gte=HIGH_POTENTIAL_THRESHOLD).count() high_potential_ratio = round( (high_potential_count / total_applications) * 100, 1 ) if total_applications > 0 else 0 total_scored_candidates = applications_with_score_query.count() scored_ratio = round( (total_scored_candidates / total_applications) * 100, 1 ) if total_applications > 0 else 0 # --- 6. CHART DATA PREPARATION --- # A. Pipeline Funnel (Scoped) stage_counts = application_queryset.values('stage').annotate(count=Count('stage')) stage_map = {item['stage']: item['count'] for item in stage_counts} application_stage = ['Applied', 'Exam', 'Interview', 'Offer', 'Hired'] application_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 = application_queryset.values('hiring_source').annotate(count=Count('stage')) source_map= {item['hiring_source']: item['count'] for item in hiring_source_counts} applications_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_applications': total_applications, 'new_applications_7days': new_applications_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 'application_stage': json.dumps(application_stage), 'application_count': json.dumps(application_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, 'applications_count_in_each_source': json.dumps(applications_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 applications_offer_view(request, slug): """View for applications in the Offer stage""" job = get_object_or_404(models.JobPosting, slug=slug) # Filter applications for this specific job and stage applications = job.offer_applications # Handle search search_query = request.GET.get('search', '') if search_query: applications = applications.filter( Q(first_name__icontains=search_query) | Q(last_name__icontains=search_query) | Q(email__icontains=search_query) | Q(phone__icontains=search_query) ) applications = applications.order_by('-created_at') context = { 'job': job, 'applications': applications, 'search_query': search_query, 'current_stage': 'Offer', } return render(request, 'recruitment/applications_offer_view.html', context) @login_required @staff_user_required def applications_hired_view(request, slug): """View for hired applications""" job = get_object_or_404(models.JobPosting, slug=slug) # Filter applications with offer_status = 'Accepted' applications = job.hired_applications # Handle search search_query = request.GET.get('search', '') if search_query: applications = applications.filter( Q(first_name__icontains=search_query) | Q(last_name__icontains=search_query) | Q(email__icontains=search_query) | Q(phone__icontains=search_query) ) applications = applications.order_by('-created_at') context = { 'job': job, 'applications': applications, 'search_query': search_query, 'current_stage': 'Hired', } return render(request, 'recruitment/applications_hired_view.html', context) @login_required @staff_user_required def update_application_status(request, job_slug, application_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) application = get_object_or_404(models.Application, slug=application_slug, job=job) if request.method == "POST": if stage_type == 'exam': status = request.POST.get("exam_status") score = request.POST.get("exam_score") application.exam_status = status application.exam_score = score application.exam_date = timezone.now() application.save(update_fields=['exam_status','exam_score', 'exam_date']) return render(request,'recruitment/partials/exam-results.html',{'application':application,'job':job}) elif stage_type == 'interview': application.interview_status = status application.interview_date = timezone.now() application.save(update_fields=['interview_status', 'interview_date']) return render(request,'recruitment/partials/interview-results.html',{'application':application,'job':job}) elif stage_type == 'offer': application.offer_status = status application.offer_date = timezone.now() application.save(update_fields=['offer_status', 'offer_date']) return render(request,'recruitment/partials/offer-results.html',{'application':application,'job':job}) return redirect('application_detail', application.slug) else: if stage_type == 'exam': return render(request,"includes/applications_update_exam_form.html",{'application':application,'job':job}) elif stage_type == 'interview': return render(request,"includes/applications_update_interview_form.html",{'application':application,'job':job}) elif stage_type == 'offer': return render(request,"includes/applications_update_offer_form.html",{'application':application,'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_applications_csv(request, job_slug, stage): """Export applications 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 applications based on stage if stage == 'hired': applications = job.applications.filter(**config['filter']) else: applications = job.applications.filter(**config['filter']) # Handle search if provided search_query = request.GET.get('search', '') if search_query: applications = applications.filter( Q(first_name__icontains=search_query) | Q(last_name__icontains=search_query) | Q(email__icontains=search_query) | Q(phone__icontains=search_query) ) applications = applications.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 application data for application in applications: row = [] # Extract data based on stage configuration for field in config['fields']: if field == 'name': row.append(application.name) elif field == 'email': row.append(application.email) elif field == 'phone': row.append(application.phone) elif field == 'created_at': row.append(application.created_at.strftime('%Y-%m-%d %H:%M') if application.created_at else '') elif field == 'stage': row.append(application.stage or '') elif field == 'exam_status': row.append(application.exam_status or '') elif field == 'exam_date': row.append(application.exam_date.strftime('%Y-%m-%d %H:%M') if application.exam_date else '') elif field == 'interview_status': row.append(application.interview_status or '') elif field == 'interview_date': row.append(application.interview_date.strftime('%Y-%m-%d %H:%M') if application.interview_date else '') elif field == 'offer_status': row.append(application.offer_status or '') elif field == 'offer_date': row.append(application.offer_date.strftime('%Y-%m-%d %H:%M') if application.offer_date else '') elif field == 'ai_score': # Extract AI score using model property try: score = application.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 = application.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 = application.screening_stage_rating row.append(rating if rating else '') except: row.append('') elif field == 'professional_category': # Extract professional category using model property try: category = application.professional_category row.append(category if category else '') except: row.append('') elif field == 'top_skills': # Extract top 3 skills using model property try: skills = application.top_3_keywords row.append(', '.join(skills) if skills else '') except: row.append('') elif field == 'strengths': # Extract strengths using model property try: strengths = application.strengths row.append(strengths if strengths else '') except: row.append('') elif field == 'weaknesses': # Extract weaknesses using model property try: weaknesses = application.weaknesses row.append(weaknesses if weaknesses else '') except: row.append('') elif field == 'join_date': row.append(application.join_date.strftime('%Y-%m-%d') if application.join_date else '') else: row.append(getattr(application, 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_applications(request, job_slug): """Sync hired applications 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'