"""Resume parsing service — Module 8.

Four-layer NER fallback chain merged on top of a lexical floor:

    Nucha BERT -> jobbert -> SkillNER -> SBERT+pgvector -> lexical (always)

All layers that succeed contribute to the merged output — "merge semantics"
rather than "first-success". This makes the fallback chain visible in the
demo and maximises recall for downstream gap analysis: different layers
catch different phrasings ("ML" vs "machine learning" vs "scikit-learn").

Each layer runs synchronously inside try/except; a layer that raises is logged
at ERROR and skipped while the rest of the chain continues. The lexical floor
always fires so the endpoint never returns empty solely because HF models are
unavailable.

Design constraints:

  * The endpoint contract of ``extract_skills_from_pdf`` is stable: a list
    of ``SkillPrediction`` dicts. Higher layers enrich confidence for skills
    the lexical layer already found, and add new skills via alias resolution.
  * Matching is done against the existing Skill catalog only. We never create
    new skills from free text — prevents catalog pollution and keeps the
    recommendation engine grounded in the curated taxonomy.
  * Proficiency is a heuristic, never an assertion. The user reviews and
    edits on the Profile page before anything is saved.

Layer selection via env var ``GAPGUIDE_PARSE_LAYERS`` (default is the full
chain; CI sets this to ``"lexical"`` via conftest.py so tests don't download
any models). See ~/.claude/plans/okay-plan-for-7a-fluffy-marshmallow.md.
"""
from __future__ import annotations

import io
import logging
import os
import re
from dataclasses import asdict, dataclass

import pdfplumber
from rapidfuzz import fuzz, process

from apps.skills.models import Skill


logger = logging.getLogger(__name__)

# Layer order — the dispatcher runs in this sequence. Merging is order-agnostic
# but the order here drives the "first to register a given skill wins the
# proficiency heuristic" rule. Lexical last (it already has the best local
# context), but we guarantee it runs last anyway in _predict_skills_from_text.
_DEFAULT_LAYERS = "nucha,jobbert,skillner,sbert,lexical"


MAX_PDF_BYTES = 5 * 1024 * 1024  # 5 MB — UCP students' CVs fit easily.
MAX_TEXT_CHARS = 60_000          # defends the downstream matcher.

# Rough heuristic — the parser is an *augmentation* layer, so we cap signals
# below 70 so the user always has room to raise after review.
_DEFAULT_PROFICIENCY = 55
_STRONG_SIGNAL_PROFICIENCY = 70

# Fuzzy layer is per-token (not partial_ratio against the full doc), so the
# threshold is genuinely "this word is essentially a typo of the skill name".
# Anything lower would flood the output with false positives like
# partial_ratio("python", "sourdough") > 0.5.
_FUZZY_TOKEN_SCORE = 88
_FUZZY_MIN_SKILL_LEN = 5   # 'Go', 'R', 'SQL' never go through fuzzy path

# Terms that indicate hands-on experience → pushes proficiency toward the top.
_EXPERIENCE_MARKERS = re.compile(
    r"\b(?:expert|advanced|proficien[tc]y?|senior|lead|architect|years?|yrs?)\b",
    re.IGNORECASE,
)

# Terms that hint at entry level → pulls proficiency down.
_NOVICE_MARKERS = re.compile(
    r"\b(?:beginner|junior|intro(?:duction)?|basic|elementary|familiar with)\b",
    re.IGNORECASE,
)

# Rank by *signal strength*, NOT numeric proficiency. NOVICE=40 is numerically
# less than DEFAULT=55 but carries a stronger claim ("beginner Python" beats
# silence), so it should win during window aggregation.
_NOVICE_PROFICIENCY = 40
_PROFICIENCY_RANK = {
    _STRONG_SIGNAL_PROFICIENCY: 2,
    _NOVICE_PROFICIENCY: 1,
    _DEFAULT_PROFICIENCY: 0,
}

# Per-layer calibration coefficient applied before the max() merge in
# _predict_skills_from_text. Each layer emits [0,1] from a different
# distribution — softmax (Nucha/JobBERT), cosine similarity above 0.65 threshold
# (SBERT), hardcoded 0.95 / rapidfuzz ratio (lexical). Raw max lets lexical's
# hardcoded 0.95 outweigh a genuine model hit of 0.92. Values are heuristic —
# a Platt-scaling rebuild from labelled data is out of FYP scope, but these at
# least stop the hardcoded layers from dominating live model output.
_LAYER_CALIBRATION = {
    'nucha': 1.0,
    'jobbert': 1.0,
    'skillner': 0.90,
    'sbert': 0.85,
    'lexical': 0.95,
}


class ResumeParseError(ValueError):
    """Raised for malformed or oversized inputs."""


@dataclass
class SkillPrediction:
    skill_id: int
    skill_name: str
    proficiency: int
    confidence: float
    matched_span: str

    def to_dict(self) -> dict:
        return asdict(self)


def _extract_text(pdf_bytes: bytes) -> str:
    if len(pdf_bytes) > MAX_PDF_BYTES:
        raise ResumeParseError(
            f"Resume exceeds the {MAX_PDF_BYTES // 1024 // 1024} MB upload limit."
        )
    if not pdf_bytes:
        raise ResumeParseError("Empty upload.")

    try:
        with pdfplumber.open(io.BytesIO(pdf_bytes)) as pdf:
            pages = []
            for page in pdf.pages:
                text = page.extract_text() or ""
                pages.append(text)
                if sum(len(p) for p in pages) > MAX_TEXT_CHARS:
                    break
            full_text = "\n".join(pages)
    except Exception as exc:
        # pdfplumber raises a grab-bag of exceptions for corrupt PDFs; treat
        # them all as user input errors rather than 500s.
        raise ResumeParseError(f"Could not read PDF: {exc}") from exc

    if not full_text.strip():
        raise ResumeParseError(
            "No text could be extracted. If your CV is a scanned image, "
            "re-export it as a text-based PDF."
        )
    return full_text[:MAX_TEXT_CHARS]


def _classify_proficiency(context: str) -> tuple[int, float]:
    """Return (proficiency, confidence_boost) from surrounding context.

    A short window around the matched span is all we get — this is purposely
    coarse. Higher layers can refine with a real classifier later.
    """
    if _EXPERIENCE_MARKERS.search(context):
        return _STRONG_SIGNAL_PROFICIENCY, 0.10
    if _NOVICE_MARKERS.search(context):
        return _NOVICE_PROFICIENCY, 0.05
    return _DEFAULT_PROFICIENCY, 0.0


def _classify_aggregated(contexts: list[str]) -> tuple[int, float]:
    """Pick the strongest signal across all window contexts for a single skill.

    Short-circuits on STRONG (the ceiling). Ranks by _PROFICIENCY_RANK, not by
    raw proficiency value, so NOVICE ("beginner X") beats DEFAULT (silence)
    even though NOVICE=40 < DEFAULT=55.
    """
    best = (_DEFAULT_PROFICIENCY, 0.0)
    best_rank = 0
    for ctx in contexts:
        prof, boost = _classify_proficiency(ctx)
        rank = _PROFICIENCY_RANK.get(prof, 0)
        if rank > best_rank:
            best, best_rank = (prof, boost), rank
            if rank == 2:  # STRONG — can't improve further.
                break
    return best


def _windows_for(text: str, pattern: str, context_chars: int = 60) -> list[str]:
    """Yield ±context_chars windows around every match of `pattern` in text."""
    windows = []
    for match in re.finditer(pattern, text.lower()):
        start = max(0, match.start() - context_chars)
        end = min(len(text), match.end() + context_chars)
        windows.append(text[start:end])
    return windows


_TOKEN_RE = re.compile(r"[A-Za-z][A-Za-z0-9+#.\-]{2,}")


def _candidate_matches(text: str, skills: list[Skill]) -> dict[int, dict]:
    """Return {skill_id: {'confidence', 'proficiency', 'matched_span'}} for
    every catalog skill that appears in the text.

    Two passes:
      1. Exact (case-insensitive) boundary-aware substring hit → high
         confidence (0.95). We do NOT include '.' in the boundary character
         class because tech names like "Node.js" already round-trip through
         ``re.escape`` and sentence-final punctuation like "... SQL." must
         still match.
      2. Token-level fuzzy match — for each token in the document that's
         long enough, rapidfuzz.fuzz.ratio against the catalog. ``partial_
         ratio`` was rejected because it false-positives aggressively
         ("python" vs "sourdough" scored ~0.5). Per-token ``ratio`` with a
         high threshold (88) only fires on near-typos.
    """
    hits: dict[int, dict] = {}

    for skill in skills:
        name = skill.skill_name.strip()
        if not name:
            continue
        needle = name.lower()
        # Word-boundary exact match — '\b' handles punctuation around
        # identifier-like tech names. '.' is intentionally excluded from the
        # boundary class so "SQL." at end-of-sentence still matches.
        pattern = r"(?<![A-Za-z0-9+#\-])" + re.escape(needle) + r"(?![A-Za-z0-9+#\-])"
        windows = _windows_for(text, pattern)
        if windows:
            prof, boost = _classify_aggregated(windows)
            # First window is kept as the human-readable matched_span — the
            # strongest-signal window may be far away, and the first occurrence
            # is what the reviewer will visually scan for.
            hits[skill.id] = {
                'confidence': round(min(0.95 + boost, 0.99), 3),
                'proficiency': prof,
                'matched_span': windows[0].strip(),
            }

    # Fuzzy pass: build the document's token set once, then for each unmatched
    # skill find its best-matching token (rapidfuzz.process.extractOne).
    tokens = list({m.group(0) for m in _TOKEN_RE.finditer(text)})
    if not tokens:
        return hits

    for skill in skills:
        if skill.id in hits:
            continue
        name = skill.skill_name.strip()
        if len(name) < _FUZZY_MIN_SKILL_LEN:
            # Short names ('Go', 'R', 'SQL') only travel the exact path.
            continue
        candidate = process.extractOne(
            name.lower(),
            [t.lower() for t in tokens],
            scorer=fuzz.ratio,
            score_cutoff=_FUZZY_TOKEN_SCORE,
        )
        if candidate is None:
            continue
        _, score, idx = candidate
        hits[skill.id] = {
            'confidence': round(score / 100.0, 3),
            'proficiency': _DEFAULT_PROFICIENCY,
            'matched_span': tokens[idx],
        }

    return hits


def _load_layer(layer_name: str):
    """Lazy-import a layer module by name. Returns None if import fails.

    Done this way (rather than an eager registry) so a CI run pinned to
    ``GAPGUIDE_PARSE_LAYERS=lexical`` never imports transformers/torch.
    """
    try:
        # `apps.accounts.ner.<name>` — lives next to this module.
        mod = __import__(
            f"apps.accounts.ner.{layer_name}",
            fromlist=["layer"],
        )
        return getattr(mod, "layer", None)
    except Exception as exc:
        logger.warning("ner.%s: import failed, skipping (%s)", layer_name, exc)
        return None


def _enabled_layers() -> list[str]:
    raw = os.environ.get("GAPGUIDE_PARSE_LAYERS", _DEFAULT_LAYERS)
    out = [s.strip() for s in raw.split(",") if s.strip()]
    # Always guarantee lexical at the end — contract says the floor fires.
    if "lexical" not in out:
        out.append("lexical")
    return out


def _find_span_for(text: str, skill_name: str) -> tuple[int, str]:
    """Search the text for `skill_name` and derive (proficiency, matched_span).

    Used to enrich non-lexical layer hits with the same surrounding-context
    proficiency heuristic the lexical layer uses. If the skill name isn't
    literally in the text (the layer inferred it from a paraphrase), fall
    back to the default proficiency + empty span.

    Multi-window aggregation: if the skill appears multiple times, the
    strongest *signal* across windows wins (STRONG > NOVICE > DEFAULT).
    """
    needle = re.escape(skill_name.lower())
    pattern = r"(?<![A-Za-z0-9+#\-])" + needle + r"(?![A-Za-z0-9+#\-])"
    windows = _windows_for(text, pattern)
    if not windows:
        return _DEFAULT_PROFICIENCY, ''
    prof, _ = _classify_aggregated(windows)
    return prof, windows[0].strip()


def _resolve_alias(alias: str, skills_by_lower: dict, all_skills: list[Skill]) -> Skill | None:
    """Map a free-text alias to a canonical Skill row.

    Exact match (case-insensitive) -> fuzzy match via rapidfuzz using the
    same threshold as the lexical layer. Returns None if nothing confident
    enough is found. Aliases too short to fuzzy-match just require an exact
    hit; keeps 'Go' and 'R' from matching arbitrary noun phrases.
    """
    if not alias:
        return None
    key = alias.strip().lower()
    hit = skills_by_lower.get(key)
    if hit is not None:
        return hit
    if len(key) < _FUZZY_MIN_SKILL_LEN:
        return None
    candidate = process.extractOne(
        key,
        list(skills_by_lower.keys()),
        scorer=fuzz.ratio,
        score_cutoff=_FUZZY_TOKEN_SCORE,
    )
    if candidate is None:
        return None
    matched_name, _, _ = candidate
    return skills_by_lower.get(matched_name)


def _predict_skills_from_text(text: str) -> tuple[dict[int, dict], list[str]]:
    """Run enabled NER layers against `text` and merge their hits.

    Returns (hits_by_skill_id, fired_layers):
      - hits_by_skill_id: {skill_id: {confidence, proficiency, matched_span}}
      - fired_layers: layers that actually returned results (in order)

    Each layer runs synchronously. A layer that raises during predict() is
    logged at ERROR (with traceback) and skipped — the rest of the chain
    continues. (Import failures are caught earlier in _load_layer and logged
    at WARNING.) If ALL layers fail, the endpoint still returns 200 with the
    lexical-floor output (since lexical is always run).
    """
    skills = list(Skill.objects.all().only('id', 'skill_name'))
    skills_by_lower = {s.skill_name.lower(): s for s in skills}

    hits: dict[int, dict] = {}
    fired: list[str] = []

    for layer_name in _enabled_layers():
        layer = _load_layer(layer_name)
        if layer is None or not layer.available():
            continue
        try:
            # Run synchronously. A previous version ran each layer in a
            # ThreadPoolExecutor for timeout support, but Django's ORM uses
            # thread-local connections — the new thread never saw the test
            # transaction's uncommitted rows, so the lexical/sbert layers
            # found nothing under pytest-django. HF client libraries already
            # apply their own request timeouts, so dropping the local one
            # costs us only a theoretical hang on a misbehaving custom layer.
            layer_out = layer.predict(text)
        except Exception as exc:
            # Keep the catch broad (a flaky HF layer must never 500 the upload)
            # but log at ERROR with the traceback — the always-on net that makes
            # a silently-skipped layer (e.g. a wiring/construction bug) visible
            # in the logs instead of vanishing.
            logger.error(
                "ner.%s: raised %s, skipping", layer_name, exc, exc_info=True,
            )
            continue

        if not layer_out:
            continue
        fired.append(layer_name)

        calibration = _LAYER_CALIBRATION.get(layer_name, 1.0)
        for alias, conf in layer_out.items():
            skill = _resolve_alias(alias, skills_by_lower, skills)
            if skill is None:
                continue
            existing = hits.get(skill.id)
            # Calibrate per-layer so a 0.92 softmax isn't dominated by a
            # hardcoded lexical 0.95. Cap below 1.0 to keep downstream plots
            # away from the edge of the [0,1] axis.
            conf_f = round(min(float(conf) * calibration, 0.99), 3)
            if existing is None:
                prof, span = _find_span_for(text, skill.skill_name)
                hits[skill.id] = {
                    'confidence': conf_f,
                    'proficiency': prof,
                    'matched_span': span,
                }
            elif conf_f > existing['confidence']:
                existing['confidence'] = conf_f

    return hits, fired


def extract_skills_from_pdf(pdf_bytes: bytes) -> list[dict]:
    """Parse a resume PDF and return ordered, catalog-resolved skill predictions.

    Kept as a list-only return to preserve the MVP test contract. The view
    uses :func:`parse_resume_envelope` instead, which also carries the list
    of layers that fired (the dispatcher's `parser_version`).
    """
    env = parse_resume_envelope(pdf_bytes)
    return env['skills']


def parse_resume_envelope(pdf_bytes: bytes) -> dict:
    """Full response envelope: ``{'skills': [...], 'parser_version': [...]}``.

    `parser_version` is a list of layer names so the frontend can render a
    real "which layers fired" chip sequence (`["nucha", "lexical"]`) rather
    than the MVP's opaque `"mvp-lexical-v1"` string.
    """
    text = _extract_text(pdf_bytes)
    scored, fired = _predict_skills_from_text(text)
    if not scored:
        return {'skills': [], 'parser_version': fired}

    skill_lookup = {
        s.id: s for s in Skill.objects.filter(id__in=scored.keys())
    }
    predictions: list[SkillPrediction] = []
    for skill_id, data in scored.items():
        skill = skill_lookup.get(skill_id)
        if skill is None:
            continue
        predictions.append(SkillPrediction(
            skill_id=skill_id,
            skill_name=skill.skill_name,
            proficiency=int(data['proficiency']),
            confidence=float(data['confidence']),
            matched_span=str(data['matched_span'])[:200],
        ))

    predictions.sort(key=lambda p: (-p.confidence, p.skill_name))
    return {
        'skills': [p.to_dict() for p in predictions],
        'parser_version': fired,
    }