"""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()] # 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", "indeed"} # 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": ("misceres~indeed-scraper", lambda p: A.build_indeed_input(p, max_items=get_settings().INDEED_MAX_JOBS), N.indeed_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 = [sj for it in items if (sj := norm(it))] 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] results = await asyncio.gather(*(_fetch_board(k, prefs, cache_mode) 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)