From 13d5b9644611474f502d291c45d34e7e03293ec1 Mon Sep 17 00:00:00 2001 From: Shawn Tretter Date: Tue, 7 Jul 2026 14:57:45 -0600 Subject: [PATCH 1/2] feat(sources): add `ats` source for live Greenhouse/Ashby role feeds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `watchlist_careers` scraper cannot read JavaScript-rendered career pages, so most modern boards yield no individual postings. This adds an `ats` source that reads structured job data directly from the public Greenhouse and Ashby JSON APIs. For each watchlist company with an `ats: {provider, token}` block, the adapter fetches that board and emits one FetchedItem per posting with `evidence_type="job"`, so real role titles flow into scoring and can reach the Act Now tier. Optional per-source filters: - `title_includes` / `title_excludes`: keep/drop postings by title term. - `us_only`: keep only US-based or remote-eligible roles, classified from structured country data (Ashby `addressCountry`, Greenhouse trailing location token) via exact token match — US cities like Milwaukee or Indianapolis are never misread as foreign. - `location_block`: substring blocklist on the display location. - `max_jobs_per_board`: cap postings per company. Additive and resilient: existing adapters and `scoring.py` are unchanged, a dead token or malformed job is skipped without failing the run, and the new source type is registered in `doctor` validation. README documents the config; adds network-free tests for parsing and country classification. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 53 ++++++++++ pathscout/doctor.py | 4 +- pathscout/fetchers.py | 220 ++++++++++++++++++++++++++++++++++++++++++ tests/test_ats.py | 161 +++++++++++++++++++++++++++++++ 4 files changed, 436 insertions(+), 2 deletions(-) create mode 100644 tests/test_ats.py diff --git a/README.md b/README.md index 5fc514f..cdc5dc8 100644 --- a/README.md +++ b/README.md @@ -162,9 +162,62 @@ The v0.2 runner supports standard-library fetches for: - `portfolio`: turns companies from `config/portfolio.json` into relationship-context observations. - `web_page`: fetches a single web page. - `rss`: fetches an RSS or Atom feed. +- `ats`: pulls live role postings directly from applicant-tracking-system JSON APIs (Greenhouse and Ashby) for watchlist companies that declare an `ats` block. `radar_portfolio` remains as a deprecated alias for one release. Use `portfolio` for new config. +### ATS live role feeds + +Most modern career pages are JavaScript-rendered, so the standard-library `watchlist_careers` +scraper cannot see individual postings. The `ats` source reads structured job data straight from +the public JSON boards that many companies already publish, so real role titles flow into scoring +(each posting is emitted with `evidence_type: "job"`, making it eligible for the `Act Now` tier). + +Add an `ats` block to any watchlist company: + +```json +{ + "name": "Example AI", + "status": "strong", + "domains": ["ai"], + "urls": { "homepage": "https://example.ai" }, + "ats": { "provider": "greenhouse", "token": "exampleai" } +} +``` + +- `provider`: `greenhouse` (`boards-api.greenhouse.io/v1/boards//jobs`) or `ashby` + (`api.ashbyhq.com/posting-api/job-board/`). +- `token`: the board slug/org name in that provider's public URL. + +Then enable the source: + +```json +{ + "id": "ats_live_roles", + "type": "ats", + "name": "Live ATS role feeds", + "enabled": true, + "config": { + "path": "config/watchlist.json", + "max_jobs_per_board": 60, + "title_includes": ["sales", "solutions engineer", "forward deployed"], + "title_excludes": ["software engineer", "recruiter"], + "us_only": true, + "location_block": ["london", "singapore"] + } +} +``` + +Optional filters (all applied per posting): + +- `title_includes`: keep only postings whose title contains one of these terms. +- `title_excludes`: drop postings whose title contains one of these terms. +- `us_only`: when true, keep only US-based or remote-eligible roles, classified from structured + country data (Ashby `addressCountry`, Greenhouse trailing location token) via exact token match + so US cities are never misread as foreign. +- `location_block`: additional substring blocklist applied to the display location. +- `max_jobs_per_board`: cap postings emitted per company (0 = no cap). + ## Commands ```bash diff --git a/pathscout/doctor.py b/pathscout/doctor.py index 54fe1de..447e9ce 100644 --- a/pathscout/doctor.py +++ b/pathscout/doctor.py @@ -10,7 +10,7 @@ from .watchlist import summarize_watchlist -SUPPORTED_SOURCES = {"manual", "watchlist", "watchlist_careers", "portfolio", "radar_portfolio", "web_page", "rss"} +SUPPORTED_SOURCES = {"manual", "watchlist", "watchlist_careers", "portfolio", "radar_portfolio", "web_page", "rss", "ats"} def validate_setup( @@ -113,7 +113,7 @@ def validate_sources(sources: list[dict[str, Any]], warnings: list[str], errors: warnings.append(f"{source_id} uses deprecated source type radar_portfolio; use portfolio") if source.get("enabled", True): active_sources.append(source) - if source_type in {"watchlist", "watchlist_careers", "portfolio", "radar_portfolio"}: + if source_type in {"watchlist", "watchlist_careers", "portfolio", "radar_portfolio", "ats"}: path = Path(source_setting(source, "path", "")) if path and not path.exists(): warnings.append(f"{source_id} references missing source file: {path}") diff --git a/pathscout/fetchers.py b/pathscout/fetchers.py index 862288a..5d968d1 100644 --- a/pathscout/fetchers.py +++ b/pathscout/fetchers.py @@ -20,6 +20,40 @@ logger = logging.getLogger("pathscout.fetchers") USER_AGENT = "PathScout role-discovery-radar/0.2" + +# Exact-match tokens for US-vs-foreign job-location classification (used by location_is_us). +US_STATE_TOKENS = { + "united states", "united states of america", "usa", "us", "u s a", + "al", "ak", "az", "ar", "ca", "co", "ct", "de", "fl", "ga", "hi", "id", "il", "in", "ia", + "ks", "ky", "la", "me", "md", "ma", "mi", "mn", "ms", "mo", "mt", "ne", "nv", "nh", "nj", + "nm", "ny", "nc", "nd", "oh", "ok", "or", "pa", "ri", "sc", "sd", "tn", "tx", "ut", "vt", + "va", "wa", "wv", "wi", "wy", "dc", + "alabama", "alaska", "arizona", "arkansas", "california", "colorado", "connecticut", + "delaware", "florida", "georgia", "hawaii", "idaho", "illinois", "indiana", "iowa", + "kansas", "kentucky", "louisiana", "maine", "maryland", "massachusetts", "michigan", + "minnesota", "mississippi", "missouri", "montana", "nebraska", "nevada", "new hampshire", + "new jersey", "new mexico", "new york", "north carolina", "north dakota", "ohio", + "oklahoma", "oregon", "pennsylvania", "rhode island", "south carolina", "south dakota", + "tennessee", "texas", "utah", "vermont", "virginia", "washington", "west virginia", + "wisconsin", "wyoming", "district of columbia", +} +FOREIGN_TOKENS = { + "india", "japan", "singapore", "australia", "new zealand", "canada", "mexico", "brazil", + "brasil", "united kingdom", "uk", "england", "scotland", "wales", "ireland", "france", + "germany", "spain", "italy", "netherlands", "belgium", "switzerland", "sweden", "norway", + "denmark", "finland", "poland", "portugal", "austria", "greece", "czech republic", "czechia", + "romania", "hungary", "turkey", "israel", "united arab emirates", "uae", "saudi arabia", + "qatar", "egypt", "south africa", "nigeria", "kenya", "south korea", "korea", "china", + "hong kong", "taiwan", "philippines", "indonesia", "thailand", "vietnam", "malaysia", + "sri lanka", "pakistan", "bangladesh", "argentina", "chile", "colombia", "peru", + "costa rica", "uruguay", "luxembourg", "monaco", + # major intl cities as a backup when no country token is present + "london", "munich", "berlin", "frankfurt", "paris", "dublin", "toronto", "vancouver", + "sydney", "melbourne", "bengaluru", "bangalore", "mumbai", "hyderabad", "seoul", "tokyo", + "tel aviv", "amsterdam", "zurich", "madrid", "barcelona", "milan", "stockholm", "dubai", + "sao paulo", "mexico city", "warsaw", "lisbon", "copenhagen", "oslo", +} + ROLE_TITLE_TERMS = [ "chief", "svp", @@ -109,6 +143,8 @@ def fetch_source( return fetch_web_page(source, cache=cache) if source_type == "rss": return fetch_rss(source, cache=cache) + if source_type == "ats": + return fetch_ats(source, cache=cache) raise ValueError(f"Unsupported source type: {source_type}") @@ -335,6 +371,190 @@ def fetch_rss(source: dict[str, Any], cache: ResponseCache | None = None) -> lis return items +def fetch_ats(source: dict[str, Any], cache: ResponseCache | None = None) -> list[FetchedItem]: + """Pull live role postings from applicant-tracking-system (ATS) JSON APIs. + + Reads the watchlist and, for each company carrying an ``ats`` block of the + form ``{"provider": "greenhouse"|"ashby", "token": "..."}``, fetches that + board's public JSON job feed and emits one FetchedItem per posting. Unlike + the HTML careers scraper (which is blind to JS-rendered boards), this reads + structured job data directly, so real role titles flow into scoring with + ``evidence_type="job"`` — making them eligible for the Act Now tier. + + Optional config: ``title_includes`` (only emit jobs whose title contains one + of these terms), ``max_jobs_per_board`` (cap postings per company). + """ + path = source_setting(source, "path", "config/watchlist.json") + with open(path, "r", encoding="utf-8") as handle: + watchlist = json.load(handle) + title_includes = [normalize_token(t) for t in (source_setting(source, "title_includes", []) or [])] + title_excludes = [normalize_token(t) for t in (source_setting(source, "title_excludes", []) or [])] + location_block = [normalize_token(t) for t in (source_setting(source, "location_block", []) or [])] + us_only = bool(source_setting(source, "us_only", False)) + max_per_board = int(source_setting(source, "max_jobs_per_board", 0) or 0) + timeout = int(source_setting(source, "timeout_seconds", 12)) + items: list[FetchedItem] = [] + for company in watchlist.get("companies", []): + if company.get("status", "watch") in {"exclude", "archive"}: + continue + ats = company.get("ats") or {} + provider = (ats.get("provider") or "").strip().lower() + token = (ats.get("token") or "").strip() + if not provider or not token: + continue + if provider == "greenhouse": + url = f"https://boards-api.greenhouse.io/v1/boards/{token}/jobs?content=true" + elif provider == "ashby": + url = f"https://api.ashbyhq.com/posting-api/job-board/{token}?includeCompensation=true" + else: + logger.warning("Unknown ATS provider %r for %s", provider, company.get("name", "unknown")) + continue + try: + body = http_get(url, timeout=timeout, cache=cache) + jobs = json.loads(body).get("jobs", []) or [] + except Exception as exc: + logger.warning( + "ATS fetch failed for %s (%s/%s): %s", company.get("name", "unknown"), provider, token, exc + ) + continue + emitted = 0 + for job in jobs: + try: + parsed = parse_ats_job(provider, job) + except Exception: + continue + if not parsed: + continue + title = parsed["title"] + if not title: + continue + job_url, location, detail = parsed["url"], parsed["location"], parsed["detail"] + title_norm = normalize_token(title) + if title_includes and not any(term in title_norm for term in title_includes): + continue + if title_excludes and any(term in title_norm for term in title_excludes): + continue + if us_only and not location_is_us(parsed["countries"], location): + continue + if location_block and any(term in normalize_token(location) for term in location_block): + continue + text_parts = [ + title, + location, + detail, + company.get("stage", ""), + " ".join(company.get("domains", [])), + company.get("watch_reason", ""), + ] + items.append( + FetchedItem( + source_id=source["id"], + source_name=source.get("name", source["id"]), + source_type="ats", + company=company.get("name", ""), + title=title, + url=job_url, + text=" ".join(part for part in text_parts if part), + evidence_type="job", + ) + ) + emitted += 1 + if max_per_board and emitted >= max_per_board: + break + return items + + +def parse_ats_job(provider: str, job: dict[str, Any]) -> dict[str, Any] | None: + """Return a parsed job dict (title/url/location/detail/countries) or None if unusable. + + ``countries`` holds structured/derived country-or-US-state tokens used for reliable + US-vs-foreign filtering (exact token match, not fragile substring search). + """ + if provider == "greenhouse": + title = (job.get("title") or "").strip() + job_url = job.get("absolute_url", "") or "" + location = (job.get("location") or {}).get("name", "") or "" + departments = ", ".join( + d.get("name", "") for d in (job.get("departments") or []) if isinstance(d, dict) and d.get("name") + ) + content = html_to_text(unescape(job.get("content", "") or "")) + detail = " ".join(part for part in [departments, content[:400]] if part) + # Greenhouse encodes country/state as the trailing comma-segment of each location name. + names = [location] + [o.get("name", "") for o in (job.get("offices") or []) if isinstance(o, dict)] + countries = _derive_country_tokens(names) + return {"title": title, "url": job_url, "location": location, "detail": detail, "countries": countries} + if provider == "ashby": + title = (job.get("title") or "").strip() + job_url = job.get("jobUrl") or job.get("applyUrl") or "" + locs = [job.get("location", "") or ""] + countries: list[str] = [] + primary_country = ((job.get("address") or {}).get("postalAddress") or {}).get("addressCountry") + if primary_country: + countries.append(primary_country) + for sec in job.get("secondaryLocations") or []: + if not isinstance(sec, dict): + continue + name = sec.get("location") + if name: + locs.append(name) + sec_country = ((sec.get("address") or {}).get("postalAddress") or {}).get("addressCountry") + if sec_country: + countries.append(sec_country) + if job.get("isRemote") or (job.get("workplaceType", "") or "").lower() == "remote": + locs.append("Remote") + location = ", ".join(dict.fromkeys(part for part in locs if part)) + team = job.get("team") or job.get("department") or job.get("departmentName") or "" + comp = "" + comp_obj = job.get("compensation") or {} + if isinstance(comp_obj, dict): + comp = comp_obj.get("compensationTierSummary") or comp_obj.get("scrapeableCompensationSalarySummary") or "" + desc = (job.get("descriptionPlain") or "")[:400] + detail = " ".join(part for part in [team, job.get("employmentType", ""), comp or "", desc] if part) + return {"title": title, "url": job_url, "location": location, "detail": detail, "countries": countries} + return None + + +def _derive_country_tokens(names: list[str]) -> list[str]: + """From display location names, extract the trailing token of each comma segment. + + "San Francisco, CA" -> "ca"; "Bengaluru, India; Mumbai, India" -> "india"; + "Singapore" -> "singapore". Used for exact-match US-vs-foreign classification. + """ + tokens: list[str] = [] + for name in names: + if not name: + continue + for segment in re.split(r"[;|]", name): + parts = [p.strip() for p in segment.split(",") if p.strip()] + if parts: + tokens.append(normalize_token(parts[-1])) + return tokens + + +def location_is_us(countries: list[str], display: str) -> bool: + """True if a job is US-based/remote-eligible; False only when a clear foreign token exists. + + Classification is exact token equality on structured/derived country tokens, so US cities + (Milwaukee, Indianapolis) can never be misread as foreign. Ambiguous/generic 'Remote' with + no country signal is kept. + """ + tokens = [normalize_token(c) for c in countries if c] + has_us = any(t in US_STATE_TOKENS for t in tokens) + has_foreign = any(t in FOREIGN_TOKENS for t in tokens) + if has_us: + return True + if has_foreign: + return False + display_norm = normalize_token(display) + if any(marker in display_norm for marker in ("united states", "remote us", "us remote", "usa")): + return True + return True + + +def normalize_token(value: str) -> str: + return " ".join((value or "").lower().replace("/", " ").replace("-", " ").split()) + + def source_setting(source: dict[str, Any], key: str, default: Any = None) -> Any: if key in source: return source[key] diff --git a/tests/test_ats.py b/tests/test_ats.py new file mode 100644 index 0000000..2d7c7e9 --- /dev/null +++ b/tests/test_ats.py @@ -0,0 +1,161 @@ +import json +import os +import tempfile +import unittest +from unittest import mock + +from pathscout import fetchers +from pathscout.fetchers import ( + _derive_country_tokens, + fetch_ats, + location_is_us, + parse_ats_job, +) + + +GREENHOUSE_RESPONSE = { + "jobs": [ + { + "title": "Enterprise Account Executive", + "absolute_url": "https://boards.example/gh/1", + "location": {"name": "San Francisco, CA | New York City, NY"}, + "departments": [{"name": "Sales"}], + "offices": [{"name": "San Francisco, CA"}], + "content": "<p>Own strategic accounts</p>", + }, + { + "title": "Software Engineer, Backend", + "absolute_url": "https://boards.example/gh/2", + "location": {"name": "Remote"}, + "content": "", + }, + { + "title": "Sales Manager", + "absolute_url": "https://boards.example/gh/3", + "location": {"name": "London, UK"}, + "content": "", + }, + ] +} + +ASHBY_RESPONSE = { + "jobs": [ + { + "title": "Solutions Architect", + "jobUrl": "https://jobs.example/ashby/1", + "location": "Denver", + "isRemote": True, + "team": "Sales", + "address": {"postalAddress": {"addressCountry": "United States"}}, + }, + { + "title": "Regional Vice President of Sales - India", + "jobUrl": "https://jobs.example/ashby/2", + "location": "India", + "isRemote": True, + "address": {"postalAddress": {"addressCountry": "India"}}, + }, + ] +} + + +class ParseAtsJobTests(unittest.TestCase): + def test_greenhouse_parse_extracts_fields_and_country(self): + parsed = parse_ats_job("greenhouse", GREENHOUSE_RESPONSE["jobs"][0]) + self.assertEqual(parsed["title"], "Enterprise Account Executive") + self.assertEqual(parsed["url"], "https://boards.example/gh/1") + self.assertIn("San Francisco", parsed["location"]) + self.assertIn("ca", parsed["countries"]) + self.assertIn("Own strategic accounts", parsed["detail"]) + + def test_ashby_parse_uses_structured_country(self): + parsed = parse_ats_job("ashby", ASHBY_RESPONSE["jobs"][1]) + self.assertEqual(parsed["title"], "Regional Vice President of Sales - India") + self.assertEqual(parsed["countries"], ["India"]) + + def test_unknown_provider_returns_none(self): + self.assertIsNone(parse_ats_job("lever", {"title": "x"})) + + +class DeriveCountryTokensTests(unittest.TestCase): + def test_us_state_and_foreign_and_citystate(self): + self.assertEqual(_derive_country_tokens(["San Francisco, CA"]), ["ca"]) + self.assertEqual( + _derive_country_tokens(["Bengaluru, India; Mumbai, India"]), ["india", "india"] + ) + self.assertEqual(_derive_country_tokens(["Singapore"]), ["singapore"]) + self.assertEqual( + _derive_country_tokens(["San Francisco, CA | New York City, NY"]), ["ca", "ny"] + ) + + +class LocationIsUsTests(unittest.TestCase): + def test_us_country_and_states_kept(self): + self.assertTrue(location_is_us(["United States"], "San Francisco")) + self.assertTrue(location_is_us(["ca", "ny"], "SF | NYC")) + self.assertTrue(location_is_us(["co"], "Denver, CO")) + + def test_foreign_dropped(self): + self.assertFalse(location_is_us(["India"], "India")) + self.assertFalse(location_is_us(["United Kingdom"], "London")) + self.assertFalse(location_is_us(["japan"], "Tokyo, Japan")) + self.assertFalse(location_is_us(["singapore"], "Singapore")) + + def test_no_substring_false_positives(self): + # "Milwaukee" contains "uk"; "Indianapolis" contains "india" — exact-token match is safe. + self.assertTrue(location_is_us(["wi"], "Milwaukee, WI")) + self.assertTrue(location_is_us(["in"], "Indianapolis, IN")) + + def test_generic_remote_kept(self): + self.assertTrue(location_is_us([], "Remote")) + + +class FetchAtsFilterTests(unittest.TestCase): + def setUp(self): + watchlist = { + "schema_version": 1, + "companies": [ + {"name": "GH Co", "status": "strong", "ats": {"provider": "greenhouse", "token": "ghco"}}, + {"name": "Ashby Co", "status": "watch", "ats": {"provider": "ashby", "token": "ashco"}}, + ], + } + handle = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) + json.dump(watchlist, handle) + handle.close() + self.watchlist_path = handle.name + + def tearDown(self): + os.unlink(self.watchlist_path) + + def _fake_http_get(self, url, *args, **kwargs): + if "greenhouse" in url: + return json.dumps(GREENHOUSE_RESPONSE) + if "ashby" in url: + return json.dumps(ASHBY_RESPONSE) + raise AssertionError("unexpected url: " + url) + + def test_filters_by_title_and_us_only(self): + source = { + "id": "ats_live", + "name": "Live ATS", + "type": "ats", + "config": { + "path": self.watchlist_path, + "title_includes": ["sales", "account exec", "solutions architect"], + "title_excludes": ["software engineer"], + "us_only": True, + }, + } + with mock.patch.object(fetchers, "http_get", side_effect=self._fake_http_get): + items = fetch_ats(source) + titles = sorted(item.title for item in items) + # Kept: US account exec (GH) + US solutions architect (Ashby). + # Dropped: Software Engineer (title_excludes), Sales Manager London (foreign), + # Regional VP Sales India (foreign). + self.assertEqual(titles, ["Enterprise Account Executive", "Solutions Architect"]) + self.assertTrue(all(item.evidence_type == "job" for item in items)) + self.assertTrue(all(item.source_type == "ats" for item in items)) + + +if __name__ == "__main__": + unittest.main() From 7eb756d5174246276aba4a3e9b6fcb1c8792a89c Mon Sep 17 00:00:00 2001 From: Shawn Tretter Date: Tue, 7 Jul 2026 16:00:19 -0600 Subject: [PATCH 2/2] feat(sources): add Lever as a third ATS provider Extends the `ats` source to read Lever's public postings API (api.lever.co/v0/postings/?mode=json) alongside Greenhouse and Ashby. Lever returns a bare JSON list (not {"jobs": [...]}) and exposes an ISO-3166 alpha-2 `country` field, which is classified directly (US -> kept; any other -> foreign) rather than via the location-token path, since ISO codes collide with US state abbreviations (Lever "CA" means Canada, not California). README documents the new provider; adds tests for Lever parsing and the ISO/state collision case. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 7 ++++--- pathscout/fetchers.py | 33 ++++++++++++++++++++++++++++++++- tests/test_ats.py | 35 ++++++++++++++++++++++++++++++++++- 3 files changed, 70 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index cdc5dc8..aacb49c 100644 --- a/README.md +++ b/README.md @@ -162,7 +162,7 @@ The v0.2 runner supports standard-library fetches for: - `portfolio`: turns companies from `config/portfolio.json` into relationship-context observations. - `web_page`: fetches a single web page. - `rss`: fetches an RSS or Atom feed. -- `ats`: pulls live role postings directly from applicant-tracking-system JSON APIs (Greenhouse and Ashby) for watchlist companies that declare an `ats` block. +- `ats`: pulls live role postings directly from applicant-tracking-system JSON APIs (Greenhouse, Ashby, and Lever) for watchlist companies that declare an `ats` block. `radar_portfolio` remains as a deprecated alias for one release. Use `portfolio` for new config. @@ -185,8 +185,9 @@ Add an `ats` block to any watchlist company: } ``` -- `provider`: `greenhouse` (`boards-api.greenhouse.io/v1/boards//jobs`) or `ashby` - (`api.ashbyhq.com/posting-api/job-board/`). +- `provider`: `greenhouse` (`boards-api.greenhouse.io/v1/boards//jobs`), `ashby` + (`api.ashbyhq.com/posting-api/job-board/`), or `lever` + (`api.lever.co/v0/postings/?mode=json`). - `token`: the board slug/org name in that provider's public URL. Then enable the source: diff --git a/pathscout/fetchers.py b/pathscout/fetchers.py index 5d968d1..c2c0241 100644 --- a/pathscout/fetchers.py +++ b/pathscout/fetchers.py @@ -52,6 +52,8 @@ "sydney", "melbourne", "bengaluru", "bangalore", "mumbai", "hyderabad", "seoul", "tokyo", "tel aviv", "amsterdam", "zurich", "madrid", "barcelona", "milan", "stockholm", "dubai", "sao paulo", "mexico city", "warsaw", "lisbon", "copenhagen", "oslo", + # sentinel appended for any non-US ISO country code (e.g. Lever's `country` field) + "__nonus__", } ROLE_TITLE_TERMS = [ @@ -406,12 +408,16 @@ def fetch_ats(source: dict[str, Any], cache: ResponseCache | None = None) -> lis url = f"https://boards-api.greenhouse.io/v1/boards/{token}/jobs?content=true" elif provider == "ashby": url = f"https://api.ashbyhq.com/posting-api/job-board/{token}?includeCompensation=true" + elif provider == "lever": + url = f"https://api.lever.co/v0/postings/{token}?mode=json" else: logger.warning("Unknown ATS provider %r for %s", provider, company.get("name", "unknown")) continue try: body = http_get(url, timeout=timeout, cache=cache) - jobs = json.loads(body).get("jobs", []) or [] + data = json.loads(body) + # Greenhouse/Ashby wrap postings in {"jobs": [...]}; Lever returns a bare list. + jobs = data if isinstance(data, list) else (data.get("jobs", []) or []) except Exception as exc: logger.warning( "ATS fetch failed for %s (%s/%s): %s", company.get("name", "unknown"), provider, token, exc @@ -511,6 +517,31 @@ def parse_ats_job(provider: str, job: dict[str, Any]) -> dict[str, Any] | None: desc = (job.get("descriptionPlain") or "")[:400] detail = " ".join(part for part in [team, job.get("employmentType", ""), comp or "", desc] if part) return {"title": title, "url": job_url, "location": location, "detail": detail, "countries": countries} + if provider == "lever": + title = (job.get("text") or "").strip() + job_url = job.get("hostedUrl") or job.get("applyUrl") or "" + cats = job.get("categories") or {} + locs = [] + if cats.get("location"): + locs.append(cats["location"]) + for loc in cats.get("allLocations") or []: + if loc: + locs.append(loc) + if (job.get("workplaceType") or "").lower() == "remote": + locs.append("Remote") + location = ", ".join(dict.fromkeys(locs)) + # Lever's `country` is an ISO-2 code. Classify it directly (US vs foreign) rather than + # via the state-token path, because ISO codes collide with US state abbrevs (CA, IN, CO). + countries = _derive_country_tokens(locs) + country = (job.get("country") or "").strip().upper() + if country in ("US", "USA"): + countries.append("united states") + elif country: + countries.append("__nonus__") + team = cats.get("team") or cats.get("department") or "" + desc = (job.get("descriptionPlain") or "")[:400] + detail = " ".join(part for part in [team, cats.get("commitment", ""), desc] if part) + return {"title": title, "url": job_url, "location": location, "detail": detail, "countries": countries} return None diff --git a/tests/test_ats.py b/tests/test_ats.py index 2d7c7e9..d5fef2f 100644 --- a/tests/test_ats.py +++ b/tests/test_ats.py @@ -73,8 +73,41 @@ def test_ashby_parse_uses_structured_country(self): self.assertEqual(parsed["title"], "Regional Vice President of Sales - India") self.assertEqual(parsed["countries"], ["India"]) + def test_lever_parse_us_job(self): + job = { + "text": "Enterprise Account Executive", + "hostedUrl": "https://jobs.lever.co/acme/1", + "country": "US", + "workplaceType": "remote", + "categories": {"location": "New York", "team": "Sales", "commitment": "Full-time"}, + "descriptionPlain": "Sell to enterprise accounts", + } + parsed = parse_ats_job("lever", job) + self.assertEqual(parsed["title"], "Enterprise Account Executive") + self.assertEqual(parsed["url"], "https://jobs.lever.co/acme/1") + self.assertIn("united states", parsed["countries"]) + self.assertTrue(location_is_us(parsed["countries"], parsed["location"])) + + def test_lever_parse_foreign_job_iso_classified(self): + job = { + "text": "Account Executive", + "hostedUrl": "https://jobs.lever.co/acme/2", + "country": "FR", + "categories": {"location": "Paris", "team": "Business"}, + } + parsed = parse_ats_job("lever", job) + # ISO "FR" -> foreign sentinel, so a Paris role is correctly dropped by us_only. + self.assertIn("__nonus__", parsed["countries"]) + self.assertFalse(location_is_us(parsed["countries"], parsed["location"])) + + def test_iso_state_collision_safe(self): + # Lever country "CA" means Canada, not California — must be treated as foreign. + job = {"text": "AE", "country": "CA", "categories": {"location": "Toronto"}} + parsed = parse_ats_job("lever", job) + self.assertFalse(location_is_us(parsed["countries"], parsed["location"])) + def test_unknown_provider_returns_none(self): - self.assertIsNone(parse_ats_job("lever", {"title": "x"})) + self.assertIsNone(parse_ats_job("workday", {"title": "x"})) class DeriveCountryTokensTests(unittest.TestCase):