2379 lines
92 KiB
Python
2379 lines
92 KiB
Python
import json
|
||
|
||
from django.utils.translation import gettext as _
|
||
from django.contrib.auth.models import User
|
||
from django.contrib.auth.decorators import login_required
|
||
from django.contrib.auth.mixins import LoginRequiredMixin
|
||
|
||
from rich import print
|
||
|
||
from django.views.decorators.csrf import csrf_exempt
|
||
from django.views.decorators.http import require_http_methods
|
||
from django.http import HttpResponse, JsonResponse
|
||
from datetime import datetime,time,timedelta
|
||
from django.views import View
|
||
from django.urls import reverse
|
||
from django.conf import settings
|
||
from django.utils import timezone
|
||
from django.db.models import FloatField,CharField, DurationField
|
||
from django.db.models.functions import Cast
|
||
from django.db.models.fields.json import KeyTextTransform
|
||
from django.db.models.expressions import ExpressionWrapper
|
||
from django.db.models import Count, Avg, F,Q
|
||
from .forms import (
|
||
CandidateExamDateForm,
|
||
InterviewForm,
|
||
ZoomMeetingForm,
|
||
JobPostingForm,
|
||
FormTemplateForm,
|
||
InterviewScheduleForm,JobPostingStatusForm,
|
||
BreakTimeFormSet,
|
||
JobPostingImageForm,
|
||
ProfileImageUploadForm,
|
||
StaffUserCreationForm,
|
||
MeetingCommentForm,
|
||
ToggleAccountForm,
|
||
|
||
)
|
||
from easyaudit.models import CRUDEvent, LoginEvent, RequestEvent
|
||
from rest_framework import viewsets
|
||
from django.contrib import messages
|
||
from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger
|
||
from .linkedin_service import LinkedInService
|
||
from .serializers import JobPostingSerializer, CandidateSerializer
|
||
from django.shortcuts import get_object_or_404, render, redirect
|
||
from django.views.generic import CreateView, UpdateView, DetailView, ListView
|
||
from .utils import (
|
||
create_zoom_meeting,
|
||
delete_zoom_meeting,
|
||
get_candidates_from_request,
|
||
update_meeting,
|
||
update_zoom_meeting,
|
||
get_zoom_meeting_details,
|
||
schedule_interviews,
|
||
get_available_time_slots,
|
||
)
|
||
from django.views.decorators.csrf import ensure_csrf_cookie
|
||
from django.views.decorators.http import require_POST
|
||
from .models import (
|
||
FormTemplate,
|
||
FormStage,
|
||
FormField,
|
||
FieldResponse,
|
||
FormSubmission,
|
||
InterviewSchedule,
|
||
BreakTime,
|
||
ZoomMeeting,
|
||
Candidate,
|
||
JobPosting,
|
||
ScheduledInterview,
|
||
JobPostingImage,
|
||
Profile,MeetingComment
|
||
)
|
||
import logging
|
||
from datastar_py.django import (
|
||
DatastarResponse,
|
||
ServerSentEventGenerator as SSE,
|
||
read_signals,
|
||
)
|
||
from django.db import transaction
|
||
from django_q.tasks import async_task
|
||
from django.db.models import Prefetch
|
||
from django.db.models import Q, Count, Avg
|
||
from django.db.models import FloatField
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class JobPostingViewSet(viewsets.ModelViewSet):
|
||
queryset = JobPosting.objects.all()
|
||
serializer_class = JobPostingSerializer
|
||
|
||
|
||
class CandidateViewSet(viewsets.ModelViewSet):
|
||
queryset = Candidate.objects.all()
|
||
serializer_class = CandidateSerializer
|
||
|
||
|
||
class ZoomMeetingCreateView(LoginRequiredMixin, CreateView):
|
||
model = ZoomMeeting
|
||
template_name = "meetings/create_meeting.html"
|
||
form_class = ZoomMeetingForm
|
||
success_url = "/"
|
||
|
||
def form_valid(self, form):
|
||
instance = form.save(commit=False)
|
||
try:
|
||
topic = instance.topic
|
||
if instance.start_time < timezone.now():
|
||
messages.error(self.request, "Start time must be in the future.")
|
||
return redirect(reverse("create_meeting",kwargs={"slug": instance.slug}))
|
||
start_time = instance.start_time
|
||
duration = instance.duration
|
||
|
||
result = create_zoom_meeting(topic, start_time, duration)
|
||
if result["status"] == "success":
|
||
instance.meeting_id = result["meeting_details"]["meeting_id"]
|
||
instance.join_url = result["meeting_details"]["join_url"]
|
||
instance.host_email = result["meeting_details"]["host_email"]
|
||
instance.password = result["meeting_details"]["password"]
|
||
instance.status = result["zoom_gateway_response"]["status"]
|
||
instance.zoom_gateway_response = result["zoom_gateway_response"]
|
||
instance.save()
|
||
messages.success(self.request, result["message"])
|
||
|
||
return redirect(reverse("list_meetings"))
|
||
else:
|
||
messages.error(self.request, result["message"])
|
||
return redirect(reverse("create_meeting",kwargs={"slug": instance.slug}))
|
||
except Exception as e:
|
||
messages.error(self.request, f"Error creating meeting: {e}")
|
||
return redirect(reverse("create_meeting",kwargs={"slug": instance.slug}))
|
||
|
||
|
||
class ZoomMeetingListView(LoginRequiredMixin, ListView):
|
||
model = ZoomMeeting
|
||
template_name = "meetings/list_meetings.html"
|
||
context_object_name = "meetings"
|
||
paginate_by = 10
|
||
|
||
def get_queryset(self):
|
||
queryset = super().get_queryset().order_by("-start_time")
|
||
|
||
# Prefetch related interview data efficiently
|
||
|
||
queryset = queryset.prefetch_related(
|
||
Prefetch(
|
||
'interview', # related_name from ZoomMeeting to ScheduledInterview
|
||
queryset=ScheduledInterview.objects.select_related('candidate', 'job'),
|
||
to_attr='interview_details' # Changed to not start with underscore
|
||
)
|
||
)
|
||
|
||
# Handle search by topic or meeting_id
|
||
search_query = self.request.GET.get("q", "") # Renamed from 'search' to 'q' for consistency
|
||
if search_query:
|
||
queryset = queryset.filter(
|
||
Q(topic__icontains=search_query) | Q(meeting_id__icontains=search_query)
|
||
)
|
||
|
||
# Handle filter by status
|
||
status_filter = self.request.GET.get("status", "")
|
||
if status_filter:
|
||
queryset = queryset.filter(status=status_filter)
|
||
|
||
# Handle search by candidate name
|
||
candidate_name = self.request.GET.get("candidate_name", "")
|
||
if candidate_name:
|
||
# Filter based on the name of the candidate associated with the meeting's interview
|
||
queryset = queryset.filter(
|
||
Q(interview__candidate__first_name__icontains=candidate_name) |
|
||
Q(interview__candidate__last_name__icontains=candidate_name)
|
||
)
|
||
|
||
return queryset
|
||
|
||
def get_context_data(self, **kwargs):
|
||
context = super().get_context_data(**kwargs)
|
||
context["search_query"] = self.request.GET.get("q", "")
|
||
context["status_filter"] = self.request.GET.get("status", "")
|
||
context["candidate_name_filter"] = self.request.GET.get("candidate_name", "")
|
||
return context
|
||
|
||
|
||
class ZoomMeetingDetailsView(LoginRequiredMixin, DetailView):
|
||
model = ZoomMeeting
|
||
template_name = "meetings/meeting_details.html"
|
||
context_object_name = "meeting"
|
||
|
||
|
||
class ZoomMeetingUpdateView(LoginRequiredMixin, UpdateView):
|
||
model = ZoomMeeting
|
||
form_class = ZoomMeetingForm
|
||
context_object_name = "meeting"
|
||
template_name = "meetings/update_meeting.html"
|
||
success_url = "/"
|
||
|
||
# def get_form_kwargs(self):
|
||
# kwargs = super().get_form_kwargs()
|
||
# # Ensure the form is initialized with the instance's current values
|
||
# if self.object:
|
||
# kwargs['initial'] = getattr(kwargs, 'initial', {})
|
||
# initial_start_time = ""
|
||
# if self.object.start_time:
|
||
# try:
|
||
# initial_start_time = self.object.start_time.strftime('%m-%d-%Y,T%H:%M')
|
||
# except AttributeError:
|
||
# print(f"Warning: start_time {self.object.start_time} is not a datetime object.")
|
||
# initial_start_time = ""
|
||
# kwargs['initial']['start_time'] = initial_start_time
|
||
# return kwargs
|
||
|
||
def form_valid(self, form):
|
||
instance = form.save(commit=False)
|
||
updated_data = {
|
||
"topic": instance.topic,
|
||
"start_time": instance.start_time.isoformat() + "Z",
|
||
"duration": instance.duration,
|
||
}
|
||
if instance.start_time < timezone.now():
|
||
messages.error(self.request, "Start time must be in the future.")
|
||
return redirect(reverse("meeting_details", kwargs={"slug": instance.slug}))
|
||
|
||
result = update_meeting(instance, updated_data)
|
||
|
||
if result["status"] == "success":
|
||
messages.success(self.request, result["message"])
|
||
else:
|
||
messages.error(self.request, result["message"])
|
||
return redirect(reverse("meeting_details", kwargs={"slug": instance.slug}))
|
||
|
||
|
||
def ZoomMeetingDeleteView(request, slug):
|
||
meeting = get_object_or_404(ZoomMeeting, slug=slug)
|
||
if "HX-Request" in request.headers:
|
||
return render(request, "meetings/delete_meeting_form.html", {"meeting": meeting,"delete_url": reverse("delete_meeting", kwargs={"slug": meeting.slug})})
|
||
if request.method == "POST":
|
||
try:
|
||
result = delete_zoom_meeting(meeting.meeting_id)
|
||
if result["status"] == "success" or "Meeting does not exist" in result["details"]["message"]:
|
||
meeting.delete()
|
||
messages.success(request, "Meeting deleted successfully.")
|
||
else:
|
||
messages.error(request, f"{result["message"]} , {result['details']["message"]}")
|
||
return redirect(reverse("list_meetings"))
|
||
except Exception as e:
|
||
messages.error(request, str(e))
|
||
return redirect(reverse("list_meetings"))
|
||
|
||
# Job Posting
|
||
# def job_list(request):
|
||
# """Display the list of job postings order by creation date descending"""
|
||
# jobs=JobPosting.objects.all().order_by('-created_at')
|
||
|
||
# # Filter by status if provided
|
||
# print(f"the request is: {request} ")
|
||
# status=request.GET.get('status')
|
||
# print(f"DEBUG: Status filter received: {status}")
|
||
# if status:
|
||
# jobs=jobs.filter(status=status)
|
||
|
||
# #pagination
|
||
# paginator=Paginator(jobs,10) # Show 10 jobs per page
|
||
# page_number=request.GET.get('page')
|
||
# page_obj=paginator.get_page(page_number)
|
||
# return render(request, 'jobs/job_list.html', {
|
||
# 'page_obj': page_obj,
|
||
# 'status_filter': status
|
||
# })
|
||
|
||
|
||
@login_required
|
||
def create_job(request):
|
||
"""Create a new job posting"""
|
||
|
||
if request.method == "POST":
|
||
form = JobPostingForm(
|
||
request.POST
|
||
)
|
||
# to check user is authenticated or not
|
||
if form.is_valid():
|
||
try:
|
||
job = form.save(commit=False)
|
||
job.save()
|
||
job_apply_url_relative=reverse('application_detail',kwargs={'slug':job.slug})
|
||
job_apply_url_absolute=request.build_absolute_uri(job_apply_url_relative)
|
||
job.application_url=job_apply_url_absolute
|
||
# FormTemplate.objects.create(job=job, is_active=False, name=job.title,created_by=request.user)
|
||
job.save()
|
||
messages.success(request, f'Job "{job.title}" created successfully!')
|
||
return redirect("job_list")
|
||
except Exception as e:
|
||
logger.error(f"Error creating job: {e}")
|
||
messages.error(request, f"Error creating job: {e}")
|
||
else:
|
||
messages.error(request, f"Please correct the errors below.{form.errors}")
|
||
else:
|
||
form = JobPostingForm()
|
||
return render(request, "jobs/create_job.html", {"form": form})
|
||
|
||
|
||
@login_required
|
||
def edit_job(request, slug):
|
||
"""Edit an existing job posting"""
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
if request.method == "POST":
|
||
form = JobPostingForm(
|
||
request.POST,
|
||
instance=job
|
||
)
|
||
if form.is_valid():
|
||
try:
|
||
form.save()
|
||
messages.success(request, f'Job "{job.title}" updated successfully!')
|
||
return redirect("job_list")
|
||
except Exception as e:
|
||
logger.error(f"Error updating job: {e}")
|
||
messages.error(request, f"Error updating job: {e}")
|
||
else:
|
||
messages.error(request, "Please correct the errors below.")
|
||
else:
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
form = JobPostingForm(
|
||
instance=job
|
||
)
|
||
return render(request, "jobs/edit_job.html", {"form": form, "job": job})
|
||
|
||
|
||
@login_required
|
||
def job_detail(request, slug):
|
||
"""View details of a specific job"""
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
# Get all candidates for this job, ordered by most recent
|
||
applicants = job.candidates.all().order_by("-created_at")
|
||
|
||
# Count candidates by stage for summary statistics
|
||
total_applicant = applicants.count()
|
||
|
||
applied_count = applicants.filter(stage="Applied").count()
|
||
|
||
exam_count=applicants.filter(stage="Exam").count
|
||
|
||
interview_count = applicants.filter(stage="Interview").count()
|
||
|
||
offer_count = applicants.filter(stage="Offer").count()
|
||
|
||
status_form = JobPostingStatusForm(instance=job)
|
||
try:
|
||
# If the related object exists, use its instance data
|
||
image_upload_form = JobPostingImageForm(instance=job.post_images)
|
||
except Exception as e:
|
||
# If the related object does NOT exist, create a blank form
|
||
image_upload_form = JobPostingImageForm()
|
||
|
||
|
||
# 2. Check for POST request (Status Update Submission)
|
||
if request.method == 'POST':
|
||
|
||
status_form = JobPostingStatusForm(request.POST, instance=job)
|
||
|
||
if status_form.is_valid():
|
||
job_status=status_form.cleaned_data['status']
|
||
form_template=job.form_template
|
||
if job_status=='ACTIVE':
|
||
form_template.is_active=True
|
||
form_template.save(update_fields=['is_active'])
|
||
else:
|
||
form_template.is_active=False
|
||
form_template.save(update_fields=['is_active'])
|
||
|
||
status_form.save()
|
||
|
||
# Add a success message
|
||
messages.success(request, f"Status for '{job.title}' updated to '{job.get_status_display()}' successfully!")
|
||
|
||
|
||
return redirect('job_detail', slug=slug)
|
||
else:
|
||
|
||
|
||
messages.error(request, "Failed to update status due to validation errors.")
|
||
|
||
|
||
# --- 2. Quality Metrics (JSON Aggregation) ---
|
||
|
||
# Filter for candidates who have been scored and annotate with a sortable score
|
||
candidates_with_score = applicants.filter(is_resume_parsed=True).annotate(
|
||
# Extract the score as TEXT
|
||
score_as_text=KeyTextTransform(
|
||
'match_score',
|
||
KeyTextTransform('resume_data', F('ai_analysis_data'))
|
||
)
|
||
).annotate(
|
||
# Cast the extracted text score to a FloatField for numerical operations
|
||
sortable_score=Cast('score_as_text', output_field=FloatField())
|
||
)
|
||
|
||
# Aggregate: Average Match Score
|
||
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)
|
||
|
||
# Metric: High Potential Count (Score >= 75)
|
||
high_potential_count = candidates_with_score.filter(
|
||
sortable_score__gte=75
|
||
).count()
|
||
high_potential_ratio = round((high_potential_count / total_applicant) * 100, 1) if total_applicant > 0 else 0
|
||
|
||
# --- 3. Time Metrics (Duration Aggregation) ---
|
||
|
||
# Metric: Average Time from Applied to Interview (T2I)
|
||
t2i_candidates = applicants.filter(
|
||
interview_date__isnull=False
|
||
).annotate(
|
||
time_to_interview=ExpressionWrapper(
|
||
F('interview_date') - F('created_at'),
|
||
output_field=DurationField()
|
||
)
|
||
)
|
||
avg_t2i_duration = t2i_candidates.aggregate(
|
||
avg_t2i=Avg('time_to_interview')
|
||
)['avg_t2i']
|
||
|
||
# Convert timedelta to days
|
||
avg_t2i_days = round(avg_t2i_duration.total_seconds() / (60*60*24), 1) if avg_t2i_duration else 0
|
||
|
||
# Metric: Average Time in Exam Stage
|
||
t_in_exam_candidates = applicants.filter(
|
||
exam_date__isnull=False, interview_date__isnull=False
|
||
).annotate(
|
||
time_in_exam=ExpressionWrapper(
|
||
F('interview_date') - F('exam_date'),
|
||
output_field=DurationField()
|
||
)
|
||
)
|
||
avg_t_in_exam_duration = t_in_exam_candidates.aggregate(
|
||
avg_t_in_exam=Avg('time_in_exam')
|
||
)['avg_t_in_exam']
|
||
|
||
# Convert timedelta to days
|
||
avg_t_in_exam_days = round(avg_t_in_exam_duration.total_seconds() / (60*60*24), 1) if avg_t_in_exam_duration else 0
|
||
|
||
category_data = applicants.filter(
|
||
ai_analysis_data__analysis_data__category__isnull=False
|
||
).values('ai_analysis_data__analysis_data__category').annotate(
|
||
candidate_count=Count('id'),
|
||
category=Cast('ai_analysis_data__analysis_data__category',output_field=CharField())
|
||
).order_by('ai_analysis_data__analysis_data__category')
|
||
# Prepare data for Chart.js
|
||
print(category_data)
|
||
categories = [item['category'] for item in category_data]
|
||
candidate_counts = [item['candidate_count'] for item in category_data]
|
||
# avg_scores = [round(item['avg_match_score'], 2) if item['avg_match_score'] is not None else 0 for item in category_data]
|
||
|
||
|
||
context = {
|
||
"job": job,
|
||
"applicants": applicants,
|
||
"total_applicants": total_applicant, # This was total_candidates in the prompt, using total_applicant for consistency
|
||
"applied_count": applied_count,
|
||
'exam_count':exam_count,
|
||
"interview_count": interview_count,
|
||
"offer_count": offer_count,
|
||
'status_form':status_form,
|
||
'image_upload_form':image_upload_form,
|
||
'categories': categories,
|
||
'candidate_counts': candidate_counts,
|
||
# 'avg_scores': avg_scores,
|
||
# New statistics
|
||
'avg_match_score': avg_match_score,
|
||
'high_potential_count': high_potential_count,
|
||
'high_potential_ratio': high_potential_ratio,
|
||
'avg_t2i_days': avg_t2i_days,
|
||
'avg_t_in_exam_days': avg_t_in_exam_days,
|
||
}
|
||
return render(request, "jobs/job_detail.html", context)
|
||
|
||
@login_required
|
||
def job_image_upload(request, slug):
|
||
#only for handling the post request
|
||
job=get_object_or_404(JobPosting,slug=slug)
|
||
try:
|
||
instance = JobPostingImage.objects.get(job=job)
|
||
except JobPostingImage.DoesNotExist:
|
||
# If it doesn't exist, create a new instance placeholder
|
||
instance = None
|
||
|
||
if request.method == 'POST':
|
||
# Pass the existing instance to the form if it exists
|
||
image_upload_form = JobPostingImageForm(request.POST, request.FILES, instance=instance)
|
||
|
||
if image_upload_form.is_valid():
|
||
|
||
# If creating a new one (instance is None), set the job link manually
|
||
if instance is None:
|
||
image_instance = image_upload_form.save(commit=False)
|
||
image_instance.job = job
|
||
image_instance.save()
|
||
messages.success(request, f"Image uploaded successfully for {job.title}.")
|
||
else:
|
||
# If updating, the form will update the instance passed to it
|
||
image_upload_form.save()
|
||
messages.success(request, f"Image updated successfully for {job.title}.")
|
||
|
||
else:
|
||
|
||
messages.error(request, "Image upload failed: Please ensure a valid image file was selected.")
|
||
return redirect('job_detail', slug=job.slug)
|
||
return redirect('job_detail', slug=job.slug)
|
||
|
||
|
||
def kaauh_career(request):
|
||
active_jobs = JobPosting.objects.select_related(
|
||
'form_template'
|
||
).filter(
|
||
status='ACTIVE',
|
||
form_template__is_active=True
|
||
)
|
||
|
||
return render(request,'jobs/career.html',{'active_jobs':active_jobs})
|
||
|
||
|
||
|
||
# job detail facing the candidate:
|
||
def application_detail(request, slug):
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
return render(request, "forms/application_detail.html", {"job": job})
|
||
|
||
|
||
from django_q.tasks import async_task
|
||
|
||
@login_required
|
||
def post_to_linkedin(request, slug):
|
||
"""Post a job to LinkedIn"""
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
if job.status != "ACTIVE":
|
||
messages.info(request, "Only active jobs can be posted to LinkedIn.")
|
||
return redirect("job_list")
|
||
|
||
if request.method == "POST":
|
||
linkedin_access_token=request.session.get("linkedin_access_token")
|
||
# Check if user is authenticated with LinkedIn
|
||
if not "linkedin_access_token":
|
||
messages.error(request, "Please authenticate with LinkedIn first.")
|
||
return redirect("linkedin_login")
|
||
try:
|
||
|
||
# Clear previous LinkedIn data for re-posting
|
||
#Prepare the job object for background processing
|
||
job.posted_to_linkedin = False
|
||
job.linkedin_post_id = ""
|
||
job.linkedin_post_url = ""
|
||
job.linkedin_post_status = "QUEUED"
|
||
job.linkedin_posted_at = None
|
||
job.save()
|
||
|
||
# ENQUEUE THE TASK
|
||
# Pass the function path, the job slug, and the token as arguments
|
||
|
||
async_task(
|
||
'recruitment.tasks.linkedin_post_task',
|
||
job.slug,
|
||
linkedin_access_token
|
||
)
|
||
|
||
messages.success(
|
||
request,
|
||
_(f"✅ Job posting process for job with JOB ID: {job.internal_job_id} started! Check the job details page in a moment for the final status.")
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error enqueuing LinkedIn post: {e}")
|
||
messages.error(request, _("Failed to start the job posting process. Please try again."))
|
||
|
||
return redirect("job_detail", slug=job.slug)
|
||
|
||
|
||
def linkedin_login(request):
|
||
"""Redirect to LinkedIn OAuth"""
|
||
service = LinkedInService()
|
||
auth_url = service.get_auth_url()
|
||
"""
|
||
It creates a special URL that:
|
||
Sends the user to LinkedIn to log in
|
||
Asks the user to grant your app permission to post on their behalf
|
||
Tells LinkedIn where to send the user back after they approve (your redirect_uri)
|
||
http://yoursite.com/linkedin/callback/?code=TEMPORARY_CODE_HERE
|
||
"""
|
||
return redirect(auth_url)
|
||
|
||
|
||
def linkedin_callback(request):
|
||
"""Handle LinkedIn OAuth callback"""
|
||
code = request.GET.get("code")
|
||
if not code:
|
||
messages.error(request, "No authorization code received from LinkedIn.")
|
||
return redirect("job_list")
|
||
|
||
try:
|
||
service = LinkedInService()
|
||
# get_access_token(code)->It makes a POST request to LinkedIn’s token endpoint with parameters
|
||
access_token = service.get_access_token(code)
|
||
request.session["linkedin_access_token"] = access_token
|
||
request.session["linkedin_authenticated"] = True
|
||
settings.LINKEDIN_IS_CONNECTED = True
|
||
messages.success(request, "Successfully authenticated with LinkedIn!")
|
||
except Exception as e:
|
||
logger.error(f"LinkedIn authentication error: {e}")
|
||
messages.error(request, f"LinkedIn authentication failed: {e}")
|
||
|
||
return redirect("job_list")
|
||
|
||
|
||
# applicant views
|
||
def applicant_job_detail(request, slug):
|
||
"""View job details for applicants"""
|
||
job=get_object_or_404(JobPosting,slug=slug,status='ACTIVE')
|
||
return render(request,'jobs/applicant_job_detail.html',{'job':job})
|
||
|
||
def application_success(request,slug):
|
||
job=get_object_or_404(JobPosting,slug=slug)
|
||
return render(request,'jobs/application_success.html',{'job':job})
|
||
|
||
@ensure_csrf_cookie
|
||
@login_required
|
||
def form_builder(request, template_slug=None):
|
||
"""Render the form builder interface"""
|
||
context = {}
|
||
if template_slug:
|
||
template = get_object_or_404(
|
||
FormTemplate, slug=template_slug
|
||
)
|
||
context['template']=template
|
||
context["template_slug"] = template.slug
|
||
context["template_name"] = template.name
|
||
return render(request, "forms/form_builder.html", context)
|
||
|
||
|
||
@csrf_exempt
|
||
@require_http_methods(["POST"])
|
||
def save_form_template(request):
|
||
"""Save a new or existing form template"""
|
||
try:
|
||
data = json.loads(request.body)
|
||
template_name = data.get("name", "Untitled Form")
|
||
stages_data = data.get("stages", [])
|
||
template_slug = data.get("template_slug")
|
||
|
||
if template_slug:
|
||
# Update existing template
|
||
template = get_object_or_404(
|
||
FormTemplate, slug=template_slug
|
||
)
|
||
template.name = template_name
|
||
template.save()
|
||
# Clear existing stages and fields
|
||
template.stages.all().delete()
|
||
else:
|
||
# Create new template
|
||
template = FormTemplate.objects.create(
|
||
name=template_name
|
||
)
|
||
|
||
# Create stages and fields
|
||
for stage_order, stage_data in enumerate(stages_data):
|
||
stage = FormStage.objects.create(
|
||
template=template,
|
||
name=stage_data["name"],
|
||
order=stage_order,
|
||
is_predefined=stage_data.get("predefined", False),
|
||
)
|
||
|
||
for field_order, field_data in enumerate(stage_data["fields"]):
|
||
options = field_data.get("options", [])
|
||
if not isinstance(options, list):
|
||
options = []
|
||
|
||
file_types = field_data.get("fileTypes", "")
|
||
max_file_size = field_data.get("maxFileSize", 5)
|
||
|
||
FormField.objects.create(
|
||
stage=stage,
|
||
label=field_data.get("label", ""),
|
||
field_type=field_data.get("type", "text"),
|
||
placeholder=field_data.get("placeholder", ""),
|
||
required=field_data.get("required", False),
|
||
order=field_order,
|
||
is_predefined=field_data.get("predefined", False),
|
||
options=options,
|
||
file_types=file_types,
|
||
max_file_size=max_file_size,
|
||
)
|
||
|
||
return JsonResponse(
|
||
{
|
||
"success": True,
|
||
"template_slug": template.slug,
|
||
"message": "Form template saved successfully!",
|
||
}
|
||
)
|
||
except Exception as e:
|
||
return JsonResponse({"success": False, "error": str(e)}, status=400)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def load_form_template(request, template_slug):
|
||
"""Load an existing form template"""
|
||
template = get_object_or_404(FormTemplate, slug=template_slug)
|
||
|
||
stages = []
|
||
for stage in template.stages.all():
|
||
fields = []
|
||
for field in stage.fields.all():
|
||
fields.append(
|
||
{
|
||
"id": field.id,
|
||
"type": field.field_type,
|
||
"label": field.label,
|
||
"placeholder": field.placeholder,
|
||
"required": field.required,
|
||
"options": field.options,
|
||
"fileTypes": field.file_types,
|
||
"maxFileSize": field.max_file_size,
|
||
"predefined": field.is_predefined,
|
||
}
|
||
)
|
||
stages.append(
|
||
{
|
||
"id": stage.id,
|
||
"name": stage.name,
|
||
"predefined": stage.is_predefined,
|
||
"fields": fields,
|
||
}
|
||
)
|
||
|
||
return JsonResponse(
|
||
{
|
||
"success": True,
|
||
"template": {
|
||
"id": template.id,
|
||
"template_slug": template.slug,
|
||
"name": template.name,
|
||
"description": template.description,
|
||
"is_active": template.is_active,
|
||
"job": template.job_id if template.job else None,
|
||
"stages": stages,
|
||
},
|
||
}
|
||
)
|
||
|
||
|
||
@login_required
|
||
def form_templates_list(request):
|
||
"""List all form templates for the current user"""
|
||
query = request.GET.get("q", "")
|
||
templates = FormTemplate.objects.filter()
|
||
|
||
if query:
|
||
templates = templates.filter(
|
||
Q(name__icontains=query) | Q(description__icontains=query)
|
||
)
|
||
|
||
templates = templates.order_by("-created_at")
|
||
paginator = Paginator(templates, 10) # Show 10 templates per page
|
||
page_number = request.GET.get("page")
|
||
page_obj = paginator.get_page(page_number)
|
||
form = FormTemplateForm()
|
||
form.fields["job"].queryset = JobPosting.objects.filter(form_template__isnull=True)
|
||
context = {"templates": page_obj, "query": query, "form": form}
|
||
return render(request, "forms/form_templates_list.html", context)
|
||
|
||
|
||
@login_required
|
||
def create_form_template(request):
|
||
"""Create a new form template"""
|
||
if request.method == "POST":
|
||
form = FormTemplateForm(request.POST)
|
||
if form.is_valid():
|
||
template = form.save(commit=False)
|
||
template.created_by = request.user
|
||
template.save()
|
||
messages.success(
|
||
request, f'Form template "{template.name}" created successfully!'
|
||
)
|
||
return redirect("form_templates_list")
|
||
else:
|
||
form = FormTemplateForm()
|
||
|
||
return render(request, "forms/create_form_template.html", {"form": form})
|
||
|
||
|
||
@login_required
|
||
@require_http_methods(["GET"])
|
||
def list_form_templates(request):
|
||
"""List all form templates for the current user"""
|
||
templates = FormTemplate.objects.filter().values(
|
||
"id", "name", "description", "created_at", "updated_at"
|
||
)
|
||
return JsonResponse({"success": True, "templates": list(templates)})
|
||
|
||
|
||
@login_required
|
||
@require_http_methods(["DELETE"])
|
||
def delete_form_template(request, template_id):
|
||
"""Delete a form template"""
|
||
template = get_object_or_404(FormTemplate, id=template_id)
|
||
template.delete()
|
||
return JsonResponse(
|
||
{"success": True, "message": "Form template deleted successfully!"}
|
||
)
|
||
|
||
|
||
def application_submit_form(request, template_slug):
|
||
"""Display the form as a step-by-step wizard"""
|
||
template = get_object_or_404(FormTemplate, slug=template_slug, is_active=True)
|
||
job_id = template.job.internal_job_id
|
||
job=template.job
|
||
is_limit_exceeded = job.is_application_limit_reached
|
||
if is_limit_exceeded:
|
||
messages.error(
|
||
request,
|
||
'Application limit reached: This job is no longer accepting new applications. Please explore other available positions.'
|
||
)
|
||
return redirect('application_detail',slug=job.slug)
|
||
if job.is_expired:
|
||
messages.error(
|
||
request,
|
||
'Application deadline passed: This job is no longer accepting new applications. Please explore other available positions.'
|
||
)
|
||
return redirect('application_detail',slug=job.slug)
|
||
|
||
return render(
|
||
request,
|
||
"forms/application_submit_form.html",
|
||
{"template_slug": template_slug, "job_id": job_id},
|
||
)
|
||
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def application_submit(request, template_slug):
|
||
"""Handle form submission"""
|
||
template = get_object_or_404(FormTemplate, slug=template_slug)
|
||
job = template.job
|
||
if request.method == "POST":
|
||
try:
|
||
with transaction.atomic():
|
||
job_posting = JobPosting.objects.select_for_update().get(form_template=template)
|
||
|
||
current_count = job_posting.candidates.count()
|
||
if current_count >= job_posting.max_applications:
|
||
template.is_active = False
|
||
template.save()
|
||
return JsonResponse(
|
||
{"success": False, "message": "Application limit reached for this job."}
|
||
)
|
||
submission = FormSubmission.objects.create(template=template)
|
||
|
||
# Process field responses
|
||
for field_id, value in request.POST.items():
|
||
if field_id.startswith("field_"):
|
||
actual_field_id = field_id.replace("field_", "")
|
||
try:
|
||
field = FormField.objects.get(
|
||
id=actual_field_id, stage__template=template
|
||
)
|
||
FieldResponse.objects.create(
|
||
submission=submission,
|
||
field=field,
|
||
value=value if value else None,
|
||
)
|
||
except FormField.DoesNotExist:
|
||
continue
|
||
|
||
# Handle file uploads
|
||
for field_id, uploaded_file in request.FILES.items():
|
||
if field_id.startswith("field_"):
|
||
actual_field_id = field_id.replace("field_", "")
|
||
try:
|
||
field = FormField.objects.get(
|
||
id=actual_field_id, stage__template=template
|
||
)
|
||
FieldResponse.objects.create(
|
||
submission=submission,
|
||
field=field,
|
||
uploaded_file=uploaded_file,
|
||
)
|
||
except FormField.DoesNotExist:
|
||
continue
|
||
try:
|
||
first_name = submission.responses.get(field__label="First Name")
|
||
last_name = submission.responses.get(field__label="Last Name")
|
||
email = submission.responses.get(field__label="Email Address")
|
||
phone = submission.responses.get(field__label="Phone Number")
|
||
address = submission.responses.get(field__label="Address")
|
||
resume = submission.responses.get(field__label="Resume Upload")
|
||
|
||
submission.applicant_name = (
|
||
f"{first_name.display_value} {last_name.display_value}"
|
||
)
|
||
submission.applicant_email = email.display_value
|
||
submission.save()
|
||
# time=timezone.now()
|
||
Candidate.objects.create(
|
||
first_name=first_name.display_value,
|
||
last_name=last_name.display_value,
|
||
email=email.display_value,
|
||
phone=phone.display_value,
|
||
address=address.display_value,
|
||
resume=resume.get_file if resume.is_file else None,
|
||
job=job
|
||
)
|
||
return JsonResponse(
|
||
{
|
||
"success": True,
|
||
"message": "Form submitted successfully!",
|
||
"redirect_url": reverse('application_success',kwargs={'slug':job.slug}),
|
||
}
|
||
)
|
||
# return redirect('application_success',slug=job.slug)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Candidate creation failed,{e}")
|
||
pass
|
||
return JsonResponse(
|
||
{
|
||
"success": True,
|
||
"message": "Form submitted successfully!",
|
||
"submission_id": submission.id,
|
||
}
|
||
)
|
||
except Exception as e:
|
||
return JsonResponse({"success": False, "error": str(e)}, status=400)
|
||
else:
|
||
# Handle GET request - this should not happen for form submission
|
||
return JsonResponse(
|
||
{"success": False, "error": "GET method not allowed for form submission"},
|
||
status=405,
|
||
)
|
||
|
||
|
||
@login_required
|
||
def form_template_submissions_list(request, slug):
|
||
"""List all submissions for a specific form template"""
|
||
template = get_object_or_404(FormTemplate, slug=slug)
|
||
|
||
submissions = FormSubmission.objects.filter(template=template).order_by(
|
||
"-submitted_at"
|
||
)
|
||
|
||
# Pagination
|
||
paginator = Paginator(submissions, 10) # Show 10 submissions per page
|
||
page_number = request.GET.get("page")
|
||
page_obj = paginator.get_page(page_number)
|
||
|
||
return render(
|
||
request,
|
||
"forms/form_template_submissions_list.html",
|
||
{"template": template, "page_obj": page_obj},
|
||
)
|
||
|
||
|
||
@login_required
|
||
def form_template_all_submissions(request, template_id):
|
||
"""Display all submissions for a form template in table format"""
|
||
template = get_object_or_404(FormTemplate, id=template_id)
|
||
print(template)
|
||
# Get all submissions for this template
|
||
submissions = FormSubmission.objects.filter(template=template).order_by("-submitted_at")
|
||
|
||
# Get all fields for this template, ordered by stage and field order
|
||
fields = FormField.objects.filter(stage__template=template).select_related('stage').order_by('stage__order', 'order')
|
||
|
||
# Pagination
|
||
paginator = Paginator(submissions, 10) # Show 10 submissions per page
|
||
page_number = request.GET.get("page")
|
||
page_obj = paginator.get_page(page_number)
|
||
|
||
return render(
|
||
request,
|
||
"forms/form_template_all_submissions.html",
|
||
{
|
||
"template": template,
|
||
"page_obj": page_obj,
|
||
"fields": fields,
|
||
},
|
||
)
|
||
|
||
|
||
@login_required
|
||
def form_submission_details(request, template_id, slug):
|
||
"""Display detailed view of a specific form submission"""
|
||
# Get the form template and verify ownership
|
||
template = get_object_or_404(FormTemplate, id=template_id)
|
||
# Get the specific submission
|
||
submission = get_object_or_404(FormSubmission, slug=slug, template=template)
|
||
|
||
# Get all stages with their fields
|
||
stages = template.stages.prefetch_related("fields").order_by("order")
|
||
|
||
# Get all responses for this submission, ordered by field order
|
||
responses = submission.responses.select_related("field").order_by("field__order")
|
||
|
||
# Group responses by stage
|
||
stage_responses = {}
|
||
for stage in stages:
|
||
stage_responses[stage.id] = {
|
||
"stage": stage,
|
||
"responses": responses.filter(field__stage=stage),
|
||
}
|
||
|
||
return render(
|
||
request,
|
||
"forms/form_submission_details.html",
|
||
{
|
||
"template": template,
|
||
"submission": submission,
|
||
"stages": stages,
|
||
"responses": responses,
|
||
"stage_responses": stage_responses,
|
||
},
|
||
)
|
||
|
||
def _handle_get_request(request, slug, job):
|
||
"""
|
||
Handles GET requests, setting up forms and restoring candidate selections
|
||
from the session for persistence.
|
||
"""
|
||
SESSION_KEY = f"schedule_candidate_ids_{slug}"
|
||
|
||
form = InterviewScheduleForm(slug=slug)
|
||
# break_formset = BreakTimeFormSet(prefix='breaktime')
|
||
|
||
selected_ids = []
|
||
|
||
# 1. Capture IDs from HTMX request and store in session (when first clicked)
|
||
if "HX-Request" in request.headers:
|
||
candidate_ids = request.GET.getlist("candidate_ids")
|
||
|
||
if candidate_ids:
|
||
request.session[SESSION_KEY] = candidate_ids
|
||
selected_ids = candidate_ids
|
||
|
||
# 2. Restore IDs from session (on refresh or navigation)
|
||
if not selected_ids:
|
||
selected_ids = request.session.get(SESSION_KEY, [])
|
||
|
||
# 3. Use the list of IDs to initialize the form
|
||
if selected_ids:
|
||
candidates_to_load = Candidate.objects.filter(pk__in=selected_ids)
|
||
form.initial["candidates"] = candidates_to_load
|
||
|
||
return render(
|
||
request,
|
||
"interviews/schedule_interviews.html",
|
||
{"form": form, "job": job},
|
||
)
|
||
|
||
|
||
def _handle_preview_submission(request, slug, job):
|
||
"""
|
||
Handles the initial POST request (Preview Schedule).
|
||
Validates forms, calculates slots, saves data to session, and renders preview.
|
||
"""
|
||
SESSION_DATA_KEY = "interview_schedule_data"
|
||
form = InterviewScheduleForm(slug, request.POST)
|
||
# break_formset = BreakTimeFormSet(request.POST,prefix='breaktime')
|
||
|
||
if form.is_valid():
|
||
# Get the form data
|
||
candidates = form.cleaned_data["candidates"]
|
||
start_date = form.cleaned_data["start_date"]
|
||
end_date = form.cleaned_data["end_date"]
|
||
working_days = form.cleaned_data["working_days"]
|
||
start_time = form.cleaned_data["start_time"]
|
||
end_time = form.cleaned_data["end_time"]
|
||
interview_duration = form.cleaned_data["interview_duration"]
|
||
buffer_time = form.cleaned_data["buffer_time"]
|
||
break_start_time = form.cleaned_data["break_start_time"]
|
||
break_end_time = form.cleaned_data["break_end_time"]
|
||
|
||
# Process break times
|
||
# breaks = []
|
||
# for break_form in break_formset:
|
||
# print(break_form.cleaned_data)
|
||
# if break_form.cleaned_data and not break_form.cleaned_data.get("DELETE"):
|
||
# breaks.append(
|
||
# {
|
||
# "start_time": break_form.cleaned_data["start_time"].strftime("%H:%M:%S"),
|
||
# "end_time": break_form.cleaned_data["end_time"].strftime("%H:%M:%S"),
|
||
# }
|
||
# )
|
||
|
||
# Create a temporary schedule object (not saved to DB)
|
||
temp_schedule = InterviewSchedule(
|
||
job=job,
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
working_days=working_days,
|
||
start_time=start_time,
|
||
end_time=end_time,
|
||
interview_duration=interview_duration,
|
||
buffer_time=buffer_time,
|
||
break_start_time=break_start_time,
|
||
break_end_time=break_end_time
|
||
)
|
||
|
||
# Get available slots (temp_breaks logic moved into get_available_time_slots if needed)
|
||
available_slots = get_available_time_slots(temp_schedule)
|
||
|
||
if len(available_slots) < len(candidates):
|
||
messages.error(
|
||
request,
|
||
f"Not enough available slots. Required: {len(candidates)}, Available: {len(available_slots)}",
|
||
)
|
||
return render(
|
||
request,
|
||
"interviews/schedule_interviews.html",
|
||
{"form": form, "job": job},
|
||
)
|
||
|
||
# Create a preview schedule
|
||
preview_schedule = []
|
||
for i, candidate in enumerate(candidates):
|
||
slot = available_slots[i]
|
||
preview_schedule.append(
|
||
{"candidate": candidate, "date": slot["date"], "time": slot["time"]}
|
||
)
|
||
|
||
# Save the form data to session for later use
|
||
schedule_data = {
|
||
"start_date": start_date.isoformat(),
|
||
"end_date": end_date.isoformat(),
|
||
"working_days": working_days,
|
||
"start_time": start_time.isoformat(),
|
||
"end_time": end_time.isoformat(),
|
||
"interview_duration": interview_duration,
|
||
"buffer_time": buffer_time,
|
||
"break_start_time": break_start_time.isoformat(),
|
||
"break_end_time": break_end_time.isoformat(),
|
||
"candidate_ids": [c.id for c in candidates],
|
||
}
|
||
request.session[SESSION_DATA_KEY] = schedule_data
|
||
|
||
# Render the preview page
|
||
return render(
|
||
request,
|
||
"interviews/preview_schedule.html",
|
||
{
|
||
"job": job,
|
||
"schedule": preview_schedule,
|
||
"start_date": start_date,
|
||
"end_date": end_date,
|
||
"working_days": working_days,
|
||
"start_time": start_time,
|
||
"end_time": end_time,
|
||
"break_start_time": break_start_time,
|
||
"break_end_time": break_end_time,
|
||
"interview_duration": interview_duration,
|
||
"buffer_time": buffer_time,
|
||
},
|
||
)
|
||
else:
|
||
# Re-render the form if validation fails
|
||
return render(
|
||
request,
|
||
"interviews/schedule_interviews.html",
|
||
{"form": form, "job": job},
|
||
)
|
||
|
||
|
||
def _handle_confirm_schedule(request, slug, job):
|
||
"""
|
||
Handles the final POST request (Confirm Schedule).
|
||
Creates the main schedule record and queues individual interviews asynchronously.
|
||
"""
|
||
|
||
|
||
SESSION_DATA_KEY = "interview_schedule_data"
|
||
SESSION_ID_KEY = f"schedule_candidate_ids_{slug}"
|
||
|
||
# 1. Get schedule data from session
|
||
schedule_data = request.session.get(SESSION_DATA_KEY)
|
||
|
||
if not schedule_data:
|
||
messages.error(request, "Session expired. Please try again.")
|
||
return redirect("schedule_interviews", slug=slug)
|
||
|
||
# 2. Create the Interview Schedule (Parent Record)
|
||
# NOTE: You MUST convert the time strings back to Python time objects here.
|
||
try:
|
||
schedule = InterviewSchedule.objects.create(
|
||
job=job,
|
||
created_by=request.user,
|
||
start_date=datetime.fromisoformat(schedule_data["start_date"]).date(),
|
||
end_date=datetime.fromisoformat(schedule_data["end_date"]).date(),
|
||
working_days=schedule_data["working_days"],
|
||
start_time=time.fromisoformat(schedule_data["start_time"]),
|
||
end_time=time.fromisoformat(schedule_data["end_time"]),
|
||
interview_duration=schedule_data["interview_duration"],
|
||
buffer_time=schedule_data["buffer_time"],
|
||
|
||
# Use the simple break times saved in the session
|
||
# If the value is None (because required=False in form), handle it gracefully
|
||
break_start_time=schedule_data.get("break_start_time"),
|
||
break_end_time=schedule_data.get("break_end_time"),
|
||
)
|
||
except Exception as e:
|
||
# Handle database creation error
|
||
messages.error(request, f"Error creating schedule: {e}")
|
||
if SESSION_ID_KEY in request.session: del request.session[SESSION_ID_KEY]
|
||
return redirect("schedule_interviews", slug=slug)
|
||
|
||
|
||
# 3. Setup candidates and get slots
|
||
candidates = Candidate.objects.filter(id__in=schedule_data["candidate_ids"])
|
||
schedule.candidates.set(candidates)
|
||
available_slots = get_available_time_slots(schedule) # This should still be synchronous and fast
|
||
|
||
# 4. Queue scheduled interviews asynchronously (FAST RESPONSE)
|
||
queued_count = 0
|
||
for i, candidate in enumerate(candidates):
|
||
if i < len(available_slots):
|
||
slot = available_slots[i]
|
||
|
||
# Dispatch the individual creation task to the background queue
|
||
async_task(
|
||
"recruitment.tasks.create_interview_and_meeting",
|
||
candidate.pk,
|
||
job.pk,
|
||
schedule.pk,
|
||
slot['date'],
|
||
slot['time'],
|
||
schedule.interview_duration,
|
||
)
|
||
queued_count += 1
|
||
|
||
# 5. Success and Cleanup (IMMEDIATE RESPONSE)
|
||
messages.success(
|
||
request,
|
||
f"Schedule successfully created. Queued {queued_count} interviews to be booked asynchronously. Check back in a moment!"
|
||
)
|
||
|
||
# Clear both session data keys upon successful completion
|
||
if SESSION_DATA_KEY in request.session: del request.session[SESSION_DATA_KEY]
|
||
if SESSION_ID_KEY in request.session: del request.session[SESSION_ID_KEY]
|
||
|
||
return redirect("job_detail", slug=slug)
|
||
|
||
def schedule_interviews_view(request, slug):
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
if request.method == "POST":
|
||
# return _handle_confirm_schedule(request, slug, job)
|
||
return _handle_preview_submission(request, slug, job)
|
||
else:
|
||
return _handle_get_request(request, slug, job)
|
||
def confirm_schedule_interviews_view(request, slug):
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
if request.method == "POST":
|
||
return _handle_confirm_schedule(request, slug, job)
|
||
|
||
|
||
@login_required
|
||
def candidate_screening_view(request, slug):
|
||
"""
|
||
Manage candidate tiers and stage transitions
|
||
"""
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
candidates = job.screening_candidates
|
||
|
||
# Get filter parameters
|
||
min_ai_score_str = request.GET.get('min_ai_score')
|
||
min_experience_str = request.GET.get('min_experience')
|
||
screening_rating = request.GET.get('screening_rating')
|
||
tier1_count_str = request.GET.get('tier1_count')
|
||
|
||
try:
|
||
# Check if the string value exists and is not an empty string before conversion
|
||
if min_ai_score_str:
|
||
min_ai_score = int(min_ai_score_str)
|
||
else:
|
||
min_ai_score = 0
|
||
|
||
if min_experience_str:
|
||
min_experience = float(min_experience_str)
|
||
else:
|
||
min_experience = 0
|
||
|
||
if tier1_count_str:
|
||
tier1_count = int(tier1_count_str)
|
||
else:
|
||
tier1_count = 0
|
||
|
||
except ValueError:
|
||
# This catches if the user enters non-numeric text (e.g., "abc")
|
||
min_ai_score = 0
|
||
min_experience = 0
|
||
tier1_count = 0
|
||
|
||
# Apply filters
|
||
if min_ai_score > 0:
|
||
candidates = candidates.filter(ai_analysis_data__analysis_data__match_score__gte=min_ai_score)
|
||
|
||
if min_experience > 0:
|
||
candidates = candidates.filter(ai_analysis_data__analysis_data__years_of_experience__gte=min_experience)
|
||
|
||
if screening_rating:
|
||
candidates = candidates.filter(ai_analysis_data__analysis_data__screening_stage_rating=screening_rating)
|
||
|
||
if tier1_count > 0:
|
||
candidates = candidates[:tier1_count]
|
||
|
||
context = {
|
||
"job": job,
|
||
"candidates": candidates,
|
||
'min_ai_score':min_ai_score,
|
||
'min_experience':min_experience,
|
||
'screening_rating':screening_rating,
|
||
'tier1_count':tier1_count,
|
||
"current_stage" : "Applied"
|
||
}
|
||
|
||
return render(request, "recruitment/candidate_screening_view.html", context)
|
||
|
||
|
||
@login_required
|
||
def candidate_exam_view(request, slug):
|
||
"""
|
||
Manage candidate tiers and stage transitions
|
||
"""
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
context = {
|
||
"job": job,
|
||
"candidates": job.exam_candidates,
|
||
'current_stage' : "Exam"
|
||
}
|
||
return render(request, "recruitment/candidate_exam_view.html", context)
|
||
|
||
@login_required
|
||
def update_candidate_exam_status(request, slug):
|
||
candidate = get_object_or_404(Candidate, slug=slug)
|
||
if request.method == "POST":
|
||
form = CandidateExamDateForm(request.POST, instance=candidate)
|
||
if form.is_valid():
|
||
form.save()
|
||
return redirect("candidate_exam_view", slug=candidate.job.slug)
|
||
else:
|
||
form = CandidateExamDateForm(request.POST, instance=candidate)
|
||
return render(request, "includes/candidate_exam_status_form.html", {"candidate": candidate,"form": form})
|
||
@login_required
|
||
def bulk_update_candidate_exam_status(request,slug):
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
status = request.headers.get('status')
|
||
if status:
|
||
for candidate in get_candidates_from_request(request):
|
||
try:
|
||
if status == "pass":
|
||
candidate.exam_status = "Passed"
|
||
candidate.stage = "Interview"
|
||
else:
|
||
candidate.exam_status = "Failed"
|
||
candidate.save()
|
||
except Exception as e:
|
||
print(e)
|
||
messages.success(request, f"Updated exam status selected candidates")
|
||
return redirect("candidate_exam_view", slug=job.slug)
|
||
|
||
def candidate_criteria_view_htmx(request, pk):
|
||
candidate = get_object_or_404(Candidate, pk=pk)
|
||
return render(request, "includes/candidate_modal_body.html", {"candidate": candidate})
|
||
|
||
|
||
@login_required
|
||
def candidate_set_exam_date(request, slug):
|
||
candidate = get_object_or_404(Candidate, slug=slug)
|
||
candidate.exam_date = timezone.now()
|
||
candidate.save()
|
||
messages.success(request, f"Set exam date for {candidate.name} to {candidate.exam_date}")
|
||
return redirect("candidate_screening_view", slug=candidate.job.slug)
|
||
|
||
@login_required
|
||
def candidate_update_status(request, slug):
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
mark_as = request.POST.get('mark_as')
|
||
if mark_as != '----------':
|
||
candidate_ids = request.POST.getlist("candidate_ids")
|
||
if c := Candidate.objects.filter(pk__in = candidate_ids):
|
||
c.update(stage=mark_as,exam_date=timezone.now(),applicant_status="Candidate" if mark_as in ["Exam","Interview","Offer"] else "Applicant")
|
||
|
||
messages.success(request, f"Candidates Updated")
|
||
response = HttpResponse(redirect("candidate_screening_view", slug=job.slug))
|
||
response.headers["HX-Refresh"] = "true"
|
||
return response
|
||
|
||
@login_required
|
||
def candidate_interview_view(request,slug):
|
||
job = get_object_or_404(JobPosting,slug=slug)
|
||
context = {"job":job,"candidates":job.interview_candidates,'current_stage':'Interview'}
|
||
return render(request,"recruitment/candidate_interview_view.html",context)
|
||
|
||
@login_required
|
||
def reschedule_meeting_for_candidate(request,slug,candidate_id,meeting_id):
|
||
job = get_object_or_404(JobPosting,slug=slug)
|
||
candidate = get_object_or_404(Candidate,pk=candidate_id)
|
||
meeting = get_object_or_404(ZoomMeeting,pk=meeting_id)
|
||
form = ZoomMeetingForm(instance=meeting)
|
||
|
||
if request.method == "POST":
|
||
form = ZoomMeetingForm(request.POST,instance=meeting)
|
||
if form.is_valid():
|
||
instance = form.save(commit=False)
|
||
updated_data = {
|
||
"topic": instance.topic,
|
||
"start_time": instance.start_time.isoformat() + "Z",
|
||
"duration": instance.duration,
|
||
}
|
||
if instance.start_time < timezone.now():
|
||
messages.error(request, "Start time must be in the future.")
|
||
return redirect("reschedule_meeting_for_candidate",slug=job.slug,candidate_id=candidate_id,meeting_id=meeting_id)
|
||
|
||
result = update_meeting(instance, updated_data)
|
||
|
||
if result["status"] == "success":
|
||
messages.success(request, result["message"])
|
||
else:
|
||
messages.error(request, result["message"])
|
||
return redirect(reverse("candidate_interview_view", kwargs={"slug": job.slug}))
|
||
|
||
context = {"job":job,"candidate":candidate,"meeting":meeting,"form":form}
|
||
return render(request,"meetings/reschedule_meeting.html",context)
|
||
|
||
|
||
@login_required
|
||
def delete_meeting_for_candidate(request,slug,candidate_pk,meeting_id):
|
||
job = get_object_or_404(JobPosting,slug=slug)
|
||
candidate = get_object_or_404(Candidate,pk=candidate_pk)
|
||
meeting = get_object_or_404(ZoomMeeting,pk=meeting_id)
|
||
if request.method == "POST":
|
||
result = delete_zoom_meeting(meeting.meeting_id)
|
||
if result["status"] == "success" or "Meeting does not exist" in result["details"]["message"]:
|
||
meeting.delete()
|
||
messages.success(request, "Meeting deleted successfully")
|
||
else:
|
||
messages.error(request, result["message"])
|
||
return redirect(reverse("candidate_interview_view", kwargs={"slug": job.slug}))
|
||
|
||
context = {"job":job,"candidate":candidate,"meeting":meeting,'delete_url':reverse("delete_meeting_for_candidate",kwargs={"slug":job.slug,"candidate_pk":candidate_pk,"meeting_id":meeting_id})}
|
||
return render(request,"meetings/delete_meeting_form.html",context)
|
||
|
||
@login_required
|
||
def interview_calendar_view(request, slug):
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
|
||
# Get all scheduled interviews for this job
|
||
scheduled_interviews = ScheduledInterview.objects.filter(
|
||
job=job
|
||
).select_related('candidate', 'zoom_meeting')
|
||
|
||
# Convert interviews to calendar events
|
||
events = []
|
||
for interview in scheduled_interviews:
|
||
# Create start datetime
|
||
start_datetime = datetime.combine(
|
||
interview.interview_date,
|
||
interview.interview_time
|
||
)
|
||
|
||
# Calculate end datetime based on interview duration
|
||
duration = interview.zoom_meeting.duration if interview.zoom_meeting else 60
|
||
end_datetime = start_datetime + timedelta(minutes=duration)
|
||
|
||
# Determine event color based on status
|
||
color = '#00636e' # Default color
|
||
if interview.status == 'confirmed':
|
||
color = '#00a86b' # Green for confirmed
|
||
elif interview.status == 'cancelled':
|
||
color = '#e74c3c' # Red for cancelled
|
||
elif interview.status == 'completed':
|
||
color = '#95a5a6' # Gray for completed
|
||
|
||
events.append({
|
||
'title': f"Interview: {interview.candidate.name}",
|
||
'start': start_datetime.isoformat(),
|
||
'end': end_datetime.isoformat(),
|
||
'url': f"{request.path}interview/{interview.id}/",
|
||
'color': color,
|
||
'extendedProps': {
|
||
'candidate': interview.candidate.name,
|
||
'email': interview.candidate.email,
|
||
'status': interview.status,
|
||
'meeting_id': interview.zoom_meeting.meeting_id if interview.zoom_meeting else None,
|
||
'join_url': interview.zoom_meeting.join_url if interview.zoom_meeting else None,
|
||
}
|
||
})
|
||
|
||
context = {
|
||
'job': job,
|
||
'events': events,
|
||
'calendar_color': '#00636e',
|
||
}
|
||
|
||
return render(request, 'recruitment/interview_calendar.html', context)
|
||
|
||
@login_required
|
||
def interview_detail_view(request, slug, interview_id):
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
interview = get_object_or_404(
|
||
ScheduledInterview,
|
||
id=interview_id,
|
||
job=job
|
||
)
|
||
|
||
context = {
|
||
'job': job,
|
||
'interview': interview,
|
||
}
|
||
|
||
return render(request, 'recruitment/interview_detail.html', context)
|
||
|
||
# Candidate Meeting Scheduling/Rescheduling Views
|
||
@require_POST
|
||
def api_schedule_candidate_meeting(request, job_slug, candidate_pk):
|
||
"""
|
||
Handle POST request to schedule a Zoom meeting for a candidate via HTMX.
|
||
Returns JSON response for modal update.
|
||
"""
|
||
job = get_object_or_404(JobPosting, slug=job_slug)
|
||
candidate = get_object_or_404(Candidate, pk=candidate_pk, job=job)
|
||
|
||
topic = f"Interview: {job.title} with {candidate.name}"
|
||
start_time_str = request.POST.get('start_time')
|
||
duration = int(request.POST.get('duration', 60))
|
||
|
||
if not start_time_str:
|
||
return JsonResponse({'success': False, 'error': 'Start time is required.'}, status=400)
|
||
|
||
try:
|
||
# Parse datetime from datetime-local input (YYYY-MM-DDTHH:MM)
|
||
# This will be in server's timezone, create_zoom_meeting will handle UTC conversion
|
||
naive_start_time = datetime.fromisoformat(start_time_str)
|
||
# Ensure it's timezone-aware if your system requires it, or let create_zoom_meeting handle it.
|
||
# For simplicity, assuming create_zoom_meeting handles naive datetimes or they are in UTC.
|
||
# If start_time is expected to be in a specific timezone, convert it here.
|
||
# e.g., start_time = timezone.make_aware(naive_start_time, timezone.get_current_timezone())
|
||
start_time = naive_start_time # Or timezone.make_aware(naive_start_time)
|
||
except ValueError:
|
||
return JsonResponse({'success': False, 'error': 'Invalid date/time format for start time.'}, status=400)
|
||
|
||
if start_time <= timezone.now():
|
||
return JsonResponse({'success': False, 'error': 'Start time must be in the future.'}, status=400)
|
||
|
||
result = create_zoom_meeting(topic=topic, start_time=start_time, duration=duration)
|
||
|
||
if result["status"] == "success":
|
||
zoom_meeting_details = result["meeting_details"]
|
||
zoom_meeting = ZoomMeeting.objects.create(
|
||
topic=topic,
|
||
start_time=start_time, # Store in local timezone
|
||
duration=duration,
|
||
meeting_id=zoom_meeting_details["meeting_id"],
|
||
join_url=zoom_meeting_details["join_url"],
|
||
password=zoom_meeting_details["password"],
|
||
# host_email=zoom_meeting_details["host_email"],
|
||
status=result["zoom_gateway_response"].get("status", "waiting"),
|
||
zoom_gateway_response=result["zoom_gateway_response"],
|
||
)
|
||
scheduled_interview = ScheduledInterview.objects.create(
|
||
candidate=candidate,
|
||
job=job,
|
||
zoom_meeting=zoom_meeting,
|
||
interview_date=start_time.date(),
|
||
interview_time=start_time.time(),
|
||
status='scheduled' # Or 'confirmed' depending on your workflow
|
||
)
|
||
messages.success(request, f"Meeting scheduled with {candidate.name}.")
|
||
|
||
# Return updated table row or a success message
|
||
# For HTMX, you might want to return a fragment of the updated table
|
||
# For now, returning JSON to indicate success and close modal
|
||
return JsonResponse({
|
||
'success': True,
|
||
'message': 'Meeting scheduled successfully!',
|
||
'join_url': zoom_meeting.join_url,
|
||
'meeting_id': zoom_meeting.meeting_id,
|
||
'candidate_name': candidate.name,
|
||
'interview_datetime': start_time.strftime("%Y-%m-%d %H:%M")
|
||
})
|
||
else:
|
||
messages.error(request, result["message"])
|
||
return JsonResponse({'success': False, 'error': result["message"]}, status=400)
|
||
|
||
|
||
def schedule_candidate_meeting(request, job_slug, candidate_pk):
|
||
"""
|
||
GET: Render modal form to schedule a meeting. (For HTMX)
|
||
POST: Handled by api_schedule_candidate_meeting.
|
||
"""
|
||
job = get_object_or_404(JobPosting, slug=job_slug)
|
||
candidate = get_object_or_404(Candidate, pk=candidate_pk, job=job)
|
||
|
||
if request.method == "POST":
|
||
return api_schedule_candidate_meeting(request, job_slug, candidate_pk)
|
||
|
||
# GET request - render the form snippet for HTMX
|
||
context = {
|
||
'job': job,
|
||
'candidate': candidate,
|
||
'action_url': reverse('api_schedule_candidate_meeting', kwargs={'job_slug': job_slug, 'candidate_pk': candidate_pk}),
|
||
'scheduled_interview': None, # Explicitly None for schedule
|
||
}
|
||
# Render just the form part, or the whole modal body content
|
||
return render(request, "includes/meeting_form.html", context)
|
||
|
||
|
||
@require_http_methods(["GET", "POST"])
|
||
def api_schedule_candidate_meeting(request, job_slug, candidate_pk):
|
||
"""
|
||
Handles GET to render form and POST to process scheduling.
|
||
"""
|
||
job = get_object_or_404(JobPosting, slug=job_slug)
|
||
candidate = get_object_or_404(Candidate, pk=candidate_pk, job=job)
|
||
|
||
if request.method == "GET":
|
||
# This GET is for HTMX to fetch the form
|
||
context = {
|
||
'job': job,
|
||
'candidate': candidate,
|
||
'action_url': reverse('api_schedule_candidate_meeting', kwargs={'job_slug': job_slug, 'candidate_pk': candidate_pk}),
|
||
'scheduled_interview': None,
|
||
}
|
||
return render(request, "includes/meeting_form.html", context)
|
||
|
||
# POST logic (remains the same)
|
||
topic = f"Interview: {job.title} with {candidate.name}"
|
||
start_time_str = request.POST.get('start_time')
|
||
duration = int(request.POST.get('duration', 60))
|
||
|
||
if not start_time_str:
|
||
return JsonResponse({'success': False, 'error': 'Start time is required.'}, status=400)
|
||
|
||
try:
|
||
naive_start_time = datetime.fromisoformat(start_time_str)
|
||
start_time = naive_start_time
|
||
except ValueError:
|
||
return JsonResponse({'success': False, 'error': 'Invalid date/time format for start time.'}, status=400)
|
||
|
||
if start_time <= timezone.now():
|
||
return JsonResponse({'success': False, 'error': 'Start time must be in the future.'}, status=400)
|
||
|
||
result = create_zoom_meeting(topic=topic, start_time=start_time, duration=duration)
|
||
|
||
if result["status"] == "success":
|
||
zoom_meeting_details = result["meeting_details"]
|
||
zoom_meeting = ZoomMeeting.objects.create(
|
||
topic=topic,
|
||
start_time=start_time,
|
||
duration=duration,
|
||
meeting_id=zoom_meeting_details["meeting_id"],
|
||
join_url=zoom_meeting_details["join_url"],
|
||
password=zoom_meeting_details["password"],
|
||
host_email=zoom_meeting_details["host_email"],
|
||
status=result["zoom_gateway_response"].get("status", "waiting"),
|
||
zoom_gateway_response=result["zoom_gateway_response"],
|
||
)
|
||
scheduled_interview = ScheduledInterview.objects.create(
|
||
candidate=candidate,
|
||
job=job,
|
||
zoom_meeting=zoom_meeting,
|
||
interview_date=start_time.date(),
|
||
interview_time=start_time.time(),
|
||
status='scheduled'
|
||
)
|
||
messages.success(request, f"Meeting scheduled with {candidate.name}.")
|
||
return JsonResponse({
|
||
'success': True,
|
||
'message': 'Meeting scheduled successfully!',
|
||
'join_url': zoom_meeting.join_url,
|
||
'meeting_id': zoom_meeting.meeting_id,
|
||
'candidate_name': candidate.name,
|
||
'interview_datetime': start_time.strftime("%Y-%m-%d %H:%M")
|
||
})
|
||
else:
|
||
messages.error(request, result["message"])
|
||
return JsonResponse({'success': False, 'error': result["message"]}, status=400)
|
||
|
||
|
||
@require_http_methods(["GET", "POST"])
|
||
def api_reschedule_candidate_meeting(request, job_slug, candidate_pk, interview_pk):
|
||
"""
|
||
Handles GET to render form and POST to process rescheduling.
|
||
"""
|
||
job = get_object_or_404(JobPosting, slug=job_slug)
|
||
scheduled_interview = get_object_or_404(
|
||
ScheduledInterview.objects.select_related('zoom_meeting'),
|
||
pk=interview_pk,
|
||
candidate__pk=candidate_pk,
|
||
job=job
|
||
)
|
||
zoom_meeting = scheduled_interview.zoom_meeting
|
||
|
||
if request.method == "GET":
|
||
# This GET is for HTMX to fetch the form
|
||
initial_data = {
|
||
'topic': zoom_meeting.topic,
|
||
'start_time': zoom_meeting.start_time.strftime('%Y-%m-%dT%H:%M'),
|
||
'duration': zoom_meeting.duration,
|
||
}
|
||
context = {
|
||
'job': job,
|
||
'candidate': scheduled_interview.candidate,
|
||
'scheduled_interview': scheduled_interview, # Pass for conditional logic in template
|
||
'initial_data': initial_data,
|
||
'action_url': reverse('api_reschedule_candidate_meeting', kwargs={'job_slug': job_slug, 'candidate_pk': candidate_pk, 'interview_pk': interview_pk})
|
||
}
|
||
return render(request, "includes/meeting_form.html", context)
|
||
|
||
# POST logic (remains the same)
|
||
new_start_time_str = request.POST.get('start_time')
|
||
new_duration = int(request.POST.get('duration', zoom_meeting.duration))
|
||
|
||
if not new_start_time_str:
|
||
return JsonResponse({'success': False, 'error': 'New start time is required.'}, status=400)
|
||
|
||
try:
|
||
naive_new_start_time = datetime.fromisoformat(new_start_time_str)
|
||
new_start_time = naive_new_start_time
|
||
except ValueError:
|
||
return JsonResponse({'success': False, 'error': 'Invalid date/time format for new start time.'}, status=400)
|
||
|
||
if new_start_time <= timezone.now():
|
||
return JsonResponse({'success': False, 'error': 'Start time must be in the future.'}, status=400)
|
||
|
||
updated_data = {
|
||
"topic": f"Interview: {job.title} with {scheduled_interview.candidate.name}",
|
||
"start_time": new_start_time.isoformat() + "Z",
|
||
"duration": new_duration,
|
||
}
|
||
|
||
result = update_zoom_meeting(zoom_meeting.meeting_id, updated_data)
|
||
|
||
if result["status"] == "success":
|
||
details_result = get_zoom_meeting_details(zoom_meeting.meeting_id)
|
||
if details_result["status"] == "success":
|
||
updated_zoom_details = details_result["meeting_details"]
|
||
zoom_meeting.topic = updated_zoom_details.get("topic", zoom_meeting.topic)
|
||
zoom_meeting.start_time = new_start_time
|
||
zoom_meeting.duration = new_duration
|
||
zoom_meeting.join_url = updated_zoom_details.get("join_url", zoom_meeting.join_url)
|
||
zoom_meeting.password = updated_zoom_details.get("password", zoom_meeting.password)
|
||
zoom_meeting.status = updated_zoom_details.get("status", zoom_meeting.status)
|
||
zoom_meeting.zoom_gateway_response = updated_zoom_details
|
||
zoom_meeting.save()
|
||
|
||
scheduled_interview.interview_date = new_start_time.date()
|
||
scheduled_interview.interview_time = new_start_time.time()
|
||
scheduled_interview.status = 'rescheduled'
|
||
scheduled_interview.save()
|
||
messages.success(request, f"Meeting for {scheduled_interview.candidate.name} rescheduled.")
|
||
else:
|
||
logger.warning(f"Zoom meeting {zoom_meeting.meeting_id} updated, but failed to fetch latest details.")
|
||
zoom_meeting.start_time = new_start_time
|
||
zoom_meeting.duration = new_duration
|
||
zoom_meeting.save()
|
||
scheduled_interview.interview_date = new_start_time.date()
|
||
scheduled_interview.interview_time = new_start_time.time()
|
||
scheduled_interview.save()
|
||
messages.success(request, f"Meeting for {scheduled_interview.candidate.name} rescheduled. (Note: Could not refresh all details from Zoom.)")
|
||
|
||
return JsonResponse({
|
||
'success': True,
|
||
'message': 'Meeting rescheduled successfully!',
|
||
'join_url': zoom_meeting.join_url,
|
||
'new_interview_datetime': new_start_time.strftime("%Y-%m-%d %H:%M")
|
||
})
|
||
else:
|
||
messages.error(request, result["message"])
|
||
return JsonResponse({'success': False, 'error': result["message"]}, status=400)
|
||
|
||
# The original schedule_candidate_meeting and reschedule_candidate_meeting (without api_ prefix)
|
||
# can be removed if their only purpose was to be called by the JS onclicks.
|
||
# If they were intended for other direct URL access, they can be kept as simple redirects
|
||
# or wrappers to the api_ versions.
|
||
# For now, let's assume the api_ versions are the primary ones for HTMX.
|
||
|
||
|
||
def reschedule_candidate_meeting(request, job_slug, candidate_pk, interview_pk):
|
||
"""
|
||
Handles GET to display a form for rescheduling a meeting.
|
||
Handles POST to process the rescheduling of a meeting.
|
||
"""
|
||
job = get_object_or_404(JobPosting, slug=job_slug)
|
||
candidate = get_object_or_404(Candidate, pk=candidate_pk, job=job)
|
||
scheduled_interview = get_object_or_404(
|
||
ScheduledInterview.objects.select_related('zoom_meeting'),
|
||
pk=interview_pk,
|
||
candidate=candidate,
|
||
job=job
|
||
)
|
||
zoom_meeting = scheduled_interview.zoom_meeting
|
||
|
||
# Determine if the candidate has other future meetings
|
||
# This helps in providing context in the template
|
||
# Note: This checks for *any* future meetings for the candidate, not just the one being rescheduled.
|
||
# If candidate.has_future_meeting is True, it implies they have at least one other upcoming meeting,
|
||
# or the specific meeting being rescheduled is itself in the future.
|
||
# We can refine this logic if needed, e.g., check for meetings *other than* the current `interview_pk`.
|
||
has_other_future_meetings = candidate.has_future_meeting
|
||
# More precise check: if the current meeting being rescheduled is in the future, then by definition
|
||
# the candidate will have a future meeting (this one). The UI might want to know if there are *others*.
|
||
# For now, `candidate.has_future_meeting` is a good general indicator.
|
||
|
||
if request.method == "POST":
|
||
form = ZoomMeetingForm(request.POST)
|
||
if form.is_valid():
|
||
new_topic = form.cleaned_data.get('topic')
|
||
new_start_time = form.cleaned_data.get('start_time')
|
||
new_duration = form.cleaned_data.get('duration')
|
||
|
||
# Use a default topic if not provided, keeping the original structure
|
||
if not new_topic:
|
||
new_topic = f"Interview: {job.title} with {candidate.name}"
|
||
|
||
# Ensure new_start_time is in the future
|
||
if new_start_time <= timezone.now():
|
||
messages.error(request, "Start time must be in the future.")
|
||
# Re-render form with error and initial data
|
||
return render(request, "recruitment/schedule_meeting_form.html", { # Reusing the same form template
|
||
'form': form,
|
||
'job': job,
|
||
'candidate': candidate,
|
||
'scheduled_interview': scheduled_interview,
|
||
'initial_topic': new_topic,
|
||
'initial_start_time': new_start_time.strftime('%Y-%m-%dT%H:%M') if new_start_time else '',
|
||
'initial_duration': new_duration,
|
||
'action_url': reverse('reschedule_candidate_meeting', kwargs={'job_slug': job_slug, 'candidate_pk': candidate_pk, 'interview_pk': interview_pk}),
|
||
'has_future_meeting': has_other_future_meetings # Pass status for template
|
||
})
|
||
|
||
# Prepare data for Zoom API update
|
||
# The update_zoom_meeting expects start_time as ISO string with 'Z'
|
||
zoom_update_data = {
|
||
"topic": new_topic,
|
||
"start_time": new_start_time.isoformat() + "Z",
|
||
"duration": new_duration,
|
||
}
|
||
|
||
# Update Zoom meeting using utility function
|
||
zoom_update_result = update_zoom_meeting(zoom_meeting.meeting_id, zoom_update_data)
|
||
|
||
if zoom_update_result["status"] == "success":
|
||
# Fetch the latest details from Zoom after successful update
|
||
details_result = get_zoom_meeting_details(zoom_meeting.meeting_id)
|
||
|
||
if details_result["status"] == "success":
|
||
updated_zoom_details = details_result["meeting_details"]
|
||
# Update local ZoomMeeting record
|
||
zoom_meeting.topic = updated_zoom_details.get("topic", new_topic)
|
||
zoom_meeting.start_time = new_start_time # Store the original datetime
|
||
zoom_meeting.duration = new_duration
|
||
zoom_meeting.join_url = updated_zoom_details.get("join_url", zoom_meeting.join_url)
|
||
zoom_meeting.password = updated_zoom_details.get("password", zoom_meeting.password)
|
||
zoom_meeting.status = updated_zoom_details.get("status", zoom_meeting.status)
|
||
zoom_meeting.zoom_gateway_response = details_result.get("meeting_details")
|
||
zoom_meeting.save()
|
||
|
||
# Update ScheduledInterview record
|
||
scheduled_interview.interview_date = new_start_time.date()
|
||
scheduled_interview.interview_time = new_start_time.time()
|
||
scheduled_interview.status = 'rescheduled' # Or 'scheduled' if you prefer
|
||
scheduled_interview.save()
|
||
messages.success(request, f"Meeting for {candidate.name} rescheduled successfully.")
|
||
else:
|
||
# If fetching details fails, update with form data and log a warning
|
||
logger.warning(
|
||
f"Successfully updated Zoom meeting {zoom_meeting.meeting_id}, but failed to fetch updated details. "
|
||
f"Error: {details_result.get('message', 'Unknown error')}"
|
||
)
|
||
# Update with form data as a fallback
|
||
zoom_meeting.topic = new_topic
|
||
zoom_meeting.start_time = new_start_time
|
||
zoom_meeting.duration = new_duration
|
||
zoom_meeting.save()
|
||
scheduled_interview.interview_date = new_start_time.date()
|
||
scheduled_interview.interview_time = new_start_time.time()
|
||
scheduled_interview.save()
|
||
messages.success(request, f"Meeting for {candidate.name} rescheduled. (Note: Could not refresh all details from Zoom.)")
|
||
|
||
return redirect('candidate_interview_view', slug=job.slug)
|
||
else:
|
||
messages.error(request, f"Failed to update Zoom meeting: {zoom_update_result['message']}")
|
||
# Re-render form with error
|
||
return render(request, "recruitment/schedule_meeting_form.html", {
|
||
'form': form,
|
||
'job': job,
|
||
'candidate': candidate,
|
||
'scheduled_interview': scheduled_interview,
|
||
'initial_topic': new_topic,
|
||
'initial_start_time': new_start_time.strftime('%Y-%m-%dT%H:%M') if new_start_time else '',
|
||
'initial_duration': new_duration,
|
||
'action_url': reverse('reschedule_candidate_meeting', kwargs={'job_slug': job_slug, 'candidate_pk': candidate_pk, 'interview_pk': interview_pk}),
|
||
'has_future_meeting': has_other_future_meetings
|
||
})
|
||
else:
|
||
# Form validation errors
|
||
return render(request, "recruitment/schedule_meeting_form.html", {
|
||
'form': form,
|
||
'job': job,
|
||
'candidate': candidate,
|
||
'scheduled_interview': scheduled_interview,
|
||
'initial_topic': request.POST.get('topic', new_topic),
|
||
'initial_start_time': request.POST.get('start_time', new_start_time.strftime('%Y-%m-%dT%H:%M') if new_start_time else ''),
|
||
'initial_duration': request.POST.get('duration', new_duration),
|
||
'action_url': reverse('reschedule_candidate_meeting', kwargs={'job_slug': job_slug, 'candidate_pk': candidate_pk, 'interview_pk': interview_pk}),
|
||
'has_future_meeting': has_other_future_meetings
|
||
})
|
||
else: # GET request
|
||
# Pre-populate form with existing meeting details
|
||
initial_data = {
|
||
'topic': zoom_meeting.topic,
|
||
'start_time': zoom_meeting.start_time.strftime('%Y-%m-%dT%H:%M'),
|
||
'duration': zoom_meeting.duration,
|
||
}
|
||
form = ZoomMeetingForm(initial=initial_data)
|
||
return render(request, "recruitment/schedule_meeting_form.html", {
|
||
'form': form,
|
||
'job': job,
|
||
'candidate': candidate,
|
||
'scheduled_interview': scheduled_interview, # Pass to template for title/differentiation
|
||
'action_url': reverse('reschedule_candidate_meeting', kwargs={'job_slug': job_slug, 'candidate_pk': candidate_pk, 'interview_pk': interview_pk}),
|
||
'has_future_meeting': has_other_future_meetings # Pass status for template
|
||
})
|
||
|
||
|
||
def schedule_meeting_for_candidate(request, slug, candidate_pk):
|
||
"""
|
||
Handles GET to display a simple form for scheduling a meeting for a candidate.
|
||
Handles POST to process the form, create the meeting, and redirect back.
|
||
"""
|
||
job = get_object_or_404(JobPosting, slug=slug)
|
||
candidate = get_object_or_404(Candidate, pk=candidate_pk, job=job)
|
||
|
||
if request.method == "POST":
|
||
form = ZoomMeetingForm(request.POST)
|
||
if form.is_valid():
|
||
topic_val = form.cleaned_data.get('topic')
|
||
start_time_val = form.cleaned_data.get('start_time')
|
||
duration_val = form.cleaned_data.get('duration')
|
||
|
||
# Use a default topic if not provided
|
||
if not topic_val:
|
||
topic_val = f"Interview: {job.title} with {candidate.name}"
|
||
|
||
# Ensure start_time is in the future
|
||
if start_time_val <= timezone.now():
|
||
messages.error(request, "Start time must be in the future.")
|
||
# Re-render form with error and initial data
|
||
return redirect('candidate_interview_view', slug=job.slug)
|
||
# return render(request, "recruitment/schedule_meeting_form.html", {
|
||
# 'form': form,
|
||
# 'job': job,
|
||
# 'candidate': candidate,
|
||
# 'initial_topic': topic_val,
|
||
# 'initial_start_time': start_time_val.strftime('%Y-%m-%dT%H:%M') if start_time_val else '',
|
||
# 'initial_duration': duration_val
|
||
# })
|
||
|
||
# Create Zoom meeting using utility function
|
||
# The create_zoom_meeting expects start_time as a datetime object
|
||
# and handles its own conversion to UTC for the API call.
|
||
zoom_creation_result = create_zoom_meeting(
|
||
topic=topic_val,
|
||
start_time=start_time_val, # Pass the datetime object
|
||
duration=duration_val
|
||
)
|
||
|
||
if zoom_creation_result["status"] == "success":
|
||
zoom_details = zoom_creation_result["meeting_details"]
|
||
zoom_meeting_instance = ZoomMeeting.objects.create(
|
||
topic=topic_val,
|
||
start_time=start_time_val, # Store the original datetime
|
||
duration=duration_val,
|
||
meeting_id=zoom_details["meeting_id"],
|
||
join_url=zoom_details["join_url"],
|
||
password=zoom_details.get("password"), # password might be None
|
||
status=zoom_creation_result["zoom_gateway_response"].get("status", "waiting"),
|
||
zoom_gateway_response=zoom_creation_result["zoom_gateway_response"],
|
||
)
|
||
# Create a ScheduledInterview record
|
||
ScheduledInterview.objects.create(
|
||
candidate=candidate,
|
||
job=job,
|
||
zoom_meeting=zoom_meeting_instance,
|
||
interview_date=start_time_val.date(),
|
||
interview_time=start_time_val.time(),
|
||
status='scheduled'
|
||
)
|
||
messages.success(request, f"Meeting scheduled with {candidate.name}.")
|
||
return redirect('candidate_interview_view', slug=job.slug)
|
||
else:
|
||
messages.error(request, f"Failed to create Zoom meeting: {zoom_creation_result['message']}")
|
||
# Re-render form with error
|
||
return render(request, "recruitment/schedule_meeting_form.html", {
|
||
'form': form,
|
||
'job': job,
|
||
'candidate': candidate,
|
||
'initial_topic': topic_val,
|
||
'initial_start_time': start_time_val.strftime('%Y-%m-%dT%H:%M') if start_time_val else '',
|
||
'initial_duration': duration_val
|
||
})
|
||
else:
|
||
# Form validation errors
|
||
return render(request, "meetings/schedule_meeting_form.html", {
|
||
'form': form,
|
||
'job': job,
|
||
'candidate': candidate,
|
||
'initial_topic': request.POST.get('topic', f"Interview: {job.title} with {candidate.name}"),
|
||
'initial_start_time': request.POST.get('start_time', ''),
|
||
'initial_duration': request.POST.get('duration', 60)
|
||
})
|
||
else: # GET request
|
||
initial_data = {
|
||
'topic': f"Interview: {job.title} with {candidate.name}",
|
||
'start_time': (timezone.now() + timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M'), # Default to 1 hour from now
|
||
'duration': 60, # Default duration
|
||
}
|
||
form = ZoomMeetingForm(initial=initial_data)
|
||
return render(request, "meetings/schedule_meeting_form.html", {
|
||
'form': form,
|
||
'job': job,
|
||
'candidate': candidate
|
||
})
|
||
|
||
|
||
from django.core.exceptions import ObjectDoesNotExist
|
||
|
||
def user_profile_image_update(request, pk):
|
||
user = get_object_or_404(User, pk=pk)
|
||
try:
|
||
instance =user.profile
|
||
|
||
except ObjectDoesNotExist as e:
|
||
Profile.objects.create(user=user)
|
||
|
||
if request.method == 'POST':
|
||
profile_form = ProfileImageUploadForm(request.POST, request.FILES, instance=user.profile)
|
||
if profile_form.is_valid():
|
||
profile_form.save()
|
||
messages.success(request, 'Image uploaded successfully')
|
||
return redirect('user_detail', pk=user.pk)
|
||
else:
|
||
messages.error(request, 'An error occurred while uploading the image. Please check the errors below.')
|
||
else:
|
||
profile_form = ProfileImageUploadForm(instance=user.profile)
|
||
|
||
context = {
|
||
'profile_form': profile_form,
|
||
'user': user,
|
||
}
|
||
return render(request, 'user/profile.html', context)
|
||
|
||
def user_detail(request, pk):
|
||
user = get_object_or_404(User, pk=pk)
|
||
|
||
try:
|
||
profile_instance = user.profile
|
||
profile_form = ProfileImageUploadForm(instance=profile_instance)
|
||
except:
|
||
profile_form = ProfileImageUploadForm()
|
||
|
||
if request.method == 'POST':
|
||
first_name=request.POST.get('first_name')
|
||
last_name=request.POST.get('last_name')
|
||
if first_name:
|
||
user.first_name=first_name
|
||
if last_name:
|
||
user.last_name=last_name
|
||
user.save()
|
||
context = {
|
||
|
||
'user': user,
|
||
'profile_form':profile_form
|
||
|
||
}
|
||
return render(request, 'user/profile.html', context)
|
||
|
||
|
||
|
||
|
||
def easy_logs(request):
|
||
"""
|
||
Function-based view to display Django Easy Audit logs with tab switching and pagination.
|
||
"""
|
||
logs_per_page = 20
|
||
|
||
|
||
active_tab = request.GET.get('tab', 'crud')
|
||
|
||
if active_tab == 'login':
|
||
queryset = LoginEvent.objects.order_by('-datetime')
|
||
tab_title = _("User Authentication")
|
||
elif active_tab == 'request':
|
||
queryset = RequestEvent.objects.order_by('-datetime')
|
||
tab_title = _("HTTP Requests")
|
||
else:
|
||
queryset = CRUDEvent.objects.order_by('-datetime')
|
||
tab_title = _("Model Changes (CRUD)")
|
||
active_tab = 'crud'
|
||
|
||
|
||
paginator = Paginator(queryset, logs_per_page)
|
||
page = request.GET.get('page')
|
||
|
||
try:
|
||
|
||
logs_page = paginator.page(page)
|
||
except PageNotAnInteger:
|
||
|
||
logs_page = paginator.page(1)
|
||
except EmptyPage:
|
||
|
||
logs_page = paginator.page(paginator.num_pages)
|
||
|
||
context = {
|
||
'logs': logs_page,
|
||
'total_count': queryset.count(),
|
||
'active_tab': active_tab,
|
||
'tab_title': tab_title,
|
||
}
|
||
|
||
return render(request, "includes/easy_logs.html", context)
|
||
|
||
|
||
|
||
from allauth.account.views import SignupView
|
||
from django.contrib.auth.decorators import user_passes_test
|
||
|
||
def is_superuser_check(user):
|
||
return user.is_superuser
|
||
|
||
|
||
@user_passes_test(is_superuser_check)
|
||
def create_staff_user(request):
|
||
if request.method == 'POST':
|
||
|
||
form = StaffUserCreationForm(request.POST)
|
||
print(form)
|
||
if form.is_valid():
|
||
form.save()
|
||
messages.success(
|
||
request,
|
||
f"Staff user {form.cleaned_data['first_name']} {form.cleaned_data['last_name']} "
|
||
f"({form.cleaned_data['email']}) created successfully!"
|
||
)
|
||
return redirect('admin_settings')
|
||
else:
|
||
form = StaffUserCreationForm()
|
||
return render(request, 'user/create_staff.html', {'form': form})
|
||
|
||
|
||
|
||
|
||
|
||
|
||
@user_passes_test(is_superuser_check)
|
||
def admin_settings(request):
|
||
staffs=User.objects.filter(is_superuser=False)
|
||
form = ToggleAccountForm()
|
||
context={
|
||
'staffs':staffs,
|
||
'form':form
|
||
}
|
||
return render(request,'user/admin_settings.html',context)
|
||
|
||
|
||
from django.contrib.auth.forms import SetPasswordForm
|
||
|
||
@user_passes_test(is_superuser_check)
|
||
def set_staff_password(request,pk):
|
||
user=get_object_or_404(User,pk=pk)
|
||
print(request.POST)
|
||
if request.method=='POST':
|
||
form = SetPasswordForm(user, data=request.POST)
|
||
if form.is_valid():
|
||
form.save()
|
||
messages.success(request,f'Password successfully changed')
|
||
return redirect('admin_settings')
|
||
else:
|
||
form=SetPasswordForm(user=user)
|
||
messages.error(request,f'Password does not match please try again.')
|
||
return redirect('admin_settings')
|
||
|
||
else:
|
||
form=SetPasswordForm(user=user)
|
||
return render(request,'user/staff_password_create.html',{'form':form,'user':user})
|
||
|
||
|
||
@user_passes_test(is_superuser_check)
|
||
def account_toggle_status(request,pk):
|
||
user=get_object_or_404(User,pk=pk)
|
||
if request.method=='POST':
|
||
print(user.is_active)
|
||
form=ToggleAccountForm(request.POST)
|
||
if form.is_valid():
|
||
if user.is_active:
|
||
user.is_active=False
|
||
user.save()
|
||
messages.success(request,f'Staff with email: {user.email} deactivated successfully')
|
||
return redirect('admin_settings')
|
||
else:
|
||
user.is_active=True
|
||
user.save()
|
||
messages.success(request,f'Staff with email: {user.email} activated successfully')
|
||
return redirect('admin_settings')
|
||
else:
|
||
messages.error(f'Please correct the error below')
|
||
|
||
|
||
# @login_required
|
||
# def user_detail(requests,pk):
|
||
# user=get_object_or_404(User,pk=pk)
|
||
# return render(requests,'user/profile.html')
|
||
|
||
|
||
@csrf_exempt
|
||
def zoom_webhook_view(request):
|
||
print(request.headers)
|
||
print(settings.ZOOM_WEBHOOK_API_KEY)
|
||
# if api_key != settings.ZOOM_WEBHOOK_API_KEY:
|
||
# return HttpResponse(status=405)
|
||
if request.method == 'POST':
|
||
try:
|
||
payload = json.loads(request.body)
|
||
async_task("recruitment.tasks.handle_zoom_webhook_event", payload)
|
||
return HttpResponse(status=200)
|
||
except Exception:
|
||
return HttpResponse(status=400)
|
||
return HttpResponse(status=405)
|
||
|
||
|
||
# Meeting Comments Views
|
||
@login_required
|
||
def add_meeting_comment(request, slug):
|
||
"""Add a comment to a meeting"""
|
||
meeting = get_object_or_404(ZoomMeeting, slug=slug)
|
||
|
||
if request.method == 'POST':
|
||
form = MeetingCommentForm(request.POST)
|
||
if form.is_valid():
|
||
comment = form.save(commit=False)
|
||
comment.meeting = meeting
|
||
comment.author = request.user
|
||
comment.save()
|
||
messages.success(request, 'Comment added successfully!')
|
||
|
||
# HTMX response - return just the comment section
|
||
if 'HX-Request' in request.headers:
|
||
return render(request, 'includes/comment_list.html', {
|
||
'comments': meeting.comments.all().order_by('-created_at'),
|
||
'meeting': meeting
|
||
})
|
||
|
||
return redirect('meeting_details', slug=slug)
|
||
else:
|
||
form = MeetingCommentForm()
|
||
|
||
context = {
|
||
'form': form,
|
||
'meeting': meeting,
|
||
}
|
||
|
||
# HTMX response - return the comment form
|
||
if 'HX-Request' in request.headers:
|
||
return render(request, 'includes/comment_form.html', context)
|
||
|
||
return redirect('meeting_details', slug=slug)
|
||
|
||
|
||
@login_required
|
||
def edit_meeting_comment(request, slug, comment_id):
|
||
"""Edit a meeting comment"""
|
||
meeting = get_object_or_404(ZoomMeeting, slug=slug)
|
||
comment = get_object_or_404(MeetingComment, id=comment_id, meeting=meeting)
|
||
|
||
# Check if user is the author
|
||
if comment.author != request.user:
|
||
messages.error(request, 'You can only edit your own comments.')
|
||
return redirect('meeting_details', slug=slug)
|
||
|
||
if request.method == 'POST':
|
||
form = MeetingCommentForm(request.POST, instance=comment)
|
||
if form.is_valid():
|
||
form.save()
|
||
messages.success(request, 'Comment updated successfully!')
|
||
|
||
# HTMX response - return just the comment section
|
||
if 'HX-Request' in request.headers:
|
||
return render(request, 'includes/comment_list.html', {
|
||
'comments': meeting.comments.all().order_by('-created_at'),
|
||
'meeting': meeting
|
||
})
|
||
|
||
return redirect('meeting_details', slug=slug)
|
||
else:
|
||
form = MeetingCommentForm(instance=comment)
|
||
print("hi")
|
||
context = {
|
||
'form': form,
|
||
'meeting': meeting,
|
||
'comment':comment
|
||
}
|
||
return render(request, 'includes/edit_comment_form.html', context)
|
||
|
||
@login_required
|
||
def delete_meeting_comment(request, slug, comment_id):
|
||
"""Delete a meeting comment"""
|
||
meeting = get_object_or_404(ZoomMeeting, slug=slug)
|
||
comment = get_object_or_404(MeetingComment, id=comment_id, meeting=meeting)
|
||
|
||
# Check if user is the author
|
||
if comment.author != request.user and not request.user.is_staff:
|
||
messages.error(request, 'You can only delete your own comments.')
|
||
return redirect('meeting_details', slug=slug)
|
||
|
||
if request.method == 'POST':
|
||
comment.delete()
|
||
messages.success(request, 'Comment deleted successfully!')
|
||
|
||
# HTMX response - return just the comment section
|
||
if 'HX-Request' in request.headers:
|
||
return render(request, 'includes/comment_list.html', {
|
||
'comments': meeting.comments.all().order_by('-created_at'),
|
||
'meeting': meeting
|
||
})
|
||
|
||
return redirect('meeting_details', slug=slug)
|
||
|
||
# HTMX response - return the delete confirmation modal
|
||
if 'HX-Request' in request.headers:
|
||
return render(request, 'includes/delete_comment_form.html', {
|
||
'meeting': meeting,
|
||
'comment': comment,
|
||
'delete_url': reverse('delete_meeting_comment', kwargs={'slug': slug, 'comment_id': comment_id})
|
||
})
|
||
|
||
return redirect('meeting_details', slug=slug)
|
||
|
||
|
||
@login_required
|
||
def set_meeting_candidate(request,slug):
|
||
meeting = get_object_or_404(ZoomMeeting, slug=slug)
|
||
if request.method == 'POST' and 'HX-Request' not in request.headers:
|
||
form = InterviewForm(request.POST)
|
||
if form.is_valid():
|
||
candidate = form.save(commit=False)
|
||
candidate.zoom_meeting = meeting
|
||
candidate.interview_date = meeting.start_time.date()
|
||
candidate.interview_time = meeting.start_time.time()
|
||
candidate.save()
|
||
messages.success(request, 'Candidate added successfully!')
|
||
return redirect('list_meetings')
|
||
job = request.GET.get("job")
|
||
form = InterviewForm()
|
||
|
||
if job:
|
||
form.fields['candidate'].queryset = Candidate.objects.filter(job=job)
|
||
|
||
else:
|
||
form.fields['candidate'].queryset = Candidate.objects.none()
|
||
form.fields['job'].widget.attrs.update({
|
||
'hx-get': reverse('set_meeting_candidate', kwargs={'slug': slug}),
|
||
'hx-target': '#div_id_candidate',
|
||
'hx-select': '#div_id_candidate',
|
||
'hx-swap': 'outerHTML'
|
||
})
|
||
context = {
|
||
"form": form,
|
||
"meeting": meeting
|
||
}
|
||
return render(request, 'meetings/set_candidate_form.html', context)
|