" # aria-label link; location lives in the card text (captured via blob mode). ("google", "Google", "playwright", { "url": "https://www.google.com/about/careers/applications/jobs/results/?location=Switzerland", "wait_for": "a[href*='jobs/results/'][aria-label*='Learn more']", "card": "li:has(a[aria-label*='Learn more about'])", "title_sel": "a[aria-label*='Learn more about']", "title_sel_attr": "aria-label", "title_strip_prefix": "Learn more about ", "link_sel": "a[href*='jobs/results/']", "link_attr": "href", "url_prefix": "https://www.google.com/about/careers/applications/", "default_location": "", "scroll_count": 5, "use_inner_text_as_blob": True, "cookie_accept": ["button:has-text('Accept all')", "button:has-text('Reject all')"], # Results are paged 20 at a time and "page" is 1-indexed — without this only the # first 20 of ~48 Zürich roles were ever scraped. "page_param": "page", "page_param_start": 2, "max_pages": 6, }), # Amazon + AWS share one board. Added 2026-08-18: AWS is the cloud the evidence base # actually supports (Swisscom migration + data products), whereas claims.json marks GCP # output-forbidden — so an AWS-native Zurich FDE req is the Google FDE lane without the # cloud gap. 32 CH roles at time of adding, incl. "Senior Forward Deployed Engineer, AWS # Forward Deployed Engineering" (Zurich). # AWS names its delivery/field org differently from every other board, so the shared # filter caught only 1 of 32 CH roles. These four terms are added narrowly and ONLY here: # "solutions architect" is the classic overscoring trap (see job_scout_overscoring_findings), # but on the AWS board it is the delivery-side title attached to migration/data work, which # is exactly SW-1. Read the JD before trusting the score on any of these. ("amazon", "Amazon / AWS", "amazon", { "countries": ["CHE"], "_title_exclude": NOISE_TITLE_EXCLUDE, }), ("apple", "Apple", "playwright", { "url": "https://jobs.apple.com/en-us/search?location=switzerland-CHE", "wait_for": "a[href*='/en-us/details/']", "card": "a[href*='/en-us/details/']", "title_attr": "text", "link_attr": "href", "url_prefix": "https://jobs.apple.com", # NO default_location. Apple's own postLocation-CHE filter leaks global reqs whose # postLocationId is postLocation-USA ("Various Locations within United States"), and # a forced "Switzerland" default relabelled every one of them as CH-eligible — 84 # phantom CH rows over 4 runs, all internships/retail, which then read as "Apple is a # bad fit" in the decision log. Read the real location off the card instead; when # Apple has no Swiss reqs the honest answer is 0. Verified 2026-08-18. "use_inner_text_as_blob": True, "scroll_count": 5, }), # Meta job links are /profile/job_details/<id>; title + location are in the link text. # NOT broken despite returning ~1 role: verified 2026-08-18 that metacareers itself # reports "1 Items" for the Zurich office filter. The board is genuinely near-empty; # don't "fix" this scraper without first checking the live item count on the page. ("meta", "Meta", "playwright", { "url": "https://www.metacareers.com/jobs?offices[0]=Zurich%2C%20Switzerland", "wait_for": "a[href*='/profile/job_details/']", "card": "a[href*='/profile/job_details/']", "title_attr": "text", "link_attr": "href", "url_prefix": "https://www.metacareers.com", "default_location": "Zurich, Switzerland", "scroll_count": 5, "use_inner_text_as_blob": True, }), # Roche: was a playwright scrape of the search-results page until 2026-08-18. That page's # ?locationsearch=Switzerland filter silently fails — the page state reports "no-results" # and the scrape harvested *recommendation-widget* cards instead, returning Shanghai, # Kyiv, Bogota and Mannheim and zero Swiss roles for months while looking healthy. The # underlying Phenom refineSearch endpoint filters correctly (134 CH roles), so hit it # directly. Title-filtered: Roche CH is overwhelmingly lab/pharma/apprenticeship reqs. ("roche", "Roche", "phenom", { "url": "https://careers.roche.com/widgets", "ref_num": "ROCHGLOBAL", "page_id": "page11-ds", "country_facet": "Switzerland", "_title_exclude": NOISE_TITLE_EXCLUDE, }), # Cisco (PhenomPeople, new careers.cisco.com domain). Keyword search surfaces CH roles. ("cisco", "Cisco", "playwright", { "url": "https://careers.cisco.com/global/en/search-results?keywords=Switzerland", "wait_for": "a[href*='/job/'], div[role='listitem']", "card": "div[role='listitem']:has(a[href*='/job/'])", "title_sel": "a[href*='/job/']", "link_sel": "a[href*='/job/']", "link_attr": "href", "url_prefix": "https://careers.cisco.com", "default_location": "Switzerland", "cookie_accept": ["#onetrust-accept-btn-handler"], "scroll_count": 5, "use_inner_text_as_blob": True, }), # --- Zürich/Zug high-comp additions (2026-05-31 list review) --- # Palantir (Lever). Verified: 221 postings on the public board. It's US/London-heavy, so # Swiss/Schwyz roles are rare but self-surface when posted (the location filter drops the # US/London bulk). No title filter: his target titles (Forward Deployed Software Engineer, # Deployment Strategist) aren't in ENG_TITLE_FILTER, so filtering would hide them. ("palantir", "Palantir", "lever", {"slug": "palantir"}), # QuantCo (Lever — note the trailing-hyphen slug "quantco-"). ~16 roles, most tagged # "Europe" (hybrid); QuantCo's continental hub is Zürich, so Europe-scope in # location_eligibility surfaces them. No title filter: target band is DS/Quant/AI/Cloud # (ENG_TITLE_FILTER would drop some); interns/frontend caught by NEGATIVE_KEYWORDS. ("quantco", "QuantCo", "lever", {"slug": "quantco-"}), # --- Bern/Thun local tier — WLB & proximity exception (comp bar relaxed; 2026-06-01) --- # Wired after live endpoint discovery. ⚠️ German citizen: RUAG classified work may require # Swiss citizenship — verify per-role before tailoring (see project_target_companies). # Swissgrid (Aarau): Magnolia CMS JSON endpoint (verified). placeOfWork is a bare city # (Aarau/Prilly/...), so loc_suffix tags it Switzerland for the CH filter. No title filter # (small board ~13 roles; lets Data Scientist / Applied-ML roles surface). ("swissgrid", "Swissgrid (Aarau)", "json", { "url": "https://www.swissgrid.ch/.rest/cloud/component-data?path=%2Fswissgrid%2Fen%2Fhome%2Fcareer%2Fjobs%2Fmain%2Fjoblist_transferred_11", "jobs_key": "jobs", "field_title": "title", "field_location": "placeOfWork", "field_url": "descriptionUrl", "field_date": "onlineSince", "loc_suffix": " Switzerland", "desc_keys": ["department", "typeOfEmployment", "entryLevel"], "_title_exclude": NOISE_TITLE_EXCLUDE, }), # RUAG (Thun/Bern/Emmen). Jobs render on the portal as anchors to jobs.ruag.ch; the first # line of each anchor is the title. All sites are Swiss, so default_location=Switzerland # passes the CH filter. ENG_TITLE_FILTER cuts the apprenticeship/Lehrstelle bulk. # Drupal portal: 20 jobs/page, server-rendered, paginated via ?page=N (0-indexed). The # first page is apprenticeship-heavy; eng roles (DevOps/Data/Cloud) are on later pages, # so we page through until a page adds nothing new (~5-6 pages). ("ruag", "RUAG (Thun/Bern)", "playwright", { "url": "https://www.ruag.ch/en/working-us/job-portal", "wait_for": "a[href*='/offene-stellen/']", "card": "a[href*='/offene-stellen/']", "title_attr": "text", "link_attr": "href", "default_location": "Switzerland", "scroll_count": 1, "page_param": "page", "max_pages": 10, "_title_exclude": NOISE_TITLE_EXCLUDE, }), # SBB (company.sbb.ch — the correct host; company-jobs.sbb.ch was wrong). AEM job filter # served as a flat JSON list; the fetch_sbb adapter replicates the user's IT + Bern-region # filter. German/generic titles, so _score_floor keeps the pre-filtered results visible. # ⚠️ DE-citizen limits may apply to some SBB security/critical-infra roles. ("sbb", "SBB", "sbb", { "topic": "IT / Telekommunikation", "region": "Bern Mittelland", "_score_floor": 2, "_title_exclude": NOISE_TITLE_EXCLUDE, }), # BKW Group (jobs.bkw.com — the real ATS host). PMS structured-data API; ~600 roles # group-wide, so fetch_bkw keeps only Berufsfeld categories Informatik/Trading/Finanzen # (IT/data + energy-trading, incl. the flagged Energiehandel roles). German/generic # titles, so _score_floor keeps the pre-filtered set visible. ("bkw", "BKW (Bern)", "bkw", {"_score_floor": 2, "_title_exclude": NOISE_TITLE_EXCLUDE}), # PostFinance (Bern). The careers site renders a small, client-side paginated board; # scrape all pages through its stable next-page control. No title filter: the board is # low-volume, and the scorer keeps unrelated banking/customer-service roles out of the # default report while retaining locally relevant engineering variants. ("postfinance", "PostFinance (Bern)", "playwright", { "url": "https://jobs.postfinance.ch/PostFinance/search?locale=de_DE", "wait_for": "a[href*='/PostFinance/job/']", "card": "a[href*='/PostFinance/job/']", "title_attr": "text", "link_attr": "href", "default_location": "Switzerland", "use_inner_text_as_blob": True, "next_button": "#pfch-pagination-next", "max_pages": 10, "_title_exclude": NOISE_TITLE_EXCLUDE, }), # BFH (Bern University of Applied Sciences). Re-added 2026-07-14: the jobs.bfh.ch domain # itself is a broken/stub SPA shell (renders "Career Center project template", nothing # else — this is what looked like a 403 previously), but the actual listing widget is # embedded and fully renders on the www.bfh.ch careers page. Card = one <div> per posting # with 3 li.teaser-text-list-item fields (department, location, employment type); location # is the 2nd. Low-volume (12 open reqs), single page. German titles score 0 under the # English-keyword scorer (same problem as SBB/BKW) — _score_floor keeps them visible # instead of hidden as "noise", e.g. "Professur Data-Driven Government", # "Co-Leitung ... digitale Souveränität". ("bfh", "BFH (Bern)", "playwright", { "url": "https://www.bfh.ch/en/about-bfh/careers/jobs/", "wait_for": ".teas-prospective-job-teaser", "card": ".teas-prospective-job-teaser", "title_sel": "h2.teaser-title", "location_sel": "li.teaser-text-list-item:nth-of-type(2)", "link_sel": "a", "default_location": "Switzerland", "_score_floor": 2, "_title_exclude": NOISE_TITLE_EXCLUDE, }), ] # Companies that resist scraping — clickable checklist in the report. # (BFH re-automated 2026-07-14 on www.bfh.ch. Dialectic dropped — crypto covered elsewhere.) MANUAL_CHECK = [ # Oracle (Tier C, requested 2026-06-06). Oracle Recruiting Cloud (ORC) resists clean # scraping: careers.oracle.com renders job tiles that are NOT anchors, so the playwright # selector pattern fails. The ORC REST endpoint works and returns data, but reliable CH # filtering needs the Switzerland geography node id, which the facet-expand call 400s on. # Wireable later as a `json` adapter once that geographyId is resolved. For now, manual: # https://eeho.fa.us2.oraclecloud.com/hcmRestApi/resources/latest/recruitingCEJobRequisitions # ?onlyData=true&expand=requisitionList.workLocation # &finder=findReqs;siteNumber=CX_45001,limit=200,sortBy=POSTING_DATES_DESC # (then client-filter requisitionList[].PrimaryLocation for Switzerland/Zürich) ("Oracle", "ORC SPA resists scraping; REST endpoint known but needs CH geographyId (see code comment). Check Switzerland tech roles manually.", "https://careers.oracle.com/en/sites/jobsearch/jobs?location=Switzerland"), ] @lru_cache(maxsize=1) def _ssl_context(): """Prefer certifi's CA bundle over the OS trust store: on Windows, urllib's default context pulls trust anchors from the system cert store, which has been observed to reject valid Workday-hosted certs as expired (stale cached root) even though the live chain verifies fine elsewhere. Falls back to the default context if certifi isn't installed (pip install -r requirements.txt covers it).""" try: import certifi return ssl.create_default_context(cafile=certifi.where()) except ImportError: return ssl.create_default_context() def http_get_json(url, headers=None, data=None, method="GET"): headers = headers or {} headers.setdefault("User-Agent", USER_AGENT) headers.setdefault("Accept", "application/json") if data is not None and isinstance(data, dict): data = json.dumps(data).encode("utf-8") headers.setdefault("Content-Type", "application/json") req = urllib.request.Request(url, data=data, headers=headers, method=method) with urllib.request.urlopen(req, timeout=30, context=_ssl_context()) as resp: return json.loads(resp.read().decode("utf-8")) def fetch_workday(args): host, site, tenant = args["host"], args["site"], args["tenant"] search_text = args.get("search_text", "") url = f"https://{host}/wday/cxs/{tenant}/{site}/jobs" jobs, offset = [], 0 while True: data = http_get_json(url, method="POST", data={ "appliedFacets": {}, "limit": 20, "offset": offset, "searchText": search_text, }) postings = data.get("jobPostings", []) for p in postings: ext = p.get("externalPath", "") jid = (p.get("bulletFields") or [ext])[0] if p.get("bulletFields") else ext jobs.append({ "id": jid, "title": p.get("title", ""), "location": p.get("locationsText", "") + " " + ext, "url": f"https://{host}{ext}", "posted": p.get("postedOn", ""), "description": "", }) total = data.get("total", 0) offset += len(postings) if not postings or offset >= total: break return jobs def fetch_ashby(args): slug = args["slug"] url = f"https://api.ashbyhq.com/posting-api/job-board/{slug}?includeCompensation=true" data = http_get_json(url) jobs = [] for j in data.get("jobs", []): secs = j.get("secondaryLocations", []) or [] sec_names = [s.get("location", "") if isinstance(s, dict) else str(s) for s in secs] loc_blob = " | ".join([j.get("location", "") or ""] + sec_names) jobs.append({ "id": j.get("id"), "title": j.get("title", ""), "location": loc_blob, "url": j.get("jobUrl"), "posted": j.get("publishedAt", ""), "description": (j.get("descriptionPlain") or "")[:2500], "department": j.get("department", ""), }) return jobs def fetch_greenhouse(args): # Title-filtered boards score title_only — skip heavy JD HTML (content=true) for those. board = args["board"] want_content = not args.get("_title_filter") url = f"https://boards-api.greenhouse.io/v1/boards/{board}/jobs" if want_content: url += "?content=true" data = http_get_json(url) jobs = [] for j in data.get("jobs", []): loc = (j.get("location") or {}).get("name", "") offices = j.get("offices") or [] office_names = " | ".join(o.get("name", "") for o in offices if isinstance(o, dict)) loc_blob = " ".join(x for x in [loc, office_names] if x) desc = "" if want_content: desc = j.get("content", "") or "" desc = re.sub(r"<[^>]+>", " ", desc) desc = re.sub(r"\s+", " ", desc).strip() jobs.append({ "id": str(j.get("id")), "title": j.get("title", ""), "location": loc_blob, "url": j.get("absolute_url"), "posted": j.get("updated_at", ""), "description": desc[:2500], }) return jobs def fetch_pcsx(args): """Eightfold PCSX search API. Microsoft uses apply.careers.microsoft.com. The same endpoint pattern is used by other PCS-hosted boards.""" domain = args["domain"] location = args.get("location", "") base = "https://apply.careers.microsoft.com/api/pcsx/search" jobs, start = [], 0 while True: url = f"{base}?domain={domain}&query=&location={urllib.parse.quote(location)}&start={start}&num=50" data = http_get_json(url, headers={"Referer": f"https://apply.careers.microsoft.com/careers?location={urllib.parse.quote(location)}"}) payload = data.get("data") or {} positions = payload.get("positions", []) or [] for p in positions: locs = p.get("locations") or [] jobs.append({ "id": str(p.get("id")), "title": p.get("name", ""), "location": " | ".join(locs), "url": f"https://jobs.careers.microsoft.com/global/en/job/{p.get('displayJobId') or p.get('id')}", "posted": p.get("postedTs", ""), "description": (p.get("description") or "")[:2000], }) # The API serves ~10 rows per call regardless of `num`, so a short page is NOT # end-of-results — page until a call comes back empty or `count` is reached. # (The old `len(positions) < 50` break capped Microsoft at its first 10 CH roles # and hid the entire Zürich Principal-FDE cluster sitting on page 2.) if not positions: break start += len(positions) total = payload.get("count") if isinstance(total, int) and start >= total: break if start >= 500: break return jobs def fetch_smartrecruiters(args): """SmartRecruiters public postings API. Used by many EU energy/commodity firms.""" company = args["company"] base = f"https://api.smartrecruiters.com/v1/companies/{company}/postings" jobs, offset = [], 0 while True: data = http_get_json(f"{base}?limit=100&offset={offset}") content = data.get("content", []) or [] for p in content: loc = p.get("location") or {} parts = [loc.get("fullLocation") or loc.get("city") or ""] if loc.get("remote"): parts.append("Remote") if loc.get("hybrid"): parts.append("Hybrid") loc_str = " ".join(x for x in parts if x) dept = (p.get("department") or {}).get("label", "") if isinstance(p.get("department"), dict) else "" func = (p.get("function") or {}).get("label", "") if isinstance(p.get("function"), dict) else "" jobs.append({ "id": str(p.get("id")), "title": p.get("name", ""), "location": loc_str, "url": f"https://jobs.smartrecruiters.com/{company}/{p.get('id')}", "posted": p.get("releasedDate", ""), "description": " ".join(filter(None, [dept, func])), }) total = data.get("totalFound", 0) offset += len(content) if not content or offset >= total or offset >= 300: break return jobs def fetch_rss(args): """Generic RSS/RDF feed parser. BIS publishes vacancies as RSS 1.0 (RDF), whose <item> elements live in the http://purl.org/rss/1.0/ namespace. Falls back to plain RSS 2.0 <item> elements. Location isn't in the feed, so default_location is required.""" import xml.etree.ElementTree as ET req = urllib.request.Request(args["url"], headers={"User-Agent": USER_AGENT}) with urllib.request.urlopen(req, timeout=30, context=_ssl_context()) as resp: root = ET.fromstring(resp.read()) ns = {"rss1": "http://purl.org/rss/1.0/", "dc": "http://purl.org/dc/elements/1.1/"} items = root.findall(".//rss1:item", ns) or root.findall(".//item") jobs = [] for it in items: def field(tag, namespaced=True): el = it.find(f"rss1:{tag}", ns) if namespaced else it.find(tag) if el is None and namespaced: el = it.find(tag) return (el.text or "").strip() if el is not None and el.text else "" link = field("link") jobs.append({ "id": link or field("title"), "title": field("title"), "location": args.get("default_location", ""), "url": link, "posted": (it.findtext("dc:date", default="", namespaces=ns) or field("date")), "description": re.sub(r"<[^>]+>", " ", field("description"))[:1500], }) return jobs # schema.org JobPosting addresses carry ISO alpha-2 country codes; the location policy # keyword lists are full names. Only the countries the policies actually test for. _ISO2_COUNTRY = { "CH": "Switzerland", "NO": "Norway", "DK": "Denmark", "DE": "Germany", "AT": "Austria", "FR": "France", "IT": "Italy", "ES": "Spain", "PL": "Poland", "NL": "Netherlands", "BE": "Belgium", "SE": "Sweden", "FI": "Finland", "GB": "United Kingdom", "UK": "United Kingdom", "IE": "Ireland", "US": "United States", "PT": "Portugal", "CZ": "Czechia", "RO": "Romania", "RS": "Serbia", "TR": "Turkey", } def fetch_teamtailor(args): """Teamtailor public JSON Feed (`https://<slug>.teamtailor.com/jobs.json`). Used by Telenor (`slug`) and Axpo (`base_url`, a Teamtailor board on a custom domain — careers.axpo.com — which no <slug>.teamtailor.com host resolves to). The public /jobs HTML page paginates and under-reports; the feed is authoritative. The feed carries no location field, so default_location is required (Teamtailor boards are per-country). Descriptions are frequently Norwegian — see the _score_floor note on the company. The feed caps at 100 items and links the next page via `next_url`; follow it, or large boards silently truncate (Axpo is ~400 roles, i.e. 4 pages). TRAP: an unclaimed slug returns Teamtailor's *demo* board (HTTP 200, plausible JSON) instead of 404. Verified 2026-07-29: `ksat` and `akerbp` both served the seed set below though neither company uses Teamtailor. Filtered here so phantom roles never reach a report — if a real board is ever dropped by this, widen the fingerprint.""" DEMO_TITLES = {"sales development manager", "team lead - csm", "social media manager", "customer success manager", "key account manager", "ux designer"} if args.get("base_url"): url = args["base_url"].rstrip("/") + "/jobs.json" else: url = f"https://{args['slug']}.teamtailor.com/jobs.json" jobs = [] for _ in range(args.get("max_pages", 10)): req = urllib.request.Request( url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"}) with urllib.request.urlopen(req, timeout=30, context=_ssl_context()) as resp: data = json.loads(resp.read().decode("utf-8", "replace")) items = data.get("items", []) or [] for it in items: # Prefer the schema.org JobPosting payload's real address over default_location. # Axpo is a pan-European trader (Madrid, Milan, Warsaw, Germany, France as well as # Baden/Zurich), so defaulting every role to the board's home country would mark # ~380 non-Swiss roles CH-eligible — the same class of bug as Apple's forced # "Switzerland" default. Telenor's feed has no _jobposting and falls back cleanly. location = args.get("default_location", "") places = ((it.get("_jobposting") or {}).get("jobLocation") or []) if isinstance(places, dict): places = [places] parts = [] for place in places: addr = (place or {}).get("address") or {} # schema.org gives ISO alpha-2 ("CH", "NO"); the location policy matches on # full country names, so expand or the role reads as location-unknown. country = _ISO2_COUNTRY.get((addr.get("addressCountry") or "").upper(), addr.get("addressCountry")) bit = ", ".join(x for x in (addr.get("addressLocality"), country) if x) if bit and bit not in parts: parts.append(bit) if parts: location = " | ".join(parts) jobs.append({ "id": it.get("id") or it.get("url", ""), "title": it.get("title", ""), "location": location, "url": it.get("url", ""), "posted": it.get("date_published", ""), "description": re.sub(r"<[^>]+>", " ", it.get("content_html", ""))[:2500], }) nxt = data.get("next_url") if not items or not nxt or nxt == url: break url = nxt titles = {j["title"].strip().lower() for j in jobs} if jobs and len(titles & DEMO_TITLES) >= 4: raise RuntimeError( f"teamtailor slug '{args['slug']}' served the demo board, not a real one " f"({len(jobs)} placeholder ads) — the company likely uses a different ATS") return jobs def fetch_getro(args): """Getro network job-board search API (POST JSON). Powers VC portfolio talent networks. NO company in COMPANIES currently uses this adapter — the Coinbase Ventures network was dropped 2026-08-18 (see the note there). Kept because it generalises to any Getro collection id. Note such boards list PORTFOLIO companies' roles, never the VC's own openings, so an org filter is usually needed. Server-side filters: searchable_locations and job_functions. Org name is folded into the title since this is a multi-company board.""" collection = args["collection"] url = f"https://api.getro.com/api/v2/collections/{collection}/search/jobs" filters = {} if args.get("locations"): filters["searchable_locations"] = args["locations"] if args.get("job_functions"): filters["job_functions"] = args["job_functions"] jobs, page = [], 0 excluded_orgs = {name.casefold() for name in args.get("_exclude_orgs", [])} while True: data = http_get_json(url, method="POST", data={ "hitsPerPage": 100, "page": page, "query": "", "filters": filters, }) res = data.get("results", {}) or {} batch = res.get("jobs", []) or [] for j in batch: org = (j.get("organization") or {}).get("name", "") if org.casefold() in excluded_orgs: continue locs = j.get("searchable_locations") or j.get("locations") or [] loc_str = " | ".join(locs) if isinstance(locs, list) else str(locs) ts = j.get("created_at") posted = "" if isinstance(ts, (int, float)): posted = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d") title = j.get("title", "") jobs.append({ "id": str(j.get("id")), "title": f"{title} @ {org}" if org else title, "location": loc_str, "url": j.get("url", ""), "posted": posted, "description": " ".join(filter(None, [org] + (j.get("skills") or []))), }) total = res.get("count", 0) page += 1 if not batch or len(jobs) >= total or page >= 10: break return jobs def fetch_amazon(args): """amazon.jobs public search JSON. Covers Amazon + AWS, which share one board. `normalized_country_code[]` is the filter that actually works; the plain `country[]` and `loc_query` params are ignored and silently return the global (US-heavy) set — verified 2026-08-18, where `country[]=CHE` returned 6,837 mostly-Seattle hits against 32 for the normalized form.""" base = "https://www.amazon.jobs/en/search.json" headers = {"User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")} jobs, offset, page_size = [], 0, 100 while True: qs = urllib.parse.urlencode({ "radius": "100mi", "offset": offset, "result_limit": page_size, "sort": "recent", "base_query": args.get("base_query", ""), }) countries = "".join(f"&normalized_country_code[]={urllib.parse.quote(c)}" for c in args.get("countries", ["CHE"])) data = http_get_json(f"{base}?{qs}{countries}", headers=headers) batch = data.get("jobs", []) or [] for j in batch: jobs.append({ "id": str(j.get("id_icims") or j.get("id") or ""), "title": j.get("title", ""), "location": j.get("normalized_location") or j.get("location") or "", "url": "https://www.amazon.jobs" + (j.get("job_path") or ""), "posted": j.get("posted_date", ""), "description": (j.get("description_short") or j.get("description") or "")[:2000], }) total = data.get("hits", 0) offset += page_size if not batch or offset >= total or offset >= args.get("max_results", 500): break return jobs def fetch_phenom(args): """Phenom People careers search (POST /widgets with ddoKey=refineSearch). Used by Roche. The public search-results page is a JS shell whose location filter silently fails: a scrape of `?locationsearch=Switzerland` returns *recommendation-widget* cards (Shanghai, Kyiv, Bogota) while the page state reports `no-results`. Verified 2026-08-18 — this is why Roche contributed 0 Swiss roles for months while looking healthy. The underlying refineSearch endpoint filters correctly (134 CH roles), so query it directly. `country_facet` is the value for the `country` facet (e.g. "Switzerland"). `ref_num` and `page_id` are tenant constants visible in any /widgets POST from the careers site.""" url = args["url"] ref_num, page_id = args["ref_num"], args.get("page_id", "page11-ds") lang = args.get("lang", "en_global") page_size = args.get("page_size", 100) headers = { "Origin": "{0.scheme}://{0.netloc}".format(urllib.parse.urlsplit(url)), "Referer": url, # Phenom 403s the default library UA. "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"), } jobs, offset = [], 0 while True: data = http_get_json(url, method="POST", headers=headers, data={ "lang": lang, "deviceType": "desktop", "country": "global", "pageName": "search-results", "ddoKey": "refineSearch", "sortBy": "", "subsearch": "", "from": offset, "jobs": True, "counts": True, "all_fields": ["country", "state", "city", "category", "type"], "size": page_size, "clearAll": False, "jdsource": "facets", "isSliderEnable": False, "pageId": page_id, "siteType": "external", "keywords": "", "global": True, "locationData": {}, "selected_fields": {"country": [args["country_facet"]]}, }) res = data.get("refineSearch", {}) or {} batch = (res.get("data", {}) or {}).get("jobs", []) or [] for j in batch: jobs.append({ "id": str(j.get("jobId") or j.get("jobSeqNo") or ""), "title": j.get("title", ""), "location": j.get("location") or j.get("cityStateCountry") or "", "url": j.get("applyUrl", ""), "posted": (j.get("postedDate") or "")[:10], "description": (j.get("descriptionTeaser") or "")[:2000], }) total = res.get("totalHits", 0) offset += page_size if not batch or offset >= total or offset >= args.get("max_results", 600): break return jobs def fetch_onlyfy(args): """onlyfy.jobs board (XING E-Recruiting / ex-Prinzip), used by Bitcoin Suisse. The candidate/job/ajax_list endpoint returns an HTML fragment listing every posting; each card carries a <a href="/job/ID">title</a> and a location cell flagged by an icon-map-marker. Titles and locations appear in document order, one of each per card, so we extract both lists and zip them. No JSON API and no headless browser needed.""" import html as _html slug = args["slug"] base = f"https://{slug}.onlyfy.jobs" url = (f"{base}/candidate/job/ajax_list" f"?display_length=100&page=1&sort=date&sort_dir=DESC&search=") req = urllib.request.Request(url, headers={ "User-Agent": USER_AGENT, "X-Requested-With": "XMLHttpRequest", }) with urllib.request.urlopen(req, timeout=30, context=_ssl_context()) as resp: page = resp.read().decode("utf-8", "replace") titles = re.findall(r'<a href="(/job/[a-z0-9]+)">(.*?)</a>', page, re.S) locs = re.findall(r'icon-map-marker[^>]*></i>\s*([^<]+)', page) if titles and locs and len(titles) != len(locs): # Misaligned regex pairs would invent wrong locations; keep titles, drop bad locs. print(f" onlyfy: title/loc count mismatch ({len(titles)} vs {len(locs)}); " f"using titles only", file=sys.stderr) locs = [""] * len(titles) jobs = [] for (href, raw_title), raw_loc in zip(titles, locs if locs else [""] * len(titles)): title = _html.unescape(re.sub(r"<[^>]+>", "", raw_title)).strip() loc = _html.unescape(raw_loc).strip() if raw_loc else "" jobs.append({ "id": href.rsplit("/", 1)[-1], "title": title, "location": loc, "url": base + href, "posted": "", "description": loc, }) return jobs def fetch_lever(args): """Lever public postings API. Palantir uses this. The board is US/London-heavy; Swiss/Zurich (Schwyz hub) roles are rare on it but will surface here when posted — location filtering downstream drops the US/London bulk. categories.allLocations captures multi-location postings; createdAt is epoch-ms.""" slug = args["slug"] data = http_get_json(f"https://api.lever.co/v0/postings/{slug}?mode=json") jobs = [] for j in data: cats = j.get("categories") or {} all_locs = cats.get("allLocations") or [] loc_blob = " | ".join(x for x in ([cats.get("location") or ""] + [str(a) for a in all_locs]) if x) ts = j.get("createdAt") posted = "" if isinstance(ts, (int, float)): posted = datetime.fromtimestamp(ts / 1000, tz=timezone.utc).strftime("%Y-%m-%d") jobs.append({ "id": j.get("id"), "title": j.get("text", ""), "location": loc_blob, "url": j.get("hostedUrl"), "posted": posted, "description": (j.get("descriptionPlain") or "")[:2500], }) return jobs def _parse_taleo_jobsearch_html(html, host, career_section): """Extract jobs from a Taleo jobsearch.ftl HTML response. Taleo does not expose a reliable public JSON search API for this board (REST returns careerSectionUnAvailable). Jobs are embedded in the page as api.fillInterface arrays and as a pipe-delimited initialHistory field. """ import html as _html jobs, seen = [], set() detail_base = f"https://{host}/careersection/{career_section}/jobdetail.ftl" # Quoted fillInterface form (already-unescaped dates in modern Taleo builds). # Observed record shape (NATO 2026-07): # '<id>','<title>','<id>','<title>', five more '<id>', # '<jobNo>','<Country-City>','false','','','','', # '<deadline>','<NATO body>','<grade>' pat = re.compile( r"'(\d+)','((?:[^'\\]|\\.)+)'," r"'\1','\2'," r"(?:'\1',){5}" r"'(\d+)','((?:[^'\\]|\\.)+)'," r"'false','','','',''," r"'((?:[^'\\]|\\.)*)','((?:[^'\\]|\\.)*)','((?:[^'\\]|\\.)*)'" ) for m in pat.finditer(html): job_no = m.group(3) if job_no in seen: continue seen.add(job_no) def _unq(s): return _html.unescape(s.replace("\\'", "'").replace("\\:", ":")) title = _unq(m.group(2)) loc = _unq(m.group(4)).replace("-", ", ") deadline = _unq(m.group(5)) body = _unq(m.group(6)) grade = _unq(m.group(7)) # List view exposes application deadline only (not a posted-on date). We store # the date part in `posted` so the report surfaces the closing date. deadline_short = deadline.split(",")[0].strip() if deadline else "" jobs.append({ "id": job_no, "title": title, "location": loc, "url": f"{detail_base}?job={job_no}&lang=en", "posted": deadline_short, "description": f"{body} {grade}".strip()[:500], }) if jobs: return jobs # Fallback: pipe-delimited initialHistory hidden field m = re.search( r'(?:name|id)="initialHistory"[^>]*value="([^"]*)"', html, re.I ) or re.search( r'value="([^"]*)"[^>]*(?:name|id)="initialHistory"', html, re.I ) if not m: return [] hist = urllib.parse.unquote(m.group(1).replace("%5C:", ":").replace("\\:", ":")) parts = hist.split("!|!") for i, p in enumerate(parts): if not re.fullmatch(r"\d{5,6}", p) or i + 1 >= len(parts): continue loc_raw = parts[i + 1] if not re.search( r"Norway|Denmark|Belgium|Germany|Italy|France|United|Luxembourg|" r"Netherlands|Poland|Spain|Portugal|Canada|Turkey|Türkiye", loc_raw, re.I, ): continue job_no = p if job_no in seen: continue title = "" for j in range(i - 1, max(-1, i - 12), -1): cand = parts[j] if (cand and not re.fullmatch(r"\d+", cand) and cand not in ("false", "true", "Apply") and len(cand) > 8): title = cand break if not title: continue seen.add(job_no) deadline = parts[i + 6] if i + 6 < len(parts) else "" body = parts[i + 7] if i + 7 < len(parts) else "" grade = parts[i + 8] if i + 8 < len(parts) else "" deadline_short = deadline.split(",")[0].strip() if deadline else "" jobs.append({ "id": job_no, "title": _html.unescape(title), "location": loc_raw.replace("-", ", "), "url": f"{detail_base}?job={job_no}&lang=en", "posted": deadline_short, "description": f"{body} {grade}".strip()[:500], }) return jobs def fetch_taleo(args): """Oracle Taleo careersection board (used by NATO civilian vacancies). The marketing page https://www.nato.int/en/work-with-us/careers/vacancies is an AEM shell; the live board is Taleo. Public REST search is unavailable, so we GET jobsearch.ftl (optionally filtered by Taleo location taxonomy IDs from the board's location dropdown) and parse embedded job records. Args: host (default nato.taleo.net), career_section (default "2"), location_ids (optional list of Taleo option values — e.g. Norway/Denmark). """ host = args.get("host", "nato.taleo.net") section = str(args.get("career_section", "2")) location_ids = args.get("location_ids") # One request per location id (Taleo query param takes a single id); merge by job no. # If no location_ids, one unfiltered first-page fetch. fetches = location_ids if location_ids else [None] by_id = {} for loc_id in fetches: url = f"https://{host}/careersection/{section}/jobsearch.ftl?lang=en" if loc_id: url += f"&location={urllib.parse.quote(str(loc_id))}" req = urllib.request.Request(url, headers={ "User-Agent": USER_AGENT, "Accept": "text/html,application/xhtml+xml", }) with urllib.request.urlopen(req, timeout=60, context=_ssl_context()) as resp: html = resp.read().decode("utf-8", "replace") for j in _parse_taleo_jobsearch_html(html, host, section): by_id[j["id"]] = j return list(by_id.values()) def fetch_json(args): """Generic JSON jobs API with configurable field names, for employer sites that expose a clean public endpoint. Verified use: Swissgrid (Magnolia CMS /.rest/cloud/component-data — {config, jobs:[...], filters}). Field names vary by site, so they're configurable: field_title/field_location/field_url/field_date. loc_suffix appends e.g. ' Switzerland' so the CH location filter matches city-only values such as "Aarau"/"Prilly" (not every Swiss town is in CH_LOCATION_KEYWORDS). desc_keys fold extra fields (department, employment type) into the description for keyword scoring. Args: url, jobs_key (default "jobs"), field_* (defaults title/location/url/date), url_prefix, loc_suffix, desc_keys.""" data = http_get_json(args["url"]) arr = data.get(args.get("jobs_key", "jobs"), []) if isinstance(data, dict) else (data or []) ft, fl = args.get("field_title", "title"), args.get("field_location", "location") fu, fd = args.get("field_url", "url"), args.get("field_date", "date") prefix, suffix = args.get("url_prefix", ""), args.get("loc_suffix", "") desc_keys = args.get("desc_keys", []) jobs = [] for j in arr: url = j.get(fu, "") or "" if url and not url.startswith("http") and prefix: url = prefix.rstrip("/") + "/" + url.lstrip("/") loc = (j.get(fl, "") or "").strip() + suffix desc = " ".join(str(j.get(k)) for k in desc_keys if j.get(k)) jobs.append({ "id": str(j.get("id") or url), "title": j.get(ft, ""), "location": loc, "url": url, "posted": j.get(fd, "") or "", "description": desc[:500], }) return jobs def fetch_sbb(args): """SBB (company.sbb.ch) AEM job filter. The whole board is served as a flat JSON list at .../jobfilter.results.json (~145 roles); the website filters client-side via each job's numbered `attributes`: '20'=Berufsfeld/topic, '110'=region, '100'=city, 'links.directlink'=the jobs.sbb.ch URL. We replicate the user's IT + Bern-region filter so only commutable IT roles surface. Titles are German/generic (Application Engineer, Network Security Engineer, OT Architekt) and won't match ENG_TITLE_FILTER or the keyword scorer, so this company is given a _score_floor in COMPANIES to keep its pre-filtered results visible. topic/region are configurable substrings.""" url = args.get("url", ("https://company.sbb.ch/content/internet/corporate/de/" "jobs-karriere/jobs/job-suche/jcr:content/parmain/" "jobfilter.results.json")) topic = args.get("topic", "IT / Telekommunikation") region = args.get("region", "Bern Mittelland") data = http_get_json(url) arr = data if isinstance(data, list) else (data.get("results") or data.get("jobs") or []) jobs = [] for j in arr: a = j.get("attributes", {}) or {} blob = " ".join(str(x) for v in a.values() for x in (v if isinstance(v, list) else [v])) if topic and topic not in blob: continue if region and region not in blob: continue region_v = " ".join(a.get("110", []) or []) city_v = " ".join(a.get("100", []) or []) field_v = " ".join(a.get("20", []) or []) jobs.append({ "id": str(j.get("id") or j.get("viewkey") or ""), "title": j.get("title", ""), "location": f"{city_v} {region_v} Schweiz".strip(), "url": (j.get("links") or {}).get("directlink", ""), "posted": j.get("start_date", "") or "", "description": (field_v + " " + (j.get("text", "") or ""))[:400], }) return jobs def fetch_bkw(args): """BKW Group (jobs.bkw.com) PMS structured-data API. The whole-group board is ~600 roles dominated by building-tech / electrical / civil-engineering trades; we keep only the Berufsfeld categories relevant to the user (Informatik / Trading / Finanzen), which surfaces IT/data plus the energy-trading roles (Quant Risk Modeller, Solution Architect Energiehandel, Energy Derivatives/Market-Risk analysts). locations[].address gives city/country. Pre-filtered + German/generic titles, so paired with a _score_floor in COMPANIES. The category allowlist is configurable.""" url = args.get("url", ("https://jobs.bkw.com/_api/v1/structureddata?" "configFromContentElement=82381&language=de-ch")) allow = [c.lower() for c in args.get("categories", ["Informatik", "Trading", "Finanzen"])] data = http_get_json(url) arr = data if isinstance(data, list) else [] if not arr and isinstance(data, dict): for v in data.values(): if isinstance(v, list) and v and isinstance(v[0], dict) and "title" in v[0]: arr = v break jobs = [] for j in arr: if j.get("type") and j.get("type") != "jobs": continue cats = [c.get("title", "") for c in (j.get("relations", {}) or {}).get("Berufsfeld", []) or []] if allow and not any(any(a in c.lower() for a in allow) for c in cats): continue locs = j.get("locations") or [] addr = (locs[0].get("address") if locs and isinstance(locs[0], dict) else {}) or {} loc = " ".join(x for x in [addr.get("city", ""), addr.get("country", "")] if x) or "Schweiz" jobs.append({ "id": str(j.get("id") or j.get("url") or ""), "title": j.get("title", ""), "location": loc, "url": j.get("url", ""), "posted": "", "description": " ".join(cats + [j.get("subtitle", "") or ""])[:300], }) return jobs # Injected before page scripts run, to mask the most common headless-detection signals. # Required for Google; harmless for the other sites. STEALTH_JS = """ Object.defineProperty(navigator, 'webdriver', {get: () => undefined}); window.chrome = {runtime: {}, loadTimes: () => {}, csi: () => {}, app: {}}; Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]}); Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en', 'de']}); const _q = navigator.permissions && navigator.permissions.query; if (_q) { navigator.permissions.query = (p) => p && p.name === 'notifications' ? Promise.resolve({state: Notification.permission}) : _q(p); } """ _playwright_singleton = {"pw": None, "browser": None} def _get_browser(): """Lazy-init a single shared headless browser. Saves ~3s per company.""" if _playwright_singleton["browser"] is not None: return _playwright_singleton["browser"] try: from playwright.sync_api import sync_playwright except ImportError as e: raise RuntimeError("playwright not installed - run: pip install -r requirements.txt") from e pw = sync_playwright().start() browser = pw.chromium.launch( headless=True, args=["--disable-blink-features=AutomationControlled"], ) _playwright_singleton["pw"] = pw _playwright_singleton["browser"] = browser return browser def _absolutize(href, prefix): """Join a possibly-relative href with the configured prefix.""" if not href or href.startswith("http"): return href cleaned = href.lstrip("./").lstrip("/") if not prefix: return href return prefix.rstrip("/") + "/" + cleaned def _strip_query_param(url, param): """Drop a single query parameter from a URL, preserving the rest. Job-card hrefs inherit the listing page's query string, so with query-param pagination the same posting would get a different URL (and therefore a different id) depending on which page it happened to land on — making every reshuffle look like a brand-new job and orphaning its decision-log entry. """ if not url or f"{param}=" not in url: return url head, _, query = url.partition("?") if not query: return url kept = [kv for kv in query.split("&") if kv and not kv.startswith(f"{param}=")] return f"{head}?{'&'.join(kept)}" if kept else head def _extract_location_from_blob(full, default=""): """Pull a short location line out of a job-card text blob. Google/Meta/Roche-style cards often dump Material-icon chrome into inner_text (e.g. 'corporate_fare' / 'place' / 'Zürich, Switzerland' / 'bar_chart'). Prefer the line after a 'place'/'location' label, else the first short line that looks like a place, else the configured default — never the first 300 chars of the whole card. """ if not full: return default lines = [ln.strip() for ln in full.splitlines() if ln.strip()] skip = { "place", "location", "corporate_fare", "bar_chart", "work", "schedule", "google", "meta", "apple", "roche", "cisco", } for i, ln in enumerate(lines): if ln.lower() in ("place", "location") and i + 1 < len(lines): cand = lines[i + 1].strip() if cand and cand.lower() not in skip and len(cand) < 120: return cand[:200] place_hints = ( CH_LOCATION_KEYWORDS + DK_LOCATION_KEYWORDS + NO_LOCATION_KEYWORDS + list(EUROPE_SCOPE_KEYWORDS) + list(GLOBAL_SCOPE_KEYWORDS) + ["remote", "hybrid", "onsite", "on-site"] ) for ln in lines: low = ln.lower() if len(ln) > 120 or low in skip: continue if any(k in low for k in place_hints): return ln[:200] return default def _close_browser(): if _playwright_singleton["browser"]: try: _playwright_singleton["browser"].close() except Exception: pass if _playwright_singleton["pw"]: try: _playwright_singleton["pw"].stop() except Exception: pass def fetch_playwright(args): """Generic headless-browser scraper. See COMPANIES entries for selector args.""" browser = _get_browser() ctx = browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", locale="en-US", viewport={"width": 1366, "height": 768}, ) ctx.add_init_script(STEALTH_JS) page = ctx.new_page() jobs = [] seen_ids = set() def scrape_current(): """Extract cards from the currently-loaded page; append new ones to `jobs`. Returns the count of newly-added (not-yet-seen) cards so a pagination loop can stop once a page contributes nothing new.""" wait_for = args.get("wait_for") if wait_for: try: page.wait_for_selector(wait_for, timeout=15000) except Exception: page.wait_for_timeout(4000) # Scroll a few times to trigger any lazy-loaded results for _ in range(args.get("scroll_count", 3)): try: page.mouse.wheel(0, 4000) page.wait_for_timeout(700) except Exception: break cards = page.locator(args["card"]) n = min(cards.count(), args.get("max_cards", 150)) added = 0 for i in range(n): card = cards.nth(i) try: title = "" if args.get("title_attr") == "text": title = (card.inner_text() or "").strip().split("\n", 1)[0][:200] elif args.get("title_attr"): title = (card.get_attribute(args["title_attr"]) or "").strip() elif args.get("title_sel"): t = card.locator(args["title_sel"]).first if t.count(): # Read either an attribute (e.g. aria-label) or the inner text if args.get("title_sel_attr"): title = (t.get_attribute(args["title_sel_attr"]) or "").strip() else: title = (t.inner_text() or "").strip() if args.get("title_strip_prefix") and title.startswith(args["title_strip_prefix"]): title = title[len(args["title_strip_prefix"]):].strip() if not title: title = (card.inner_text() or "").strip().split("\n", 1)[0][:200] location = args.get("default_location", "") if args.get("location_sel"): lsel = card.locator(args["location_sel"]).first if lsel.count(): location = (lsel.inner_text() or location).strip() link_el = card if not args.get("link_sel") else card.locator(args["link_sel"]).first href = (link_el.get_attribute(args.get("link_attr", "href")) or "") if link_el.count() else "" href = _absolutize(href, args.get("url_prefix", "")) if args.get("page_param"): href = _strip_query_param(href, args["page_param"]) if not title: continue jid = href or f"{page.url}#{i}" if jid in seen_ids: continue seen_ids.add(jid) added += 1 description = "" if args.get("use_inner_text_as_blob"): # Full card text for keyword scoring; location via structured extract. full = (card.inner_text() or "") description = full[:2000] extracted = _extract_location_from_blob( full, default=args.get("default_location", "") ) # Prefer a real location line over a default that is empty or a # polluted leftover; keep default_location when extract finds nothing. if extracted: location = extracted elif not location: location = args.get("default_location", "") jobs.append({ "id": jid, "title": title, "location": location, "url": href or page.url, "posted": "", "description": description, }) except Exception: continue return added try: page.goto(args["url"], timeout=45000, wait_until="domcontentloaded") # Optional cookie banner acceptance (once, on the first page) for sel in args.get("cookie_accept", []) or []: try: btn = page.locator(sel).first if btn.is_visible(timeout=2000): btn.click() page.wait_for_timeout(500) except Exception: pass # Optional query-param pagination (e.g. Drupal "?page=N"). The base URL is the first # page (already loaded); fetch successive pages until one adds no new cards. # `page_param_start` is the value the SECOND page takes: 1 for 0-indexed boards # (Drupal), 2 for 1-indexed ones (Google, where "?page=1" is just page one again). page_param = args.get("page_param") if page_param: base = args["url"] joiner = "&" if "?" in base else "?" second = args.get("page_param_start", 1) for p in range(args.get("max_pages", 8)): if p > 0: page.goto(f"{base}{joiner}{page_param}={second + p - 1}", timeout=45000, wait_until="domcontentloaded") added = scrape_current() if p > 0 and added == 0: break else: # Optional in-page pagination for client-side job boards. This is distinct from # query-param pagination above: the URL does not change when moving between pages. next_button = args.get("next_button") if next_button: for _ in range(args.get("max_pages", 8)): added = scrape_current() button = page.locator(next_button) if button.count() != 1: break try: if not button.is_visible() or button.is_disabled(): break button.click() page.wait_for_timeout(args.get("next_wait_ms", 800)) except Exception: break # A page with no unseen cards signals a loop or exhausted pagination. if added == 0: break else: scrape_current() finally: ctx.close() return jobs ADAPTERS = { "workday": fetch_workday, "ashby": fetch_ashby, "greenhouse": fetch_greenhouse, "pcsx": fetch_pcsx, "smartrecruiters": fetch_smartrecruiters, "rss": fetch_rss, "teamtailor": fetch_teamtailor, "getro": fetch_getro, "amazon": fetch_amazon, "phenom": fetch_phenom, "onlyfy": fetch_onlyfy, "lever": fetch_lever, "taleo": fetch_taleo, "json": fetch_json, "sbb": fetch_sbb, "bkw": fetch_bkw, "playwright": fetch_playwright, } def location_eligibility(loc_text, policy="default"): """Decide whether a posting's location is in-scope. Returns (eligible, in_ch, is_remote, is_relocation). Default policy (most companies): - CH onsite/hybrid - Denmark onsite/reloc - Europe/EMEA/global remote scope (work-from CH or DK is realistic) - CH-remote (Switzerland + remote wording) - Multi-country allowlists WITHOUT CH/DK/Europe/global scope are NOT eligible (e.g. Kraken "UK | Brazil | Ireland | ..." with no Switzerland/Europe) Overrides (args["_location_policy"]): - norway_only — Equinor: Norway only - dk_no — NATO-style: Denmark + Norway only (no CH) """ if not loc_text: return False, False, False, False low = loc_text.lower() in_ch = any(k in low for k in CH_LOCATION_KEYWORDS) in_dk = any(k in low for k in DK_LOCATION_KEYWORDS) in_no = any(k in low for k in NO_LOCATION_KEYWORDS) is_us_only = any(p in low for p in US_ONLY_PATTERNS) and not in_ch has_remote = bool(re.search(r"\bremote\b", low)) or "work from anywhere" in low europe_scope = any(k in low for k in EUROPE_SCOPE_KEYWORDS) global_scope = any(k in low for k in GLOBAL_SCOPE_KEYWORDS) and not is_us_only # Pan-regional scope counts even without the word "remote" (e.g. QuantCo "Europe"). is_europe_or_global = (europe_scope or global_scope) and not is_us_only is_ch_remote = in_ch and has_remote if policy == "norway_only": eligible = in_no and not is_us_only return eligible, False, False, eligible if policy == "dk_no": # NATO etc.: Denmark + Norway only — CH onsite is out (no NATO seat in CH). eligible = (in_dk or in_no) and not is_us_only return eligible, False, False, eligible # default is_relocation = in_dk is_remote = is_europe_or_global or is_ch_remote eligible = (in_ch or in_dk or is_europe_or_global) and not (is_us_only and not in_ch) return eligible, in_ch, is_remote, is_relocation @lru_cache(maxsize=512) def _kw_pattern(kw): """Word-boundary regex for a keyword. Plain substring matching produced false hits ('rag' inside 'sto[rag]e'/'tet[rag]on', 'intern' inside 'inte[rnal]'); we instead require the keyword not be flanked by alphanumerics. Keywords that begin/end on a non-word char (c#, .net, c++) skip that side's guard so they still match.""" esc = re.escape(kw.strip()) left = r"(?<![a-z0-9])" if kw.strip()[:1].isalnum() else "" right = r"(?![a-z0-9])" if kw.strip()[-1:].isalnum() else "" return re.compile(left + esc + right) def _kw_in(kw, text): return bool(_kw_pattern(kw).search(text)) def score_job(job, title_only=False): # Title carries the real signal; the JD body is full of company boilerplate (every # Kraken post mentions crypto/blockchain/trading, every cloud post mentions python). # So title matches score at full weight and body-only matches at half (min 1) — enough # to surface a role without letting boilerplate inflate it. Negatives count fully # wherever they appear (a disqualifier in the body still disqualifies). Title-filtered # boards pass title_only=True and skip body scoring entirely. title = (job.get("title") or "").lower() desc = "" if title_only else (job.get("description") or "").lower() score, pos, neg = 0, [], [] for kw, w in POSITIVE_KEYWORDS.items(): if _kw_in(kw, title): score += w pos.append(kw) elif desc and _kw_in(kw, desc): score += max(1, w // 2) pos.append(kw) for kw, w in NEGATIVE_KEYWORDS.items(): if _kw_in(kw, title) or (desc and _kw_in(kw, desc)): score += w neg.append(kw) return score, pos, neg def _atomic_write_json(path, data): """Write JSON via temp + os.replace so a crash mid-write cannot truncate state files.""" path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") os.replace(tmp, path) def load_seen(): if STATE_FILE.exists(): return json.loads(STATE_FILE.read_text(encoding="utf-8")) return {} def save_seen(seen): _atomic_write_json(STATE_FILE, seen) def load_decisions(): """Decision log keyed by job URL: {url: {company, title, decision, note, date}}. Decisions persist across runs so we don't re-evaluate roles we've already judged (shortlist / skip / applied / paused / rejected — free-text, not enforced).""" if DECISIONS_FILE.exists(): return json.loads(DECISIONS_FILE.read_text(encoding="utf-8")) return {} def save_decisions(decisions): _atomic_write_json(DECISIONS_FILE, decisions) def load_last_scrape(): """Per-company last successful scraped counts — used to flag sudden 0-job deaths.""" if LAST_SCRAPE_FILE.exists(): try: return json.loads(LAST_SCRAPE_FILE.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): return {} return {} def save_last_scrape(data): _atomic_write_json(LAST_SCRAPE_FILE, data) def _parse_posted(s): """Best-effort parse of an adapter's `posted` field into a date, across the mix of formats the boards use (ISO 8601 incl. trailing Z, YYYY-MM-DD, DD.MM.YYYY). Returns None for unparseable values (e.g. Workday's relative "Posted 5 Days Ago", or empty).""" if not s or not isinstance(s, str): return None s = s.strip() try: return datetime.fromisoformat(s.replace("Z", "+00:00")).date() except ValueError: pass for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%Y/%m/%d", "%d/%m/%Y", "%d-%b-%Y", "%d-%B-%Y"): try: # %b/%B need full token (e.g. 09-Aug-2026), not a fixed 10-char slice token = s.split(",")[0].strip() return datetime.strptime(token if "%b" in fmt or "%B" in fmt else s[:10], fmt).date() except ValueError: pass m = re.search(r"\d{4}-\d{2}-\d{2}", s) if m: try: return datetime.strptime(m.group(0), "%Y-%m-%d").date() except ValueError: pass return None def write_stats_table(stats, total_secs): """Render the per-company scan stats as a markdown table (+ a totals row).""" out = ["## Scan stats\n", "| Company | Scraped | CH/Remote | Match ≥2 | Newest posting | Time (s) |", "|---|--:|--:|--:|:--|--:|"] t_scraped = t_elig = t_match = 0 newest_all = None for s in stats: name = s["company"] + (" ⚠️" if s.get("error") else "") newest = s["newest"].isoformat() if s["newest"] else "—" out.append(f"| {name} | {s['scraped']:,} | {s['eligible']:,} | " f"{s['match']:,} | {newest} | {s['secs']:.1f} |") t_scraped += s["scraped"]; t_elig += s["eligible"]; t_match += s["match"] if s["newest"] and (newest_all is None or s["newest"] > newest_all): newest_all = s["newest"] out.append(f"| **Total ({len(stats)})** | **{t_scraped:,}** | **{t_elig:,}** | " f"**{t_match:,}** | **{newest_all.isoformat() if newest_all else '—'}** | " f"**{total_secs:.1f}** |") out.append("") return out def write_report(path, results, errors, new_only, include_weak, stats=None, total_secs=0.0, decisions=None, hide_decided=False): decisions = decisions or {} today = datetime.now().strftime("%Y-%m-%d") n_new = sum(1 for r in results if r["is_new"]) n_match = sum(1 for r in results if r["score"] >= 2) lines = [ f"# Job scout report {today}{' (new only)' if new_only else ''}\n", f"Automated coverage: **{len(COMPANIES)}** companies. Manual checks: **{len(MANUAL_CHECK)}**.", f"Eligible (CH/remote): **{len(results)}** · interest matches (score ≥ 2): " f"**{n_match}** · **{n_new}** new since last run\n", ] if stats: lines += write_stats_table(stats, total_secs) if errors: lines.append("## Errors\n") for company, err in errors: lines.append(f"- **{company}**: {err}") lines.append("") strong = [r for r in results if r["score"] >= 6] medium = [r for r in results if 2 <= r["score"] < 6] weak = [r for r in results if r["score"] < 2] if not include_weak and weak: lines.append(f"\n_Hiding {len(weak)} weak/noise roles (score < 2). Use --include-weak to show._") n_decided = sum(1 for r in results if r["url"] in decisions) if n_decided: shown = "hidden" if hide_decided else "tagged inline" lines.append(f"_{n_decided} role(s) already in the decision log ({shown}; " f"see state/decisions.json)._") buckets = [("Strong fit (score >= 6)", strong), ("Medium fit (score 2-5)", medium)] if include_weak: buckets.append(("Weak / noise (score < 2)", weak)) for bucket_name, bucket in buckets: shown = [r for r in bucket if not (hide_decided and r["url"] in decisions)] if not shown: continue lines.append(f"\n## {bucket_name} - {len(shown)} role(s)\n") for r in shown: d = decisions.get(r["url"]) new_tag = " [NEW]" if r["is_new"] else "" decided_tag = f" — 🗂 {d['decision'].upper()}" if d else "" loc_tag = ("CH" if r["in_ch"] else "Remote" if r["remote"] else "Relocation" if r.get("relocation") else "?") lines.append(f"### [{r['score']}] {r['company']} - {r['title']}{new_tag}{decided_tag}") lines.append(f"- Location: {r['location']} *({loc_tag})*") if r.get("posted"): lines.append(f"- Posted: {r['posted']}") lines.append(f"- URL: {r['url']}") if d: note = f" — {d['note']}" if d.get("note") else "" lines.append(f"- 🗂 Decision: **{d['decision']}**{note} ({d.get('date','')})") if r["pos"]: lines.append(f"- Positive: {', '.join(r['pos'])}") if r["neg"]: lines.append(f"- Negative: {', '.join(r['neg'])}") lines.append("") if MANUAL_CHECK: lines.append("\n## Manual check (companies without scrapable APIs)\n") lines.append("These use Cloudflare-protected sites, custom GraphQL APIs, or JS-rendered SPAs.") lines.append("Open each link, scan for new postings since your last quarterly review:\n") for name, note, url in MANUAL_CHECK: lines.append(f"- [ ] **{name}** — {note}: <{url}>") lines.append("") path.write_text("\n".join(lines), encoding="utf-8") def _fetch_company(cid, display, adapter, args): """Fetch one company; returns (cid, display, adapter, args, jobs|None, error|None, secs).""" t0 = time.perf_counter() try: jobs = ADAPTERS[adapter](args) return cid, display, adapter, args, jobs, None, time.perf_counter() - t0 except Exception as e: kind = "unexpected" if not isinstance( e, (urllib.error.URLError, urllib.error.HTTPError, ValueError, RuntimeError) ) else "error" msg = f"{kind}: {e!r}" if kind == "unexpected" else repr(e) return cid, display, adapter, args, None, msg, time.perf_counter() - t0 def _process_company(cid, display, args, jobs, seen, today, last_scrape): """Filter, score, and update seen for one company's job list. Returns (results, stat_dict, warnings) where warnings is a list of (display, msg). """ warnings = [] scraped = len(jobs) prev = (last_scrape.get(cid) or {}).get("scraped", 0) if scraped == 0: if prev and prev > 0: warnings.append(( display, f"0 jobs returned (was {prev} last run — possible dead board/slug/selectors)", )) else: warnings.append(( display, "0 jobs returned (verify board slug/selectors if this is unexpected)", )) # Inclusion allowlist (fails closed — only for boards too large to score in full). title_filter = args.get("_title_filter") if title_filter: jobs = [j for j in jobs if any(_kw_in(k, (j.get("title") or "").lower()) for k in title_filter)] # Exclusion denylist (fails open — preferred; see NOISE_TITLE_EXCLUDE). title_exclude = args.get("_title_exclude") if title_exclude: jobs = [j for j in jobs if not any(_kw_in(k, (j.get("title") or "").lower()) for k in title_exclude)] dates = [d for j in jobs if (d := _parse_posted(j.get("posted")))] newest = max(dates) if dates else None policy = args.get("_location_policy", "default") company_seen = seen.setdefault(cid, {}) title_seen = set() results = [] eligible = match = 0 for j in jobs: jid = str(j.get("id") or j.get("url")) ok, in_ch, is_remote, is_relocation = location_eligibility( j.get("location", ""), policy=policy ) if not ok: continue # Collapse the same role posted once per remote country (title differs only # by a "| Country | Remote" suffix) — dedupe on the title before the first "|". norm_title = re.sub(r"\s+", " ", (j.get("title") or "").split("|")[0]).strip().lower() if norm_title in title_seen: continue title_seen.add(norm_title) eligible += 1 is_new = jid not in company_seen score, pos, neg = score_job(j, title_only=bool(title_filter)) # Pre-filtered boards (e.g. SBB) with German/generic titles: _score_floor keeps # already-relevant results out of the hidden weak bucket. floor = args.get("_score_floor") if floor is not None and score < floor: score = floor if score >= 2: match += 1 results.append({ "company": display, "company_id": cid, "title": j["title"], "location": j["location"], "url": j["url"], "posted": j.get("posted", ""), "score": score, "pos": pos, "neg": neg, "in_ch": in_ch, "remote": is_remote, "relocation": is_relocation, "is_new": is_new, }) if is_new: company_seen[jid] = {"title": j["title"], "first_seen": today} stat = { "company": display, "scraped": scraped, "eligible": eligible, "match": match, "newest": newest, "error": False, } return results, stat, warnings def main(): today = datetime.now(timezone.utc).strftime("%Y-%m-%d") # Record a decision and exit: --decide "<url>" <status> [note words...] if "--decide" in sys.argv: rest = sys.argv[sys.argv.index("--decide") + 1:] if len(rest) < 2: print('Usage: --decide "<url>" <status> [note...]', file=sys.stderr) return url, status, note = rest[0], rest[1], " ".join(rest[2:]) decisions = load_decisions() prev = decisions.get(url, {}) # Preserve any company/title already stored (filled when a report run tagged the URL). decisions[url] = { "company": prev.get("company", ""), "title": prev.get("title", ""), "decision": status, "note": note, "date": today, } save_decisions(decisions) print(f"Recorded: {status} — {url}", file=sys.stderr) return only, new_only, include_weak, hide_decided = None, False, False, False for arg in sys.argv[1:]: if arg in ("--help", "-h"): print(__doc__) return if arg == "--new-only": new_only = True elif arg == "--include-weak": include_weak = True elif arg == "--hide-decided": hide_decided = True elif arg.startswith("--only="): only = arg.split("=", 1)[1] targets = [(cid, display, adapter, args) for cid, display, adapter, args in COMPANIES if not only or cid == only] if only and not targets: known = ", ".join(c[0] for c in COMPANIES) print(f"Unknown --only={only!r}. Known ids: {known}", file=sys.stderr) sys.exit(2) seen = load_seen() decisions = load_decisions() last_scrape = load_last_scrape() all_results, errors, stats = [], [], [] new_last_scrape = dict(last_scrape) run_start = time.perf_counter() # API adapters in parallel; playwright serial (shared browser is not thread-safe). api_targets = [t for t in targets if t[2] != "playwright"] pw_targets = [t for t in targets if t[2] == "playwright"] # Preserve COMPANIES order in stats/report via ordered result map. fetch_results = {} try: if api_targets: print(f"Fetching {len(api_targets)} API boards " f"({API_FETCH_WORKERS} workers)...", file=sys.stderr) with ThreadPoolExecutor(max_workers=API_FETCH_WORKERS) as pool: futs = { pool.submit(_fetch_company, cid, display, adapter, args): cid for cid, display, adapter, args in api_targets } for fut in as_completed(futs): cid, display, adapter, args, jobs, err, secs = fut.result() print(f" {display}: " f"{'ERROR' if err else f'{len(jobs)} jobs'} ({secs:.1f}s)", file=sys.stderr) fetch_results[cid] = (display, args, jobs, err, secs) for cid, display, adapter, args in pw_targets: print(f"Fetching {display} (playwright)...", file=sys.stderr) _, display, _, args, jobs, err, secs = _fetch_company( cid, display, adapter, args ) print(f" {display}: " f"{'ERROR' if err else f'{len(jobs)} jobs'} ({secs:.1f}s)", file=sys.stderr) fetch_results[cid] = (display, args, jobs, err, secs) # Process in COMPANIES order for stable stats tables. for cid, display, adapter, args in targets: display, args, jobs, err, secs = fetch_results[cid] if err is not None: errors.append((display, err)) stats.append({ "company": display, "scraped": 0, "eligible": 0, "match": 0, "newest": None, "secs": secs, "error": True, }) continue results, stat, warnings = _process_company( cid, display, args, jobs, seen, today, last_scrape ) stat["secs"] = secs for w in warnings: errors.append(w) # Soft warning only when 0 jobs — still mark row with ⚠️ if was previously >0 prev = (last_scrape.get(cid) or {}).get("scraped", 0) if stat["scraped"] == 0 and prev > 0: stat["error"] = True stats.append(stat) all_results.extend(results) new_last_scrape[cid] = { "scraped": stat["scraped"], "date": today, "display": display, } finally: save_seen(seen) save_last_scrape(new_last_scrape) _close_browser() total_secs = time.perf_counter() - run_start if new_only: all_results = [r for r in all_results if r["is_new"]] all_results.sort(key=lambda r: (-r["score"], r["company"], r["title"])) REPORTS_DIR.mkdir(parents=True, exist_ok=True) report_path = REPORTS_DIR / f"{today}.md" write_report(report_path, all_results, errors, new_only, include_weak, stats=stats, total_secs=total_secs, decisions=decisions, hide_decided=hide_decided) # Plain data dump of every eligible role, unfiltered by keyword score — fit judgment # for these is done in conversation against the profile, not by the keyword scorer. json_path = REPORTS_DIR / f"{today}.json" dump = [{ "company": r["company"], "title": r["title"], "location": r["location"], "url": r["url"], "posted": r["posted"], "is_new": r["is_new"], "decision": decisions.get(r["url"]), "keyword_signal": {"score": r["score"], "pos": r["pos"], "neg": r["neg"]}, "loc_flags": { "in_ch": r["in_ch"], "remote": r["remote"], "relocation": r["relocation"], }, } for r in all_results] _atomic_write_json(json_path, dump) n_new = sum(1 for r in all_results if r["is_new"]) print(f"\nReport written: {report_path}", file=sys.stderr) print(f"Total matches: {len(all_results)} ({n_new} new) | " f"scanned {len(stats)} companies in {total_secs:.1f}s", file=sys.stderr) if errors: print(f"Errors/warnings: {len(errors)} - see report", file=sys.stderr) # === Adapter coverage (refreshed 2026-07-18) ================================== # 34 companies automated across 14 adapter types; 1 in MANUAL_CHECK (Oracle). # # Location policy: # default — CH + Denmark + Europe/EMEA/global remote (work-from CH or DK) # norway_only — Equinor # dk_no — NATO (Denmark + Norway only; no CH) # # Automated (COMPANIES above): # workday nvidia, novartis, equinor (norway_only) # ashby kraken, openai, confluent, snowflake # greenhouse anthropic, gitlab, grafana, databricks, datadog, elastic, fivetran # pcsx microsoft # smartrecruiters metgroup, ldc # rss bis # amazon amazon (Amazon + AWS share one board) # phenom roche (Cisco still uses the playwright path) # teamtailor telenor, axpo (custom domain via base_url) # getro (no company configured — adapter kept for future VC talent networks) # onlyfy bitcoin_suisse # lever palantir, quantco # taleo nato (dk_no; nato.taleo.net §2 — marketing page is nato.int/vacancies) # json swissgrid # sbb sbb # bkw bkw # playwright google, apple, meta, cisco, ruag, postfinance, bfh # # MANUAL_CHECK: Oracle (ORC needs CH geographyId). # ============================================================================== if __name__ == "__main_

"""Job scout for Dennis's quarterly target companies. Pulls latest openings from companies via public ATS APIs (Workday/Ashby/Greenhouse/ SmartRecruiters/Lever/Eightfold/RSS) and, for JS-rendered careers sites, a headless-browser (playwright) adapter. Filters by location policy, scores fit against profile keywords, tracks which job IDs we've already seen, writes a markdown report. Location policy (default for most companies): - Switzerland onsite/hybrid (CH) - Denmark onsite/reloc - CH-remote and Europe/EMEA/global remote (work-from CH or DK is realistic) - NOT multi-country allowlists that omit CH/DK/Europe (e.g. "UK | Brazil | Ireland | ...") Company overrides via args["_location_policy"]: - "norway_only" — Equinor: Norway only - "dk_no" — NATO-style: Denmark + Norway only (no CH — no NATO seat in CH) Usage: py scout.py # Pull all configured companies (strong + medium only) py scout.py --only=nvidia # Pull a single company by id py scout.py --new-only # Report only jobs not seen before py scout.py --include-weak # Include weak/noise bucket (default hidden) py scout.py --hide-decided # Drop roles already in the decision log (undecided-only view) py scout.py --decide "" [note...] # Record a decision and exit # status is free-text: shortlist | skip | applied | paused | ... State : state/seen_jobs.json · state/decisions.json · state/last_scrape.json Output: reports/YYYY-MM-DD.md (+ .json dump of every eligible role) To add a company: append to COMPANIES with one of the existing adapter types. A few sites resist scraping even headless and stay in MANUAL_CHECK (surfaced as a report checklist). See the adapter-coverage notes at the bottom for the current automated/manual split. """ import json import os import re import ssl import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed from functools import lru_cache import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).parent STATE_FILE = ROOT / "state" / "seen_jobs.json" DECISIONS_FILE = ROOT / "state" / "decisions.json" LAST_SCRAPE_FILE = ROOT / "state" / "last_scrape.json" REPORTS_DIR = ROOT / "reports" USER_AGENT = "Mozilla/5.0 (compatible; job-scout/0.1)" API_FETCH_WORKERS = 8 CH_LOCATION_KEYWORDS = [ "switzerland", "zurich", "zürich", "basel", "bern", "geneva", "genf", "lausanne", "zug", "rüschlikon", "stäfa", "schweiz", "suisse", ] DK_LOCATION_KEYWORDS = [ "denmark", "copenhagen", "danmark", "københavn", "koebenhavn", ] NO_LOCATION_KEYWORDS = [ "norway", "oslo", "stavanger", "bergen", "trondheim", "norge", ] US_ONLY_PATTERNS = [ "remote - us", "remote, us", "remote-us", "us remote", "us-remote", "remote-friendly us", "remote (us)", "united states - remote", "remote, united states", ] # Pan-regional remote scope: work-from CH or DK is realistic. EUROPE_SCOPE_KEYWORDS = [ "europe", "emea", "european", "eu-wide", "eu wide", "remote europe", "europe remote", "remote - europe", "remote, europe", ] GLOBAL_SCOPE_KEYWORDS = [ "global", "worldwide", "work from anywhere", "anywhere in the world", ] POSITIVE_KEYWORDS = { "genai": 3, "generative ai": 3, "llm": 3, "large language model": 3, "applied ai": 3, "applied ml": 3, "ai engineer": 3, "ml engineer": 3, "mlops": 3, "ai platform": 3, "ml platform": 3, "python": 2, "java": 2, "data engineer": 2, "data engineering": 2, # "data scientist" scored modestly (medium, not strong) — secondary to his data-eng/ # platform thesis, but the targeted band at boutiques like QuantCo (see target memory). "data scientist": 2, "data science": 2, "solutions architect": 2, "platform engineer": 2, "ai infrastructure": 2, "inference": 2, "rag": 2, "agentic": 2, # Spelled-out "Artificial Intelligence" (int'l orgs like BIS write it out instead of # "AI") and analytics-engineer variants — was missing entirely, dropped a genuinely # on-thesis BIS role (Senior Data & Analytics Engineer - AI) into the noise bucket. "artificial intelligence": 3, "analytics engineer": 2, "data analytics": 2, "kubernetes": 1, "docker": 1, "etl": 1, "pipeline": 1, # Core CV lane — DevOps / data-platform / cloud (was scoring 0; surfaced only via "senior") "data platform": 3, "platform engineering": 2, "devops": 2, "sre": 2, "site reliability": 2, "cloud engineer": 2, "cloud": 1, "software engineer": 1, # Technical-architect pivot targets (cloud/data/platform = build on his stack; rank above # bare "solutions architect" pre-sales). Generic "architect" catches the long tail. "cloud architect": 3, "data architect": 3, "platform architect": 3, "enterprise architect": 2, "architect": 1, "crypto": 2, "blockchain": 2, "web3": 2, "solidity": 3, # Trading / quant-finance — explicit user interest (energy/finance/crypto trading) "trading": 2, "trader": 2, "quant": 2, "quantitative": 2, "market data": 2, "low latency": 2, "low-latency": 2, "fix protocol": 2, "brokerage": 2, "commodity": 1, "execution": 1, # "solutions architect" (plural) already scored above; add singular + adjacent stack "solution architect": 2, "c#": 1, ".net": 1, # Forward-deployed / field / resident-architect lane — vendor "deploy-to-customer" roles # where Dennis's multi-country ramp is the differentiator (see user_international_mobility). # Travel-from-Bern fits his mobility appetite; no relocation required. "forward deployed": 3, "forward-deployed": 3, "field engineer": 3, "resident architect": 3, "resident engineer": 3, "resident solutions architect": 3, "customer engineer": 2, "senior": 1, "staff": 1, "lead": 1, "principal": 1, } NEGATIVE_KEYWORDS = { "cuda": -3, "kernel driver": -3, "gpu programming": -3, "compiler engineer": -3, "pytorch internals": -3, "jax internals": -3, "rdma": -2, "infiniband": -2, "nccl": -3, "hpc cluster": -2, "frontend": -3, "front-end": -3, "react native": -3, "ios engineer": -3, "android engineer": -3, "mobile engineer": -3, "ui engineer": -2, "ux engineer": -2, "verilog": -3, "vhdl": -3, "asic": -3, "rtl design": -3, "physical design": -3, "silicon": -2, "expert c++": -2, "5+ years c++": -2, "deep c++": -2, "intern": -5, "internship": -5, "graduate program": -3, "junior": -3, } # Title prefilter for high-volume boards (all-remote tech orgs + commodity traders that # post mostly non-tech roles). Only keep titles containing one of these specific role # phrases — kept tight so "Sales Engineer"/"Staff Accountant"/"Data Privacy Counsel" # don't leak in. Matched as case-insensitive substrings against the title only. # Inclusion allowlist. Applied ONLY to boards too large to score in full (>~200 roles: # Databricks, Snowflake, Datadog, Elastic, Fivetran, Louis Dreyfus, Palantir). Everywhere else # use NOISE_TITLE_EXCLUDE — see the design note there for why fail-open is the default. ENG_TITLE_FILTER = [ "data engineer", "data engineering", "data platform", "platform engineer", "data infrastructure", "data architect", "analytics engineer", "mlops", "ml engineer", "ml platform", "machine learning engineer", "site reliability", "sre", "backend engineer", "back-end engineer", "devops engineer", "cloud engineer", "software engineer", "infrastructure engineer", "kafka", "streaming", "big data", "quantitative developer", "quant developer", # Forward-deployed / field / resident lane (vendor boards) — see user_international_mobility. # "resident" alone catches Resident Solutions Architect/Engineer without opening the gate to # all pre-sales SAs (the overscoring trap); "customer engineer" is Google's field-eng term. "forward deployed", "forward-deployed", "field engineer", "resident", "customer engineer", # Added 2026-08-18 after the CH boards were title-filtered: without these, real near-misses # were dropped (BFH "Wissenschaftliche Mitarbeit Data-Driven Government", Novartis "Director # & Group Head (AI-Systems & Scale)"). Deliberately NOT adding bare "ai" — it substring-matches # Maintenance/Training/Chair and floods every board. "artificial intelligence", "ai-systems", "ai platform", "ai engineer", "ai architect", "data scientist", "data-driven", "data science", ] # Generic non-tech title exclusions, shared by every board that uses exclusion-mode filtering. # # DESIGN NOTE (2026-08-18). Boards are filtered one of two ways: # * inclusion (`_title_filter`) — an allowlist, used ONLY on boards too large to score in # full (Databricks ~800, OpenAI ~730, Palantir ~300). It fails CLOSED: a great-fit role # with a title nobody anticipated is dropped at fetch time and never reaches the scorer, # so it appears in no report and in no JSON dump. That is an acceptable trade only where # volume forces it. # * exclusion (`_title_exclude`) — this list. It fails OPEN: everything survives unless it # is *clearly* not an engineering role, and the score + the weak/noise bucket decide what # surfaces. Prefer this. The final call belongs to the reviewer, not to a keyword gate. # # Keep these terms unambiguous. Anything that could plausibly attach to a technical role # (analyst, manager, specialist, consultant, architect, lead) must NOT go in here. NOISE_TITLE_EXCLUDE = [ # Facilities / retail / hospitality / admin "hausdienst", "reinigung", "empfang", "hauswart", "chef de partie", "koch", "küche", "restaurant", "catering", "fahrer", "logistik mitarbeiter", "lagerist", "verkauf", "verkäufer", "retail", "barista", "security guard", "sicherheitsdienst", # Care / health / teaching-of-non-tech "physiotherapie", "pflege", "pflegefach", "hebamme", "ergotherapie", "psychologie", "medical representative", "nurse", "arzt", "ärztin", "dentist", # Arts / music / sport "violine", "klavier", "musik", "dozierende*r violine", "sport", # Back-office "fundraising", "buchhaltung", "accountant", "accounting", "payroll", "steuer", "recruiter", "talent acquisition", "human resources", "personalwesen", "legal counsel", "rechtsanwalt", "rechtsreferendar", "notar", "jurist", "kommunikation", "public relations", "übersetzer", "translator", # Early-career / non-role listings "lehrstelle", "praktikum", "praktikant", "internship", "intern ", "trainee", "apprenti", "ausbildung", "schnupper", "where we're hiring", "talent community", # Non-software engineering trades "elektroplaner", "elektroinstallat", "sanitär", "heizung", "maler", "schreiner", "hochspannung", "wasserbau", "strassenbau", "holzbau", "bauingenieur", "bauleiter", ] # id, display, adapter, adapter_args COMPANIES = [ ("nvidia", "NVIDIA", "workday", { "host": "nvidia.wd5.myworkdayjobs.com", "tenant": "nvidia", "site": "NVIDIAExternalCareerSite", "search_text": "Switzerland", }), ("kraken", "Kraken", "ashby", {"slug": "kraken.com"}), ("openai", "OpenAI", "ashby", {"slug": "openai"}), ("anthropic", "Anthropic", "greenhouse", {"board": "anthropic"}), ("novartis", "Novartis", "workday", { "host": "novartis.wd3.myworkdayjobs.com", "tenant": "novartis", "site": "Novartis_Careers", "search_text": "Switzerland", "_title_exclude": NOISE_TITLE_EXCLUDE, }), # PCSX (Eightfold) — Microsoft has a public position search endpoint ("microsoft", "Microsoft", "pcsx", { "domain": "microsoft.com", "location": "Switzerland", }), # --- Data-infra US tech (his exact stack; mostly all-remote — title-filtered to eng/data) --- # Dropped: ClickHouse (Glassdoor 3.3, 36% recommend, toxic-culture flag — 2026-05). # Dropped: HashiCorp — acquired by IBM (closed 2025); greenhouse/ashby/lever boards all 404, # roles folded into IBM's careers (no clean public ATS API). 2026-06-06. ("confluent", "Confluent", "ashby", {"slug": "confluent", "_title_exclude": NOISE_TITLE_EXCLUDE}), ("gitlab", "GitLab", "greenhouse", {"board": "gitlab", "_title_exclude": NOISE_TITLE_EXCLUDE}), ("grafana", "Grafana Labs","greenhouse",{"board": "grafanalabs", "_title_exclude": NOISE_TITLE_EXCLUDE}), # Added 2026-06-06 (Tier A/B data-infra). Databricks/Snowflake/Datadog have Zürich offices # (Swiss-scale comp, clears bar); Elastic/Fivetran are remote-EU (verify CH-equiv comp — # may be geo-banded below 180k, like Grafana). All title-filtered (boards are 160-760 roles). ("databricks","Databricks","greenhouse", {"board": "databricks", "_title_filter": ENG_TITLE_FILTER}), # Zürich SWE + remote-EU; DK via default location policy ("snowflake", "Snowflake", "ashby", {"slug": "snowflake", "_title_filter": ENG_TITLE_FILTER}), # Zürich "Observe" observability SWE roles ("datadog", "Datadog", "greenhouse", {"board": "datadog", "_title_filter": ENG_TITLE_FILTER}), # Zürich branch + remote-EU ("elastic", "Elastic", "greenhouse", {"board": "elastic", "_title_filter": ENG_TITLE_FILTER}), # remote-first; ELK = his stack # dbt Labs -> Fivetran: the "dbtlabsinc" board 404'd from 2026-08-18. getdbt.com/careers now # redirects to a page serving Fivetran's Greenhouse listings (gh_jid links) — the two merged, so # the dbt roles live on board "fivetran". EMEA presence is Dublin/London/Serbia, no CH office, so # most roles fall out on the location policy; kept for remote-EU analytics-eng reqs. 2026-08-18. ("fivetran", "Fivetran (ex-dbt Labs)", "greenhouse", {"board": "fivetran", "_title_filter": ENG_TITLE_FILTER}), # --- Energy / commodity trading (SmartRecruiters; title-filtered to tech roles) --- # Dropped: Vitol (Glassdoor 3.5, 55% recommend, grueling-hours/toxic flag — 2026-05). # Dropped: Sygnum (Glassdoor 3.4, 51% recommend, comp 2.3/5 — below 180k bar — 2026-05). # Axpo (Baden/Zurich) — Teamtailor on a custom domain; no .teamtailor.com host # resolves, so this uses base_url. Added 2026-08-18: memory names Axpo/Alpiq as the # energy-trading targets but only MET was ever configured, and MET+LDC have produced # 0 CH-eligible roles across 1,464 scrapes. ~400 roles, German-speaking region (which is # what rules out the Geneva traders), incl. a "Forward Deployed AI Engineer" at add time. ("axpo", "Axpo", "teamtailor", { "base_url": "https://careers.axpo.com", "default_location": "Switzerland", "_title_exclude": NOISE_TITLE_EXCLUDE, }), ("metgroup", "MET Group", "smartrecruiters", {"company": "METGroup", "_title_exclude": NOISE_TITLE_EXCLUDE}), ("ldc", "Louis Dreyfus","smartrecruiters",{"company": "LouisDreyfusCompany", "_title_filter": ENG_TITLE_FILTER}), # Equinor (Workday) — Norway energy major; lived/worked in NO before. Outlier location # policy: Norway only (not CH / Europe-remote). Small board (~15); no title filter. ("equinor", "Equinor", "workday", { "host": "equinor.wd3.myworkdayjobs.com", "tenant": "equinor", "site": "EQNR", "_location_policy": "norway_only", # Norway-only filter already scopes the board; titles are often generic/energy # ops and score 0 under the English eng keyword list — keep them visible. "_score_floor": 2, }), # Telenor Norge (Teamtailor) — Nordic telco; Swisscom is the direct domain analogue # and Dennis worked in Norway before (Vizrt, Bergen). NOTE: the old Workday tenant # telenorgroup.wd3.myworkdayjobs.com is dead (HTTP 422 on /wday/cxs, 2026-07-29) — # Telenor migrated to Teamtailor. Slug telenorgroup is a near-empty group-HQ board; # telenornorway is the real Norway board (redirects to careers.telenor.no). # CAVEAT: most Telenor Norge postings require Norwegian ("norsk og engelsk") and are # Oslo/Trondheim onsite-hybrid — verify both per role before tailoring. # Norwegian-language titles/descriptions score ~0 under the English eng keyword # list, so keep a score floor (same pattern as Equinor). ("telenor", "Telenor Norge", "teamtailor", { "slug": "telenornorway", "default_location": "Oslo/Trondheim, Norway", "_location_policy": "norway_only", "_score_floor": 2, }), # NATO civilian vacancies (Oracle Taleo). Public marketing page is # https://www.nato.int/en/work-with-us/careers/vacancies — jobs actually live on # nato.taleo.net career section 2. Location policy dk_no: Denmark + Norway only # (no NATO seat in CH). Server-side location taxonomy IDs filter the board before # parse; dk_no is a second safety net on the location string. # Taleo option values (jobsearch.ftl dropdown): Norway=14470035151, Denmark=16270035151. ("nato", "NATO", "taleo", { "host": "nato.taleo.net", "career_section": "2", "location_ids": ["14470035151", "16270035151"], # Norway, Denmark "_location_policy": "dk_no", # Staff Officer / NATO-body titles score unevenly under eng keywords — keep # pre-filtered NO/DK roles visible (same pattern as Equinor/SBB). "_score_floor": 2, }), # International org — BIS (Basel), commutable from Bern, salary net of Swiss tax. # Low-volume RSS feed; no title filter (Innovation Hub roles can be oddly titled). ("bis", "BIS (Basel)","rss", { "url": "https://www.bis.org/doclist/vacancies.rss", "default_location": "Basel, Switzerland", }), # Dropped: Coinbase Ventures web3 talent network (Getro collection 1625) — 2026-08-18. # Never carried Coinbase's own roles, only portfolio companies. Across ~3 months # it surfaced 3 distinct CH-eligible roles, ALL from Ashby (recruiting software, # ruled out 2026-07-28) and all decided "skip". With Ashby excluded the board has # returned 0 jobs on every run since, reported as a false error each time. # fetch_getro is retained for a future VC talent network worth tracking. # Bitcoin Suisse (Zug) uses the onlyfy.jobs ATS. No title filter — small crypto # firm, only a handful of CH roles; let scoring rank them (CH filter does the rest). ("bitcoin_suisse", "Bitcoin Suisse", "onlyfy", {"slug": "bitcoin-suisse"}), # Headless-browser scrapers — slower (3-15s per company) but covers JS-rendered sites. # Google actively bot-detects; the STEALTH_JS init script (applied to every context) # is what makes its job list render. Cards are

with a "Learn more about " # aria-label link; location lives in the card text (captured via blob mode). ("google", "Google", "playwright", { "url": "https://www.google.com/about/careers/applications/jobs/results/?location=Switzerland", "wait_for": "a[href*='jobs/results/'][aria-label*='Learn more']", "card": "li:has(a[aria-label*='Learn more about'])", "title_sel": "a[aria-label*='Learn more about']", "title_sel_attr": "aria-label", "title_strip_prefix": "Learn more about ", "link_sel": "a[href*='jobs/results/']", "link_attr": "href", "url_prefix": "https://www.google.com/about/careers/applications/", "default_location": "", "scroll_count": 5, "use_inner_text_as_blob": True, "cookie_accept": ["button:has-text('Accept all')", "button:has-text('Reject all')"], # Results are paged 20 at a time and "page" is 1-indexed — without this only the # first 20 of ~48 Zürich roles were ever scraped. "page_param": "page", "page_param_start": 2, "max_pages": 6, }), # Amazon + AWS share one board. Added 2026-08-18: AWS is the cloud the evidence base # actually supports (Swisscom migration + data products), whereas claims.json marks GCP # output-forbidden — so an AWS-native Zurich FDE req is the Google FDE lane without the # cloud gap. 32 CH roles at time of adding, incl. "Senior Forward Deployed Engineer, AWS # Forward Deployed Engineering" (Zurich). # AWS names its delivery/field org differently from every other board, so the shared # filter caught only 1 of 32 CH roles. These four terms are added narrowly and ONLY here: # "solutions architect" is the classic overscoring trap (see job_scout_overscoring_findings), # but on the AWS board it is the delivery-side title attached to migration/data work, which # is exactly SW-1. Read the JD before trusting the score on any of these. ("amazon", "Amazon / AWS", "amazon", { "countries": ["CHE"], "_title_exclude": NOISE_TITLE_EXCLUDE, }), ("apple", "Apple", "playwright", { "url": "https://jobs.apple.com/en-us/search?location=switzerland-CHE", "wait_for": "a[href*='/en-us/details/']", "card": "a[href*='/en-us/details/']", "title_attr": "text", "link_attr": "href", "url_prefix": "https://jobs.apple.com", # NO default_location. Apple's own postLocation-CHE filter leaks global reqs whose # postLocationId is postLocation-USA ("Various Locations within United States"), and # a forced "Switzerland" default relabelled every one of them as CH-eligible — 84 # phantom CH rows over 4 runs, all internships/retail, which then read as "Apple is a # bad fit" in the decision log. Read the real location off the card instead; when # Apple has no Swiss reqs the honest answer is 0. Verified 2026-08-18. "use_inner_text_as_blob": True, "scroll_count": 5, }), # Meta job links are /profile/job_details/<id>; title + location are in the link text. # NOT broken despite returning ~1 role: verified 2026-08-18 that metacareers itself # reports "1 Items" for the Zurich office filter. The board is genuinely near-empty; # don't "fix" this scraper without first checking the live item count on the page. ("meta", "Meta", "playwright", { "url": "https://www.metacareers.com/jobs?offices[0]=Zurich%2C%20Switzerland", "wait_for": "a[href*='/profile/job_details/']", "card": "a[href*='/profile/job_details/']", "title_attr": "text", "link_attr": "href", "url_prefix": "https://www.metacareers.com", "default_location": "Zurich, Switzerland", "scroll_count": 5, "use_inner_text_as_blob": True, }), # Roche: was a playwright scrape of the search-results page until 2026-08-18. That page's # ?locationsearch=Switzerland filter silently fails — the page state reports "no-results" # and the scrape harvested *recommendation-widget* cards instead, returning Shanghai, # Kyiv, Bogota and Mannheim and zero Swiss roles for months while looking healthy. The # underlying Phenom refineSearch endpoint filters correctly (134 CH roles), so hit it # directly. Title-filtered: Roche CH is overwhelmingly lab/pharma/apprenticeship reqs. ("roche", "Roche", "phenom", { "url": "https://careers.roche.com/widgets", "ref_num": "ROCHGLOBAL", "page_id": "page11-ds", "country_facet": "Switzerland", "_title_exclude": NOISE_TITLE_EXCLUDE, }), # Cisco (PhenomPeople, new careers.cisco.com domain). Keyword search surfaces CH roles. ("cisco", "Cisco", "playwright", { "url": "https://careers.cisco.com/global/en/search-results?keywords=Switzerland", "wait_for": "a[href*='/job/'], div[role='listitem']", "card": "div[role='listitem']:has(a[href*='/job/'])", "title_sel": "a[href*='/job/']", "link_sel": "a[href*='/job/']", "link_attr": "href", "url_prefix": "https://careers.cisco.com", "default_location": "Switzerland", "cookie_accept": ["#onetrust-accept-btn-handler"], "scroll_count": 5, "use_inner_text_as_blob": True, }), # --- Zürich/Zug high-comp additions (2026-05-31 list review) --- # Palantir (Lever). Verified: 221 postings on the public board. It's US/London-heavy, so # Swiss/Schwyz roles are rare but self-surface when posted (the location filter drops the # US/London bulk). No title filter: his target titles (Forward Deployed Software Engineer, # Deployment Strategist) aren't in ENG_TITLE_FILTER, so filtering would hide them. ("palantir", "Palantir", "lever", {"slug": "palantir"}), # QuantCo (Lever — note the trailing-hyphen slug "quantco-"). ~16 roles, most tagged # "Europe" (hybrid); QuantCo's continental hub is Zürich, so Europe-scope in # location_eligibility surfaces them. No title filter: target band is DS/Quant/AI/Cloud # (ENG_TITLE_FILTER would drop some); interns/frontend caught by NEGATIVE_KEYWORDS. ("quantco", "QuantCo", "lever", {"slug": "quantco-"}), # --- Bern/Thun local tier — WLB & proximity exception (comp bar relaxed; 2026-06-01) --- # Wired after live endpoint discovery. ⚠️ German citizen: RUAG classified work may require # Swiss citizenship — verify per-role before tailoring (see project_target_companies). # Swissgrid (Aarau): Magnolia CMS JSON endpoint (verified). placeOfWork is a bare city # (Aarau/Prilly/...), so loc_suffix tags it Switzerland for the CH filter. No title filter # (small board ~13 roles; lets Data Scientist / Applied-ML roles surface). ("swissgrid", "Swissgrid (Aarau)", "json", { "url": "https://www.swissgrid.ch/.rest/cloud/component-data?path=%2Fswissgrid%2Fen%2Fhome%2Fcareer%2Fjobs%2Fmain%2Fjoblist_transferred_11", "jobs_key": "jobs", "field_title": "title", "field_location": "placeOfWork", "field_url": "descriptionUrl", "field_date": "onlineSince", "loc_suffix": " Switzerland", "desc_keys": ["department", "typeOfEmployment", "entryLevel"], "_title_exclude": NOISE_TITLE_EXCLUDE, }), # RUAG (Thun/Bern/Emmen). Jobs render on the portal as anchors to jobs.ruag.ch; the first # line of each anchor is the title. All sites are Swiss, so default_location=Switzerland # passes the CH filter. ENG_TITLE_FILTER cuts the apprenticeship/Lehrstelle bulk. # Drupal portal: 20 jobs/page, server-rendered, paginated via ?page=N (0-indexed). The # first page is apprenticeship-heavy; eng roles (DevOps/Data/Cloud) are on later pages, # so we page through until a page adds nothing new (~5-6 pages). ("ruag", "RUAG (Thun/Bern)", "playwright", { "url": "https://www.ruag.ch/en/working-us/job-portal", "wait_for": "a[href*='/offene-stellen/']", "card": "a[href*='/offene-stellen/']", "title_attr": "text", "link_attr": "href", "default_location": "Switzerland", "scroll_count": 1, "page_param": "page", "max_pages": 10, "_title_exclude": NOISE_TITLE_EXCLUDE, }), # SBB (company.sbb.ch — the correct host; company-jobs.sbb.ch was wrong). AEM job filter # served as a flat JSON list; the fetch_sbb adapter replicates the user's IT + Bern-region # filter. German/generic titles, so _score_floor keeps the pre-filtered results visible. # ⚠️ DE-citizen limits may apply to some SBB security/critical-infra roles. ("sbb", "SBB", "sbb", { "topic": "IT / Telekommunikation", "region": "Bern Mittelland", "_score_floor": 2, "_title_exclude": NOISE_TITLE_EXCLUDE, }), # BKW Group (jobs.bkw.com — the real ATS host). PMS structured-data API; ~600 roles # group-wide, so fetch_bkw keeps only Berufsfeld categories Informatik/Trading/Finanzen # (IT/data + energy-trading, incl. the flagged Energiehandel roles). German/generic # titles, so _score_floor keeps the pre-filtered set visible. ("bkw", "BKW (Bern)", "bkw", {"_score_floor": 2, "_title_exclude": NOISE_TITLE_EXCLUDE}), # PostFinance (Bern). The careers site renders a small, client-side paginated board; # scrape all pages through its stable next-page control. No title filter: the board is # low-volume, and the scorer keeps unrelated banking/customer-service roles out of the # default report while retaining locally relevant engineering variants. ("postfinance", "PostFinance (Bern)", "playwright", { "url": "https://jobs.postfinance.ch/PostFinance/search?locale=de_DE", "wait_for": "a[href*='/PostFinance/job/']", "card": "a[href*='/PostFinance/job/']", "title_attr": "text", "link_attr": "href", "default_location": "Switzerland", "use_inner_text_as_blob": True, "next_button": "#pfch-pagination-next", "max_pages": 10, "_title_exclude": NOISE_TITLE_EXCLUDE, }), # BFH (Bern University of Applied Sciences). Re-added 2026-07-14: the jobs.bfh.ch domain # itself is a broken/stub SPA shell (renders "Career Center project template", nothing # else — this is what looked like a 403 previously), but the actual listing widget is # embedded and fully renders on the www.bfh.ch careers page. Card = one <div> per posting # with 3 li.teaser-text-list-item fields (department, location, employment type); location # is the 2nd. Low-volume (12 open reqs), single page. German titles score 0 under the # English-keyword scorer (same problem as SBB/BKW) — _score_floor keeps them visible # instead of hidden as "noise", e.g. "Professur Data-Driven Government", # "Co-Leitung ... digitale Souveränität". ("bfh", "BFH (Bern)", "playwright", { "url": "https://www.bfh.ch/en/about-bfh/careers/jobs/", "wait_for": ".teas-prospective-job-teaser", "card": ".teas-prospective-job-teaser", "title_sel": "h2.teaser-title", "location_sel": "li.teaser-text-list-item:nth-of-type(2)", "link_sel": "a", "default_location": "Switzerland", "_score_floor": 2, "_title_exclude": NOISE_TITLE_EXCLUDE, }), ] # Companies that resist scraping — clickable checklist in the report. # (BFH re-automated 2026-07-14 on www.bfh.ch. Dialectic dropped — crypto covered elsewhere.) MANUAL_CHECK = [ # Oracle (Tier C, requested 2026-06-06). Oracle Recruiting Cloud (ORC) resists clean # scraping: careers.oracle.com renders job tiles that are NOT anchors, so the playwright # selector pattern fails. The ORC REST endpoint works and returns data, but reliable CH # filtering needs the Switzerland geography node id, which the facet-expand call 400s on. # Wireable later as a `json` adapter once that geographyId is resolved. For now, manual: # https://eeho.fa.us2.oraclecloud.com/hcmRestApi/resources/latest/recruitingCEJobRequisitions # ?onlyData=true&expand=requisitionList.workLocation # &finder=findReqs;siteNumber=CX_45001,limit=200,sortBy=POSTING_DATES_DESC # (then client-filter requisitionList[].PrimaryLocation for Switzerland/Zürich) ("Oracle", "ORC SPA resists scraping; REST endpoint known but needs CH geographyId (see code comment). Check Switzerland tech roles manually.", "https://careers.oracle.com/en/sites/jobsearch/jobs?location=Switzerland"), ] @lru_cache(maxsize=1) def _ssl_context(): """Prefer certifi's CA bundle over the OS trust store: on Windows, urllib's default context pulls trust anchors from the system cert store, which has been observed to reject valid Workday-hosted certs as expired (stale cached root) even though the live chain verifies fine elsewhere. Falls back to the default context if certifi isn't installed (pip install -r requirements.txt covers it).""" try: import certifi return ssl.create_default_context(cafile=certifi.where()) except ImportError: return ssl.create_default_context() def http_get_json(url, headers=None, data=None, method="GET"): headers = headers or {} headers.setdefault("User-Agent", USER_AGENT) headers.setdefault("Accept", "application/json") if data is not None and isinstance(data, dict): data = json.dumps(data).encode("utf-8") headers.setdefault("Content-Type", "application/json") req = urllib.request.Request(url, data=data, headers=headers, method=method) with urllib.request.urlopen(req, timeout=30, context=_ssl_context()) as resp: return json.loads(resp.read().decode("utf-8")) def fetch_workday(args): host, site, tenant = args["host"], args["site"], args["tenant"] search_text = args.get("search_text", "") url = f"https://{host}/wday/cxs/{tenant}/{site}/jobs" jobs, offset = [], 0 while True: data = http_get_json(url, method="POST", data={ "appliedFacets": {}, "limit": 20, "offset": offset, "searchText": search_text, }) postings = data.get("jobPostings", []) for p in postings: ext = p.get("externalPath", "") jid = (p.get("bulletFields") or [ext])[0] if p.get("bulletFields") else ext jobs.append({ "id": jid, "title": p.get("title", ""), "location": p.get("locationsText", "") + " " + ext, "url": f"https://{host}{ext}", "posted": p.get("postedOn", ""), "description": "", }) total = data.get("total", 0) offset += len(postings) if not postings or offset >= total: break return jobs def fetch_ashby(args): slug = args["slug"] url = f"https://api.ashbyhq.com/posting-api/job-board/{slug}?includeCompensation=true" data = http_get_json(url) jobs = [] for j in data.get("jobs", []): secs = j.get("secondaryLocations", []) or [] sec_names = [s.get("location", "") if isinstance(s, dict) else str(s) for s in secs] loc_blob = " | ".join([j.get("location", "") or ""] + sec_names) jobs.append({ "id": j.get("id"), "title": j.get("title", ""), "location": loc_blob, "url": j.get("jobUrl"), "posted": j.get("publishedAt", ""), "description": (j.get("descriptionPlain") or "")[:2500], "department": j.get("department", ""), }) return jobs def fetch_greenhouse(args): # Title-filtered boards score title_only — skip heavy JD HTML (content=true) for those. board = args["board"] want_content = not args.get("_title_filter") url = f"https://boards-api.greenhouse.io/v1/boards/{board}/jobs" if want_content: url += "?content=true" data = http_get_json(url) jobs = [] for j in data.get("jobs", []): loc = (j.get("location") or {}).get("name", "") offices = j.get("offices") or [] office_names = " | ".join(o.get("name", "") for o in offices if isinstance(o, dict)) loc_blob = " ".join(x for x in [loc, office_names] if x) desc = "" if want_content: desc = j.get("content", "") or "" desc = re.sub(r"<[^>]+>", " ", desc) desc = re.sub(r"\s+", " ", desc).strip() jobs.append({ "id": str(j.get("id")), "title": j.get("title", ""), "location": loc_blob, "url": j.get("absolute_url"), "posted": j.get("updated_at", ""), "description": desc[:2500], }) return jobs def fetch_pcsx(args): """Eightfold PCSX search API. Microsoft uses apply.careers.microsoft.com. The same endpoint pattern is used by other PCS-hosted boards.""" domain = args["domain"] location = args.get("location", "") base = "https://apply.careers.microsoft.com/api/pcsx/search" jobs, start = [], 0 while True: url = f"{base}?domain={domain}&query=&location={urllib.parse.quote(location)}&start={start}&num=50" data = http_get_json(url, headers={"Referer": f"https://apply.careers.microsoft.com/careers?location={urllib.parse.quote(location)}"}) payload = data.get("data") or {} positions = payload.get("positions", []) or [] for p in positions: locs = p.get("locations") or [] jobs.append({ "id": str(p.get("id")), "title": p.get("name", ""), "location": " | ".join(locs), "url": f"https://jobs.careers.microsoft.com/global/en/job/{p.get('displayJobId') or p.get('id')}", "posted": p.get("postedTs", ""), "description": (p.get("description") or "")[:2000], }) # The API serves ~10 rows per call regardless of `num`, so a short page is NOT # end-of-results — page until a call comes back empty or `count` is reached. # (The old `len(positions) < 50` break capped Microsoft at its first 10 CH roles # and hid the entire Zürich Principal-FDE cluster sitting on page 2.) if not positions: break start += len(positions) total = payload.get("count") if isinstance(total, int) and start >= total: break if start >= 500: break return jobs def fetch_smartrecruiters(args): """SmartRecruiters public postings API. Used by many EU energy/commodity firms.""" company = args["company"] base = f"https://api.smartrecruiters.com/v1/companies/{company}/postings" jobs, offset = [], 0 while True: data = http_get_json(f"{base}?limit=100&offset={offset}") content = data.get("content", []) or [] for p in content: loc = p.get("location") or {} parts = [loc.get("fullLocation") or loc.get("city") or ""] if loc.get("remote"): parts.append("Remote") if loc.get("hybrid"): parts.append("Hybrid") loc_str = " ".join(x for x in parts if x) dept = (p.get("department") or {}).get("label", "") if isinstance(p.get("department"), dict) else "" func = (p.get("function") or {}).get("label", "") if isinstance(p.get("function"), dict) else "" jobs.append({ "id": str(p.get("id")), "title": p.get("name", ""), "location": loc_str, "url": f"https://jobs.smartrecruiters.com/{company}/{p.get('id')}", "posted": p.get("releasedDate", ""), "description": " ".join(filter(None, [dept, func])), }) total = data.get("totalFound", 0) offset += len(content) if not content or offset >= total or offset >= 300: break return jobs def fetch_rss(args): """Generic RSS/RDF feed parser. BIS publishes vacancies as RSS 1.0 (RDF), whose <item> elements live in the http://purl.org/rss/1.0/ namespace. Falls back to plain RSS 2.0 <item> elements. Location isn't in the feed, so default_location is required.""" import xml.etree.ElementTree as ET req = urllib.request.Request(args["url"], headers={"User-Agent": USER_AGENT}) with urllib.request.urlopen(req, timeout=30, context=_ssl_context()) as resp: root = ET.fromstring(resp.read()) ns = {"rss1": "http://purl.org/rss/1.0/", "dc": "http://purl.org/dc/elements/1.1/"} items = root.findall(".//rss1:item", ns) or root.findall(".//item") jobs = [] for it in items: def field(tag, namespaced=True): el = it.find(f"rss1:{tag}", ns) if namespaced else it.find(tag) if el is None and namespaced: el = it.find(tag) return (el.text or "").strip() if el is not None and el.text else "" link = field("link") jobs.append({ "id": link or field("title"), "title": field("title"), "location": args.get("default_location", ""), "url": link, "posted": (it.findtext("dc:date", default="", namespaces=ns) or field("date")), "description": re.sub(r"<[^>]+>", " ", field("description"))[:1500], }) return jobs # schema.org JobPosting addresses carry ISO alpha-2 country codes; the location policy # keyword lists are full names. Only the countries the policies actually test for. _ISO2_COUNTRY = { "CH": "Switzerland", "NO": "Norway", "DK": "Denmark", "DE": "Germany", "AT": "Austria", "FR": "France", "IT": "Italy", "ES": "Spain", "PL": "Poland", "NL": "Netherlands", "BE": "Belgium", "SE": "Sweden", "FI": "Finland", "GB": "United Kingdom", "UK": "United Kingdom", "IE": "Ireland", "US": "United States", "PT": "Portugal", "CZ": "Czechia", "RO": "Romania", "RS": "Serbia", "TR": "Turkey", } def fetch_teamtailor(args): """Teamtailor public JSON Feed (`https://<slug>.teamtailor.com/jobs.json`). Used by Telenor (`slug`) and Axpo (`base_url`, a Teamtailor board on a custom domain — careers.axpo.com — which no <slug>.teamtailor.com host resolves to). The public /jobs HTML page paginates and under-reports; the feed is authoritative. The feed carries no location field, so default_location is required (Teamtailor boards are per-country). Descriptions are frequently Norwegian — see the _score_floor note on the company. The feed caps at 100 items and links the next page via `next_url`; follow it, or large boards silently truncate (Axpo is ~400 roles, i.e. 4 pages). TRAP: an unclaimed slug returns Teamtailor's *demo* board (HTTP 200, plausible JSON) instead of 404. Verified 2026-07-29: `ksat` and `akerbp` both served the seed set below though neither company uses Teamtailor. Filtered here so phantom roles never reach a report — if a real board is ever dropped by this, widen the fingerprint.""" DEMO_TITLES = {"sales development manager", "team lead - csm", "social media manager", "customer success manager", "key account manager", "ux designer"} if args.get("base_url"): url = args["base_url"].rstrip("/") + "/jobs.json" else: url = f"https://{args['slug']}.teamtailor.com/jobs.json" jobs = [] for _ in range(args.get("max_pages", 10)): req = urllib.request.Request( url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"}) with urllib.request.urlopen(req, timeout=30, context=_ssl_context()) as resp: data = json.loads(resp.read().decode("utf-8", "replace")) items = data.get("items", []) or [] for it in items: # Prefer the schema.org JobPosting payload's real address over default_location. # Axpo is a pan-European trader (Madrid, Milan, Warsaw, Germany, France as well as # Baden/Zurich), so defaulting every role to the board's home country would mark # ~380 non-Swiss roles CH-eligible — the same class of bug as Apple's forced # "Switzerland" default. Telenor's feed has no _jobposting and falls back cleanly. location = args.get("default_location", "") places = ((it.get("_jobposting") or {}).get("jobLocation") or []) if isinstance(places, dict): places = [places] parts = [] for place in places: addr = (place or {}).get("address") or {} # schema.org gives ISO alpha-2 ("CH", "NO"); the location policy matches on # full country names, so expand or the role reads as location-unknown. country = _ISO2_COUNTRY.get((addr.get("addressCountry") or "").upper(), addr.get("addressCountry")) bit = ", ".join(x for x in (addr.get("addressLocality"), country) if x) if bit and bit not in parts: parts.append(bit) if parts: location = " | ".join(parts) jobs.append({ "id": it.get("id") or it.get("url", ""), "title": it.get("title", ""), "location": location, "url": it.get("url", ""), "posted": it.get("date_published", ""), "description": re.sub(r"<[^>]+>", " ", it.get("content_html", ""))[:2500], }) nxt = data.get("next_url") if not items or not nxt or nxt == url: break url = nxt titles = {j["title"].strip().lower() for j in jobs} if jobs and len(titles & DEMO_TITLES) >= 4: raise RuntimeError( f"teamtailor slug '{args['slug']}' served the demo board, not a real one " f"({len(jobs)} placeholder ads) — the company likely uses a different ATS") return jobs def fetch_getro(args): """Getro network job-board search API (POST JSON). Powers VC portfolio talent networks. NO company in COMPANIES currently uses this adapter — the Coinbase Ventures network was dropped 2026-08-18 (see the note there). Kept because it generalises to any Getro collection id. Note such boards list PORTFOLIO companies' roles, never the VC's own openings, so an org filter is usually needed. Server-side filters: searchable_locations and job_functions. Org name is folded into the title since this is a multi-company board.""" collection = args["collection"] url = f"https://api.getro.com/api/v2/collections/{collection}/search/jobs" filters = {} if args.get("locations"): filters["searchable_locations"] = args["locations"] if args.get("job_functions"): filters["job_functions"] = args["job_functions"] jobs, page = [], 0 excluded_orgs = {name.casefold() for name in args.get("_exclude_orgs", [])} while True: data = http_get_json(url, method="POST", data={ "hitsPerPage": 100, "page": page, "query": "", "filters": filters, }) res = data.get("results", {}) or {} batch = res.get("jobs", []) or [] for j in batch: org = (j.get("organization") or {}).get("name", "") if org.casefold() in excluded_orgs: continue locs = j.get("searchable_locations") or j.get("locations") or [] loc_str = " | ".join(locs) if isinstance(locs, list) else str(locs) ts = j.get("created_at") posted = "" if isinstance(ts, (int, float)): posted = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d") title = j.get("title", "") jobs.append({ "id": str(j.get("id")), "title": f"{title} @ {org}" if org else title, "location": loc_str, "url": j.get("url", ""), "posted": posted, "description": " ".join(filter(None, [org] + (j.get("skills") or []))), }) total = res.get("count", 0) page += 1 if not batch or len(jobs) >= total or page >= 10: break return jobs def fetch_amazon(args): """amazon.jobs public search JSON. Covers Amazon + AWS, which share one board. `normalized_country_code[]` is the filter that actually works; the plain `country[]` and `loc_query` params are ignored and silently return the global (US-heavy) set — verified 2026-08-18, where `country[]=CHE` returned 6,837 mostly-Seattle hits against 32 for the normalized form.""" base = "https://www.amazon.jobs/en/search.json" headers = {"User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")} jobs, offset, page_size = [], 0, 100 while True: qs = urllib.parse.urlencode({ "radius": "100mi", "offset": offset, "result_limit": page_size, "sort": "recent", "base_query": args.get("base_query", ""), }) countries = "".join(f"&normalized_country_code[]={urllib.parse.quote(c)}" for c in args.get("countries", ["CHE"])) data = http_get_json(f"{base}?{qs}{countries}", headers=headers) batch = data.get("jobs", []) or [] for j in batch: jobs.append({ "id": str(j.get("id_icims") or j.get("id") or ""), "title": j.get("title", ""), "location": j.get("normalized_location") or j.get("location") or "", "url": "https://www.amazon.jobs" + (j.get("job_path") or ""), "posted": j.get("posted_date", ""), "description": (j.get("description_short") or j.get("description") or "")[:2000], }) total = data.get("hits", 0) offset += page_size if not batch or offset >= total or offset >= args.get("max_results", 500): break return jobs def fetch_phenom(args): """Phenom People careers search (POST /widgets with ddoKey=refineSearch). Used by Roche. The public search-results page is a JS shell whose location filter silently fails: a scrape of `?locationsearch=Switzerland` returns *recommendation-widget* cards (Shanghai, Kyiv, Bogota) while the page state reports `no-results`. Verified 2026-08-18 — this is why Roche contributed 0 Swiss roles for months while looking healthy. The underlying refineSearch endpoint filters correctly (134 CH roles), so query it directly. `country_facet` is the value for the `country` facet (e.g. "Switzerland"). `ref_num` and `page_id` are tenant constants visible in any /widgets POST from the careers site.""" url = args["url"] ref_num, page_id = args["ref_num"], args.get("page_id", "page11-ds") lang = args.get("lang", "en_global") page_size = args.get("page_size", 100) headers = { "Origin": "{0.scheme}://{0.netloc}".format(urllib.parse.urlsplit(url)), "Referer": url, # Phenom 403s the default library UA. "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"), } jobs, offset = [], 0 while True: data = http_get_json(url, method="POST", headers=headers, data={ "lang": lang, "deviceType": "desktop", "country": "global", "pageName": "search-results", "ddoKey": "refineSearch", "sortBy": "", "subsearch": "", "from": offset, "jobs": True, "counts": True, "all_fields": ["country", "state", "city", "category", "type"], "size": page_size, "clearAll": False, "jdsource": "facets", "isSliderEnable": False, "pageId": page_id, "siteType": "external", "keywords": "", "global": True, "locationData": {}, "selected_fields": {"country": [args["country_facet"]]}, }) res = data.get("refineSearch", {}) or {} batch = (res.get("data", {}) or {}).get("jobs", []) or [] for j in batch: jobs.append({ "id": str(j.get("jobId") or j.get("jobSeqNo") or ""), "title": j.get("title", ""), "location": j.get("location") or j.get("cityStateCountry") or "", "url": j.get("applyUrl", ""), "posted": (j.get("postedDate") or "")[:10], "description": (j.get("descriptionTeaser") or "")[:2000], }) total = res.get("totalHits", 0) offset += page_size if not batch or offset >= total or offset >= args.get("max_results", 600): break return jobs def fetch_onlyfy(args): """onlyfy.jobs board (XING E-Recruiting / ex-Prinzip), used by Bitcoin Suisse. The candidate/job/ajax_list endpoint returns an HTML fragment listing every posting; each card carries a <a href="/job/ID">title</a> and a location cell flagged by an icon-map-marker. Titles and locations appear in document order, one of each per card, so we extract both lists and zip them. No JSON API and no headless browser needed.""" import html as _html slug = args["slug"] base = f"https://{slug}.onlyfy.jobs" url = (f"{base}/candidate/job/ajax_list" f"?display_length=100&page=1&sort=date&sort_dir=DESC&search=") req = urllib.request.Request(url, headers={ "User-Agent": USER_AGENT, "X-Requested-With": "XMLHttpRequest", }) with urllib.request.urlopen(req, timeout=30, context=_ssl_context()) as resp: page = resp.read().decode("utf-8", "replace") titles = re.findall(r'<a href="(/job/[a-z0-9]+)">(.*?)</a>', page, re.S) locs = re.findall(r'icon-map-marker[^>]*></i>\s*([^<]+)', page) if titles and locs and len(titles) != len(locs): # Misaligned regex pairs would invent wrong locations; keep titles, drop bad locs. print(f" onlyfy: title/loc count mismatch ({len(titles)} vs {len(locs)}); " f"using titles only", file=sys.stderr) locs = [""] * len(titles) jobs = [] for (href, raw_title), raw_loc in zip(titles, locs if locs else [""] * len(titles)): title = _html.unescape(re.sub(r"<[^>]+>", "", raw_title)).strip() loc = _html.unescape(raw_loc).strip() if raw_loc else "" jobs.append({ "id": href.rsplit("/", 1)[-1], "title": title, "location": loc, "url": base + href, "posted": "", "description": loc, }) return jobs def fetch_lever(args): """Lever public postings API. Palantir uses this. The board is US/London-heavy; Swiss/Zurich (Schwyz hub) roles are rare on it but will surface here when posted — location filtering downstream drops the US/London bulk. categories.allLocations captures multi-location postings; createdAt is epoch-ms.""" slug = args["slug"] data = http_get_json(f"https://api.lever.co/v0/postings/{slug}?mode=json") jobs = [] for j in data: cats = j.get("categories") or {} all_locs = cats.get("allLocations") or [] loc_blob = " | ".join(x for x in ([cats.get("location") or ""] + [str(a) for a in all_locs]) if x) ts = j.get("createdAt") posted = "" if isinstance(ts, (int, float)): posted = datetime.fromtimestamp(ts / 1000, tz=timezone.utc).strftime("%Y-%m-%d") jobs.append({ "id": j.get("id"), "title": j.get("text", ""), "location": loc_blob, "url": j.get("hostedUrl"), "posted": posted, "description": (j.get("descriptionPlain") or "")[:2500], }) return jobs def _parse_taleo_jobsearch_html(html, host, career_section): """Extract jobs from a Taleo jobsearch.ftl HTML response. Taleo does not expose a reliable public JSON search API for this board (REST returns careerSectionUnAvailable). Jobs are embedded in the page as api.fillInterface arrays and as a pipe-delimited initialHistory field. """ import html as _html jobs, seen = [], set() detail_base = f"https://{host}/careersection/{career_section}/jobdetail.ftl" # Quoted fillInterface form (already-unescaped dates in modern Taleo builds). # Observed record shape (NATO 2026-07): # '<id>','<title>','<id>','<title>', five more '<id>', # '<jobNo>','<Country-City>','false','','','','', # '<deadline>','<NATO body>','<grade>' pat = re.compile( r"'(\d+)','((?:[^'\\]|\\.)+)'," r"'\1','\2'," r"(?:'\1',){5}" r"'(\d+)','((?:[^'\\]|\\.)+)'," r"'false','','','',''," r"'((?:[^'\\]|\\.)*)','((?:[^'\\]|\\.)*)','((?:[^'\\]|\\.)*)'" ) for m in pat.finditer(html): job_no = m.group(3) if job_no in seen: continue seen.add(job_no) def _unq(s): return _html.unescape(s.replace("\\'", "'").replace("\\:", ":")) title = _unq(m.group(2)) loc = _unq(m.group(4)).replace("-", ", ") deadline = _unq(m.group(5)) body = _unq(m.group(6)) grade = _unq(m.group(7)) # List view exposes application deadline only (not a posted-on date). We store # the date part in `posted` so the report surfaces the closing date. deadline_short = deadline.split(",")[0].strip() if deadline else "" jobs.append({ "id": job_no, "title": title, "location": loc, "url": f"{detail_base}?job={job_no}&lang=en", "posted": deadline_short, "description": f"{body} {grade}".strip()[:500], }) if jobs: return jobs # Fallback: pipe-delimited initialHistory hidden field m = re.search( r'(?:name|id)="initialHistory"[^>]*value="([^"]*)"', html, re.I ) or re.search( r'value="([^"]*)"[^>]*(?:name|id)="initialHistory"', html, re.I ) if not m: return [] hist = urllib.parse.unquote(m.group(1).replace("%5C:", ":").replace("\\:", ":")) parts = hist.split("!|!") for i, p in enumerate(parts): if not re.fullmatch(r"\d{5,6}", p) or i + 1 >= len(parts): continue loc_raw = parts[i + 1] if not re.search( r"Norway|Denmark|Belgium|Germany|Italy|France|United|Luxembourg|" r"Netherlands|Poland|Spain|Portugal|Canada|Turkey|Türkiye", loc_raw, re.I, ): continue job_no = p if job_no in seen: continue title = "" for j in range(i - 1, max(-1, i - 12), -1): cand = parts[j] if (cand and not re.fullmatch(r"\d+", cand) and cand not in ("false", "true", "Apply") and len(cand) > 8): title = cand break if not title: continue seen.add(job_no) deadline = parts[i + 6] if i + 6 < len(parts) else "" body = parts[i + 7] if i + 7 < len(parts) else "" grade = parts[i + 8] if i + 8 < len(parts) else "" deadline_short = deadline.split(",")[0].strip() if deadline else "" jobs.append({ "id": job_no, "title": _html.unescape(title), "location": loc_raw.replace("-", ", "), "url": f"{detail_base}?job={job_no}&lang=en", "posted": deadline_short, "description": f"{body} {grade}".strip()[:500], }) return jobs def fetch_taleo(args): """Oracle Taleo careersection board (used by NATO civilian vacancies). The marketing page https://www.nato.int/en/work-with-us/careers/vacancies is an AEM shell; the live board is Taleo. Public REST search is unavailable, so we GET jobsearch.ftl (optionally filtered by Taleo location taxonomy IDs from the board's location dropdown) and parse embedded job records. Args: host (default nato.taleo.net), career_section (default "2"), location_ids (optional list of Taleo option values — e.g. Norway/Denmark). """ host = args.get("host", "nato.taleo.net") section = str(args.get("career_section", "2")) location_ids = args.get("location_ids") # One request per location id (Taleo query param takes a single id); merge by job no. # If no location_ids, one unfiltered first-page fetch. fetches = location_ids if location_ids else [None] by_id = {} for loc_id in fetches: url = f"https://{host}/careersection/{section}/jobsearch.ftl?lang=en" if loc_id: url += f"&location={urllib.parse.quote(str(loc_id))}" req = urllib.request.Request(url, headers={ "User-Agent": USER_AGENT, "Accept": "text/html,application/xhtml+xml", }) with urllib.request.urlopen(req, timeout=60, context=_ssl_context()) as resp: html = resp.read().decode("utf-8", "replace") for j in _parse_taleo_jobsearch_html(html, host, section): by_id[j["id"]] = j return list(by_id.values()) def fetch_json(args): """Generic JSON jobs API with configurable field names, for employer sites that expose a clean public endpoint. Verified use: Swissgrid (Magnolia CMS /.rest/cloud/component-data — {config, jobs:[...], filters}). Field names vary by site, so they're configurable: field_title/field_location/field_url/field_date. loc_suffix appends e.g. ' Switzerland' so the CH location filter matches city-only values such as "Aarau"/"Prilly" (not every Swiss town is in CH_LOCATION_KEYWORDS). desc_keys fold extra fields (department, employment type) into the description for keyword scoring. Args: url, jobs_key (default "jobs"), field_* (defaults title/location/url/date), url_prefix, loc_suffix, desc_keys.""" data = http_get_json(args["url"]) arr = data.get(args.get("jobs_key", "jobs"), []) if isinstance(data, dict) else (data or []) ft, fl = args.get("field_title", "title"), args.get("field_location", "location") fu, fd = args.get("field_url", "url"), args.get("field_date", "date") prefix, suffix = args.get("url_prefix", ""), args.get("loc_suffix", "") desc_keys = args.get("desc_keys", []) jobs = [] for j in arr: url = j.get(fu, "") or "" if url and not url.startswith("http") and prefix: url = prefix.rstrip("/") + "/" + url.lstrip("/") loc = (j.get(fl, "") or "").strip() + suffix desc = " ".join(str(j.get(k)) for k in desc_keys if j.get(k)) jobs.append({ "id": str(j.get("id") or url), "title": j.get(ft, ""), "location": loc, "url": url, "posted": j.get(fd, "") or "", "description": desc[:500], }) return jobs def fetch_sbb(args): """SBB (company.sbb.ch) AEM job filter. The whole board is served as a flat JSON list at .../jobfilter.results.json (~145 roles); the website filters client-side via each job's numbered `attributes`: '20'=Berufsfeld/topic, '110'=region, '100'=city, 'links.directlink'=the jobs.sbb.ch URL. We replicate the user's IT + Bern-region filter so only commutable IT roles surface. Titles are German/generic (Application Engineer, Network Security Engineer, OT Architekt) and won't match ENG_TITLE_FILTER or the keyword scorer, so this company is given a _score_floor in COMPANIES to keep its pre-filtered results visible. topic/region are configurable substrings.""" url = args.get("url", ("https://company.sbb.ch/content/internet/corporate/de/" "jobs-karriere/jobs/job-suche/jcr:content/parmain/" "jobfilter.results.json")) topic = args.get("topic", "IT / Telekommunikation") region = args.get("region", "Bern Mittelland") data = http_get_json(url) arr = data if isinstance(data, list) else (data.get("results") or data.get("jobs") or []) jobs = [] for j in arr: a = j.get("attributes", {}) or {} blob = " ".join(str(x) for v in a.values() for x in (v if isinstance(v, list) else [v])) if topic and topic not in blob: continue if region and region not in blob: continue region_v = " ".join(a.get("110", []) or []) city_v = " ".join(a.get("100", []) or []) field_v = " ".join(a.get("20", []) or []) jobs.append({ "id": str(j.get("id") or j.get("viewkey") or ""), "title": j.get("title", ""), "location": f"{city_v} {region_v} Schweiz".strip(), "url": (j.get("links") or {}).get("directlink", ""), "posted": j.get("start_date", "") or "", "description": (field_v + " " + (j.get("text", "") or ""))[:400], }) return jobs def fetch_bkw(args): """BKW Group (jobs.bkw.com) PMS structured-data API. The whole-group board is ~600 roles dominated by building-tech / electrical / civil-engineering trades; we keep only the Berufsfeld categories relevant to the user (Informatik / Trading / Finanzen), which surfaces IT/data plus the energy-trading roles (Quant Risk Modeller, Solution Architect Energiehandel, Energy Derivatives/Market-Risk analysts). locations[].address gives city/country. Pre-filtered + German/generic titles, so paired with a _score_floor in COMPANIES. The category allowlist is configurable.""" url = args.get("url", ("https://jobs.bkw.com/_api/v1/structureddata?" "configFromContentElement=82381&language=de-ch")) allow = [c.lower() for c in args.get("categories", ["Informatik", "Trading", "Finanzen"])] data = http_get_json(url) arr = data if isinstance(data, list) else [] if not arr and isinstance(data, dict): for v in data.values(): if isinstance(v, list) and v and isinstance(v[0], dict) and "title" in v[0]: arr = v break jobs = [] for j in arr: if j.get("type") and j.get("type") != "jobs": continue cats = [c.get("title", "") for c in (j.get("relations", {}) or {}).get("Berufsfeld", []) or []] if allow and not any(any(a in c.lower() for a in allow) for c in cats): continue locs = j.get("locations") or [] addr = (locs[0].get("address") if locs and isinstance(locs[0], dict) else {}) or {} loc = " ".join(x for x in [addr.get("city", ""), addr.get("country", "")] if x) or "Schweiz" jobs.append({ "id": str(j.get("id") or j.get("url") or ""), "title": j.get("title", ""), "location": loc, "url": j.get("url", ""), "posted": "", "description": " ".join(cats + [j.get("subtitle", "") or ""])[:300], }) return jobs # Injected before page scripts run, to mask the most common headless-detection signals. # Required for Google; harmless for the other sites. STEALTH_JS = """ Object.defineProperty(navigator, 'webdriver', {get: () => undefined}); window.chrome = {runtime: {}, loadTimes: () => {}, csi: () => {}, app: {}}; Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]}); Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en', 'de']}); const _q = navigator.permissions && navigator.permissions.query; if (_q) { navigator.permissions.query = (p) => p && p.name === 'notifications' ? Promise.resolve({state: Notification.permission}) : _q(p); } """ _playwright_singleton = {"pw": None, "browser": None} def _get_browser(): """Lazy-init a single shared headless browser. Saves ~3s per company.""" if _playwright_singleton["browser"] is not None: return _playwright_singleton["browser"] try: from playwright.sync_api import sync_playwright except ImportError as e: raise RuntimeError("playwright not installed - run: pip install -r requirements.txt") from e pw = sync_playwright().start() browser = pw.chromium.launch( headless=True, args=["--disable-blink-features=AutomationControlled"], ) _playwright_singleton["pw"] = pw _playwright_singleton["browser"] = browser return browser def _absolutize(href, prefix): """Join a possibly-relative href with the configured prefix.""" if not href or href.startswith("http"): return href cleaned = href.lstrip("./").lstrip("/") if not prefix: return href return prefix.rstrip("/") + "/" + cleaned def _strip_query_param(url, param): """Drop a single query parameter from a URL, preserving the rest. Job-card hrefs inherit the listing page's query string, so with query-param pagination the same posting would get a different URL (and therefore a different id) depending on which page it happened to land on — making every reshuffle look like a brand-new job and orphaning its decision-log entry. """ if not url or f"{param}=" not in url: return url head, _, query = url.partition("?") if not query: return url kept = [kv for kv in query.split("&") if kv and not kv.startswith(f"{param}=")] return f"{head}?{'&'.join(kept)}" if kept else head def _extract_location_from_blob(full, default=""): """Pull a short location line out of a job-card text blob. Google/Meta/Roche-style cards often dump Material-icon chrome into inner_text (e.g. 'corporate_fare' / 'place' / 'Zürich, Switzerland' / 'bar_chart'). Prefer the line after a 'place'/'location' label, else the first short line that looks like a place, else the configured default — never the first 300 chars of the whole card. """ if not full: return default lines = [ln.strip() for ln in full.splitlines() if ln.strip()] skip = { "place", "location", "corporate_fare", "bar_chart", "work", "schedule", "google", "meta", "apple", "roche", "cisco", } for i, ln in enumerate(lines): if ln.lower() in ("place", "location") and i + 1 < len(lines): cand = lines[i + 1].strip() if cand and cand.lower() not in skip and len(cand) < 120: return cand[:200] place_hints = ( CH_LOCATION_KEYWORDS + DK_LOCATION_KEYWORDS + NO_LOCATION_KEYWORDS + list(EUROPE_SCOPE_KEYWORDS) + list(GLOBAL_SCOPE_KEYWORDS) + ["remote", "hybrid", "onsite", "on-site"] ) for ln in lines: low = ln.lower() if len(ln) > 120 or low in skip: continue if any(k in low for k in place_hints): return ln[:200] return default def _close_browser(): if _playwright_singleton["browser"]: try: _playwright_singleton["browser"].close() except Exception: pass if _playwright_singleton["pw"]: try: _playwright_singleton["pw"].stop() except Exception: pass def fetch_playwright(args): """Generic headless-browser scraper. See COMPANIES entries for selector args.""" browser = _get_browser() ctx = browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", locale="en-US", viewport={"width": 1366, "height": 768}, ) ctx.add_init_script(STEALTH_JS) page = ctx.new_page() jobs = [] seen_ids = set() def scrape_current(): """Extract cards from the currently-loaded page; append new ones to `jobs`. Returns the count of newly-added (not-yet-seen) cards so a pagination loop can stop once a page contributes nothing new.""" wait_for = args.get("wait_for") if wait_for: try: page.wait_for_selector(wait_for, timeout=15000) except Exception: page.wait_for_timeout(4000) # Scroll a few times to trigger any lazy-loaded results for _ in range(args.get("scroll_count", 3)): try: page.mouse.wheel(0, 4000) page.wait_for_timeout(700) except Exception: break cards = page.locator(args["card"]) n = min(cards.count(), args.get("max_cards", 150)) added = 0 for i in range(n): card = cards.nth(i) try: title = "" if args.get("title_attr") == "text": title = (card.inner_text() or "").strip().split("\n", 1)[0][:200] elif args.get("title_attr"): title = (card.get_attribute(args["title_attr"]) or "").strip() elif args.get("title_sel"): t = card.locator(args["title_sel"]).first if t.count(): # Read either an attribute (e.g. aria-label) or the inner text if args.get("title_sel_attr"): title = (t.get_attribute(args["title_sel_attr"]) or "").strip() else: title = (t.inner_text() or "").strip() if args.get("title_strip_prefix") and title.startswith(args["title_strip_prefix"]): title = title[len(args["title_strip_prefix"]):].strip() if not title: title = (card.inner_text() or "").strip().split("\n", 1)[0][:200] location = args.get("default_location", "") if args.get("location_sel"): lsel = card.locator(args["location_sel"]).first if lsel.count(): location = (lsel.inner_text() or location).strip() link_el = card if not args.get("link_sel") else card.locator(args["link_sel"]).first href = (link_el.get_attribute(args.get("link_attr", "href")) or "") if link_el.count() else "" href = _absolutize(href, args.get("url_prefix", "")) if args.get("page_param"): href = _strip_query_param(href, args["page_param"]) if not title: continue jid = href or f"{page.url}#{i}" if jid in seen_ids: continue seen_ids.add(jid) added += 1 description = "" if args.get("use_inner_text_as_blob"): # Full card text for keyword scoring; location via structured extract. full = (card.inner_text() or "") description = full[:2000] extracted = _extract_location_from_blob( full, default=args.get("default_location", "") ) # Prefer a real location line over a default that is empty or a # polluted leftover; keep default_location when extract finds nothing. if extracted: location = extracted elif not location: location = args.get("default_location", "") jobs.append({ "id": jid, "title": title, "location": location, "url": href or page.url, "posted": "", "description": description, }) except Exception: continue return added try: page.goto(args["url"], timeout=45000, wait_until="domcontentloaded") # Optional cookie banner acceptance (once, on the first page) for sel in args.get("cookie_accept", []) or []: try: btn = page.locator(sel).first if btn.is_visible(timeout=2000): btn.click() page.wait_for_timeout(500) except Exception: pass # Optional query-param pagination (e.g. Drupal "?page=N"). The base URL is the first # page (already loaded); fetch successive pages until one adds no new cards. # `page_param_start` is the value the SECOND page takes: 1 for 0-indexed boards # (Drupal), 2 for 1-indexed ones (Google, where "?page=1" is just page one again). page_param = args.get("page_param") if page_param: base = args["url"] joiner = "&" if "?" in base else "?" second = args.get("page_param_start", 1) for p in range(args.get("max_pages", 8)): if p > 0: page.goto(f"{base}{joiner}{page_param}={second + p - 1}", timeout=45000, wait_until="domcontentloaded") added = scrape_current() if p > 0 and added == 0: break else: # Optional in-page pagination for client-side job boards. This is distinct from # query-param pagination above: the URL does not change when moving between pages. next_button = args.get("next_button") if next_button: for _ in range(args.get("max_pages", 8)): added = scrape_current() button = page.locator(next_button) if button.count() != 1: break try: if not button.is_visible() or button.is_disabled(): break button.click() page.wait_for_timeout(args.get("next_wait_ms", 800)) except Exception: break # A page with no unseen cards signals a loop or exhausted pagination. if added == 0: break else: scrape_current() finally: ctx.close() return jobs ADAPTERS = { "workday": fetch_workday, "ashby": fetch_ashby, "greenhouse": fetch_greenhouse, "pcsx": fetch_pcsx, "smartrecruiters": fetch_smartrecruiters, "rss": fetch_rss, "teamtailor": fetch_teamtailor, "getro": fetch_getro, "amazon": fetch_amazon, "phenom": fetch_phenom, "onlyfy": fetch_onlyfy, "lever": fetch_lever, "taleo": fetch_taleo, "json": fetch_json, "sbb": fetch_sbb, "bkw": fetch_bkw, "playwright": fetch_playwright, } def location_eligibility(loc_text, policy="default"): """Decide whether a posting's location is in-scope. Returns (eligible, in_ch, is_remote, is_relocation). Default policy (most companies): - CH onsite/hybrid - Denmark onsite/reloc - Europe/EMEA/global remote scope (work-from CH or DK is realistic) - CH-remote (Switzerland + remote wording) - Multi-country allowlists WITHOUT CH/DK/Europe/global scope are NOT eligible (e.g. Kraken "UK | Brazil | Ireland | ..." with no Switzerland/Europe) Overrides (args["_location_policy"]): - norway_only — Equinor: Norway only - dk_no — NATO-style: Denmark + Norway only (no CH) """ if not loc_text: return False, False, False, False low = loc_text.lower() in_ch = any(k in low for k in CH_LOCATION_KEYWORDS) in_dk = any(k in low for k in DK_LOCATION_KEYWORDS) in_no = any(k in low for k in NO_LOCATION_KEYWORDS) is_us_only = any(p in low for p in US_ONLY_PATTERNS) and not in_ch has_remote = bool(re.search(r"\bremote\b", low)) or "work from anywhere" in low europe_scope = any(k in low for k in EUROPE_SCOPE_KEYWORDS) global_scope = any(k in low for k in GLOBAL_SCOPE_KEYWORDS) and not is_us_only # Pan-regional scope counts even without the word "remote" (e.g. QuantCo "Europe"). is_europe_or_global = (europe_scope or global_scope) and not is_us_only is_ch_remote = in_ch and has_remote if policy == "norway_only": eligible = in_no and not is_us_only return eligible, False, False, eligible if policy == "dk_no": # NATO etc.: Denmark + Norway only — CH onsite is out (no NATO seat in CH). eligible = (in_dk or in_no) and not is_us_only return eligible, False, False, eligible # default is_relocation = in_dk is_remote = is_europe_or_global or is_ch_remote eligible = (in_ch or in_dk or is_europe_or_global) and not (is_us_only and not in_ch) return eligible, in_ch, is_remote, is_relocation @lru_cache(maxsize=512) def _kw_pattern(kw): """Word-boundary regex for a keyword. Plain substring matching produced false hits ('rag' inside 'sto[rag]e'/'tet[rag]on', 'intern' inside 'inte[rnal]'); we instead require the keyword not be flanked by alphanumerics. Keywords that begin/end on a non-word char (c#, .net, c++) skip that side's guard so they still match.""" esc = re.escape(kw.strip()) left = r"(?<![a-z0-9])" if kw.strip()[:1].isalnum() else "" right = r"(?![a-z0-9])" if kw.strip()[-1:].isalnum() else "" return re.compile(left + esc + right) def _kw_in(kw, text): return bool(_kw_pattern(kw).search(text)) def score_job(job, title_only=False): # Title carries the real signal; the JD body is full of company boilerplate (every # Kraken post mentions crypto/blockchain/trading, every cloud post mentions python). # So title matches score at full weight and body-only matches at half (min 1) — enough # to surface a role without letting boilerplate inflate it. Negatives count fully # wherever they appear (a disqualifier in the body still disqualifies). Title-filtered # boards pass title_only=True and skip body scoring entirely. title = (job.get("title") or "").lower() desc = "" if title_only else (job.get("description") or "").lower() score, pos, neg = 0, [], [] for kw, w in POSITIVE_KEYWORDS.items(): if _kw_in(kw, title): score += w pos.append(kw) elif desc and _kw_in(kw, desc): score += max(1, w // 2) pos.append(kw) for kw, w in NEGATIVE_KEYWORDS.items(): if _kw_in(kw, title) or (desc and _kw_in(kw, desc)): score += w neg.append(kw) return score, pos, neg def _atomic_write_json(path, data): """Write JSON via temp + os.replace so a crash mid-write cannot truncate state files.""" path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") os.replace(tmp, path) def load_seen(): if STATE_FILE.exists(): return json.loads(STATE_FILE.read_text(encoding="utf-8")) return {} def save_seen(seen): _atomic_write_json(STATE_FILE, seen) def load_decisions(): """Decision log keyed by job URL: {url: {company, title, decision, note, date}}. Decisions persist across runs so we don't re-evaluate roles we've already judged (shortlist / skip / applied / paused / rejected — free-text, not enforced).""" if DECISIONS_FILE.exists(): return json.loads(DECISIONS_FILE.read_text(encoding="utf-8")) return {} def save_decisions(decisions): _atomic_write_json(DECISIONS_FILE, decisions) def load_last_scrape(): """Per-company last successful scraped counts — used to flag sudden 0-job deaths.""" if LAST_SCRAPE_FILE.exists(): try: return json.loads(LAST_SCRAPE_FILE.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): return {} return {} def save_last_scrape(data): _atomic_write_json(LAST_SCRAPE_FILE, data) def _parse_posted(s): """Best-effort parse of an adapter's `posted` field into a date, across the mix of formats the boards use (ISO 8601 incl. trailing Z, YYYY-MM-DD, DD.MM.YYYY). Returns None for unparseable values (e.g. Workday's relative "Posted 5 Days Ago", or empty).""" if not s or not isinstance(s, str): return None s = s.strip() try: return datetime.fromisoformat(s.replace("Z", "+00:00")).date() except ValueError: pass for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%Y/%m/%d", "%d/%m/%Y", "%d-%b-%Y", "%d-%B-%Y"): try: # %b/%B need full token (e.g. 09-Aug-2026), not a fixed 10-char slice token = s.split(",")[0].strip() return datetime.strptime(token if "%b" in fmt or "%B" in fmt else s[:10], fmt).date() except ValueError: pass m = re.search(r"\d{4}-\d{2}-\d{2}", s) if m: try: return datetime.strptime(m.group(0), "%Y-%m-%d").date() except ValueError: pass return None def write_stats_table(stats, total_secs): """Render the per-company scan stats as a markdown table (+ a totals row).""" out = ["## Scan stats\n", "| Company | Scraped | CH/Remote | Match ≥2 | Newest posting | Time (s) |", "|---|--:|--:|--:|:--|--:|"] t_scraped = t_elig = t_match = 0 newest_all = None for s in stats: name = s["company"] + (" ⚠️" if s.get("error") else "") newest = s["newest"].isoformat() if s["newest"] else "—" out.append(f"| {name} | {s['scraped']:,} | {s['eligible']:,} | " f"{s['match']:,} | {newest} | {s['secs']:.1f} |") t_scraped += s["scraped"]; t_elig += s["eligible"]; t_match += s["match"] if s["newest"] and (newest_all is None or s["newest"] > newest_all): newest_all = s["newest"] out.append(f"| **Total ({len(stats)})** | **{t_scraped:,}** | **{t_elig:,}** | " f"**{t_match:,}** | **{newest_all.isoformat() if newest_all else '—'}** | " f"**{total_secs:.1f}** |") out.append("") return out def write_report(path, results, errors, new_only, include_weak, stats=None, total_secs=0.0, decisions=None, hide_decided=False): decisions = decisions or {} today = datetime.now().strftime("%Y-%m-%d") n_new = sum(1 for r in results if r["is_new"]) n_match = sum(1 for r in results if r["score"] >= 2) lines = [ f"# Job scout report {today}{' (new only)' if new_only else ''}\n", f"Automated coverage: **{len(COMPANIES)}** companies. Manual checks: **{len(MANUAL_CHECK)}**.", f"Eligible (CH/remote): **{len(results)}** · interest matches (score ≥ 2): " f"**{n_match}** · **{n_new}** new since last run\n", ] if stats: lines += write_stats_table(stats, total_secs) if errors: lines.append("## Errors\n") for company, err in errors: lines.append(f"- **{company}**: {err}") lines.append("") strong = [r for r in results if r["score"] >= 6] medium = [r for r in results if 2 <= r["score"] < 6] weak = [r for r in results if r["score"] < 2] if not include_weak and weak: lines.append(f"\n_Hiding {len(weak)} weak/noise roles (score < 2). Use --include-weak to show._") n_decided = sum(1 for r in results if r["url"] in decisions) if n_decided: shown = "hidden" if hide_decided else "tagged inline" lines.append(f"_{n_decided} role(s) already in the decision log ({shown}; " f"see state/decisions.json)._") buckets = [("Strong fit (score >= 6)", strong), ("Medium fit (score 2-5)", medium)] if include_weak: buckets.append(("Weak / noise (score < 2)", weak)) for bucket_name, bucket in buckets: shown = [r for r in bucket if not (hide_decided and r["url"] in decisions)] if not shown: continue lines.append(f"\n## {bucket_name} - {len(shown)} role(s)\n") for r in shown: d = decisions.get(r["url"]) new_tag = " [NEW]" if r["is_new"] else "" decided_tag = f" — 🗂 {d['decision'].upper()}" if d else "" loc_tag = ("CH" if r["in_ch"] else "Remote" if r["remote"] else "Relocation" if r.get("relocation") else "?") lines.append(f"### [{r['score']}] {r['company']} - {r['title']}{new_tag}{decided_tag}") lines.append(f"- Location: {r['location']} *({loc_tag})*") if r.get("posted"): lines.append(f"- Posted: {r['posted']}") lines.append(f"- URL: {r['url']}") if d: note = f" — {d['note']}" if d.get("note") else "" lines.append(f"- 🗂 Decision: **{d['decision']}**{note} ({d.get('date','')})") if r["pos"]: lines.append(f"- Positive: {', '.join(r['pos'])}") if r["neg"]: lines.append(f"- Negative: {', '.join(r['neg'])}") lines.append("") if MANUAL_CHECK: lines.append("\n## Manual check (companies without scrapable APIs)\n") lines.append("These use Cloudflare-protected sites, custom GraphQL APIs, or JS-rendered SPAs.") lines.append("Open each link, scan for new postings since your last quarterly review:\n") for name, note, url in MANUAL_CHECK: lines.append(f"- [ ] **{name}** — {note}: <{url}>") lines.append("") path.write_text("\n".join(lines), encoding="utf-8") def _fetch_company(cid, display, adapter, args): """Fetch one company; returns (cid, display, adapter, args, jobs|None, error|None, secs).""" t0 = time.perf_counter() try: jobs = ADAPTERS[adapter](args) return cid, display, adapter, args, jobs, None, time.perf_counter() - t0 except Exception as e: kind = "unexpected" if not isinstance( e, (urllib.error.URLError, urllib.error.HTTPError, ValueError, RuntimeError) ) else "error" msg = f"{kind}: {e!r}" if kind == "unexpected" else repr(e) return cid, display, adapter, args, None, msg, time.perf_counter() - t0 def _process_company(cid, display, args, jobs, seen, today, last_scrape): """Filter, score, and update seen for one company's job list. Returns (results, stat_dict, warnings) where warnings is a list of (display, msg). """ warnings = [] scraped = len(jobs) prev = (last_scrape.get(cid) or {}).get("scraped", 0) if scraped == 0: if prev and prev > 0: warnings.append(( display, f"0 jobs returned (was {prev} last run — possible dead board/slug/selectors)", )) else: warnings.append(( display, "0 jobs returned (verify board slug/selectors if this is unexpected)", )) # Inclusion allowlist (fails closed — only for boards too large to score in full). title_filter = args.get("_title_filter") if title_filter: jobs = [j for j in jobs if any(_kw_in(k, (j.get("title") or "").lower()) for k in title_filter)] # Exclusion denylist (fails open — preferred; see NOISE_TITLE_EXCLUDE). title_exclude = args.get("_title_exclude") if title_exclude: jobs = [j for j in jobs if not any(_kw_in(k, (j.get("title") or "").lower()) for k in title_exclude)] dates = [d for j in jobs if (d := _parse_posted(j.get("posted")))] newest = max(dates) if dates else None policy = args.get("_location_policy", "default") company_seen = seen.setdefault(cid, {}) title_seen = set() results = [] eligible = match = 0 for j in jobs: jid = str(j.get("id") or j.get("url")) ok, in_ch, is_remote, is_relocation = location_eligibility( j.get("location", ""), policy=policy ) if not ok: continue # Collapse the same role posted once per remote country (title differs only # by a "| Country | Remote" suffix) — dedupe on the title before the first "|". norm_title = re.sub(r"\s+", " ", (j.get("title") or "").split("|")[0]).strip().lower() if norm_title in title_seen: continue title_seen.add(norm_title) eligible += 1 is_new = jid not in company_seen score, pos, neg = score_job(j, title_only=bool(title_filter)) # Pre-filtered boards (e.g. SBB) with German/generic titles: _score_floor keeps # already-relevant results out of the hidden weak bucket. floor = args.get("_score_floor") if floor is not None and score < floor: score = floor if score >= 2: match += 1 results.append({ "company": display, "company_id": cid, "title": j["title"], "location": j["location"], "url": j["url"], "posted": j.get("posted", ""), "score": score, "pos": pos, "neg": neg, "in_ch": in_ch, "remote": is_remote, "relocation": is_relocation, "is_new": is_new, }) if is_new: company_seen[jid] = {"title": j["title"], "first_seen": today} stat = { "company": display, "scraped": scraped, "eligible": eligible, "match": match, "newest": newest, "error": False, } return results, stat, warnings def main(): today = datetime.now(timezone.utc).strftime("%Y-%m-%d") # Record a decision and exit: --decide "<url>" <status> [note words...] if "--decide" in sys.argv: rest = sys.argv[sys.argv.index("--decide") + 1:] if len(rest) < 2: print('Usage: --decide "<url>" <status> [note...]', file=sys.stderr) return url, status, note = rest[0], rest[1], " ".join(rest[2:]) decisions = load_decisions() prev = decisions.get(url, {}) # Preserve any company/title already stored (filled when a report run tagged the URL). decisions[url] = { "company": prev.get("company", ""), "title": prev.get("title", ""), "decision": status, "note": note, "date": today, } save_decisions(decisions) print(f"Recorded: {status} — {url}", file=sys.stderr) return only, new_only, include_weak, hide_decided = None, False, False, False for arg in sys.argv[1:]: if arg in ("--help", "-h"): print(__doc__) return if arg == "--new-only": new_only = True elif arg == "--include-weak": include_weak = True elif arg == "--hide-decided": hide_decided = True elif arg.startswith("--only="): only = arg.split("=", 1)[1] targets = [(cid, display, adapter, args) for cid, display, adapter, args in COMPANIES if not only or cid == only] if only and not targets: known = ", ".join(c[0] for c in COMPANIES) print(f"Unknown --only={only!r}. Known ids: {known}", file=sys.stderr) sys.exit(2) seen = load_seen() decisions = load_decisions() last_scrape = load_last_scrape() all_results, errors, stats = [], [], [] new_last_scrape = dict(last_scrape) run_start = time.perf_counter() # API adapters in parallel; playwright serial (shared browser is not thread-safe). api_targets = [t for t in targets if t[2] != "playwright"] pw_targets = [t for t in targets if t[2] == "playwright"] # Preserve COMPANIES order in stats/report via ordered result map. fetch_results = {} try: if api_targets: print(f"Fetching {len(api_targets)} API boards " f"({API_FETCH_WORKERS} workers)...", file=sys.stderr) with ThreadPoolExecutor(max_workers=API_FETCH_WORKERS) as pool: futs = { pool.submit(_fetch_company, cid, display, adapter, args): cid for cid, display, adapter, args in api_targets } for fut in as_completed(futs): cid, display, adapter, args, jobs, err, secs = fut.result() print(f" {display}: " f"{'ERROR' if err else f'{len(jobs)} jobs'} ({secs:.1f}s)", file=sys.stderr) fetch_results[cid] = (display, args, jobs, err, secs) for cid, display, adapter, args in pw_targets: print(f"Fetching {display} (playwright)...", file=sys.stderr) _, display, _, args, jobs, err, secs = _fetch_company( cid, display, adapter, args ) print(f" {display}: " f"{'ERROR' if err else f'{len(jobs)} jobs'} ({secs:.1f}s)", file=sys.stderr) fetch_results[cid] = (display, args, jobs, err, secs) # Process in COMPANIES order for stable stats tables. for cid, display, adapter, args in targets: display, args, jobs, err, secs = fetch_results[cid] if err is not None: errors.append((display, err)) stats.append({ "company": display, "scraped": 0, "eligible": 0, "match": 0, "newest": None, "secs": secs, "error": True, }) continue results, stat, warnings = _process_company( cid, display, args, jobs, seen, today, last_scrape ) stat["secs"] = secs for w in warnings: errors.append(w) # Soft warning only when 0 jobs — still mark row with ⚠️ if was previously >0 prev = (last_scrape.get(cid) or {}).get("scraped", 0) if stat["scraped"] == 0 and prev > 0: stat["error"] = True stats.append(stat) all_results.extend(results) new_last_scrape[cid] = { "scraped": stat["scraped"], "date": today, "display": display, } finally: save_seen(seen) save_last_scrape(new_last_scrape) _close_browser() total_secs = time.perf_counter() - run_start if new_only: all_results = [r for r in all_results if r["is_new"]] all_results.sort(key=lambda r: (-r["score"], r["company"], r["title"])) REPORTS_DIR.mkdir(parents=True, exist_ok=True) report_path = REPORTS_DIR / f"{today}.md" write_report(report_path, all_results, errors, new_only, include_weak, stats=stats, total_secs=total_secs, decisions=decisions, hide_decided=hide_decided) # Plain data dump of every eligible role, unfiltered by keyword score — fit judgment # for these is done in conversation against the profile, not by the keyword scorer. json_path = REPORTS_DIR / f"{today}.json" dump = [{ "company": r["company"], "title": r["title"], "location": r["location"], "url": r["url"], "posted": r["posted"], "is_new": r["is_new"], "decision": decisions.get(r["url"]), "keyword_signal": {"score": r["score"], "pos": r["pos"], "neg": r["neg"]}, "loc_flags": { "in_ch": r["in_ch"], "remote": r["remote"], "relocation": r["relocation"], }, } for r in all_results] _atomic_write_json(json_path, dump) n_new = sum(1 for r in all_results if r["is_new"]) print(f"\nReport written: {report_path}", file=sys.stderr) print(f"Total matches: {len(all_results)} ({n_new} new) | " f"scanned {len(stats)} companies in {total_secs:.1f}s", file=sys.stderr) if errors: print(f"Errors/warnings: {len(errors)} - see report", file=sys.stderr) # === Adapter coverage (refreshed 2026-07-18) ================================== # 34 companies automated across 14 adapter types; 1 in MANUAL_CHECK (Oracle). # # Location policy: # default — CH + Denmark + Europe/EMEA/global remote (work-from CH or DK) # norway_only — Equinor # dk_no — NATO (Denmark + Norway only; no CH) # # Automated (COMPANIES above): # workday nvidia, novartis, equinor (norway_only) # ashby kraken, openai, confluent, snowflake # greenhouse anthropic, gitlab, grafana, databricks, datadog, elastic, fivetran # pcsx microsoft # smartrecruiters metgroup, ldc # rss bis # amazon amazon (Amazon + AWS share one board) # phenom roche (Cisco still uses the playwright path) # teamtailor telenor, axpo (custom domain via base_url) # getro (no company configured — adapter kept for future VC talent networks) # onlyfy bitcoin_suisse # lever palantir, quantco # taleo nato (dk_no; nato.taleo.net §2 — marketing page is nato.int/vacancies) # json swissgrid # sbb sbb # bkw bkw # playwright google, apple, meta, cisco, ruag, postfinance, bfh # # MANUAL_CHECK: Oracle (ORC needs CH geographyId). # ============================================================================== if __name__ == "__main__": main()