diff --git a/app/agent/session.py b/app/agent/session.py index 59995de..3c681e5 100644 --- a/app/agent/session.py +++ b/app/agent/session.py @@ -46,6 +46,11 @@ class MatchmakingAgentSession: "tailor_resume": self.handle_tailor_resume, "submit_application": self.handle_submit_application, "get_apply_proof": self.handle_get_apply_proof, + # per-opportunity state — persists across deck refreshes (applied / dismissed / saved docs) + "dismiss_opportunity": self.handle_dismiss_opportunity, + "restore_opportunity": self.handle_restore_opportunity, + "mark_applied": self.handle_mark_applied, + "save_apply_docs": self.handle_save_apply_docs, } handler = handlers.get(action) if not handler: @@ -86,7 +91,19 @@ class MatchmakingAgentSession: await self.push("agent_thinking", message="Fetching fresh roles across job boards…" if fresh else "Scanning live roles across job boards…") + # Search cursor: re-running the SAME query advances the page (LinkedIn) + reuses the Naukri + # incremental stateKey, so each run fetches GENUINELY NEW jobs — not the same set re-charged. + from app.db import repo as _repo + sig = _repo.search_signature(prefs) + page = await _repo.get_search_cursor(self.user_id, sig) + prefs["_page"] = page + # colon-free key (state-store record id charset is alphanumeric/_-.) + prefs["_naukri_state"] = f"{self.user_id}_{sig}".replace(":", "_") sweep = await search.run_sweep(prefs, fresh=fresh) + # Safety net: drop anything already shown / applied / dismissed (covers boards that can't + # paginate, so the same posting never reappears even if a board re-returns it). + seen = await _repo.get_seen_ids(self.user_id) + candidates = [o for o in sweep["opportunities"] if o.get("id") not in seen] # Engine: assemble FULL context (profile + prefs). Cheap rankers SIFT 90 → top ~18 (white-box ‖ # embedding vibe); then Opus reads only those and curates the honest shortlist + report cards. from app.engine import curate as _curate @@ -96,7 +113,7 @@ class MatchmakingAgentSession: await self.push("agent_thinking", message="Scoring roles against your profile…") # Run the blocking LLM work (embeddings + Opus) OFF the event loop, or the long sync OpenAI # call freezes the loop and the Redis response can't publish (→ the loader hangs forever). - top, dbg = await asyncio.to_thread(_sift.sift, prefs, user_context, sweep["opportunities"]) + top, dbg = await asyncio.to_thread(_sift.sift, prefs, user_context, candidates) await self.push("agent_thinking", message="Scout is reading your top roles…") curated = await asyncio.to_thread(_curate.curate, prefs, user_context, top) # Opus's call is final if curated is None: # safety net: sift + templated cards @@ -107,11 +124,18 @@ class MatchmakingAgentSession: result = {"opportunities": curated, "sources": sweep["sources"], "shortlisted": len(curated), "scanned": dbg["scored"], "engine": engine, "has_profile": bool(user_context and user_context.get("skills"))} + # Persist the RAW deck + ADVANCE the cursor (page+1) so the next run of this query fetches new + # jobs. Per-opportunity state (applied/dismissed/saved docs) lives separately, applied on read. + clean_prefs = {k: v for k, v in prefs.items() + if k not in ("user_context", "_fresh", "_page", "_naukri_state")} + await _repo.save_feed(self.user_id, clean_prefs, result, query_sig=sig, cursor_page=page + 1) + # Mark the shown deck as SEEN so it never reappears in a future run. + await _repo.mark_seen(self.user_id, [o.get("id") for o in (result["opportunities"] or []) if o.get("id")]) + # Hand the UI a state-aware view: dismissed dropped, applied + saved-doc ids annotated. + states = await _repo.get_opportunity_states(self.user_id) + result["opportunities"] = _repo.annotate_and_filter(result["opportunities"], states) + result["shortlisted"] = len(result["opportunities"]) await self.push("agent_data", action="search_complete", data=result) - # Persist the feed so it survives navigation/refresh (replayed by handle_get_feed). - from app.db.repo import save_feed - clean_prefs = {k: v for k, v in prefs.items() if k not in ("user_context", "_fresh")} - await save_feed(self.user_id, clean_prefs, result) async def handle_suggest_bubbles(self, params: dict): """Fine-tune bubbling — stage-aware (broad→narrow→role) bubbles from the picks so far.""" @@ -131,3 +155,35 @@ class MatchmakingAgentSession: async def handle_get_apply_proof(self, params: dict): await self._stub("apply_proof", "return captured confirmation screenshots") + + # ── per-opportunity state (persists across deck refreshes) ── + async def handle_dismiss_opportunity(self, params: dict): + """Mark a posting stale/dismissed → it never resurfaces in the deck.""" + from app.db.repo import set_opportunity_state + await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status="dismissed") + await self.push("agent_data", action="opportunity_state_saved", + data={"opportunity_id": params.get("opportunity_id"), "status": "dismissed"}) + + async def handle_restore_opportunity(self, params: dict): + """Undo a dismiss/pass → clear the status so the role returns to the deck.""" + from app.db.repo import set_opportunity_state + await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status=None) + await self.push("agent_data", action="opportunity_state_saved", + data={"opportunity_id": params.get("opportunity_id"), "status": None}) + + async def handle_mark_applied(self, params: dict): + """Record that the user applied to a role (sticks across reloads).""" + from app.db.repo import set_opportunity_state + await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status="applied") + await self.push("agent_data", action="opportunity_state_saved", + data={"opportunity_id": params.get("opportunity_id"), "status": "applied"}) + + async def handle_save_apply_docs(self, params: dict): + """Remember the resume-builder document ids generated for a role so we never re-pay to + re-tailor / re-generate them. Only the provided ids are written.""" + from app.db.repo import set_opportunity_state + fields = {k: params[k] for k in ("tailored_resume_id", "tailored_version_id", "cover_letter_id") + if params.get(k)} + await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), **fields) + await self.push("agent_data", action="opportunity_state_saved", + data={"opportunity_id": params.get("opportunity_id"), **fields}) diff --git a/app/db/models.py b/app/db/models.py index 6415339..80fa04a 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime -from sqlalchemy import JSON, DateTime, Integer, String, func +from sqlalchemy import JSON, Boolean, DateTime, Integer, String, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column @@ -22,6 +22,28 @@ class UserFeed(Base): sources: Mapped[dict] = mapped_column(JSON, default=dict) # per-board counts engine: Mapped[str] = mapped_column(String, default="") # "opus" | "fallback:…" scanned: Mapped[int] = mapped_column(Integer, default=0) + # Search cursor — advances per re-run of the SAME query so each run fetches NEW jobs (LinkedIn + # page++, Naukri incremental stateKey). Resets when the query signature changes. + query_sig: Mapped[str] = mapped_column(String, default="") + cursor_page: Mapped[int] = mapped_column(Integer, default=1) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + +class OpportunityState(Base): + """Per-(user, opportunity) state that SURVIVES deck refreshes. Persists three things: + `applied`/`dismissed` status, and the resume-builder document ids already generated for the + role (so we never re-tailor / re-generate a paid document twice). Keyed by (user, opportunity).""" + __tablename__ = "opportunity_state" + + user_id: Mapped[str] = mapped_column(String, primary_key=True) + opportunity_id: Mapped[str] = mapped_column(String, primary_key=True) + status: Mapped[str | None] = mapped_column(String, nullable=True) # 'applied' | 'dismissed' | None + tailored_resume_id: Mapped[str | None] = mapped_column(String, nullable=True) # resume-builder resume id + tailored_version_id: Mapped[str | None] = mapped_column(String, nullable=True) # the tailored version + cover_letter_id: Mapped[str | None] = mapped_column(String, nullable=True) # the generated cover letter + seen: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # shown in a deck → exclude from new runs updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now() ) diff --git a/app/db/repo.py b/app/db/repo.py index 3b36eb1..aadceec 100644 --- a/app/db/repo.py +++ b/app/db/repo.py @@ -2,16 +2,32 @@ when the DB is off or unreachable (persistence is optional; the engine must never crash on it).""" from __future__ import annotations +import hashlib +import json import logging from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert as pg_insert -from app.db.models import Base, UserFeed +from app.db.models import Base, OpportunityState, UserFeed from app.db.session import engine, session_factory logger = logging.getLogger(__name__) +# fields a caller may set on an opportunity's state (status + the generated document ids) +_STATE_FIELDS = ("status", "tailored_resume_id", "tailored_version_id", "cover_letter_id") + +# the search-defining pref fields — same values → same signature → cursor advances (new jobs); +# change any → new signature → cursor resets to page 1. +_SIG_FIELDS = ("title", "role", "location", "industry", "experience", "years", "work_mode", "remote") + + +def search_signature(prefs: dict) -> str: + """Stable short hash of the search-defining prefs (ignores transient flags / user_context).""" + payload = {k: prefs.get(k) for k in _SIG_FIELDS} + blob = json.dumps(payload, sort_keys=True, default=str) + return hashlib.sha1(blob.encode()).hexdigest()[:16] + async def init_db() -> None: """Create tables if the DB is configured (best-effort; logged, never fatal).""" @@ -27,8 +43,8 @@ async def init_db() -> None: logger.warning("feed store init failed (persistence off): %s", e) -async def save_feed(user_id: str, prefs: dict, result: dict) -> None: - """Upsert the user's last completed search. No-op if the DB is off.""" +async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str = "", cursor_page: int = 1) -> None: + """Upsert the user's last completed search + the advanced search cursor. No-op if the DB is off.""" factory = session_factory() if factory is None or not user_id: return @@ -39,13 +55,15 @@ async def save_feed(user_id: str, prefs: dict, result: dict) -> None: "sources": result.get("sources") or {}, "engine": result.get("engine") or "", "scanned": int(result.get("scanned") or 0), + "query_sig": query_sig, + "cursor_page": cursor_page, } try: async with factory() as s: stmt = pg_insert(UserFeed).values(**row) stmt = stmt.on_conflict_do_update( index_elements=[UserFeed.user_id], - set_={k: row[k] for k in ("prefs", "opportunities", "sources", "engine", "scanned")}, + set_={k: row[k] for k in ("prefs", "opportunities", "sources", "engine", "scanned", "query_sig", "cursor_page")}, ) await s.execute(stmt) await s.commit() @@ -66,13 +84,135 @@ async def get_feed(user_id: str) -> dict | None: return None if not row: return None + states = await get_opportunity_states(user_id) + opps = annotate_and_filter(row.opportunities or [], states) return { - "opportunities": row.opportunities or [], + "opportunities": opps, "sources": row.sources or {}, "prefs": row.prefs or {}, "engine": row.engine, "scanned": row.scanned, - "shortlisted": len(row.opportunities or []), + "shortlisted": len(opps), "cached": True, "updated_at": row.updated_at.isoformat() if row.updated_at else None, } + + +async def set_opportunity_state(user_id: str, opportunity_id: str, **fields) -> None: + """Upsert per-opportunity state (status and/or document ids). No-op if the DB is off. + Only the provided fields are written, so partial updates don't clobber existing values.""" + factory = session_factory() + if factory is None or not user_id or not opportunity_id: + return + vals = {k: fields[k] for k in _STATE_FIELDS if k in fields} + if not vals: + return + row = {"user_id": user_id, "opportunity_id": opportunity_id, **vals} + try: + async with factory() as s: + stmt = pg_insert(OpportunityState).values(**row) + stmt = stmt.on_conflict_do_update( + index_elements=[OpportunityState.user_id, OpportunityState.opportunity_id], + set_=vals, + ) + await s.execute(stmt) + await s.commit() + except Exception as e: # noqa: BLE001 + logger.warning("set_opportunity_state failed (non-fatal): %s", e) + + +async def get_opportunity_states(user_id: str) -> dict[str, dict]: + """{opportunity_id: {status, tailored_resume_id, tailored_version_id, cover_letter_id}}.""" + factory = session_factory() + if factory is None or not user_id: + return {} + try: + async with factory() as s: + rows = (await s.execute( + select(OpportunityState).where(OpportunityState.user_id == user_id) + )).scalars().all() + except Exception as e: # noqa: BLE001 + logger.warning("get_opportunity_states failed (non-fatal): %s", e) + return {} + return { + r.opportunity_id: { + "status": r.status, + "tailored_resume_id": r.tailored_resume_id, + "tailored_version_id": r.tailored_version_id, + "cover_letter_id": r.cover_letter_id, + } + for r in rows + } + + +def annotate_and_filter(opportunities: list[dict], states: dict[str, dict]) -> list[dict]: + """Drop dismissed opportunities and annotate the rest with their persisted state (applied flag + + already-generated document ids) so the UI can restore without re-tailoring.""" + out: list[dict] = [] + for o in opportunities: + st = states.get(o.get("id")) + if st and st.get("status") == "dismissed": + continue # never resurface + if st: + o = {**o, + "applied": st.get("status") == "applied", + "tailored_resume_id": st.get("tailored_resume_id"), + "tailored_version_id": st.get("tailored_version_id"), + "cover_letter_id": st.get("cover_letter_id")} + out.append(o) + return out + + +async def get_search_cursor(user_id: str, query_sig: str) -> int: + """The next LinkedIn start-page for this query. Same signature as last run → advance; a new + query (different signature) → start at page 1. Returns 1 when the DB is off or query changed.""" + factory = session_factory() + if factory is None or not user_id: + return 1 + try: + async with factory() as s: + row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none() + except Exception as e: # noqa: BLE001 + logger.warning("get_search_cursor failed (non-fatal): %s", e) + return 1 + if not row or row.query_sig != query_sig: + return 1 # new/changed query → fresh cursor + return int(row.cursor_page or 1) + + +async def get_seen_ids(user_id: str) -> set[str]: + """Every opportunity the user has already been shown / acted on — excluded from new runs.""" + factory = session_factory() + if factory is None or not user_id: + return set() + try: + async with factory() as s: + rows = (await s.execute( + select(OpportunityState.opportunity_id).where(OpportunityState.user_id == user_id) + )).scalars().all() + return set(rows) + except Exception as e: # noqa: BLE001 + logger.warning("get_seen_ids failed (non-fatal): %s", e) + return set() + + +async def mark_seen(user_id: str, opportunity_ids: list[str]) -> None: + """Record that these opportunities were shown in a deck (so future runs skip them). No-op on DB off. + Sets seen=true without touching any existing status/document ids.""" + factory = session_factory() + if factory is None or not user_id or not opportunity_ids: + return + rows = [{"user_id": user_id, "opportunity_id": oid, "seen": True} for oid in opportunity_ids if oid] + if not rows: + return + try: + async with factory() as s: + stmt = pg_insert(OpportunityState).values(rows) + stmt = stmt.on_conflict_do_update( + index_elements=[OpportunityState.user_id, OpportunityState.opportunity_id], + set_={"seen": True}, + ) + await s.execute(stmt) + await s.commit() + except Exception as e: # noqa: BLE001 + logger.warning("mark_seen failed (non-fatal): %s", e) diff --git a/app/engine/apify_client.py b/app/engine/apify_client.py index aa9716c..3bbee33 100644 --- a/app/engine/apify_client.py +++ b/app/engine/apify_client.py @@ -47,6 +47,44 @@ def _cache_write(path: Path, items: list[dict]) -> None: # would double the cost for no gain (an empty result is usually a real "0 matches"). _MAX_ATTEMPTS = 2 _BACKOFF = 2.0 +_RUNS_BASE = "https://api.apify.com/v2" +_POLL = 3.0 # async-run poll interval (s) + + +async def _run_actor_async(actor_id: str, run_input: dict, *, timeout: float, token: str) -> list[dict]: + """Start an ASYNC run, poll to completion, return its dataset items. Needed for incremental / + stateful actors (state persisted across runs via `stateKey`) whose crawl exceeds the run-sync + window — that window aborts the run mid-crawl. No cache: the actor itself is the dedup.""" + async with httpx.AsyncClient(timeout=60.0) as client: + r = await client.post(f"{_BASE}/{actor_id}/runs", params={"token": token}, json=run_input) + r.raise_for_status() + run = r.json().get("data", {}) + run_id = run.get("id") + ds_id = run.get("defaultDatasetId") + if not run_id: + return [] + elapsed = 0.0 + while elapsed < timeout: + await asyncio.sleep(_POLL) + elapsed += _POLL + s = await client.get(f"{_RUNS_BASE}/actor-runs/{run_id}", params={"token": token}) + data = s.json().get("data", {}) + status = data.get("status") + ds_id = data.get("defaultDatasetId") or ds_id + if status == "SUCCEEDED": + break + if status in ("FAILED", "ABORTED", "TIMED-OUT", "TIMING-OUT"): + logger.warning("apify async run %s ended %s", actor_id, status) + return [] + else: + logger.warning("apify async run %s did not finish within %ss", actor_id, timeout) + return [] + d = await client.get(f"{_RUNS_BASE}/datasets/{ds_id}/items", params={"token": token, "format": "json"}) + d.raise_for_status() + items = d.json() + n = len(items) if isinstance(items, list) else 0 + logger.info("apify async run %s SUCCEEDED — %d items (incremental)", actor_id, n) + return items if isinstance(items, list) else [] async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0, @@ -70,6 +108,11 @@ async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0, token = get_settings().APIFY_TOKEN if not token: raise RuntimeError("APIFY_TOKEN not configured") + + # Incremental/stateful actors need an async run (the sync window aborts the longer crawl). + if run_input.get("incremental"): + return await _run_actor_async(actor_id, run_input, timeout=timeout, token=token) + url = f"{_BASE}/{actor_id}/run-sync-get-dataset-items" async with httpx.AsyncClient(timeout=timeout) as client: diff --git a/app/engine/board_adapters/adapters.py b/app/engine/board_adapters/adapters.py index 09052a0..cc9005a 100644 --- a/app/engine/board_adapters/adapters.py +++ b/app/engine/board_adapters/adapters.py @@ -67,12 +67,13 @@ def build_naukri_input(prefs: Prefs, *, max_jobs: int = 80, freshness: str = "7" # ── Naukri (blackfalcondata feed — city-precise, real offsite applyRedirectUrl) ── -def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details: bool = True) -> dict: +def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details: bool = True, + state_key: str | None = None) -> dict: # City-filtered at the board (hyper-relevant + cheap). fetch_details=True surfaces the # offsite `applyRedirectUrl`; skipReposts dedups for cost. Narrowing (dept/industry/exp) # stays off here — that's engine ranking later. city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "") - return { + out: dict[str, Any] = { "keyword": C.title_of(prefs) or C.build_keyword(prefs), "location": city, # precise city, e.g. "New Delhi" "maxResults": max_jobs, @@ -80,6 +81,13 @@ def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details: "skipReposts": True, "sortBy": "date", } + # Incremental mode: the actor persists what it has emitted under `stateKey` (a named state store) + # and returns only NEW / changed jobs on each run — so re-running the same search costs us only + # genuinely new postings, with ZERO re-fetch. Requires an ASYNC run (apify_client routes it there). + if state_key: + out["incremental"] = True + out["stateKey"] = state_key + return out # ── Foundit / Monster India (city-precise; apply_url is the offsite redirect) ── @@ -103,7 +111,7 @@ def build_foundit_input(prefs: Prefs, *, results_wanted: int = 50) -> dict: # ── LinkedIn (takes the band, not years; ₹ comp incompatible) ──────────────── def build_linkedin_input(prefs: Prefs, *, max_items: int = 80, posted: str = "week", - recall: bool = False) -> dict: + recall: bool = False, page: int = 1) -> dict: # recall=True (retrieve stage): title + COUNTRY-level location only — drop the narrowing # filters (easyApply / workplaceType / experienceLevel / industryIds / freshness), which # become ENGINE ranking later. City-level location + a wrong industryId zero the actor out. @@ -111,6 +119,7 @@ def build_linkedin_input(prefs: Prefs, *, max_items: int = 80, posted: str = "we "jobTitles": [C.title_of(prefs)], "maxItems": max_items, "sortBy": "date", + "page": page, # Start Page Number — advance per re-run for NEW jobs } cities: list[str] = [] # bare city names ("New Delhi") — harvestapi's location format diff --git a/app/engine/normalize.py b/app/engine/normalize.py index ee8398f..42b9d42 100644 --- a/app/engine/normalize.py +++ b/app/engine/normalize.py @@ -46,11 +46,33 @@ def _compact(d: dict) -> dict: return {k: v for k, v in d.items() if v not in (None, "", [], {})} -def _logo(org: str) -> dict: +def _logo(org: str, url: str | None = None) -> dict: org = (org or "").strip() label = org[0].upper() if org else "?" bg = _PALETTE[sum(map(ord, org)) % len(_PALETTE)] if org else "#9CA3AF" - return {"label": label, "bg": bg} + d = {"label": label, "bg": bg} + # Real company logo from the actor (used by the UI; the label/bg stay as the fallback). + if url and isinstance(url, str) and url.startswith("http"): + d["url"] = url + return d + + +# actor field names for the company logo, across boards + nesting (Naukri=logoPath, +# LinkedIn=company.logo, Indeed=companyLogo, …). First http(s) value wins; None → letter badge. +_LOGO_KEYS = ("logoPath", "logoPathV3", "logo", "companyLogo", "company_logo", "logoUrl", "image", "companyImage") + + +def _logo_url(it: dict) -> str | None: + scopes = [it, it.get("company") or {}, it.get("jobDetails") or {}, + (it.get("jobDetails") or {}).get("companyDetail") or {}] + for sc in scopes: + if not isinstance(sc, dict): + continue + for k in _LOGO_KEYS: + v = sc.get(k) + if isinstance(v, str) and v.startswith("http"): + return v + return None def _digits(s: str) -> int | None: @@ -114,11 +136,16 @@ def naukri_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None: sd = jd.get("salaryDetail") or {} if isinstance(sd, dict) and not sd.get("hideSalary") and sd.get("maximumSalary"): lpa, pay = _salary_from_amount(int(sd["maximumSalary"]), per_year=True) + # Offsite employer/ATS apply link first; the Naukri listing (staticUrl) is the fallback. + redirect = ( + jd.get("applyRedirectUrl") or jd.get("companyApplyUrl") + or it.get("applyRedirectUrl") or it.get("companyApplyUrl") + ) return { "id": f"naukri-{jd.get('jobId')}", "title": title, "organization": org, - "logo": _logo(org), + "logo": _logo(org, _logo_url(it)), "location_city": city, "location_country": "India", "matchScore": 0, # placeholder — set after merge @@ -129,7 +156,8 @@ def naukri_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None: "tags": skills[:3], "applicants": jd.get("applyCount"), "note": f"Live from {board}", - "apply_url": jd.get("staticUrl"), + "apply_url": redirect or jd.get("staticUrl"), # offsite employer link first, Naukri listing fallback + "offsite_apply": bool(redirect), "qx": 10, } @@ -151,14 +179,14 @@ def indeed_to_scoutjob(it: dict, board: str = "Indeed") -> dict | None: pass sal = it.get("salary") or "" lpa, pay = _salary_from_amount(_digits(sal), per_year="year" in sal.lower()) if sal else (None, "Not disclosed") - apply_url = it.get("externalApplyLink") - if not apply_url or apply_url == "None": - apply_url = it.get("url") + ext = it.get("externalApplyLink") + offsite = bool(ext and ext != "None") # externalApplyLink = real offsite employer link + apply_url = ext if offsite else it.get("url") # offsite first, Indeed listing fallback return { "id": f"indeed-{it.get('id')}", "title": title, "organization": org, - "logo": _logo(org), + "logo": _logo(org, _logo_url(it)), "location_city": city, "location_country": "India", "matchScore": 0, @@ -169,6 +197,7 @@ def indeed_to_scoutjob(it: dict, board: str = "Indeed") -> dict | None: "applicants": None, "note": f"Live from {board}", "apply_url": apply_url, + "offsite_apply": offsite, "qx": 10, } @@ -195,11 +224,19 @@ def linkedin_to_scoutjob(it: dict, board: str = "LinkedIn") -> dict | None: "skills": industries, "company_about": _plain(co.get("description"), 700) if isinstance(co, dict) else "", }) + # Prefer the OFFSITE employer/ATS apply link over the LinkedIn listing — the + # board URL can go stale, the employer's own apply page is the durable target. + _apply_method = it.get("applyMethod") or {} + offsite = ( + it.get("companyApplyUrl") + or (_apply_method.get("companyApplyUrl") if isinstance(_apply_method, dict) else None) + or it.get("applyRedirectUrl") + ) return { "id": f"linkedin-{it.get('id')}", "title": title, "organization": org, - "logo": _logo(org), + "logo": _logo(org, _logo_url(it)), "location_city": city, "location_country": country, "location_mode": mode, @@ -211,7 +248,8 @@ def linkedin_to_scoutjob(it: dict, board: str = "LinkedIn") -> dict | None: "tags": industries[:3], "applicants": it.get("applicants"), "note": f"Live from {board}", - "apply_url": it.get("linkedinUrl"), + "apply_url": offsite or it.get("linkedinUrl"), # offsite employer link first, board link fallback + "offsite_apply": bool(offsite), "details": details, "qx": 10, } @@ -231,7 +269,7 @@ def ats_to_scoutjob(it: dict, board: str = "Direct") -> dict | None: "id": f"ats-{it.get('global_id') or it.get('job_id')}", "title": title, "organization": org, - "logo": _logo(org), + "logo": _logo(org, _logo_url(it)), "location_city": city, "location_country": None, "location_mode": mode, @@ -244,6 +282,7 @@ def ats_to_scoutjob(it: dict, board: str = "Direct") -> dict | None: "applicants": None, "note": f"Live from {board}", # direct employer posting → real offsite apply link "apply_url": it.get("apply_url") or it.get("url"), + "offsite_apply": bool(it.get("apply_url") or it.get("url")), # ATS direct posting = employer's own apply page "qx": 10, } @@ -288,7 +327,7 @@ def naukri_feed_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None: "id": f"naukri-{it.get('jobId')}", "title": title.title() if title.isupper() else title, "organization": org, - "logo": _logo(org), + "logo": _logo(org, _logo_url(it)), "location_city": city, "location_country": "India", "location_mode": _WORKMODE.get((it.get("wfhType") or "").lower()), @@ -335,7 +374,7 @@ def foundit_to_scoutjob(it: dict, board: str = "Foundit") -> dict | None: "id": f"foundit-{it.get('job_id')}", "title": title, "organization": org, - "logo": _logo(org), + "logo": _logo(org, _logo_url(it)), "location_city": city, "location_country": "India", "matchScore": 0, diff --git a/app/engine/search.py b/app/engine/search.py index 08bf214..ef12647 100644 --- a/app/engine/search.py +++ b/app/engine/search.py @@ -25,13 +25,15 @@ def _ats_companies() -> list[dict]: # all city-filtered at the board. The rest stay registered (code ready) but off unless enabled. BOARDS = { "naukri": ("blackfalcondata~naukri-jobs-feed", - lambda p: A.build_naukri_feed_input(p, max_jobs=get_settings().NAUKRI_MAX_JOBS), + lambda p: A.build_naukri_feed_input(p, max_jobs=get_settings().NAUKRI_MAX_JOBS, + state_key=p.get("_naukri_state")), 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", - lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS), + lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS, + page=int(p.get("_page") or 1)), N.linkedin_to_scoutjob), # ── registered fallbacks / other lanes (off unless added to BOARDS_ENABLED) ── "ats": ("bovi~greenhouse-lever-ashby-job-scraper",