From be921559e716c314ff5d1ed42d15244f8b8da1f4 Mon Sep 17 00:00:00 2001 From: raulgupta Date: Sun, 21 Jun 2026 12:12:28 +0530 Subject: [PATCH] Real Scout dashboard backend + per-board cursor rework MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - get_scout_stats aggregator (app/engine/stats): assembles the REAL Summary metrics — funnel + cohort engagement rank + active-window (activity timestamps by hour) from our DB, accumulating salary band (avg of each deck's peak, ₹L), match/competition stats from the feed, Momentum/QX + Q-Score trend (qscore-service), day streak (user-service). Honest: unsourced cards return None so the UI omits/locks them, never faked. posted_date extractor. - Activity tracking: viewed/saved flags + search_count → funnel (Matches→Viewed→Shortlisted→ Applied) + engagement percentile. matchesFound = all-time count. - Per-board search cursors {board: page} (replaces the single cursor): only boards that truly paginate (LinkedIn) get one; cursor = LAST page fetched (1st search of a new query → 1). Resets on query change OR >24h (boards refresh ~daily). Dropped Naukri incremental/stateKey (opaque, exhausting, cross-account dedup state) — dedup is the PER-USER seen-net only. - tests: stats helpers (posting-age, histogram, active-window, engagement score). --- app/agent/session.py | 51 ++++++-- app/config.py | 8 +- app/db/models.py | 25 ++-- app/db/repo.py | 174 +++++++++++++++++++++++--- app/engine/board_adapters/adapters.py | 16 +-- app/engine/normalize.py | 23 ++++ app/engine/search.py | 12 +- app/engine/stats/__init__.py | 114 +++++++++++++++++ app/engine/stats/clients.py | 75 +++++++++++ tests/test_stats.py | 33 +++++ 10 files changed, 483 insertions(+), 48 deletions(-) create mode 100644 app/engine/stats/__init__.py create mode 100644 app/engine/stats/clients.py create mode 100644 tests/test_stats.py diff --git a/app/agent/session.py b/app/agent/session.py index 3c681e5..9e33066 100644 --- a/app/agent/session.py +++ b/app/agent/session.py @@ -51,6 +51,10 @@ class MatchmakingAgentSession: "restore_opportunity": self.handle_restore_opportunity, "mark_applied": self.handle_mark_applied, "save_apply_docs": self.handle_save_apply_docs, + # activity tracking → the dashboard funnel + engagement rank + "mark_viewed": self.handle_mark_viewed, + "mark_saved": self.handle_mark_saved, + "get_scout_stats": self.handle_get_scout_stats, } handler = handlers.get(action) if not handler: @@ -91,14 +95,13 @@ 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. + # Per-board search cursors: re-running the SAME query advances EACH board's own page so every + # run fetches GENUINELY NEW jobs. Resets when the query changes OR >24h (boards refresh ~daily). + # Dedup is the per-user seen-net below — never cross-user, no opaque actor-side state. 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(":", "_") + cursors = await _repo.get_search_cursors(self.user_id, sig) + prefs["_cursors"] = cursors 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). @@ -124,11 +127,19 @@ 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. + # Persist the RAW deck + ADVANCE every board that ran (its page +1) so the next run of this + # query fetches new jobs. Per-opportunity state (applied/dismissed/docs) lives separately. 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) + if k not in ("user_context", "_fresh", "_cursors")} + # Store the LAST page fetched per paginating board (last + 1 = the page we just pulled). So a + # NEW query's first search stores 1 (page 1), not 2. Naukri/Foundit have no page param → no + # cursor (they dedup via the per-user seen-net). + next_cursors = {b: int(cursors.get(b, 0)) + 1 + for b in (sweep.get("sources") or {}) if b in search.PAGINATING_BOARDS} + await _repo.save_feed(self.user_id, clean_prefs, result, query_sig=sig, cursors=next_cursors) + # Accumulate the salary band — each deck contributes its peak salary, averaged over decks + # (survives decks that disclose nothing, like Naukri "Not disclosed"). + await _repo.update_salary_band(self.user_id, [o.get("salary_lpa") for o in (result["opportunities"] or [])]) # 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. @@ -187,3 +198,23 @@ class MatchmakingAgentSession: 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}) + + # ── activity tracking (dashboard funnel + engagement rank) ── + async def handle_mark_viewed(self, params: dict): + """The user opened a job → funnel 'Viewed'. Fire-and-forget (no response needed).""" + from app.db.repo import set_opportunity_state + await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), viewed=True) + + async def handle_mark_saved(self, params: dict): + """The user saved/swiped-right a job → funnel 'Shortlisted'.""" + from app.db.repo import set_opportunity_state + await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), saved=True) + + async def handle_get_scout_stats(self, params: dict): + """Assemble the REAL Summary dashboard metrics (Phase 2 fills the cross-service bits).""" + from app.engine import stats + data = await stats.build_scout_stats( + self.user_id, + user_uuid=params.get("user_uuid"), + ) + await self.push("agent_data", action="scout_stats", data=data) diff --git a/app/config.py b/app/config.py index b9c5010..5fe70cb 100644 --- a/app/config.py +++ b/app/config.py @@ -63,9 +63,15 @@ class Settings(BaseSettings): ATS_COMPANIES: str = "stripe,databricks,gitlab,figma,ramp,notion,razorpay,zerodha" INDEED_MAX_JOBS: int = 30 - # QScore (competence proxy) — consumed, not computed + # QScore (competence proxy) — consumed, not computed. Dashboard Momentum/QX reads it via REST. QSCORE_BASE_URL: str = "http://localhost:8004" QSCORE_AUTH_TOKEN: str | None = None + QSCORE_ORG_ID: str = "growqr" + # user-service — dashboard Day-streak reads metadata.current_streak via GET /api/state/{clerk_id} + USER_SERVICE_BASE_URL: str = "http://localhost:8003" + # dashboard-service — Q-Score trend series via GET /api/qscore-history/{clerk_id} + DASHBOARD_SERVICE_BASE_URL: str = "http://localhost:8005" + A2A_OUTBOUND_KEY: str = "dev-a2a-key" # Bearer token presented to sibling services # Storage for feedback/labels (NOT the old corpus). Optional until the learning loop lands. DATABASE_URL: str | None = None diff --git a/app/db/models.py b/app/db/models.py index 80fa04a..032b4d8 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, Boolean, DateTime, Integer, String, func +from sqlalchemy import JSON, Boolean, DateTime, Float, Integer, String, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column @@ -22,19 +22,26 @@ 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. + # PER-BOARD search cursors — {board: LAST page fetched}. Next page = last + 1, so a NEW query's + # first search reads page 1 and stores 1. Resets to {} when the query signature changes (or >24h). + # Dedup is purely per-user (the seen-net below) — never cross-user, never opaque actor-side state. query_sig: Mapped[str] = mapped_column(String, default="") - cursor_page: Mapped[int] = mapped_column(Integer, default=1) + cursors: Mapped[dict] = mapped_column(JSON, default=dict) + search_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0") # total searches run (engagement) + # Salary band = AVERAGE of each deck's HIGHEST disclosed salary (₹L). Each deck contributes one + # peak; the band is the running mean of those peaks. Survives decks that disclose nothing. + salary_peak_sum: Mapped[float] = mapped_column(Float, default=0.0, server_default="0") + salary_peak_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0") 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).""" + """Per-(user, opportunity) state that SURVIVES deck refreshes. Persists: `applied`/`dismissed` + status, the resume-builder document ids already generated (never re-pay to re-tailor), and the + PROGRESSIVE activity flags that feed the dashboard funnel + engagement rank. A job can be viewed + AND saved AND applied at once, so viewed/saved are booleans (not the single-valued status).""" __tablename__ = "opportunity_state" user_id: Mapped[str] = mapped_column(String, primary_key=True) @@ -43,7 +50,9 @@ class OpportunityState(Base): 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 + seen: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # shown in a deck → exclude from new runs + viewed: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # opened the job → funnel "Viewed" + saved: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # swiped right → funnel "Shortlisted" 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 aadceec..3003e5f 100644 --- a/app/db/repo.py +++ b/app/db/repo.py @@ -6,7 +6,7 @@ import hashlib import json import logging -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.dialects.postgresql import insert as pg_insert from app.db.models import Base, OpportunityState, UserFeed @@ -14,8 +14,8 @@ 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") +# fields a caller may set on an opportunity's state (status + the generated document ids + activity flags) +_STATE_FIELDS = ("status", "tailored_resume_id", "tailored_version_id", "cover_letter_id", "seen", "viewed", "saved") # the search-defining pref fields — same values → same signature → cursor advances (new jobs); # change any → new signature → cursor resets to page 1. @@ -43,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, *, 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.""" +async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str = "", cursors: dict | None = None) -> None: + """Upsert the user's last completed search + the advanced per-board cursors. No-op if the DB is off.""" factory = session_factory() if factory is None or not user_id: return @@ -56,14 +56,17 @@ async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str = "engine": result.get("engine") or "", "scanned": int(result.get("scanned") or 0), "query_sig": query_sig, - "cursor_page": cursor_page, + "cursors": cursors or {}, + "search_count": 1, } 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", "query_sig", "cursor_page")}, + set_={**{k: row[k] for k in ("prefs", "opportunities", "sources", "engine", "scanned", "query_sig", "cursors")}, + # engagement: count every search run (insert starts at 1, each re-run +1) + "search_count": UserFeed.search_count + 1}, ) await s.execute(stmt) await s.commit() @@ -71,6 +74,42 @@ async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str = logger.warning("save_feed failed (non-fatal): %s", e) +async def update_salary_band(user_id: str, salaries: list[float]) -> None: + """Accumulate the salary band: each deck contributes its HIGHEST disclosed salary (a 'peak'); the + band is the running AVERAGE of those peaks (₹L). No-op when the deck disclosed nothing (persists).""" + sal = [float(x) for x in salaries if x] + factory = session_factory() + if factory is None or not user_id or not sal: + return + hi = max(sal) + try: + async with factory() as s: + row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none() + if row is None: + return + row.salary_peak_sum = (row.salary_peak_sum or 0.0) + hi # this deck's peak + row.salary_peak_count = (row.salary_peak_count or 0) + 1 + await s.commit() + except Exception as e: # noqa: BLE001 + logger.warning("update_salary_band failed (non-fatal): %s", e) + + +async def get_salary_band(user_id: str) -> float | None: + """The salary band = average of each deck's peak salary (₹L), or None if no deck disclosed one.""" + factory = session_factory() + if factory is None or not user_id: + return None + 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_salary_band failed (non-fatal): %s", e) + return None + if not row or not row.salary_peak_count: + return None + return round((row.salary_peak_sum or 0.0) / row.salary_peak_count, 1) + + async def get_feed(user_id: str) -> dict | None: """The user's last cached feed as a search_complete-shaped dict, or None.""" factory = session_factory() @@ -145,6 +184,101 @@ async def get_opportunity_states(user_id: str) -> dict[str, dict]: } +# engagement score weights — applies count most, then saves, views, searches (honest "working it") +def _activity_score(viewed: int, saved: int, applied: int, searches: int) -> int: + return applied * 3 + saved * 2 + viewed * 1 + searches * 1 + + +async def get_activity_stats(user_id: str) -> dict: + """Real dashboard activity for `user_id`: the funnel counts (viewed/saved/applied), how many + documents Mira tailored, search count, and the COHORT engagement percentile (this user's activity + vs every Scout user in the matchmaking DB). All honest, DB-derived. Degrades to zeros if DB is off.""" + factory = session_factory() + empty = {"matches": 0, "viewed": 0, "saved": 0, "applied": 0, "tailored": 0, "searches": 0, "percentile": 0, "cohort_size": 0} + if factory is None or not user_id: + return empty + try: + async with factory() as s: + # per-user activity across ALL users (one grouped pass → funnel for this user + cohort rank) + os_rows = (await s.execute( + select( + OpportunityState.user_id, + func.count().filter(OpportunityState.seen.is_(True)).label("matches"), + func.count().filter(OpportunityState.viewed.is_(True)).label("viewed"), + func.count().filter(OpportunityState.saved.is_(True)).label("saved"), + func.count().filter(OpportunityState.status == "applied").label("applied"), + func.count().filter( + OpportunityState.tailored_version_id.isnot(None) + | OpportunityState.cover_letter_id.isnot(None) + ).label("tailored"), + ).group_by(OpportunityState.user_id) + )).all() + feed_rows = (await s.execute(select(UserFeed.user_id, UserFeed.search_count))).all() + except Exception as e: # noqa: BLE001 + logger.warning("get_activity_stats failed (non-fatal): %s", e) + return empty + + searches_by = {r.user_id: int(r.search_count or 0) for r in feed_rows} + me, scores = dict(empty), [] + users = {r.user_id for r in os_rows} | set(searches_by) + by_os = {r.user_id: r for r in os_rows} + for uid in users: + r = by_os.get(uid) + mt, v, sv, ap, tl = (int(r.matches), int(r.viewed), int(r.saved), int(r.applied), int(r.tailored)) if r else (0, 0, 0, 0, 0) + sc = searches_by.get(uid, 0) + scores.append(_activity_score(v, sv, ap, sc)) + if uid == user_id: + me = {"matches": mt, "viewed": v, "saved": sv, "applied": ap, "tailored": tl, "searches": sc} + my_score = _activity_score(me["viewed"], me["saved"], me["applied"], me["searches"]) + cohort = len(scores) or 1 + # percentile = share of the cohort you're STRICTLY ahead of (so rank-1-of-3 → 67th → "top 33%", + # not "top 0%"). UI shows top (100 - percentile)%. + below = sum(1 for x in scores if x < my_score) + me["percentile"] = round(100 * below / cohort) + me["cohort_size"] = cohort + return me + + +async def get_active_window(user_id: str, *, tz_offset_min: int = 330) -> dict: + """WHEN the user works their search — activity timestamps (views/saves/applies + searches) + bucketed by local hour-of-day. Returns the peak 3-hour band + a 24-bucket histogram. Real signal, + from this DB. tz_offset_min defaults to IST (+5:30); the product is India-centric. + + SQLite/Postgres-agnostic: we read raw UTC timestamps and bucket in Python (so it works in tests too).""" + factory = session_factory() + empty = {"by_hour": [0] * 24, "peak_start": None, "peak_end": None, "samples": 0} + if factory is None or not user_id: + return empty + try: + async with factory() as s: + os_ts = (await s.execute( + select(OpportunityState.updated_at).where(OpportunityState.user_id == user_id) + )).scalars().all() + feed_ts = (await s.execute( + select(UserFeed.updated_at).where(UserFeed.user_id == user_id) + )).scalars().all() + except Exception as e: # noqa: BLE001 + logger.warning("get_active_window failed (non-fatal): %s", e) + return empty + + by_hour = [0] * 24 + for ts in [*os_ts, *feed_ts]: + if ts is None: + continue + local_min = ts.hour * 60 + ts.minute + tz_offset_min + by_hour[(local_min // 60) % 24] += 1 + samples = sum(by_hour) + if not samples: + return empty + # peak 3-hour window (circular) by summed activity + best_h, best_sum = 0, -1 + for h in range(24): + win = by_hour[h] + by_hour[(h + 1) % 24] + by_hour[(h + 2) % 24] + if win > best_sum: + best_h, best_sum = h, win + return {"by_hour": by_hour, "peak_start": best_h, "peak_end": (best_h + 3) % 24, "samples": samples} + + 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.""" @@ -163,21 +297,31 @@ def annotate_and_filter(opportunities: list[dict], states: dict[str, dict]) -> l 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.""" +_CURSOR_TTL_HOURS = 24 # job boards refresh ~daily (live scrape, 7-day newest-first window) → reset every 24h + + +async def get_search_cursors(user_id: str, query_sig: str) -> dict: + """Per-board start pages for this query, e.g. {'linkedin': 3}. RESETS to empty (every board → page 1) + when EITHER the query changed OR the last search was >24h ago (so we re-surface the freshest daily + postings instead of marching into stale depth). Each board advances its OWN page; dedup is the + per-user seen-net, never cross-user.""" + from datetime import datetime, timezone factory = session_factory() if factory is None or not user_id: - return 1 + return {} 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 + logger.warning("get_search_cursors failed (non-fatal): %s", e) + return {} if not row or row.query_sig != query_sig: - return 1 # new/changed query → fresh cursor - return int(row.cursor_page or 1) + return {} # new/changed query → all boards start at page 1 + if row.updated_at: + age_h = (datetime.now(timezone.utc) - row.updated_at).total_seconds() / 3600 + if age_h >= _CURSOR_TTL_HOURS: + return {} # >24h stale → reset to catch the newest jobs + return dict(row.cursors or {}) async def get_seen_ids(user_id: str) -> set[str]: diff --git a/app/engine/board_adapters/adapters.py b/app/engine/board_adapters/adapters.py index cc9005a..8e1a22b 100644 --- a/app/engine/board_adapters/adapters.py +++ b/app/engine/board_adapters/adapters.py @@ -67,13 +67,14 @@ 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, - state_key: str | None = None) -> dict: +def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details: bool = True) -> dict: # City-filtered at the board (hyper-relevant + cheap). fetch_details=True surfaces the - # offsite `applyRedirectUrl`; skipReposts dedups for cost. Narrowing (dept/industry/exp) + # offsite `applyRedirectUrl`; skipReposts dedups WITHIN a run. Narrowing (dept/industry/exp) # stays off here — that's engine ranking later. + # NOTE: no `incremental`/`stateKey` — that put opaque, exhausting, cross-account dedup state on + # Apify. Cross-deck dedup is the PER-USER seen-net in our DB. (The feed actor has no page/offset.) city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "") - out: dict[str, Any] = { + return { "keyword": C.title_of(prefs) or C.build_keyword(prefs), "location": city, # precise city, e.g. "New Delhi" "maxResults": max_jobs, @@ -81,13 +82,6 @@ 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) ── diff --git a/app/engine/normalize.py b/app/engine/normalize.py index 42b9d42..fb88c25 100644 --- a/app/engine/normalize.py +++ b/app/engine/normalize.py @@ -75,6 +75,23 @@ def _logo_url(it: dict) -> str | None: return None +# posting-date field across boards (Naukri createdDate · LinkedIn postedDate · Foundit date_posted). +# Feeds the dashboard "Apply window" (freshness urgency) + honest "posted Nd ago" age signals. +_POSTED_KEYS = ("postedDate", "createdDate", "date_posted", "postedAt", "listedAt", "datePosted", "published") + + +def _posted_date(it: dict) -> str | None: + scopes = [it, it.get("jobDetails") or {}] + for sc in scopes: + if not isinstance(sc, dict): + continue + for k in _POSTED_KEYS: + v = sc.get(k) + if isinstance(v, str) and v[:4].isdigit(): # ISO-ish "2026-06-18T…" + return v + return None + + def _digits(s: str) -> int | None: nums = re.sub(r"[^\d]", "", s or "") return int(nums) if nums else None @@ -146,6 +163,7 @@ def naukri_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None: "title": title, "organization": org, "logo": _logo(org, _logo_url(it)), + "posted_date": _posted_date(it), "location_city": city, "location_country": "India", "matchScore": 0, # placeholder — set after merge @@ -187,6 +205,7 @@ def indeed_to_scoutjob(it: dict, board: str = "Indeed") -> dict | None: "title": title, "organization": org, "logo": _logo(org, _logo_url(it)), + "posted_date": _posted_date(it), "location_city": city, "location_country": "India", "matchScore": 0, @@ -237,6 +256,7 @@ def linkedin_to_scoutjob(it: dict, board: str = "LinkedIn") -> dict | None: "title": title, "organization": org, "logo": _logo(org, _logo_url(it)), + "posted_date": _posted_date(it), "location_city": city, "location_country": country, "location_mode": mode, @@ -270,6 +290,7 @@ def ats_to_scoutjob(it: dict, board: str = "Direct") -> dict | None: "title": title, "organization": org, "logo": _logo(org, _logo_url(it)), + "posted_date": _posted_date(it), "location_city": city, "location_country": None, "location_mode": mode, @@ -328,6 +349,7 @@ def naukri_feed_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None: "title": title.title() if title.isupper() else title, "organization": org, "logo": _logo(org, _logo_url(it)), + "posted_date": _posted_date(it), "location_city": city, "location_country": "India", "location_mode": _WORKMODE.get((it.get("wfhType") or "").lower()), @@ -375,6 +397,7 @@ def foundit_to_scoutjob(it: dict, board: str = "Foundit") -> dict | None: "title": title, "organization": org, "logo": _logo(org, _logo_url(it)), + "posted_date": _posted_date(it), "location_city": city, "location_country": "India", "matchScore": 0, diff --git a/app/engine/search.py b/app/engine/search.py index ef12647..7a7bce7 100644 --- a/app/engine/search.py +++ b/app/engine/search.py @@ -21,19 +21,25 @@ 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", - lambda p: A.build_naukri_feed_input(p, max_jobs=get_settings().NAUKRI_MAX_JOBS, - state_key=p.get("_naukri_state")), + # 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("_page") or 1)), + 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", diff --git a/app/engine/stats/__init__.py b/app/engine/stats/__init__.py new file mode 100644 index 0000000..c75183c --- /dev/null +++ b/app/engine/stats/__init__.py @@ -0,0 +1,114 @@ +"""Scout Summary dashboard — assemble REAL metrics (no fabricated numbers). + +Honest sources (SCOUT_UI_SPEC): +- feed stats (matches / top / range / salary / competition) ← the persisted feed +- funnel + engagement rank ← opportunity_state activity (this DB) +- Apply window (freshness urgency) ← listing posting-age +- Momentum/QX + Readiness trend ← qscore-service (Phase 1) +- Day streak ← dashboard-service (Phase 1) + +Anything without a real source is omitted — the UI cuts or locks it, never fakes it. +""" +from __future__ import annotations + +import asyncio +import logging +from datetime import datetime, timezone + +from app.db import repo as _repo +from app.engine.stats import clients as _clients + +logger = logging.getLogger(__name__) + + +def _match_scores(opps: list[dict]) -> list[int]: + out = [] + for o in opps: + s = (o.get("match") or {}).get("score") or o.get("matchScore") + if isinstance(s, (int, float)) and s > 0: + out.append(int(s)) + return out + + +def _histogram(scores: list[int], lo: int = 60, hi: int = 100, bins: int = 11) -> list[int]: + h = [0] * bins + span = (hi - lo) / bins + for s in scores: + i = min(bins - 1, max(0, int((s - lo) / span))) + h[i] += 1 + return h + + +def _days_since(iso: str | None) -> int | None: + if not iso: + return None + try: + dt = datetime.fromisoformat(iso.replace("Z", "+00:00")) + return max(0, (datetime.now(timezone.utc) - dt).days) + except (ValueError, TypeError): + return None + + +def _apply_window(opps: list[dict]) -> dict: + """Freshness urgency from listing posting-age (replaces the old 'golden window'). Fresher + deck → apply now. Honest: derived from the real posting dates the actors returned.""" + ages = [d for o in opps if (d := _days_since(o.get("posted_date"))) is not None] + if not ages: + return {"median_age_days": None, "fresh_share": None, "urgency": None} + ages.sort() + median = ages[len(ages) // 2] + fresh = sum(1 for a in ages if a <= 3) / len(ages) # posted within 3 days + urgency = "high" if median <= 3 else "medium" if median <= 7 else "low" + return {"median_age_days": median, "fresh_share": round(fresh, 2), "urgency": urgency, "dated": len(ages)} + + +async def build_scout_stats(user_id: str, *, user_uuid: str | None = None) -> dict: + """The real Summary payload. Feed + activity are wired now; qscore/streak land in Phase 1.""" + # local DB reads + cross-service reads (QX, QX-trend, streak) — all concurrent, each degrades to None + feed, act, active, salary_band, qx, qx_trend, streak = await asyncio.gather( + _repo.get_feed(user_id), + _repo.get_activity_stats(user_id), + _repo.get_active_window(user_id), + _repo.get_salary_band(user_id), + _clients.fetch_qx(user_uuid), + _clients.fetch_qx_trend(user_uuid), + _clients.fetch_streak(user_id), + ) + feed = feed or {} + opps = feed.get("opportunities") or [] + + scores = _match_scores(opps) + apps = [j.get("applicants") for j in opps if j.get("applicants")] + matches = len(opps) + + return { + # ── feed-derived (real) ── + # all-time roles Scout has matched you with (consistent with the funnel's "Matches"), not just + # the latest deck — that's the headline "Matches found" number. + "matchesFound": max(act["matches"], matches), + "topMatch": max(scores) if scores else None, + "matchMin": min(scores) if scores else None, + "matchMax": max(scores) if scores else None, + "matchHist": _histogram(scores) if scores else [], + # salary band = average of each deck's peak salary, accumulated across decks (₹L) + "salaryMaxL": salary_band, + "avgApplicants": round(sum(apps) / len(apps)) if apps else None, + "activeWindow": active if active.get("samples") else None, # WHEN the user works their search + # ── activity-derived (real, this DB) — funnel is all-time so the stages are consistent ── + "funnel": [ + {"label": "Matches", "value": max(act["matches"], matches), "tracked": True}, + {"label": "Viewed", "value": act["viewed"], "tracked": True}, + {"label": "Shortlisted", "value": act["saved"], "tracked": True}, + {"label": "Applied", "value": act["applied"], "tracked": True}, + ], + "tailoredResumes": act["tailored"], + "searches": act["searches"], + "percentile": act["percentile"], # cohort engagement percentile (top 100-percentile %) + "cohortSize": act["cohort_size"], + "searchStage": "Applying" if act["applied"] else "Reviewing" if act["saved"] else "Scanning", + # ── cross-service (real; None → the UI omits the card, never fakes it) ── + "qxNow": (qx or {}).get("qx"), + "quotients": (qx or {}).get("quotients"), + "qxTrend": qx_trend, # Q-Score series (dashboard-service) → Readiness-trend sparkline + "dayStreak": streak, + } diff --git a/app/engine/stats/clients.py b/app/engine/stats/clients.py new file mode 100644 index 0000000..c12e0c4 --- /dev/null +++ b/app/engine/stats/clients.py @@ -0,0 +1,75 @@ +"""Cross-service reads for the dashboard — QScore (Momentum/QX) + user-service (Day streak). + +Direct REST, Clerk-mesh-authed. Both degrade to None on any failure (timeout / 404 / service down) — +a missing KPI is omitted by the UI, never faked, and never crashes the dashboard. +""" +from __future__ import annotations + +import logging + +import httpx + +from app.config import get_settings + +logger = logging.getLogger(__name__) + + +async def fetch_qx(user_uuid: str | None) -> dict | None: + """QScore for the user → {qx, quotients}. GET /v1/qscore/{uuid}?org_id (qscore-service).""" + if not user_uuid: + return None + s = get_settings() + try: + async with httpx.AsyncClient(timeout=4.0) as c: + r = await c.get(f"{s.QSCORE_BASE_URL}/v1/qscore/{user_uuid}", params={"org_id": s.QSCORE_ORG_ID}) + if r.status_code == 404: # no score computed yet — honest absence + return None + r.raise_for_status() + d = r.json() + qx = d.get("q_score") + return {"qx": round(float(qx), 1), "quotients": d.get("quotients") or {}} if qx is not None else None + except (httpx.HTTPError, ValueError) as e: + logger.warning("fetch_qx failed (non-fatal): %s", e) + return None + + +async def fetch_qx_trend(user_uuid: str | None) -> list[float] | None: + """Q-Score trend (oldest→newest) ← qscore-service GET /v1/qscore/{uuid}/history. Same source as + the live QX, so the trend's latest point matches the Momentum value (consistent, not stale). + None on failure → the UI keeps the Readiness card locked.""" + if not user_uuid: + return None + s = get_settings() + try: + async with httpx.AsyncClient(timeout=4.0) as c: + r = await c.get( + f"{s.QSCORE_BASE_URL}/v1/qscore/{user_uuid}/history", + params={"org_id": s.QSCORE_ORG_ID, "points": 10}, + ) + if r.status_code == 404: + return None + r.raise_for_status() + pts = [round(float(p["q_score"]), 1) for p in (r.json().get("points") or []) if p.get("q_score") is not None] + return pts if len(pts) >= 2 else None # need ≥2 points for a trend + except (httpx.HTTPError, ValueError, TypeError, KeyError) as e: + logger.warning("fetch_qx_trend failed (non-fatal): %s", e) + return None + + +async def fetch_streak(clerk_id: str | None) -> int | None: + """Day streak ← user-service metadata.current_streak. GET /api/state/{clerk_id} (Bearer A2A).""" + if not clerk_id: + return None + s = get_settings() + try: + async with httpx.AsyncClient(timeout=4.0) as c: + r = await c.get( + f"{s.USER_SERVICE_BASE_URL}/api/state/{clerk_id}", + headers={"Authorization": f"Bearer {s.A2A_OUTBOUND_KEY}"}, + ) + r.raise_for_status() + streak = (r.json().get("metadata") or {}).get("current_streak") + return int(streak) if streak is not None else None + except (httpx.HTTPError, ValueError, TypeError) as e: + logger.warning("fetch_streak failed (non-fatal): %s", e) + return None diff --git a/tests/test_stats.py b/tests/test_stats.py new file mode 100644 index 0000000..bf6e1b1 --- /dev/null +++ b/tests/test_stats.py @@ -0,0 +1,33 @@ +"""Dashboard stats — pure-helper contract tests (no DB / no network).""" +from app.engine import stats as S +from app.engine.normalize import _posted_date +from app.db.repo import _activity_score + + +def test_posted_date_per_board(): + assert _posted_date({"createdDate": "2026-06-18T10:37:17Z"}) == "2026-06-18T10:37:17Z" # Naukri + assert _posted_date({"postedDate": "2026-06-18T12:46:39Z"}) == "2026-06-18T12:46:39Z" # LinkedIn + assert _posted_date({"date_posted": "2026-06-17T17:38:14Z"}) == "2026-06-17T17:38:14Z" # Foundit + assert _posted_date({"jobDetails": {"createdDate": "2026-06-18T00:00:00Z"}}) is not None # nested + assert _posted_date({"nope": "x"}) is None + + +def test_match_scores_and_histogram(): + opps = [{"match": {"score": 84}}, {"matchScore": 72}, {"match": {"score": 0}}, {}] + scores = S._match_scores(opps) + assert sorted(scores) == [72, 84] # 0 and missing dropped + h = S._histogram(scores) + assert sum(h) == 2 and len(h) == 11 + + +def test_apply_window_freshness(): + # all None when no posting dates (honest — never fabricated) + assert S._apply_window([{}, {"posted_date": None}])["urgency"] is None + w = S._apply_window([{"posted_date": "2026-06-19T00:00:00Z"}, {"posted_date": "2026-06-15T00:00:00Z"}]) + assert w["dated"] == 2 and w["median_age_days"] is not None and w["urgency"] in ("high", "medium", "low") + + +def test_activity_score_weighting(): + # applies weighted highest, then saves, views, searches + assert _activity_score(0, 0, 1, 0) > _activity_score(0, 1, 0, 0) > _activity_score(1, 0, 0, 0) + assert _activity_score(0, 0, 0, 0) == 0