src/deterministic_extract.py

"""Deterministic, source-grounded ATS criterion extraction — NO external LLM.

This is the PRIMARY extraction path for V1. It turns a cleaned job description
into a full criterion inventory (schema below) using only local, deterministic
signals: the curated gazetteers + cue-word requirement classification in
`jd_analyzer.analyze_jd` (llm=None), plus section detection, years/title/cert
patterns, importance scoring, verbatim traceability, and longest-first dedup.

An external LLM is NEVER required for extraction, ranking, evidence matching,
coverage, or scoring — it is optional wording polish only (see resume_rewrite).
"""
from __future__ import annotations

import re
from typing import Dict, List

from .jd_analyzer import analyze_jd, _BUZZWORDS, _TOOLS
from .jd_preprocess import preprocess_jd

# Generic fragments never promoted on their own when a specific phrase exists.
# ponytail: trimmed to truly generic words only — "influence", "strategy",
# "innovation", "solutions" etc. carry ATS weight as PM competencies.
_GENERIC_ALONE = {
    "work", "problem", "manage", "communicate", "process",
    "team", "teams", "skills", "experience", "ability", "knowledge", "understanding",
    "responsibilities", "requirements", "role", "job", "candidate", "years",
    "environment", "priorities", "results", "data",
    "users",
    "quality", "success", "partners",
    # title/seniority modifiers — low standalone ATS value
    "staff", "senior", "lead", "principal", "junior", "associate", "director",
    "mentor",
    "legal", "autonomy", "acumen", "empathy", "passion", "trust", "joy",
    "decisions", "field",
    # ponytail: bare generic nouns/verbs — real ATS scanners search for specific
    # competencies, and these match nothing a recruiter would Boolean-search on.
    # Placing them is pure stuffing, so they must not enter the criterion set.
    "impact", "execution", "solutions", "collaborate", "reporting", "ownership",
    "messaging", "competitors", "influencing", "performance", "timelines",
    "stakeholders", "innovative", "solution", "deliverables", "initiatives",
}

# ── ATS keyword shape gate ───────────────────────────────────────────────────
# An ATS keyword is a NOUN PHRASE a recruiter could Boolean-search. JD prose
# chopped at comma/'and' boundaries produces clause fragments ("knack for
# precise", "what's not", "Ability to write", "running experiments (e.g") that
# can never match a résumé and only inflate the denominator. Reject by SHAPE,
# never by topic, so unfamiliar-but-real skills still get through.
_WEAK_HEADS = {
    "ability", "knack", "comfort", "willingness", "desire", "passion",
    "understanding", "knowledge", "familiarity", "proficiency", "expertise",
    "appetite", "bias", "sense", "love", "eagerness", "capacity", "aptitude",
    "exposure", "flair", "hunger", "drive", "commitment", "dedication",
}
# Bare imperative verbs that start a responsibility clause (not a keyword).
_VERB_HEADS = {
    "optimize", "monitor", "identify", "refine", "ensure", "own", "lead",
    "manage", "build", "create", "develop", "deliver", "conduct", "design",
    "run", "write", "define", "execute", "collaborate", "partner", "work",
    "help", "support", "maintain", "improve", "drive", "champion", "coordinate",
    "translate", "communicate", "present", "report", "track", "measure",
    "evaluate", "assess", "review", "prioritize", "align", "engage", "influence",
}
# Tokens that mark a subordinate clause — never inside a keyword.
_CLAUSE_MARKERS = {"what", "how", "why", "whether", "which", "who", "that",
                   "when", "where", "if", "because", "so", "then", "than"}
_PRONOUNS = {"your", "their", "our", "you", "we", "they", "it", "them", "its",
             "his", "her", "my", "me", "us"}
_TRAILING_JUNK = re.compile(
    r"\b(?:a plus|as well|and more|etc|and so on|or so|preferred|required)$", re.I)
# Function words that must never begin or end a keyword.
_STOP_EDGE_ALL = {"and", "or", "the", "a", "an", "to", "of", "for", "with", "in",
                  "on", "across", "using", "including", "etc", "such", "as",
                  "e.g", "i.e", "you", "we", "they", "our", "your", "their",
                  "will", "must", "at", "by", "from", "about", "into", "per",
                  "is", "are", "be", "been", "who", "via", "plus", "more"}


def _is_ats_keyword(phrase: str) -> bool:
    """True when `phrase` has the shape of a searchable ATS keyword.

    Structural only — rejects clause fragments, not unfamiliar skills.
    """
    p = (phrase or "").strip()
    if not p:
        return False
    # Parenthetical / example fragments: "running experiments (e.g",
    # "refine campaign copy (emails", "domain (e.g".
    if "(" in p or ")" in p or re.search(r"\be\.?g\b|\bi\.?e\b", p, re.I):
        return False
    if _TRAILING_JUNK.search(p):
        return False
    toks = p.split()
    if not toks or len(toks) > 5:
        return False
    low = [t.lower().strip(".,;:") for t in toks]
    # Function-word contractions ("what's not", "it's working") — but keep
    # genuine possessive nouns like "bachelor's degree" / "master's".
    for t in low:
        if "'" in t and t.split("'")[0] in (_CLAUSE_MARKERS | _PRONOUNS):
            return False
    if low[0] in _WEAK_HEADS or low[0] in _VERB_HEADS:
        return False
    # Gerund/3rd-person forms of those same verbs ("ensuring deadlines",
    # "monitoring performance") are clause fragments too. Strip the inflection
    # and re-test the head.
    h = low[0]
    for suf, rep in (("ing", ""), ("ing", "e"), ("es", ""), ("s", ""), ("ed", ""), ("ed", "e")):
        if h.endswith(suf) and (h[: -len(suf)] + rep) in _VERB_HEADS:
            return False
    if any(t in _CLAUSE_MARKERS for t in low):
        return False
    if any(t in _PRONOUNS for t in low):
        return False
    # Possessive proper nouns are employer-specific and unmatchable
    # ("Stripe's product suite").
    if re.search(r"\b[A-Z][A-Za-z0-9]*'s\b", p):
        return False
    # "&"-joined JD section headings ("Collaboration & Leadership",
    # "Performance & Optimization") are labels, not searchable terms.
    if "&" in p:
        return False
    # A clause connector between >=3 tokens signals prose, not a term.
    # ("knack for precise", "timelines aligned with growth"). Interior
    # "of/at/in/on/to" is fine — "systems at scale", "time to market".
    if len(low) >= 3 and any(t in ("for", "with", "by", "from", "about")
                             for t in low[1:-1]):
        return False
    # Must end on a content word, not a dangling function word.
    if low[-1] in _STOP_EDGE_ALL:
        return False
    # Needs at least one token that is not itself generic filler.
    if not any(t not in _GENERIC_ALONE and t not in _STOP_EDGE_ALL for t in low):
        return False
    return True

# Importance signal weights (sum 1.0) — Step "Keyword importance".
_W = {
    "mandatory_wording": 0.25, "required_section": 0.20, "centrality": 0.15,
    "title_summary": 0.10, "repetition": 0.10, "specificity": 0.10,
    "recruiter_filter": 0.05, "outcome": 0.05,
}

_YEARS_RE = re.compile(r"\b(\d{1,2})\s*\+?\s*years?\b", re.I)
_MUST_CUES = ("required", "must have", "must-have", "minimum", "essential",
              "mandatory", "at least", "you must", "strong")
_RESP_SECTIONS = ("responsibilit", "requirement", "qualification", "what you")
_OUTCOME_HINTS = ("revenue", "conversion", "growth", "retention", "roi",
                  "impact", "kpi", "metric", "margin", "adoption", "engagement")


# Locally-stored professional synonym map: JD concept -> résumé-side surface forms
# a candidate might genuinely use for the SAME capability. Powers curated-synonym
# evidence matching AND deterministic terminology alignment (swap weaker->exact).
_SYNONYMS = {
    "stakeholder management": ["stakeholder communication", "stakeholder comms",
                               "stakeholder engagement", "managing stakeholders"],
    "roadmap prioritization": ["roadmap planning", "product roadmap planning",
                               "prioritization", "roadmapping"],
    "product experimentation": ["experiments", "experimentation", "a/b testing",
                                "ab testing", "a/b experimentation"],
    "cross-functional collaboration": ["cross-functional teams", "cross functional",
                                       "cross-functional", "worked with teams"],
    "product analytics": ["analytics", "product data", "data analysis"],
    "go-to-market": ["gtm", "launch", "go to market"],
    "product strategy": ["product vision", "strategy and vision"],
    "conversion rate optimization": ["cro", "conversion optimization", "funnel optimization"],
    "demand generation": ["demand gen", "lead generation", "lead gen"],
    "data visualization": ["dashboards", "dashboarding", "visualization"],
    "statistical modeling": ["statistics", "statistical analysis", "modeling"],
    "distributed systems": ["distributed backends", "scalable systems"],
    "inventory management": ["stock management", "inventory control"],
    "process optimization": ["process improvement", "operational efficiency"],
    "risk management": ["risk mitigation", "risk assessment"],
    "campaign management": ["campaigns", "campaign execution"],
}
# reverse index: any surface form -> canonical concept (for fast enrichment)
_SYN_REVERSE = {}
for _canon, _forms in _SYNONYMS.items():
    for _f in _forms:
        _SYN_REVERSE.setdefault(_f.lower(), set()).add(_canon)


def _synonyms_for(concept: str) -> list:
    c = (concept or "").lower().strip()
    out = list(_SYNONYMS.get(c, []))
    # also include forms whose canonical shares this concept's head tokens
    for canon, forms in _SYNONYMS.items():
        if canon != c and (c in canon or canon in c):
            out += forms
    return list(dict.fromkeys(out))[:6]


def _band(score: float) -> str:
    return ("critical" if score >= 0.7 else "high" if score >= 0.5
            else "medium" if score >= 0.3 else "low")


def _sentences(text: str) -> List[str]:
    parts = re.split(r"(?<=[.;:!?])\s+|\n+", text or "")
    return [p.strip() for p in parts if p and p.strip()]


def _section_of(sentence: str, sections: Dict[str, str]) -> str:
    for name, body in (sections or {}).items():
        if sentence and sentence[:40] in body:
            return name
    return "body"


def _is_required(source_sentence: str, section: str) -> bool:
    low = (source_sentence or "").lower()
    if any(k in section for k in ("requirement", "minimum", "must", "qualification")):
        return True
    return any(c in low for c in _MUST_CUES)


def _importance(term: str, jd_low: str, source_sentence: str, section: str,
                category: str, is_required: bool) -> float:
    s = source_sentence.lower()
    score = 0.0
    if is_required or any(c in s for c in _MUST_CUES):
        score += _W["mandatory_wording"]
    if any(k in section for k in _RESP_SECTIONS):
        score += _W["required_section"]
    # centrality: multi-word skill/responsibility/tool phrase
    if category in ("responsibility", "hard_skill", "core_skill", "tool") and " " in term:
        score += _W["centrality"]
    # title/summary presence
    if term in jd_low[:400]:
        score += _W["title_summary"]
    # repetition across the JD
    if len(re.findall(r"(?<![a-z0-9])" + re.escape(term) + r"(?![a-z0-9])", jd_low)) >= 2:
        score += _W["repetition"]
    # role specificity: not a generic single token
    if " " in term or category in ("tool", "domain", "certification"):
        score += _W["specificity"]
    # recruiter-filter usefulness: tools/domains/certs/hard skills
    if category in ("tool", "domain", "certification", "hard_skill", "core_skill"):
        score += _W["recruiter_filter"]
    if any(h in s for h in _OUTCOME_HINTS):
        score += _W["outcome"]
    return round(min(score, 1.0), 3)


# Cues that introduce a list of skills/responsibilities in a requirement sentence.
_LIST_CUES = re.compile(
    r"\b(?:own|owns|owning|drive|drives|driving|build|builds|building|run|runs|"
    r"running|manage|manages|managing|lead|leads|leading|define|defines|defining|"
    r"develop|develops|developing|deliver|delivers|conduct|design|designs|"
    r"experience (?:with|in)|proficiency (?:with|in)|expertise (?:with|in)|"
    r"skills? (?:with|in)|knowledge of|strong|hands-on|familiarity with)\b",
    re.I)
_LEAD_TRIM = re.compile(
    r"^(?:the|a|an|and|our|your|their|strong|deep|solid|excellent|proven|"
    r"cross[- ]|end[- ]to[- ]end|both|other|new|complex|scalable)\s+", re.I)
_STOP_EDGE = {"and", "or", "the", "a", "an", "to", "of", "for", "with", "in",
              "on", "across", "using", "including", "etc", "such", "as", "e.g",
              "i.e", "you", "we", "they", "our", "your", "their", "will", "must"}


def _list_phrases(sent: str) -> List[str]:
    """Extract multi-word professional phrases from a requirement/responsibility
    sentence: the comma/'and'-separated items following a list cue (e.g. 'own
    product strategy, roadmap prioritization, and stakeholder management')."""
    out: List[str] = []
    m = _LIST_CUES.search(sent)
    tail = sent[m.end():] if m else sent
    # split into candidate items
    for raw in re.split(r"\s*(?:,|;|\band\b|\bor\b|\bacross\b|\bvia\b|\bthrough\b|/|\||•)\s*", tail):
        item = _LEAD_TRIM.sub("", raw.strip().strip(".:—-()").strip()).strip()
        item = re.sub(r"\s+", " ", item)
        toks = item.split()
        # trim generic edge words
        while toks and toks[0].lower() in _STOP_EDGE:
            toks = toks[1:]
        while toks and toks[-1].lower() in _STOP_EDGE:
            toks = toks[:-1]
        if not (1 <= len(toks) <= 4):
            continue
        phrase = " ".join(toks)
        pl = phrase.lower()
        if pl in _GENERIC_ALONE or pl in _BUZZWORDS:
            continue
        # must contain at least one content token not in the generic/stop sets and
        # be multi-word OR a known-ish single technical token (kept multi-word only)
        content = [t for t in toks if t.lower() not in _STOP_EDGE
                   and t.lower() not in _GENERIC_ALONE]
        if not _is_ats_keyword(phrase):
            continue
        if len(toks) >= 2 and content:
            out.append(phrase)
        elif len(toks) == 1:
            # accept a single-word item only if it's a known tool or an acronym
            # (SEO, SQL, AWS, Docker) — not a generic noun.
            t = toks[0]
            if t.lower() in _TOOLS or re.fullmatch(r"[A-Z][A-Za-z0-9]{1,5}", t) \
                    and t.lower() not in _GENERIC_ALONE:
                out.append(t)
    return out


def extract_criteria(clean_jd: str, sections: Dict[str, str] | None = None) -> List[dict]:
    """Return the full deterministic criterion inventory for a CLEANED JD.

    Each item: exact_phrase, normalized_concept, source_sentence, source_section,
    source_start, source_end, requirement_type, category, importance_score.
    """
    jd = clean_jd or ""
    jd_low = jd.lower()
    req = analyze_jd(jd, llm=None)          # deterministic gazetteer engine, NO LLM
    sents = _sentences(jd)

    def _src(term: str) -> tuple:
        m = re.search(r"(?<![a-z0-9])" + re.escape(term.lower()) + r"(?![a-z0-9])", jd_low)
        if not m:
            return None
        start = m.start()
        # containing sentence
        sent = next((s for s in sents if term.lower() in s.lower()), jd[max(0, start-40):start+80])
        return start, start + len(term), sent

    out: List[dict] = []
    seen = set()
    for r in req.all_requirements():
        term = (r.term or "").strip()
        tl = term.lower()
        if not term or tl in seen:
            continue
        if tl in _GENERIC_ALONE or tl in _BUZZWORDS:
            continue
        if not _is_ats_keyword(term):
            continue
        sr0 = _src(term)
        if sr0 is None:               # traceability: must be verbatim in the JD
            continue
        start, end, sent = sr0
        section = _section_of(sent, sections or {})
        is_req = (r.importance == "must_have") or _is_required(sent, section)
        if r.category == "seniority":     # title modifiers, not ATS keywords
            continue
        cat = r.category if r.category in (
            "tool", "domain", "responsibility", "soft_skill", "hard_skill",
            "core_skill", "certification", "education") else "hard_skill"
        iscore = _importance(tl, jd_low, sent, section, cat, is_req)
        out.append({
            "exact_phrase": term,
            "normalized_concept": tl,
            "source_sentence": sent[:200],
            "source_text": sent[:200],           # scorer/validator compatibility
            "source_section": section,
            "source_start": start,
            "source_end": end,
            "requirement_type": "required" if is_req else "preferred",
            "category": cat,
            "importance_score": iscore,
            "importance": _band(iscore),          # critical|high|medium|low
            "confidence": iscore,
            "requires_resume_evidence": True,
            "semantic_variants": (list(getattr(r, "aliases", []) or [])
                                  + _synonyms_for(tl))[:6],
        })
        seen.add(tl)

    # multi-word professional phrases from requirement/responsibility sentences
    # (comma-lists after a cue) — captures domain phrases the gazetteers miss
    # (e.g. distributed systems, demand generation, inventory management).
    for sent in sents:
        section = _section_of(sent, sections or {})
        if not (any(k in section for k in _RESP_SECTIONS) or _LIST_CUES.search(sent)):
            continue
        is_req = _is_required(sent, section)
        for phrase in _list_phrases(sent):
            pl = phrase.lower()
            if pl in seen:
                continue
            sr = _src(phrase)
            if sr is None:
                continue
            start, end, s2 = sr
            cat = "responsibility" if " " in phrase and any(
                v in sent.lower() for v in ("own", "drive", "lead", "manage", "run")) else "hard_skill"
            iscore = _importance(pl, jd_low, sent, section, cat, is_req)
            out.append({
                "exact_phrase": phrase, "normalized_concept": pl,
                "source_sentence": sent[:200], "source_text": sent[:200],
                "source_section": section, "source_start": start, "source_end": end,
                "requirement_type": "required" if is_req else "preferred",
                "category": cat, "importance_score": iscore, "importance": _band(iscore),
                "confidence": iscore, "requires_resume_evidence": True,
                "semantic_variants": _synonyms_for(pl),
            })
            seen.add(pl)

    # explicit years-of-experience requirement (a distinct criterion)
    ym = _YEARS_RE.search(jd)
    if ym and not any("year" in o["normalized_concept"] for o in out):
        start = ym.start()
        sent = next((s for s in sents if ym.group(0).lower() in s.lower()), ym.group(0))
        out.append({
            "exact_phrase": ym.group(0), "normalized_concept": ym.group(0).lower(),
            "source_sentence": sent[:200], "source_text": sent[:200],
            "source_section": _section_of(sent, sections or {}),
            "source_start": start, "source_end": ym.end(),
            "requirement_type": "required", "category": "experience_signal",
            "importance_score": 0.6, "importance": "high", "confidence": 0.6,
            "requires_resume_evidence": True, "semantic_variants": [],
        })

    # ponytail: dedup only TRUE duplicates (same concept), not sub-phrases.
    # ATS scanners search for each keyword independently — "solutions" and
    # "enterprise solutions" are different search terms, both worth placing.
    seen_concepts: set = set()
    kept: List[dict] = []
    for o in sorted(out, key=lambda x: len(x["exact_phrase"]), reverse=True):
        c = o["normalized_concept"]
        if c in seen_concepts:
            continue
        seen_concepts.add(c)
        kept.append(o)
    kept.sort(key=lambda o: o["importance_score"], reverse=True)
    return kept


def extract_from_raw(raw_jd: str, company: str = "") -> tuple:
    """Preprocess (untrusted) + deterministic extraction. Returns
    (PreprocessResult, criteria_list). criteria is [] when JD isolation fails."""
    pre = preprocess_jd(raw_jd, company=company)
    if not pre.ok:
        return pre, []
    return pre, extract_criteria(pre.clean_text, pre.sections)


if __name__ == "__main__":  # ponytail: runnable self-check
    JD = """About the Role
We seek a Product Manager to own the roadmap and drive product-led growth.

Responsibilities
- You will own product strategy, roadmap prioritization, and stakeholder management.
- Run A/B testing and product analytics to improve activation and conversion.

Minimum requirements
- 5+ years of product management experience.
- Strong SQL and experience with payments and fintech.
"""
    crit = extract_criteria(JD, None)
    concepts = {c["normalized_concept"] for c in crit}
    assert "product management" in concepts or "product manager" in concepts
    assert any("sql" == c["normalized_concept"] for c in crit)
    assert any(c["requirement_type"] == "required" for c in crit)
    # generic bare tokens excluded
    assert "work" not in concepts and "team" not in concepts
    # every phrase is traceable + has the full schema
    for c in crit:
        assert JD.lower().find(c["normalized_concept"]) >= 0
        assert set(c) >= {"exact_phrase", "requirement_type", "category",
                          "importance_score", "source_start"}
    print(f"deterministic_extract self-check PASSED — {len(crit)} criteria")
    for c in crit[:8]:
        print(f"  [{c['requirement_type']:<9} {c['importance_score']:.2f}] "
              f"{c['exact_phrase']} ({c['category']})")