matchmaking-v2/app/engine/search.py

"""On-demand multi-board sweep: ScoutPrefs → all boards (parallel) → merge + dedup → ScoutJob[].

This is the retrieve layer (final shape). Slice-1a has NO engine — ranking is a placeholder
spread; the §3 cascade slots in here later without changing the output contract.
Degrade-don't-break: a board that errors/needs setup is skipped; the others still return.
"""
from __future__ import annotations

import asyncio
import logging

from app.config import get_settings
from app.engine.apify_client import run_actor
from app.engine.board_adapters import adapters as A
from app.engine.board_adapters import coerce as C
from app.engine import normalize as N

logger = logging.getLogger(__name__)

def _ats_companies() -> list[dict]:
    return [{"company": c.strip()} for c in get_settings().ATS_COMPANIES.split(",") if c.strip()]


def pool_decision(n_fresh: int, force_fresh: bool, floor: int, reorder: int) -> str:
    """The warm-pool gate. Given how many FRESH, UNSEEN jobs are banked, decide where candidates come from:
      'sweep'       — blocking Apify fetch (pool below the hard floor, or a forced fresh search)
      'pool+refill' — serve from the pool NOW + top it up in the BACKGROUND (dipped below the reorder mark)
      'pool'        — serve from the pool, no fetch at all (pool healthy)
    Reorder EARLY and floor HIGH so the pool never runs thin (floor ≤ reorder)."""
    if force_fresh or n_fresh < floor:
        return "sweep"
    if n_fresh < reorder:
        return "pool+refill"
    return "pool"


# Boards whose actor takes a real page/offset → they get a per-board cursor that advances each run.
# Everyone else (Naukri/Foundit feeds — no page param) relies purely on the per-user seen-net; we do
# NOT track a meaningless cursor for them.
PAGINATING_BOARDS = {"linkedin"}   # valig-indeed uses `limit` (no page) → seen-net, not a cursor

# board key → (actor_id, build_input, normalizer). "Balanced" India stack = naukri+foundit+linkedin,
# all city-filtered at the board. The rest stay registered (code ready) but off unless enabled.
BOARDS = {
    "naukri": ("blackfalcondata~naukri-jobs-feed",
               # no page/offset param → per-user seen-net handles dedup (NOT actor-side incremental).
               lambda p: A.build_naukri_feed_input(p, max_jobs=get_settings().NAUKRI_MAX_JOBS),
               N.naukri_feed_to_scoutjob),
    "foundit": ("shahidirfan~Foundit-Jobs-Scraper",
                lambda p: A.build_foundit_input(p, results_wanted=get_settings().FOUNDIT_MAX_JOBS),
                N.foundit_to_scoutjob),
    "linkedin": ("harvestapi~linkedin-job-search",
                 # cursor stores the LAST page fetched (0 = none yet) → next page = last + 1.
                 lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS,
                                                  page=int((p.get("_cursors") or {}).get("linkedin", 0)) + 1),
                 N.linkedin_to_scoutjob),
    # ── registered fallbacks / other lanes (off unless added to BOARDS_ENABLED) ──
    "ats": ("bovi~greenhouse-lever-ashby-job-scraper",
            lambda p: A.build_ats_input(p, recall=True, companies=_ats_companies(),
                                        max_per_company=get_settings().ATS_MAX_PER_COMPANY),
            N.ats_to_scoutjob),
    "naukri_v1": ("muhammetakkurtt~naukri-job-scraper",
                  lambda p: A.build_naukri_input(p, recall=True, max_jobs=get_settings().NAUKRI_MAX_JOBS),
                  N.naukri_to_scoutjob),
    "indeed": ("valig~indeed-jobs-scraper",   # ~7s vs misceres ~52s; richer fields + offsite jobUrl
               lambda p: A.build_indeed_input(p, max_items=get_settings().INDEED_MAX_JOBS),
               N.indeed_to_scoutjob),
    "wellfound": ("blackfalcondata~wellfound-scraper",  # registered, OFF — US-startup-heavy, not the India fit
                  lambda p: A.build_wellfound_input(p, max_results=get_settings().WELLFOUND_MAX_JOBS),
                  N.wellfound_to_scoutjob),
    "workindia": ("shahidirfan~workindia-jobs-scraper",
                  # India blue/grey-collar non-tech; no page param → per-user seen-net dedups.
                  lambda p: A.build_workindia_input(p, results_wanted=get_settings().WORKINDIA_MAX_JOBS),
                  N.workindia_to_scoutjob),
}


async def _fetch_board(key: str, prefs: dict, cache_mode: str | None = None):
    actor, build, norm = BOARDS[key]
    items = await run_actor(actor, build(prefs), cache_mode=cache_mode)
    jobs = []
    for it in items:
        sj = norm(it)
        if not sj:
            continue
        if not sj.get("seniority_level"):                 # board omitted it → infer from the title
            sj["seniority_level"] = N._seniority_from_title(sj.get("title"))
        jobs.append(sj)
    return key, jobs


async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict:
    # fresh=True forces a live fetch + cache overwrite (dev "fresh search" toggle);
    # otherwise honor the configured cache mode (dev replays cached results for $0).
    cache_mode = "refresh" if fresh else None
    enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip() in BOARDS]
    cap = get_settings().BOARD_TIMEOUT_S

    async def _timed(k: str):
        # A single slow/hung board must NOT hold the whole search hostage — cap it and degrade.
        try:
            return await asyncio.wait_for(_fetch_board(k, prefs, cache_mode), timeout=cap)
        except asyncio.TimeoutError:
            logger.warning("board %s exceeded %ss — skipped (search stays fast)", k, cap)
            return k, []

    results = await asyncio.gather(*(_timed(k) for k in enabled), return_exceptions=True)

    merged: list[dict] = []
    seen: set[str] = set()
    sources: dict[str, int] = {}
    for r in results:
        if isinstance(r, Exception):
            logger.warning("board fetch failed: %s", r)
            continue
        key, jobs = r
        sources[key] = len(jobs)
        for j in jobs:
            dk = (j.get("apply_url") or "").strip() or f"{j['organization']}|{j['title']}|{j.get('location_city')}".lower()
            if dk in seen:
                continue
            seen.add(dk)
            merged.append(j)

    # No ranking here — the engine (app/engine/rank.py) scores + selects after the sweep.
    return {"opportunities": merged, "sources": sources}


def has_mvq(prefs: dict) -> bool:
    return C.has_mvq(prefs)