1795 lines
59 KiB
Python
1795 lines
59 KiB
Python
from django.db import models
|
|
from django.urls import reverse
|
|
from typing import List,Dict,Any
|
|
from django.utils import timezone
|
|
from django.db.models import FloatField,CharField,IntegerField
|
|
from django.db.models.functions import Cast,Coalesce
|
|
from django.db.models import F
|
|
from django.contrib.auth.models import User
|
|
from django.core.validators import URLValidator
|
|
from django_countries.fields import CountryField
|
|
from django.core.exceptions import ValidationError
|
|
from django_ckeditor_5.fields import CKEditor5Field
|
|
from django.utils.html import strip_tags
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django_extensions.db.fields import RandomCharField
|
|
from .validators import validate_hash_tags, validate_image_size
|
|
|
|
|
|
class Base(models.Model):
|
|
created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Created at"))
|
|
updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated at"))
|
|
slug = RandomCharField(
|
|
length=8, unique=True, editable=False, verbose_name=_("Slug")
|
|
)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
class Profile(models.Model):
|
|
profile_image = models.ImageField(null=True, blank=True, upload_to="profile_pic/",validators=[validate_image_size])
|
|
designation = models.CharField(max_length=100, blank=True,null=True)
|
|
phone=models.CharField(blank=True,null=True,verbose_name=_("Phone Number"),max_length=12)
|
|
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="profile")
|
|
|
|
def __str__(self):
|
|
return f"image for user {self.user}"
|
|
|
|
|
|
class JobPosting(Base):
|
|
# Basic Job Information
|
|
|
|
JOB_TYPES = [
|
|
("FULL_TIME", "Full-time"),
|
|
("PART_TIME", "Part-time"),
|
|
("CONTRACT", "Contract"),
|
|
("INTERNSHIP", "Internship"),
|
|
("FACULTY", "Faculty"),
|
|
("TEMPORARY", "Temporary"),
|
|
]
|
|
|
|
WORKPLACE_TYPES = [
|
|
("ON_SITE", "On-site"),
|
|
("REMOTE", "Remote"),
|
|
("HYBRID", "Hybrid"),
|
|
]
|
|
|
|
users=models.ManyToManyField(
|
|
User,
|
|
blank=True,related_name="jobs_assigned",
|
|
verbose_name=_("Internal Participant"),
|
|
help_text=_("Internal staff involved in the recruitment process for this job"),
|
|
)
|
|
|
|
participants=models.ManyToManyField('Participants',
|
|
blank=True,related_name="jobs_participating",
|
|
verbose_name=_("External Participant"),
|
|
help_text=_("External participants involved in the recruitment process for this job"),
|
|
)
|
|
|
|
# Core Fields
|
|
title = models.CharField(max_length=200)
|
|
department = models.CharField(max_length=100, blank=True)
|
|
job_type = models.CharField(max_length=20, choices=JOB_TYPES, default="FULL_TIME")
|
|
workplace_type = models.CharField(
|
|
max_length=20, choices=WORKPLACE_TYPES, default="ON_SITE"
|
|
)
|
|
|
|
# Location
|
|
location_city = models.CharField(max_length=100, blank=True)
|
|
location_state = models.CharField(max_length=100, blank=True)
|
|
location_country = models.CharField(max_length=100, default="Saudia Arabia")
|
|
|
|
# Job Details
|
|
description = CKEditor5Field(
|
|
'Description',
|
|
config_name='extends' # Matches the config name you defined in settings.py
|
|
)
|
|
|
|
qualifications = CKEditor5Field(blank=True, null=True,
|
|
config_name='extends'
|
|
)
|
|
salary_range = models.CharField(
|
|
max_length=200, blank=True, help_text="e.g., $60,000 - $80,000"
|
|
)
|
|
benefits = CKEditor5Field(blank=True, null=True, config_name='extends')
|
|
|
|
# Application Information ---job detail apply link for the candidates
|
|
application_url = models.URLField(
|
|
validators=[URLValidator()],
|
|
help_text="URL where candidates apply",
|
|
null=True,
|
|
blank=True,
|
|
)
|
|
|
|
application_deadline = models.DateField(db_index=True) # Added index
|
|
application_instructions = CKEditor5Field(
|
|
blank=True, null=True, config_name='extends'
|
|
)
|
|
|
|
# Internal Tracking
|
|
internal_job_id = models.CharField(max_length=50, editable=False)
|
|
|
|
created_by = models.CharField(
|
|
max_length=100, blank=True, help_text="Name of person who created this job"
|
|
)
|
|
|
|
# Status Fields
|
|
STATUS_CHOICES = [
|
|
("DRAFT", "Draft"),
|
|
("ACTIVE", "Active"),
|
|
("CLOSED", "Closed"),
|
|
("CANCELLED", "Cancelled"),
|
|
("ARCHIVED", "Archived"),
|
|
]
|
|
status = models.CharField(
|
|
db_index=True, max_length=20, choices=STATUS_CHOICES, default="DRAFT" # Added index
|
|
)
|
|
|
|
# hashtags for social media
|
|
hash_tags = models.CharField(
|
|
max_length=200,
|
|
blank=True,
|
|
help_text="Comma-separated hashtags for linkedin post like #hiring,#jobopening",
|
|
validators=[validate_hash_tags],
|
|
)
|
|
|
|
# LinkedIn Integration Fields
|
|
linkedin_post_id = models.CharField(
|
|
max_length=200, blank=True, help_text="LinkedIn post ID after posting"
|
|
)
|
|
linkedin_post_url = models.URLField(
|
|
blank=True, help_text="Direct URL to LinkedIn post"
|
|
)
|
|
posted_to_linkedin = models.BooleanField(default=False)
|
|
linkedin_post_status = models.CharField(
|
|
max_length=50, blank=True, help_text="Status of LinkedIn posting"
|
|
)
|
|
linkedin_posted_at = models.DateTimeField(null=True, blank=True)
|
|
linkedin_post_formated_data=models.TextField(null=True,blank=True)
|
|
|
|
published_at = models.DateTimeField(db_index=True, null=True, blank=True) # Added index
|
|
# University Specific Fields
|
|
position_number = models.CharField(
|
|
max_length=50, blank=True, help_text="University position number"
|
|
)
|
|
reporting_to = models.CharField(
|
|
max_length=100, blank=True, help_text="Who this position reports to"
|
|
)
|
|
|
|
open_positions = models.PositiveIntegerField(
|
|
default=1, help_text="Number of open positions for this job"
|
|
)
|
|
|
|
source = models.ForeignKey(
|
|
"Source",
|
|
on_delete=models.SET_NULL, # Recommended: If a source is deleted, job's source is set to NULL
|
|
related_name="job_postings",
|
|
null=True,
|
|
blank=True,
|
|
help_text="The system or channel from which this job posting originated or was first published.",
|
|
db_index=True # Explicitly index ForeignKey
|
|
)
|
|
max_applications = models.PositiveIntegerField(
|
|
default=1000, help_text="Maximum number of applications allowed", null=True, blank=True
|
|
)
|
|
hiring_agency = models.ManyToManyField(
|
|
"HiringAgency",
|
|
blank=True,
|
|
related_name="jobs",
|
|
verbose_name=_("Hiring Agency"),
|
|
help_text=_(
|
|
"External agency responsible for sourcing candidates for this role"
|
|
),
|
|
)
|
|
cancel_reason = models.TextField(
|
|
blank=True,
|
|
help_text=_("Reason for canceling the job posting"),
|
|
verbose_name=_("Cancel Reason"),
|
|
)
|
|
cancelled_by = models.CharField(
|
|
max_length=100,
|
|
blank=True,
|
|
help_text=_("Name of person who cancelled this job"),
|
|
verbose_name=_("Cancelled By"),
|
|
)
|
|
cancelled_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
verbose_name = "Job Posting"
|
|
verbose_name_plural = "Job Postings"
|
|
indexes = [
|
|
models.Index(fields=['status', 'created_at', 'title']),
|
|
models.Index(fields=['slug']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.title} - {self.internal_job_id}-{self.get_status_display()}"
|
|
|
|
def get_source(self):
|
|
return self.source.name if self.source else "System"
|
|
|
|
def save(self, *args, **kwargs):
|
|
from django.db import transaction
|
|
|
|
# Generate unique internal job ID if not exists
|
|
with transaction.atomic():
|
|
if not self.internal_job_id:
|
|
prefix = "KAAUH"
|
|
year = timezone.now().year
|
|
# Get next sequential number
|
|
last_job = (
|
|
JobPosting.objects.select_for_update().filter(
|
|
internal_job_id__startswith=f"{prefix}-{year}-"
|
|
)
|
|
.order_by("internal_job_id")
|
|
.last()
|
|
)
|
|
|
|
if last_job:
|
|
last_num = int(last_job.internal_job_id.split("-")[-1])
|
|
next_num = last_num + 1
|
|
else:
|
|
next_num = 1
|
|
|
|
self.internal_job_id = f"{prefix}-{year}-{next_num:06d}"
|
|
|
|
super().save(*args, **kwargs)
|
|
|
|
def get_location_display(self):
|
|
"""Return formatted location string"""
|
|
parts = []
|
|
if self.location_city:
|
|
parts.append(self.location_city)
|
|
if self.location_state:
|
|
parts.append(self.location_state)
|
|
if self.location_country:
|
|
parts.append(self.location_country)
|
|
return ", ".join(parts) if parts else "Not specified"
|
|
|
|
@property
|
|
def is_expired(self):
|
|
"""Check if application deadline has passed"""
|
|
if self.application_deadline:
|
|
return self.application_deadline < timezone.now().date()
|
|
return False
|
|
|
|
def publish(self):
|
|
self.status = "PUBLISHED"
|
|
self.published_at = timezone.now()
|
|
self.application_url = reverse(
|
|
"form_wizard", kwargs={"slug": self.form_template.slug}
|
|
)
|
|
self.save()
|
|
|
|
def _check_content(self, field_value):
|
|
"""Helper to check if a field contains meaningful content."""
|
|
if not field_value:
|
|
return False
|
|
|
|
# 1. Replace the common HTML non-breaking space entity with a standard space.
|
|
content = field_value.replace(' ', ' ')
|
|
|
|
# 2. Remove all HTML tags (leaving only text and remaining spaces).
|
|
stripped = strip_tags(content)
|
|
|
|
# 3. Use .strip() to remove ALL leading/trailing whitespace, including the ones from step 1.
|
|
final_content = stripped.strip()
|
|
|
|
# Return True if any content remains after stripping tags and spaces.
|
|
return bool(final_content)
|
|
|
|
@property
|
|
def has_description_content(self):
|
|
"""Returns True if the description field has meaningful content."""
|
|
return self._check_content(self.description)
|
|
|
|
@property
|
|
def has_qualifications_content(self):
|
|
"""Returns True if the qualifications field has meaningful content."""
|
|
return self._check_content(self.qualifications)
|
|
|
|
# Add similar properties for benefits and application_instructions
|
|
@property
|
|
def has_benefits_content(self):
|
|
return self._check_content(self.benefits)
|
|
|
|
@property
|
|
def has_application_instructions_content(self):
|
|
return self._check_content(self.application_instructions)
|
|
|
|
@property
|
|
def current_applications_count(self):
|
|
"""Returns the current number of candidates associated with this job."""
|
|
return self.candidates.count()
|
|
|
|
@property
|
|
def is_application_limit_reached(self):
|
|
"""Checks if the current application count meets or exceeds the max limit."""
|
|
if self.max_applications == 0:
|
|
return True
|
|
|
|
return self.current_applications_count >= self.max_applications
|
|
|
|
@property
|
|
def all_candidates(self):
|
|
return self.candidates.annotate(
|
|
sortable_score=Coalesce(Cast('ai_analysis_data__analysis_data__match_score', output_field=IntegerField()),
|
|
0)).order_by('-sortable_score')
|
|
|
|
@property
|
|
def screening_candidates(self):
|
|
return self.all_candidates.filter(stage="Applied")
|
|
|
|
@property
|
|
def exam_candidates(self):
|
|
return self.all_candidates.filter(stage="Exam")
|
|
|
|
@property
|
|
def interview_candidates(self):
|
|
return self.all_candidates.filter(stage="Interview")
|
|
|
|
@property
|
|
def offer_candidates(self):
|
|
return self.all_candidates.filter(stage="Offer")
|
|
@property
|
|
def accepted_candidates(self):
|
|
return self.all_candidates.filter(offer_status="Accepted")
|
|
@property
|
|
def hired_candidates(self):
|
|
return self.all_candidates.filter(stage="Hired")
|
|
|
|
# counts
|
|
@property
|
|
def all_candidates_count(self):
|
|
return self.candidates.annotate(
|
|
sortable_score=Cast('ai_analysis_data__match_score', output_field=CharField())).order_by(
|
|
'-sortable_score').count() or 0
|
|
|
|
@property
|
|
def screening_candidates_count(self):
|
|
return self.all_candidates.filter(stage="Applied").count() or 0
|
|
|
|
@property
|
|
def exam_candidates_count(self):
|
|
return self.all_candidates.filter(stage="Exam").count() or 0
|
|
|
|
@property
|
|
def interview_candidates_count(self):
|
|
return self.all_candidates.filter(stage="Interview").count() or 0
|
|
|
|
@property
|
|
def offer_candidates_count(self):
|
|
return self.all_candidates.filter(stage="Offer").count() or 0
|
|
|
|
@property
|
|
def hired_candidates_count(self):
|
|
return self.all_candidates.filter(stage="Hired").count() or 0
|
|
|
|
@property
|
|
def vacancy_fill_rate(self):
|
|
total_positions = self.open_positions
|
|
|
|
no_of_positions_filled = self.candidates.filter(stage__in=['HIRED']).count()
|
|
|
|
if total_positions > 0:
|
|
vacancy_fill_rate = no_of_positions_filled / total_positions
|
|
else:
|
|
vacancy_fill_rate = 0.0
|
|
|
|
return vacancy_fill_rate
|
|
|
|
|
|
|
|
class JobPostingImage(models.Model):
|
|
job=models.OneToOneField('JobPosting',on_delete=models.CASCADE,related_name='post_images')
|
|
post_image = models.ImageField(upload_to='post/',validators=[validate_image_size])
|
|
|
|
|
|
class Candidate(Base):
|
|
class Stage(models.TextChoices):
|
|
APPLIED = "Applied", _("Applied")
|
|
EXAM = "Exam", _("Exam")
|
|
INTERVIEW = "Interview", _("Interview")
|
|
OFFER = "Offer", _("Offer")
|
|
HIRED = "Hired", _("Hired")
|
|
|
|
class ExamStatus(models.TextChoices):
|
|
PASSED = "Passed", _("Passed")
|
|
FAILED = "Failed", _("Failed")
|
|
|
|
class Status(models.TextChoices):
|
|
ACCEPTED = "Accepted", _("Accepted")
|
|
REJECTED = "Rejected", _("Rejected")
|
|
|
|
class ApplicantType(models.TextChoices):
|
|
APPLICANT = "Applicant", _("Applicant")
|
|
CANDIDATE = "Candidate", _("Candidate")
|
|
|
|
# Stage transition validation constants
|
|
STAGE_SEQUENCE = {
|
|
"Applied": ["Exam", "Interview", "Offer"],
|
|
"Exam": ["Interview", "Offer"],
|
|
"Interview": ["Offer"],
|
|
"Offer": [], # Final stage - no further transitions
|
|
}
|
|
|
|
job = models.ForeignKey(
|
|
JobPosting,
|
|
on_delete=models.CASCADE,
|
|
related_name="candidates",
|
|
verbose_name=_("Job"),
|
|
)
|
|
first_name = models.CharField(max_length=255, verbose_name=_("First Name"))
|
|
last_name = models.CharField(max_length=255, verbose_name=_("Last Name"))
|
|
email = models.EmailField(db_index=True, verbose_name=_("Email")) # Added index
|
|
phone = models.CharField(max_length=20, verbose_name=_("Phone"))
|
|
address = models.TextField(max_length=200, verbose_name=_("Address"))
|
|
resume = models.FileField(upload_to="resumes/", verbose_name=_("Resume"))
|
|
is_resume_parsed = models.BooleanField(
|
|
default=False, verbose_name=_("Resume Parsed")
|
|
)
|
|
is_potential_candidate = models.BooleanField(
|
|
default=False, verbose_name=_("Potential Candidate")
|
|
)
|
|
parsed_summary = models.TextField(blank=True, verbose_name=_("Parsed Summary"))
|
|
applied = models.BooleanField(default=False, verbose_name=_("Applied"))
|
|
stage = models.CharField(
|
|
db_index=True, max_length=100, # Added index
|
|
default="Applied",
|
|
choices=Stage.choices,
|
|
verbose_name=_("Stage"),
|
|
)
|
|
applicant_status = models.CharField(
|
|
choices=ApplicantType.choices,
|
|
default="Applicant",
|
|
max_length=100,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Applicant Status"),
|
|
)
|
|
exam_date = models.DateTimeField(null=True, blank=True, verbose_name=_("Exam Date"))
|
|
exam_status = models.CharField(
|
|
choices=ExamStatus.choices,
|
|
max_length=100,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Exam Status"),
|
|
)
|
|
interview_date = models.DateTimeField(
|
|
null=True, blank=True, verbose_name=_("Interview Date")
|
|
)
|
|
interview_status = models.CharField(
|
|
choices=ExamStatus.choices,
|
|
max_length=100,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Interview Status"),
|
|
)
|
|
offer_date = models.DateField(null=True, blank=True, verbose_name=_("Offer Date"))
|
|
offer_status = models.CharField(
|
|
choices=Status.choices,
|
|
max_length=100,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Offer Status"),
|
|
)
|
|
hired_date = models.DateField(null=True, blank=True, verbose_name=_("Hired Date"))
|
|
join_date = models.DateField(null=True, blank=True, verbose_name=_("Join Date"))
|
|
ai_analysis_data = models.JSONField(
|
|
verbose_name="AI Analysis Data",
|
|
default=dict,
|
|
help_text="Full JSON output from the resume scoring model."
|
|
)# {'resume_data': {}, 'analysis_data': {}}
|
|
|
|
retry = models.SmallIntegerField(verbose_name="Resume Parsing Retry",default=3)
|
|
hiring_source = models.CharField(
|
|
max_length=255,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Hiring Source"),
|
|
choices=[("Public", "Public"), ("Internal", "Internal"), ("Agency", "Agency")],
|
|
default="Public",
|
|
)
|
|
hiring_agency = models.ForeignKey(
|
|
"HiringAgency",
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name="candidates",
|
|
verbose_name=_("Hiring Agency"),
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Candidate")
|
|
verbose_name_plural = _("Candidates")
|
|
indexes = [
|
|
models.Index(fields=['stage']),
|
|
models.Index(fields=['created_at']),
|
|
]
|
|
def set_field(self, key: str, value: Any):
|
|
"""
|
|
Generic method to set any single key-value pair and save.
|
|
"""
|
|
self.ai_analysis_data[key] = value
|
|
# self.save(update_fields=['ai_analysis_data'])
|
|
|
|
# ====================================================================
|
|
# ✨ PROPERTIES (GETTERS)
|
|
# ====================================================================
|
|
@property
|
|
def resume_data(self):
|
|
return self.ai_analysis_data.get('resume_data', {})
|
|
@property
|
|
def analysis_data(self):
|
|
return self.ai_analysis_data.get('analysis_data', {})
|
|
@property
|
|
def match_score(self) -> int:
|
|
"""1. A score from 0 to 100 representing how well the candidate fits the role."""
|
|
return self.analysis_data.get('match_score', 0)
|
|
|
|
@property
|
|
def years_of_experience(self) -> float:
|
|
"""4. The total number of years of professional experience as a numerical value."""
|
|
return self.analysis_data.get('years_of_experience', 0.0)
|
|
|
|
@property
|
|
def soft_skills_score(self) -> int:
|
|
"""15. A score (0-100) for inferred non-technical skills."""
|
|
return self.analysis_data.get('soft_skills_score', 0)
|
|
|
|
@property
|
|
def industry_match_score(self) -> int:
|
|
"""16. A score (0-100) for the relevance of the candidate's industry experience."""
|
|
# Renamed to clarify: experience_industry_match
|
|
return self.analysis_data.get('experience_industry_match', 0)
|
|
|
|
# --- Properties for Funnel & Screening Efficiency ---
|
|
|
|
@property
|
|
def min_requirements_met(self) -> bool:
|
|
"""14. Boolean (true/false) indicating if all non-negotiable minimum requirements are met."""
|
|
return self.analysis_data.get('min_req_met_bool', False)
|
|
|
|
@property
|
|
def screening_stage_rating(self) -> str:
|
|
"""13. A standardized rating (e.g., "A - Highly Qualified", "B - Qualified")."""
|
|
return self.analysis_data.get('screening_stage_rating', 'N/A')
|
|
|
|
@property
|
|
def top_3_keywords(self) -> List[str]:
|
|
"""10. A list of the three most dominant and relevant technical skills or technologies."""
|
|
return self.analysis_data.get('top_3_keywords', [])
|
|
|
|
@property
|
|
def most_recent_job_title(self) -> str:
|
|
"""8. The candidate's most recent or current professional job title."""
|
|
return self.analysis_data.get('most_recent_job_title', 'N/A')
|
|
|
|
# --- Properties for Structured Detail ---
|
|
|
|
@property
|
|
def criteria_checklist(self) -> Dict[str, str]:
|
|
"""5 & 6. An object rating the candidate's match for each specific criterion."""
|
|
return self.analysis_data.get('criteria_checklist', {})
|
|
|
|
@property
|
|
def professional_category(self) -> str:
|
|
"""7. The most fitting professional field or category for the individual."""
|
|
return self.analysis_data.get('category', 'N/A')
|
|
|
|
@property
|
|
def language_fluency(self) -> List[Dict[str, str]]:
|
|
"""12. A list of languages and their fluency levels mentioned."""
|
|
return self.analysis_data.get('language_fluency', [])
|
|
|
|
# --- Properties for Summaries and Narrative ---
|
|
|
|
@property
|
|
def strengths(self) -> str:
|
|
"""2. A brief summary of why the candidate is a strong fit."""
|
|
return self.analysis_data.get('strengths', '')
|
|
|
|
@property
|
|
def weaknesses(self) -> str:
|
|
"""3. A brief summary of where the candidate falls short or what criteria are missing."""
|
|
return self.analysis_data.get('weaknesses', '')
|
|
|
|
@property
|
|
def job_fit_narrative(self) -> str:
|
|
"""11. A single, concise sentence summarizing the core fit."""
|
|
return self.analysis_data.get('job_fit_narrative', '')
|
|
|
|
@property
|
|
def recommendation(self) -> str:
|
|
"""9. Provide a detailed final recommendation for the candidate."""
|
|
# Using a more descriptive name to avoid conflict with potential built-in methods
|
|
return self.analysis_data.get('recommendation', '')
|
|
|
|
@property
|
|
def name(self):
|
|
return f"{self.first_name} {self.last_name}"
|
|
|
|
@property
|
|
def full_name(self):
|
|
return self.name
|
|
|
|
@property
|
|
def get_file_size(self):
|
|
if self.resume:
|
|
return self.resume.size
|
|
return 0
|
|
|
|
|
|
def save(self, *args, **kwargs):
|
|
"""Override save to ensure validation is called"""
|
|
self.clean() # Call validation before saving
|
|
super().save(*args, **kwargs)
|
|
|
|
def get_available_stages(self):
|
|
"""Get list of stages this candidate can transition to"""
|
|
if not self.pk: # New record
|
|
return ["Applied"]
|
|
|
|
old_stage = self.__class__.objects.get(pk=self.pk).stage
|
|
return self.STAGE_SEQUENCE.get(old_stage, [])
|
|
|
|
@property
|
|
def submission(self):
|
|
return FormSubmission.objects.filter(template__job=self.job).first()
|
|
@property
|
|
def responses(self):
|
|
if self.submission:
|
|
return self.submission.responses.all()
|
|
return []
|
|
def __str__(self):
|
|
return self.full_name
|
|
|
|
@property
|
|
def get_meetings(self):
|
|
return self.scheduled_interviews.all()
|
|
@property
|
|
def get_latest_meeting(self):
|
|
schedule = self.scheduled_interviews.order_by('-created_at').first()
|
|
if schedule:
|
|
return schedule.zoom_meeting
|
|
return None
|
|
|
|
@property
|
|
def has_future_meeting(self):
|
|
"""
|
|
Checks if the candidate has any scheduled interviews for a future date/time.
|
|
"""
|
|
# Ensure timezone.now() is used for comparison
|
|
now = timezone.now()
|
|
# Check if any related ScheduledInterview has a future interview_date and interview_time
|
|
# We need to combine date and time for a proper datetime comparison if they are separate fields
|
|
future_meetings = self.scheduled_interviews.filter(
|
|
interview_date__gt=now.date()
|
|
).filter(
|
|
interview_time__gte=now.time()
|
|
).exists()
|
|
|
|
# Also check for interviews happening later today
|
|
today_future_meetings = self.scheduled_interviews.filter(
|
|
interview_date=now.date(),
|
|
interview_time__gte=now.time()
|
|
).exists()
|
|
|
|
return future_meetings or today_future_meetings
|
|
|
|
@property
|
|
def check_and_retry_ai_scoring(self):
|
|
"""
|
|
Triggers an immediate save ONLY if:
|
|
1. The resume hasn't been parsed yet.
|
|
2. At least 5 minutes have passed since the last attempt.
|
|
Returns True if a save was performed, False otherwise.
|
|
"""
|
|
from datetime import timedelta
|
|
|
|
min_delay = timedelta(minutes=5)
|
|
|
|
time_since_last_attempt = timezone.now() - self.created_at
|
|
|
|
if not self.is_resume_parsed and time_since_last_attempt >= min_delay:
|
|
|
|
# 1. Update the retry timestamp
|
|
self.last_retry_attempt = timezone.now()
|
|
|
|
|
|
self.save()
|
|
|
|
return True
|
|
|
|
return False
|
|
|
|
# @property
|
|
# def time_to_hire(self):
|
|
# time_to_hire=self.hired_date-self.created_at
|
|
# return time_to_hire
|
|
|
|
|
|
|
|
class TrainingMaterial(Base):
|
|
title = models.CharField(max_length=255, verbose_name=_("Title"))
|
|
content = CKEditor5Field(blank=True, verbose_name=_("Content"),config_name='extends')
|
|
video_link = models.URLField(blank=True, verbose_name=_("Video Link"))
|
|
file = models.FileField(
|
|
upload_to="training_materials/", blank=True, verbose_name=_("File")
|
|
)
|
|
created_by = models.ForeignKey(
|
|
User, on_delete=models.SET_NULL, null=True, verbose_name=_("Created by")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Training Material")
|
|
verbose_name_plural = _("Training Materials")
|
|
|
|
def __str__(self):
|
|
return self.title
|
|
|
|
|
|
class ZoomMeeting(Base):
|
|
class MeetingStatus(models.TextChoices):
|
|
WAITING = "waiting", _("Waiting")
|
|
STARTED = "started", _("Started")
|
|
ENDED = "ended", _("Ended")
|
|
CANCELLED = "cancelled",_("Cancelled")
|
|
# Basic meeting details
|
|
topic = models.CharField(max_length=255, verbose_name=_("Topic"))
|
|
meeting_id = models.CharField(
|
|
db_index=True, max_length=20, unique=True, verbose_name=_("Meeting ID") # Added index
|
|
) # Unique identifier for the meeting
|
|
start_time = models.DateTimeField(db_index=True, verbose_name=_("Start Time")) # Added index
|
|
duration = models.PositiveIntegerField(
|
|
verbose_name=_("Duration")
|
|
) # Duration in minutes
|
|
timezone = models.CharField(max_length=50, verbose_name=_("Timezone"))
|
|
join_url = models.URLField(
|
|
verbose_name=_("Join URL")
|
|
) # URL for participants to join
|
|
participant_video = models.BooleanField(
|
|
default=True, verbose_name=_("Participant Video")
|
|
)
|
|
password = models.CharField(
|
|
max_length=20, blank=True, null=True, verbose_name=_("Password")
|
|
)
|
|
join_before_host = models.BooleanField(
|
|
default=False, verbose_name=_("Join Before Host")
|
|
)
|
|
mute_upon_entry = models.BooleanField(
|
|
default=False, verbose_name=_("Mute Upon Entry")
|
|
)
|
|
waiting_room = models.BooleanField(default=False, verbose_name=_("Waiting Room"))
|
|
|
|
zoom_gateway_response = models.JSONField(
|
|
blank=True, null=True, verbose_name=_("Zoom Gateway Response")
|
|
)
|
|
status = models.CharField(
|
|
db_index=True, max_length=20, # Added index
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Status"),
|
|
default=MeetingStatus.WAITING,
|
|
)
|
|
# Timestamps
|
|
|
|
def __str__(self):
|
|
return self.topic
|
|
|
|
@property
|
|
def get_job(self):
|
|
try:
|
|
job=self.interview.job.first()
|
|
return job
|
|
except:
|
|
return None
|
|
@property
|
|
def get_candidate(self):
|
|
try:
|
|
candidate=self.interview.candidate.first()
|
|
return candidate
|
|
except:
|
|
return None
|
|
|
|
@property
|
|
def get_external_participants(self):
|
|
try:
|
|
interview=self.interview.first()
|
|
if interview:
|
|
return interview.job.participants.all()
|
|
return None
|
|
except:
|
|
return None
|
|
@property
|
|
def get_users_participants(self):
|
|
try:
|
|
interview=self.interview.first()
|
|
if interview:
|
|
return interview.job.users.all()
|
|
return None
|
|
except:
|
|
return None
|
|
|
|
|
|
|
|
class MeetingComment(Base):
|
|
"""
|
|
Model for storing meeting comments/notes
|
|
"""
|
|
meeting = models.ForeignKey(
|
|
ZoomMeeting,
|
|
on_delete=models.CASCADE,
|
|
related_name="comments",
|
|
verbose_name=_("Meeting")
|
|
)
|
|
author = models.ForeignKey(
|
|
User,
|
|
on_delete=models.CASCADE,
|
|
related_name="meeting_comments",
|
|
verbose_name=_("Author")
|
|
)
|
|
content = CKEditor5Field(
|
|
verbose_name=_("Content"),
|
|
config_name='extends'
|
|
)
|
|
# Inherited from Base: created_at, updated_at, slug
|
|
|
|
class Meta:
|
|
verbose_name = _("Meeting Comment")
|
|
verbose_name_plural = _("Meeting Comments")
|
|
ordering = ['-created_at']
|
|
|
|
def __str__(self):
|
|
return f"Comment by {self.author.get_username()} on {self.meeting.topic}"
|
|
|
|
|
|
class FormTemplate(Base):
|
|
"""
|
|
Represents a complete form template with multiple stages
|
|
"""
|
|
|
|
job = models.OneToOneField(
|
|
JobPosting, on_delete=models.CASCADE, related_name="form_template", db_index=True
|
|
)
|
|
name = models.CharField(max_length=200, help_text="Name of the form template")
|
|
description = models.TextField(
|
|
blank=True, help_text="Description of the form template"
|
|
)
|
|
created_by = models.ForeignKey(
|
|
User, on_delete=models.CASCADE, related_name="form_templates",null=True,blank=True, db_index=True
|
|
)
|
|
is_active = models.BooleanField(
|
|
default=False, help_text="Whether this template is active"
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
verbose_name = "Form Template"
|
|
verbose_name_plural = "Form Templates"
|
|
indexes = [
|
|
models.Index(fields=['created_at']),
|
|
models.Index(fields=['is_active']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
def get_stage_count(self):
|
|
return self.stages.count()
|
|
|
|
def get_field_count(self):
|
|
return sum(stage.fields.count() for stage in self.stages.all())
|
|
|
|
|
|
class FormStage(Base):
|
|
"""
|
|
Represents a stage/section within a form template
|
|
"""
|
|
|
|
template = models.ForeignKey(
|
|
FormTemplate, on_delete=models.CASCADE, related_name="stages", db_index=True
|
|
)
|
|
name = models.CharField(max_length=200, help_text="Name of the stage")
|
|
order = models.PositiveIntegerField(
|
|
default=0, help_text="Order of the stage in the form"
|
|
)
|
|
is_predefined = models.BooleanField(
|
|
default=False, help_text="Whether this is a default resume stage"
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ["order"]
|
|
verbose_name = "Form Stage"
|
|
verbose_name_plural = "Form Stages"
|
|
|
|
def __str__(self):
|
|
return f"{self.template.name} - {self.name}"
|
|
|
|
def clean(self):
|
|
if self.order < 0:
|
|
raise ValidationError("Order must be a positive integer")
|
|
|
|
|
|
class FormField(Base):
|
|
"""
|
|
Represents a single field within a form stage
|
|
"""
|
|
|
|
FIELD_TYPES = [
|
|
("text", "Text Input"),
|
|
("email", "Email"),
|
|
("phone", "Phone"),
|
|
("textarea", "Text Area"),
|
|
("file", "File Upload"),
|
|
("date", "Date Picker"),
|
|
("select", "Dropdown"),
|
|
("radio", "Radio Buttons"),
|
|
("checkbox", "Checkboxes"),
|
|
]
|
|
|
|
stage = models.ForeignKey(
|
|
FormStage, on_delete=models.CASCADE, related_name="fields", db_index=True
|
|
)
|
|
label = models.CharField(max_length=200, help_text="Label for the field")
|
|
field_type = models.CharField(
|
|
max_length=20, choices=FIELD_TYPES, help_text="Type of the field"
|
|
)
|
|
placeholder = models.CharField(
|
|
max_length=200, blank=True, help_text="Placeholder text"
|
|
)
|
|
required = models.BooleanField(
|
|
default=False, help_text="Whether the field is required"
|
|
)
|
|
order = models.PositiveIntegerField(
|
|
default=0, help_text="Order of the field in the stage"
|
|
)
|
|
is_predefined = models.BooleanField(
|
|
default=False, help_text="Whether this is a default field"
|
|
)
|
|
|
|
# For selection fields (select, radio, checkbox)
|
|
options = models.JSONField(
|
|
default=list,
|
|
blank=True,
|
|
help_text="Options for selection fields (stored as JSON array)",
|
|
)
|
|
|
|
# For file upload fields
|
|
file_types = models.CharField(
|
|
max_length=200,
|
|
blank=True,
|
|
help_text="Allowed file types (comma-separated, e.g., '.pdf,.doc,.docx')",
|
|
)
|
|
max_file_size = models.PositiveIntegerField(
|
|
default=5, help_text="Maximum file size in MB (default: 5MB)"
|
|
)
|
|
multiple_files = models.BooleanField(
|
|
default=False, help_text="Allow multiple files to be uploaded"
|
|
)
|
|
max_files = models.PositiveIntegerField(
|
|
default=1,
|
|
help_text="Maximum number of files allowed (when multiple_files is True)",
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ["order"]
|
|
verbose_name = "Form Field"
|
|
verbose_name_plural = "Form Fields"
|
|
|
|
def clean(self):
|
|
# Validate options for selection fields
|
|
if self.field_type in ["select", "radio", "checkbox"]:
|
|
if not isinstance(self.options, list):
|
|
raise ValidationError("Options must be a list for selection fields")
|
|
else:
|
|
# Clear options for non-selection fields
|
|
if self.options:
|
|
self.options = []
|
|
|
|
# Validate file settings for file fields
|
|
if self.field_type == "file":
|
|
if not self.file_types:
|
|
self.file_types = ".pdf,.doc,.docx"
|
|
if self.max_file_size <= 0:
|
|
raise ValidationError("Max file size must be greater than 0")
|
|
if self.multiple_files and self.max_files <= 0:
|
|
raise ValidationError(
|
|
"Max files must be greater than 0 when multiple files are allowed"
|
|
)
|
|
if not self.multiple_files:
|
|
self.max_files = 1
|
|
else:
|
|
# Clear file settings for non-file fields
|
|
self.file_types = ""
|
|
self.max_file_size = 0
|
|
self.multiple_files = False
|
|
self.max_files = 1
|
|
|
|
# Validate order
|
|
if self.order < 0:
|
|
raise ValidationError("Order must be a positive integer")
|
|
|
|
def __str__(self):
|
|
return f"{self.stage.template.name} - {self.stage.name} - {self.label}"
|
|
|
|
|
|
class FormSubmission(Base):
|
|
"""
|
|
Represents a completed form submission by an applicant
|
|
"""
|
|
|
|
template = models.ForeignKey(
|
|
FormTemplate, on_delete=models.CASCADE, related_name="submissions", db_index=True
|
|
)
|
|
submitted_by = models.ForeignKey(
|
|
User,
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name="form_submissions",
|
|
db_index=True
|
|
)
|
|
submitted_at = models.DateTimeField(db_index=True, auto_now_add=True) # Added index
|
|
applicant_name = models.CharField(
|
|
max_length=200, blank=True, help_text="Name of the applicant"
|
|
)
|
|
applicant_email = models.EmailField(db_index=True, blank=True, help_text="Email of the applicant") # Added index
|
|
|
|
class Meta:
|
|
ordering = ["-submitted_at"]
|
|
verbose_name = "Form Submission"
|
|
verbose_name_plural = "Form Submissions"
|
|
indexes = [
|
|
models.Index(fields=['submitted_at']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"Submission for {self.template.name} - {self.submitted_at.strftime('%Y-%m-%d %H:%M')}"
|
|
|
|
|
|
class FieldResponse(Base):
|
|
"""
|
|
Represents a response to a specific field in a form submission
|
|
"""
|
|
|
|
submission = models.ForeignKey(
|
|
FormSubmission, on_delete=models.CASCADE, related_name="responses", db_index=True
|
|
)
|
|
field = models.ForeignKey(
|
|
FormField, on_delete=models.CASCADE, related_name="responses", db_index=True
|
|
)
|
|
|
|
# Store the response value as JSON to handle different data types
|
|
value = models.JSONField(
|
|
null=True, blank=True, help_text="Response value (stored as JSON)"
|
|
)
|
|
|
|
# For file uploads, store the file path
|
|
uploaded_file = models.FileField(upload_to="form_uploads/", null=True, blank=True)
|
|
|
|
class Meta:
|
|
verbose_name = "Field Response"
|
|
verbose_name_plural = "Field Responses"
|
|
indexes = [
|
|
models.Index(fields=['submission']),
|
|
models.Index(fields=['field']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"Response to {self.field.label} in {self.submission}"
|
|
|
|
@property
|
|
def is_file(self):
|
|
if self.uploaded_file:
|
|
return True
|
|
return False
|
|
|
|
@property
|
|
def get_file(self):
|
|
if self.is_file:
|
|
return self.uploaded_file
|
|
return None
|
|
|
|
@property
|
|
def get_file_size(self):
|
|
if self.is_file:
|
|
return self.uploaded_file.size
|
|
return 0
|
|
|
|
@property
|
|
def display_value(self):
|
|
"""Return a human-readable representation of the response value"""
|
|
if self.is_file:
|
|
return f"File: {self.uploaded_file.name}"
|
|
elif self.value is None:
|
|
return ""
|
|
elif isinstance(self.value, list):
|
|
return ", ".join(str(v) for v in self.value)
|
|
else:
|
|
return str(self.value)
|
|
|
|
|
|
# Optional: Create a model for form templates that can be shared across organizations
|
|
class SharedFormTemplate(Base):
|
|
"""
|
|
Represents a form template that can be shared across different organizations/users
|
|
"""
|
|
|
|
template = models.OneToOneField(FormTemplate, on_delete=models.CASCADE)
|
|
is_public = models.BooleanField(
|
|
default=False, help_text="Whether this template is publicly available"
|
|
)
|
|
shared_with = models.ManyToManyField(
|
|
User, blank=True, related_name="shared_templates"
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = "Shared Form Template"
|
|
verbose_name_plural = "Shared Form Templates"
|
|
|
|
def __str__(self):
|
|
return f"Shared: {self.template.name}"
|
|
|
|
|
|
class Source(Base):
|
|
name = models.CharField(
|
|
max_length=100,
|
|
unique=True,
|
|
verbose_name=_("Source Name"),
|
|
help_text=_("e.g., ATS, ERP "),
|
|
)
|
|
source_type = models.CharField(
|
|
max_length=100, verbose_name=_("Source Type"), help_text=_("e.g., ATS, ERP ")
|
|
)
|
|
description = models.TextField(
|
|
blank=True,
|
|
verbose_name=_("Description"),
|
|
help_text=_("A description of the source"),
|
|
)
|
|
ip_address = models.GenericIPAddressField(
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("IP Address"),
|
|
help_text=_("The IP address of the source"),
|
|
)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
|
|
# Integration specific fields
|
|
api_key = models.CharField(
|
|
max_length=255,
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("API Key"),
|
|
help_text=_("API key for authentication (will be encrypted)"),
|
|
)
|
|
api_secret = models.CharField(
|
|
max_length=255,
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("API Secret"),
|
|
help_text=_("API secret for authentication (will be encrypted)"),
|
|
)
|
|
trusted_ips = models.TextField(
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("Trusted IP Addresses"),
|
|
help_text=_("Comma-separated list of trusted IP addresses"),
|
|
)
|
|
is_active = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Active"),
|
|
help_text=_("Whether this source is active for integration"),
|
|
)
|
|
integration_version = models.CharField(
|
|
max_length=50,
|
|
blank=True,
|
|
verbose_name=_("Integration Version"),
|
|
help_text=_("Version of the integration protocol"),
|
|
)
|
|
last_sync_at = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Last Sync At"),
|
|
help_text=_("Timestamp of the last successful synchronization"),
|
|
)
|
|
sync_status = models.CharField(
|
|
max_length=20,
|
|
blank=True,
|
|
choices=[
|
|
("IDLE", "Idle"),
|
|
("SYNCING", "Syncing"),
|
|
("ERROR", "Error"),
|
|
("DISABLED", "Disabled"),
|
|
],
|
|
default="IDLE",
|
|
verbose_name=_("Sync Status"),
|
|
)
|
|
|
|
# Outbound sync configuration
|
|
sync_endpoint = models.URLField(
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("Sync Endpoint"),
|
|
help_text=_("Endpoint URL for sending candidate data (for outbound sync)"),
|
|
)
|
|
sync_method = models.CharField(
|
|
max_length=10,
|
|
blank=True,
|
|
choices=[
|
|
("POST", "POST"),
|
|
("PUT", "PUT"),
|
|
],
|
|
default="POST",
|
|
verbose_name=_("Sync Method"),
|
|
help_text=_("HTTP method for outbound sync requests"),
|
|
)
|
|
test_method = models.CharField(
|
|
max_length=10,
|
|
blank=True,
|
|
choices=[
|
|
("GET", "GET"),
|
|
("POST", "POST"),
|
|
],
|
|
default="GET",
|
|
verbose_name=_("Test Method"),
|
|
help_text=_("HTTP method for connection testing"),
|
|
)
|
|
custom_headers = models.TextField(
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("Custom Headers"),
|
|
help_text=_("JSON object with custom HTTP headers for sync requests"),
|
|
)
|
|
supports_outbound_sync = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("Supports Outbound Sync"),
|
|
help_text=_("Whether this source supports receiving candidate data from ATS"),
|
|
)
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class Meta:
|
|
verbose_name = _("Source")
|
|
verbose_name_plural = _("Sources")
|
|
ordering = ["name"]
|
|
|
|
|
|
class IntegrationLog(Base):
|
|
"""
|
|
Log all integration requests and responses for audit and debugging purposes
|
|
"""
|
|
|
|
class ActionChoices(models.TextChoices):
|
|
REQUEST = "REQUEST", _("Request")
|
|
RESPONSE = "RESPONSE", _("Response")
|
|
ERROR = "ERROR", _("Error")
|
|
SYNC = "SYNC", _("Sync")
|
|
CREATE_JOB = "CREATE_JOB", _("Create Job")
|
|
UPDATE_JOB = "UPDATE_JOB", _("Update Job")
|
|
|
|
source = models.ForeignKey(
|
|
Source,
|
|
on_delete=models.CASCADE,
|
|
related_name="integration_logs",
|
|
verbose_name=_("Source"),
|
|
)
|
|
action = models.CharField(
|
|
max_length=20, choices=ActionChoices.choices, verbose_name=_("Action")
|
|
)
|
|
endpoint = models.CharField(max_length=255, blank=True, verbose_name=_("Endpoint"))
|
|
method = models.CharField(max_length=50, blank=True, verbose_name=_("HTTP Method"))
|
|
request_data = models.JSONField(
|
|
blank=True, null=True, verbose_name=_("Request Data")
|
|
)
|
|
response_data = models.JSONField(
|
|
blank=True, null=True, verbose_name=_("Response Data")
|
|
)
|
|
status_code = models.CharField(
|
|
max_length=10, blank=True, verbose_name=_("Status Code")
|
|
)
|
|
error_message = models.TextField(blank=True, verbose_name=_("Error Message"))
|
|
ip_address = models.GenericIPAddressField(verbose_name=_("IP Address"))
|
|
user_agent = models.CharField(
|
|
max_length=255, blank=True, verbose_name=_("User Agent")
|
|
)
|
|
processing_time = models.FloatField(
|
|
null=True, blank=True, verbose_name=_("Processing Time (seconds)")
|
|
)
|
|
|
|
def __str__(self):
|
|
return f"{self.source.name} - {self.action} - {self.created_at}"
|
|
|
|
class Meta:
|
|
ordering = ["-created_at"]
|
|
verbose_name = _("Integration Log")
|
|
verbose_name_plural = _("Integration Logs")
|
|
|
|
@property
|
|
def is_successful(self):
|
|
"""Check if the integration action was successful"""
|
|
if self.action == self.ActionChoices.ERROR:
|
|
return False
|
|
if self.action == self.ActionChoices.REQUEST:
|
|
return True # Requests are always logged, success depends on response
|
|
if self.status_code and self.status_code.startswith("2"):
|
|
return True
|
|
return False
|
|
|
|
|
|
class HiringAgency(Base):
|
|
name = models.CharField(max_length=200, unique=True, verbose_name=_("Agency Name"))
|
|
contact_person = models.CharField(
|
|
max_length=150, blank=True, verbose_name=_("Contact Person")
|
|
)
|
|
email = models.EmailField(blank=True)
|
|
phone = models.CharField(max_length=20, blank=True)
|
|
website = models.URLField(blank=True)
|
|
notes = models.TextField(blank=True, help_text=_("Internal notes about the agency"))
|
|
country = CountryField(blank=True, null=True, blank_label=_("Select country"))
|
|
address = models.TextField(blank=True, null=True)
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class Meta:
|
|
verbose_name = _("Hiring Agency")
|
|
verbose_name_plural = _("Hiring Agencies")
|
|
ordering = ["name"]
|
|
|
|
|
|
class AgencyJobAssignment(Base):
|
|
"""Assigns specific jobs to agencies with limits and deadlines"""
|
|
|
|
class AssignmentStatus(models.TextChoices):
|
|
ACTIVE = "ACTIVE", _("Active")
|
|
COMPLETED = "COMPLETED", _("Completed")
|
|
EXPIRED = "EXPIRED", _("Expired")
|
|
CANCELLED = "CANCELLED", _("Cancelled")
|
|
|
|
agency = models.ForeignKey(
|
|
HiringAgency,
|
|
on_delete=models.CASCADE,
|
|
related_name="job_assignments",
|
|
verbose_name=_("Agency")
|
|
)
|
|
job = models.ForeignKey(
|
|
JobPosting,
|
|
on_delete=models.CASCADE,
|
|
related_name="agency_assignments",
|
|
verbose_name=_("Job")
|
|
)
|
|
|
|
# Limits & Controls
|
|
max_candidates = models.PositiveIntegerField(
|
|
verbose_name=_("Maximum Candidates"),
|
|
help_text=_("Maximum candidates agency can submit for this job")
|
|
)
|
|
candidates_submitted = models.PositiveIntegerField(
|
|
default=0,
|
|
verbose_name=_("Candidates Submitted"),
|
|
help_text=_("Number of candidates submitted so far")
|
|
)
|
|
|
|
# Timeline
|
|
assigned_date = models.DateTimeField(auto_now_add=True, verbose_name=_("Assigned Date"))
|
|
deadline_date = models.DateTimeField(
|
|
verbose_name=_("Deadline Date"),
|
|
help_text=_("Deadline for agency to submit candidates")
|
|
)
|
|
|
|
# Status & Extensions
|
|
is_active = models.BooleanField(default=True, verbose_name=_("Is Active"))
|
|
status = models.CharField(
|
|
max_length=20,
|
|
choices=AssignmentStatus.choices,
|
|
default=AssignmentStatus.ACTIVE,
|
|
verbose_name=_("Status")
|
|
)
|
|
|
|
# Extension tracking
|
|
deadline_extended = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("Deadline Extended")
|
|
)
|
|
original_deadline = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Original Deadline"),
|
|
help_text=_("Original deadline before extensions")
|
|
)
|
|
|
|
# Admin notes
|
|
admin_notes = models.TextField(
|
|
blank=True,
|
|
verbose_name=_("Admin Notes"),
|
|
help_text=_("Internal notes about this assignment")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Agency Job Assignment")
|
|
verbose_name_plural = _("Agency Job Assignments")
|
|
ordering = ["-created_at"]
|
|
indexes = [
|
|
models.Index(fields=['agency', 'status']),
|
|
models.Index(fields=['job', 'status']),
|
|
models.Index(fields=['deadline_date']),
|
|
models.Index(fields=['is_active']),
|
|
]
|
|
unique_together = ['agency', 'job'] # Prevent duplicate assignments
|
|
|
|
def __str__(self):
|
|
return f"{self.agency.name} - {self.job.title}"
|
|
|
|
@property
|
|
def days_remaining(self):
|
|
"""Calculate days remaining until deadline"""
|
|
if not self.deadline_date:
|
|
return 0
|
|
delta = self.deadline_date.date() - timezone.now().date()
|
|
return max(0, delta.days)
|
|
|
|
@property
|
|
def is_currently_active(self):
|
|
"""Check if assignment is currently active"""
|
|
return (
|
|
self.status == 'ACTIVE' and
|
|
self.deadline_date and
|
|
self.deadline_date > timezone.now() and
|
|
self.candidates_submitted < self.max_candidates
|
|
)
|
|
|
|
@property
|
|
def can_submit(self):
|
|
"""Check if candidates can still be submitted"""
|
|
return self.is_currently_active
|
|
|
|
def clean(self):
|
|
"""Validate assignment constraints"""
|
|
if self.deadline_date and self.deadline_date <= timezone.now():
|
|
raise ValidationError(_("Deadline date must be in the future"))
|
|
|
|
if self.max_candidates <= 0:
|
|
raise ValidationError(_("Maximum candidates must be greater than 0"))
|
|
|
|
if self.candidates_submitted > self.max_candidates:
|
|
raise ValidationError(_("Candidates submitted cannot exceed maximum candidates"))
|
|
|
|
@property
|
|
def remaining_slots(self):
|
|
"""Return number of remaining candidate slots"""
|
|
return max(0, self.max_candidates - self.candidates_submitted)
|
|
|
|
@property
|
|
def is_expired(self):
|
|
"""Check if assignment has expired"""
|
|
return self.deadline_date and self.deadline_date <= timezone.now()
|
|
|
|
@property
|
|
def is_full(self):
|
|
"""Check if assignment has reached maximum candidates"""
|
|
return self.candidates_submitted >= self.max_candidates
|
|
|
|
@property
|
|
def can_submit(self):
|
|
"""Check if agency can still submit candidates"""
|
|
return (self.is_active and
|
|
not self.is_expired and
|
|
not self.is_full and
|
|
self.status == self.AssignmentStatus.ACTIVE)
|
|
|
|
def increment_submission_count(self):
|
|
"""Safely increment the submitted candidates count"""
|
|
if self.can_submit:
|
|
self.candidates_submitted += 1
|
|
self.save(update_fields=['candidates_submitted'])
|
|
|
|
# Check if assignment is now complete
|
|
# if self.candidates_submitted >= self.max_candidates:
|
|
# self.status = self.AssignmentStatus.COMPLETED
|
|
# self.save(update_fields=['status'])
|
|
return True
|
|
return False
|
|
|
|
def extend_deadline(self, new_deadline):
|
|
"""Extend the deadline for this assignment"""
|
|
# Convert database deadline to timezone-aware for comparison
|
|
deadline_aware = timezone.make_aware(self.deadline_date) if timezone.is_naive(self.deadline_date) else self.deadline_date
|
|
if new_deadline > deadline_aware:
|
|
if not self.deadline_extended:
|
|
self.original_deadline = self.deadline_date
|
|
self.deadline_extended = True
|
|
self.deadline_date = new_deadline
|
|
self.save(update_fields=['deadline_date', 'original_deadline', 'deadline_extended'])
|
|
return True
|
|
return False
|
|
|
|
|
|
class AgencyAccessLink(Base):
|
|
"""Secure access links for agencies to submit candidates"""
|
|
|
|
assignment = models.OneToOneField(
|
|
AgencyJobAssignment,
|
|
on_delete=models.CASCADE,
|
|
related_name="access_link",
|
|
verbose_name=_("Assignment")
|
|
)
|
|
|
|
# Security
|
|
unique_token = models.CharField(
|
|
max_length=64,
|
|
unique=True,
|
|
editable=False,
|
|
verbose_name=_("Unique Token")
|
|
)
|
|
access_password = models.CharField(
|
|
max_length=32,
|
|
verbose_name=_("Access Password"),
|
|
help_text=_("Password for agency access")
|
|
)
|
|
|
|
# Timeline
|
|
created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Created At"))
|
|
expires_at = models.DateTimeField(
|
|
verbose_name=_("Expires At"),
|
|
help_text=_("When this access link expires")
|
|
)
|
|
last_accessed = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Last Accessed")
|
|
)
|
|
|
|
# Usage tracking
|
|
access_count = models.PositiveIntegerField(
|
|
default=0,
|
|
verbose_name=_("Access Count")
|
|
)
|
|
is_active = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Is Active")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Agency Access Link")
|
|
verbose_name_plural = _("Agency Access Links")
|
|
ordering = ["-created_at"]
|
|
indexes = [
|
|
models.Index(fields=['unique_token']),
|
|
models.Index(fields=['expires_at']),
|
|
models.Index(fields=['is_active']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"Access Link for {self.assignment}"
|
|
|
|
def clean(self):
|
|
"""Validate access link constraints"""
|
|
if self.expires_at and self.expires_at <= timezone.now():
|
|
raise ValidationError(_("Expiration date must be in the future"))
|
|
|
|
@property
|
|
def is_expired(self):
|
|
"""Check if access link has expired"""
|
|
return self.expires_at and self.expires_at <= timezone.now()
|
|
|
|
@property
|
|
def is_valid(self):
|
|
"""Check if access link is valid and active"""
|
|
return self.is_active and not self.is_expired
|
|
|
|
def record_access(self):
|
|
"""Record an access to this link"""
|
|
self.last_accessed = timezone.now()
|
|
self.access_count += 1
|
|
self.save(update_fields=['last_accessed', 'access_count'])
|
|
|
|
def generate_token(self):
|
|
"""Generate a unique secure token"""
|
|
import secrets
|
|
self.unique_token = secrets.token_urlsafe(48)
|
|
|
|
def generate_password(self):
|
|
"""Generate a random password"""
|
|
import secrets
|
|
import string
|
|
alphabet = string.ascii_letters + string.digits
|
|
self.access_password = ''.join(secrets.choice(alphabet) for _ in range(12))
|
|
|
|
def save(self, *args, **kwargs):
|
|
"""Override save to generate token and password if not set"""
|
|
if not self.unique_token:
|
|
self.generate_token()
|
|
if not self.access_password:
|
|
self.generate_password()
|
|
super().save(*args, **kwargs)
|
|
|
|
|
|
|
|
|
|
class BreakTime(models.Model):
|
|
"""Model to store break times for a schedule"""
|
|
|
|
start_time = models.TimeField(verbose_name=_("Start Time"))
|
|
end_time = models.TimeField(verbose_name=_("End Time"))
|
|
|
|
def __str__(self):
|
|
return f"{self.start_time} - {self.end_time}"
|
|
|
|
|
|
class InterviewSchedule(Base):
|
|
"""Stores the scheduling criteria for interviews"""
|
|
|
|
job = models.ForeignKey(
|
|
JobPosting, on_delete=models.CASCADE, related_name="interview_schedules", db_index=True
|
|
)
|
|
candidates = models.ManyToManyField(Candidate, related_name="interview_schedules", blank=True,null=True)
|
|
start_date = models.DateField(db_index=True, verbose_name=_("Start Date")) # Added index
|
|
end_date = models.DateField(db_index=True, verbose_name=_("End Date")) # Added index
|
|
working_days = models.JSONField(
|
|
verbose_name=_("Working Days")
|
|
) # Store days of week as [0,1,2,3,4] for Mon-Fri
|
|
start_time = models.TimeField(verbose_name=_("Start Time"))
|
|
end_time = models.TimeField(verbose_name=_("End Time"))
|
|
|
|
break_start_time = models.TimeField(verbose_name=_("Break Start Time"),null=True,blank=True)
|
|
break_end_time = models.TimeField(verbose_name=_("Break End Time"),null=True,blank=True)
|
|
|
|
interview_duration = models.PositiveIntegerField(
|
|
verbose_name=_("Interview Duration (minutes)")
|
|
)
|
|
buffer_time = models.PositiveIntegerField(
|
|
verbose_name=_("Buffer Time (minutes)"), default=0
|
|
)
|
|
created_by = models.ForeignKey(User, on_delete=models.CASCADE, db_index=True) # Added index
|
|
|
|
def __str__(self):
|
|
return f"Interview Schedule for {self.job.title}"
|
|
|
|
class Meta:
|
|
indexes = [
|
|
models.Index(fields=['start_date']),
|
|
models.Index(fields=['end_date']),
|
|
models.Index(fields=['created_by']),
|
|
]
|
|
|
|
|
|
|
|
class ScheduledInterview(Base):
|
|
"""Stores individual scheduled interviews"""
|
|
|
|
candidate = models.ForeignKey(
|
|
Candidate,
|
|
on_delete=models.CASCADE,
|
|
related_name="scheduled_interviews",
|
|
db_index=True
|
|
)
|
|
|
|
|
|
job = models.ForeignKey(
|
|
"JobPosting", on_delete=models.CASCADE, related_name="scheduled_interviews", db_index=True
|
|
)
|
|
zoom_meeting = models.OneToOneField(
|
|
ZoomMeeting, on_delete=models.CASCADE, related_name="interview", db_index=True
|
|
)
|
|
schedule = models.ForeignKey(
|
|
InterviewSchedule, on_delete=models.CASCADE, related_name="interviews",null=True,blank=True, db_index=True
|
|
)
|
|
|
|
interview_date = models.DateField(db_index=True, verbose_name=_("Interview Date")) # Added index
|
|
interview_time = models.TimeField(verbose_name=_("Interview Time"))
|
|
status = models.CharField(
|
|
db_index=True, max_length=20, # Added index
|
|
choices=[
|
|
("scheduled", _("Scheduled")),
|
|
("confirmed", _("Confirmed")),
|
|
("cancelled", _("Cancelled")),
|
|
("completed", _("Completed")),
|
|
],
|
|
default="scheduled",
|
|
)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
def __str__(self):
|
|
return f"Interview with {self.candidate.name} for {self.job.title}"
|
|
|
|
class Meta:
|
|
indexes = [
|
|
models.Index(fields=['job', 'status']),
|
|
models.Index(fields=['interview_date', 'interview_time']),
|
|
models.Index(fields=['candidate', 'job']),
|
|
]
|
|
|
|
class Notification(models.Model):
|
|
"""
|
|
Model to store system notifications, primarily for emails.
|
|
"""
|
|
class NotificationType(models.TextChoices):
|
|
EMAIL = "email", _("Email")
|
|
IN_APP = "in_app", _("In-App") # For future expansion
|
|
|
|
class Status(models.TextChoices):
|
|
PENDING = "pending", _("Pending")
|
|
SENT = "sent", _("Sent")
|
|
READ = "read", _("Read")
|
|
FAILED = "failed", _("Failed")
|
|
RETRYING = "retrying", _("Retrying")
|
|
|
|
recipient = models.ForeignKey(
|
|
User,
|
|
on_delete=models.CASCADE,
|
|
related_name="notifications",
|
|
verbose_name=_("Recipient")
|
|
)
|
|
message = models.TextField(verbose_name=_("Notification Message"))
|
|
notification_type = models.CharField(
|
|
max_length=20,
|
|
choices=NotificationType.choices,
|
|
default=NotificationType.EMAIL,
|
|
verbose_name=_("Notification Type")
|
|
)
|
|
status = models.CharField(
|
|
max_length=20,
|
|
choices=Status.choices,
|
|
default=Status.PENDING,
|
|
verbose_name=_("Status")
|
|
)
|
|
related_meeting = models.ForeignKey(
|
|
ZoomMeeting,
|
|
on_delete=models.CASCADE,
|
|
related_name="notifications",
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Related Meeting")
|
|
)
|
|
scheduled_for = models.DateTimeField(
|
|
verbose_name=_("Scheduled Send Time"),
|
|
help_text=_("The date and time this notification is scheduled to be sent.")
|
|
)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
attempts = models.PositiveIntegerField(default=0, verbose_name=_("Send Attempts"))
|
|
last_error = models.TextField(blank=True, verbose_name=_("Last Error Message"))
|
|
|
|
class Meta:
|
|
ordering = ["-scheduled_for", "-created_at"]
|
|
verbose_name = _("Notification")
|
|
verbose_name_plural = _("Notifications")
|
|
indexes = [
|
|
models.Index(fields=['status', 'scheduled_for']),
|
|
models.Index(fields=['recipient']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"Notification for {self.recipient.get_username()} ({self.get_status_display()})"
|
|
|
|
def mark_as_sent(self):
|
|
self.status = Notification.Status.SENT
|
|
self.last_error = ""
|
|
self.save(update_fields=['status', 'last_error'])
|
|
|
|
def mark_as_failed(self, error_message=""):
|
|
self.status = Notification.Status.FAILED
|
|
self.last_error = error_message
|
|
self.attempts += 1
|
|
self.save(update_fields=['status', 'last_error', 'attempts'])
|
|
|
|
|
|
|
|
class Participants(Base):
|
|
"""Model to store Participants details"""
|
|
name = models.CharField(max_length=255, verbose_name=_("Participant Name"),null=True,blank=True)
|
|
email= models.EmailField(verbose_name=_("Email"))
|
|
phone = models.CharField(max_length=12,verbose_name=_("Phone Number"),null=True,blank=True)
|
|
designation = models.CharField(
|
|
max_length=100, blank=True, verbose_name=_("Designation"),null=True
|
|
)
|
|
|
|
def __str__(self):
|
|
return f"{self.name} - {self.email}"
|
|
|
|
|