b1e9cdd182
- A1: fire the hard filter (Apify is precise); embedding vibe-ranker (embed.py) blends with the floor-free white-box (rank.py) to sift ~90 -> top 18 - A2: Opus curator (curate.py) reads the 18 -> honest <=count + Gemini-voiced report cards; graceful fallback to the white-box + templated cards - LLM via opencode.ai/zen gateway (llm.py): Opus chat + direct-OpenAI embeddings - Run LLM work off the event loop (asyncio.to_thread) so the Redis response publishes - C1: dedicated Postgres (app/db/) persists the per-user feed; get_scout_feed replays it so matches survive navigation/refresh - match contract + report-card fields (schema.py); skills.py; tests/
76 lines
3.2 KiB
Python
76 lines
3.2 KiB
Python
"""Phase 1 — factor primitives + end-to-end shortlist on CACHED data ($0)."""
|
|
import glob
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from app.engine import normalize as N
|
|
from app.engine import rank as R
|
|
|
|
_NORM = {"blackfalcondata": N.naukri_feed_to_scoutjob, "shahidirfan": N.foundit_to_scoutjob,
|
|
"harvestapi": N.linkedin_to_scoutjob}
|
|
|
|
|
|
def _cached_jobs() -> list[dict]:
|
|
jobs = []
|
|
for f in glob.glob(".apify_cache/*.json"):
|
|
fn = _NORM.get(f.split("/")[-1].split("_")[0])
|
|
if not fn:
|
|
continue
|
|
for it in json.load(open(f)):
|
|
sj = fn(it)
|
|
if sj:
|
|
jobs.append(sj)
|
|
return jobs
|
|
|
|
|
|
# ── factor primitives ──
|
|
def test_location_gate_and_tiers():
|
|
p = R.profile_from({"title": "PM", "location": ["New Delhi · India"]})
|
|
assert R.f_location(p, {"location_city": "New Delhi", "location_country": "India"}) == 1.0
|
|
assert R.f_location(p, {"location_city": "Bengaluru", "location_country": "India"}) == 0.6
|
|
assert R.f_location(p, {"location_city": "London", "location_country": "UK"}) == 0.0
|
|
assert R.f_location(p, {"location_mode": "remote"}) == 1.0
|
|
p_open = R.profile_from({"title": "PM"})
|
|
assert R.f_location(p_open, {"location_city": "X", "location_country": "Y"}) is None # Anywhere → absent
|
|
|
|
|
|
def test_experience_band():
|
|
p = R.profile_from({"title": "PM", "years": 5})
|
|
assert R.f_experience(p, {"seniority_level": "mid"}) == 1.0 # 5 in [3,8]
|
|
assert R.f_experience(p, {"seniority_level": "lead"}) < 1.0 # under-qualified
|
|
assert R.f_experience(p, {"seniority_level": None}) is None # absent
|
|
|
|
|
|
def test_e2e_ranks_all_gate_passing_no_early_cut():
|
|
jobs = _cached_jobs()
|
|
if len(jobs) < 30:
|
|
pytest.skip("no cached sweep data — run a search first")
|
|
prefs = {"title": "Product Manager", "role": ["Product"], "location": ["New Delhi · India"],
|
|
"experience": ["Mid-level"], "stretch": "balanced"}
|
|
res = R.rank(prefs, None, jobs)
|
|
opp = res["opportunities"]
|
|
# the cheap white-box does NOT cut to 25 — it ranks ALL gate-passing jobs (cut is Phase 3)
|
|
assert len(opp) >= 1
|
|
assert res["ranked"] == len(opp) and res["scanned"] == len(jobs)
|
|
scores = [j["matchScore"] for j in opp]
|
|
assert scores == sorted(scores, reverse=True) # ranked best-first
|
|
for j in opp:
|
|
m = j["match"]
|
|
assert m["fit"] in ("fit", "stretch") and isinstance(m["breakdown"], list)
|
|
assert m["score"] == j["matchScore"]
|
|
# determinism
|
|
res2 = R.rank(prefs, None, _cached_jobs())
|
|
assert [j["id"] for j in res2["opportunities"]] == [j["id"] for j in opp]
|
|
|
|
|
|
def test_select_is_separate_and_capped():
|
|
# the final cut (Phase 3) — caps at ~25, floor so never empty, by fused matchScore
|
|
ranked = [{"id": str(i), "matchScore": 100 - i} for i in range(80)]
|
|
sel = R.select(ranked, "balanced")
|
|
assert 1 <= len(sel) <= R.CAP
|
|
assert [j["matchScore"] for j in sel] == sorted((j["matchScore"] for j in sel), reverse=True)
|
|
# floor: even if nothing clears the bar, return at least `floor`
|
|
low = [{"id": str(i), "matchScore": 10} for i in range(40)]
|
|
assert len(R.select(low, "safe")) == R.FLOOR
|