engine/jd_parser.py

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