# ============================================================================= # 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