Spaces:
Sleeping
Sleeping
File size: 3,097 Bytes
14ca0c1 fea3c1a 14ca0c1 82d026c fea3c1a 14ca0c1 82d026c 14ca0c1 82d026c 14ca0c1 7972b6e 14ca0c1 fea3c1a 82d026c 7972b6e 14ca0c1 82d026c fea3c1a 14ca0c1 82d026c 7972b6e 14ca0c1 fea3c1a 14ca0c1 922b856 14ca0c1 922b856 14ca0c1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | 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
|