""" recommender.py ────────────── Student recommendations, moved here from ifind/lib/recommendation so everything that touches vectors and the HNSW graph lives in one service. For one student: run their BERT vector through the WHOLE graph (k-NN), fetch the hit internships (scraped 'internships' and employer-posted 'internships.this-platform'), score each one with the weighted cosine (0.4 TF-IDF + 0.6 BERT, vectors are L2-normalised so the dot product is the cosine), drop scores under the threshold, keep the top N and save them on the user as `recommendedInternships`. Triggered by (see app.py): - /encode-resume with user_id + recommend a student's resume or profile changed - /vectorize-hnsw, /vectorize-platform new internships entered the graph -> every student is re-run - /recommend-all admin refresh (all students, or the given user_ids) """ import logging import threading from datetime import datetime, timezone from typing import Any, Dict, Iterable, List, Optional, Tuple import numpy as np from vectorizer_hnsw_pipeline import ( APPROVED_STATUSES, LIVE_COLLECTION, PLATFORM_COLLECTION, deadline_passed, get_db, get_search_manager, purge_expired, ) log = logging.getLogger("recommender") TOP_N = 20 THRESHOLD = 0.1 # Small nudges for the student's location (the profile "preferences" city / state): they only reorder close matches. CITY_BOOST = 0.08 STATE_BOOST = 0.04 REMOTE_BOOST = 0.03 TFIDF_WEIGHT = 0.4 BERT_WEIGHT = 0.6 K = TOP_N * 2 * 2 # candidates pulled from the graph (same as the old HNSWStrategy: topN*2, doubled) # One fan-out at a time; requests that arrive meanwhile are folded into one extra pass at the end. _FANOUT_LOCK = threading.Lock() _fanout_dirty = False def _visible_platform_filter() -> Dict[str, Any]: """Same rule as ifind lib/employer/visibility.ts visibleToStudentsFilter().""" now = datetime.now(timezone.utc) return { "status": "published", "moderation.status": {"$in": list(APPROVED_STATUSES)}, "isActive": {"$ne": False}, "bannedByCompany": {"$ne": True}, # company banned by a moderator (ifind companies.restriction.status == "banned") "$or": [{"deadlineDate": None}, {"deadlineDate": {"$gt": now}}], } def _norm(v) -> str: return str(v or "").strip().lower() def location_boost(location: Optional[Dict[str, Any]], is_remote: bool, addresses: Optional[list]) -> float: """Boost for an internship that is remote, or in the student's city / state. 0 when the student has no location.""" if not location: return 0.0 city, state = _norm(location.get("city")), _norm(location.get("state")) if not city and not state: return 0.0 boost = REMOTE_BOOST if is_remote else 0.0 for a in addresses or []: if not isinstance(a, dict): continue if city and _norm(a.get("city")) == city: boost = max(boost, CITY_BOOST) elif state and _norm(a.get("state")) == state: boost = max(boost, STATE_BOOST) return boost class _Candidates: """Candidate vectors by id, kept across users in one run (the same internships come up again and again).""" def __init__(self, db): self.db = db self.cache: Dict[str, Optional[Tuple[np.ndarray, np.ndarray]]] = {} self.where: Dict[str, Tuple[bool, list]] = {} # id -> (isRemote, address[]) for the location boost def get(self, ids: Iterable[str]) -> Dict[str, Tuple[np.ndarray, np.ndarray]]: from bson import ObjectId ids = list(ids) missing = [i for i in ids if i not in self.cache] if missing: oids = [ObjectId(i) for i in missing] proj = {"tfidf_vector": 1, "bert_vector": 1, "isRemote": 1, "address": 1, "deadlineDate": 1} now = datetime.now(timezone.utc) found: Dict[str, Tuple[np.ndarray, np.ndarray]] = {} live = {"_id": {"$in": oids}, "$or": [{"isActive": True}, {"isActive": {"$exists": False}}]} plat = {**_visible_platform_filter(), "_id": {"$in": oids}} for col, q in ((LIVE_COLLECTION, live), (PLATFORM_COLLECTION, plat)): for d in self.db[col].find(q, proj): if deadline_passed(d.get("deadlineDate"), now): continue # expired: never recommended, whatever the graph still holds if isinstance(d.get("bert_vector"), list) and isinstance(d.get("tfidf_vector"), list): self.where[str(d["_id"])] = (bool(d.get("isRemote")), d.get("address") if isinstance(d.get("address"), list) else []) found[str(d["_id"])] = ( np.asarray(d["tfidf_vector"], dtype=np.float32), np.asarray(d["bert_vector"], dtype=np.float32), ) for i in missing: self.cache[i] = found.get(i) # None = not recommendable right now (inactive, paused, no vectors) return {i: self.cache[i] for i in ids if self.cache.get(i) is not None} def _brute_force_ids(db) -> List[str]: """Graph unavailable: every recommendable internship (the old BruteForceStrategy). ponytail: O(n), fine as a fallback.""" ids = [str(d["_id"]) for d in db[LIVE_COLLECTION].find( {"$or": [{"isActive": True}, {"isActive": {"$exists": False}}], "bert_vector": {"$type": "array"}}, {"_id": 1, "deadlineDate": 1}) if not deadline_passed(d.get("deadlineDate"))] ids += [str(d["_id"]) for d in db[PLATFORM_COLLECTION].find( {**_visible_platform_filter(), "bert_vector": {"$type": "array"}}, {"_id": 1, "deadlineDate": 1}) if not deadline_passed(d.get("deadlineDate"))] return ids def score_user(db, mgr, cands: _Candidates, tfidf: List[float], bert: List[float], location: Optional[Dict[str, Any]] = None) -> List[Tuple[str, float]]: """Run one student's vectors through the graph and return the top N (id, score), best first.""" try: ids = [i for i, _ in mgr.search(bert, K)] except Exception as err: log.error("Graph search failed, falling back to a full scan: %s", err) ids = _brute_force_ids(db) u_tfidf = np.asarray(tfidf, dtype=np.float32) u_bert = np.asarray(bert, dtype=np.float32) scored = [] for i, (c_tfidf, c_bert) in cands.get(ids).items(): t = float(u_tfidf @ c_tfidf) if c_tfidf.shape == u_tfidf.shape else 0.0 b = float(u_bert @ c_bert) if c_bert.shape == u_bert.shape else 0.0 score = t * TFIDF_WEIGHT + b * BERT_WEIGHT if score >= THRESHOLD: remote, addresses = cands.where.get(i, (False, [])) scored.append((i, score + location_boost(location, remote, addresses))) scored.sort(key=lambda x: x[1], reverse=True) return scored[:TOP_N] def save_recommendations(db, user_id, recs: List[Tuple[str, float]]) -> None: from bson import ObjectId db["users"].update_one( {"_id": ObjectId(str(user_id))}, {"$set": { "recommendedInternships.updatedAt": datetime.now(timezone.utc), "recommendedInternships.recommendedList": [ObjectId(i) for i, _ in recs], "recommendedInternships.recommendedScores": [{"id": ObjectId(i), "score": round(s, 3)} for i, s in recs], }}, ) def recommend_user(user_id: str, tfidf: List[float], bert: List[float], db=None) -> int: """Run ONE student through the whole graph and save their list. Returns how many were recommended.""" db = db if db is not None else get_db() from bson import ObjectId mgr = get_search_manager(db, dim=len(bert)) graph_count = mgr.index.get_current_count() if mgr and mgr.index else len(mgr.uid_to_id) user_doc = db["users"].find_one({"_id": ObjectId(str(user_id))}, {"username": 1, "email": 1, "name": 1, "city": 1, "state": 1}) username = user_doc.get("username") or user_doc.get("email") or user_doc.get("name") or str(user_id) if user_doc else str(user_id) log.info("Running recommender for student '%s' (id=%s) against %d graph internships...", username, user_id, graph_count) recs = score_user(db, mgr, _Candidates(db), tfidf, bert, user_doc) save_recommendations(db, user_id, recs) top_score = round(recs[0][1], 3) if recs else 0.0 log.ok("Recommended %d internships for student '%s' (top match: %.3f)", len(recs), username, top_score) return len(recs) def recommend_users(user_ids: Optional[List[str]] = None, db=None) -> Dict[str, int]: """Run every student with resume vectors (or just `user_ids`) through the whole graph.""" from bson import ObjectId db = db if db is not None else get_db() stats = {"users": 0, "updated": 0, "errors": 0} query: Dict[str, Any] = {"resume.bert_vector": {"$type": "array"}, "resume.tfidf_vector": {"$type": "array"}} if user_ids is not None: oids = [] for u in user_ids: try: oids.append(ObjectId(u)) except Exception: stats["errors"] += 1 query["_id"] = {"$in": oids} mgr = get_search_manager(db) graph_count = mgr.index.get_current_count() if mgr and mgr.index else (len(mgr.uid_to_id) if mgr else 0) total_users = db["users"].count_documents(query) target_desc = f"{total_users} specified student(s)" if user_ids is not None else f"every student ({total_users} with active resumes)" log.info("Running recommender fan-out on %s against %d graph internships (top %d per student)...", target_desc, graph_count, TOP_N) if total_users == 0: log.info("No matching students found in database — 0 recommendations needed") return stats cands = _Candidates(db) cursor = db["users"].find( query, {"username": 1, "email": 1, "name": 1, "resume.tfidf_vector": 1, "resume.bert_vector": 1, "city": 1, "state": 1} ) idx = 0 for user in cursor: idx += 1 stats["users"] += 1 username = user.get("username") or user.get("email") or user.get("name") or str(user["_id"]) try: bert = user["resume"]["bert_vector"] if mgr is None or mgr.dim != len(bert): mgr = get_search_manager(db, dim=len(bert)) recs = score_user(db, mgr, cands, user["resume"]["tfidf_vector"], bert, user) save_recommendations(db, user["_id"], recs) stats["updated"] += 1 top_score = round(recs[0][1], 3) if recs else 0.0 log.info("[%d/%d] Recommended for student '%s': %d internships matched (top score: %.3f)", idx, total_users, username, len(recs), top_score) except Exception as err: log.error("[%d/%d] Recommendation failed for student '%s' (id=%s): %s", idx, total_users, username, user.get("_id"), err) stats["errors"] += 1 if stats["errors"] == 0: log.ok("Recommender complete: %d/%d students updated successfully, 0 failed (graph size: %d internships)", stats["updated"], stats["users"], graph_count) else: log.warn("Recommender finished with issues: %d updated, %d failed out of %d students", stats["updated"], stats["errors"], stats["users"]) return stats def run_fanout(user_ids: Optional[List[str]] = None) -> Dict[str, Any]: """ recommend_users() with coalescing: when one is already running, ask it to go once more when it finishes instead of piling up (each approval in a moderator batch would otherwise start its own pass). A targeted run (user_ids) is never coalesced. """ global _fanout_dirty try: purge_expired() # expired internships leave the collection and the graph before anyone is scored except Exception as err: log.error("Expiry sweep failed (the recommender still skips expired internships): %s", err) if user_ids is not None: return recommend_users(user_ids) if not _FANOUT_LOCK.acquire(blocking=False): _fanout_dirty = True log.info("Recommender fan-out already in progress — marked dirty to re-run on completion") return {"status": "coalesced"} try: pass_count = 1 while True: _fanout_dirty = False log.info("Starting recommender fan-out pass #%d on all students...", pass_count) stats = recommend_users() if not _fanout_dirty: return stats pass_count += 1 log.info("Fan-out marked dirty during pass #%d, running subsequent pass...", pass_count - 1) finally: _FANOUT_LOCK.release()