Spaces:
Sleeping
Sleeping
| import json | |
| from bs4 import BeautifulSoup | |
| import requests | |
| from io import BytesIO | |
| from PyPDF2 import PdfReader | |
| from src.data_processor import DataProcessor | |
| from src.logger import Logger | |
| import logging | |
| from model.cv_job_matcher import CVJobMatcher | |
| class JDProcessor: | |
| """Processes job descriptions and candidate CVs from JSON data.""" | |
| def __init__(self, jd_file_path): | |
| self.jd_file_path = jd_file_path | |
| self.error_logger = Logger("Error Logger", see_time=True, console_log=True, level=logging.ERROR) | |
| def load_jd_data(self): | |
| """Loads the job description data from a JSON file.""" | |
| if isinstance(self.jd_file_path, dict): | |
| return self.jd_file_path | |
| try: | |
| with open(self.jd_file_path, "r", encoding="utf-8") as file: | |
| return json.load(file) | |
| except Exception as e: | |
| self.error_logger.log_message(f"Failed to load job description file: {e}") | |
| return None | |
| def clean_html(html_content): | |
| """Cleans HTML content by removing tags and replacing <br> with newlines.""" | |
| soup = BeautifulSoup(html_content, 'html.parser') | |
| for br in soup.find_all("br"): | |
| br.replace_with("\n") | |
| return soup.get_text() | |
| def extract_text_from_pdf(self, pdf_url): | |
| """Extracts text content from a PDF file at a given URL.""" | |
| if pdf_url.startswith("http"): | |
| return self._extract_text_from_remote_pdf(pdf_url) | |
| else: | |
| return self._extract_text_from_local_pdf(pdf_url) | |
| def _extract_text_from_remote_pdf(self, pdf_url): | |
| try: | |
| response = requests.get(pdf_url) | |
| pdf_reader = PdfReader(BytesIO(response.content)) | |
| pdf_text = " ".join(page.extract_text() for page in pdf_reader.pages) | |
| return pdf_text | |
| except Exception as e: | |
| self.error_logger.log_message(f"Error fetching PDF from {pdf_url}: {e}") | |
| return "" | |
| def _extract_text_from_local_pdf(self, pdf_path): | |
| try: | |
| with open(pdf_path, "rb") as file: | |
| pdf_reader = PdfReader(file) | |
| pdf_text = "".join(page.extract_text() for page in pdf_reader.pages) | |
| return pdf_text | |
| except Exception as e: | |
| self.error_logger.log_message(f"Error reading PDF from {pdf_path}: {e}") | |
| return "" | |
| def extract_jd_and_cvs(self): | |
| """Extracts job description and CVs from loaded JSON data.""" | |
| data = self.load_jd_data() | |
| if not data: | |
| return None | |
| jd_text = self.clean_html(data.get("description", "")) | |
| cvs = [] | |
| for job_user in data.get("JobUser", []): | |
| user_id = job_user.get("userId", "NO ID FOUND") | |
| cover_letter_text = self.clean_html(job_user.get("coverLetter", "")) | |
| pdf_url = job_user.get("cv") | |
| pdf_text = self.extract_text_from_pdf(pdf_url) if pdf_url else "" | |
| cvs.append({ | |
| "userId": user_id, | |
| "cover_letter": cover_letter_text, | |
| "cv_content": pdf_text | |
| }) | |
| return {"JD": jd_text, "CVs": cvs} | |
| if __name__ == "__main__": | |
| # jd_processor = JDProcessor("data/job_data.json") | |
| # print(jd_processor.extract_jd_and_cvs()) | |
| jd_processor = JDProcessor("test_data/jd.json") | |
| print(jd_processor.extract_jd_and_cvs()) | |