162 lines
8.2 KiB
Python
162 lines
8.2 KiB
Python
"""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,
|
|
exhausted: bool = False) -> 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).
|
|
|
|
EXHAUSTION guard: when the query is tapped out (a recent refill added almost no NEW jobs), a blocking
|
|
sweep would just re-buy duplicates — so as long as the pool isn't EMPTY, serve what's there instead of
|
|
block-fetching. This is what makes a high floor safe on niche queries. A forced-fresh always sweeps."""
|
|
if force_fresh:
|
|
return "sweep"
|
|
if exhausted and n_fresh > 0:
|
|
return "pool" # tapped out but we have something → serve, don't re-buy dupes
|
|
if 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.
|
|
# NO board paginates: curious_coder only honors start=0 (start>0 → empty), and the rest are date feeds.
|
|
# Freshness comes from the boards' "latest by date" + the per-user seen-net — NOT page cursors.
|
|
PAGINATING_BOARDS: set[str] = set()
|
|
|
|
# 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": ("curious_coder~linkedin-jobs-scraper", # structured industry+jobFunction+applicants
|
|
# NO cursor — actor only honors start=0; freshness via date sort + the seen-net.
|
|
lambda p: A.build_linkedin_cc_input(p, count=get_settings().LINKEDIN_MAX_JOBS),
|
|
N.linkedin_cc_to_scoutjob),
|
|
"linkedin_v1": ("harvestapi~linkedin-job-search", # registered fallback (jobTitles input)
|
|
lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS),
|
|
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),
|
|
"timesjobs": ("shahidirfan~Timesjobs-Scraper", # India full-signal; no page param → seen-net dedups.
|
|
lambda p: A.build_timesjobs_input(p, results_wanted=get_settings().TIMESJOBS_MAX_JOBS),
|
|
N.timesjobs_to_scoutjob),
|
|
}
|
|
|
|
|
|
async def _fetch_board(key: str, prefs: dict, cache_mode: str | None = None):
|
|
actor, build, norm = BOARDS[key]
|
|
inp = build(prefs)
|
|
items = await run_actor(actor, inp, cache_mode=cache_mode)
|
|
if not items:
|
|
# Scrape-based boards (Naukri-feed, curious_coder LinkedIn) intermittently return 0 under the
|
|
# site's anti-scrape rate-limiting, then recover. One fresh retry rescues the transient case
|
|
# (a genuinely-empty query just returns 0 again — cheap). RCA'd: same input flakes 0 then 15.
|
|
logger.info("board %s returned 0 — retrying once (transient scrape flake)", key)
|
|
items = await run_actor(actor, inp, cache_mode="refresh")
|
|
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] = {}
|
|
failed_boards = 0
|
|
for r in results:
|
|
if isinstance(r, Exception):
|
|
failed_boards += 1
|
|
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,
|
|
"all_boards_failed": bool(enabled) and failed_boards == len(enabled),
|
|
"failed_boards": failed_boards,
|
|
"enabled_boards": len(enabled),
|
|
}
|
|
|
|
|
|
def has_mvq(prefs: dict) -> bool:
|
|
return C.has_mvq(prefs)
|