Blame for scripts/run_ranking.py

"""End-to-end ranking pipeline.

This is THE reproduce command listed in submission_metadata.yaml.

Usage:
    python scripts/run_ranking.py \\
        --candidates "<path-to-candidates.jsonl>" \\
        --out        "<path-to-submission.csv>"

It streams the candidate pool, builds BM25 + (optional) dense indices,
fuses them with RRF, computes the full feature set on the shortlist, and
applies the trained ranker (or the deterministic fallback). The result
is a 100-row CSV that the validator checks for compliance.

If a precomputed artifact is missing, the pipeline rebuilds it. All
artifacts live under `artifacts/`.
"""
from __future__ import annotations

import argparse
import json
import os
import pickle
import re
import sys
import time
from pathlib import Path

import numpy as np
import pandas as pd

PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

# Determinism
os.environ.setdefault("PYTHONHASHSEED", "0")
os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")

from redrob import config
from redrob.blueprint import build_blueprint, save_blueprint
from redrob.data import load_or_build_parquet
from redrob.retrieval.bm25 import build_bm25, bm25_search
from redrob.retrieval.dense import (
    build_dense_index, dense_search, dense_model_name,
)
from redrob.retrieval.rrf import rrf_fuse
from redrob.rank.predict import rank_shortlist
from redrob.submit.write import write_submission


def _save_id_map(path: Path, ids: list) -> None:
    with open(path, "w", encoding="utf-8") as f:
        json.dump(ids, f)


def _load_id_map(path: Path) -> list:
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def run(args) -> None:
    t_start = time.time()
    print(f"[{time.time()-t_start:5.1f}s] starting pipeline")

    # 1) Blueprint
    save_blueprint()
    bp = build_blueprint()
    queries = bp["query_terms"]
    print(f"[{time.time()-t_start:5.1f}s] blueprint saved ({len(config.CORE_COMPETENCIES)} core skills)")

    # 2) Load or build the canonicalised parquet
    df = load_or_build_parquet(args.candidates)
    print(f"[{time.time()-t_start:5.1f}s] canonicalised: {len(df):,} candidates")

    # 3) BM25 sparse retrieval
    bm25, _ = build_bm25(force=False)
    sparse_hits = bm25_search(queries, top_k=config.BM25_TOP_K, bm25_obj=(bm25, _))
    sparse_rankings = list(sparse_hits.values())
    print(f"[{time.time()-t_start:5.1f}s] BM25 ready (top-{config.BM25_TOP_K} per query)")

    # 4) Dense retrieval (or fallback)
    dense_rankings: list = []
    if not args.no_dense:
        try:
            t_dense = time.time()
            print(f"[{time.time()-t_start:5.1f}s] building dense index (model={dense_model_name()}) ...")
            # If the cache exists, use it (fast). Otherwise encode and cache.
            emb, ids = build_dense_index(force=False)
            dense_hits = dense_search(queries, top_k=config.DENSE_TOP_K, matrix=(emb, ids))
            dense_rankings = list(dense_hits.values())
            print(f"[{time.time()-t_start:5.1f}s] dense ready in {time.time()-t_dense:.1f}s")
        except Exception as e:  # noqa: BLE001
            print(f"[{time.time()-t_start:5.1f}s] dense unavailable ({e}); using BM25-only")
            dense_rankings = []
    else:
        print(f"[{time.time()-t_start:5.1f}s] dense path disabled by --no_dense")

    # 5) Structured high-recall gate as SOFT score (saturating)
    gate_terms_lc = {t.lower() for t in config.STRUCTURED_GATE_TERMS}
    df_reset = df.reset_index(drop=True)

    def _gate_text_score(text):
        if not text:
            return 0
        tl = text.lower()
        n_hits = sum(1 for g in gate_terms_lc if g in tl)
        return float(min(1.0, n_hits / 5.0))

    def _gate_skill_score(skills):
        if skills is None:
            return 0.0
        if hasattr(skills, "tolist"):
            skills = skills.tolist()
        n = sum(1 for s in (skills or []) if (s or "").lower() in gate_terms_lc)
        return float(min(1.0, n / 3.0))

    text_scores = df_reset["text_corpus"].fillna("").apply(_gate_text_score).to_numpy()
    skill_scores = df_reset["skill_names"].apply(_gate_skill_score).to_numpy()
    gate_score = np.maximum(text_scores, skill_scores)
    structured_hit = (gate_score > 0.0).astype(np.float32)
    print(f"[{time.time()-t_start:5.1f}s] structured gate: median score {np.median(gate_score):.2f}, "
          f"hits {int(structured_hit.sum()):,}/{len(df):,}")

    # 6) RRF over all channels with per-channel weights
    # BM25 (reliable) > structured-gate (binary OK) > dense (noisy in this domain)
    rankings = list(sparse_rankings) + list(dense_rankings)
    if dense_rankings:
        channel_weights = [1.2] * len(sparse_rankings) + [0.8] * len(dense_rankings)
    else:
        channel_weights = [1.0] * len(sparse_rankings)
    fused = rrf_fuse(rankings, k=config.RRF_K, top_n=config.SHORTLIST_N,
                     channel_weights=channel_weights)
    shortlist_ids = [cid for cid, _score, _r in fused]
    shortlist_set = set(shortlist_ids)
    if args.verbose:
        print(f"[{time.time()-t_start:5.1f}s] RRF shortlist: {len(shortlist_ids):,}")

    # Add top structured-gate candidates not already in shortlist (sorted by gate_score)
    struct_candidates = df_reset.loc[gate_score > 0.0].copy()
    struct_candidates = struct_candidates.assign(_g=gate_score[gate_score > 0.0])
    struct_candidates = struct_candidates.sort_values("_g", ascending=False)
    added = 0
    for cid in struct_candidates["candidate_id"].tolist():
        if cid not in shortlist_set:
            shortlist_ids.append(cid)
            shortlist_set.add(cid)
            added += 1
            if added >= args.structured_top_up:
                break
    if args.verbose:
        print(f"[{time.time()-t_start:5.1f}s] +{added:,} from structured gate (total shortlist: {len(shortlist_ids):,})")
    struct_ids = df_reset.loc[gate_score > 0.0, "candidate_id"].tolist()
    gate_score_map = {cid: float(s) for cid, s in zip(df_reset["candidate_id"].tolist(), gate_score)}

    # ----------------------------------------------------------------------
    # Plain-language Tier-5 retrieval top-up (added by super-smart-agent review).
    # Some excellent candidates don't say RAG/Pinecone in skills, but their
    # career DESCRIPTIONS mention shipping ranking/search/recsys systems.
    # Optimised: pre-compile regexes, use only the first N rows for speed,
    # and score on text_corpus (already-built column) instead of building
    # new strings per candidate.
    # ----------------------------------------------------------------------
    _SHIP_KW = re.compile(
        r"\b(built|shipped|launched|scaled|deployed|owned|led|designed|"
        r"implemented|migrated|productionized)\b",
        re.IGNORECASE,
    )
    _RETRIEVAL_KW = re.compile(
        r"\b(matching|personalization|personalisation|marketplace|"
        r"recruiter engagement|click.?through|engagement|"
        r"index|refresh|search system|relevance|re-ranking|reranking|"
        r"semantic|embeddings|indexing|indexed|ranker)\b",
        re.IGNORECASE,
    )
    _PRODUCT_TITLE_KW = re.compile(
        r"\b(ai engineer|machine learning|ml engineer|"
        r"applied scientist|research engineer|search engineer|"
        r"ranking engineer|recommendation|nlp engineer|data scientist)\b",
        re.IGNORECASE,
    )
    _PLAIN_LANG_TOPUP = 200  # how many extra candidates to add

    plain_lang_candidates = []
    _current_titles = df_reset["current_title"].fillna("").str.lower().tolist()
    _text_corpuses = df_reset["text_corpus"].fillna("").str[:1000].tolist()  # cap length
    _cids_all = df_reset["candidate_id"].tolist()
    n_total = len(_cids_all)
    # Scan only candidates not already in shortlist (faster loop)
    scan_indices = [i for i, cid in enumerate(_cids_all) if cid not in shortlist_set]

    for i in scan_indices:
        # Quick title pre-filter
        if not _PRODUCT_TITLE_KW.search(_current_titles[i] or ""):
            continue
        # Use the pre-built text_corpus (head 1000 chars) — fast substring match
        text = (_text_corpuses[i] or "").lower()
        if not text:
            continue
        n_ship = len(_SHIP_KW.findall(text))
        if n_ship < 1:
            continue
        n_ret = len(_RETRIEVAL_KW.findall(text))
        if n_ret >= 1:
            plain_lang_candidates.append((n_ship + n_ret, _cids_all[i]))

    plain_lang_candidates.sort(reverse=True)
    added = 0
    for _, cid in plain_lang_candidates[:_PLAIN_LANG_TOPUP]:
        if cid not in shortlist_set:
            shortlist_ids.append(cid)
            shortlist_set.add(cid)
            added += 1
    if args.verbose:
        print(f"[{time.time()-t_start:5.1f}s] +{added:,} from plain-language top-up "
              f"(of {len(plain_lang_candidates):,} eligible; scanned {len(scan_indices):,})")

    # 7) Build the shortlist dataframe
    shortlist_df = df_reset[df_reset["candidate_id"].isin(shortlist_set)].copy()

    # 8) Per-channel scores for the shortlist (sparse/dense/rrf)
    sparse_score_by_id: dict = {}
    for ranking in sparse_rankings:
        for cid, sc in ranking:
            if cid in shortlist_set:
                sparse_score_by_id[cid] = max(sparse_score_by_id.get(cid, 0.0), sc)
    dense_score_by_id: dict = {}
    for ranking in dense_rankings:
        for cid, sc in ranking:
            if cid in shortlist_set:
                dense_score_by_id[cid] = max(dense_score_by_id.get(cid, 0.0), sc)
    rrf_score_by_id: dict = {cid: s for cid, s, _r in fused}

    shortlist_df["sparse_score"] = shortlist_df["candidate_id"].map(lambda c: sparse_score_by_id.get(c, 0.0))
    shortlist_df["dense_score"] = shortlist_df["candidate_id"].map(lambda c: dense_score_by_id.get(c, 0.0))
    shortlist_df["rrf_score"] = shortlist_df["candidate_id"].map(lambda c: rrf_score_by_id.get(c, 0.0))
    shortlist_df["structured_hit"] = shortlist_df["candidate_id"].map(lambda c: gate_score_map.get(c, 0.0))

    print(f"[{time.time()-t_start:5.1f}s] computing features for {len(shortlist_df):,} shortlist candidates")

    # 9) Rank (uses LightGBM if model exists, else deterministic fallback)
    use_lgbm = not args.no_lgbm
    if not use_lgbm:
        print(f"[{time.time()-t_start:5.1f}s] ranker: deterministic fallback (--no_lgbm)")
    else:
        print(f"[{time.time()-t_start:5.1f}s] ranker: LightGBM (auto-trains if missing)")
    ranked = rank_shortlist(shortlist_df, use_lgbm=use_lgbm)

    # ----------------------------------------------------------------------
    # v11: Final logistics eligibility filter.
    # JD: "Location: Pune/Noida-preferred but flexible. Outside India:
    # case-by-case, but we don't sponsor work visas."
    # A non-India candidate who is not willing to relocate is, for this
    # company, functionally unreachable. Remove them from the ranked
    # shortlist BEFORE selecting top-100, so they are replaced by the
    # next eligible candidate from below. This does NOT change top-10
    # or top-50 (all six removed are in ranks 51–100) and preserves any
    # non-India candidate who IS willing to relocate.
    # ----------------------------------------------------------------------
    country_arr = ranked["country"].fillna("").astype(str).to_numpy()
    reloc_arr = ranked["willing_to_relocate"].fillna(0).astype(float).to_numpy()
    eligible_mask = ~((country_arr != "India") & (reloc_arr == 0))
    n_removed = int((~eligible_mask).sum())
    print(f"[{time.time()-t_start:5.1f}s] v11 eligibility filter: kept "
          f"{int(eligible_mask.sum()):,}, removed {n_removed} non-India + not-willing-to-relocate")

    # 10) Top-100 (selected from the eligible subset)
    top = ranked[eligible_mask].head(100).copy()
    print(f"[{time.time()-t_start:5.1f}s] top-100 selected")

    # 11) Honeypot self-check
    n_honeypot = int((top["honeypot_penalty"] >= config.HONEYPOT_HARD_EXCLUDE).sum())
    n_low_title = int((top["title_fit_blend"] < 0.10).sum())
    n_services_top50 = int(top.head(50)["is_services_company"].astype(int).sum()) if "is_services_company" in top.columns else 0
    print(f"[{time.time()-t_start:5.1f}s] self-check: honeypot_hard={n_honeypot}, low_title_top100={n_low_title}, services_top50={n_services_top50}")

    # 12) Write CSV
    out = write_submission(top, out_path=args.out, validate=True)
    print(f"[{time.time()-t_start:5.1f}s] submission written: {out}")
    print(f"[{time.time()-t_start:5.1f}s] DONE")


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument(
        "--candidates",
        default=str(config.DEFAULT_CANDIDATES_JSONL),
        help="Path to candidates.jsonl",
    )
    ap.add_argument(
        "--out",
        default=str(config.DEFAULT_OUTPUT_CSV),
        help="Path to submission CSV",
    )
    ap.add_argument(
        "--no_dense",
        action="store_true",
        help="Disable sentence-transformers (use BM25-only retrieval)",
    )
    ap.add_argument(
        "--no_lgbm",
        action="store_true",
        help="Use the deterministic scorer instead of the LightGBM ranker",
    )
    ap.add_argument(
        "--structured_top_up",
        type=int,
        default=2000,
        help="How many structured-gate candidates to add beyond the RRF shortlist",
    )
    ap.add_argument("--verbose", action="store_true")
    args = ap.parse_args()
    run(args)


if __name__ == "__main__":
    main()