import asyncio import threading from typing import Dict, Any, List, Optional import logging from datetime import datetime import os import pymongo from dotenv import load_dotenv from bson.codec_options import CodecOptions, DatetimeConversion from .job_api_service import JobAPIService from .opensearch_service import OpenSearchService from .embeddings import Embedder from .mongo_service import mongo_service from .ai_service import AIService # Load environment variables load_dotenv() # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class JobMatchingService: """Service for job matching with candidate screening and background processing""" def __init__(self): self.job_api_service = JobAPIService() self.opensearch_service = OpenSearchService() self.embedder = Embedder() self.ai_service = AIService() def process_job_matching_async(self, job_emp_id: str) -> Dict[str, Any]: """ Start background job matching process and return immediately Args: job_emp_id: The employer job ID to process Returns: Dict with success status - actual processing happens in background """ try: # Start background processing in a separate thread thread = threading.Thread( target=self._background_job_matching, args=(job_emp_id,), daemon=True ) thread.start() return { "success": True, "message": f"Job matching process started for job ID: {job_emp_id}", "job_id": job_emp_id, "status": "processing_started" } except Exception as e: logger.error(f"Error starting job matching process: {str(e)}") return { "success": False, "error": str(e), "job_id": job_emp_id } def process_specific_candidates_async(self, job_emp_id: str, jobseeker_ids: List[int]) -> Dict[str, Any]: """ Start background processing for specific candidate analysis Args: job_emp_id: The employer job ID to process jobseeker_ids: List of specific jobseeker IDs to analyze Returns: Dict with success status - actual processing happens in background """ try: # Start background processing in a separate thread thread = threading.Thread( target=self._background_specific_candidates_analysis, args=(job_emp_id, jobseeker_ids), daemon=True ) thread.start() return { "success": True, "message": f"Specific candidate analysis started for job ID: {job_emp_id}", "job_id": job_emp_id, "jobseeker_count": len(jobseeker_ids), "status": "processing_started" } except Exception as e: logger.error(f"Error starting specific candidate analysis: {str(e)}") return { "success": False, "error": str(e), "job_id": job_emp_id } def _background_job_matching(self, job_emp_id: str): """ Background process for job matching - runs in separate thread Args: job_emp_id: The employer job ID to process """ try: logger.info(f"Starting background job matching for job ID: {job_emp_id}") # Step 1: Get job details from job API job_details = self._get_job_details(job_emp_id) if not job_details: logger.error(f"Could not retrieve job details for job ID: {job_emp_id}") return # Step 2: Search candidates using both OpenSearch and Vector search candidates = self._search_candidates(job_details) if not candidates: logger.info(f"No matching candidates found for job ID: {job_emp_id}") return # Step 3: Process each candidate processed_count = 0 for candidate in candidates: try: # Generate AI analysis for the candidate analysis = self._generate_candidate_analysis(job_details, candidate) if analysis: # Only store if matching percentage is 50% or higher if analysis.get("matching_percentage", 0) >= 50.0: # Store screening result self._store_screening_result(job_emp_id, candidate, analysis) processed_count += 1 else: logger.info(f"Candidate {candidate.get('resume_id')} below 50% threshold: {analysis.get('matching_percentage')}%") except Exception as e: logger.error(f"Error processing candidate {candidate.get('resume_id')}: {str(e)}") continue logger.info(f"Job matching completed for job ID: {job_emp_id}. Processed {processed_count} candidates.") except Exception as e: logger.error(f"Error in background job matching for job ID {job_emp_id}: {str(e)}") def _background_specific_candidates_analysis(self, job_emp_id: str, jobseeker_ids: List[int]): """ Background process for specific candidate analysis - runs in separate thread Args: job_emp_id: The employer job ID to process jobseeker_ids: List of specific jobseeker IDs to analyze """ try: logger.info(f"Starting specific candidate analysis for job ID: {job_emp_id}, candidates: {len(jobseeker_ids)}") # Step 1: Get job details from job API job_details = self._get_job_details(job_emp_id) if not job_details: logger.error(f"Could not retrieve job details for job ID: {job_emp_id}") return # For type 1 specific analysis, disable country filtering logger.info(f"Type 1 specific analysis - country filtering disabled for job {job_emp_id}") if job_details.get('country_header_code'): logger.info(f"Original job country: {job_details['country_header_code']} - will be ignored for type 1") job_details['country_header_code'] = None # Remove country filter for type 1 # Step 2: Process each specific jobseeker processed_count = 0 indexed_count = 0 for jobseeker_id in jobseeker_ids: try: # Convert to string for consistent handling jobseeker_id_str = str(jobseeker_id) # Check if jobseeker is indexed, index if needed if not self._ensure_jobseeker_indexed(jobseeker_id_str): logger.warning(f"Could not ensure indexing for jobseeker: {jobseeker_id_str}") continue indexed_count += 1 # Get candidate details candidate_data = self._get_candidate_details(jobseeker_id_str) if not candidate_data: logger.warning(f"Could not get candidate details for jobseeker: {jobseeker_id_str}") continue # Create candidate object for analysis (no scores since this is direct analysis) candidate = { "resume_id": jobseeker_id_str, "keyword_score": 0, "keyword_percentage": 0, "vector_score": 0, "vector_percentage": 0, "combined_score": 0 # Will be set by AI analysis } # Generate AI analysis analysis = self._generate_candidate_analysis(job_details, candidate) if analysis: # Store all screening results for specific candidates (no minimum threshold) self._store_specific_screening_result(job_emp_id, candidate, analysis) processed_count += 1 except Exception as e: logger.error(f"Error processing specific candidate {jobseeker_id}: {str(e)}") continue logger.info(f"Specific candidate analysis completed for job ID: {job_emp_id}. Indexed: {indexed_count}, Processed: {processed_count}") except Exception as e: logger.error(f"Error in background specific candidate analysis for job ID {job_emp_id}: {str(e)}") def _get_job_details(self, job_emp_id: str) -> Optional[Dict[str, Any]]: """ Get job details from job API and extract required fields Args: job_emp_id: The employer job ID Returns: Dictionary with job details or None if failed """ try: result = self.job_api_service.get_employer_job(job_emp_id) if not result.get('success'): logger.error(f"Failed to get job details: {result.get('error')}") return None job_data = result.get('data', {}) # Check if job_data is a list (invalid job ID case) if isinstance(job_data, list): logger.error(f"Invalid job ID {job_emp_id}: API returned empty list instead of job data") return None # Check if job_data is not a dictionary if not isinstance(job_data, dict): logger.error(f"Invalid job ID {job_emp_id}: API returned {type(job_data)} instead of dictionary") return None # Check if data is nested under 'job' key if 'job' in job_data: job_data = job_data['job'] # Extract required fields with multiple possible field names job_details = { 'job_emp_id': job_emp_id, 'job_title': (job_data.get('job_title', '') or job_data.get('title', '') or job_data.get('position', '') or job_data.get('job_position', '')), 'job_desc': (job_data.get('job_description', '') or job_data.get('job_desc', '') or job_data.get('description', '') or job_data.get('job_details', '')), 'skill_list': (job_data.get('skill_list', '') or job_data.get('skills', '') or job_data.get('required_skills', '')), 'skill_list_ar': job_data.get('skill_list_ar', ''), 'workplace_type': (job_data.get('workplace_type', '') or job_data.get('work_type', '') or job_data.get('remote_type', '')), 'job_type': (job_data.get('job_type', '') or job_data.get('employment_type', '') or job_data.get('type', '')), 'country_name': (job_data.get('country_name', '') or job_data.get('country', '') or job_data.get('location', '')), 'country_id': job_data.get('country_id'), 'country_header_code': None # Will be populated below } # Get country_header_code from country_update table if country_id is available if job_details.get('country_id'): try: # Use auto datetime conversion to avoid BSON errors codec_options = CodecOptions(datetime_conversion=DatetimeConversion.DATETIME_AUTO) # Get collection with proper codec options client = mongo_service.client db = client[mongo_service.db.name].with_options(codec_options=codec_options) country_collection = db['country_update'] # Query only string fields to avoid datetime issues country_doc = country_collection.find_one( {'id': job_details['country_id']}, {'id': 1, 'name': 1, 'header_code': 1} ) if country_doc and country_doc.get('header_code'): job_details['country_header_code'] = country_doc['header_code'].lower() logger.info(f"Found country header code: {job_details['country_header_code']} for country_id: {job_details['country_id']}") else: logger.warning(f"No header_code found for country_id: {job_details['country_id']}") except Exception as e: logger.error(f"Error getting country header code: {str(e)}") # No fallback mapping - rely on database data only logger.warning(f"Could not determine country_header_code for country_id: {job_details.get('country_id')}") job_details['country_header_code'] = None # Add logging for final country status if job_details.get('country_header_code'): logger.info(f"Country filtering enabled for job {job_emp_id}: {job_details['country_header_code']}") else: logger.warning(f"No country filtering for job {job_emp_id} - will search all countries") # Validate essential job information is present if not job_details.get('job_title'): logger.error(f"Invalid job ID {job_emp_id}: No job title found in API response") return None logger.info(f"Successfully retrieved job details for job ID: {job_emp_id}") return job_details except Exception as e: logger.error(f"Error getting job details for job ID {job_emp_id}: {str(e)}") return None def _search_candidates_without_country_filter(self, job_details: Dict[str, Any], jobseeker_ids: List[str] = None) -> List[Dict[str, Any]]: """ Search for candidates without country filtering (for type 1 and type 3) Args: job_details: Job details dictionary jobseeker_ids: Optional list of specific jobseeker IDs to analyze Returns: List of matching candidates without country restrictions """ try: # Create search query from job details search_query = self._build_search_query(job_details) logger.info(f"Search query built without country filtering: '{search_query}'") # For type 1 and type 3, search without country filtering logger.info("Country filtering disabled for type 1 and type 3") # If specific jobseeker IDs provided, search only those if jobseeker_ids: logger.info(f"Searching specific {len(jobseeker_ids)} candidates: {jobseeker_ids}") # Get candidates by specific IDs (no country filter) keyword_results = self.opensearch_service.search_by_ids(jobseeker_ids) logger.info(f"OpenSearch returned {len(keyword_results) if keyword_results else 0} specific candidates") else: # Search all candidates (no country filter) keyword_results = self.opensearch_service.search(search_query, top_k=100) logger.info(f"OpenSearch returned {len(keyword_results) if keyword_results else 0} results") # Step 2: Filter OpenSearch results by absolute minimum score (still apply quality filter) absolute_min_score = float(os.getenv("JOB_MATCHING_ABSOLUTE_MIN_SCORE", 0.3)) filtered_keyword_results = [r for r in keyword_results if r.get("keyword_score", 0) >= absolute_min_score] if keyword_results else [] logger.info(f"OpenSearch results above {absolute_min_score} threshold: {len(filtered_keyword_results)}") # Step 3: Use filtered OpenSearch results for Vector search (no country filter) if filtered_keyword_results: # Extract resume IDs from filtered OpenSearch results filtered_resume_ids = [r.get("resume_id") for r in filtered_keyword_results if r.get("resume_id")] logger.info(f"Performing vector search on {len(filtered_resume_ids)} pre-filtered candidates without country restriction") # Search using Vector search without country filter vector_results = self.embedder.search_filtered_candidates(search_query, filtered_resume_ids, top_k=100) logger.info(f"Vector search returned {len(vector_results.get('result', [])) if vector_results else 0} results from filtered candidates") # Step 4: Merge and rank results final_candidates = self._merge_and_rank_results(filtered_keyword_results, vector_results.get('result', []), job_details) return final_candidates else: logger.info("No candidates met minimum quality threshold") return [] except Exception as e: logger.error(f"Error searching candidates without country filter: {str(e)}") return [] def _search_candidates_for_specific_analysis(self, job_details: Dict[str, Any], jobseeker_ids: List[str] = None) -> List[Dict[str, Any]]: """ Search for candidates for specific analysis without country filtering (type 1) Args: job_details: Job details dictionary jobseeker_ids: Optional list of specific jobseeker IDs to analyze Returns: List of matching candidates without country restrictions """ try: # Create search query from job details search_query = self._build_search_query(job_details) logger.info(f"Search query built for specific analysis: '{search_query}'") # For specific analysis, search without country filtering logger.info("Country filtering disabled for specific analysis (type 1)") # If specific jobseeker IDs provided, search only those if jobseeker_ids: logger.info(f"Searching specific {len(jobseeker_ids)} candidates: {jobseeker_ids}") # Get candidates by specific IDs (no country filter) keyword_results = self.opensearch_service.search_by_ids(jobseeker_ids) logger.info(f"OpenSearch returned {len(keyword_results) if keyword_results else 0} specific candidates") else: # Search all candidates (no country filter) keyword_results = self.opensearch_service.search(search_query, top_k=100) logger.info(f"OpenSearch returned {len(keyword_results) if keyword_results else 0} results") # Step 2: Filter OpenSearch results by absolute minimum score (still apply quality filter) absolute_min_score = float(os.getenv("JOB_MATCHING_ABSOLUTE_MIN_SCORE", 0.3)) filtered_keyword_results = [r for r in keyword_results if r.get("keyword_score", 0) >= absolute_min_score] if keyword_results else [] logger.info(f"OpenSearch results above {absolute_min_score} threshold: {len(filtered_keyword_results)}") # Step 3: Use filtered OpenSearch results for Vector search (no country filter) if filtered_keyword_results: # Extract resume IDs from filtered OpenSearch results filtered_resume_ids = [r.get("resume_id") for r in filtered_keyword_results if r.get("resume_id")] logger.info(f"Performing vector search on {len(filtered_resume_ids)} pre-filtered candidates without country restriction") # Search using Vector search without country filter vector_results = self.embedder.search_filtered_candidates(search_query, filtered_resume_ids, top_k=100) logger.info(f"Vector search returned {len(vector_results.get('result', [])) if vector_results else 0} results from filtered candidates") # Step 4: Merge and rank results final_candidates = self._merge_and_rank_results(filtered_keyword_results, vector_results.get('result', []), job_details) return final_candidates else: logger.info("No candidates met minimum quality threshold for specific analysis") return [] except Exception as e: logger.error(f"Error searching candidates for specific analysis: {str(e)}") return [] def _search_candidates(self, job_details: Dict[str, Any]) -> List[Dict[str, Any]]: """ Search for candidates using both OpenSearch and Vector search with country filtering Args: job_details: Job details dictionary Returns: List of matching candidates with combined scores """ try: # Create search query from job details search_query = self._build_search_query(job_details) logger.info(f"Search query built: '{search_query}'") # Get country filter from job details for mandatory country matching (type 2 only) country_header_code = job_details.get('country_header_code') if country_header_code: logger.info(f"Country filtering enabled for screening type 2: {country_header_code}") else: logger.warning("No country_header_code found - search will include all countries") # Step 1: Search using OpenSearch (keyword search) with country filter if country_header_code: keyword_results = self.opensearch_service.search_with_country_filter(search_query, country_header_code, top_k=100) logger.info(f"OpenSearch with country filter ({country_header_code}) returned {len(keyword_results) if keyword_results else 0} results") else: keyword_results = self.opensearch_service.search(search_query, top_k=100) logger.info(f"OpenSearch returned {len(keyword_results) if keyword_results else 0} results") # Step 2: Filter OpenSearch results by absolute minimum score absolute_min_score = float(os.getenv("JOB_MATCHING_ABSOLUTE_MIN_SCORE", 0.3)) filtered_keyword_results = [r for r in keyword_results if r.get("keyword_score", 0) >= absolute_min_score] if keyword_results else [] logger.info(f"OpenSearch results above {absolute_min_score} threshold: {len(filtered_keyword_results)}") # Step 3: Use filtered OpenSearch results for Vector search (double filtering) if filtered_keyword_results: # Extract resume IDs from filtered OpenSearch results filtered_resume_ids = [r.get("resume_id") for r in filtered_keyword_results if r.get("resume_id")] logger.info(f"Performing vector search on {len(filtered_resume_ids)} pre-filtered candidates") # Search using Vector search with country filter if country_header_code: # Create country filters for vector search vector_filters = {'country_header_code': country_header_code} vector_results = self.embedder.search_with_filters(search_query, vector_filters, top_k=100) logger.info(f"Vector search with country filter ({country_header_code}) returned {len(vector_results.get('result', [])) if vector_results else 0} results") else: # Fallback to filtered candidate search if no country filter vector_results = self.embedder.search_filtered_candidates(search_query, filtered_resume_ids, top_k=100) logger.info(f"Vector search returned {len(vector_results.get('result', [])) if vector_results else 0} results from filtered candidates") else: logger.info("No candidates passed OpenSearch filtering, skipping vector search") vector_results = {"result": []} # Combine and filter results candidates = self._combine_and_filter_results(filtered_keyword_results, vector_results) logger.info(f"Found {len(candidates)} matching candidates for job: {job_details['job_title']}") return candidates except Exception as e: logger.error(f"Error searching candidates: {str(e)}") return [] def _build_search_query(self, job_details: Dict[str, Any]) -> str: """ Build search query from job details Args: job_details: Job details dictionary Returns: Combined search query string """ query_parts = [] # Add job title if job_details.get('job_title'): query_parts.append(job_details['job_title']) # Add skills if job_details.get('skill_list'): query_parts.append(job_details['skill_list']) # Add job description (first 200 characters) if job_details.get('job_desc'): job_desc = job_details['job_desc'][:200] query_parts.append(job_desc) return " ".join(query_parts) def _combine_and_filter_results(self, keyword_results: List[Dict], vector_results: Dict) -> List[Dict[str, Any]]: """ Combine pre-filtered OpenSearch and Vector search results, apply percentage threshold Args: keyword_results: Pre-filtered results from OpenSearch (already above absolute minimum) vector_results: Results from Qdrant vector search on filtered candidates Returns: List of filtered candidates with combined scores """ candidates = {} # Use dict to avoid duplicates # Process OpenSearch results (already filtered by absolute minimum) if keyword_results: logger.info(f"Processing {len(keyword_results)} pre-filtered OpenSearch results") if keyword_results: max_keyword_score = max([r.get("keyword_score", 0) for r in keyword_results]) logger.info(f"Max OpenSearch score: {max_keyword_score}") for result in keyword_results: resume_id = result.get("resume_id") score = result.get("keyword_score", 0) match_percentage = (score / max_keyword_score) * 100 if max_keyword_score > 0 else 0 # Only include if percentage threshold or higher match min_threshold = float(os.getenv("JOB_MATCHING_MIN_PERCENTAGE", 50.0)) if match_percentage >= min_threshold and resume_id: candidates[resume_id] = { "resume_id": resume_id, "keyword_score": score, "keyword_percentage": round(match_percentage, 2), "vector_score": 0, "vector_percentage": 0, "combined_score": match_percentage } logger.info(f"OpenSearch candidates above {min_threshold}% threshold: {len([c for c in candidates.values() if c['keyword_percentage'] >= min_threshold])}") else: logger.info("No pre-filtered OpenSearch results received") # Process Vector search results (from filtered candidates only) vector_data = vector_results.get("result", []) if vector_results else [] if vector_data: logger.info(f"Processing {len(vector_data)} vector results from filtered candidates") # No need for absolute minimum filter as these are already filtered candidates if vector_data: max_vector_score = max([r.get("score", 0) for r in vector_data]) logger.info(f"Max vector score: {max_vector_score}") for result in vector_data: resume_id = result.get("payload", {}).get("resume_id") score = result.get("score", 0) match_percentage = (score / max_vector_score) * 100 if max_vector_score > 0 else 0 # Only include if percentage threshold or higher match min_threshold = float(os.getenv("JOB_MATCHING_MIN_PERCENTAGE", 50.0)) if match_percentage >= min_threshold and resume_id: if resume_id in candidates: # Update existing candidate with vector scores candidates[resume_id]["vector_score"] = score candidates[resume_id]["vector_percentage"] = round(match_percentage, 2) # Combine scores (weighted average) candidates[resume_id]["combined_score"] = round( (candidates[resume_id]["keyword_percentage"] + match_percentage) / 2, 2 ) else: # This shouldn't happen as vector search is on filtered candidates logger.warning(f"Vector result found for resume_id {resume_id} not in OpenSearch results") candidates[resume_id] = { "resume_id": resume_id, "keyword_score": 0, "keyword_percentage": 0, "vector_score": score, "vector_percentage": round(match_percentage, 2), "combined_score": match_percentage } logger.info(f"Vector candidates above {min_threshold}% threshold: {len([c for c in candidates.values() if c['vector_percentage'] >= min_threshold])}") else: logger.info("No vector results received from filtered candidates") # Convert to list and sort by combined score candidate_list = list(candidates.values()) candidate_list.sort(key=lambda x: x["combined_score"], reverse=True) logger.info(f"After double filtering (absolute + {os.getenv('JOB_MATCHING_MIN_PERCENTAGE', '50.0')}%): {len(candidate_list)} candidates") return candidate_list def _generate_candidate_analysis(self, job_details: Dict[str, Any], candidate: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ Generate AI analysis for a candidate including strengths, weaknesses, and justification Args: job_details: Job details dictionary candidate: Candidate information with scores Returns: Analysis dictionary or None if failed """ try: # Get detailed candidate data from database candidate_data = self._get_candidate_details(candidate["resume_id"]) # If no database data available, return None (will use direct method) if not candidate_data: logger.info(f"No database candidate data found for {candidate['resume_id']}, will use direct analysis method") return None # Create analysis prompt prompt = self._create_analysis_prompt(job_details, candidate_data, candidate) # Get AI analysis system_prompt = "You are an expert HR analyst. Analyze the candidate's fit for the job and provide detailed, professional assessment." analysis_text = self.ai_service.ai_response( question=prompt, system_prompt=system_prompt, max_tokens=1000 ) # Parse the AI response parsed_analysis = self._parse_ai_analysis(analysis_text, candidate["combined_score"]) return parsed_analysis except Exception as e: logger.error(f"Error generating candidate analysis for {candidate.get('resume_id')}: {str(e)}") return None def _get_candidate_details(self, resume_id: str) -> Optional[Dict[str, Any]]: """ Get detailed candidate information from MongoDB Args: resume_id: Resume/jobseeker ID Returns: Candidate details or None if not found """ try: # Get complete jobseeker data candidate_data = mongo_service.get_jobseeker_complete_data(resume_id) return candidate_data except Exception as e: logger.error(f"Error getting candidate details for {resume_id}: {str(e)}") return None def _create_analysis_prompt(self, job_details: Dict[str, Any], candidate_data: Dict[str, Any], candidate_scores: Dict[str, Any]) -> str: """ Create prompt for AI analysis Args: job_details: Job requirements candidate_data: Candidate information candidate_scores: Matching scores Returns: Formatted prompt string """ try: # Extract candidate information safely basic_details = candidate_data.get("basic_details", {}) if candidate_data else {} employment_details = candidate_data.get("employment_details", []) if candidate_data else [] education_details = candidate_data.get("education_details", []) if candidate_data else [] # logger.info(f"Creating prompt - Basic details keys: {list(basic_details.keys()) if basic_details else 'None'}") # logger.info(f"Creating prompt - Employment count: {len(employment_details) if employment_details else 0}") # logger.info(f"Creating prompt - Education count: {len(education_details) if education_details else 0}") # Format candidate experience safely experience_text = "" if employment_details and isinstance(employment_details, list): for emp in employment_details[:3]: # Top 3 experiences if isinstance(emp, dict) and emp.get("role") and emp.get("company"): experience_text += f"- {emp['role']} at {emp['company']}\n" elif isinstance(emp, dict) and emp.get("designation") and emp.get("company_name"): experience_text += f"- {emp['designation']} at {emp['company_name']}\n" # Format candidate education safely education_text = "" if education_details and isinstance(education_details, list): for edu in education_details[:2]: # Top 2 educations if isinstance(edu, dict) and edu.get("degree") and edu.get("institution"): education_text += f"- {edu['degree']} from {edu['institution']}\n" # Safe access to job details job_title = job_details.get('job_title', 'N/A') if job_details else 'N/A' skill_list = job_details.get('skill_list', 'N/A') if job_details else 'N/A' job_type = job_details.get('job_type', 'N/A') if job_details else 'N/A' workplace_type = job_details.get('workplace_type', 'N/A') if job_details else 'N/A' country_name = job_details.get('country_name', 'N/A') if job_details else 'N/A' job_desc = job_details.get('job_desc', 'N/A') if job_details else 'N/A' # Safe access to candidate scores combined_score = candidate_scores.get('combined_score', 0) if candidate_scores else 0 # Calculate total experience years from employment details total_experience_years = 0 if employment_details and isinstance(employment_details, list): for emp in employment_details: if isinstance(emp, dict): # Try to extract duration information duration = emp.get('job_experience_duration', 0) if isinstance(duration, (int, float)): total_experience_years += duration elif isinstance(duration, str) and duration.replace('.', '').isdigit(): total_experience_years += float(duration) # Build detailed experience text with responsibilities company = emp.get('company_name', emp.get('company', 'N/A')) designation = emp.get('designation', emp.get('role', 'N/A')) job_desc = emp.get('job_desc', emp.get('job_description', 'No description'))[:200] experience_text += f"• {designation} at {company}\n Responsibilities: {job_desc}...\n\n" # Enhanced prompt with detailed scoring criteria prompt = f""" You are an expert HR analyst. Analyze this candidate for the job position with PRECISE percentage scoring. JOB REQUIREMENTS: - Position: {job_title} - Required Skills: {skill_list} - Job Type: {job_type} - Workplace: {workplace_type} - Location: {country_name} - Description: {job_desc[:400] if job_desc != 'N/A' else 'N/A'} CANDIDATE PROFILE: - Name: {basic_details.get('first_name', '')} {basic_details.get('last_name', '')} - Total Experience: ~{total_experience_years} years - Key Skills: {basic_details.get('key_skills', 'N/A')} - Profile Summary: {basic_details.get('profile_summary', 'N/A')[:300] if basic_details.get('profile_summary') else 'N/A'} WORK EXPERIENCE: {experience_text if experience_text else 'No detailed experience data available'} EDUCATION: {education_text if education_text else 'No education data available'} PRECISE SCORING GUIDELINES (Use EXACT percentages): 95-100%: Perfect match - ALL required skills + exceeds experience + perfect education + additional value 85-94%: Excellent match - Most required skills + strong experience + good education + some extras 75-84%: Very good match - Core skills present + adequate experience + meets education requirements 65-74%: Good match - Some key skills + reasonable experience + acceptable education background 55-64%: Fair match - Few required skills + limited relevant experience + basic education 45-54%: Below average - Missing many skills + insufficient experience + education gaps 35-44%: Poor match - Wrong skill set + little relevant experience + education mismatch 25-34%: Very poor match - No relevant skills + unrelated background + major gaps 15-24%: Not suitable - Completely different field + no matching qualifications 0-14%: Totally unsuitable - No relevance to job requirements whatsoever CRITICAL INSTRUCTIONS: 1. Be PRECISE - use specific percentages like 73%, 82%, 91%, not round numbers like 70%, 80%, 90% 2. Consider SKILL OVERLAP percentage between job requirements and candidate skills 3. Factor in EXPERIENCE RELEVANCE and years of experience 4. Evaluate EDUCATION ALIGNMENT with job requirements 5. Avoid generic 60-65% range - be specific based on actual qualifications Please provide analysis in this EXACT format: STRENGTHS: [List 3-5 specific strengths with examples from their background] WEAKNESSES: [List 3-5 specific gaps compared to job requirements] MATCHING_PERCENTAGE: [Provide ONE precise percentage (0-100) - be specific like 78% or 86%, NOT generic like 65%] MATCH_JUSTIFICATION: [2-3 detailed sentences explaining the EXACT percentage with specific skill/experience examples] """ return prompt except Exception as e: logger.error(f"Error creating analysis prompt: {str(e)}") logger.error(f"Job details: {job_details}") logger.error(f"Candidate data: {candidate_data}") logger.error(f"Candidate scores: {candidate_scores}") raise e def _parse_ai_analysis(self, analysis_text: str, fallback_percentage: float) -> Dict[str, Any]: """ Parse AI analysis response into structured data Args: analysis_text: Raw AI response fallback_percentage: Fallback percentage if parsing fails Returns: Structured analysis dictionary """ try: analysis = { "strengths": [], "weaknesses": [], "matching_percentage": fallback_percentage, "match_justification": "" } # Split into sections sections = analysis_text.split('\n\n') for section in sections: section = section.strip() if section.startswith('STRENGTHS:'): strengths_text = section.replace('STRENGTHS:', '').strip() # Convert to array by splitting on line breaks and filtering empty lines strengths_list = [s.strip().lstrip('-•').strip() for s in strengths_text.split('\n') if s.strip() and not s.strip().startswith('[') and not s.strip().endswith(']')] analysis["strengths"] = strengths_list elif section.startswith('WEAKNESSES:'): weaknesses_text = section.replace('WEAKNESSES:', '').strip() # Convert to array by splitting on line breaks and filtering empty lines weaknesses_list = [w.strip().lstrip('-•').strip() for w in weaknesses_text.split('\n') if w.strip() and not w.strip().startswith('[') and not w.strip().endswith(']')] analysis["weaknesses"] = weaknesses_list elif section.startswith('MATCHING_PERCENTAGE:'): percentage_text = section.replace('MATCHING_PERCENTAGE:', '').strip() # Extract number from text (including decimals) import re numbers = re.findall(r'\d+\.?\d*', percentage_text) if numbers: analysis["matching_percentage"] = min(100.0, max(0.0, float(numbers[0]))) # Allow 0-100 range as float elif section.startswith('MATCH_JUSTIFICATION:'): analysis["match_justification"] = section.replace('MATCH_JUSTIFICATION:', '').strip() # Only return analysis if all required fields have meaningful content if (not analysis["strengths"] or not analysis["weaknesses"] or not analysis["match_justification"] or len(analysis["strengths"]) == 0 or len(analysis["weaknesses"]) == 0 or analysis["match_justification"].strip() == ""): logger.warning("AI analysis incomplete - skipping candidate") return None return analysis except Exception as e: logger.error(f"Error parsing AI analysis: {str(e)}") # Return None instead of fallback data - let the candidate be skipped return None def _store_screening_result(self, job_emp_id: str, candidate: Dict[str, Any], analysis: Dict[str, Any]): """ Store screening result in MongoDB screenings collection Check for duplicates before inserting Args: job_emp_id: Job employer ID candidate: Candidate information analysis: AI analysis result """ try: # Get database connection from services.database import get_database db = get_database() screenings_collection = db['screenings'] # Check if screening already exists for this job + candidate + type=2 existing_screening = screenings_collection.find_one({ "job_id": job_emp_id, "jobseeker_id": int(candidate["resume_id"]), "type": 2 }) if existing_screening: logger.info(f"Screening already exists for candidate {candidate['resume_id']} and job {job_emp_id}, skipping insert") return True # Get candidate details for jobseeker_details array - store ALL available data candidate_data = self._get_candidate_details(candidate["resume_id"]) jobseeker_details = [] if candidate_data: # Extract ALL details for comprehensive jobseeker_details storage basic_details = candidate_data.get("basic_details", {}) employment_details = candidate_data.get("employment_details", []) education_details = candidate_data.get("education_details", []) certification_details = candidate_data.get("certification_details", []) course_details = candidate_data.get("course_details", []) project_details = candidate_data.get("project_details", []) language_details = candidate_data.get("language_details", []) hobby_details = candidate_data.get("hobby_details", []) reference_details = candidate_data.get("reference_details", []) portfolio_details = candidate_data.get("portfolio_details", []) social_media_details = candidate_data.get("social_media_details", []) # Format comprehensive basic_details with ALL available information basic_details_array = [] if basic_details: # Include ALL fields from basic_details - comprehensive storage basic_detail = { # Personal Information "name": f"{basic_details.get('first_name', '')} {basic_details.get('last_name', '')}".strip(), "first_name": basic_details.get('first_name', ''), "last_name": basic_details.get('last_name', ''), "email": candidate_data.get('email', '') or basic_details.get('email', ''), "phone": candidate_data.get('mobile_number', '') or basic_details.get('phone', ''), "mobile_number": candidate_data.get('mobile_number', ''), "birth_date": basic_details.get('birth_date', ''), "gender_id": basic_details.get('gender_id', ''), "marital_status": basic_details.get('marital_status', ''), "nationality_id": basic_details.get('nationality_id', ''), "religion_id": basic_details.get('religion_id', ''), # Professional Information "current_designation": basic_details.get('current_desigation', '') or basic_details.get('current_designation', ''), "current_company": basic_details.get('current_company', ''), "total_experience_year": basic_details.get('total_experience_year', ''), "total_experience_month": basic_details.get('total_experience_month', ''), "total_experience": f"{basic_details.get('total_experience_year', '')} {basic_details.get('total_experience_month', '')}".strip(), "industry_id": basic_details.get('industry_id', ''), "functional_area_id": basic_details.get('functional_area_id', ''), "profile_type_id": basic_details.get('profile_type_id', ''), # Skills and Profile "key_skills": basic_details.get('key_skills', '').split(',') if basic_details.get('key_skills') else [], "key_skills_id": basic_details.get('key_skills_id', []), "profile_summary": basic_details.get('profile_summary', ''), "cv_headline": basic_details.get('cv_headline', ''), # Salary Information "current_salary": basic_details.get('current_salary', ''), "current_salary_period": basic_details.get('current_salary_period', ''), "current_salary_currency": basic_details.get('current_salary_currency', ''), "expected_salary": basic_details.get('expected_salary', ''), "expected_salary_period": basic_details.get('expected_salary_period', ''), "expected_salary_currency": basic_details.get('expected_salary_currency', ''), # Location Information "current_location": basic_details.get('current_location', ''), "prefered_location": basic_details.get('prefered_location', ''), "country_id": basic_details.get('country_id', ''), "com_address": basic_details.get('com_address', ''), # Visa and Legal Status "visa_status_id": basic_details.get('visa_status_id', ''), "visa_residence_location": basic_details.get('visa_residence_location', ''), "visa_validity_month": basic_details.get('visa_validity_month', ''), "visa_validity_year": basic_details.get('visa_validity_year', ''), "license_opt": basic_details.get('license_opt', ''), # System Information "profile_completion": basic_details.get('profile_completion', ''), "profile_updated_date": basic_details.get('profile_updated_date', ''), "system_original_resume": basic_details.get('system_original_resume', ''), "system_resume_file": basic_details.get('system_resume_file', ''), "original_file_name": basic_details.get('original_file_name', ''), "resume_file": basic_details.get('resume_file', ''), "resume_added_date": basic_details.get('resume_added_date', ''), "profile_pic": basic_details.get('profile_pic', ''), # Language Information "known_language_id": basic_details.get('known_language_id', []), "known_languages": basic_details.get('known_languages', ''), # Additional Profile Data "cv_push": basic_details.get('cv_push', ''), "cv_push_id": basic_details.get('cv_push_id', ''), "cv_download_count": basic_details.get('cv_download_count', ''), "applied_job_count": basic_details.get('applied_job_count', ''), "connection_count": basic_details.get('connection_count', ''), "profile_upload": basic_details.get('profile_upload', ''), "resume_upload": basic_details.get('resume_upload', ''), "es_push": basic_details.get('es_push', ''), "status": basic_details.get('status', ''), "is_deleted": basic_details.get('is_deleted', ''), "created_at": basic_details.get('created_at', ''), "updated_at": basic_details.get('updated_at', ''), "id": basic_details.get('id', '') } basic_details_array.append(basic_detail) # Store comprehensive jobseeker details with ALL available information jobseeker_detail = { "basic_details": basic_details_array, "experience": employment_details if employment_details else [], # ALL employment details "education": education_details if education_details else [], # ALL education details "certifications": certification_details if certification_details else [], # ALL certifications "courses": course_details if course_details else [], # ALL courses "projects": project_details if project_details else [], # ALL projects "languages": language_details if language_details else [], # ALL languages "hobbies": hobby_details if hobby_details else [], # ALL hobbies "references": reference_details if reference_details else [], # ALL references "portfolio": portfolio_details if portfolio_details else [], # ALL portfolio items "social_media": social_media_details if social_media_details else [], # ALL social media # Additional top-level candidate information "email": candidate_data.get('email', ''), "mobile_number": candidate_data.get('mobile_number', ''), "country_name": candidate_data.get('country_name', ''), "country_header_code": candidate_data.get('country_header_code', ''), "status": candidate_data.get('status', ''), "indexed": candidate_data.get('indexed', ''), "country_id": candidate_data.get('country_id', ''), "is_subscribed": candidate_data.get('is_subscribed', 0), # Add is_subscribed from jobseekers table # Complete candidate profile data (if available) "additional_info": { k: v for k, v in candidate_data.items() if k not in ['basic_details', 'employment_details', 'education_details', 'certification_details', 'course_details', 'project_details', 'language_details', 'hobby_details', 'reference_details', 'portfolio_details', 'social_media_details'] } } jobseeker_details.append(jobseeker_detail) # Create new screening data with jobseeker_details screening_data = { "job_id": job_emp_id, "jobseeker_id": int(candidate["resume_id"]), # Store as integer "status": 1, "type": 2, # As specified - integer type "strengths": analysis["strengths"], # Now stored as array "weaknesses": analysis["weaknesses"], # Now stored as array "matching_percentage": round(analysis["matching_percentage"], 2), # Round to 2 decimal places "match_justification": analysis["match_justification"], "jobseeker_details": jobseeker_details, # Add jobseeker_details for type 2 "created_at": datetime.utcnow(), "processed_at": datetime.utcnow() } # Insert new screening record result = screenings_collection.insert_one(screening_data) logger.info(f"Successfully stored new screening result for candidate {candidate['resume_id']} and job {job_emp_id}") return True except Exception as e: logger.error(f"Error storing screening result: {str(e)}") return False def _ensure_jobseeker_indexed(self, jobseeker_id: str) -> bool: """ Ensure jobseeker is indexed in OpenSearch, index if needed Args: jobseeker_id: Jobseeker ID to check/index Returns: bool: True if indexed successfully, False otherwise """ try: # Check if already indexed jobseeker_id_int = int(jobseeker_id) jobseeker = mongo_service.get_collection('jobseekers').find_one( {"id": jobseeker_id_int, "status": 1} ) if not jobseeker: logger.warning(f"Jobseeker {jobseeker_id} not found or inactive") return False # Check if already indexed if jobseeker.get("indexed") == 1: logger.info(f"Jobseeker {jobseeker_id} already indexed") return True # Index the jobseeker success = self.opensearch_service.add_resume(jobseeker_id) if success: logger.info(f"Successfully indexed jobseeker {jobseeker_id}") return True else: logger.error(f"Failed to index jobseeker {jobseeker_id}") return False except Exception as e: logger.error(f"Error ensuring jobseeker {jobseeker_id} is indexed: {str(e)}") return False def _store_specific_screening_result(self, job_emp_id: str, candidate: Dict[str, Any], analysis: Dict[str, Any]): """ Store screening result for specific candidate analysis with type=1 Check for duplicates before inserting Args: job_emp_id: Job employer ID candidate: Candidate information analysis: AI analysis result """ try: # Get database connection from services.database import get_database db = get_database() screenings_collection = db['screenings'] # Check if screening already exists for this job + candidate + type=1 existing_screening = screenings_collection.find_one({ "job_id": job_emp_id, "jobseeker_id": int(candidate["resume_id"]), "type": 1 }) if existing_screening: logger.info(f"Specific screening already exists for candidate {candidate['resume_id']} and job {job_emp_id}, skipping insert") return True # Get candidate details for jobseeker_details array - store ALL available data for type 1 candidate_data = self._get_candidate_details(candidate["resume_id"]) jobseeker_details = [] if candidate_data: # Extract ALL details for comprehensive jobseeker_details storage (type 1) basic_details = candidate_data.get("basic_details", {}) employment_details = candidate_data.get("employment_details", []) education_details = candidate_data.get("education_details", []) certification_details = candidate_data.get("certification_details", []) course_details = candidate_data.get("course_details", []) project_details = candidate_data.get("project_details", []) language_details = candidate_data.get("language_details", []) hobby_details = candidate_data.get("hobby_details", []) reference_details = candidate_data.get("reference_details", []) portfolio_details = candidate_data.get("portfolio_details", []) social_media_details = candidate_data.get("social_media_details", []) # Format comprehensive basic_details with ALL available information for type 1 basic_details_array = [] if basic_details: # Include ALL fields from basic_details - comprehensive storage basic_detail = { # Personal Information "name": f"{basic_details.get('first_name', '')} {basic_details.get('last_name', '')}".strip(), "first_name": basic_details.get('first_name', ''), "last_name": basic_details.get('last_name', ''), "email": candidate_data.get('email', '') or basic_details.get('email', ''), "phone": candidate_data.get('mobile_number', '') or basic_details.get('phone', ''), "mobile_number": candidate_data.get('mobile_number', ''), "birth_date": basic_details.get('birth_date', ''), "gender_id": basic_details.get('gender_id', ''), "marital_status": basic_details.get('marital_status', ''), "nationality_id": basic_details.get('nationality_id', ''), "religion_id": basic_details.get('religion_id', ''), # Professional Information "current_designation": basic_details.get('current_desigation', '') or basic_details.get('current_designation', ''), "current_company": basic_details.get('current_company', ''), "total_experience_year": basic_details.get('total_experience_year', ''), "total_experience_month": basic_details.get('total_experience_month', ''), "total_experience": f"{basic_details.get('total_experience_year', '')} {basic_details.get('total_experience_month', '')}".strip(), "industry_id": basic_details.get('industry_id', ''), "functional_area_id": basic_details.get('functional_area_id', ''), "profile_type_id": basic_details.get('profile_type_id', ''), # Skills and Profile "key_skills": basic_details.get('key_skills', '').split(',') if basic_details.get('key_skills') else [], "key_skills_id": basic_details.get('key_skills_id', []), "profile_summary": basic_details.get('profile_summary', ''), "cv_headline": basic_details.get('cv_headline', ''), # Salary Information "current_salary": basic_details.get('current_salary', ''), "current_salary_period": basic_details.get('current_salary_period', ''), "current_salary_currency": basic_details.get('current_salary_currency', ''), "expected_salary": basic_details.get('expected_salary', ''), "expected_salary_period": basic_details.get('expected_salary_period', ''), "expected_salary_currency": basic_details.get('expected_salary_currency', ''), # Location Information "current_location": basic_details.get('current_location', ''), "prefered_location": basic_details.get('prefered_location', ''), "country_id": basic_details.get('country_id', ''), "com_address": basic_details.get('com_address', ''), # Visa and Legal Status "visa_status_id": basic_details.get('visa_status_id', ''), "visa_residence_location": basic_details.get('visa_residence_location', ''), "visa_validity_month": basic_details.get('visa_validity_month', ''), "visa_validity_year": basic_details.get('visa_validity_year', ''), "license_opt": basic_details.get('license_opt', ''), # System Information "profile_completion": basic_details.get('profile_completion', ''), "profile_updated_date": basic_details.get('profile_updated_date', ''), "system_original_resume": basic_details.get('system_original_resume', ''), "system_resume_file": basic_details.get('system_resume_file', ''), "original_file_name": basic_details.get('original_file_name', ''), "resume_file": basic_details.get('resume_file', ''), "resume_added_date": basic_details.get('resume_added_date', ''), "profile_pic": basic_details.get('profile_pic', ''), # Language Information "known_language_id": basic_details.get('known_language_id', []), "known_languages": basic_details.get('known_languages', ''), # Additional Profile Data "cv_push": basic_details.get('cv_push', ''), "cv_push_id": basic_details.get('cv_push_id', ''), "cv_download_count": basic_details.get('cv_download_count', ''), "applied_job_count": basic_details.get('applied_job_count', ''), "connection_count": basic_details.get('connection_count', ''), "profile_upload": basic_details.get('profile_upload', ''), "resume_upload": basic_details.get('resume_upload', ''), "es_push": basic_details.get('es_push', ''), "status": basic_details.get('status', ''), "is_deleted": basic_details.get('is_deleted', ''), "created_at": basic_details.get('created_at', ''), "updated_at": basic_details.get('updated_at', ''), "id": basic_details.get('id', '') } basic_details_array.append(basic_detail) # Store comprehensive jobseeker details with ALL available information for type 1 jobseeker_detail = { "basic_details": basic_details_array, "experience": employment_details if employment_details else [], # ALL employment details "education": education_details if education_details else [], # ALL education details "certifications": certification_details if certification_details else [], # ALL certifications "courses": course_details if course_details else [], # ALL courses "projects": project_details if project_details else [], # ALL projects "languages": language_details if language_details else [], # ALL languages "hobbies": hobby_details if hobby_details else [], # ALL hobbies "references": reference_details if reference_details else [], # ALL references "portfolio": portfolio_details if portfolio_details else [], # ALL portfolio items "social_media": social_media_details if social_media_details else [], # ALL social media # Additional top-level candidate information "email": candidate_data.get('email', ''), "mobile_number": candidate_data.get('mobile_number', ''), "country_name": candidate_data.get('country_name', ''), "country_header_code": candidate_data.get('country_header_code', ''), "status": candidate_data.get('status', ''), "indexed": candidate_data.get('indexed', ''), "country_id": candidate_data.get('country_id', ''), "is_subscribed": candidate_data.get('is_subscribed', 0), # Add is_subscribed from jobseekers table # Complete candidate profile data (if available) "additional_info": { k: v for k, v in candidate_data.items() if k not in ['basic_details', 'employment_details', 'education_details', 'certification_details', 'course_details', 'project_details', 'language_details', 'hobby_details', 'reference_details', 'portfolio_details', 'social_media_details'] } } jobseeker_details.append(jobseeker_detail) # Create new screening data with type=1 screening_data = { "job_id": job_emp_id, "jobseeker_id": int(candidate["resume_id"]), # Store as integer "status": 1, "type": 1, # Type 1 for specific candidate analysis "strengths": analysis["strengths"], # Now stored as array "weaknesses": analysis["weaknesses"], # Now stored as array "matching_percentage": round(analysis["matching_percentage"], 2), # Round to 2 decimal places "match_justification": analysis["match_justification"], "jobseeker_details": jobseeker_details, # Add jobseeker_details for type 1 "created_at": datetime.utcnow(), "processed_at": datetime.utcnow() } # Insert new screening record result = screenings_collection.insert_one(screening_data) logger.info(f"Successfully stored specific screening result for candidate {candidate['resume_id']} and job {job_emp_id}") return True except Exception as e: logger.error(f"Error storing specific screening result: {str(e)}") return False def analyze_single_candidate_for_upload(self, job_emp_id: str, jobseeker_id: int, parsed_data: Dict[str, Any], analysis_type: int = 3) -> Dict[str, Any]: """ Analyze a single candidate for upload with AI analysis and store with specified type Args: job_emp_id: The employer job ID jobseeker_id: The jobseeker ID from jobseekers_bulk_upload_cv.id parsed_data: Parsed resume data analysis_type: Type of analysis (3 for upload analysis) Returns: Dict: Analysis result with success status """ try: # Get job details using the existing method job_details = self._get_job_details(job_emp_id) if not job_details: return { "success": False, "error": f"Failed to get job details for {job_emp_id}" } # Generate AI analysis using existing method # Create candidate object for analysis - same format as types 1 & 2 candidate = { "resume_id": str(jobseeker_id), "keyword_score": 0, "keyword_percentage": 0, "vector_score": 0, "vector_percentage": 0, "combined_score": 0 # Will be set by AI analysis } # Get complete candidate data from all database tables for type 3 complete_candidate_data = self._get_complete_bulk_upload_candidate_data(jobseeker_id) if not complete_candidate_data: logger.warning(f"Could not fetch complete candidate data for jobseeker {jobseeker_id}, using parsed data only") # Fallback to parsed data formatting complete_candidate_data = self._format_parsed_data_for_storage(parsed_data, jobseeker_id) # Generate AI analysis using the same method as types 1 & 2 analysis_result = self._generate_candidate_analysis(job_details, candidate) # If analysis fails, try with complete candidate data directly if not analysis_result: analysis_result = self._generate_candidate_analysis_direct(job_details, complete_candidate_data, candidate) if analysis_result: # Store screening result with specified type and complete candidate data storage_success = self._store_upload_screening_result( job_emp_id=job_emp_id, jobseeker_id=jobseeker_id, analysis=analysis_result, analysis_type=analysis_type, candidate_data=complete_candidate_data # Pass complete candidate data for jobseeker_details ) if storage_success: logger.info(f"Successfully analyzed and stored upload candidate {jobseeker_id} for job {job_emp_id} with type {analysis_type}") return { "success": True, "jobseeker_id": jobseeker_id, "analysis": analysis_result } else: return { "success": False, "error": "Failed to store analysis result" } else: return { "success": False, "error": "Failed to generate AI analysis" } except Exception as e: logger.error(f"Error analyzing single candidate for upload: {str(e)}") return { "success": False, "error": str(e) } def _store_upload_screening_result(self, job_emp_id: str, jobseeker_id: int, analysis: Dict[str, Any], analysis_type: int, candidate_data: Dict[str, Any] = None): """ Store screening result for upload analysis with jobseeker_details for type 3 Check for duplicates before inserting Args: job_emp_id: Job employer ID jobseeker_id: Jobseeker ID from bulk upload collection analysis: AI analysis result analysis_type: Type of analysis (3 for upload) candidate_data: Candidate data to store in jobseeker_details array """ try: # Get database connection from services.database import get_database db = get_database() screenings_collection = db['screenings'] # Check if screening already exists for this job + candidate + type existing_screening = screenings_collection.find_one({ "job_id": job_emp_id, "jobseeker_id": jobseeker_id, # Already integer "type": analysis_type }) if existing_screening: logger.info(f"Upload screening already exists for candidate {jobseeker_id} and job {job_emp_id} with type {analysis_type}, skipping insert") return True # Create comprehensive jobseeker_details array for type 3 with ALL available data jobseeker_details = [] if candidate_data and analysis_type == 3: # Extract ALL details for comprehensive jobseeker_details storage (type 3) basic_details = candidate_data.get("basic_details", {}) employment_details = candidate_data.get("employment_details", []) education_details = candidate_data.get("education_details", []) certification_details = candidate_data.get("certification_details", []) course_details = candidate_data.get("course_details", []) project_details = candidate_data.get("project_details", []) language_details = candidate_data.get("language_details", []) hobby_details = candidate_data.get("hobby_details", []) reference_details = candidate_data.get("reference_details", []) portfolio_details = candidate_data.get("portfolio_details", []) social_media_details = candidate_data.get("social_media_details", []) # For type 3, get email and mobile from jobseekers_bulk_upload_cv table email = "" mobile = "" try: from services.database import get_database db = get_database() bulk_upload_data = db['jobseekers_bulk_upload_cv'].find_one({"id": jobseeker_id}) if bulk_upload_data: email = bulk_upload_data.get('email', '') mobile = bulk_upload_data.get('mobile_number', '') logger.info(f"Found bulk upload data for jobseeker {jobseeker_id}: email={email}, mobile={mobile}") else: logger.warning(f"No bulk upload data found for jobseeker {jobseeker_id}") except Exception as e: logger.error(f"Error getting bulk upload data for jobseeker {jobseeker_id}: {str(e)}") # Fallback to candidate_data if bulk upload data not found if not email: email = candidate_data.get('email', '') or basic_details.get('email', '') if not mobile: mobile = candidate_data.get('mobile_number', '') or basic_details.get('phone', '') # Format comprehensive basic_details with ALL available information for type 3 basic_details_array = [] if basic_details or candidate_data: # Get name from basic_details first, then fallback to additional_info first_name = basic_details.get('first_name', '') if basic_details else '' last_name = basic_details.get('last_name', '') if basic_details else '' # If basic_details names are empty, try to get from additional_info or parse from bulk upload if not first_name and not last_name and candidate_data: additional_info = candidate_data.get('additional_info', {}) cv_filename = additional_info.get('cv_filename', '') # Try to extract name from CV filename if available if cv_filename and not cv_filename.startswith('uploads/'): # Remove file extension and common separators name_from_file = cv_filename.replace('.pdf', '').replace('.doc', '').replace('.docx', '').replace('_', ' ').replace('-', ' ').strip() name_parts = name_from_file.split() if len(name_parts) >= 2: first_name = name_parts[0] last_name = ' '.join(name_parts[1:]) # If still no name, try to get from parsed basic_details parsed_basic = candidate_data.get('basic_details', {}) if not first_name and parsed_basic.get('first_name'): first_name = parsed_basic.get('first_name', '') if not last_name and parsed_basic.get('last_name'): last_name = parsed_basic.get('last_name', '') # Use email and mobile from the bulk upload lookup we did earlier # But also fallback to parsed data if those are empty final_email = email final_mobile = mobile if not final_email and candidate_data: parsed_basic = candidate_data.get('basic_details', {}) final_email = parsed_basic.get('email', '') if not final_mobile and candidate_data: parsed_basic = candidate_data.get('basic_details', {}) final_mobile = parsed_basic.get('phone', '') # Include ALL fields from basic_details - comprehensive storage basic_detail = { # Personal Information "name": f"{first_name} {last_name}".strip(), "first_name": first_name, "last_name": last_name, "email": final_email, "phone": final_mobile, "mobile_number": final_mobile, "birth_date": basic_details.get('birth_date', ''), "gender_id": basic_details.get('gender_id', ''), "marital_status": basic_details.get('marital_status', ''), "nationality_id": basic_details.get('nationality_id', ''), "religion_id": basic_details.get('religion_id', ''), # Professional Information "current_designation": basic_details.get('current_desigation', '') or basic_details.get('current_designation', ''), "current_company": basic_details.get('current_company', ''), "total_experience_year": basic_details.get('total_experience_year', ''), "total_experience_month": basic_details.get('total_experience_month', ''), "total_experience": f"{basic_details.get('total_experience_year', '')} {basic_details.get('total_experience_month', '')}".strip(), "industry_id": basic_details.get('industry_id', ''), "functional_area_id": basic_details.get('functional_area_id', ''), "profile_type_id": basic_details.get('profile_type_id', ''), # Skills and Profile "key_skills": basic_details.get('key_skills', '').split(',') if basic_details.get('key_skills') else [], "key_skills_id": basic_details.get('key_skills_id', []), "profile_summary": basic_details.get('profile_summary', ''), "cv_headline": basic_details.get('cv_headline', ''), # Salary Information "current_salary": basic_details.get('current_salary', ''), "current_salary_period": basic_details.get('current_salary_period', ''), "current_salary_currency": basic_details.get('current_salary_currency', ''), "expected_salary": basic_details.get('expected_salary', ''), "expected_salary_period": basic_details.get('expected_salary_period', ''), "expected_salary_currency": basic_details.get('expected_salary_currency', ''), # Location Information "current_location": basic_details.get('current_location', ''), "prefered_location": basic_details.get('prefered_location', ''), "country_id": basic_details.get('country_id', ''), "com_address": basic_details.get('com_address', ''), # Visa and Legal Status "visa_status_id": basic_details.get('visa_status_id', ''), "visa_residence_location": basic_details.get('visa_residence_location', ''), "visa_validity_month": basic_details.get('visa_validity_month', ''), "visa_validity_year": basic_details.get('visa_validity_year', ''), "license_opt": basic_details.get('license_opt', ''), # System Information "profile_completion": basic_details.get('profile_completion', ''), "profile_updated_date": basic_details.get('profile_updated_date', ''), "system_original_resume": basic_details.get('system_original_resume', ''), "system_resume_file": basic_details.get('system_resume_file', ''), "original_file_name": basic_details.get('original_file_name', ''), "resume_file": basic_details.get('resume_file', ''), "resume_added_date": basic_details.get('resume_added_date', ''), "profile_pic": basic_details.get('profile_pic', ''), # Language Information "known_language_id": basic_details.get('known_language_id', []), "known_languages": basic_details.get('known_languages', ''), # Additional Profile Data "cv_push": basic_details.get('cv_push', ''), "cv_push_id": basic_details.get('cv_push_id', ''), "cv_download_count": basic_details.get('cv_download_count', ''), "applied_job_count": basic_details.get('applied_job_count', ''), "connection_count": basic_details.get('connection_count', ''), "profile_upload": basic_details.get('profile_upload', ''), "resume_upload": basic_details.get('resume_upload', ''), "es_push": basic_details.get('es_push', ''), "status": basic_details.get('status', ''), "is_deleted": basic_details.get('is_deleted', ''), "created_at": basic_details.get('created_at', ''), "updated_at": basic_details.get('updated_at', ''), "id": basic_details.get('id', '') } basic_details_array.append(basic_detail) # Store comprehensive jobseeker details with ALL available information for type 3 jobseeker_detail = { "basic_details": basic_details_array, "experience": employment_details if employment_details else [], # ALL employment details "education": education_details if education_details else [], # ALL education details "certifications": certification_details if certification_details else [], # ALL certifications "courses": course_details if course_details else [], # ALL courses "projects": project_details if project_details else [], # ALL projects "languages": language_details if language_details else [], # ALL languages "hobbies": hobby_details if hobby_details else [], # ALL hobbies "references": reference_details if reference_details else [], # ALL references "portfolio": portfolio_details if portfolio_details else [], # ALL portfolio items "social_media": social_media_details if social_media_details else [], # ALL social media # Additional top-level candidate information "email": final_email, "mobile_number": final_mobile, "country_name": candidate_data.get('country_name', ''), "country_header_code": candidate_data.get('country_header_code', ''), "status": candidate_data.get('status', ''), "is_subscribed": candidate_data.get('is_subscribed', 0), "indexed": candidate_data.get('indexed', ''), "country_id": candidate_data.get('country_id', ''), # Complete candidate profile data (if available) "additional_info": { k: v for k, v in candidate_data.items() if k not in ['basic_details', 'employment_details', 'education_details', 'certification_details', 'course_details', 'project_details', 'language_details', 'hobby_details', 'reference_details', 'portfolio_details', 'social_media_details'] } } jobseeker_details.append(jobseeker_detail) # Create new screening data screening_data = { "job_id": job_emp_id, "jobseeker_id": jobseeker_id, # Store as integer (already int) "status": 1, "type": analysis_type, "strengths": analysis["strengths"], # Now stored as array "weaknesses": analysis["weaknesses"], # Now stored as array "matching_percentage": round(analysis["matching_percentage"], 2), # Store all percentages including 0-100% "match_justification": analysis["match_justification"], "created_at": datetime.utcnow(), "processed_at": datetime.utcnow() } # Add jobseeker_details array for type 3 if analysis_type == 3 and candidate_data: screening_data["jobseeker_details"] = jobseeker_details elif analysis_type == 3: # Ensure jobseeker_details is always present for type 3 even if empty screening_data["jobseeker_details"] = [] # Insert new screening record result = screenings_collection.insert_one(screening_data) logger.info(f"Successfully stored upload screening result for candidate {jobseeker_id} and job {job_emp_id} with type {analysis_type}") return True except Exception as e: logger.error(f"Error storing upload screening result: {str(e)}") return False def _generate_candidate_analysis_direct(self, job_details: Dict[str, Any], candidate_data: Dict[str, Any], candidate: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ Generate AI analysis for a candidate using direct data (for upload scenarios) Args: job_details: Job details dictionary candidate_data: Direct candidate data candidate: Candidate information with scores Returns: Analysis dictionary or None if failed """ try: # Debug logging to identify the issue logger.info(f"Starting AI analysis for candidate {candidate.get('resume_id', 'unknown') if candidate else 'None candidate'}") logger.info(f"Job details available: {bool(job_details)}") logger.info(f"Candidate data available: {bool(candidate_data)}") logger.info(f"Candidate object: {candidate}") # Create analysis prompt prompt = self._create_analysis_prompt(job_details, candidate_data, candidate) # Debug logging logger.info(f"Generated analysis prompt for candidate {candidate.get('resume_id', 'unknown')}") # Get AI analysis system_prompt = "You are an expert HR analyst. Analyze the candidate's fit for the job and provide detailed, professional assessment." analysis_text = self.ai_service.ai_response( question=prompt, system_prompt=system_prompt, max_tokens=1000 ) # Parse the AI response (allow all percentages 0-100% for upload analysis) parsed_analysis = self._parse_ai_analysis(analysis_text, 50.0) # Default fallback return parsed_analysis except Exception as e: # Use safer key access for error logging candidate_id = candidate.get('resume_id', 'unknown') if candidate else 'unknown' logger.error(f"Error generating direct candidate analysis for {candidate_id}: {str(e)}") return None def _get_complete_bulk_upload_candidate_data(self, jobseeker_id: int) -> Dict[str, Any]: """ Get complete candidate data from all bulk upload related tables Args: jobseeker_id: The jobseeker ID from bulk upload collection Returns: Complete candidate data dictionary """ try: from services.database import get_database db = get_database() # Initialize complete candidate data candidate_data = { "basic_details": {}, "employment_details": [], "education_details": [], "certification_details": [], "course_details": [], "project_details": [], "language_details": [], "hobby_details": [], "reference_details": [], "portfolio_details": [], "social_media_details": [] } # Get data from jobseekers_bulk_upload_cv main_cv_data = db['jobseekers_bulk_upload_cv'].find_one({"id": jobseeker_id}) if main_cv_data: candidate_data["email"] = main_cv_data.get('email', '') candidate_data["mobile_number"] = main_cv_data.get('mobile_number', '') candidate_data["cv_filename"] = main_cv_data.get('cv_filename', '') candidate_data["cvfile_store_path"] = main_cv_data.get('cvfile_store_path', '') candidate_data["last_login_date"] = main_cv_data.get('last_login_date', '') # Get basic details from jobseeker_basic_details_bulk_upload_cv basic_details = db['jobseeker_basic_details_bulk_upload_cv'].find_one({"jobseeker_id": jobseeker_id}) if basic_details: candidate_data["basic_details"] = basic_details # Get employment details from jobseeker_employment_details_bulk_upload_cv employment_details = list(db['jobseeker_employment_details_bulk_upload_cv'].find({"jobseeker_id": jobseeker_id})) candidate_data["employment_details"] = employment_details # Get education details from jobseeker_education_details_bulk_upload_cv education_details = list(db['jobseeker_education_details_bulk_upload_cv'].find({"jobseeker_id": jobseeker_id})) candidate_data["education_details"] = education_details # Get course details from jobseeker_course_details_bulk_upload_cv course_details = list(db['jobseeker_course_details_bulk_upload_cv'].find({"jobseeker_id": jobseeker_id})) candidate_data["course_details"] = course_details # Get project details from jobseeker_project_details_bulk_upload_cv project_details = list(db['jobseeker_project_details_bulk_upload_cv'].find({"jobseeker_id": jobseeker_id})) candidate_data["project_details"] = project_details # Get certification details from jobseeker_certification_details_bulk_upload_cv certification_details = list(db['jobseeker_certification_details_bulk_upload_cv'].find({"jobseeker_id": jobseeker_id})) candidate_data["certification_details"] = certification_details logger.info(f"Successfully fetched complete candidate data for jobseeker {jobseeker_id}") return candidate_data except Exception as e: logger.error(f"Error fetching complete candidate data for jobseeker {jobseeker_id}: {str(e)}") return None def _format_parsed_data_for_storage(self, parsed_data: Dict[str, Any], jobseeker_id: int) -> Dict[str, Any]: """ Format parsed data to match the expected storage structure Args: parsed_data: The parsed resume data jobseeker_id: The jobseeker ID Returns: Formatted candidate data dictionary """ try: # Convert parsed data to same format as database candidate data formatted_candidate_data = { "basic_details": { "jobseeker_id": jobseeker_id, "first_name": parsed_data.get("name", "").split()[0] if parsed_data.get("name") else "", "last_name": " ".join(parsed_data.get("name", "").split()[1:]) if parsed_data.get("name") and len(parsed_data.get("name", "").split()) > 1 else "", "full_name": parsed_data.get("name", ""), "email": parsed_data.get("email", ""), "phone": parsed_data.get("phone", ""), "key_skills": ",".join(parsed_data.get("skills", [])) if parsed_data.get("skills") else "", "profile_summary": parsed_data.get("profile_summary", "") if parsed_data.get("profile_summary") else "", "date_of_birth": parsed_data.get("date_of_birth"), "nationality": parsed_data.get("nationality"), "languages": parsed_data.get("languages", []), "status": 1, "is_deleted": 0, "created_at": parsed_data.get("created_at", ""), "updated_at": parsed_data.get("updated_at", "") }, "email": parsed_data.get("email", ""), "mobile_number": parsed_data.get("phone", ""), "employment_details": [], "education_details": [], "certification_details": [], "course_details": [], "project_details": [], "additional_info": { "email": parsed_data.get("email", ""), "mobile_number": parsed_data.get("phone", ""), "cv_filename": f"Parsed Resume - {parsed_data.get('name', 'Unknown')}.pdf", "cvfile_store_path": f"uploads/parsed/{jobseeker_id}/resume.pdf", "last_login_date": parsed_data.get("created_at", "") }, "project_details": [], "language_details": [], "hobby_details": [], "reference_details": [], "portfolio_details": [], "social_media_details": [], "email": parsed_data.get("email", ""), "mobile_number": parsed_data.get("phone", "") } # Convert experience data to match database format if parsed_data.get("experience") and isinstance(parsed_data["experience"], list): for exp in parsed_data["experience"]: if isinstance(exp, dict): formatted_exp = { "jobseeker_id": jobseeker_id, "designation": exp.get("role", "") or exp.get("position", ""), "company_name": exp.get("company", ""), "start_date": exp.get("start_date"), "end_date": exp.get("end_date"), "job_description": exp.get("job_description", "") or exp.get("description", ""), "responsibilities": exp.get("responsibilities", []) } formatted_candidate_data["employment_details"].append(formatted_exp) # Convert education data to match database format if parsed_data.get("education") and isinstance(parsed_data["education"], list): for edu in parsed_data["education"]: if isinstance(edu, dict): formatted_edu = { "jobseeker_id": jobseeker_id, "degree": edu.get("degree", ""), "institute_name": edu.get("institution", ""), "field_of_study": edu.get("field_of_study", ""), "start_date": edu.get("start_date"), "end_date": edu.get("end_date"), "completion_year": edu.get("completion_year"), "grade": edu.get("grade") } formatted_candidate_data["education_details"].append(formatted_edu) # Convert certifications data if available if parsed_data.get("certifications") and isinstance(parsed_data["certifications"], list): for cert in parsed_data["certifications"]: if isinstance(cert, dict): formatted_cert = { "jobseeker_id": jobseeker_id, "certificate_name": cert.get("name", "") or cert.get("title", ""), "institute_name": cert.get("issuer", "") or cert.get("organization", ""), "issue_date": cert.get("issue_date"), "expiry_date": cert.get("expiry_date"), "credential_id": cert.get("credential_id"), "url": cert.get("url") } formatted_candidate_data["certification_details"].append(formatted_cert) # Convert projects data if available if parsed_data.get("projects") and isinstance(parsed_data["projects"], list): for proj in parsed_data["projects"]: if isinstance(proj, dict): formatted_proj = { "jobseeker_id": jobseeker_id, "project_title": proj.get("title", "") or proj.get("name", ""), "organization_name": proj.get("organization", "") or proj.get("company", ""), "project_description": proj.get("description", ""), "project_url": proj.get("url", ""), "start_date": proj.get("start_date"), "end_date": proj.get("end_date"), "technologies": proj.get("technologies", []) } formatted_candidate_data["project_details"].append(formatted_proj) logger.info(f"Successfully formatted parsed data for jobseeker {jobseeker_id}") return formatted_candidate_data except Exception as e: logger.error(f"Error formatting parsed data for jobseeker {jobseeker_id}: {str(e)}") return { "basic_details": {}, "employment_details": [], "education_details": [], "certification_details": [], "course_details": [], "project_details": [], "language_details": [], "hobby_details": [], "reference_details": [], "portfolio_details": [], "social_media_details": [], "email": parsed_data.get("email", ""), "mobile_number": parsed_data.get("phone", "") }