diff --git a/backend/app/database.py b/backend/app/database.py index 787d654..e35b5ff 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -314,86 +314,79 @@ def _norm_loc_value(v: Any) -> str: return s.casefold() if s else "" -def _remote_substring_ors() -> List[Dict[str, Any]]: - r = "remote" - return [ - {_M_CITY: {"$regex": r, "$options": "i"}}, - {_M_COUNTY: {"$regex": r, "$options": "i"}}, - ] - - -def _field_contains_substr_regex( - field: str, needle_cf: str -) -> Optional[Dict[str, Any]]: - if not needle_cf: - return None - return {field: {"$regex": re.escape(needle_cf), "$options": "i"}} - - -def _expr_haystack_contains_mongo_subfield( - haystack_casefold: str, dollar_field: str -) -> Optional[Dict[str, Any]]: - """True when haystack (user string) contains the job’s city/county (Python: job in user). - - Requires a non-empty job field: MongoDB matches an empty substring at index 0 for - ``$indexOfCP``, which would incorrectly match every document if city/county were missing. - """ - if not haystack_casefold: - return None - needle = {"$ifNull": [{"$toLower": dollar_field}, ""]} - return { - "$expr": { - "$and": [ - {"$gt": [{"$strLenCP": needle}, 0]}, - {"$gte": [{"$indexOfCP": [haystack_casefold, needle]}, 0]}, - ] - } - } - - -def _location_or_clauses_for_one_user(user: dict) -> List[Dict[str, Any]]: - """Superset of matching_service._job_matches_user_location, on classifier_metadata fields.""" - uc = _norm_loc_value(user.get("city")) - up = _norm_loc_value(user.get("province")) - ors: List[Dict[str, Any]] = list(_remote_substring_ors()) - if not uc or not up: - return ors - for field in (_M_CITY, _M_COUNTY, _M_PROVINCE): - f_c = _field_contains_substr_regex(field, uc) - if f_c is not None: - ors.append(f_c) - f_p = _field_contains_substr_regex(field, up) - if f_p is not None: - ors.append(f_p) - for hay, fpath in ( - (uc, "$classifier_metadata.city"), - (uc, "$classifier_metadata.county"), - (uc, "$classifier_metadata.province"), - (up, "$classifier_metadata.city"), - (up, "$classifier_metadata.county"), - (up, "$classifier_metadata.province"), - ): - ex = _expr_haystack_contains_mongo_subfield(hay, fpath) - if ex is not None: - ors.append(ex) - return ors +# Case-insensitive collation used by both the location-aware indexes and the +# location filter query. ``strength: 2`` makes string comparisons ignore case and +# diacritics (e.g. "Nairobi" == "NAIROBI" == "nairobi"). The query MUST pass this +# same collation to ``find()`` for the planner to pick the matching index. +LOCATION_COLLATION: Dict[str, Any] = {"locale": "en", "strength": 2} + +# Value (not a flag) — included in the $in clause so fully-remote postings always match. +_REMOTE_LOCATION_VALUE = "Remote" + + +def _stripped_user_locations(user: dict) -> List[str]: + """City + province for one user, stripped. Case is preserved — the query's + collation handles case-insensitive matching at the index level.""" + locations: List[str] = [] + for field_name in ("city", "province"): + value = user.get(field_name) + if value is None: + continue + stripped = str(value).strip() + if stripped: + locations.append(stripped) + return locations def build_mongo_filter_active_and_location( users: Sequence[dict], ) -> Optional[Dict[str, Any]]: - """ - is_active and (OR of all per-user location clauses). None if the caller should - use active-only (no user context or empty list). + """is_active AND (city/county/province ∈ {user_cities ∪ user_provinces ∪ 'Remote'}). + + Returns ``None`` if the caller should use the active-only filter (no users supplied). + If all users have empty city/province, the location $in collapses to ``{'Remote'}`` + so the resulting filter restricts to remote-tagged jobs — same as the previous + regex-based implementation. + + Match semantics: **case-insensitive equality** via Mongo collation + (``LOCATION_COLLATION``). Callers (the ranked-jobs find) MUST pass the same + collation to ``find()`` for the planner to use the matching + ``is_active_{city,county,province}_ci_-_id`` compound index — otherwise the + planner falls back to a collection scan and the filter still works but isn't + index-served. + + Multi-user requests are folded into a single ``$in`` per field — no per-user + clause duplication, which keeps the query small for batch traffic. + + Behavior change vs. the previous ``$regex``/``$expr/$indexOfCP`` implementation: + a job with ``city = "Nairobi East"`` no longer matches a user whose city is + ``"Nairobi"``. The collation gives us case-insensitive equality, not substring. """ if not users: return None - parts: List[Dict[str, Any]] = [] - for u in users: - parts.extend(_location_or_clauses_for_one_user(u)) - if not parts: - return RANKED_JOBS_ACTIVE_FILTER - return {"$and": [RANKED_JOBS_ACTIVE_FILTER, {"$or": parts}]} + + # When users have no usable city/province, location_values ends up as just + # {_REMOTE_LOCATION_VALUE} — the resulting filter restricts to remote-tagged jobs. + # That matches the previous behavior of _location_or_clauses_for_one_user (which + # returned only the remote-substring clauses when city or province was missing). + location_values: set[str] = {_REMOTE_LOCATION_VALUE} + for user in users: + for location in _stripped_user_locations(user): + location_values.add(location) + + mongo_in_clause = {"$in": sorted(location_values)} + return { + "$and": [ + RANKED_JOBS_ACTIVE_FILTER, + { + "$or": [ + {_M_CITY: mongo_in_clause}, + {_M_COUNTY: mongo_in_clause}, + {_M_PROVINCE: mongo_in_clause}, + ] + }, + ] + } def build_job_dict_from_ranked(rd: Dict[str, Any]) -> Optional[Dict[str, Any]]: @@ -577,7 +570,8 @@ async def get_all_jobs_with_timing(users: Optional[Sequence[dict]] = None): jobs_retrieval_filter_applied, jobs_find_use_projection """ t_total = time.perf_counter() - t0 = time.perf_counter() + + t_filter_build = time.perf_counter() filt: Dict[str, Any] = RANKED_JOBS_ACTIVE_FILTER retrieval_applied = False if JOBS_RETRIEVAL_FILTER and users: @@ -585,17 +579,40 @@ async def get_all_jobs_with_timing(users: Optional[Sequence[dict]] = None): if built is not None and built != RANKED_JOBS_ACTIVE_FILTER: filt = built retrieval_applied = True + filter_build_ms = _ms(t_filter_build) + + t_cursor_create = time.perf_counter() col = db[MONGO_JOBS_COLLECTION] + # When the location filter is applied we MUST pass the matching collation so + # the planner can use the case-insensitive is_active_{city,county,province}_ci + # compound indexes. Without the collation the same query falls back to a + # collection scan. When no location filter is applied we skip the collation + # so Mongo can pick the simpler is_active_-_id index. + find_kwargs: Dict[str, Any] = {} if JOBS_FIND_USE_PROJECTION: - cursor = col.find(filt, RANKED_JOB_FIND_PROJECTION) - else: - cursor = col.find(filt) + find_kwargs["projection"] = RANKED_JOB_FIND_PROJECTION + if retrieval_applied: + find_kwargs["collation"] = LOCATION_COLLATION + cursor = col.find(filt, **find_kwargs) if retrieval_applied: cursor = cursor.sort([("_id", -1)]) if JOBS_RETRIEVAL_LIMIT > 0: cursor = cursor.limit(JOBS_RETRIEVAL_LIMIT) - ranked_docs = [d async for d in cursor] - mongo_ranked_find_ms = _ms(t0) + cursor_create_ms = _ms(t_cursor_create) + + # Drain the cursor: time-to-first-doc (server exec + first network batch) vs + # remaining drain (the rest of the network transfer + iter overhead). + t_drain_start = time.perf_counter() + ranked_docs: List[dict] = [] + cursor_first_doc_ms = 0.0 + first = True + async for d in cursor: + if first: + cursor_first_doc_ms = _ms(t_drain_start) + first = False + ranked_docs.append(d) + cursor_drain_remaining_ms = _ms(t_drain_start) - cursor_first_doc_ms + mongo_ranked_find_ms = _ms(t_filter_build) # legacy total kept for back-compat t0 = time.perf_counter() jobs: List[dict] = [] @@ -616,8 +633,31 @@ async def get_all_jobs_with_timing(users: Optional[Sequence[dict]] = None): len(ranked_docs), skipped, ) + try: + from app.match_timing_log import log_match_step as _log_match_step + _log_match_step( + "mongo /jobs", + "fetch sub-stages", + n_jobs=len(jobs), + n_ranked_raw=len(ranked_docs), + filter_build_ms=filter_build_ms, + cursor_create_ms=cursor_create_ms, + cursor_first_doc_ms=cursor_first_doc_ms, + cursor_drain_remaining_ms=cursor_drain_remaining_ms, + python_build_jobs_ms=python_build_jobs_ms, + get_all_jobs_total_ms=total_ms, + retrieval_filter_applied=retrieval_applied, + projection_enabled=JOBS_FIND_USE_PROJECTION, + ) + except Exception: + pass + return jobs, { "mongo_ranked_find_ms": mongo_ranked_find_ms, + "filter_build_ms": filter_build_ms, + "cursor_create_ms": cursor_create_ms, + "cursor_first_doc_ms": cursor_first_doc_ms, + "cursor_drain_remaining_ms": cursor_drain_remaining_ms, "python_build_jobs_ms": python_build_jobs_ms, "n_ranked_raw": len(ranked_docs), "n_jobs": len(jobs), @@ -758,6 +798,40 @@ def build_jobs_browse_filter( [("is_active", ASCENDING), ("classifier_metadata.source_platform", ASCENDING)], name="is_active_source_platform", ), + # Location-aware match retrieval. is_active + case-insensitive equality on the + # writer's existing classifier_metadata.{city,county,province} fields. The + # ``collation`` on each index (strength=2 = case + diacritic insensitive) is what + # makes the index serve case-insensitive equality. The query MUST pass the same + # collation (LOCATION_COLLATION) to find() — without that the planner ignores + # this index and falls back to a collection scan. Sort by _id desc so the + # JOBS_RETRIEVAL_LIMIT cap takes the most-recent matches. + IndexModel( + [ + ("is_active", ASCENDING), + ("classifier_metadata.city", ASCENDING), + ("_id", DESCENDING), + ], + name="is_active_city_ci_-_id", + collation=LOCATION_COLLATION, + ), + IndexModel( + [ + ("is_active", ASCENDING), + ("classifier_metadata.county", ASCENDING), + ("_id", DESCENDING), + ], + name="is_active_county_ci_-_id", + collation=LOCATION_COLLATION, + ), + IndexModel( + [ + ("is_active", ASCENDING), + ("classifier_metadata.province", ASCENDING), + ("_id", DESCENDING), + ], + name="is_active_province_ci_-_id", + collation=LOCATION_COLLATION, + ), ] diff --git a/backend/app/services/bm25_scoring/bm25library.py b/backend/app/services/bm25_scoring/bm25library.py index 008ed71..1b51552 100644 --- a/backend/app/services/bm25_scoring/bm25library.py +++ b/backend/app/services/bm25_scoring/bm25library.py @@ -137,6 +137,18 @@ def avg_doc_length(docs: List[List[str]]) -> float: return float(sum(len(d) for d in docs)) / len(docs) +def _sanitize_corpus_for_bm25(corpus: List[List[str]]) -> List[List[str]]: + """Avoid rank_bm25's ZeroDivisionError when every doc is empty. + + rank_bm25._calc_idf divides by len(self.idf); if all docs tokenize to [] the + IDF dict ends up empty and BM25Okapi construction throws. Substitute a single + sentinel token in any empty doc so the corpus has a nonzero IDF — downstream + queries score these docs as 0 against any real query token, preserving + behavior. + """ + return [doc if doc else ["__empty_doc__"] for doc in corpus] + + def build_okapi_indexes( skills_corpus: List[List[str]], full_corpus: List[List[str]], @@ -147,8 +159,8 @@ def build_okapi_indexes( """Return ``(skills_bm25, full_bm25)`` ``BM25Okapi`` instances.""" _require_rank_bm25() return ( - BM25Okapi(skills_corpus, k1=k1, b=b), - BM25Okapi(full_corpus, k1=k1, b=b), + BM25Okapi(_sanitize_corpus_for_bm25(skills_corpus), k1=k1, b=b), + BM25Okapi(_sanitize_corpus_for_bm25(full_corpus), k1=k1, b=b), ) diff --git a/backend/app/services/match_concat_gemini_ce_service.py b/backend/app/services/match_concat_gemini_ce_service.py index 3187952..bf96a85 100644 --- a/backend/app/services/match_concat_gemini_ce_service.py +++ b/backend/app/services/match_concat_gemini_ce_service.py @@ -15,6 +15,9 @@ import logging import os import threading +import time + +from app.match_timing_log import log_match_step from typing import Any, Dict, List, Optional import numpy as np @@ -272,6 +275,9 @@ def run_match_concat_gemini_ce( rt = max(1, int(retrieve_top_k)) fk = max(1, int(final_top_k)) + t_engine_start = time.perf_counter() + + t0 = time.perf_counter() job_rows: List[Dict[str, Any]] = [] vectors: List[np.ndarray] = [] for j in jobs: @@ -290,6 +296,7 @@ def run_match_concat_gemini_ce( # Post-secondary education gate: aligned with job_rows, used to skip candidates per user. job_requires_ps = [job_requires_post_secondary(j) for j in job_rows] + prep_job_vectors_ms = (time.perf_counter() - t0) * 1000.0 if not job_rows: empty_summary = { @@ -319,10 +326,13 @@ def run_match_concat_gemini_ce( for u in users ] + t0 = time.perf_counter() j_mat = np.stack(vectors, axis=0).astype(np.float64) j_norm = l2_normalize_rows(j_mat.astype(np.float32)).astype(np.float64) jid_list = [str(j.get("uuid") or "") for j in job_rows] + job_mat_build_ms = (time.perf_counter() - t0) * 1000.0 + t0 = time.perf_counter() if user_unit_vectors is not None: u_norm = np.asarray(user_unit_vectors, dtype=np.float64) if ( @@ -333,14 +343,22 @@ def run_match_concat_gemini_ce( raise RuntimeError( f"user_unit_vectors shape {u_norm.shape} != ({len(users)}, {EMBEDDING_DIM})" ) + gemini_embed_ms = 0.0 # supplied by caller else: u_norm = embed_user_unit_vectors(users) + gemini_embed_ms = (time.perf_counter() - t0) * 1000.0 matcher = _get_matcher() reranker = _get_reranker() + cosine_shortlist_ms = 0.0 + score_pair_ms = 0.0 + ce_rerank_ms = 0.0 + format_user_recs_ms = 0.0 + out_results: List[Dict[str, Any]] = [] for i, user in enumerate(users): + t_user = time.perf_counter() sim_row = (u_norm[i : i + 1] @ j_norm.T).reshape(-1) order = _sorted_indices_desc(sim_row) user_no_ps = user_lacks_post_secondary(user) @@ -374,7 +392,9 @@ def run_match_concat_gemini_ce( for r_i, row in enumerate(cosine_recs, start=1): row["rank"] = r_i + cosine_shortlist_ms += (time.perf_counter() - t_user) * 1000.0 + t_ce = time.perf_counter() labels = user_skill_labels_for_concat(user) reranked = rerank_cosine_recommendations( labels, @@ -382,7 +402,9 @@ def run_match_concat_gemini_ce( reranker=reranker, final_top_k=fk, ) + ce_rerank_ms += (time.perf_counter() - t_ce) * 1000.0 + t_fmt = time.perf_counter() recs: List[Dict[str, Any]] = [] for row in reranked: recs.append( @@ -429,5 +451,26 @@ def run_match_concat_gemini_ce( "config_summary": cfg, } ) + format_user_recs_ms += (time.perf_counter() - t_fmt) * 1000.0 + + engine_total_ms = (time.perf_counter() - t_engine_start) * 1000.0 + try: + log_match_step( + "engine concat_gemini_ce", + "sub-stages (sum across all users)", + n_users=len(users), + n_jobs_with_emb=n_with_emb, + retrieve_top_k=rt, + final_top_k=fk, + prep_job_vectors_ms=prep_job_vectors_ms, + job_mat_build_ms=job_mat_build_ms, + gemini_embed_ms=gemini_embed_ms, + cosine_shortlist_ms=cosine_shortlist_ms, + ce_rerank_ms=ce_rerank_ms, + format_user_recs_ms=format_user_recs_ms, + engine_total_ms=engine_total_ms, + ) + except Exception: + pass return out_results