cpps_core.py

"""CPPS core — shared parsing, skill normalization, scoring, and debug helpers.

Single source of truth for the production pipeline and the import-backed notebooks.

Phase 1: L6 (centralized blacklist/whitelist), L7 (unified alias map),
         L9 (debug log), L10 (pipeline consistency).
Phase 2: L1 (adaptive threshold), L3 (semantic synonyms),
         L4 (POS filtering), L12 (dynamic top_n).
"""

from __future__ import annotations

import logging
import re
from collections import Counter
from dataclasses import dataclass, field
from typing import Dict, Iterable, List, Optional, Tuple

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# L6 — Centralized noise blacklist (union of prior NB3 + NB8 sets)
# ---------------------------------------------------------------------------

NON_SKILL_WORDS: set = {
    # JD framing
    "hiring", "looking", "required", "preferred", "experience", "years",
    "education", "degree", "field", "related", "bachelor", "master",
    "candidate", "position", "role", "opportunity",
    # Connective / stopword-like
    "of", "in", "and", "or", "the", "a", "an", "to", "for", "with",
    "on", "at", "by", "is", "are", "we", "our", "their", "your",
    # Generic nouns that leak through
    "skills", "data", "analysis", "deep", "platforms", "team",
    "work", "tools", "strong", "good", "best", "knowledge",
    # Section labels
    "required skills", "preferred skills", "related field",
    "qualifications", "requirements", "responsibilities",
    "team player", "ability", "experience in", "knowledge of",
    "good to have", "nice to have", "must have",
    "certifications", "overview", "company", "about", "benefits",
    "perks", "salary", "date", "hands", "engineers", "implement",
    "controls", "specifications",
    "load", "e.g", "eg",
}

JD_NOISE_PHRASES: set = {
    "company overview", "role overview", "key responsibilities",
    "required qualifications", "preferred qualifications",
    "technical skills", "overview", "responsibilities",
    "qualifications", "requirements", "benefits", "perks",
    "job description", "about us", "about the role",
    "what you will do", "what we look for",
    "job title", "must have", "good to have", "nice to have",
    "compensation", "work mode", "hybrid", "remote", "onsite",
    "about the company", "minimum qualifications", "basic qualifications",
    "years of experience", "strong communication", "excellent communication",
    "or related field", "related field", "and related field",
    "about the team", "job responsibilities", "skills and capabilities",
    "maintain consistent communication", "including personal banking",
    "equal opportunity employer", "sexual orientation", "national origin",
    "race", "religion", "gender", "age", "color",
    "business requirements", "technical specifications", "strong controls",
    "bachelor s degree", "hands on experience",
    "deployment of data", "analytics technologies", "strategy to reporting",
}

# Substring-filtered noise ("ability to work with X" -> reject anything
# containing "ability", "experience", etc.)
SKILL_NOISE_SUBSTRINGS: List[str] = [
    "ability", "experience", "knowledge",
    "responsible", "working with", "understanding",
    "proficiency", "familiarity", "expertise",
]

JD_NOISE_SUBSTRINGS: List[str] = [
    "about the",
    "job responsibilities",
    "responsibilit",
    "equal opportunity",
    "sexual orientation",
    "national origin",
    "interpersonal",
    "communication",
    "results-driven",
    "including ",
    "maintain consistent",
]

JD_TOKEN_BLACKLIST: set = {
    "you", "your", "we", "our", "us", "them", "they", "the", "a", "an",
    "this", "that", "these", "those", "both", "regards",
}

JD_ACTION_VERBS: set = {
    "collaborate", "analyze", "design", "develop", "monitor", "provide",
    "translate", "document", "possess", "create", "creating", "eagerly",
    "maintain", "conduct", "work", "stay", "transform", "test", "understand",
}

JD_GENERIC_TAIL_TOKENS: set = {
    "team", "teams", "needs", "objectives", "products", "users", "training",
    "support", "insights", "trends", "documentation", "capabilities",
    "levels", "assignments", "timelines", "efforts", "progress", "future",
    "growth", "engagement", "history", "solutions", "businesses", "consumers",
}

TECH_SIGNAL_TOKENS: set = {
    "access", "agile", "ai", "alteryx", "amazon", "analytics",
    "api", "artificial", "aws", "azure", "bi", "cloud", "dashboard",
    "data", "database", "docker", "excel", "etl", "gcp", "google",
    "java", "javascript", "kubernetes", "machine", "microsoft", "model",
    "mongodb", "mysql", "office", "oracle", "pipeline", "postgresql",
    "power", "python", "reporting", "scikit-learn", "snowflake", "sql",
    "tableau", "tensorflow", "teradata", "visualization",
}

JD_BUCKET_WEIGHTS: Dict[str, float] = {
    "required": 1.0,
    "preferred": 0.75,
    "optional": 0.5,
}

MATCH_SCORE_BY_TYPE: Dict[str, Tuple[float, float]] = {
    "exact": (1.0, 1.0),
    "alias": (1.0, 1.0),
    "synonym": (0.9, 0.9),
    "semantic": (0.55, 0.80),
    "partial": (0.30, 0.50),
}

SKILL_SCORE_COMPONENT_WEIGHTS: Dict[str, float] = {
    "weighted_jd_coverage": 0.45,
    "avg_match_confidence": 0.35,
    "resume_pool_coverage": 0.20,
}

SKILL_CLUSTERS: Dict[str, List[str]] = {
    "python_ecosystem": ["python", "pandas", "numpy", "scipy", "matplotlib"],
    "cloud_platforms": ["aws", "azure", "gcp", "cloud computing", "google cloud platform"],
    "web_frontend": ["react", "angular", "vue", "html", "css", "javascript"],
    "databases": ["sql", "mysql", "postgresql", "mongodb", "redis"],
}

# Soft skills — not matchable via SBERT against technical JDs.
SOFT_SKILL_BLACKLIST: set = {
    "communication", "teamwork", "leadership", "problem-solving",
    "problem solving", "collaboration", "adaptability", "creativity",
    "time management", "critical thinking", "interpersonal skills",
}

# Whitelist: phrases matching these patterns are ALWAYS kept even if short
# or stopword-ish (e.g. "C++", "C#", "SQL", "React.js").
SHORT_SKILL_WHITELIST: set = {
    "sql", "aws", "gcp", "api", "apis", "etl", "bi", "ui", "ux",
    "ml", "ai", "nlp", "cv", "sap", "erp", "crm", "css", "html",
}

WHITELIST_PATTERNS: List[re.Pattern] = [
    re.compile(r"^[a-z]+\+\+$"),          # C++
    re.compile(r"^[a-z]+#$"),              # C#, F#
    re.compile(r"^[a-z]+\.[a-z]+$"),       # react.js, node.js, vue.js
]


def is_whitelisted(phrase: str) -> bool:
    """Return True if phrase matches a tech-skill whitelist pattern."""
    if not phrase:
        return False
    p = phrase.strip().lower()
    return p in SHORT_SKILL_WHITELIST or any(pat.match(p) for pat in WHITELIST_PATTERNS)


# ---------------------------------------------------------------------------
# L7 — Unified alias / normalization map
# ---------------------------------------------------------------------------

SKILL_ALIAS_MAP: Dict[str, str] = {
    # Abbreviations
    "ml": "machine learning",
    "dl": "deep learning",
    "nlp": "natural language processing",
    "cv": "computer vision",
    "ai": "artificial intelligence",
    "js": "javascript",
    "ts": "typescript",
    "py": "python",
    "tf": "tensorflow",
    "k8s": "kubernetes",
    "gcp": "google cloud platform",
    "aws": "amazon web services",
    "gke": "google kubernetes engine",
    "eks": "amazon elastic kubernetes service",
    "ci/cd": "continuous integration",
    "oop": "object-oriented programming",
    "db": "database",
    "rdbms": "relational database",
    "eda": "exploratory data analysis",

    # Common variants / spellings
    "sklearn": "scikit-learn",
    "scikit learn": "scikit-learn",
    "postgres": "postgresql",
    "mongo": "mongodb",
    "powerbi": "power bi",
    "power-bi": "power bi",
    "reactjs": "react",
    "react.js": "react",
    "nodejs": "node.js",
    "node js": "node.js",
    "vuejs": "vue",
    "vue.js": "vue",
    "angularjs": "angular",
    "angular.js": "angular",
    "c++": "cpp",
    "c#": "csharp",
    ".net": "dotnet",
    "dot net": "dotnet",
    "tensor flow": "tensorflow",
    "py torch": "pytorch",
    "torch": "pytorch",

    # Conceptual equivalences (lightweight)
    "data wrangling": "data manipulation",
    "data cleaning": "data preprocessing",
    "etl": "data pipeline",
    "rest api": "rest",
    "restful api": "rest",
    "restful apis": "rest",
    "amazon web services": "aws",
    "google cloud": "google cloud platform",
    "google cloud services": "google cloud platform",
    "postgres sql": "postgresql",
    "mongo db": "mongodb",
    "expressjs": "express",
    "react js": "react",
    "java script": "javascript",
    "scikit learn library": "scikit-learn",

    # Frontend / backend / full-stack variants
    "frontend": "frontend development",
    "front-end": "frontend development",
    "front end": "frontend development",
    "backend": "backend development",
    "back-end": "backend development",
    "back end": "backend development",
    "full-stack": "full stack development",
    "fullstack": "full stack development",
    "full stack": "full stack development",
    "dashboarding tools": "dashboarding",
    "dashboard tools": "dashboarding",
    "rest apis": "rest",
}


def normalize_skill(skill: str) -> str:
    """Resolve abbreviations/variants to canonical form."""
    if not skill:
        return ""
    key = skill.lower().strip()
    return SKILL_ALIAS_MAP.get(key, key)


# ---------------------------------------------------------------------------
# Text normalization (shared by NB3 + NB8)
# ---------------------------------------------------------------------------

def normalize_text_for_skills(text: Optional[str]) -> str:
    """Normalize skill-focused text while preserving skill-list boundaries."""
    if not isinstance(text, str) or not text.strip():
        return ""
    cleaned = text.strip().lower()
    cleaned = cleaned.replace("\r\n", "\n").replace("\r", "\n")
    cleaned = re.sub(r"[\n;|]+", ", ", cleaned)
    cleaned = re.sub(r"[^a-z0-9,\+#\./\- ]+", " ", cleaned)
    cleaned = re.sub(r"\s*,\s*", ", ", cleaned)
    cleaned = re.sub(r"\s+", " ", cleaned)
    return cleaned.strip(" ,")


# Backward-compat alias for NB3 callers using `normalize_text`.
normalize_text = normalize_text_for_skills


def normalize_phrase(phrase: Optional[str]) -> str:
    """Normalize a single skill phrase for comparison and output."""
    normalized = normalize_text_for_skills(phrase)
    normalized = normalized.replace(",", " ")
    normalized = re.sub(r"\s+", " ", normalized)
    return normalized.strip()


# ---------------------------------------------------------------------------
# Skill filtering / cleaning
# ---------------------------------------------------------------------------

def is_valid_skill(phrase: str) -> bool:
    """Return True if phrase passes the unified blacklist/whitelist filter."""
    if not phrase:
        return False
    p = phrase.strip().lower()
    if len(p) < 2:
        return False
    if is_whitelisted(p):
        return True
    if p in NON_SKILL_WORDS:
        return False
    if p in JD_NOISE_PHRASES:
        return False
    if p in SOFT_SKILL_BLACKLIST:
        return False
    if len(p.split()) > 3:
        return False
    if any(sub in p for sub in SKILL_NOISE_SUBSTRINGS):
        return False
    return True


def is_valid_jd_skill(phrase: str) -> bool:
    """JD-specific filter that removes legal/boilerplate and soft-skill noise."""
    p = normalize_skill(normalize_phrase(phrase))
    if not is_valid_skill(p):
        return False
    tokens = p.split()
    if any(token in JD_TOKEN_BLACKLIST for token in tokens):
        return False
    if tokens and tokens[0] in JD_ACTION_VERBS:
        return False
    if tokens and tokens[-1] in JD_GENERIC_TAIL_TOKENS:
        return False
    if any(sub in p for sub in JD_NOISE_SUBSTRINGS):
        return False
    if re.search(r"\b(?:gender|religion|race|color|age|disability)\b", p):
        return False
    if re.search(r"\b\d+\+?\s*years?\b", p):
        return False
    if re.search(r"\b(?:engineers?|implement|hands?|degree|bachelor|master|phd|date)\b", p):
        return False
    if len(tokens) == 1 and not (
        is_whitelisted(p) or
        p in SKILL_ALIAS_MAP or
        p in SKILL_ALIAS_MAP.values() or
        p in SEMANTIC_SYNONYMS or
        p in _SYNONYM_REVERSE or
        p in TECH_SIGNAL_TOKENS
    ):
        return False
    return True


def clean_skills(skills: Iterable[str], cap: Optional[int] = 30) -> List[str]:
    """Filter noise, normalize aliases, deduplicate, cap length."""
    cleaned: List[str] = []
    for s in skills:
        if not s:
            continue
        p = s.lower().strip()
        if not is_valid_skill(p):
            continue
        p = normalize_skill(p)
        if p not in cleaned:
            cleaned.append(p)
    if cap is None:
        return cleaned
    return cleaned[:cap]


def has_technical_signal(phrase: str) -> bool:
    """Return True when a phrase contains at least one technical anchor."""
    p = normalize_skill(normalize_phrase(phrase))
    if not p:
        return False
    if is_whitelisted(p) or p in SKILL_ALIAS_MAP or p in SKILL_ALIAS_MAP.values():
        return True
    if p in SEMANTIC_SYNONYMS or p in _SYNONYM_REVERSE:
        return True
    tokens = {tok for tok in re.split(r"[\s/\-]+", p) if tok}
    return bool(tokens & TECH_SIGNAL_TOKENS)


def compute_token_overlap_ratio(phrase_a: str, phrase_b: str) -> float:
    """Compute overlap on normalized, non-trivial tokens."""
    stop_tokens = NON_SKILL_WORDS | JD_TOKEN_BLACKLIST | {"s"}
    tokens_a = {
        tok for tok in re.split(r"[\s/\-]+", normalize_phrase(phrase_a))
        if tok and tok not in stop_tokens
    }
    tokens_b = {
        tok for tok in re.split(r"[\s/\-]+", normalize_phrase(phrase_b))
        if tok and tok not in stop_tokens
    }
    if not tokens_a or not tokens_b:
        return 0.0
    return len(tokens_a & tokens_b) / float(max(len(tokens_a), len(tokens_b)))


def _cluster_names_for_skill(phrase: str) -> set:
    tokens = {tok for tok in re.split(r"[\s/\-]+", normalize_skill(normalize_phrase(phrase))) if tok}
    normalized = normalize_skill(normalize_phrase(phrase))
    clusters = set()
    for name, members in SKILL_CLUSTERS.items():
        normalized_members = {normalize_skill(normalize_phrase(member)) for member in members}
        member_tokens = set().union(*(member.split() for member in normalized_members))
        if normalized in normalized_members or tokens & member_tokens:
            clusters.add(name)
    return clusters


def skills_share_cluster(left: str, right: str) -> bool:
    """Return True when two skills belong to at least one ontology cluster."""
    return bool(_cluster_names_for_skill(left) & _cluster_names_for_skill(right))


def clamp_match_score(match_type: str, similarity: float) -> float:
    """Clamp score into the configured range for a match tier."""
    low, high = MATCH_SCORE_BY_TYPE.get(match_type, (0.0, 1.0))
    if low == high:
        return round(low, 4)
    bounded = max(low, min(float(similarity or 0.0), high))
    return round(bounded, 4)


def split_skill_line(raw_line: str) -> List[str]:
    """Split multi-skill resume lines ('Languages: Python, Java') into tokens.

    Skips soft-skills categories entirely (non-technical, not useful).
    """
    if not raw_line:
        return []
    if ":" in raw_line:
        prefix = raw_line.split(":", 1)[0].strip().lower()
        if "soft" in prefix:
            return []
        raw_line = raw_line.split(":", 1)[1]
    tokens = re.split(r"[,;|]+", raw_line)
    results: List[str] = []
    for token in tokens:
        token = token.strip()
        if token and len(token) >= 2:
            results.append(token)
    return results


def deduplicate_phrases(phrases: Iterable[str]) -> List[str]:
    """Remove exact and substring-redundant duplicates while preserving order."""
    unique: List[str] = []
    for phrase in phrases:
        np_ = normalize_phrase(phrase)
        if len(np_) < 2 or np_ in NON_SKILL_WORDS or np_ in unique:
            continue
        redundant = False
        to_remove: List[str] = []
        np_tokens = np_.split()
        for existing in unique:
            existing_tokens = existing.split()
            if np_ == existing:
                redundant = True
                break
            if np_ in existing and len(np_tokens) <= len(existing_tokens):
                to_remove.append(existing)
            elif existing in np_ and len(existing_tokens) <= len(np_tokens):
                redundant = True
                break
        if redundant:
            continue
        if to_remove:
            unique = [item for item in unique if item not in to_remove]
        unique.append(np_)
    return unique


def merge_unique_skills(*skill_groups: Iterable[str], cap: Optional[int] = None) -> List[str]:
    """Merge multiple skill collections into a cleaned, ordered list."""
    merged: List[str] = []
    for group in skill_groups:
        for phrase in group or []:
            normalized = normalize_skill(normalize_phrase(phrase))
            if normalized and is_valid_skill(normalized) and normalized not in merged:
                merged.append(normalized)
                if cap is not None and len(merged) >= cap:
                    return merged
    return merged


# ---------------------------------------------------------------------------
# L8 — Multi-source skill aggregation helper
# ---------------------------------------------------------------------------

_TECH_STACK_RE = re.compile(r"tech\s*stack\s*[:\-]\s*(.+)", re.IGNORECASE)

RESUME_SECTION_ALIASES: Dict[str, Tuple[str, ...]] = {
    "skills": (
        "skills", "technical skills", "core skills", "key skills", "skill set",
        "technical expertise", "technologies", "tools and technologies",
        "tools & technologies", "programming languages", "languages",
        "competencies", "core competencies", "technical proficiency",
        "technical toolkit", "technology stack", "tech stack",
    ),
    "projects": (
        "project", "projects", "academic project", "academic projects",
        "personal project", "personal projects", "key projects",
        "project experience", "notable projects", "project work",
        "projects & achievements", "major projects", "side projects",
        "project details", "project detail",
    ),
    "certifications": (
        "certification", "certifications", "certificate", "certificates",
        "licenses & certifications", "professional certifications",
        "courses & certifications", "online courses", "training",
        "courses", "professional development",
    ),
    "internships": (
        "internship", "internships", "internship experience",
        "industrial training", "industry experience", "summer internship",
        "summer internships", "internship & training", "internship details",
    ),
    "work_experience": (
        "work experience", "experience", "professional experience",
        "employment history", "employment", "career history",
        "relevant experience", "job experience", "work history",
    ),
    "education": (
        "education", "educational background", "academic background",
        "academic qualification", "academic qualifications", "qualifications",
        "academic details", "scholastic details", "academic history",
    ),
    "achievements": (
        "achievement", "achievements", "accomplishment", "accomplishments",
        "award", "awards", "honors", "honours", "awards & honors",
        "awards & achievements", "recognition",
        "extra-curricular activities", "extracurricular activities",
        "extracurriculars", "leadership", "activities", "competition",
        "competitions", "hackathon", "hackathons",
    ),
}


def _compile_section_patterns() -> Dict[str, re.Pattern]:
    patterns: Dict[str, re.Pattern] = {}
    for section_key, aliases in RESUME_SECTION_ALIASES.items():
        escaped = "|".join(re.escape(alias) for alias in aliases)
        patterns[section_key] = re.compile(rf"^\s*(?:{escaped})\s*[:\-]?\s*$", re.IGNORECASE)
    return patterns


SECTION_PATTERNS: Dict[str, re.Pattern] = _compile_section_patterns()


def split_resume_into_sections(raw_text: str) -> Dict[str, List[str]]:
    """Split resume text into structured sections using shared header patterns."""
    sections = {key: [] for key in RESUME_SECTION_ALIASES}
    if not raw_text:
        return sections

    current_section = None
    for line in raw_text.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        matched_section = None
        for section_key, pattern in SECTION_PATTERNS.items():
            if pattern.match(stripped):
                matched_section = section_key
                break
        if matched_section:
            current_section = matched_section
            continue
        if current_section is not None:
            sections[current_section].append(stripped)
    return sections


def extract_skills_from_section(section_entries: Iterable) -> List[str]:
    """Pull skill-like tokens from a free-form section (projects, internships,
    work_experience). Looks for 'Tech Stack:' / 'Technologies:' / 'Skills:'
    lines and comma-separated lists.
    """
    results: List[str] = []
    if not section_entries:
        return results
    for entry in section_entries:
        text = entry if isinstance(entry, str) else (
            " ".join(str(v) for v in entry.values()) if isinstance(entry, dict) else ""
        )
        if not text:
            continue
        for line in text.splitlines():
            line = line.strip()
            m = _TECH_STACK_RE.search(line)
            payload = None
            if m:
                payload = m.group(1)
            elif ":" in line and any(
                kw in line.lower().split(":", 1)[0]
                for kw in ("technolog", "skill", "stack", "tools")
            ):
                payload = line.split(":", 1)[1]
            if payload:
                for tok in split_skill_line(payload):
                    p = normalize_skill(normalize_phrase(tok))
                    if is_valid_skill(p) and p not in results:
                        results.append(p)
    return results


_KNOWN_FALLBACK_SKILLS: Optional[List[str]] = None


def _get_known_fallback_skills() -> List[str]:
    global _KNOWN_FALLBACK_SKILLS
    if _KNOWN_FALLBACK_SKILLS is None:
        phrases = (
            list(SKILL_ALIAS_MAP.keys()) +
            list(SKILL_ALIAS_MAP.values()) +
            list(SEMANTIC_SYNONYMS.keys()) +
            [syn for syns in SEMANTIC_SYNONYMS.values() for syn in syns]
        )
        _KNOWN_FALLBACK_SKILLS = sorted({
            normalize_skill(phrase)
            for phrase in phrases
            if phrase and is_valid_skill(normalize_phrase(phrase))
        }, key=lambda item: (-len(item.split()), -len(item), item))
    return list(_KNOWN_FALLBACK_SKILLS)


def extract_skills_from_full_text(text: str, cap: int = 25) -> List[str]:
    """Fallback skill mining for resumes without reliable sectioning."""
    if not text or not text.strip():
        return []

    candidates: List[str] = []
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        for chunk in re.split(r"[,;|\u2022/\t]+", stripped):
            phrase = normalize_skill(normalize_phrase(chunk))
            if phrase and is_valid_skill(phrase):
                candidates.append(phrase)

    normalized_text = f" {normalize_text_for_skills(text)} "
    for phrase in _get_known_fallback_skills():
        if f" {phrase} " in normalized_text:
            candidates.append(phrase)

    return clean_skills(deduplicate_phrases(candidates), cap=cap)


def clean_jd_skill_phrases(phrases: Iterable[str], cap: int = 40) -> List[str]:
    """Normalize and filter JD skill phrases more aggressively than resume skills."""
    cleaned: List[str] = []
    for phrase in phrases or []:
        normalized = normalize_skill(normalize_phrase(phrase))
        if not normalized or not is_valid_jd_skill(normalized):
            continue
        if normalized not in cleaned:
            cleaned.append(normalized)
            if len(cleaned) >= cap:
                break
    return cleaned


# ---------------------------------------------------------------------------
# L9 — Debug & transparency layer
# ---------------------------------------------------------------------------

@dataclass
class DebugLog:
    """Structured debug log accumulated through the skill-extraction pipeline.

    Attach to a pipeline run, populate at each stage, and emit under a
    `_debug` key in the API response when `debug=True`.
    """
    raw_resume_skills: List[str] = field(default_factory=list)
    filtered_resume_skills: List[str] = field(default_factory=list)
    normalized_resume_skills: List[str] = field(default_factory=list)
    jd_skills_raw: List[str] = field(default_factory=list)
    jd_skills_filtered: List[str] = field(default_factory=list)
    required_skills: List[str] = field(default_factory=list)
    preferred_skills: List[str] = field(default_factory=list)
    optional_skills: List[str] = field(default_factory=list)
    rejected_jd_phrases: List[Dict] = field(default_factory=list)
    exact_matches: List[Dict] = field(default_factory=list)
    alias_matches: List[Dict] = field(default_factory=list)
    semantic_matches: List[Dict] = field(default_factory=list)
    synonym_matches: List[Dict] = field(default_factory=list)
    partial_matches: List[Dict] = field(default_factory=list)
    rejected_matches: List[Dict] = field(default_factory=list)
    final_missing: List[str] = field(default_factory=list)
    resume_skill_source_counts: Dict[str, int] = field(default_factory=dict)
    jd_bucket_counts: Dict[str, int] = field(default_factory=dict)
    match_type_counts: Dict[str, int] = field(default_factory=dict)
    skills_score_S_sk: float = 0.0
    weighted_jd_total: float = 0.0
    weighted_match_total: float = 0.0
    weighted_jd_coverage: float = 0.0
    avg_match_confidence: float = 0.0
    resume_pool_coverage: float = 0.0
    empty_resume_sections: List[str] = field(default_factory=list)
    notes: List[str] = field(default_factory=list)

    def note(self, message: str) -> None:
        self.notes.append(message)

    def to_dict(self) -> Dict:
        return {
            "raw_resume_skills": list(self.raw_resume_skills),
            "filtered_resume_skills": list(self.filtered_resume_skills),
            "normalized_resume_skills": list(self.normalized_resume_skills),
            "jd_skills_raw": list(self.jd_skills_raw),
            "jd_skills_filtered": list(self.jd_skills_filtered),
            "raw_jd_skills": list(self.jd_skills_raw),
            "filtered_jd_skills": list(self.jd_skills_filtered),
            "required_skills": list(self.required_skills),
            "preferred_skills": list(self.preferred_skills),
            "optional_skills": list(self.optional_skills),
            "rejected_jd_phrases": list(self.rejected_jd_phrases),
            "exact_matches": list(self.exact_matches),
            "alias_matches": list(self.alias_matches),
            "semantic_matches": list(self.semantic_matches),
            "synonym_matches": list(self.synonym_matches),
            "partial_matches": list(self.partial_matches),
            "rejected_matches": list(self.rejected_matches),
            "final_missing": list(self.final_missing),
            "resume_skill_source_counts": dict(self.resume_skill_source_counts),
            "jd_bucket_counts": dict(self.jd_bucket_counts),
            "match_type_counts": dict(self.match_type_counts),
            "skills_score_S_sk": float(self.skills_score_S_sk),
            "weighted_jd_total": float(self.weighted_jd_total),
            "weighted_match_total": float(self.weighted_match_total),
            "weighted_jd_coverage": float(self.weighted_jd_coverage),
            "avg_match_confidence": float(self.avg_match_confidence),
            "resume_pool_coverage": float(self.resume_pool_coverage),
            "empty_resume_sections": list(self.empty_resume_sections),
            "notes": list(self.notes),
        }


# ---------------------------------------------------------------------------
# L1 — Adaptive cosine-similarity threshold (Phase 2)
# ---------------------------------------------------------------------------

# Default fallback when adaptive thresholding is not used.
DEFAULT_MATCH_THRESHOLD = 0.75

# Length-band thresholds: shorter phrases need tighter matching (less context
# makes SBERT similarity noisier), longer phrases can tolerate lower scores
# because there is more semantic signal.
_THRESHOLD_BY_WORDS: List[Tuple[int, float]] = [
    (1, 0.65),   # single-word: "Python", "SQL" — strictest (short phrases are ambiguous)
    (2, 0.58),   # bigrams: "machine learning" — moderate
    (3, 0.52),   # 3+ words: "natural language processing" — more context allows lower threshold
]


def compute_adaptive_threshold(jd_phrase: str) -> float:
    """Return a cosine-similarity threshold adapted to *jd_phrase* length.

    Shorter JD phrases are inherently more ambiguous in SBERT space, so they
    require a higher similarity to count as a match.  Longer phrases carry
    more semantic context, allowing a lower threshold.
    """
    word_count = len(jd_phrase.strip().split())
    for max_words, threshold in _THRESHOLD_BY_WORDS:
        if word_count <= max_words:
            return threshold
    return _THRESHOLD_BY_WORDS[-1][1]  # longest band


# ---------------------------------------------------------------------------
# L3 — Semantic synonym map (Phase 2)
# ---------------------------------------------------------------------------

SEMANTIC_SYNONYMS: Dict[str, List[str]] = {
    # Data / ML tools
    "pandas": ["data manipulation", "dataframes", "data wrangling"],
    "numpy": ["numerical computing", "numerical python", "array computing"],
    "scipy": ["scientific computing", "scientific python"],
    "matplotlib": ["data visualization", "plotting"],
    "seaborn": ["data visualization", "statistical plotting"],
    "scikit-learn": ["machine learning library", "sklearn"],
    "tensorflow": ["deep learning framework", "tf"],
    "pytorch": ["deep learning framework", "torch"],
    "keras": ["deep learning framework", "neural network library"],
    "opencv": ["computer vision library", "image processing"],

    # Databases
    "mongodb": ["nosql", "document database", "nosql database"],
    "postgresql": ["postgres", "relational database"],
    "mysql": ["relational database", "sql database"],
    "redis": ["in-memory database", "caching", "key-value store"],
    "elasticsearch": ["search engine", "full-text search"],

    # Web / frontend
    "react": ["reactjs", "react.js", "frontend framework"],
    "angular": ["angularjs", "angular.js", "frontend framework"],
    "vue": ["vuejs", "vue.js", "frontend framework"],
    "node.js": ["nodejs", "server-side javascript", "backend javascript"],
    "express": ["expressjs", "node framework", "web framework"],
    "django": ["python web framework", "web framework"],
    "flask": ["python web framework", "micro framework"],
    "fastapi": ["python web framework", "async web framework"],

    # DevOps / Cloud
    "docker": ["containerization", "containers", "container platform"],
    "kubernetes": ["k8s", "container orchestration", "orchestration"],
    "terraform": ["infrastructure as code", "iac"],
    "jenkins": ["ci/cd", "continuous integration", "build automation"],
    "github actions": ["ci/cd", "continuous integration"],
    "aws": ["amazon web services", "cloud computing"],
    "gcp": ["google cloud platform", "cloud computing"],
    "azure": ["microsoft azure", "cloud computing"],

    # Languages / general
    "python": ["py"],
    "javascript": ["js", "ecmascript"],
    "typescript": ["ts"],
    "java": ["jvm"],
    "golang": ["go programming", "go language"],

    # Concepts
    "deep learning": ["neural networks", "dl"],
    "machine learning": ["ml", "predictive modeling"],
    "natural language processing": ["nlp", "text mining", "text analytics"],
    "computer vision": ["cv", "image recognition", "image classification"],
    "data pipeline": ["etl", "data engineering", "data workflow"],
    "rest": ["rest api", "restful api", "restful apis"],
    "graphql": ["graph query language", "api query language"],
    "microservices": ["micro services", "service-oriented architecture"],
    "object-oriented programming": ["oop", "java", "cpp", "csharp", "polymorphism"],
    "software development": ["software engineering", "application development", "coding"],
    "web development": ["frontend development", "full stack development", "web application"],
    "dashboard development": ["data visualization", "reporting", "business intelligence"],
    "data cleaning": ["data preprocessing", "data wrangling", "data preparation"],
    "data extraction": ["data mining", "data scraping", "etl"],
    "predictive models": ["machine learning models", "predictive analytics", "ml models"],

    # Frontend / Full-stack
    "frontend development": ["frontend", "front-end development", "ui development", "client-side"],
    "backend development": ["backend", "back-end development", "server-side"],
    "full stack development": ["full stack", "fullstack", "full-stack", "frontend backend"],
    "rest apis": ["rest", "restful api", "api development", "rest api"],

    # BI / Dashboarding tools
    "tableau": ["data visualization", "business intelligence", "dashboarding"],
    "power bi": ["data visualization", "business intelligence", "powerbi", "dashboarding"],
    "dashboarding": ["dashboard development", "data visualization", "reporting", "dashboarding tools"],
    "business reporting": ["data visualization", "reporting", "business intelligence"],
}

# Build reverse lookup: synonym -> set of canonical skills that claim it.
_SYNONYM_REVERSE: Dict[str, List[str]] = {}
for _canonical, _syns in SEMANTIC_SYNONYMS.items():
    for _syn in _syns:
        _s = _syn.lower()
        _SYNONYM_REVERSE.setdefault(_s, []).append(_canonical)


def get_synonyms(phrase: str) -> List[str]:
    """Return all known synonyms for *phrase* (canonical or reverse lookup)."""
    key = phrase.strip().lower()
    # Forward: phrase is a canonical skill
    results = list(SEMANTIC_SYNONYMS.get(key, []))
    # Reverse: phrase is a synonym of some canonical skill(s)
    for canonical in _SYNONYM_REVERSE.get(key, []):
        if canonical not in results:
            results.append(canonical)
    return results


def is_synonym_match(phrase_a: str, phrase_b: str) -> bool:
    """Return True if *phrase_a* and *phrase_b* are synonyms of each other."""
    a = phrase_a.strip().lower()
    b = phrase_b.strip().lower()
    if a == b:
        return True
    # a's synonyms contain b?
    if b in [s.lower() for s in get_synonyms(a)]:
        return True
    # b's synonyms contain a?
    if a in [s.lower() for s in get_synonyms(b)]:
        return True
    return False


# ---------------------------------------------------------------------------
# L4 — POS-based skill filtering (Phase 2)
# ---------------------------------------------------------------------------

# Lazy-loaded spaCy model.  We only import once and cache it.  If the model
# is not installed the feature degrades gracefully (no POS filtering).
_spacy_nlp = None  # type: ignore[assignment]
_SPACY_AVAILABLE: Optional[bool] = None


def _load_spacy():
    """Load spaCy en_core_web_sm lazily. Sets _SPACY_AVAILABLE flag."""
    global _spacy_nlp, _SPACY_AVAILABLE
    if _SPACY_AVAILABLE is not None:
        return
    try:
        import spacy
        _spacy_nlp = spacy.load("en_core_web_sm", disable=["parser", "ner"])
        _SPACY_AVAILABLE = True
        logger.info("spaCy en_core_web_sm loaded — POS filtering enabled.")
    except Exception:
        _SPACY_AVAILABLE = False
        logger.info("spaCy model not available — POS filtering disabled (graceful fallback).")


def filter_non_skill_phrases_pos(phrases: List[str]) -> List[str]:
    """Keep only noun-bearing phrases using spaCy POS tagging.

    If spaCy is unavailable, returns the input unchanged (graceful fallback).
    """
    _load_spacy()
    if not _SPACY_AVAILABLE or _spacy_nlp is None:
        return list(phrases)

    filtered: List[str] = []
    for phrase in phrases:
        # Always keep whitelisted phrases (acronyms, C++, etc.)
        if is_whitelisted(phrase):
            filtered.append(phrase)
            continue
        doc = _spacy_nlp(phrase)
        # Keep if at least one token is a noun or proper noun
        has_noun = any(t.pos_ in ("NOUN", "PROPN") for t in doc)
        # Reject if ALL tokens are non-noun (verb, adverb, adjective, etc.)
        all_non_noun = all(t.pos_ in ("VERB", "ADV", "ADJ", "ADP", "DET", "AUX", "SCONJ", "CCONJ") for t in doc)
        if has_noun and not all_non_noun:
            filtered.append(phrase)
    return filtered


# ---------------------------------------------------------------------------
# L12 — Dynamic top_n tuning (Phase 2)
# ---------------------------------------------------------------------------

def compute_optimal_top_n(text: str, requested_top_n: int = 30) -> int:
    """Scale KeyBERT top_n with text complexity (unique-word proxy).

    Short texts get fewer extractions to avoid noise padding;
    rich texts get more to capture diverse skills.
    """
    if not text or not text.strip():
        return 3
    words = text.split()
    word_count = len(words)
    unique_words = len(set(w.lower() for w in words))
    if unique_words < 20:
        optimal = 5
    elif unique_words < 50:
        optimal = 15
    else:
        optimal = min(30, unique_words // 3)
    # Never exceed explicit request, floor at 3
    return max(3, min(optimal, requested_top_n, max(3, word_count // 4)))


def compute_phrase_specificity_weight(phrase: str) -> float:
    """Weight JD phrases by specificity so longer phrases matter slightly more."""
    word_count = len(normalize_phrase(phrase).split())
    if word_count <= 1:
        return 1.0
    if word_count == 2:
        return 1.25
    return 1.5


def compute_weighted_skill_score(
    jd_entries: Iterable[Dict],
    matched_pairs: Iterable[Dict],
    total_resume_phrases: int,
) -> Dict:
    """Compute the root-PDF weighted skill score and compatibility metrics."""
    jd_list: List[Dict] = []
    for entry in jd_entries or []:
        phrase = normalize_skill(normalize_phrase(entry.get("phrase", "")))
        bucket = str(entry.get("bucket", "required")).strip().lower() or "required"
        if not phrase or bucket not in JD_BUCKET_WEIGHTS:
            continue
        jd_list.append({"phrase": phrase, "bucket": bucket})

    matched_list = list(matched_pairs or [])
    total_weight = round(sum(JD_BUCKET_WEIGHTS[item["bucket"]] for item in jd_list), 6)
    matched_weight = 0.0
    matched_resume_phrases = 0
    confidences: List[float] = []
    seen_resume = set()
    match_type_counts: Counter = Counter()
    bucket_counts: Counter = Counter(item["bucket"] for item in jd_list)

    unmatched_by_phrase: Counter = Counter((item["phrase"], item["bucket"]) for item in jd_list)
    for pair in matched_list:
        jd_phrase = normalize_skill(normalize_phrase(pair.get("jd_phrase", "")))
        bucket = str(pair.get("jd_bucket", "required")).strip().lower() or "required"
        key = (jd_phrase, bucket)
        if not jd_phrase or key not in unmatched_by_phrase or unmatched_by_phrase[key] <= 0:
            continue
        unmatched_by_phrase[key] -= 1
        matched_weight += JD_BUCKET_WEIGHTS.get(bucket, 1.0)
        confidences.append(max(0.0, min(float(pair.get("similarity", 0.0) or 0.0), 1.0)))
        match_type = str(pair.get("match_type", "semantic")).strip().lower() or "semantic"
        match_type_counts[match_type] += 1
        resume_phrase = normalize_skill(normalize_phrase(pair.get("resume_phrase", "")))
        if resume_phrase and resume_phrase not in seen_resume:
            seen_resume.add(resume_phrase)
            matched_resume_phrases += 1

    avg_match_confidence = sum(confidences) / len(confidences) if confidences else 0.0
    resume_pool_coverage = (
        matched_resume_phrases / float(total_resume_phrases)
        if total_resume_phrases > 0 else 0.0
    )
    weighted_jd_coverage = (matched_weight / total_weight) if total_weight else 0.0
    score = (
        SKILL_SCORE_COMPONENT_WEIGHTS["weighted_jd_coverage"] * weighted_jd_coverage +
        SKILL_SCORE_COMPONENT_WEIGHTS["avg_match_confidence"] * avg_match_confidence +
        SKILL_SCORE_COMPONENT_WEIGHTS["resume_pool_coverage"] * resume_pool_coverage
    )

    return {
        "score": round(min(max(score, 0.0), 1.0), 4),
        "matched_count": len(matched_list),
        "total_jd_count": len(jd_list),
        "weighted_jd_total": round(total_weight, 4),
        "weighted_match_total": round(matched_weight, 4),
        "weighted_jd_coverage": round(weighted_jd_coverage, 4),
        "avg_match_confidence": round(avg_match_confidence, 4),
        "matched_resume_phrases": matched_resume_phrases,
        "total_resume_phrases": int(total_resume_phrases),
        "resume_pool_coverage": round(resume_pool_coverage, 4),
        "jd_bucket_counts": {bucket: int(bucket_counts.get(bucket, 0)) for bucket in JD_BUCKET_WEIGHTS},
        "match_type_counts": dict(match_type_counts),
        "score_component_weights": dict(SKILL_SCORE_COMPONENT_WEIGHTS),
    }


def compute_hybrid_skill_score(
    jd_phrases: Iterable[str],
    matched_pairs: Iterable[Dict],
    total_resume_phrases: int,
) -> Dict[str, float]:
    """Compute the authoritative hybrid skills score."""
    jd_list = [normalize_skill(normalize_phrase(item)) for item in jd_phrases or [] if normalize_phrase(item)]
    matched_list = list(matched_pairs or [])

    total_jd_weight = round(sum(compute_phrase_specificity_weight(item) for item in jd_list), 6)
    matched_jd_weight = 0.0
    matched_resume_phrases = 0
    confidences: List[float] = []
    seen_jd: Counter = Counter()
    seen_resume = set()

    for pair in matched_list:
        jd_phrase = normalize_skill(normalize_phrase(pair.get("jd_phrase", "")))
        resume_phrase = normalize_skill(normalize_phrase(pair.get("resume_phrase", "")))
        if not jd_phrase or jd_phrase not in jd_list:
            continue
        if seen_jd[jd_phrase] >= jd_list.count(jd_phrase):
            continue
        seen_jd[jd_phrase] += 1
        matched_jd_weight += compute_phrase_specificity_weight(jd_phrase)
        confidence = float(pair.get("similarity", 0.0) or 0.0)
        confidences.append(max(0.0, min(confidence, 1.0)))
        if resume_phrase and resume_phrase not in seen_resume:
            seen_resume.add(resume_phrase)
            matched_resume_phrases += 1

    weighted_jd_coverage = matched_jd_weight / total_jd_weight if total_jd_weight else 0.0
    avg_match_confidence = sum(confidences) / len(confidences) if confidences else 0.0
    resume_pool_coverage = (
        matched_resume_phrases / float(total_resume_phrases)
        if total_resume_phrases > 0 else 0.0
    )
    score = (
        SKILL_SCORE_COMPONENT_WEIGHTS["weighted_jd_coverage"] * weighted_jd_coverage +
        SKILL_SCORE_COMPONENT_WEIGHTS["avg_match_confidence"] * avg_match_confidence +
        SKILL_SCORE_COMPONENT_WEIGHTS["resume_pool_coverage"] * resume_pool_coverage
    )

    return {
        "score": round(min(score, 1.0), 4),
        "matched_count": len(matched_list),
        "total_jd_count": len(jd_list),
        "weighted_jd_total": round(total_jd_weight, 4),
        "weighted_match_total": round(matched_jd_weight, 4),
        "weighted_jd_coverage": round(weighted_jd_coverage, 4),
        "avg_match_confidence": round(avg_match_confidence, 4),
        "matched_resume_phrases": matched_resume_phrases,
        "total_resume_phrases": int(total_resume_phrases),
        "resume_pool_coverage": round(resume_pool_coverage, 4),
    }


__all__ = [
    # Phase 1 — L6/L7/L9/L10
    "NON_SKILL_WORDS",
    "JD_NOISE_PHRASES",
    "SKILL_NOISE_SUBSTRINGS",
    "SOFT_SKILL_BLACKLIST",
    "WHITELIST_PATTERNS",
    "SKILL_ALIAS_MAP",
    "JD_BUCKET_WEIGHTS",
    "MATCH_SCORE_BY_TYPE",
    "SKILL_SCORE_COMPONENT_WEIGHTS",
    "SKILL_CLUSTERS",
    "is_whitelisted",
    "is_valid_skill",
    "has_technical_signal",
    "normalize_skill",
    "normalize_text_for_skills",
    "normalize_text",
    "normalize_phrase",
    "clean_skills",
    "compute_token_overlap_ratio",
    "skills_share_cluster",
    "clamp_match_score",
    "split_skill_line",
    "deduplicate_phrases",
    "merge_unique_skills",
    "RESUME_SECTION_ALIASES",
    "SECTION_PATTERNS",
    "split_resume_into_sections",
    "extract_skills_from_section",
    "extract_skills_from_full_text",
    "clean_jd_skill_phrases",
    "DebugLog",
    # Phase 2 — L1/L3/L4/L12
    "DEFAULT_MATCH_THRESHOLD",
    "compute_adaptive_threshold",
    "SEMANTIC_SYNONYMS",
    "get_synonyms",
    "is_synonym_match",
    "filter_non_skill_phrases_pos",
    "compute_optimal_top_n",
    "compute_phrase_specificity_weight",
    "compute_weighted_skill_score",
    "compute_hybrid_skill_score",
    "is_valid_jd_skill",
]