Spaces:
Sleeping
Sleeping
| # ============================================================================= | |
| # engine/jd_parser.py — Job Description Parser | |
| # ============================================================================= | |
| # Parses raw Job Description text into a structured dict that feeds directly | |
| # into the scoring engine (matcher.py). | |
| # | |
| # Handles JDs from three sources: | |
| # 1. Plain text string (copy-pasted from Naukri / LinkedIn) | |
| # 2. .txt file | |
| # 3. .pdf file | |
| # | |
| # The output schema matches the job_descriptions table in database.py exactly. | |
| # Every field that matcher.py reads is guaranteed to be present — either | |
| # extracted or set to a safe default — so downstream code never KeyErrors. | |
| # | |
| # PUBLIC API (the only functions called from outside this file): | |
| # | |
| # parse_jd_text(jd_text, jd_title) → dict | |
| # Takes raw JD string + a title, returns complete DB-ready dict. | |
| # | |
| # load_jd_from_file(file_path, jd_title) → dict | |
| # Reads a .txt or .pdf file, calls parse_jd_text, returns the same dict. | |
| # | |
| # load_all_jds_from_folder(folder_path) → list[dict] | |
| # Batch-loads every .txt and .pdf from a folder. | |
| # | |
| # INTERNAL FUNCTIONS (do not call directly): | |
| # _extract_required_skills(text) | |
| # _extract_preferred_skills(text) | |
| # _extract_required_experience(text) | |
| # _extract_education_requirement(text) | |
| # _extract_jd_title_from_text(text) | |
| # _clean_jd_text(text) | |
| # _section(text, keywords) | |
| # ============================================================================= | |
| import os | |
| import re | |
| import sys | |
| import json | |
| import pdfplumber | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| import config | |
| from engine.extractor import ( | |
| _SKILLS_TAXONOMY, | |
| _SOFT_SKILLS, | |
| _DEGREE_KEYWORDS, | |
| _find_section as _resume_section, | |
| ) | |
| # ============================================================================= | |
| # SECTION FINDER | |
| # ============================================================================= | |
| def _section(text: str, keywords: list) -> str: | |
| """ | |
| Extract a named section from JD text. | |
| JD section headers are typically written as: | |
| "Requirements:", "Required Skills:", "What We're Looking For:" | |
| This handles mixed capitalisation and trailing punctuation. | |
| Returns the section body as a string, or empty string if not found. | |
| """ | |
| lines = text.split("\n") | |
| result = [] | |
| active = False | |
| # A new section starts when a short line ends with : or is ALL CAPS | |
| section_boundary = re.compile( | |
| r"^(?:[A-Z][A-Za-z\s&/\-]{1,50}[:\-]|[A-Z\s]{4,40})$" | |
| ) | |
| for line in lines: | |
| stripped = line.strip() | |
| # Check if this line is a header matching our target keywords | |
| if any(kw.lower() in stripped.lower() for kw in keywords): | |
| active = True | |
| continue | |
| if active: | |
| # Stop at the next section boundary (unless it's the first line) | |
| if result and section_boundary.match(stripped) and len(stripped) > 3: | |
| break | |
| result.append(line) | |
| return "\n".join(result).strip() | |
| # ============================================================================= | |
| # TEXT CLEANER | |
| # ============================================================================= | |
| def _clean_jd_text(text: str) -> str: | |
| """ | |
| Normalise raw JD text for consistent downstream parsing. | |
| - Normalise line endings | |
| - Collapse 3+ blank lines to 2 | |
| - Remove zero-width / non-printable characters | |
| - Preserve all meaningful content | |
| """ | |
| if not text: | |
| return "" | |
| # Remove non-printable characters | |
| text = re.sub(r"[^\x09\x0A\x0D\x20-\x7E\u00A0-\uFFFF]", " ", text) | |
| # Normalise line endings | |
| text = text.replace("\r\n", "\n").replace("\r", "\n") | |
| # Strip trailing whitespace per line | |
| lines = [l.rstrip() for l in text.split("\n")] | |
| # Collapse excessive blank lines | |
| cleaned, blank_count = [], 0 | |
| for line in lines: | |
| if line.strip() == "": | |
| blank_count += 1 | |
| if blank_count <= 2: | |
| cleaned.append("") | |
| else: | |
| blank_count = 0 | |
| cleaned.append(line) | |
| return "\n".join(cleaned).strip() | |
| # ============================================================================= | |
| # TITLE EXTRACTION | |
| # ============================================================================= | |
| def _extract_jd_title_from_text(text: str) -> str: | |
| """ | |
| Infer a job title from the JD text when no explicit title is supplied. | |
| Strategy: | |
| 1. Look for "Job Title:", "Role:", "Position:" labels | |
| 2. Fall back to the first non-empty line (most JDs start with the role) | |
| 3. Return "Untitled JD" if nothing sensible is found | |
| """ | |
| # Pattern: "Job Title: Senior Data Analyst" | |
| label_match = re.search( | |
| r"(?:job\s+title|role|position|designation)\s*[:\-]\s*(.+)", | |
| text[:500], re.IGNORECASE | |
| ) | |
| if label_match: | |
| return label_match.group(1).strip().title() | |
| # Fall back to first meaningful line | |
| for line in text.split("\n"): | |
| line = line.strip() | |
| if (line and | |
| 3 < len(line) < 80 and | |
| not re.match(r"^(about|we are|company|dear|hello)", line, re.I)): | |
| return line.title() | |
| return "Untitled JD" | |
| # ============================================================================= | |
| # SKILLS EXTRACTION | |
| # ============================================================================= | |
| def _extract_required_skills(text: str) -> list: | |
| """ | |
| Extract explicitly required skills from the JD. | |
| Two-pass approach: | |
| Pass 1 — look in the Requirements / Must-Have section for taxonomy matches. | |
| Pass 2 — scan the full JD text for taxonomy matches, lower confidence. | |
| Skills found in Pass 1 are tagged required; Pass 2 additions are treated | |
| as inferred (both lists are returned merged, deduped, sorted). | |
| Returns: list of skill strings with Title casing. | |
| """ | |
| # Sections that signal required / mandatory skills | |
| required_section = _section(text, [ | |
| "requirements", "required skills", "must have", "mandatory", | |
| "what you need", "qualifications", "technical skills", | |
| "skills required", "key skills", "essential", | |
| ]) | |
| source = required_section if required_section else text | |
| found = set() | |
| text_lower = source.lower() | |
| for skill in _SKILLS_TAXONOMY: | |
| pattern = r"\b" + re.escape(skill) + r"\b" | |
| if re.search(pattern, text_lower): | |
| found.add(skill.upper() if len(skill) <= 3 else skill.title()) | |
| return sorted(found) | |
| def _extract_preferred_skills(text: str, required: list) -> list: | |
| """ | |
| Extract preferred / good-to-have skills that are NOT already in required. | |
| These come from sections like "Nice to Have", "Preferred", "Bonus Points". | |
| Returns: list of skill strings not in required. | |
| """ | |
| preferred_section = _section(text, [ | |
| "preferred", "nice to have", "good to have", "bonus", | |
| "desirable", "advantageous", "plus", "additional", | |
| ]) | |
| if not preferred_section: | |
| return [] | |
| required_lower = {s.lower() for s in required} | |
| found = set() | |
| section_lower = preferred_section.lower() | |
| for skill in _SKILLS_TAXONOMY: | |
| pattern = r"\b" + re.escape(skill) + r"\b" | |
| if re.search(pattern, section_lower): | |
| display = skill.upper() if len(skill) <= 3 else skill.title() | |
| if display.lower() not in required_lower: | |
| found.add(display) | |
| return sorted(found) | |
| # ============================================================================= | |
| # EXPERIENCE EXTRACTION | |
| # ============================================================================= | |
| def _extract_required_experience(text: str) -> float: | |
| """ | |
| Extract the minimum years of experience required by the JD. | |
| Handles patterns like: | |
| "3+ years", "minimum 5 years", "2-4 years experience", | |
| "at least 3 years", "3 to 5 years" | |
| Returns the lower bound as a float (e.g. "3-5 years" → 3.0). | |
| Returns 0.0 if no experience requirement is found. | |
| """ | |
| patterns = [ | |
| # "3+ years", "3 or more years" | |
| r"(\d+(?:\.\d+)?)\s*\+\s*years?", | |
| # "minimum 3 years", "at least 3 years" | |
| r"(?:minimum|min|at\s+least)\s+(\d+(?:\.\d+)?)\s*years?", | |
| # "3-5 years", "3 to 5 years" — take lower bound | |
| r"(\d+(?:\.\d+)?)\s*(?:–|-|to)\s*\d+\s*years?", | |
| # "3 years of experience" | |
| r"(\d+(?:\.\d+)?)\s*years?\s+(?:of\s+)?(?:experience|exp)", | |
| # "experience: 3 years" | |
| r"experience\s*[:\-]\s*(\d+(?:\.\d+)?)\s*years?", | |
| ] | |
| text_lower = text.lower() | |
| for pattern in patterns: | |
| match = re.search(pattern, text_lower) | |
| if match: | |
| try: | |
| return float(match.group(1)) | |
| except (ValueError, IndexError): | |
| continue | |
| return 0.0 | |
| # ============================================================================= | |
| # EDUCATION EXTRACTION | |
| # ============================================================================= | |
| def _extract_education_requirement(text: str) -> str: | |
| """ | |
| Extract the minimum education qualification required by the JD. | |
| Returns a human-readable string like "B.Tech / M.Tech" or | |
| "MBA / PGDM" rather than a structured dict, since JDs express | |
| this in highly variable ways. | |
| Returns empty string if no requirement found. | |
| """ | |
| edu_section = _section(text, [ | |
| "education", "qualification", "academic", "degree", | |
| ]) | |
| source = edu_section if edu_section else text[:1000] | |
| found = [] | |
| source_lower = source.lower() | |
| for deg in sorted(_DEGREE_KEYWORDS, key=len, reverse=True): | |
| if re.search(r"\b" + re.escape(deg) + r"\b", source_lower): | |
| found.append(deg.upper() if len(deg) <= 5 else deg.title()) | |
| # Deduplicate while preserving order | |
| seen, unique = set(), [] | |
| for d in found: | |
| if d.lower() not in seen: | |
| seen.add(d.lower()) | |
| unique.append(d) | |
| return " / ".join(unique[:4]) if unique else "" | |
| # ============================================================================= | |
| # MASTER PARSER | |
| # ============================================================================= | |
| def parse_jd_text(jd_text: str, jd_title: str = "") -> dict: | |
| """ | |
| Parse raw Job Description text into a structured, DB-ready dict. | |
| This is the primary public function. Call this with any JD text — | |
| whether pasted from Naukri, LinkedIn, or loaded from a file — | |
| and receive a complete dict matching the job_descriptions table schema. | |
| All fields are always present. Extraction failures produce safe defaults | |
| (empty list, 0.0, empty string) so matcher.py never encounters KeyErrors. | |
| Args: | |
| jd_text : Raw JD string. Can be noisy — cleaning is applied internally. | |
| jd_title : Human-readable job title. If empty, inferred from jd_text. | |
| Returns: | |
| { | |
| "jd_title" : str, | |
| "jd_text" : str, # cleaned | |
| "required_skills" : str, # JSON list | |
| "preferred_skills" : str, # JSON list | |
| "required_experience" : float, | |
| "education_requirement" : str, | |
| } | |
| Example: | |
| jd = open("data/jds/data_analyst.txt").read() | |
| record = parse_jd_text(jd, "Senior Data Analyst") | |
| jd_id = insert_jd(record) | |
| """ | |
| if not jd_text or not jd_text.strip(): | |
| raise ValueError( | |
| "jd_text is empty. Provide the full text of the job description." | |
| ) | |
| # Step 1 — Clean | |
| cleaned = _clean_jd_text(jd_text) | |
| # Step 2 — Title | |
| title = jd_title.strip() if jd_title.strip() else _extract_jd_title_from_text(cleaned) | |
| # Step 3 — Skills | |
| required_skills = _extract_required_skills(cleaned) | |
| preferred_skills = _extract_preferred_skills(cleaned, required_skills) | |
| # Step 4 — Experience | |
| required_experience = _extract_required_experience(cleaned) | |
| # Step 5 — Education | |
| education_requirement = _extract_education_requirement(cleaned) | |
| return { | |
| "jd_title" : title, | |
| "jd_text" : cleaned, | |
| "required_skills" : json.dumps(required_skills), | |
| "preferred_skills" : json.dumps(preferred_skills), | |
| "required_experience" : required_experience, | |
| "education_requirement" : education_requirement, | |
| } | |
| # ============================================================================= | |
| # FILE LOADERS | |
| # ============================================================================= | |
| def load_jd_from_file(file_path: str, jd_title: str = "") -> dict: | |
| """ | |
| Load a JD from a .txt or .pdf file and parse it. | |
| Args: | |
| file_path : Absolute or relative path to the JD file. | |
| jd_title : Optional title override. If empty, inferred from file | |
| content or derived from the filename. | |
| Returns: | |
| Parsed JD dict — same schema as parse_jd_text(). | |
| Raises: | |
| FileNotFoundError : If the file does not exist. | |
| ValueError : If the file format is not .txt or .pdf. | |
| """ | |
| ext = os.path.splitext(file_path)[1].lower() | |
| if ext not in (".txt", ".pdf", ".docx"): | |
| raise ValueError( | |
| f"Unsupported JD file format: '{ext}'\n" | |
| f"Supported formats: .txt, .pdf, .docx" | |
| ) | |
| if not os.path.exists(file_path): | |
| raise FileNotFoundError( | |
| f"JD file not found: {file_path}\n" | |
| f"Place JD files in: {config.JD_DIR}" | |
| ) | |
| # Read raw text | |
| if ext == ".txt": | |
| with open(file_path, "r", encoding="utf-8", errors="replace") as f: | |
| raw_text = f.read() | |
| elif ext == ".docx": | |
| try: | |
| from docx import Document as _DocxDocument | |
| doc = _DocxDocument(file_path) | |
| raw_text = "\n".join(p.text for p in doc.paragraphs if p.text.strip()) | |
| except ImportError: | |
| raise ImportError( | |
| "python-docx is required to read .docx JD files.\n" | |
| "Install it with: pip install python-docx" | |
| ) | |
| else: # .pdf | |
| pages = [] | |
| with pdfplumber.open(file_path) as pdf: | |
| for page in pdf.pages: | |
| text = page.extract_text() | |
| if text: | |
| pages.append(text) | |
| raw_text = "\n\n".join(pages) | |
| # Derive title from filename if not provided and not in content | |
| if not jd_title: | |
| import re as _re | |
| stem = os.path.splitext(os.path.basename(file_path))[0] | |
| stem = stem.replace("_", " ").replace("-", " ") | |
| # Remove trailing generic suffixes like "JD", "jd", "job description" | |
| stem = _re.sub(r'\s+[Jj][Dd]$', '', stem).strip() | |
| stem = _re.sub(r'\s+[Jj]ob\s+[Dd]escription$', '', stem, flags=_re.IGNORECASE).strip() | |
| jd_title = stem.title() | |
| return parse_jd_text(raw_text, jd_title) | |
| def load_all_jds_from_folder(folder_path: str) -> list: | |
| """ | |
| Batch-load and parse every .txt and .pdf JD from a folder. | |
| Files that fail to parse are included in the results with an | |
| "error" key so the pipeline continues without crashing. | |
| Args: | |
| folder_path : Path to folder containing JD files. | |
| Returns: | |
| List of dicts. Each dict is either a fully parsed JD record | |
| or an error record: | |
| { | |
| "file_name" : "data_analyst_jd.txt", | |
| "error" : "FileNotFoundError: ...", | |
| } | |
| """ | |
| if not os.path.exists(folder_path): | |
| raise FileNotFoundError( | |
| f"JD folder not found: {folder_path}\n" | |
| f"Create it and add .txt or .pdf JD files." | |
| ) | |
| supported = sorted([ | |
| f for f in os.listdir(folder_path) | |
| if os.path.splitext(f)[1].lower() in (".txt", ".pdf", ".docx") | |
| and not f.startswith(".") | |
| ]) | |
| if not supported: | |
| return [] | |
| results = [] | |
| for fname in supported: | |
| fpath = os.path.join(folder_path, fname) | |
| try: | |
| record = load_jd_from_file(fpath) | |
| record["file_name"] = fname | |
| results.append(record) | |
| except Exception as e: | |
| results.append({ | |
| "file_name": fname, | |
| "error" : f"{type(e).__name__}: {e}", | |
| }) | |
| return results |