From 34e2224f80154fa4f9b745e7bfc5cf20c92dd178 Mon Sep 17 00:00:00 2001 From: Faheed Date: Sat, 22 Nov 2025 19:43:36 +0300 Subject: [PATCH] revrting to a commit --- recruitment/models.py | 2452 ----------------------------------------- recruitment/views.py | 1 - 2 files changed, 2453 deletions(-) diff --git a/recruitment/models.py b/recruitment/models.py index 3ddfea0..0f7cc7e 100644 --- a/recruitment/models.py +++ b/recruitment/models.py @@ -2381,2455 +2381,3 @@ class Document(Base): if self.file: return self.file.name.split(".")[-1].upper() return "" - -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 AbstractUser -from django.contrib.auth import get_user_model -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 -from django.contrib.auth.models import AbstractUser -from django.contrib.contenttypes.fields import GenericForeignKey -from django.contrib.contenttypes.models import ContentType - - -class CustomUser(AbstractUser): - """Custom user model extending AbstractUser""" - - USER_TYPES = [ - ("staff", _("Staff")), - ("agency", _("Agency")), - ("candidate", _("Candidate")), - ] - - user_type = models.CharField( - max_length=20, choices=USER_TYPES, default="staff", verbose_name=_("User Type") - ) - phone = models.CharField( - max_length=20, blank=True, null=True, verbose_name=_("Phone") - ) - - class Meta: - verbose_name = _("User") - verbose_name_plural = _("Users") - - -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 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) - 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"), - ) - - 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.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_candidates(self): - return self.applications.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.applications.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 Person(Base): - """Model to store personal information that can be reused across multiple applications""" - - GENDER_CHOICES = [ - ("M", _("Male")), - ("F", _("Female")), - ("O", _("Other")), - ("P", _("Prefer not to say")), - ] - - # Personal Information - first_name = models.CharField(max_length=255, verbose_name=_("First Name")) - 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"), - help_text=_("Unique email address for the person") - ) - phone = models.CharField(max_length=20, blank=True, null=True, verbose_name=_("Phone")) - 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") - ) - 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.SET_NULL, - 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") - ) - 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"]), - ] - - 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) - @property - def belong_to_an_agency(self): - if self.agency: - return True - else: - return False - - -class Application(Base): - """Model to store job-specific application data""" - - class Stage(models.TextChoices): - APPLIED = "Applied", _("Applied") - EXAM = "Exam", _("Exam") - INTERVIEW = "Interview", _("Interview") - 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": ["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"), - ) - 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"), - ) - - # Optional linking to user account (for candidate portal access) - # user = models.OneToOneField( - # User, - # on_delete=models.SET_NULL, - # related_name="application_profile", - # verbose_name=_("User Account"), - # null=True, - # blank=True, - # ) - - 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"]), - ] - unique_together = [["person", "job"]] # Prevent duplicate applications - - def __str__(self): - return f"{self.person.full_name} - {self.job.title}" - - # ==================================================================== - # ✨ PROPERTIES (GETTERS) - Migrated from Candidate - # ==================================================================== - @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.""" - return self.analysis_data.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.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") - - @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", []) - - @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.""" - return self.analysis_data.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 get_latest_meeting(self): - # """Legacy compatibility - get latest meeting for this application""" - # #get parent interview location modal: - - # schedule=self.scheduled_interviews.order_by("-created_at").first() - - - # if schedule: - # print(schedule) - # interview_location=schedule.interview_location - # else: - # return None - # if interview_location and interview_location.location_type=='Remote': - # meeting = interview_location.zoommeetingdetails - - # return meeting - # else: - # meeting = interview_location.onsitelocationdetails - # return meeting - @property - def get_latest_meeting(self): - """ - Retrieves the most specific location details (subclass instance) - of the latest ScheduledInterview for this application, or None. - """ - # 1. Get the latest ScheduledInterview - schedule = self.scheduled_interviews.order_by("-created_at").first() - - # Check if a schedule exists and if it has an interview location - if not schedule or not schedule.interview_location: - return None - - # Get the base location instance - interview_location = schedule.interview_location - - # 2. Safely retrieve the specific subclass details - - # Determine the expected subclass accessor name based on the location_type - if interview_location.location_type == 'Remote': - accessor_name = 'zoommeetingdetails' - else: # Assumes 'Onsite' or any other type defaults to Onsite - accessor_name = 'onsitelocationdetails' - - # Use getattr to safely retrieve the specific meeting object (subclass instance). - # If the accessor exists but points to None (because the subclass record was deleted), - # or if the accessor name is wrong for the object's true type, it will return None. - meeting_details = getattr(interview_location, accessor_name, None) - - return meeting_details - - - @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) - -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 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): - 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(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 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"), - ) - inteview= models.ForeignKey( - 'InterviewSchedule', - on_delete=models.CASCADE, - related_name="notifications", - null=True, - blank=True, - verbose_name=_("Related Interview"), - ) - 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}" - - -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, - null=True, - blank=True, - 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 clean(self): - """Validate message constraints""" - super().clean() - - # For job-related messages, ensure recipient is assigned to the job - if self.job and not self.recipient: - if self.job.assigned_to: - self.recipient = self.job.assigned_to - else: - raise ValidationError(_("Job is not assigned to any user. Please assign the job first.")) - - # Validate sender can message this recipient based on user types - # if self.sender and self.recipient: - # self._validate_messaging_permissions() - - 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, user=self.sender).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"), - ) - object_id = models.PositiveIntegerField( - verbose_name=_("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"), - ) - 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"), - ) - - class Meta: - verbose_name = _("Document") - verbose_name_plural = _("Documents") - ordering = ["-created_at"] - indexes = [ - models.Index(fields=["content_type", "object_id", "document_type", "created_at"]), - ] - - 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 InterviewLocation(Base): - """ - Base model for all interview location/meeting details (remote or onsite) - using Multi-Table Inheritance. - """ - class LocationType(models.TextChoices): - REMOTE = 'Remote', _('Remote (e.g., Zoom, Google Meet)') - ONSITE = 'Onsite', _('In-Person (Physical Location)') - - class Status(models.TextChoices): - """Defines the possible real-time statuses for any interview location/meeting.""" - WAITING = "waiting", _("Waiting") - STARTED = "started", _("Started") - ENDED = "ended", _("Ended") - CANCELLED = "cancelled", _("Cancelled") - - location_type = models.CharField( - max_length=10, - choices=LocationType.choices, - verbose_name=_("Location Type"), - db_index=True - ) - - details_url = models.URLField( - verbose_name=_("Meeting/Location URL"), - max_length=2048, - blank=True, - null=True - ) - - topic = models.CharField( # Renamed from 'description' to 'topic' to match your input - max_length=255, - verbose_name=_("Location/Meeting Topic"), - blank=True, - help_text=_("e.g., 'Zoom Topic: Software Interview' or 'Main Conference Room, 3rd Floor'") - ) - - timezone = models.CharField( - max_length=50, - verbose_name=_("Timezone"), - default='UTC' - ) - - def __str__(self): - # Use 'topic' instead of 'description' - return f"{self.get_location_type_display()} - {self.topic[:50]}" - - class Meta: - verbose_name = _("Interview Location") - verbose_name_plural = _("Interview Locations") - - -class ZoomMeetingDetails(InterviewLocation): - """Concrete model for remote interviews (Zoom specifics).""" - - status = models.CharField( - db_index=True, - max_length=20, - choices=InterviewLocation.Status.choices, - default=InterviewLocation.Status.WAITING, - ) - start_time = models.DateTimeField( - db_index=True, verbose_name=_("Start Time") - ) - duration = models.PositiveIntegerField( - verbose_name=_("Duration (minutes)") - ) - meeting_id = models.CharField( - db_index=True, - max_length=50, - unique=True, - verbose_name=_("External Meeting ID") - ) - password = models.CharField( - max_length=20, blank=True, null=True, verbose_name=_("Password") - ) - zoom_gateway_response = models.JSONField( - blank=True, null=True, verbose_name=_("Zoom Gateway Response") - ) - participant_video = models.BooleanField( - default=True, verbose_name=_("Participant Video") - ) - join_before_host = models.BooleanField( - default=False, verbose_name=_("Join Before Host") - ) - - host_email=models.CharField(null=True,blank=True) - mute_upon_entry = models.BooleanField( - default=False, verbose_name=_("Mute Upon Entry") - ) - waiting_room = models.BooleanField(default=False, verbose_name=_("Waiting Room")) - - # *** REVERTED TO @classmethod (Factory Method) for cleaner instantiation *** - # @classmethod - # def create(cls, **kwargs): - # """Factory method to ensure location_type is set to REMOTE.""" - # return cls(location_type=InterviewLocation.LocationType.REMOTE, **kwargs) - - class Meta: - verbose_name = _("Zoom Meeting Details") - verbose_name_plural = _("Zoom Meeting Details") - - -class OnsiteLocationDetails(InterviewLocation): - """Concrete model for onsite interviews (Room/Address specifics).""" - - physical_address = models.CharField( - max_length=255, - verbose_name=_("Physical Address"), - blank=True, - null=True - ) - room_number = models.CharField( - max_length=50, - verbose_name=_("Room Number/Name"), - blank=True, - null=True - ) - start_time = models.DateTimeField( - db_index=True, verbose_name=_("Start Time") - ) - duration = models.PositiveIntegerField( - verbose_name=_("Duration (minutes)") - ) - status = models.CharField( - db_index=True, - max_length=20, - choices=InterviewLocation.Status.choices, - default=InterviewLocation.Status.WAITING, - ) - - # *** REVERTED TO @classmethod (Factory Method) for cleaner instantiation *** - # @classmethod - # def create(cls, **kwargs): - # """Factory method to ensure location_type is set to ONSITE.""" - # return cls(location_type=InterviewLocation.LocationType.ONSITE, **kwargs) - - class Meta: - verbose_name = _("Onsite Location Details") - verbose_name_plural = _("Onsite Location Details") - - -# --- 2. Scheduling Models --- - -class InterviewSchedule(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. - template_location = models.ForeignKey( - InterviewLocation, - on_delete=models.SET_NULL, - related_name="schedule_templates", - null=True, - blank=True, - verbose_name=_("Location Template (Zoom/Onsite)") - ) - - # NOTE: schedule_interview_type field is needed in the form, - # but not on the model itself if we use template_location. - # If you want to keep it: - schedule_interview_type = models.CharField( - max_length=10, - choices=InterviewLocation.LocationType.choices, - verbose_name=_("Interview Type"), - default=InterviewLocation.LocationType.REMOTE - ) - - 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") - ) - - 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 - ) - - def __str__(self): - return f"Schedule for {self.job.title}" - - -class ScheduledInterview(Base): - """Stores individual scheduled interviews (whether bulk or individually created).""" - - class InterviewStatus(models.TextChoices): - SCHEDULED = "scheduled", _("Scheduled") - CONFIRMED = "confirmed", _("Confirmed") - CANCELLED = "cancelled", _("Cancelled") - COMPLETED = "completed", _("Completed") - - 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_location = models.OneToOneField( - InterviewLocation, - on_delete=models.SET_NULL, - related_name="scheduled_interview", - null=True, - blank=True, - db_index=True, - verbose_name=_("Meeting/Location Details") - ) - - # Link back to the bulk schedule template (optional if individually created) - schedule = models.ForeignKey( - InterviewSchedule, - 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")) - - status = models.CharField( - db_index=True, - max_length=20, - choices=InterviewStatus.choices, - default=InterviewStatus.SCHEDULED, - ) - - 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"]), - ] - - -# --- 3. Interview Notes Model (Fixed) --- - -class InterviewNote(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') - - 1 - interview = models.ForeignKey( - ScheduledInterview, - on_delete=models.CASCADE, - related_name="notes", - verbose_name=_("Scheduled Interview"), - db_index=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()} on {self.interview.id}" \ No newline at end of file diff --git a/recruitment/views.py b/recruitment/views.py index 67efa73..65b5ceb 100644 --- a/recruitment/views.py +++ b/recruitment/views.py @@ -5784,7 +5784,6 @@ class MeetingListView(ListView): paginate_by = 100 def get_queryset(self): - print("hoo") # Start with a base queryset, ensuring an InterviewLocation link exists. queryset = super().get_queryset().filter(interview_location__isnull=False).select_related( 'interview_location',