src/data_processor.py

import logging
from src.logger import Logger
from src.base_data_processor import BaseDataProcessor
from typing import Dict, List, Optional, Tuple


class DataProcessor(BaseDataProcessor):
    def __init__(self):
        super().__init__()
        self.info_logger = Logger("Data Processor INFO", see_time=True, console_log=False, level=logging.INFO)

    def process_data(self, job_data: Dict, cv_data_list: List[Dict[str, Optional[str]]]) -> Tuple[Dict[str, str], List[Dict[str, Optional[str]]]]:
        """
        Process job description and CV data to extract relevant information.

        Args:
            job_data (Dict): job description data
            cv_data_list (List[Dict[str, Optional[str]]]): list of CV data dicts with 'user_id', 'cover_letter', and 'cv_content'

        Returns:
            Tuple[Dict[str, str], List[Dict[str, Optional[str]]]]: formatted job description with extracted details and processed CV data
        """
        # Process and log job description details
        job_data = self.clean_text(job_data)
        job_skills = self.extract_skills(job_data)
        job_experience = self.extract_experience(job_data)
        job_degrees = self.extract_degrees(job_data)
        
        with open("job_data.txt", "w", encoding="utf-8") as f:
            f.write(str(job_data))
                
        updated_job_data = {
            "skills": job_skills,
            "experience": job_experience,
            "min_education": self.format_education(job_degrees),
        }
        self.info_logger.log_message(f"Updated job data: {updated_job_data}")
        
        # Process each CV entry in cv_data_list
        processed_cvs = []
        for cv_data in cv_data_list:
            job_id = cv_data.get("userId", "NO ID FOUND")
            
            
            cover_letter = self.clean_text(cv_data.get("cover_letter", ""))
            cv_content = self.clean_text(cv_data.get("cv_content", ""))
            
            with open(f"{job_id}_cv.txt", "w") as f:
                f.write(cv_content + " " + cover_letter)
            # self.info_logger.log_message(f"Processing CV data: {cv_data}")
            
            # Extract details from the combined CV text
            combined_cv_text = cv_content + " " + cover_letter 
            education = self.extract_degrees(combined_cv_text)
            # format education to a singular format such as b.s to bachelor of science
            education = self.format_education(education)
            cv_details = {
                "userId": job_id,
                "email": self.extract_email(combined_cv_text),
                "skills": self.extract_skills(combined_cv_text),
                "education": self.extract_degrees(combined_cv_text),
                "experience": self.extract_experience(combined_cv_text),
            }
            processed_cvs.append(cv_details)
            self.info_logger.log_message(f"Processed CV data: {cv_details}")
        
        self.info_logger.log_message(f"Processed job data and {len(processed_cvs)} CV(s)")
        return updated_job_data, processed_cvs