marwan/kaauh_ats
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
kaauh_ats/recruitment/models.py

2648 lines
86 KiB
Python
Raw Permalink Blame History

import os
from django.db import models
from django.urls import reverse
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from django.utils.html import strip_tags
from django.core.validators import URLValidator
from django.core.exceptions import ValidationError
from django.contrib.auth.models import AbstractUser
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.db.models import F, Value, IntegerField, Q
from django.db.models.functions import Cast, Coalesce
from django.db.models.fields.json import KeyTransform, KeyTextTransform
from django_countries.fields import CountryField
from django_ckeditor_5.fields import CKEditor5Field
from django_extensions.db.fields import RandomCharField
from django.contrib.postgres.validators import MinValueValidator, MaxValueValidator
from secured_fields import EncryptedCharField,EncryptedTextField
from typing import List, Dict, Any
from .validators import validate_hash_tags, validate_image_size
class EmailContent(models.Model):
subject = models.CharField(max_length=255, verbose_name=_("Subject"))
message = CKEditor5Field(verbose_name=_("Message Body"))
class Meta:
verbose_name = _("Email Content")
verbose_name_plural = _("Email Contents")
def __str__(self):
return self.subject
class CustomUser(AbstractUser):
"""Custom user model extending AbstractUser"""
USER_TYPES = [
("staff", _("Staff")),
("agency", _("Agency")),
("candidate", _("Candidate")),
]
first_name=EncryptedCharField(_("first name"), max_length=150, blank=True,searchable=True)
user_type = models.CharField(
max_length=20,
choices=USER_TYPES,
default="staff",
verbose_name=_("User Type"),
db_index=True, # Added index for user_type filtering
)
phone = EncryptedCharField(
blank=True, null=True, verbose_name=_("Phone"),searchable=True
)
profile_image = models.ImageField(
null=True,
blank=True,
upload_to="profile_pic/",
validators=[validate_image_size],
verbose_name=_("Profile Image"),
)
designation = models.CharField(
max_length=100, blank=True, null=True, verbose_name=_("Designation")
)
email = models.EmailField(
unique=True,
db_index=True, # Added explicit index
error_messages={
"unique": _("A user with this email already exists."),
},
)
class Meta:
verbose_name = _("User")
verbose_name_plural = _("Users")
indexes = [
models.Index(fields=["user_type", "is_active"]),
models.Index(fields=["email"]),
]
@property
def get_unread_message_count(self):
message_list = Message.objects.filter(Q(recipient=self), is_read=False)
return message_list.count() or 0
User = get_user_model()
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 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")),
]
# 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 applicants apply",
null=True,
blank=True,
)
application_deadline = models.DateField(db_index=True)
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 applicants 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)
assigned_to = models.ForeignKey(
User,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="assigned_jobs",
help_text=_("The user who has been assigned to this job"),
verbose_name=_("Assigned To"),
)
ai_parsed = models.BooleanField(
default=False,
help_text=_("Whether the job posting has been parsed by AI"),
verbose_name=_("AI Parsed"),
)
# Field to store the generated zip file
cv_zip_file = models.FileField(upload_to="job_zips/", null=True, blank=True)
# Field to track if the background task has completed
zip_created = models.BooleanField(default=False)
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"]),
models.Index(
fields=["assigned_to", "status"]
), # Added for assigned jobs queries
models.Index(
fields=["application_deadline", "status"]
), # Added for deadline filtering
models.Index(
fields=["created_by", "created_at"]
), # Added for creator queries
]
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}"
if self.department:
self.department = self.department.title()
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("&nbsp;", " ")
# 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 applications associated with this job."""
return self.applications.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_applications(self):
# 1. Define the safe JSON extraction and conversion expression
safe_score_expression = Cast(
Coalesce(
# Extract the score explicitly as a text string (KeyTextTransform)
KeyTextTransform(
"match_score", KeyTransform("analysis_data_en", "ai_analysis_data")
),
Value("0"), # Replace SQL NULL (from missing score) with the string '0'
),
output_field=IntegerField(), # Cast the resulting string ('90' or '0') to an integer
)
# 2. Annotate the score using the safe expression
return self.applications.annotate(
sortable_score=safe_score_expression
).order_by("-sortable_score")
@property
def screening_applications(self):
return self.all_applications.filter(stage="Applied")
@property
def exam_applications(self):
return self.all_applications.filter(stage="Exam")
@property
def interview_applications(self):
return self.all_applications.filter(stage="Interview")
@property
def document_review_applications(self):
return self.all_applications.filter(stage="Document Review")
@property
def offer_applications(self):
return self.all_applications.filter(stage="Offer")
@property
def accepted_applications(self):
return self.all_applications.filter(offer_status="Accepted")
@property
def hired_applications(self):
return self.all_applications.filter(stage="Hired")
# counts
@property
def all_applications_count(self):
return self.all_applications.count()
@property
def screening_applications_count(self):
return self.all_applications.filter(stage="Applied").count() or 0
@property
def exam_applications_count(self):
return self.all_applications.filter(stage="Exam").count() or 0
@property
def interview_applications_count(self):
return self.all_applications.filter(stage="Interview").count() or 0
@property
def document_review_applications_count(self):
return self.all_applications.filter(stage="Document Review").count() or 0
@property
def offer_applications_count(self):
return self.all_applications.filter(stage="Offer").count() or 0
@property
def hired_applications_count(self):
return self.all_applications.filter(stage="Hired").count() or 0
@property
def source_sync_data(self):
if self.source:
return [
{
"first_name": x.person.first_name,
"middle_name": x.person.middle_name,
"last_name": x.person.last_name,
"email": x.person.email,
"phone": x.person.phone,
"date_of_birth": str(x.person.date_of_birth)
if x.person.date_of_birth
else "",
"nationality": str(x.person.nationality),
"gpa": x.person.gpa,
}
for x in self.hired_applications.all()
]
return []
@property
def vacancy_fill_rate(self):
total_positions = self.open_positions
print(total_positions)
no_of_positions_filled = self.applications.filter(stage__in=["Hired"]).count()
print(no_of_positions_filled)
if total_positions > 0:
vacancy_fill_rate = no_of_positions_filled / total_positions
else:
vacancy_fill_rate = 0.0
return vacancy_fill_rate
def has_already_applied_to_this_job(self, person):
"""Check if a given person has already applied to this job."""
return self.applications.filter(person=person).exists()
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 Person(Base):
"""Model to store personal information that can be reused across multiple applications"""
GENDER_CHOICES = [
("M", _("Male")),
("F", _("Female")),
]
# Personal Information
first_name = EncryptedCharField(max_length=255, verbose_name=_("First Name"),searchable=True)
last_name = models.CharField(max_length=255, verbose_name=_("Last Name"))
middle_name = models.CharField(
max_length=255, blank=True, null=True, verbose_name=_("Middle Name")
)
email = models.EmailField(
unique=True,
db_index=True,
verbose_name=_("Email"),
)
phone = EncryptedCharField(
blank=True, null=True, verbose_name=_("Phone"),searchable=True
)
date_of_birth = models.DateField(
null=True, blank=True, verbose_name=_("Date of Birth")
)
gender = models.CharField(
max_length=1,
choices=GENDER_CHOICES,
blank=True,
null=True,
verbose_name=_("Gender"),
)
gpa = models.DecimalField(
max_digits=3,
decimal_places=2,
verbose_name=_("GPA"),
help_text=_("GPA must be between 0 and 4."),
validators=[MinValueValidator(0), MaxValueValidator(4)],
)
national_id = EncryptedCharField(
help_text=_("Enter the national id or iqama number")
)
nationality = CountryField(blank=True, null=True, verbose_name=_("Nationality"))
address = models.TextField(blank=True, null=True, verbose_name=_("Address"))
# Optional linking to user account
user = models.OneToOneField(
User,
on_delete=models.CASCADE,
related_name="person_profile",
verbose_name=_("User Account"),
null=True,
blank=True,
)
# Profile information
profile_image = models.ImageField(
null=True,
blank=True,
upload_to="profile_pic/",
validators=[validate_image_size],
verbose_name=_("Profile Image"),
)
linkedin_profile = models.URLField(
blank=True, null=True, verbose_name=_("LinkedIn Profile URL")
)
agency = models.ForeignKey(
"HiringAgency",
on_delete=models.SET_NULL,
null=True,
blank=True,
verbose_name=_("Hiring Agency"),
)
def delete(self, *args, **kwargs):
"""
Custom delete method to ensure the associated User account is also deleted.
"""
# 1. Delete the associated User account first, if it exists
if self.user:
self.user.delete()
# 2. Call the original delete method for the Person instance
super().delete(*args, **kwargs)
class Meta:
verbose_name = _("Person")
verbose_name_plural = _("People")
indexes = [
models.Index(fields=["email"]),
models.Index(fields=["first_name", "last_name"]),
models.Index(fields=["created_at"]),
models.Index(
fields=["agency", "created_at"]
), # OPTIMIZED: For agency person queries
]
def __str__(self):
return f"{self.first_name} {self.last_name}"
@property
def full_name(self):
return f"{self.first_name} {self.last_name}"
@property
def age(self):
"""Calculate age from date of birth"""
if self.date_of_birth:
today = timezone.now().date()
return (
today.year
- self.date_of_birth.year
- (
(today.month, today.day)
< (self.date_of_birth.month, self.date_of_birth.day)
)
)
return None
@property
def documents(self):
"""Return all documents associated with this Person"""
from django.contrib.contenttypes.models import ContentType
content_type = ContentType.objects.get_for_model(self.__class__)
return Document.objects.filter(content_type=content_type, object_id=self.id)
class Application(Base):
"""Model to store job-specific application data"""
class Stage(models.TextChoices):
APPLIED = "Applied", _("Applied")
EXAM = "Exam", _("Exam")
INTERVIEW = "Interview", _("Interview")
DOCUMENT_REVIEW = "Document Review", _("Document Review")
OFFER = "Offer", _("Offer")
HIRED = "Hired", _("Hired")
REJECTED = "Rejected", _("Rejected")
class ExamStatus(models.TextChoices):
PASSED = "Passed", _("Passed")
FAILED = "Failed", _("Failed")
class OfferStatus(models.TextChoices):
ACCEPTED = "Accepted", _("Accepted")
REJECTED = "Rejected", _("Rejected")
PENDING = "Pending", _("Pending")
class ApplicantType(models.TextChoices):
APPLICANT = "Applicant", _("Applicant")
CANDIDATE = "Candidate", _("Candidate")
# Stage transition validation constants
STAGE_SEQUENCE = {
"Applied": ["Exam", "Interview", "Offer", "Rejected"],
"Exam": ["Interview", "Offer", "Rejected"],
"Interview": ["Document Review", "Offer", "Rejected"],
"Document Review": ["Offer", "Rejected"],
"Offer": ["Hired", "Rejected"],
"Rejected": [], # Final stage - no further transitions
"Hired": [], # Final stage - no further transitions
}
# Core relationships
person = models.ForeignKey(
Person,
on_delete=models.CASCADE,
related_name="applications",
verbose_name=_("Person"),
)
job = models.ForeignKey(
JobPosting,
on_delete=models.CASCADE,
related_name="applications",
verbose_name=_("Job"),
)
# Application-specific data
resume = models.FileField(upload_to="resumes/", verbose_name=_("Resume"))
cover_letter = models.FileField(
upload_to="cover_letters/",
blank=True,
null=True,
verbose_name=_("Cover Letter"),
)
is_resume_parsed = models.BooleanField(
default=False, verbose_name=_("Resume Parsed")
)
parsed_summary = models.TextField(blank=True, verbose_name=_("Parsed Summary"))
# Workflow fields
applied = models.BooleanField(default=False, verbose_name=_("Applied"))
stage = models.CharField(
db_index=True,
max_length=20,
default="Applied",
choices=Stage.choices,
verbose_name=_("Stage"),
)
applicant_status = models.CharField(
choices=ApplicantType.choices,
default="Applicant",
max_length=20,
null=True,
blank=True,
verbose_name=_("Applicant Status"),
)
# Timeline fields
exam_date = models.DateTimeField(null=True, blank=True, verbose_name=_("Exam Date"))
exam_status = models.CharField(
choices=ExamStatus.choices,
max_length=20,
null=True,
blank=True,
verbose_name=_("Exam Status"),
)
exam_score = models.FloatField(null=True, blank=True, verbose_name=_("Exam Score"))
interview_date = models.DateTimeField(
null=True, blank=True, verbose_name=_("Interview Date")
)
interview_status = models.CharField(
choices=ExamStatus.choices,
max_length=20,
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=OfferStatus.choices,
max_length=20,
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
ai_analysis_data = models.JSONField(
verbose_name="AI Analysis Data",
default=dict,
help_text="Full JSON output from the resume scoring model.",
null=True,
blank=True,
)
retry = models.SmallIntegerField(verbose_name="Resume Parsing Retry", default=3)
# Source tracking
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="applications",
verbose_name=_("Hiring Agency"),
)
class Meta:
verbose_name = _("Application")
verbose_name_plural = _("Applications")
indexes = [
models.Index(fields=["person", "job"]),
models.Index(fields=["stage"]),
models.Index(fields=["created_at"]),
models.Index(fields=["person", "stage", "created_at"]),
models.Index(
fields=["job", "stage", "created_at"]
), # OPTIMIZED: For job detail statistics
]
unique_together = [["person", "job"]] # Prevent duplicate applications
def __str__(self):
return f"{self.person.full_name} - {self.job.title}"
@property
def resume_data_en(self):
return self.ai_analysis_data.get("resume_data_en", {})
@property
def resume_data_ar(self):
return self.ai_analysis_data.get("resume_data_ar", {})
@property
def analysis_data_en(self):
return self.ai_analysis_data.get("analysis_data_en", {})
@property
def analysis_data_ar(self):
return self.ai_analysis_data.get("analysis_data_ar", {})
@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_en.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_en.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_en.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."""
return self.analysis_data_en.get("experience_industry_match", 0)
@property
def min_requirements_met(self) -> bool:
"""14. Boolean (true/false) indicating if all non-negotiable minimum requirements are met."""
return self.analysis_data_en.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_en.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_en.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_en.get("most_recent_job_title", "N/A")
@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_en.get("criteria_checklist", {})
@property
def professional_category(self) -> str:
"""7. The most fitting professional field or category for the individual."""
return self.analysis_data_en.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_en.get("language_fluency", [])
@property
def strengths(self) -> str:
"""2. A brief summary of why the candidate is a strong fit."""
return self.analysis_data_en.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_en.get("weaknesses", "")
@property
def job_fit_narrative(self) -> str:
"""11. A single, concise sentence summarizing the core fit."""
return self.analysis_data_en.get("job_fit_narrative", "")
@property
def recommendation(self) -> str:
"""9. Provide a detailed final recommendation for the candidate."""
return self.analysis_data_en.get("recommendation", "")
# for arabic
@property
def min_requirements_met_ar(self) -> bool:
"""14. Boolean (true/false) indicating if all non-negotiable minimum requirements are met."""
return self.analysis_data_ar.get("min_req_met_bool", False)
@property
def screening_stage_rating_ar(self) -> str:
"""13. A standardized rating (e.g., "A - Highly Qualified", "B - Qualified")."""
return self.analysis_data_ar.get("screening_stage_rating", "N/A")
@property
def top_3_keywords_ar(self) -> List[str]:
"""10. A list of the three most dominant and relevant technical skills or technologies."""
return self.analysis_data_ar.get("top_3_keywords", [])
@property
def most_recent_job_title_ar(self) -> str:
"""8. The candidate's most recent or current professional job title."""
return self.analysis_data_ar.get("most_recent_job_title", "N/A")
@property
def criteria_checklist_ar(self) -> Dict[str, str]:
"""5 & 6. An object rating the candidate's match for each specific criterion."""
return self.analysis_data_ar.get("criteria_checklist", {})
@property
def professional_category_ar(self) -> str:
"""7. The most fitting professional field or category for the individual."""
return self.analysis_data_ar.get("category", "N/A")
@property
def language_fluency_ar(self) -> List[Dict[str, str]]:
"""12. A list of languages and their fluency levels mentioned."""
return self.analysis_data_ar.get("language_fluency", [])
@property
def strengths_ar(self) -> str:
"""2. A brief summary of why the candidate is a strong fit."""
return self.analysis_data_ar.get("strengths", "")
@property
def weaknesses_ar(self) -> str:
"""3. A brief summary of where the candidate falls short or what criteria are missing."""
return self.analysis_data_ar.get("weaknesses", "")
@property
def job_fit_narrative_ar(self) -> str:
"""11. A single, concise sentence summarizing the core fit."""
return self.analysis_data_ar.get("job_fit_narrative", "")
@property
def recommendation_ar(self) -> str:
"""9. Provide a detailed final recommendation for the candidate."""
return self.analysis_data_ar.get("recommendation", "")
# ====================================================================
# 🔄 HELPER METHODS
# ====================================================================
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
def get_available_stages(self):
"""Get list of stages this application 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, [])
def save(self, *args, **kwargs):
"""Override save to ensure validation is called"""
self.clean() # Call validation before saving
super().save(*args, **kwargs)
# ====================================================================
# 📋 LEGACY COMPATIBILITY PROPERTIES
# ====================================================================
# These properties maintain compatibility with existing code that expects Candidate model
@property
def first_name(self):
"""Legacy compatibility - delegates to person.first_name"""
return self.person.first_name
@property
def last_name(self):
"""Legacy compatibility - delegates to person.last_name"""
return self.person.last_name
@property
def email(self):
"""Legacy compatibility - delegates to person.email"""
return self.person.email
@property
def phone(self):
"""Legacy compatibility - delegates to person.phone if available"""
return self.person.phone or ""
@property
def address(self):
"""Legacy compatibility - delegates to person.address if available"""
return self.person.address or ""
@property
def name(self):
"""Legacy compatibility - delegates to person.full_name"""
return self.person.full_name
@property
def full_name(self):
"""Legacy compatibility - delegates to person.full_name"""
return self.person.full_name
@property
def get_file_size(self):
"""Legacy compatibility - returns resume file size"""
if self.resume:
return self.resume.size
return 0
@property
def submission(self):
"""Legacy compatibility - get form submission for this application"""
return FormSubmission.objects.filter(template__job=self.job).first()
@property
def responses(self):
"""Legacy compatibility - get form responses for this application"""
if self.submission:
return self.submission.responses.all()
return []
@property
def get_meetings(self):
"""Legacy compatibility - get scheduled interviews for this application"""
return self.scheduled_interviews.all()
@property
def has_future_meeting(self):
"""Legacy compatibility - check for future meetings"""
now = timezone.now()
future_meetings = (
self.scheduled_interviews.filter(interview_date__gt=now.date())
.filter(interview_time__gte=now.time())
.exists()
)
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 scoring_timeout(self):
"""Legacy compatibility - check scoring timeout"""
return timezone.now() <= (self.created_at + timezone.timedelta(minutes=5))
@property
def get_interview_date(self):
"""Legacy compatibility - get interview date"""
if hasattr(self, "scheduled_interview") and self.scheduled_interview:
return self.scheduled_interviews.first().interview_date
return None
@property
def get_interview_time(self):
"""Legacy compatibility - get interview time"""
if hasattr(self, "scheduled_interview") and self.scheduled_interview:
return self.scheduled_interviews.first().interview_time
return None
@property
def time_to_hire_days(self):
"""Legacy compatibility - calculate time to hire"""
if self.hired_date and self.created_at:
time_to_hire = self.hired_date - self.created_at.date()
return time_to_hire.days
return 0
@property
def documents(self):
"""Return all documents associated with this Application"""
from django.contrib.contenttypes.models import ContentType
content_type = ContentType.objects.get_for_model(self.__class__)
return Document.objects.filter(content_type=content_type, object_id=self.id)
@property
def belong_to_an_agency(self):
if self.hiring_agency:
return True
else:
return False
@property
def is_active(self):
deadline = self.job.application_deadline
now = timezone.now().date()
if deadline > now:
return True
else:
return False
class Interview(Base):
class LocationType(models.TextChoices):
REMOTE = "Remote", _("Remote (e.g., Zoom, Google Meet)")
ONSITE = "Onsite", _("In-Person (Physical Location)")
class Status(models.TextChoices):
WAITING = "waiting", _("Waiting")
STARTED = "started", _("Started")
UPDATED = "updated", _("Updated")
DELETED = "deleted", _("Deleted")
ENDED = "ended", _("Ended")
class InterviewResult(models.TextChoices):
PASSED="passed",_("Passed")
FAILED="failed",_("Failed")
ON_HOLD="on_hold",_("ON Hold")
location_type = models.CharField(
max_length=10,
choices=LocationType.choices,
verbose_name=_("Location Type"),
db_index=True,
)
interview_result=models.CharField(
max_length=10,
choices=InterviewResult.choices,
verbose_name=_("Interview Result"),
null=True,
blank=True,
default='on_hold'
)
result_comments=models.TextField(
null=True,
blank=True
)
# Common fields
topic = models.CharField(
max_length=255,
verbose_name=_("Meeting/Location Topic"),
blank=True,
help_text=_("e.g., 'Zoom Topic: Software Interview' or 'Main Conference Room'"),
)
join_url = models.URLField(
verbose_name=_("Meeting/Location URL"), max_length=2048, blank=True, null=True
)
timezone = models.CharField(
max_length=50, verbose_name=_("Timezone"), default="UTC"
)
start_time = models.DateTimeField(db_index=True, verbose_name=_("Start Time"))
duration = models.PositiveIntegerField(verbose_name=_("Duration (minutes)"))
status = models.CharField(
max_length=20, choices=Status.choices, default=Status.WAITING, db_index=True
)
cancelled_at = models.DateTimeField(
null=True, blank=True, verbose_name=_("Cancelled At")
)
cancelled_reason = models.TextField(
blank=True, null=True, verbose_name=_("Cancellation Reason")
)
# Remote-specific (nullable)
meeting_id = models.CharField(
max_length=50,
unique=True,
null=True,
blank=True,
verbose_name=_("External Meeting ID"),
)
password = models.CharField(max_length=20, blank=True, null=True)
zoom_gateway_response = models.JSONField(blank=True, null=True)
details_url = models.JSONField(blank=True, null=True)
participant_video = models.BooleanField(default=True)
join_before_host = models.BooleanField(default=False)
host_email = models.CharField(max_length=255, blank=True, null=True)
mute_upon_entry = models.BooleanField(default=False)
waiting_room = models.BooleanField(default=False)
# Onsite-specific (nullable)
physical_address = models.CharField(max_length=255, blank=True, null=True)
room_number = models.CharField(max_length=50, blank=True, null=True)
def __str__(self):
return f"{self.get_location_type_display()} - {self.topic[:50]}"
class Meta:
verbose_name = _("Interview Location")
verbose_name_plural = _("Interview Locations")
def clean(self):
# Optional: add validation
if self.location_type == self.LocationType.REMOTE:
if not self.details_url:
raise ValidationError(_("Remote interviews require a meeting URL."))
if not self.meeting_id:
raise ValidationError(
_("Meeting ID is required for remote interviews.")
)
elif self.location_type == self.LocationType.ONSITE:
if not (self.physical_address or self.room_number):
raise ValidationError(
_("Onsite interviews require at least an address or room.")
)
# --- 2. Scheduling Models ---
class BulkInterviewTemplate(Base):
"""Stores the TEMPLATE criteria for BULK interview generation."""
# We need a field to store the template location details linked to this bulk schedule.
# This location object contains the generic Zoom/Onsite info to be cloned.
interview = models.ForeignKey(
Interview,
on_delete=models.SET_NULL,
related_name="schedule_templates",
null=True,
blank=True,
verbose_name=_("Location Template (Zoom/Onsite)"),
)
job = models.ForeignKey(
JobPosting,
on_delete=models.CASCADE,
related_name="interview_schedules",
db_index=True,
)
applications = models.ManyToManyField(
Application, related_name="interview_schedules", blank=True
)
start_date = models.DateField(db_index=True, verbose_name=_("Start Date"))
end_date = models.DateField(db_index=True, verbose_name=_("End Date"))
working_days = models.JSONField(verbose_name=_("Working Days"))
topic = models.CharField(max_length=255, verbose_name=_("Interview Topic"))
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
)
schedule_interview_type = models.CharField(
max_length=10,
choices=[
("Remote", "Remote (e.g., Zoom)"),
("Onsite", "In-Person (Physical Location)"),
],
default="Onsite",
verbose_name=_("Interview Type"),
)
physical_address = models.CharField(max_length=255, blank=True, null=True)
created_by = models.ForeignKey(User, on_delete=models.CASCADE, db_index=True)
def __str__(self):
return f"Schedule for {self.job.title}"
class ScheduledInterview(Base):
"""Stores individual scheduled interviews (whether bulk or individually created)."""
class InterviewTypeChoice(models.TextChoices):
REMOTE = "Remote", _("Remote (e.g., Zoom, Google Meet)")
ONSITE = "Onsite", _("In-Person (Physical Location)")
class InterviewStatus(models.TextChoices):
SCHEDULED = "scheduled", _("Scheduled")
CONFIRMED = "confirmed", _("Confirmed")
CANCELLED = "cancelled", _("Cancelled")
COMPLETED = "completed", _("Completed")
cancelled_at = models.DateTimeField(null=True, blank=True, verbose_name=_("Cancelled At"))
cancelled_reason = models.TextField(blank=True, null=True, verbose_name=_("Cancellation Reason"))
application = models.ForeignKey(
Application,
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,
)
# Links to the specific, individual location/meeting details for THIS interview
interview = models.OneToOneField(
Interview,
on_delete=models.CASCADE,
related_name="scheduled_interview",
null=True,
blank=True,
db_index=True,
verbose_name=_("Interview/Meeting"),
)
# Link back to the bulk schedule template (optional if individually created)
schedule = models.ForeignKey(
BulkInterviewTemplate,
on_delete=models.SET_NULL,
related_name="interviews",
null=True,
blank=True,
db_index=True,
)
participants = models.ManyToManyField("Participants", blank=True)
system_users = models.ManyToManyField(
User, related_name="attended_interviews", blank=True
)
interview_date = models.DateField(db_index=True, verbose_name=_("Interview Date"))
interview_time = models.TimeField(verbose_name=_("Interview Time"))
interview_type = models.CharField(
max_length=20,
choices=InterviewTypeChoice.choices,
default=InterviewTypeChoice.REMOTE,
)
status = models.CharField(
db_index=True,
max_length=20,
choices=InterviewStatus.choices,
default=InterviewStatus.SCHEDULED,
)
interview_questions = models.JSONField(
verbose_name=_("Question Data"),
blank=True,null=True
)
def __str__(self):
return (
f"Interview with {self.application.person.full_name} for {self.job.title}"
)
class Meta:
indexes = [
models.Index(fields=["job", "status"]),
models.Index(fields=["interview_date", "interview_time"]),
models.Index(fields=["application", "job"]),
]
@property
def get_schedule_type(self):
if self.schedule:
return self.schedule.schedule_interview_type
else:
return self.interview_location.location_type
@property
def get_schedule_status(self):
return self.status
@property
def get_meeting_details(self):
return self.interview_location
# --- 3. Interview Notes Model (Fixed) ---
class Note(Base):
"""Model for storing notes, feedback, or comments related to a specific ScheduledInterview."""
class NoteType(models.TextChoices):
FEEDBACK = "Feedback", _("Candidate Feedback")
LOGISTICS = "Logistics", _("Logistical Note")
GENERAL = "General", _("General Comment")
application = models.ForeignKey(
Application,
on_delete=models.CASCADE,
related_name="notes",
verbose_name=_("Application"),
db_index=True,
null=True,
blank=True,
)
interview = models.ForeignKey(
Interview,
on_delete=models.CASCADE,
related_name="notes",
verbose_name=_("Scheduled Interview"),
db_index=True,
null=True,
blank=True,
)
author = models.ForeignKey(
User,
on_delete=models.CASCADE,
related_name="interview_notes",
verbose_name=_("Author"),
db_index=True,
)
note_type = models.CharField(
max_length=50,
choices=NoteType.choices,
default=NoteType.FEEDBACK,
verbose_name=_("Note Type"),
)
content = CKEditor5Field(verbose_name=_("Content/Feedback"), config_name="extends")
class Meta:
verbose_name = _("Interview Note")
verbose_name_plural = _("Interview Notes")
ordering = ["created_at"]
def __str__(self):
return f"{self.get_note_type_display()} by {self.author.get_username()}"
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)",
)
is_required = models.BooleanField(default=False)
required_message = models.CharField(max_length=255, blank=True)
min_length = models.IntegerField(null=True, blank=True)
max_length = models.IntegerField(null=True, blank=True)
validation_pattern = models.CharField(max_length=50, blank=True, choices=[
('', 'None'),
('email', 'Email'),
('phone', 'Phone'),
('url', 'URL'),
('number', 'Number'),
('alpha', 'Letters Only'),
('alphanum', 'Letters & Numbers'),
('custom', 'Custom')
])
custom_pattern = models.CharField(max_length=255, blank=True)
min_value = models.CharField(max_length=50, blank=True) # For dates and numbers
max_value = models.CharField(max_length=50, blank=True) # For dates and numbers
min_file_size = models.FloatField(null=True, blank=True)
min_image_width = models.IntegerField(null=True, blank=True)
min_image_height = models.IntegerField(null=True, blank=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.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=_("Name of the source"),
)
source_type = models.CharField(
max_length=100, verbose_name=_("Source Type"), help_text=_("Type of the source")
)
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"),
("SUCCESS", "Success"),
("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.JSONField(
blank=True,
null=True,
verbose_name=_("Custom Headers"),
help_text=_("JSON object with custom HTTP headers for sync requests"),
default=dict,
)
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):
user = models.OneToOneField(
User,
on_delete=models.CASCADE,
related_name="agency_profile",
verbose_name=_("User"),
null=True,
blank=True,
)
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(unique=True)
phone = EncryptedCharField(max_length=20, blank=True,null=True,searchable=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)
generated_password = models.CharField(
max_length=255,
blank=True,
null=True,
help_text=_("Generated password for agency user account"),
)
def __str__(self):
return self.name
class Meta:
verbose_name = _("Hiring Agency")
verbose_name_plural = _("Hiring Agencies")
ordering = ["name"]
def delete(self, *args, **kwargs):
"""
Custom delete method to ensure the associated User account is also deleted.
"""
# 1. Delete the associated User account first, if it exists
if self.user:
self.user.delete()
# 2. Call the original delete method for the Agency instance
super().delete(*args, **kwargs)
class AgencyJobAssignment(Base):
"""Assigns specific jobs to agencies with limits and deadlines"""
class AssignmentStatus(models.TextChoices):
ACTIVE = "ACTIVE", _("Active")
COMPLETED = "COMPLETED", _("Completed")
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"),
)
# Cancellation tracking
cancelled_at = models.DateTimeField(
null=True,
blank=True,
verbose_name=_("Cancelled At")
)
cancelled_by = models.CharField(
max_length=100,
blank=True,
null=True,
verbose_name=_("Cancelled By"),
help_text=_("Name of person who cancelled this assignment")
)
cancel_reason = models.TextField(
blank=True,
null=True,
verbose_name=_("Cancel Reason"),
help_text=_("Reason for cancelling 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"])
return True
return False
@property
def applications_submited_count(self):
"""Return the number of applications submitted by the agency for this job"""
return Application.objects.filter(
hiring_agency=self.agency, job=self.job
).count()
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 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"),
)
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 = EncryptedCharField(
max_length=12, verbose_name=_("Phone Number"), null=True, blank=True,searchable=True
)
designation = models.CharField(
max_length=100, blank=True, verbose_name=_("Designation"), null=True
)
def __str__(self):
return f"{self.name} - {self.email}"
class Message(Base):
"""Model for messaging between different user types"""
class MessageType(models.TextChoices):
DIRECT = "direct", _("Direct Message")
JOB_RELATED = "job_related", _("Job Related")
SYSTEM = "system", _("System Notification")
sender = models.ForeignKey(
User,
on_delete=models.CASCADE,
related_name="sent_messages",
verbose_name=_("Sender"),
)
recipient = models.ForeignKey(
User,
on_delete=models.CASCADE,
related_name="received_messages",
null=True,
blank=True,
verbose_name=_("Recipient"),
)
job = models.ForeignKey(
JobPosting,
on_delete=models.CASCADE,
related_name="messages",
verbose_name=_("Related Job"),
)
subject = models.CharField(max_length=200, verbose_name=_("Subject"))
content = models.TextField(verbose_name=_("Message Content"))
message_type = models.CharField(
max_length=20,
choices=MessageType.choices,
default=MessageType.DIRECT,
verbose_name=_("Message Type"),
)
is_read = models.BooleanField(default=False, verbose_name=_("Is Read"))
read_at = models.DateTimeField(null=True, blank=True, verbose_name=_("Read At"))
class Meta:
verbose_name = _("Message")
verbose_name_plural = _("Messages")
ordering = ["-created_at"]
indexes = [
models.Index(fields=["sender", "created_at"]),
models.Index(fields=["recipient", "is_read", "created_at"]),
models.Index(fields=["job", "created_at"]),
models.Index(fields=["message_type", "created_at"]),
]
def __str__(self):
return f"Message from {self.sender.get_username()} to {self.recipient.get_username() if self.recipient else 'N/A'}"
def mark_as_read(self):
"""Mark message as read and set read timestamp"""
if not self.is_read:
self.is_read = True
self.read_at = timezone.now()
self.save(update_fields=["is_read", "read_at"])
@property
def is_job_related(self):
"""Check if message is related to a job"""
return self.job is not None
def get_auto_recipient(self):
"""Get auto recipient based on job assignment"""
if self.job and self.job.assigned_to:
return self.job.assigned_to
return None
def _validate_messaging_permissions(self):
"""Validate if sender can message recipient based on user types"""
sender_type = self.sender.user_type
recipient_type = self.recipient.user_type
# Staff can message anyone
if sender_type == "staff":
return
# Agency users can only message staff or their own candidates
if sender_type == "agency":
if recipient_type not in ["staff", "candidate"]:
raise ValidationError(
_("Agencies can only message staff or candidates.")
)
# If messaging a candidate, ensure candidate is from their agency
if recipient_type == "candidate" and self.job:
if not self.job.hiring_agency.filter(user=self.sender).exists():
raise ValidationError(
_("You can only message candidates from your assigned jobs.")
)
# Candidate users can only message staff
if sender_type == "candidate":
if recipient_type != "staff":
raise ValidationError(_("Candidates can only message staff."))
# If job-related, ensure candidate applied for the job
if self.job:
if not Application.objects.filter(
job=self.job,
person=self.sender, # TODO:fix this
).exists():
raise ValidationError(
_("You can only message about jobs you have applied for.")
)
def save(self, *args, **kwargs):
"""Override save to handle auto-recipient logic"""
self.clean()
super().save(*args, **kwargs)
class Document(Base):
"""Model for storing documents using Generic Foreign Key"""
class DocumentType(models.TextChoices):
RESUME = "resume", _("Resume")
COVER_LETTER = "cover_letter", _("Cover Letter")
CERTIFICATE = "certificate", _("Certificate")
ID_DOCUMENT = "id_document", _("ID Document")
PASSPORT = "passport", _("Passport")
EDUCATION = "education", _("Education Document")
EXPERIENCE = "experience", _("Experience Letter")
OTHER = "other", _("Other")
# Generic Foreign Key fields
content_type = models.ForeignKey(
ContentType,
on_delete=models.CASCADE,
verbose_name=_("Content Type"),
db_index=True, # Added index for foreign key
)
object_id = models.PositiveIntegerField(
verbose_name=_("Object ID"),
db_index=True, # Added index for object_id
)
content_object = GenericForeignKey("content_type", "object_id")
file = models.FileField(
upload_to="documents/%Y/%m/",
verbose_name=_("Document File"),
validators=[validate_image_size],
)
document_type = models.CharField(
max_length=20,
choices=DocumentType.choices,
default=DocumentType.OTHER,
verbose_name=_("Document Type"),
db_index=True, # Added index for document_type filtering
)
description = models.CharField(
max_length=200,
blank=True,
verbose_name=_("Description"),
)
uploaded_by = models.ForeignKey(
User,
on_delete=models.SET_NULL,
null=True,
blank=True,
verbose_name=_("Uploaded By"),
db_index=True, # Added index for foreign key
)
class Meta:
verbose_name = _("Document")
verbose_name_plural = _("Documents")
ordering = ["-created_at"]
indexes = [
models.Index(
fields=["content_type", "object_id", "document_type", "created_at"]
),
models.Index(
fields=["document_type", "created_at"]
), # Added for document type filtering
models.Index(
fields=["uploaded_by", "created_at"]
), # Added for user document queries
]
def delete(self, *args, **kwargs):
if self.file:
if os.path.isfile(self.file.path):
os.remove(self.file.path)
super().delete(*args, **kwargs)
def __str__(self):
try:
if hasattr(self.content_object, "full_name"):
object_name = self.content_object.full_name
elif hasattr(self.content_object, "title"):
object_name = self.content_object.title
elif hasattr(self.content_object, "__str__"):
object_name = str(self.content_object)
else:
object_name = f"Object {self.object_id}"
return f"{self.get_document_type_display()} - {object_name}"
except:
return f"{self.get_document_type_display()} - {self.object_id}"
@property
def file_size(self):
"""Return file size in human readable format"""
if self.file:
size = self.file.size
if size < 1024:
return f"{size} bytes"
elif size < 1024 * 1024:
return f"{size / 1024:.1f} KB"
else:
return f"{size / (1024 * 1024):.1f} MB"
return "0 bytes"
@property
def file_extension(self):
"""Return file extension"""
if self.file:
return self.file.name.split(".")[-1].upper()
return ""
class Settings(Base):
"""Model to store key-value pair settings"""
name = models.CharField(
max_length=100,
verbose_name=_("Friendly Name"),
help_text=_("A human-readable name (e.g., 'Zoom')"),
null=True, blank=True
)
key = models.CharField(
max_length=100,
unique=True,
verbose_name=_("Setting Key"),
help_text=_("Unique key for the setting"),
)
value = EncryptedTextField(
verbose_name=_("Setting Value"),
help_text=_("Value for the setting"),
)
class Meta:
verbose_name = _("Setting")
verbose_name_plural = _("Settings")
ordering = ["key"]
def __str__(self):
return f"{self.key}: {self.value[:50]}{'...' if len(self.value) > 50 else ''}"
Reference in New Issue View Git Blame Copy Permalink