from __future__ import annotations import hashlib import json import logging import os import re import threading import time from dataclasses import dataclass from typing import Any, Dict, List, Optional, Sequence, Set, Tuple import numpy as np from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity from sentence_transformers import SentenceTransformer from supabase import create_client logging.basicConfig(level=logging.INFO) logger = logging.getLogger("skillone-api") def _env(*names: str, default: str = "") -> str: for name in names: value = os.getenv(name) if value: return value.strip() return default SUPABASE_URL = _env("SUPABASE_URL", "VITE_SUPABASE_URL") SUPABASE_KEY = _env( "SUPABASE_SERVICE_ROLE_KEY", "SUPABASE_SECRET_KEY", "SUPABASE_PUBLISHABLE_KEY", "VITE_SUPABASE_PUBLISHABLE_KEY", "VITE_SUPABASE_ANON_KEY", ) FRONTEND_ORIGINS = [o.strip() for o in _env("FRONTEND_ORIGINS", default="https://prasannasais.github.io,http://localhost:5173,http://127.0.0.1:5173").split(",") if o.strip()] if not SUPABASE_URL: raise RuntimeError("Missing SUPABASE_URL (or VITE_SUPABASE_URL).") if not SUPABASE_KEY: raise RuntimeError("Missing Supabase key. Use SUPABASE_SERVICE_ROLE_KEY / SUPABASE_SECRET_KEY in Hugging Face secrets for backend access.") supabase = create_client(SUPABASE_URL, SUPABASE_KEY) app = FastAPI( title="SkillOne API", description="AI-driven knowledge graph model for personalized learning path generation", version="3.0.0", ) app.add_middleware( CORSMiddleware, allow_origins=FRONTEND_ORIGINS, allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) class ModelConfig: TFIDF_WEIGHT = 0.40 SEMANTIC_WEIGHT = 0.60 MIN_RELEVANCE_SCORE = 0.30 EDUCATION_EXACT_MATCH_BOOST = 1.8 EDUCATION_ADJACENT_BOOST = 1.3 EDUCATION_MISMATCH_PENALTY = 0.5 DIFFICULTY_EXACT_MATCH_BOOST = 1.6 DIFFICULTY_PROGRESSIVE_BOOST = 1.3 DIFFICULTY_MISMATCH_PENALTY = 0.4 CAREER_KEYWORD_MATCH_BOOST = 2.5 SKILL_KEYWORD_MATCH_BOOST = 1.2 MIN_PATH_LENGTH = 6 MAX_PATH_LENGTH = 12 OPTIMAL_PATH_LENGTH = 8 config = ModelConfig() class LearnerProfile(BaseModel): learner_id: str = Field(..., min_length=1) career_goal: str = Field(..., min_length=1) education_level: str = Field(..., min_length=1) desired_skills: List[str] = Field(default_factory=list) interests: Optional[List[str]] = Field(default_factory=list) proficiency_level: Optional[str] = "Beginner" @dataclass class CourseCache: signature: str courses: List[Dict[str, Any]] course_texts: List[str] vectorizer: TfidfVectorizer tfidf_matrix: Any embeddings: np.ndarray course_by_id: Dict[Any, Dict[str, Any]] id_to_index: Dict[Any, int] prereq_graph: Dict[Any, Set[Any]] level_map: Dict[str, int] difficulty_map: Dict[str, int] _CACHE_LOCK = threading.Lock() _COURSE_CACHE: Optional[CourseCache] = None SEMANTIC_MODEL = SentenceTransformer("all-MiniLM-L6-v2") def _norm_text(value: Any) -> str: if value is None: return "" if isinstance(value, str): value = value.strip() return re.sub(r"\s+", " ", value) return re.sub(r"\s+", " ", str(value).strip()) def _safe_list(value: Any) -> List[str]: if value is None: return [] if isinstance(value, list): return [str(x).strip() for x in value if str(x).strip()] if isinstance(value, str): raw = value.strip() if not raw: return [] try: parsed = json.loads(raw) if isinstance(parsed, list): return [str(x).strip() for x in parsed if str(x).strip()] except Exception: pass return [p.strip() for p in re.split(r"[,\|;]", raw) if p.strip()] return [str(value).strip()] if str(value).strip() else [] def _course_signature(courses: Sequence[Dict[str, Any]]) -> str: items = [] for c in courses: cid = str(c.get("id", "")) updated = str(c.get("updated_at") or c.get("modified_at") or c.get("created_at") or "") items.append(f"{cid}:{updated}") blob = "|".join(sorted(items)) return hashlib.sha256(blob.encode("utf-8")).hexdigest() def _build_maps(courses: Sequence[Dict[str, Any]]) -> Tuple[Dict[Any, int], Dict[Any, Set[Any]], Dict[str, int], Dict[str, int]]: id_to_index = {c["id"]: i for i, c in enumerate(courses) if "id" in c} prereq_graph: Dict[Any, Set[Any]] = {} for course in courses: cid = course.get("id") prereqs = _safe_list(course.get("prerequisite_course_ids")) prereq_graph[cid] = {p for p in prereqs if p in id_to_index} level_map = {"High School": 1, "Undergraduate": 2, "Graduate": 3, "Professional": 4} difficulty_map = {"Beginner": 1, "Intermediate": 2, "Advanced": 3} return id_to_index, prereq_graph, level_map, difficulty_map def _build_course_text(course: Dict[str, Any]) -> str: tags = _safe_list(course.get("tags")) title = _norm_text(course.get("title")) desc = _norm_text(course.get("description")) return f"{title} {desc} {' '.join(tags)}".strip() def _fetch_courses() -> List[Dict[str, Any]]: try: response = supabase.table("courses").select("*").order("id").execute() return response.data or [] except Exception as exc: logger.exception("Course fetch failed") raise HTTPException(status_code=500, detail=f"Failed to retrieve courses: {exc}") from exc def _build_cache(courses: List[Dict[str, Any]]) -> CourseCache: signature = _course_signature(courses) course_texts = [_build_course_text(c) for c in courses] vectorizer = TfidfVectorizer( stop_words="english", max_features=min(5000, max(1000, len(courses) * 30)), ngram_range=(1, 2), ) tfidf_matrix = vectorizer.fit_transform(course_texts) if courses else vectorizer.fit_transform(["placeholder"]) embeddings = ( SEMANTIC_MODEL.encode( course_texts if courses else ["placeholder"], batch_size=64, show_progress_bar=False, normalize_embeddings=True, ) if courses else np.zeros((1, 384), dtype=np.float32) ) id_to_index, prereq_graph, level_map, difficulty_map = _build_maps(courses) course_by_id = {c["id"]: c for c in courses if "id" in c} logger.info("Built cache for %d courses", len(courses)) return CourseCache( signature=signature, courses=courses, course_texts=course_texts, vectorizer=vectorizer, tfidf_matrix=tfidf_matrix, embeddings=np.asarray(embeddings), course_by_id=course_by_id, id_to_index=id_to_index, prereq_graph=prereq_graph, level_map=level_map, difficulty_map=difficulty_map, ) def get_course_cache() -> CourseCache: global _COURSE_CACHE with _CACHE_LOCK: courses = _fetch_courses() signature = _course_signature(courses) if _COURSE_CACHE is None or _COURSE_CACHE.signature != signature: _COURSE_CACHE = _build_cache(courses) return _COURSE_CACHE def _token_set(text: str) -> Set[str]: return {t for t in re.split(r"\W+", text.lower()) if len(t) > 1} def _rank_courses(profile: LearnerProfile, cache: CourseCache) -> np.ndarray: learner_text = _norm_text( " ".join( [ profile.career_goal, " ".join(profile.desired_skills or []), " ".join(profile.interests or []), ] ) ) learner_tfidf = cache.vectorizer.transform([learner_text]) tfidf_scores = cosine_similarity(learner_tfidf, cache.tfidf_matrix)[0] learner_embedding = SEMANTIC_MODEL.encode( learner_text or " ", show_progress_bar=False, normalize_embeddings=True, ) semantic_scores = np.dot(cache.embeddings, np.asarray(learner_embedding, dtype=np.float32)) combined_scores = (config.TFIDF_WEIGHT * tfidf_scores) + (config.SEMANTIC_WEIGHT * semantic_scores) learner_level = cache.level_map.get(profile.education_level, 2) learner_difficulty = cache.difficulty_map.get(profile.proficiency_level or "Beginner", 1) career_keywords = _token_set(profile.career_goal) skill_keywords = _token_set(" ".join(profile.desired_skills or [])) for i, course in enumerate(cache.courses): course_level = cache.level_map.get(course.get("education_level", "Undergraduate"), 2) course_difficulty = cache.difficulty_map.get(course.get("difficulty_level", "Beginner"), 1) tags = _token_set(" ".join(_safe_list(course.get("tags")))) title_tokens = _token_set(course.get("title", "")) level_diff = abs(course_level - learner_level) if level_diff == 0: combined_scores[i] *= config.EDUCATION_EXACT_MATCH_BOOST elif level_diff == 1: combined_scores[i] *= config.EDUCATION_ADJACENT_BOOST else: combined_scores[i] *= config.EDUCATION_MISMATCH_PENALTY difficulty_diff = abs(course_difficulty - learner_difficulty) if difficulty_diff == 0: combined_scores[i] *= config.DIFFICULTY_EXACT_MATCH_BOOST elif difficulty_diff == 1 and course_difficulty >= learner_difficulty: combined_scores[i] *= config.DIFFICULTY_PROGRESSIVE_BOOST elif difficulty_diff > 1: combined_scores[i] *= config.DIFFICULTY_MISMATCH_PENALTY career_overlap = len(career_keywords & (tags | title_tokens)) skill_overlap = len(skill_keywords & (tags | title_tokens)) if career_overlap: combined_scores[i] *= (1 + career_overlap * 0.20) * config.CAREER_KEYWORD_MATCH_BOOST if skill_overlap: combined_scores[i] *= (1 + skill_overlap * 0.15) * config.SKILL_KEYWORD_MATCH_BOOST if tags and title_tokens: combined_scores[i] *= 1.02 return combined_scores def _prereq_order(course_id: Any, prereq_graph: Dict[Any, Set[Any]], selected: Set[Any], seen_global: Set[Any]) -> List[Any]: order: List[Any] = [] stack: List[Tuple[Any, int]] = [(course_id, 0)] visiting: Set[Any] = set() local_seen: Set[Any] = set() while stack: node, state = stack.pop() if state == 0: if node in local_seen or node in seen_global: continue local_seen.add(node) visiting.add(node) stack.append((node, 1)) for prereq in prereq_graph.get(node, set()): if prereq in selected and prereq not in seen_global and prereq not in local_seen: stack.append((prereq, 0)) else: visiting.discard(node) if node not in seen_global: order.append(node) return order def _build_path(cache: CourseCache, ranked_indices: List[int], min_length: int, max_length: int) -> List[int]: if not ranked_indices: return [] selected_ids = [cache.courses[i]["id"] for i in ranked_indices] selected_set = set(selected_ids) added: Set[Any] = set() path: List[int] = [] for idx in ranked_indices: cid = cache.courses[idx]["id"] closure = _prereq_order(cid, cache.prereq_graph, selected_set, added) for node in closure: if node in added: continue if node in cache.id_to_index: path.append(cache.id_to_index[node]) added.add(node) if len(path) >= max_length: break if len(path) >= max_length: break if len(path) < min_length: for idx in ranked_indices: if idx not in path: path.append(idx) if len(path) >= min_length: break deduped: List[int] = [] seen: Set[int] = set() for idx in path: if idx not in seen: deduped.append(idx) seen.add(idx) if len(deduped) >= max_length: break return deduped[:max_length] async def save_learner_profile(profile: LearnerProfile) -> None: payload = { "learner_id": profile.learner_id, "career_goal": profile.career_goal, "education_level": profile.education_level, "desired_skills": profile.desired_skills, "interests": profile.interests or [], "proficiency_level": profile.proficiency_level, } try: supabase.table("learner_profiles").upsert(payload).execute() except Exception as exc: logger.exception("save_learner_profile failed: %s", exc) async def save_learning_path(learner_id: str, path: List[str], scores: Dict[str, float], reasoning: str) -> None: payload = { "learner_id": learner_id, "course_sequence": path, "relevance_scores": scores, "reasoning": reasoning, } try: supabase.table("learning_paths").delete().eq("learner_id", learner_id).execute() supabase.table("learning_paths").insert(payload).execute() except Exception as exc: logger.exception("save_learning_path failed: %s", exc) async def log_career_goal(career_goal: str) -> None: try: current = supabase.table("career_goal_logs").select("career_goal, frequency").eq("career_goal", career_goal).limit(1).execute() if current.data: frequency = int(current.data[0].get("frequency", 0)) + 1 supabase.table("career_goal_logs").update({"frequency": frequency}).eq("career_goal", career_goal).execute() else: supabase.table("career_goal_logs").insert({"career_goal": career_goal, "frequency": 1}).execute() except Exception as exc: logger.exception("log_career_goal failed: %s", exc) @app.get("/api/career-goals/suggestions") async def get_career_goal_suggestions(query: str = Query(..., min_length=1, max_length=100)): try: response = ( supabase.table("career_goal_logs") .select("career_goal, frequency") .ilike("career_goal", f"%{query}%") .order("frequency", desc=True) .limit(10) .execute() ) suggestions = [row["career_goal"] for row in (response.data or [])] return {"suggestions": suggestions} except Exception as exc: logger.exception("suggestions failed: %s", exc) return {"suggestions": [], "error": str(exc)} @app.post("/api/generate-learning-path") async def generate_learning_path(profile: LearnerProfile): try: await save_learner_profile(profile) cache = get_course_cache() if not cache.courses: raise HTTPException(status_code=404, detail="No courses found") scores = _rank_courses(profile, cache) top_ranked = np.argsort(scores)[::-1].tolist() relevant = [i for i in top_ranked if scores[i] >= config.MIN_RELEVANCE_SCORE] if len(relevant) < config.MIN_PATH_LENGTH: relevant = top_ranked[: max(config.MIN_PATH_LENGTH, min(len(top_ranked), config.OPTIMAL_PATH_LENGTH))] uniq_path_idxs = _build_path(cache, relevant, config.MIN_PATH_LENGTH, config.MAX_PATH_LENGTH) if not uniq_path_idxs: uniq_path_idxs = top_ranked[: min(config.OPTIMAL_PATH_LENGTH, len(top_ranked))] top_path = [cache.courses[i]["id"] for i in uniq_path_idxs] path_scores = {str(cache.courses[i]["id"]): float(scores[i]) for i in uniq_path_idxs} matched_skills: List[str] = [] learner_skill_tokens = [s.lower() for s in (profile.desired_skills or [])] for skill in learner_skill_tokens: for i in uniq_path_idxs: course_tags = " ".join(_safe_list(cache.courses[i].get("tags"))).lower() course_title = str(cache.courses[i].get("title", "")).lower() if skill in course_tags or skill in course_title: matched_skills.append(skill) break reasoning = ( f"Path generated using TF-IDF ({config.TFIDF_WEIGHT}) + semantic embeddings ({config.SEMANTIC_WEIGHT}), " f"education alignment, difficulty alignment, keyword overlap, and prerequisite-aware iterative ordering. " f"Matched skills: {', '.join(matched_skills[:5]) or 'none'}. " f"Career goal: {profile.career_goal}" ) await save_learning_path(profile.learner_id, top_path, path_scores, reasoning) await log_career_goal(profile.career_goal) return { "learning_path": top_path, "scores": path_scores, "reasoning": reasoning, "total_courses": len(top_path), "pathway_details": [ { "course_id": cache.courses[i]["id"], "title": cache.courses[i].get("title", ""), "difficulty": cache.courses[i].get("difficulty_level", "Beginner"), "score": float(scores[i]), } for i in uniq_path_idxs ], } except HTTPException: raise except Exception as exc: logger.exception("generate_learning_path failed: %s", exc) raise HTTPException(status_code=500, detail=str(exc)) from exc @app.get("/health") async def health_check(): status = { "status": "healthy", "database": "connected", "ml_models": "ready", "config": { "tfidf_weight": config.TFIDF_WEIGHT, "semantic_weight": config.SEMANTIC_WEIGHT, "min_relevance": config.MIN_RELEVANCE_SCORE, "path_length_range": f"{config.MIN_PATH_LENGTH}-{config.MAX_PATH_LENGTH}", "rls_safe_backend": bool( SUPABASE_KEY.startswith("sb_secret_") or "service_role" in SUPABASE_KEY.lower() ), }, } try: status["course_cache_size"] = len(get_course_cache().courses) except Exception as exc: status["database"] = "degraded" status["cache_error"] = str(exc) return status def _keep_alive() -> None: while True: try: time.sleep(600) logger.info("keep-alive tick") except Exception as exc: logger.error("keep-alive error: %s", exc) threading.Thread(target=_keep_alive, daemon=True).start() if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8000")))