""" Module 2 — Learning Path Recommendation API Ports the pipeline built in notebooks 01-04 (skill normalization, GLiNER extraction, gap analysis, course recommendation) into a single FastAPI service, deployable the same way as Module 1 / Module 3 (Docker on HuggingFace Spaces). Run locally: pip install -r requirements.txt uvicorn app:app --reload --port 8000 Required data files (already generated by notebooks 01-04), expected under DATA_DIR (default ./data/processed): skill_vocab.pkl, alias_lookup.pkl, vocab_norms.pkl, vocab_canonical.pkl, skill_graph.pkl, skill_emb_lookup.pkl, courses.pkl, course_embeddings.npy, course_id_to_idx.pkl, course_lookup.pkl, extractor_config.json """ import json import os import re import datetime from pathlib import Path from typing import Optional import numpy as np import pandas as pd import networkx as nx from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from sklearn.metrics.pairwise import cosine_similarity DATA_DIR = Path(os.getenv("DATA_DIR", "data/processed")) # ── Module-level state, populated by load_models() on startup ───────────── skill_vocab = {} alias_lookup = {} VOCAB_NORMS = [] VOCAB_CANONICAL = [] skill_graph: nx.DiGraph = nx.DiGraph() skill_emb_lookup = {} courses: pd.DataFrame = pd.DataFrame() course_emb_matrix = None course_id_to_idx = {} course_lookup = {} gliner_model = None sbert_model = None SKILL_LABELS = [] MODEL_NAME = "" MODELS_LOADED = False # ── Skill normalization (Notebook 01/02) ─────────────────────────────────── def _norm(text: str) -> str: t = str(text).lower().strip() t = re.sub(r"[\-\.\s/]+", " ", t) t = re.sub(r"[^a-z0-9\s#+]", "", t) return t.strip() def normalize_skill(raw: str, threshold: float = 0.82) -> dict: import jellyfish raw_norm = _norm(raw) if not raw_norm or len(raw_norm) < 2: return {"canonical_name": raw.strip(), "difficulty": 3.0, "match_score": 0.0, "match_type": "too_short"} if raw_norm in alias_lookup: canonical = alias_lookup[raw_norm] entry = skill_vocab.get(_norm(canonical), {}) return {"canonical_name": canonical, "difficulty": entry.get("difficulty_score", 3.0), "match_score": 1.0, "match_type": "exact"} best_score, best_idx = 0.0, -1 for i, vn in enumerate(VOCAB_NORMS): s = jellyfish.jaro_winkler_similarity(raw_norm, vn) if s > best_score: best_score, best_idx = s, i if best_score >= threshold and best_idx >= 0: canonical = VOCAB_CANONICAL[best_idx] entry = skill_vocab[VOCAB_NORMS[best_idx]] return {"canonical_name": canonical, "difficulty": entry.get("difficulty_score", 3.0), "match_score": round(best_score, 3), "match_type": "fuzzy"} return {"canonical_name": raw.strip(), "difficulty": 3.0, "match_score": round(best_score, 3), "match_type": "unmatched"} # ── Text extraction (Notebook 02) ─────────────────────────────────────────── def extract_text_from_pdf(pdf_bytes: bytes) -> str: text = "" try: import fitz # PyMuPDF doc = fitz.open(stream=pdf_bytes, filetype="pdf") parts = [] for page in doc: blocks = page.get_text("blocks") blocks.sort(key=lambda b: (round(b[1] / 50), b[0])) for b in blocks: parts.append(b[4].strip()) text = "\n".join(p for p in parts if p) doc.close() if len(text.split()) > 30: return text except Exception: pass try: import io from pdfminer.high_level import extract_text as pm_extract text = pm_extract(io.BytesIO(pdf_bytes)) if text.strip(): return text except Exception: pass return text def extract_text_from_docx(docx_bytes: bytes) -> str: import io from docx import Document doc = Document(io.BytesIO(docx_bytes)) parts = [para.text.strip() for para in doc.paragraphs if para.text.strip()] for table in doc.tables: for row in table.rows: for cell in row.cells: if cell.text.strip(): parts.append(cell.text.strip()) return "\n".join(parts) def extract_text_from_file(file_bytes: bytes, filename: str) -> str: ext = Path(filename).suffix.lower() if ext == ".pdf": return extract_text_from_pdf(file_bytes) elif ext in (".docx", ".doc"): return extract_text_from_docx(file_bytes) elif ext in (".txt", ".rtf", ".md"): return file_bytes.decode("utf-8", errors="replace") else: try: return extract_text_from_pdf(file_bytes) except Exception: return file_bytes.decode("utf-8", errors="replace") def clean_extracted_text(text: str) -> str: text = re.sub(r"[^\x00-\x7F]+", " ", text) text = re.sub(r"\b[\w.%+-]+@[\w.-]+\.[a-zA-Z]{2,}\b", "[EMAIL]", text) text = re.sub(r"\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b", "[PHONE]", text) text = re.sub(r"https?://\S+|www\.\S+", "[URL]", text) text = re.sub(r"[•·▪►◦‣⁃]", "-", text) text = re.sub(r"[ \t]{2,}", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() def chunk_text(text: str, max_tokens: int = 400, overlap: int = 50) -> list: words = text.split() if len(words) <= max_tokens: return [text] chunks = [] start = 0 while start < len(words): end = min(start + max_tokens, len(words)) chunks.append(" ".join(words[start:end])) start += max_tokens - overlap return chunks def extract_skills_from_text(text: str, confidence_threshold: float = 0.4, dedup: bool = True) -> list: text = clean_extracted_text(text) if not text.strip(): return [] chunks = chunk_text(text) all_entities = [] for chunk in chunks: try: entities = gliner_model.predict_entities(chunk, SKILL_LABELS, threshold=confidence_threshold) all_entities.extend(entities) except Exception: continue seen_canonical = {} results = [] for ent in all_entities: raw = ent["text"].strip() if len(raw) < 2: continue if re.match(r"^(the|and|or|in|of|to|for|with|a|an|is|are|was|were)$", raw.lower()): continue norm_result = normalize_skill(raw) canonical = norm_result["canonical_name"] record = { "raw_text": raw, "canonical_name": canonical, "label": ent["label"], "gliner_score": round(ent["score"], 3), "match_type": norm_result["match_type"], "match_score": norm_result.get("match_score", 1.0), "difficulty": norm_result["difficulty"], } if dedup: if canonical not in seen_canonical or ent["score"] > seen_canonical[canonical]["gliner_score"]: seen_canonical[canonical] = record else: results.append(record) if dedup: results = list(seen_canonical.values()) results.sort(key=lambda x: -x["gliner_score"]) return results # ── Resume context / proficiency estimation (Notebook 02) ───────────────── YEARS_PATTERNS = [ r"(\d+)\+?\s*(?:years?|yrs?)\s+(?:of\s+)?(?:experience\s+(?:in|with|using)\s+)?([\w\s\.\+\#]{2,30})", r"([\w\s\.\+\#]{2,30})\s+(?:for\s+)?(\d+)\+?\s*(?:years?|yrs?)", ] DATE_PATTERN = re.compile( r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec|january|february|march|" r"april|june|july|august|september|october|november|december)" r"[\s,]*(\d{4})", re.IGNORECASE, ) CERT_PATTERNS = [ r"(certified|certification|certificate)\s+(?:in\s+)?([\w\s\+\#\.]{3,40})", r"([\w\s\+\#\.]{3,40})\s+(?:certified|certification|certificate)", r"(AWS\s+certified|Google\s+certified|Microsoft\s+certified)[\w\s]+", ] def extract_context_features(text: str, skill_name: str) -> dict: skill_lower = skill_name.lower() text_lower = text.lower() result = {"years_experience": None, "last_used_months": 0, "is_certified": False, "context_snippet": ""} idx = text_lower.find(skill_lower) if idx < 0: return result window = text_lower[max(0, idx - 200): idx + 200] result["context_snippet"] = window.strip() for pattern in YEARS_PATTERNS: matches = re.findall(pattern, window, re.IGNORECASE) for m in matches: try: years = int(m[0]) if str(m[0]).isdigit() else int(m[1]) result["years_experience"] = years break except (ValueError, IndexError): continue if result["years_experience"]: break date_matches = DATE_PATTERN.findall(window) if date_matches: try: latest_year = max(int(y) for y in date_matches) now = datetime.datetime.now() months_ago = (now.year - latest_year) * 12 result["last_used_months"] = max(0, months_ago) except ValueError: pass for pattern in CERT_PATTERNS: if re.search(pattern, window, re.IGNORECASE): result["is_certified"] = True break return result def estimate_proficiency(skill_name: str, resume_text: str, total_exp_years: Optional[int] = None) -> dict: ctx = extract_context_features(resume_text, skill_name) years = ctx["years_experience"] if years is None: years = int(total_exp_years * 0.6) if total_exp_years else 2 years = min(years, 20) if years < 1: base = 1.0 elif years < 2: base = 2.0 elif years < 4: base = 3.0 elif years < 7: base = 4.0 else: base = 5.0 if ctx["is_certified"]: base = min(5.0, base + 1.0) cert_confidence = 0.9 else: cert_confidence = 0.75 if years >= 3 else 0.6 months_ago = ctx["last_used_months"] if months_ago > 0: decay = max(0.5, 1.0 - months_ago / 36.0) base = base * decay final_score = round(min(5.0, max(0.5, base)), 1) confidence = round(min(0.95, max(0.4, cert_confidence)), 2) return { "proficiency_score": final_score, "years_of_experience": years, "is_certified": ctx["is_certified"], "last_used_months": months_ago, "confidence_score": confidence, } def extract_resume_skills(resume_text: str, total_exp_years: Optional[int] = None, confidence_threshold: float = 0.4) -> list: raw_skills = extract_skills_from_text(resume_text, confidence_threshold=confidence_threshold) if total_exp_years is None: year_matches = re.findall(r"(\d+)\+?\s*(?:years?|yrs?)\s+(?:of\s+)?(?:total\s+)?experience", resume_text, re.IGNORECASE) if year_matches: total_exp_years = max(int(y) for y in year_matches) results = [] for skill in raw_skills: canonical = skill["canonical_name"] prof = estimate_proficiency(canonical, resume_text, total_exp_years) results.append({ "raw_text": skill["raw_text"], "canonical_name": canonical, "label": skill["label"], "gliner_score": skill["gliner_score"], "match_type": skill["match_type"], "proficiency_score": prof["proficiency_score"], "years_experience": prof["years_of_experience"], "is_certified": prof["is_certified"], "last_used_months": prof["last_used_months"], "confidence_score": prof["confidence_score"], "difficulty": skill["difficulty"], }) results.sort(key=lambda x: -x["proficiency_score"]) return results # ── JD extraction: criticality + required proficiency (Notebook 02) ─────── CRITICALITY_RULES = [ (1.0, ["required", "must have", "must-have", "essential", "mandatory", "expert-level", "you must"]), (0.8, ["strong", "strong experience", "solid experience", "hands-on"]), (0.7, ["preferred", "ideally", "we prefer", "experience with", "familiarity with", "proficient"]), (0.5, ["nice to have", "nice-to-have", "bonus", "a plus", "beneficial", "desirable"]), (0.3, ["optional", "good to have", "knowledge of"]), ] PROFICIENCY_RULES = [ (5.0, ["expert", "expert-level", "mastery", "deep expertise", "extensive"]), (4.0, ["advanced", "senior-level", "strong", "solid", "proven"]), (3.5, ["proficient", "proficiency", "good knowledge", "working knowledge"]), (3.0, ["intermediate", "mid-level", "competent"]), (2.0, ["familiar", "familiarity", "basic", "some experience", "exposure"]), (1.5, ["beginner", "entry-level", "introductory", "learning"]), ] def get_criticality(skill_name: str, jd_text: str) -> tuple: skill_lower = skill_name.lower() jd_lower = jd_text.lower() idx = jd_lower.find(skill_lower) if idx < 0: return ("inferred", 0.7) lines = jd_lower[:idx + len(skill_lower)].split("\n") context = "\n".join(lines[-3:]) for score, keywords in CRITICALITY_RULES: if any(kw in context for kw in keywords): label = {1.0: "required", 0.8: "strong", 0.7: "preferred", 0.5: "nice_to_have", 0.3: "optional"}.get(score, "preferred") return (label, score) position_ratio = idx / max(1, len(jd_text)) if position_ratio < 0.4: return ("required", 1.0) return ("preferred", 0.7) def get_required_proficiency(skill_name: str, jd_text: str, level_hint: str = "") -> float: jd_lower = jd_text.lower() skill_lower = skill_name.lower() idx = jd_lower.find(skill_lower) context = "" if idx >= 0: context = jd_lower[max(0, idx - 150): idx + 100] for score, keywords in PROFICIENCY_RULES: if any(kw in context for kw in keywords): return score level_defaults = { "junior": 2.0, "entry": 2.0, "mid": 3.0, "mid-level": 3.0, "senior": 4.0, "lead": 4.5, "principal": 5.0, "staff": 5.0, } level_lower = level_hint.lower() for key, default in level_defaults.items(): if key in level_lower: return default return 3.0 def extract_jd_skills(jd_text: str, level_hint: str = "", confidence_threshold: float = 0.35) -> list: raw_skills = extract_skills_from_text(jd_text, confidence_threshold=confidence_threshold) results = [] for skill in raw_skills: canonical = skill["canonical_name"] crit_label, crit_score = get_criticality(canonical, jd_text) req_prof = get_required_proficiency(canonical, jd_text, level_hint) results.append({ "raw_text": skill["raw_text"], "canonical_name": canonical, "label": skill["label"], "gliner_score": skill["gliner_score"], "match_type": skill["match_type"], "criticality_label": crit_label, "criticality_score": crit_score, "required_proficiency": req_prof, "difficulty": skill["difficulty"], "is_must_have": crit_score >= 0.8, }) results.sort(key=lambda x: (-x["criticality_score"], -x["gliner_score"])) return results # ── Semantic similarity + transferability (Notebook 03) ─────────────────── def semantic_similarity(skill_a: str, skill_b: str) -> float: emb_a = skill_emb_lookup.get(skill_a) emb_b = skill_emb_lookup.get(skill_b) if emb_a is None or emb_b is None: vecs = sbert_model.encode([skill_a if emb_a is None else "", skill_b if emb_b is None else ""]) emb_a = vecs[0] if emb_a is None else emb_a emb_b = vecs[1] if emb_b is None else emb_b sim = cosine_similarity([emb_a], [emb_b])[0][0] return float(sim) def max_semantic_similarity(target_skill: str, employee_skills: list) -> float: emp_embs = [] for s in employee_skills: skill_name = s["canonical_name"] if isinstance(s, dict) else s vec = skill_emb_lookup.get(skill_name) if vec is not None: emp_embs.append(vec) if len(emp_embs) == 0: return 0.0 target_emb = skill_emb_lookup.get(target_skill) if target_emb is None: target_emb = sbert_model.encode([target_skill])[0] similarities = [ float(np.dot(target_emb, e_emb) / (np.linalg.norm(target_emb) * np.linalg.norm(e_emb))) for e_emb in emp_embs ] return max(similarities) if len(similarities) > 0 else 0.0 def compute_transferability(target_skill: str, employee_skills: list) -> float: emp_skill_set = set(employee_skills) related_via_graph = set() if skill_graph.has_node(target_skill): related_via_graph.update(skill_graph.predecessors(target_skill)) related_via_graph.update(skill_graph.successors(target_skill)) related_via_semantic = set() for emp_skill in employee_skills: sim = semantic_similarity(target_skill, emp_skill) if sim >= 0.50: related_via_semantic.add(emp_skill) all_related = related_via_graph.union(related_via_semantic) overlapping = all_related.intersection(emp_skill_set) transferability = min(1.0, len(overlapping) / 5.0) return round(transferability, 3) def estimate_learning_hours(gap_size: float, difficulty: float, transferability: float) -> dict: gap_size = max(0.1, float(gap_size)) difficulty = max(0.5, float(difficulty)) transferability = max(0.0, min(1.0, float(transferability))) base_hours = gap_size * 20 difficulty_mult = difficulty / 3.0 transfer_discount = 1.0 - (transferability * 0.30) mean_hours = base_hours * difficulty_mult * transfer_discount diff_var = (difficulty / 5.0) * 0.30 ind_var = 0.20 std = mean_hours * (diff_var + ind_var) ci_lo = max(1.0, mean_hours - 1.96 * std) ci_hi = mean_hours + 1.96 * std return { "mean_hours": round(mean_hours, 1), "ci_low": round(ci_lo, 1), "ci_high": round(ci_hi, 1), "weeks_at_5h": round(mean_hours / 5, 1), } def compute_priority(criticality: float, gap_size: float, transferability: float) -> float: return round((criticality * gap_size) / (transferability + 0.1), 3) class GapAnalyzer: def analyze(self, employee_skills: list, jd_skills: list, max_gaps: int = 15) -> dict: emp_prof = {s["canonical_name"]: s["proficiency_score"] for s in employee_skills} emp_all = list(emp_prof.keys()) gaps = [] matched = [] for req in jd_skills: canonical = req["canonical_name"] req_prof = req.get("required_proficiency", 3.0) criticality = req.get("criticality_score", 0.7) difficulty = req.get("difficulty", 3.0) current_prof = emp_prof.get(canonical, 0.0) max_sem_sim = max_semantic_similarity(canonical, emp_all) sem_credit = max(0.0, (max_sem_sim - 0.3) / 0.7) * 0.40 if max_sem_sim > 0.3 else 0.0 sem_credit = min(0.40, sem_credit) raw_gap = req_prof - current_prof if current_prof > 0: effective_gap = max(0.0, raw_gap) else: effective_gap = max(0.0, raw_gap * (1.0 - sem_credit)) if effective_gap <= 0.05: matched.append({ "canonical_name": canonical, "current_proficiency": current_prof, "required_proficiency": req_prof, "criticality_score": criticality, "status": "met", }) continue transferability = compute_transferability(canonical, emp_all) learn_time = estimate_learning_hours(effective_gap, difficulty, transferability) priority = compute_priority(criticality, effective_gap, transferability) gaps.append({ "canonical_name": canonical, "criticality_label": req.get("criticality_label", "preferred"), "criticality_score": criticality, "current_proficiency": current_prof, "required_proficiency": req_prof, "raw_gap": round(raw_gap, 2), "effective_gap": round(effective_gap, 2), "semantic_similarity": round(max_sem_sim, 3), "semantic_credit": round(sem_credit, 3), "transferability": transferability, "difficulty": difficulty, "learning_hours_mean": learn_time["mean_hours"], "learning_hours_ci_low": learn_time["ci_low"], "learning_hours_ci_high": learn_time["ci_high"], "weeks_at_5h": learn_time["weeks_at_5h"], "priority_score": priority, "is_must_have": req.get("is_must_have", False), }) gaps.sort(key=lambda x: (-int(x["is_must_have"]), -x["priority_score"])) gaps = gaps[:max_gaps] all_reqs = jd_skills if all_reqs: total_crit = sum(r["criticality_score"] for r in all_reqs) met_crit = sum(m["criticality_score"] for m in matched) readiness = round((met_crit / total_crit) * 100, 1) if total_crit > 0 else 0.0 else: readiness = 100.0 total_learn_hrs = sum(g["learning_hours_mean"] for g in gaps) total_weeks = round(total_learn_hrs / 5, 1) return { "gaps": gaps, "matched_skills": matched, "n_gaps": len(gaps), "n_matched": len(matched), "job_readiness": readiness, "total_learn_hours": round(total_learn_hrs, 1), "total_weeks_at_5h": total_weeks, } # ── Course recommendation (Notebook 04) ──────────────────────────────────── def search_courses_for_skill(skill_name: str, top_k: int = 8, difficulty_filter: Optional[str] = None) -> list: query_emb = skill_emb_lookup.get(skill_name) if query_emb is None: query_emb = sbert_model.encode([skill_name])[0] sims = cosine_similarity([query_emb], course_emb_matrix)[0] if difficulty_filter: mask = (courses["difficulty_label"] == difficulty_filter).values sims = np.where(mask, sims, -1.0) top_indices = np.argsort(sims)[::-1][:top_k * 2] results = [] for idx in top_indices: if sims[idx] < 0.25: continue course_row = courses.iloc[idx] results.append({ "course_id": course_row["course_id"], "course_name": course_row["course_name"], "semantic_score": float(sims[idx]), "difficulty": course_row["difficulty_label"], "difficulty_score": course_row["difficulty_score"], "duration_hours": float(course_row["duration_hours"]), "price_usd": float(course_row["price_usd"]), "rating": float(course_row["rating"]), "target_skills": course_row["target_skills_list"], "prereq_skills": course_row["prereq_skills_list"], "prereq_course_ids": course_row["prereq_course_ids_list"], "prereq_course_names": course_row["prereq_course_names_list"], "is_free": bool(course_row["is_free"]), }) return results[:top_k] def compute_composite_score(course: dict, gap: dict, employee_skills: list) -> float: target_skills = course.get("target_skills", []) gap_skill = gap["canonical_name"] if gap_skill in target_skills: coverage = min(1.0, course["difficulty_score"] / max(1.0, gap["effective_gap"] * 2)) else: coverage = course["semantic_score"] * 0.5 criticality = gap["criticality_score"] value = criticality * coverage time_cost = course["duration_hours"] * 0.01 money_cost = course["price_usd"] * 0.001 diff_cost = course["difficulty_score"] * 0.1 rating_boost = (course["rating"] - 4.0) * 0.2 value += rating_boost total_cost = max(0.01, time_cost + money_cost + diff_cost) return round(value / total_cost, 4) class CourseRecommender: def recommend(self, gaps: list, employee_skills: list, max_hours: float = 200.0, max_budget: float = 500.0, top_k_per_gap: int = 6) -> dict: total_time = 0.0 total_cost = 0.0 covered_skills = set(s["canonical_name"] if isinstance(s, dict) else s for s in employee_skills) selected_courses = [] gap_to_course = {} all_candidates = [] for gap in gaps: skill = gap["canonical_name"] candidates = search_courses_for_skill(skill, top_k=top_k_per_gap) for c in candidates: c["for_gap"] = skill c["gap_criticality"] = gap["criticality_score"] c["effective_gap"] = gap["effective_gap"] c["composite_score"] = compute_composite_score(c, gap, employee_skills) all_candidates.append(c) all_candidates.sort(key=lambda x: -x["composite_score"]) for cand in all_candidates: cid = cand["course_id"] if any(s["course_id"] == cid for s in selected_courses): continue if cand["for_gap"] in gap_to_course: continue if total_time + cand["duration_hours"] > max_hours: continue if total_cost + cand["price_usd"] > max_budget: continue prereq_skills = cand.get("prereq_skills", []) _missing_prereqs = [p for p in prereq_skills if p not in covered_skills] prereq_courses_to_add = [] for prereq_cid in cand.get("prereq_course_ids", []): if any(s["course_id"] == prereq_cid for s in selected_courses): continue prereq_row = courses[courses["course_id"] == prereq_cid] if prereq_row.empty: continue pr = prereq_row.iloc[0] prereq_course = { "course_id": pr["course_id"], "course_name": pr["course_name"], "difficulty": pr["difficulty_label"], "difficulty_score": float(pr["difficulty_score"]), "duration_hours": float(pr["duration_hours"]), "price_usd": float(pr["price_usd"]), "rating": float(pr["rating"]), "target_skills": pr["target_skills_list"], "prereq_skills": pr["prereq_skills_list"], "prereq_course_ids": pr["prereq_course_ids_list"], "is_free": bool(pr["is_free"]), "for_gap": cand["for_gap"], "is_prerequisite_course": True, "composite_score": cand["composite_score"] * 0.8, "semantic_score": 0.0, } if total_time + prereq_course["duration_hours"] <= max_hours and \ total_cost + prereq_course["price_usd"] <= max_budget: prereq_courses_to_add.append(prereq_course) total_time += prereq_course["duration_hours"] total_cost += prereq_course["price_usd"] for skill in prereq_course["target_skills"]: covered_skills.add(skill) selected_courses.extend(prereq_courses_to_add) total_time += cand["duration_hours"] total_cost += cand["price_usd"] for skill in cand.get("target_skills", []): covered_skills.add(skill) gap_to_course[cand["for_gap"]] = cid cand["is_prerequisite_course"] = False selected_courses.append(cand) ordered = self._topological_sort(selected_courses) covered_gaps = set(gap_to_course.keys()) total_gaps = len(gaps) coverage_pct = round(len(covered_gaps) / total_gaps * 100, 1) if total_gaps > 0 else 0.0 return { "learning_path": ordered, "total_hours": round(total_time, 1), "total_cost_usd": round(total_cost, 2), "n_courses": len(ordered), "gap_coverage_pct": coverage_pct, "covered_gaps": list(covered_gaps), "uncovered_gaps": [g["canonical_name"] for g in gaps if g["canonical_name"] not in covered_gaps], } def _topological_sort(self, selected_courses: list) -> list: if not selected_courses: return [] selected_ids = {c["course_id"] for c in selected_courses} G_path = nx.DiGraph() for course in selected_courses: G_path.add_node(course["course_id"], difficulty_score=course.get("difficulty_score", 3)) for course in selected_courses: for prereq_id in course.get("prereq_course_ids", []): if prereq_id in selected_ids: G_path.add_edge(prereq_id, course["course_id"]) try: order = list(nx.topological_sort(G_path)) except nx.NetworkXUnfeasible: order = [c["course_id"] for c in sorted(selected_courses, key=lambda x: x.get("difficulty_score", 3))] id_to_course = {c["course_id"]: c for c in selected_courses} return [id_to_course[cid] for cid in order if cid in id_to_course] analyzer = GapAnalyzer() recommender = CourseRecommender() # ── Model / artifact loading ──────────────────────────────────────────────── def load_models(): global skill_vocab, alias_lookup, VOCAB_NORMS, VOCAB_CANONICAL, skill_graph global skill_emb_lookup, courses, course_emb_matrix, course_id_to_idx, course_lookup global gliner_model, sbert_model, SKILL_LABELS, MODEL_NAME, MODELS_LOADED import pickle from gliner import GLiNER from sentence_transformers import SentenceTransformer with open(DATA_DIR / "skill_vocab.pkl", "rb") as f: skill_vocab = pickle.load(f) with open(DATA_DIR / "alias_lookup.pkl", "rb") as f: alias_lookup = pickle.load(f) with open(DATA_DIR / "vocab_norms.pkl", "rb") as f: VOCAB_NORMS = pickle.load(f) with open(DATA_DIR / "vocab_canonical.pkl", "rb") as f: VOCAB_CANONICAL = pickle.load(f) with open(DATA_DIR / "skill_graph.pkl", "rb") as f: skill_graph = pickle.load(f) with open(DATA_DIR / "course_id_to_idx.pkl", "rb") as f: course_id_to_idx = pickle.load(f) with open(DATA_DIR / "course_lookup.pkl", "rb") as f: course_lookup = pickle.load(f) # skill_emb_lookup / courses are loaded from JSON rather than pickle: # raw numpy arrays and pandas DataFrames pickled under one numpy/pandas # major version can fail to unpickle under another (e.g. numpy 1.x vs # 2.x internal module layout). JSON is version-agnostic; the arrays are # rebuilt fresh against whatever numpy is actually installed here. with open(DATA_DIR / "skill_emb_lookup.json") as f: raw_skill_emb = json.load(f) skill_emb_lookup = {name: np.array(vec, dtype=float) for name, vec in raw_skill_emb.items()} with open(DATA_DIR / "courses.json") as f: course_records = json.load(f) courses = pd.DataFrame(course_records) course_emb_matrix = np.load(DATA_DIR / "course_embeddings.npy") with open(DATA_DIR / "extractor_config.json") as f: cfg = json.load(f) SKILL_LABELS = cfg["skill_labels"] MODEL_NAME = cfg["model_name"] try: gliner_model = GLiNER.from_pretrained("gliner-community/gliner_large-v2.5") except Exception: gliner_model = GLiNER.from_pretrained("urchade/gliner_base") sbert_model = SentenceTransformer("all-mpnet-base-v2") MODELS_LOADED = True def to_jsonable(obj): """Recursively convert numpy scalars/arrays (which sneak in via pandas columns and CSV-derived dicts) into native Python types that FastAPI's default JSON encoder can serialize.""" if isinstance(obj, dict): return {k: to_jsonable(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [to_jsonable(v) for v in obj] if isinstance(obj, np.generic): return obj.item() if isinstance(obj, np.ndarray): return obj.tolist() return obj def run_full_pipeline(resume_text: str, jd_text: str, level_hint: str = "", max_hours: float = 200.0, max_budget: float = 500.0) -> dict: resume_skills = extract_resume_skills(resume_text) jd_skills = extract_jd_skills(jd_text, level_hint=level_hint) gap_report = analyzer.analyze(resume_skills, jd_skills) rec_report = recommender.recommend(gap_report["gaps"], resume_skills, max_hours=max_hours, max_budget=max_budget) return { "resume_skills": resume_skills, "jd_skills": jd_skills, "gap_analysis": gap_report, "learning_path": rec_report, "meta": { "n_resume_skills_found": len(resume_skills), "n_jd_skills_found": len(jd_skills), }, } # ── FastAPI app ────────────────────────────────────────────────────────── app = FastAPI(title="Module 2 — Learning Path Recommendation API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) @app.on_event("startup") def _on_startup(): load_models() class AnalyzeTextRequest(BaseModel): resume_text: str jd_text: str level_hint: str = "" max_hours: float = 200.0 max_budget: float = 500.0 @app.get("/health") def health(): return { "status": "ok" if MODELS_LOADED else "loading", "gliner_model": MODEL_NAME, "n_skills_in_vocab": len(skill_vocab), "n_courses": len(courses), } @app.post("/analyze-text") def analyze_text(payload: AnalyzeTextRequest): if not MODELS_LOADED: raise HTTPException(status_code=503, detail="Models are still loading, try again shortly.") if not payload.resume_text.strip() or not payload.jd_text.strip(): raise HTTPException(status_code=400, detail="resume_text and jd_text must not be empty.") try: result = run_full_pipeline( resume_text=payload.resume_text, jd_text=payload.jd_text, level_hint=payload.level_hint, max_hours=payload.max_hours, max_budget=payload.max_budget, ) return to_jsonable(result) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/analyze") async def analyze_file( file: UploadFile = File(...), jd_text: str = Form(...), level_hint: str = Form(""), max_hours: float = Form(200.0), max_budget: float = Form(500.0), ): if not MODELS_LOADED: raise HTTPException(status_code=503, detail="Models are still loading, try again shortly.") file_bytes = await file.read() resume_text = extract_text_from_file(file_bytes, file.filename or "resume.txt") if not resume_text.strip(): raise HTTPException(status_code=400, detail="Could not extract text from the uploaded file.") try: result = run_full_pipeline( resume_text=resume_text, jd_text=jd_text, level_hint=level_hint, max_hours=max_hours, max_budget=max_budget, ) return to_jsonable(result) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/courses") def browse_courses(skill: str, top_k: int = 8): if not MODELS_LOADED: raise HTTPException(status_code=503, detail="Models are still loading, try again shortly.") return to_jsonable({"skill": skill, "results": search_courses_for_skill(skill, top_k=top_k)}) @app.get("/skills") def browse_skills(query: str): if not MODELS_LOADED: raise HTTPException(status_code=503, detail="Models are still loading, try again shortly.") return to_jsonable(normalize_skill(query)) # ── Gradio UI (required by HF Spaces' Gradio SDK / ZeroGPU hosting) ──────── # This Space runs on the free ZeroGPU tier, which is Gradio-SDK-only. The # REST API above is unaffected — every /health, /analyze-text, /analyze, # /courses, /skills route still works exactly as-is. We just also mount a # minimal Gradio form at /ui so the Space has a browsable UI, per the # officially documented pattern for embedding a FastAPI app inside a Gradio # Space: https://www.gradio.app/guides/fastapi-app-with-the-gradio-client import gradio as gr import spaces @spaces.GPU def _zerogpu_registration_stub(): """ Never called in the real request path. HF's ZeroGPU runtime refuses to start a Space ("No @spaces.GPU function detected during startup") unless at least one function is decorated with @spaces.GPU, even though our GLiNER/SBERT workload is CPU-only and never needs GPU allocation. This stub exists purely to satisfy that platform check without pulling any real request into the shared (5 min/day free) GPU queue. """ return True def _gradio_analyze(resume_text, jd_text, level_hint, max_hours, max_budget): if not MODELS_LOADED: return {"error": "Models are still loading, try again shortly."} if not resume_text.strip() or not jd_text.strip(): return {"error": "resume_text and jd_text must not be empty."} try: result = run_full_pipeline( resume_text=resume_text, jd_text=jd_text, level_hint=level_hint, max_hours=float(max_hours), max_budget=float(max_budget), ) return to_jsonable(result) except Exception as e: return {"error": str(e)} demo = gr.Interface( fn=_gradio_analyze, inputs=[ gr.Textbox(label="Resume text", lines=8), gr.Textbox(label="Job description text", lines=8), gr.Textbox(label="Level hint", value="Mid-Level"), gr.Number(label="Max hours", value=200), gr.Number(label="Max budget (USD)", value=500), ], outputs=gr.JSON(label="Result"), title="Module 2 — Learning Path Recommendation", description=( "This Space also exposes a REST API used by the frontend: " "GET /health, POST /analyze-text, POST /analyze, GET /courses, GET /skills." ), ) app = gr.mount_gradio_app(app, demo, path="/ui") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)