from django.db import models from django.urls import reverse from typing import List,Dict,Any from django.utils import timezone from django.db.models import FloatField,CharField,IntegerField from django.db.models.functions import Cast,Coalesce from django.db.models import F from django.contrib.auth.models import User from django.core.validators import URLValidator from django_countries.fields import CountryField from django.core.exceptions import ValidationError from django_ckeditor_5.fields import CKEditor5Field from django.utils.html import strip_tags from django.utils.translation import gettext_lazy as _ from django_extensions.db.fields import RandomCharField from .validators import validate_hash_tags, validate_image_size class Base(models.Model): created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Created at")) updated_at = models.DateTimeField(auto_now=True, verbose_name=_("Updated at")) slug = RandomCharField( length=8, unique=True, editable=False, verbose_name=_("Slug") ) class Meta: abstract = True class Profile(models.Model): profile_image = models.ImageField(null=True, blank=True, upload_to="profile_pic/",validators=[validate_image_size]) 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"), ] # 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) published_at = models.DateTimeField(db_index=True, null=True, blank=True) # Added index # University Specific Fields position_number = models.CharField( max_length=50, blank=True, help_text="University position number" ) reporting_to = models.CharField( max_length=100, blank=True, help_text="Who this position reports to" ) open_positions = models.PositiveIntegerField( default=1, help_text="Number of open positions for this job" ) source = models.ForeignKey( "Source", on_delete=models.SET_NULL, # Recommended: If a source is deleted, job's source is set to NULL related_name="job_postings", null=True, blank=True, help_text="The system or channel from which this job posting originated or was first published.", db_index=True # Explicitly index ForeignKey ) max_applications = models.PositiveIntegerField( default=1000, help_text="Maximum number of applications allowed", null=True, blank=True ) hiring_agency = models.ManyToManyField( "HiringAgency", blank=True, related_name="jobs", verbose_name=_("Hiring Agency"), help_text=_( "External agency responsible for sourcing candidates for this role" ), ) cancel_reason = models.TextField( blank=True, help_text=_("Reason for canceling the job posting"), verbose_name=_("Cancel Reason"), ) cancelled_by = models.CharField( max_length=100, blank=True, help_text=_("Name of person who cancelled this job"), verbose_name=_("Cancelled By"), ) cancelled_at = models.DateTimeField(null=True, blank=True) class Meta: ordering = ["-created_at"] verbose_name = "Job Posting" verbose_name_plural = "Job Postings" indexes = [ models.Index(fields=['status', 'created_at', 'title']), models.Index(fields=['slug']), ] def __str__(self): return f"{self.title} - {self.internal_job_id}-{self.get_status_display()}" def get_source(self): return self.source.name if self.source else "System" def save(self, *args, **kwargs): # from django.db import transaction # Generate unique internal job ID if not exists # with transaction.atomic(): # if not self.internal_job_id: # prefix = "KAAUH" # year = timezone.now().year # # Get next sequential number # last_job = ( # JobPosting.objects.select_for_update().filter( # internal_job_id__startswith=f"{prefix}-{year}-" # ) # .order_by("internal_job_id") # .last() # ) # if last_job: # last_num = int(last_job.internal_job_id.split("-")[-1]) # next_num = last_num + 1 # else: # next_num = 1 # self.internal_job_id = f"{prefix}-{year}-{next_num:06d}" super().save(*args, **kwargs) def get_location_display(self): """Return formatted location string""" parts = [] if self.location_city: parts.append(self.location_city) if self.location_state: parts.append(self.location_state) if self.location_country: parts.append(self.location_country) return ", ".join(parts) if parts else "Not specified" @property def is_expired(self): """Check if application deadline has passed""" if self.application_deadline: return self.application_deadline < timezone.now().date() return False def publish(self): self.status = "PUBLISHED" self.published_at = timezone.now() self.application_url = reverse( "form_wizard", kwargs={"slug": self.form_template.slug} ) self.save() def _check_content(self, field_value): """Helper to check if a field contains meaningful content.""" if not field_value: return False # 1. Replace the common HTML non-breaking space entity with a standard space. content = field_value.replace(' ', ' ') # 2. Remove all HTML tags (leaving only text and remaining spaces). stripped = strip_tags(content) # 3. Use .strip() to remove ALL leading/trailing whitespace, including the ones from step 1. final_content = stripped.strip() # Return True if any content remains after stripping tags and spaces. return bool(final_content) @property def has_description_content(self): """Returns True if the description field has meaningful content.""" return self._check_content(self.description) @property def has_qualifications_content(self): """Returns True if the qualifications field has meaningful content.""" return self._check_content(self.qualifications) # Add similar properties for benefits and application_instructions @property def has_benefits_content(self): return self._check_content(self.benefits) @property def has_application_instructions_content(self): return self._check_content(self.application_instructions) @property def current_applications_count(self): """Returns the current number of candidates associated with this job.""" return self.candidates.count() @property def is_application_limit_reached(self): """Checks if the current application count meets or exceeds the max limit.""" if self.max_applications == 0: return True return self.current_applications_count >= self.max_applications @property def all_candidates(self): return self.candidates.annotate( sortable_score=Coalesce(Cast('ai_analysis_data__analysis_data__match_score', output_field=IntegerField()), 0)).order_by('-sortable_score') @property def screening_candidates(self): return self.all_candidates.filter(stage="Applied") @property def exam_candidates(self): return self.all_candidates.filter(stage="Exam") @property def interview_candidates(self): return self.all_candidates.filter(stage="Interview") @property def offer_candidates(self): return self.all_candidates.filter(stage="Offer") # 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() @property def screening_candidates_count(self): return self.all_candidates.filter(stage="Applied").count() @property def exam_candidates_count(self): return self.all_candidates.filter(stage="Exam").count() @property def interview_candidates_count(self): return self.all_candidates.filter(stage="Interview").count() @property def offer_candidates_count(self): return self.all_candidates.filter(stage="Offer").count() class JobPostingImage(models.Model): job=models.OneToOneField('JobPosting',on_delete=models.CASCADE,related_name='post_images') post_image = models.ImageField(upload_to='post/',validators=[validate_image_size]) class Candidate(Base): class Stage(models.TextChoices): APPLIED = "Applied", _("Applied") EXAM = "Exam", _("Exam") INTERVIEW = "Interview", _("Interview") OFFER = "Offer", _("Offer") class ExamStatus(models.TextChoices): PASSED = "Passed", _("Passed") FAILED = "Failed", _("Failed") class Status(models.TextChoices): ACCEPTED = "Accepted", _("Accepted") REJECTED = "Rejected", _("Rejected") class ApplicantType(models.TextChoices): APPLICANT = "Applicant", _("Applicant") CANDIDATE = "Candidate", _("Candidate") # Stage transition validation constants STAGE_SEQUENCE = { "Applied": ["Exam", "Interview", "Offer"], "Exam": ["Interview", "Offer"], "Interview": ["Offer"], "Offer": [], # Final stage - no further transitions } job = models.ForeignKey( JobPosting, on_delete=models.CASCADE, related_name="candidates", verbose_name=_("Job"), ) first_name = models.CharField(max_length=255, verbose_name=_("First Name")) last_name = models.CharField(max_length=255, verbose_name=_("Last Name")) email = models.EmailField(db_index=True, verbose_name=_("Email")) # Added index phone = models.CharField(max_length=20, verbose_name=_("Phone")) address = models.TextField(max_length=200, verbose_name=_("Address")) resume = models.FileField(upload_to="resumes/", verbose_name=_("Resume")) is_resume_parsed = models.BooleanField( default=False, verbose_name=_("Resume Parsed") ) is_potential_candidate = models.BooleanField( default=False, verbose_name=_("Potential Candidate") ) parsed_summary = models.TextField(blank=True, verbose_name=_("Parsed Summary")) applied = models.BooleanField(default=False, verbose_name=_("Applied")) stage = models.CharField( db_index=True, max_length=100, # Added index default="Applied", choices=Stage.choices, verbose_name=_("Stage"), ) applicant_status = models.CharField( choices=ApplicantType.choices, default="Applicant", max_length=100, null=True, blank=True, verbose_name=_("Applicant Status"), ) exam_date = models.DateTimeField(null=True, blank=True, verbose_name=_("Exam Date")) exam_status = models.CharField( choices=ExamStatus.choices, max_length=100, null=True, blank=True, verbose_name=_("Exam Status"), ) interview_date = models.DateTimeField( null=True, blank=True, verbose_name=_("Interview Date") ) interview_status = models.CharField( choices=ExamStatus.choices, max_length=100, null=True, blank=True, verbose_name=_("Interview Status"), ) offer_date = models.DateField(null=True, blank=True, verbose_name=_("Offer Date")) offer_status = models.CharField( choices=Status.choices, max_length=100, null=True, blank=True, verbose_name=_("Offer Status"), ) join_date = models.DateField(null=True, blank=True, verbose_name=_("Join Date")) ai_analysis_data = models.JSONField( verbose_name="AI Analysis Data", default=dict, help_text="Full JSON output from the resume scoring model." ) # Scoring fields (populated by signal) # match_score = models.IntegerField(db_index=True, null=True, blank=True) # Added index # strengths = models.TextField(blank=True) # weaknesses = models.TextField(blank=True) # criteria_checklist = models.JSONField(default=dict, blank=True) # major_category_name = models.TextField(db_index=True, blank=True, verbose_name=_("Major Category Name")) # Added index # recommendation = models.TextField(blank=True, verbose_name=_("Recommendation")) submitted_by_agency = models.ForeignKey( "HiringAgency", on_delete=models.SET_NULL, null=True, blank=True, related_name="submitted_candidates", verbose_name=_("Submitted by Agency"), ) retry = models.SmallIntegerField(verbose_name="Resume Parsing Retry",default=3) class Meta: verbose_name = _("Candidate") verbose_name_plural = _("Candidates") indexes = [ models.Index(fields=['stage']), models.Index(fields=['created_at']), ] def set_field(self, key: str, value: Any): """ Generic method to set any single key-value pair and save. """ self.ai_analysis_data[key] = value # self.save(update_fields=['ai_analysis_data']) # ==================================================================== # ✨ PROPERTIES (GETTERS) # ==================================================================== @property def resume_data(self): return self.ai_analysis_data.get('resume_data', {}) @property def analysis_data(self): return self.ai_analysis_data.get('analysis_data', {}) @property def match_score(self) -> int: """1. A score from 0 to 100 representing how well the candidate fits the role.""" return self.analysis_data.get('match_score', 0) @property def years_of_experience(self) -> float: """4. The total number of years of professional experience as a numerical value.""" return self.analysis_data.get('years_of_experience', 0.0) @property def soft_skills_score(self) -> int: """15. A score (0-100) for inferred non-technical skills.""" return self.analysis_data.get('soft_skills_score', 0) @property def industry_match_score(self) -> int: """16. A score (0-100) for the relevance of the candidate's industry experience.""" # Renamed to clarify: experience_industry_match return self.analysis_data.get('experience_industry_match', 0) # --- Properties for Funnel & Screening Efficiency --- @property def min_requirements_met(self) -> bool: """14. Boolean (true/false) indicating if all non-negotiable minimum requirements are met.""" return self.analysis_data.get('min_req_met_bool', False) @property def screening_stage_rating(self) -> str: """13. A standardized rating (e.g., "A - Highly Qualified", "B - Qualified").""" return self.analysis_data.get('screening_stage_rating', 'N/A') @property def top_3_keywords(self) -> List[str]: """10. A list of the three most dominant and relevant technical skills or technologies.""" return self.analysis_data.get('top_3_keywords', []) @property def most_recent_job_title(self) -> str: """8. The candidate's most recent or current professional job title.""" return self.analysis_data.get('most_recent_job_title', 'N/A') # --- Properties for Structured Detail --- @property def criteria_checklist(self) -> Dict[str, str]: """5 & 6. An object rating the candidate's match for each specific criterion.""" return self.analysis_data.get('criteria_checklist', {}) @property def professional_category(self) -> str: """7. The most fitting professional field or category for the individual.""" return self.analysis_data.get('category', 'N/A') @property def language_fluency(self) -> List[Dict[str, str]]: """12. A list of languages and their fluency levels mentioned.""" return self.analysis_data.get('language_fluency', []) # --- Properties for Summaries and Narrative --- @property def strengths(self) -> str: """2. A brief summary of why the candidate is a strong fit.""" return self.analysis_data.get('strengths', '') @property def weaknesses(self) -> str: """3. A brief summary of where the candidate falls short or what criteria are missing.""" return self.analysis_data.get('weaknesses', '') @property def job_fit_narrative(self) -> str: """11. A single, concise sentence summarizing the core fit.""" return self.analysis_data.get('job_fit_narrative', '') @property def recommendation(self) -> str: """9. Provide a detailed final recommendation for the candidate.""" # Using a more descriptive name to avoid conflict with potential built-in methods return self.analysis_data.get('recommendation', '') @property def name(self): return f"{self.first_name} {self.last_name}" @property def full_name(self): return self.name @property def get_file_size(self): if self.resume: return self.resume.size return 0 def save(self, *args, **kwargs): """Override save to ensure validation is called""" self.clean() # Call validation before saving super().save(*args, **kwargs) def get_available_stages(self): """Get list of stages this candidate can transition to""" if not self.pk: # New record return ["Applied"] old_stage = self.__class__.objects.get(pk=self.pk).stage return self.STAGE_SEQUENCE.get(old_stage, []) @property def submission(self): return FormSubmission.objects.filter(template__job=self.job).first() @property def responses(self): if self.submission: return self.submission.responses.all() return [] def __str__(self): return self.full_name @property def get_meetings(self): return self.scheduled_interviews.all() @property def get_latest_meeting(self): schedule = self.scheduled_interviews.order_by('-created_at').first() if schedule: return schedule.zoom_meeting return None @property def has_future_meeting(self): """ Checks if the candidate has any scheduled interviews for a future date/time. """ # Ensure timezone.now() is used for comparison now = timezone.now() # Check if any related ScheduledInterview has a future interview_date and interview_time # We need to combine date and time for a proper datetime comparison if they are separate fields future_meetings = self.scheduled_interviews.filter( interview_date__gt=now.date() ).filter( interview_time__gte=now.time() ).exists() # Also check for interviews happening later today today_future_meetings = self.scheduled_interviews.filter( interview_date=now.date(), interview_time__gte=now.time() ).exists() return future_meetings or today_future_meetings class TrainingMaterial(Base): title = models.CharField(max_length=255, verbose_name=_("Title")) content = CKEditor5Field(blank=True, verbose_name=_("Content"),config_name='extends') video_link = models.URLField(blank=True, verbose_name=_("Video Link")) file = models.FileField( upload_to="training_materials/", blank=True, verbose_name=_("File") ) created_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, verbose_name=_("Created by") ) class Meta: verbose_name = _("Training Material") verbose_name_plural = _("Training Materials") def __str__(self): return self.title class ZoomMeeting(Base): class MeetingStatus(models.TextChoices): WAITING = "waiting", _("Waiting") STARTED = "started", _("Started") ENDED = "ended", _("Ended") CANCELLED = "cancelled",_("Cancelled") # Basic meeting details topic = models.CharField(max_length=255, verbose_name=_("Topic")) meeting_id = models.CharField( db_index=True, max_length=20, unique=True, verbose_name=_("Meeting ID") # Added index ) # Unique identifier for the meeting start_time = models.DateTimeField(db_index=True, verbose_name=_("Start Time")) # Added index duration = models.PositiveIntegerField( verbose_name=_("Duration") ) # Duration in minutes timezone = models.CharField(max_length=50, verbose_name=_("Timezone")) join_url = models.URLField( verbose_name=_("Join URL") ) # URL for participants to join participant_video = models.BooleanField( default=True, verbose_name=_("Participant Video") ) password = models.CharField( max_length=20, blank=True, null=True, verbose_name=_("Password") ) join_before_host = models.BooleanField( default=False, verbose_name=_("Join Before Host") ) mute_upon_entry = models.BooleanField( default=False, verbose_name=_("Mute Upon Entry") ) waiting_room = models.BooleanField(default=False, verbose_name=_("Waiting Room")) zoom_gateway_response = models.JSONField( blank=True, null=True, verbose_name=_("Zoom Gateway Response") ) status = models.CharField( db_index=True, max_length=20, # Added index null=True, blank=True, verbose_name=_("Status"), default=MeetingStatus.WAITING, ) # Timestamps def __str__(self): return self.topic class MeetingComment(Base): """ Model for storing meeting comments/notes """ meeting = models.ForeignKey( ZoomMeeting, on_delete=models.CASCADE, related_name="comments", verbose_name=_("Meeting") ) author = models.ForeignKey( User, on_delete=models.CASCADE, related_name="meeting_comments", verbose_name=_("Author") ) content = CKEditor5Field( verbose_name=_("Content"), config_name='extends' ) # Inherited from Base: created_at, updated_at, slug class Meta: verbose_name = _("Meeting Comment") verbose_name_plural = _("Meeting Comments") ordering = ['-created_at'] def __str__(self): return f"Comment by {self.author.get_username()} on {self.meeting.topic}" class FormTemplate(Base): """ Represents a complete form template with multiple stages """ job = models.OneToOneField( JobPosting, on_delete=models.CASCADE, related_name="form_template", db_index=True ) name = models.CharField(max_length=200, help_text="Name of the form template") description = models.TextField( blank=True, help_text="Description of the form template" ) created_by = models.ForeignKey( User, on_delete=models.CASCADE, related_name="form_templates",null=True,blank=True, db_index=True ) is_active = models.BooleanField( default=False, help_text="Whether this template is active" ) class Meta: ordering = ["-created_at"] verbose_name = "Form Template" verbose_name_plural = "Form Templates" indexes = [ models.Index(fields=['created_at']), models.Index(fields=['is_active']), ] def __str__(self): return self.name def get_stage_count(self): return self.stages.count() def get_field_count(self): return sum(stage.fields.count() for stage in self.stages.all()) class FormStage(Base): """ Represents a stage/section within a form template """ template = models.ForeignKey( FormTemplate, on_delete=models.CASCADE, related_name="stages", db_index=True ) name = models.CharField(max_length=200, help_text="Name of the stage") order = models.PositiveIntegerField( default=0, help_text="Order of the stage in the form" ) is_predefined = models.BooleanField( default=False, help_text="Whether this is a default resume stage" ) class Meta: ordering = ["order"] verbose_name = "Form Stage" verbose_name_plural = "Form Stages" def __str__(self): return f"{self.template.name} - {self.name}" def clean(self): if self.order < 0: raise ValidationError("Order must be a positive integer") class FormField(Base): """ Represents a single field within a form stage """ FIELD_TYPES = [ ("text", "Text Input"), ("email", "Email"), ("phone", "Phone"), ("textarea", "Text Area"), ("file", "File Upload"), ("date", "Date Picker"), ("select", "Dropdown"), ("radio", "Radio Buttons"), ("checkbox", "Checkboxes"), ] stage = models.ForeignKey( FormStage, on_delete=models.CASCADE, related_name="fields", db_index=True ) label = models.CharField(max_length=200, help_text="Label for the field") field_type = models.CharField( max_length=20, choices=FIELD_TYPES, help_text="Type of the field" ) placeholder = models.CharField( max_length=200, blank=True, help_text="Placeholder text" ) required = models.BooleanField( default=False, help_text="Whether the field is required" ) order = models.PositiveIntegerField( default=0, help_text="Order of the field in the stage" ) is_predefined = models.BooleanField( default=False, help_text="Whether this is a default field" ) # For selection fields (select, radio, checkbox) options = models.JSONField( default=list, blank=True, help_text="Options for selection fields (stored as JSON array)", ) # For file upload fields file_types = models.CharField( max_length=200, blank=True, help_text="Allowed file types (comma-separated, e.g., '.pdf,.doc,.docx')", ) max_file_size = models.PositiveIntegerField( default=5, help_text="Maximum file size in MB (default: 5MB)" ) multiple_files = models.BooleanField( default=False, help_text="Allow multiple files to be uploaded" ) max_files = models.PositiveIntegerField( default=1, help_text="Maximum number of files allowed (when multiple_files is True)", ) class Meta: ordering = ["order"] verbose_name = "Form Field" verbose_name_plural = "Form Fields" def clean(self): # Validate options for selection fields if self.field_type in ["select", "radio", "checkbox"]: if not isinstance(self.options, list): raise ValidationError("Options must be a list for selection fields") else: # Clear options for non-selection fields if self.options: self.options = [] # Validate file settings for file fields if self.field_type == "file": if not self.file_types: self.file_types = ".pdf,.doc,.docx" if self.max_file_size <= 0: raise ValidationError("Max file size must be greater than 0") if self.multiple_files and self.max_files <= 0: raise ValidationError( "Max files must be greater than 0 when multiple files are allowed" ) if not self.multiple_files: self.max_files = 1 else: # Clear file settings for non-file fields self.file_types = "" self.max_file_size = 0 self.multiple_files = False self.max_files = 1 # Validate order if self.order < 0: raise ValidationError("Order must be a positive integer") def __str__(self): return f"{self.stage.template.name} - {self.stage.name} - {self.label}" class FormSubmission(Base): """ Represents a completed form submission by an applicant """ template = models.ForeignKey( FormTemplate, on_delete=models.CASCADE, related_name="submissions", db_index=True ) submitted_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name="form_submissions", db_index=True ) submitted_at = models.DateTimeField(db_index=True, auto_now_add=True) # Added index applicant_name = models.CharField( max_length=200, blank=True, help_text="Name of the applicant" ) applicant_email = models.EmailField(db_index=True, blank=True, help_text="Email of the applicant") # Added index class Meta: ordering = ["-submitted_at"] verbose_name = "Form Submission" verbose_name_plural = "Form Submissions" indexes = [ models.Index(fields=['submitted_at']), ] def __str__(self): return f"Submission for {self.template.name} - {self.submitted_at.strftime('%Y-%m-%d %H:%M')}" class FieldResponse(Base): """ Represents a response to a specific field in a form submission """ submission = models.ForeignKey( FormSubmission, on_delete=models.CASCADE, related_name="responses", db_index=True ) field = models.ForeignKey( FormField, on_delete=models.CASCADE, related_name="responses", db_index=True ) # Store the response value as JSON to handle different data types value = models.JSONField( null=True, blank=True, help_text="Response value (stored as JSON)" ) # For file uploads, store the file path uploaded_file = models.FileField(upload_to="form_uploads/", null=True, blank=True) class Meta: verbose_name = "Field Response" verbose_name_plural = "Field Responses" indexes = [ models.Index(fields=['submission']), models.Index(fields=['field']), ] def __str__(self): return f"Response to {self.field.label} in {self.submission}" @property def is_file(self): if self.uploaded_file: return True return False @property def get_file(self): if self.is_file: return self.uploaded_file return None @property def get_file_size(self): if self.is_file: return self.uploaded_file.size return 0 @property def display_value(self): """Return a human-readable representation of the response value""" if self.is_file: return f"File: {self.uploaded_file.name}" elif self.value is None: return "" elif isinstance(self.value, list): return ", ".join(str(v) for v in self.value) else: return str(self.value) # Optional: Create a model for form templates that can be shared across organizations class SharedFormTemplate(Base): """ Represents a form template that can be shared across different organizations/users """ template = models.OneToOneField(FormTemplate, on_delete=models.CASCADE) is_public = models.BooleanField( default=False, help_text="Whether this template is publicly available" ) shared_with = models.ManyToManyField( User, blank=True, related_name="shared_templates" ) class Meta: verbose_name = "Shared Form Template" verbose_name_plural = "Shared Form Templates" def __str__(self): return f"Shared: {self.template.name}" class Source(Base): name = models.CharField( max_length=100, unique=True, verbose_name=_("Source Name"), help_text=_("e.g., ATS, ERP "), ) source_type = models.CharField( max_length=100, verbose_name=_("Source Type"), help_text=_("e.g., ATS, ERP ") ) description = models.TextField( blank=True, verbose_name=_("Description"), help_text=_("A description of the source"), ) ip_address = models.GenericIPAddressField( blank=True, null=True, verbose_name=_("IP Address"), help_text=_("The IP address of the source"), ) created_at = models.DateTimeField(auto_now_add=True) # Integration specific fields api_key = models.CharField( max_length=255, blank=True, null=True, verbose_name=_("API Key"), help_text=_("API key for authentication (will be encrypted)"), ) api_secret = models.CharField( max_length=255, blank=True, null=True, verbose_name=_("API Secret"), help_text=_("API secret for authentication (will be encrypted)"), ) trusted_ips = models.TextField( blank=True, null=True, verbose_name=_("Trusted IP Addresses"), help_text=_("Comma-separated list of trusted IP addresses"), ) is_active = models.BooleanField( default=True, verbose_name=_("Active"), help_text=_("Whether this source is active for integration"), ) integration_version = models.CharField( max_length=50, blank=True, verbose_name=_("Integration Version"), help_text=_("Version of the integration protocol"), ) last_sync_at = models.DateTimeField( null=True, blank=True, verbose_name=_("Last Sync At"), help_text=_("Timestamp of the last successful synchronization"), ) sync_status = models.CharField( max_length=20, blank=True, choices=[ ("IDLE", "Idle"), ("SYNCING", "Syncing"), ("ERROR", "Error"), ("DISABLED", "Disabled"), ], default="IDLE", verbose_name=_("Sync Status"), ) 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=10, blank=True, verbose_name=_("HTTP Method")) request_data = models.JSONField( blank=True, null=True, verbose_name=_("Request Data") ) response_data = models.JSONField( blank=True, null=True, verbose_name=_("Response Data") ) status_code = models.CharField( max_length=10, blank=True, verbose_name=_("Status Code") ) error_message = models.TextField(blank=True, verbose_name=_("Error Message")) ip_address = models.GenericIPAddressField(verbose_name=_("IP Address")) user_agent = models.CharField( max_length=255, blank=True, verbose_name=_("User Agent") ) processing_time = models.FloatField( null=True, blank=True, verbose_name=_("Processing Time (seconds)") ) def __str__(self): return f"{self.source.name} - {self.action} - {self.created_at}" class Meta: ordering = ["-created_at"] verbose_name = _("Integration Log") verbose_name_plural = _("Integration Logs") @property def is_successful(self): """Check if the integration action was successful""" if self.action == self.ActionChoices.ERROR: return False if self.action == self.ActionChoices.REQUEST: return True # Requests are always logged, success depends on response if self.status_code and self.status_code.startswith("2"): return True return False class HiringAgency(Base): name = models.CharField(max_length=200, unique=True, verbose_name=_("Agency Name")) contact_person = models.CharField( max_length=150, blank=True, verbose_name=_("Contact Person") ) email = models.EmailField(blank=True) phone = models.CharField(max_length=20, blank=True) website = models.URLField(blank=True) notes = models.TextField(blank=True, help_text=_("Internal notes about the agency")) country = CountryField(blank=True, null=True, blank_label=_("Select country")) address = models.TextField(blank=True, null=True) def __str__(self): return self.name class Meta: verbose_name = _("Hiring Agency") verbose_name_plural = _("Hiring Agencies") ordering = ["name"] class BreakTime(models.Model): """Model to store break times for a schedule""" start_time = models.TimeField(verbose_name=_("Start Time")) end_time = models.TimeField(verbose_name=_("End Time")) def __str__(self): return f"{self.start_time} - {self.end_time}" class InterviewSchedule(Base): """Stores the scheduling criteria for interviews""" job = models.ForeignKey( JobPosting, on_delete=models.CASCADE, related_name="interview_schedules", db_index=True ) candidates = models.ManyToManyField(Candidate, related_name="interview_schedules", blank=True,null=True) start_date = models.DateField(db_index=True, verbose_name=_("Start Date")) # Added index end_date = models.DateField(db_index=True, verbose_name=_("End Date")) # Added index working_days = models.JSONField( verbose_name=_("Working Days") ) # Store days of week as [0,1,2,3,4] for Mon-Fri start_time = models.TimeField(verbose_name=_("Start Time")) end_time = models.TimeField(verbose_name=_("End Time")) break_start_time = models.TimeField(verbose_name=_("Break Start Time"),null=True,blank=True) break_end_time = models.TimeField(verbose_name=_("Break End Time"),null=True,blank=True) interview_duration = models.PositiveIntegerField( verbose_name=_("Interview Duration (minutes)") ) buffer_time = models.PositiveIntegerField( verbose_name=_("Buffer Time (minutes)"), default=0 ) created_by = models.ForeignKey(User, on_delete=models.CASCADE, db_index=True) # Added index def __str__(self): return f"Interview Schedule for {self.job.title}" class Meta: indexes = [ models.Index(fields=['start_date']), models.Index(fields=['end_date']), models.Index(fields=['created_by']), ] class ScheduledInterview(Base): """Stores individual scheduled interviews""" candidate = models.ForeignKey( Candidate, on_delete=models.CASCADE, related_name="scheduled_interviews", db_index=True ) job = models.ForeignKey( "JobPosting", on_delete=models.CASCADE, related_name="scheduled_interviews", db_index=True ) zoom_meeting = models.OneToOneField( ZoomMeeting, on_delete=models.CASCADE, related_name="interview", db_index=True ) schedule = models.ForeignKey( InterviewSchedule, on_delete=models.CASCADE, related_name="interviews",null=True,blank=True, db_index=True ) interview_date = models.DateField(db_index=True, verbose_name=_("Interview Date")) # Added index interview_time = models.TimeField(verbose_name=_("Interview Time")) status = models.CharField( db_index=True, max_length=20, # Added index choices=[ ("scheduled", _("Scheduled")), ("confirmed", _("Confirmed")), ("cancelled", _("Cancelled")), ("completed", _("Completed")), ], default="scheduled", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) def __str__(self): return f"Interview with {self.candidate.name} for {self.job.title}" class Meta: indexes = [ models.Index(fields=['job', 'status']), models.Index(fields=['interview_date', 'interview_time']), models.Index(fields=['candidate', 'job']), ]