import os from pymongo import MongoClient from dotenv import load_dotenv from typing import Dict, Any, List from datetime import datetime, timezone from bson import ObjectId from urllib.parse import quote_plus from .mongo_service import MongoService from schemas.cv_models import ( JobseekerCV, JobseekerBasicDetails, JobseekerEducationDetails, JobseekerEmploymentDetails, JobseekerCourseDetails, JobseekerProjectDetails, JobseekerCertificationDetails ) # Load environment variables load_dotenv() class CVStorageService: def __init__(self): # Use MongoService for database connection self.mongo_service = MongoService() self.client = self.mongo_service.client self.db = self.mongo_service.db if not self.mongo_service.is_connected(): return # Collection references if self.db is not None: self.jobseekers_cv = self.db["jobseekers_bulk_upload_cv"] self.basic_details = self.db["jobseeker_basic_details_bulk_upload_cv"] self.course_details = self.db["jobseeker_course_details_bulk_upload_cv"] self.education_details = self.db["jobseeker_education_details_bulk_upload_cv"] self.employment_details = self.db["jobseeker_employment_details_bulk_upload_cv"] self.project_details = self.db["jobseeker_project_details_bulk_upload_cv"] self.certification_details = self.db["jobseeker_certification_details_bulk_upload_cv"] else: self.jobseekers_cv = None self.basic_details = None self.course_details = None self.education_details = None self.employment_details = None self.project_details = None self.certification_details = None def get_next_sequence_value(self, collection_name: str) -> int: """Get next auto-increment ID for a collection""" if self.db is None: raise Exception("Database not connected") counter = self.db.counters.find_one_and_update( {"_id": collection_name}, {"$inc": {"seq": 1}}, upsert=True, return_document=True ) return counter["seq"] def store_jobseeker_cv(self, filename: str, raw_text: str, parsed_data: Dict[str, Any], job_emp_id: str, file_path: str = "") -> int: """Store main CV record and return jobseeker_id""" if self.jobseekers_cv is None: raise Exception("Database not connected or collection not available") if job_emp_id is None or job_emp_id.strip() == "": raise Exception("job_emp_id is required and cannot be empty") jobseeker_id = self.get_next_sequence_value("jobseekers_bulk_upload_cv") # Create model instance with null handling cv_data = JobseekerCV( id=jobseeker_id, email=parsed_data.get("email") if parsed_data.get("email") is not None else "", mobile_number=parsed_data.get("phone") if parsed_data.get("phone") is not None else "", last_login_date=datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), cv_filename=filename if filename is not None else "", cv_raw_text=raw_text if raw_text is not None else "", cvfile_store_path=file_path if file_path is not None else "" ) # Convert to dict and add ObjectId cv_record = cv_data.dict() cv_record["_id"] = ObjectId() # Add required job_emp_id cv_record["job_emp_id"] = job_emp_id self.jobseekers_cv.insert_one(cv_record) return jobseeker_id def store_basic_details(self, jobseeker_id: int, parsed_data: Dict[str, Any], file_path: str = "") -> int: """Store basic details""" if self.basic_details is None: raise Exception("Database not connected or collection not available") basic_id = self.get_next_sequence_value("jobseeker_basic_details_bulk_upload_cv") # Parse name for first and last name with strict null handling name = parsed_data.get("name") if parsed_data.get("name") is not None else "" name_parts = name.split() if name else [] first_name = name_parts[0] if name_parts else "" last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else "" # Safely extract skills with proper type handling skills_list = parsed_data.get("skills", []) safe_skills = [] if skills_list and isinstance(skills_list, list): for skill in skills_list: if isinstance(skill, str): safe_skills.append(skill) elif isinstance(skill, dict): # If skill is a dict, try to get name or title field skill_name = skill.get("name") or skill.get("title") or skill.get("skill", "") if skill_name: safe_skills.append(str(skill_name)) else: # Convert other types to string safe_skills.append(str(skill)) # Safely extract languages with proper type handling languages_list = parsed_data.get("languages", []) safe_languages = [] if languages_list and isinstance(languages_list, list): for lang in languages_list: if isinstance(lang, str): safe_languages.append(lang) elif isinstance(lang, dict): # If language is a dict, try to get name or language field lang_name = lang.get("name") or lang.get("language") or lang.get("title", "") if lang_name: safe_languages.append(str(lang_name)) else: # Convert other types to string safe_languages.append(str(lang)) # Create model instance with strict null handling basic_data = JobseekerBasicDetails( id=basic_id, jobseeker_id=jobseeker_id, full_name=name, first_name=first_name, last_name=last_name, resume_added_date=datetime.now(timezone.utc).strftime("%Y-%m-%d"), resume_file=file_path if file_path is not None else "", # Add file path to basic details key_skills=",".join(safe_skills), key_skills_custome=safe_skills, profile_summary=parsed_data.get("profile_summary") if parsed_data.get("profile_summary") is not None else "", email=parsed_data.get("email") if parsed_data.get("email") is not None else "", phone=parsed_data.get("phone") if parsed_data.get("phone") is not None else "", date_of_birth=parsed_data.get("date_of_birth") if parsed_data.get("date_of_birth") is not None else None, nationality=parsed_data.get("nationality") if parsed_data.get("nationality") is not None else None, languages=safe_languages ) # Convert to dict and add ObjectId basic_record = basic_data.dict() basic_record["_id"] = ObjectId() self.basic_details.insert_one(basic_record) return basic_id def store_education_details(self, jobseeker_id: int, education_list: List[Dict[str, Any]]) -> List[int]: """Store education details""" if self.education_details is None: raise Exception("Database not connected or collection not available") education_ids = [] for education in education_list: if not education or all(v is None for v in education.values()): continue education_id = self.get_next_sequence_value("jobseeker_education_details_bulk_upload_cv") education_record = { "_id": ObjectId(), "id": education_id, "jobseeker_id": jobseeker_id, "institute_name": education.get("institution") if education.get("institution") is not None else "", "completion_year": None, "highest_education_id": 0, "degree_id": 0, "specialization_id": 0, "status": 1, "is_deleted": 0, "updated_at": datetime.now(timezone.utc), "created_at": datetime.now(timezone.utc), "degree": education.get("degree") if education.get("degree") is not None else "", "field_of_study": education.get("field_of_study") if education.get("field_of_study") is not None else "", "start_date": education.get("start_date") if education.get("start_date") is not None else None, "end_date": education.get("end_date") if education.get("end_date") is not None else None, "grade": education.get("grade") if education.get("grade") is not None else None } self.education_details.insert_one(education_record) education_ids.append(education_id) return education_ids def store_employment_details(self, jobseeker_id: int, experience_list: List[Dict[str, Any]]) -> List[int]: """Store employment/experience details""" if self.employment_details is None: raise Exception("Database not connected or collection not available") employment_ids = [] for experience in experience_list: if not experience or all(v is None for v in experience.values()): continue employment_id = self.get_next_sequence_value("jobseeker_employment_details_bulk_upload_cv") employment_record = { "_id": ObjectId(), "id": employment_id, "jobseeker_id": jobseeker_id, "company_name": experience.get("company") if experience.get("company") is not None else "", "designation": experience.get("role") if experience.get("role") is not None else "", "job_desc": [experience.get("responsibilities")] if experience.get("responsibilities") is not None else [], "from_period": None, "to_period": None, "currently_working": 0, "status": 1, "is_deleted": 0, "updated_at": datetime.now(timezone.utc), "created_at": datetime.now(timezone.utc), "start_date": experience.get("start_date") if experience.get("start_date") is not None else None, "end_date": experience.get("end_date") if experience.get("end_date") is not None else None, "job_description": experience.get("job_description") if experience.get("job_description") is not None else None } self.employment_details.insert_one(employment_record) employment_ids.append(employment_id) return employment_ids def store_course_details(self, jobseeker_id: int, courses: List[str]) -> List[int]: """Store course details (if any courses mentioned in skills/education)""" if self.course_details is None: raise Exception("Database not connected or collection not available") course_ids = [] for course in courses: if not course or (isinstance(course, str) and course.strip() == "") or course is None: continue course_id = self.get_next_sequence_value("jobseeker_course_details_bulk_upload_cv") course_record = { "_id": ObjectId(), "id": course_id, "jobseeker_id": jobseeker_id, "course_name": course if course is not None else "", "completion_location": "", "completion_year": 0, "institute_name": "", "key_skills_id": [], "key_skills_name": [], "key_skills": "", "status": 1, "is_deleted": 0, "updated_at": datetime.now(timezone.utc), "created_at": datetime.now(timezone.utc), "provider": None, "completion_date": None, "duration": None } self.course_details.insert_one(course_record) course_ids.append(course_id) return course_ids def store_project_details(self, jobseeker_id: int, projects: List[Dict[str, Any]]) -> List[int]: """Store project details (extracted from experience or separate section)""" if self.project_details is None: raise Exception("Database not connected or collection not available") project_ids = [] for project in projects: if not project or all(v is None for v in project.values()): continue project_id = self.get_next_sequence_value("jobseeker_project_details_bulk_upload_cv") project_record = { "_id": ObjectId(), "id": project_id, "jobseeker_id": jobseeker_id, "project_title": project.get("project_name") if project.get("project_name") is not None else "", "organization_name": "", "project_url": project.get("url") if project.get("url") is not None else "", "project_description": project.get("description") if project.get("description") is not None else "", "project_start_date": None, "project_end_date": None, "status": 1, "is_deleted": 0, "updated_at": datetime.now(timezone.utc), "created_at": datetime.now(timezone.utc), "technologies": project.get("technologies") if project.get("technologies") is not None and isinstance(project.get("technologies"), list) else [], "start_date": project.get("start_date") if project.get("start_date") is not None else None, "end_date": project.get("end_date") if project.get("end_date") is not None else None } self.project_details.insert_one(project_record) project_ids.append(project_id) return project_ids def store_certification_details(self, jobseeker_id: int, certifications: List[Dict[str, Any]]) -> List[int]: """Store certification details""" if self.certification_details is None: raise Exception("Database not connected or collection not available") certification_ids = [] for cert in certifications: if not cert or all(v is None for v in cert.values()): continue cert_id = self.get_next_sequence_value("jobseeker_certification_details_bulk_upload_cv") cert_record = { "_id": ObjectId(), "id": cert_id, "jobseeker_id": jobseeker_id, "certificate_name": cert.get("certification_name") if cert.get("certification_name") is not None else "", "completion_location": "", "completion_year": 0, "institute_name": cert.get("issuing_organization") if cert.get("issuing_organization") is not None else "", "status": 1, "is_deleted": 0, "updated_at": datetime.now(timezone.utc), "created_at": datetime.now(timezone.utc), "issue_date": cert.get("issue_date") if cert.get("issue_date") is not None else None, "expiry_date": cert.get("expiry_date") if cert.get("expiry_date") is not None else None, "credential_id": cert.get("credential_id") if cert.get("credential_id") is not None else None, "url": cert.get("url") if cert.get("url") is not None else None } self.certification_details.insert_one(cert_record) certification_ids.append(cert_id) return certification_ids def store_complete_cv_data(self, filename: str, raw_text: str, parsed_data: Dict[str, Any], job_emp_id: str, file_path: str = "") -> Dict[str, Any]: """Store complete CV data across all collections""" try: # Check database connection first if self.db is None or self.client is None: return { "success": False, "error": "Database not connected. Cannot store CV data." } # Validate required job_emp_id if job_emp_id is None or job_emp_id.strip() == "": return { "success": False, "error": "job_emp_id is required and cannot be empty" } # 1. Store main CV record with required job_emp_id and file path jobseeker_id = self.store_jobseeker_cv(filename, raw_text, parsed_data, job_emp_id, file_path) # 2. Store basic details with file path basic_id = self.store_basic_details(jobseeker_id, parsed_data, file_path) # 3. Store education details education_ids = [] if parsed_data.get("education") is not None and isinstance(parsed_data["education"], list): education_ids = self.store_education_details(jobseeker_id, parsed_data["education"]) # 4. Store employment details employment_ids = [] if parsed_data.get("experience") is not None and isinstance(parsed_data["experience"], list): employment_ids = self.store_employment_details(jobseeker_id, parsed_data["experience"]) # 5. Store course details (from skills if any course-like items) course_ids = [] # You can extract course-like items from skills or create separate logic # 6. Store project details (if available) project_ids = [] if parsed_data.get("projects") is not None and isinstance(parsed_data["projects"], list): project_ids = self.store_project_details(jobseeker_id, parsed_data["projects"]) # 7. Store certification details (if available) certification_ids = [] if parsed_data.get("certifications") is not None and isinstance(parsed_data["certifications"], list): certification_ids = self.store_certification_details(jobseeker_id, parsed_data["certifications"]) result = { "success": True, "jobseeker_id": jobseeker_id, "stored_records": { "basic_details_id": basic_id, "education_ids": education_ids, "employment_ids": employment_ids, "course_ids": course_ids, "project_ids": project_ids, "certification_ids": certification_ids } } return result except Exception as e: error_msg = f"Error storing CV data: {str(e)}" return { "success": False, "error": error_msg }