#!/usr/bin/env python3 # ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : scripts/probe_ats.py # Rôle : Sondage massif de slugs candidats sur les 10 ATS supportés — # un slug est CONFIRMÉ si l'API répond ≥1 offre ET ≥1 offre au # Québec (réutilise is_quebec_location du projet). Émet des entrées # prêtes pour data/feeds-*.json (scripts/gen_connectors.py). # Créé : 2026-08-18 Modifié : 2026-08-22 # ============================================================================= """Sonde des slugs candidats sur les 9 plateformes ATS de Job·Ka. Usage : python3 scripts/probe_ats.py candidats.jsonl --out confirmed.json \ [--rejected rejected.jsonl] [--workers 8] Entrée (JSONL, ou JSON liste) — un candidat par ligne : {"slug": "acme", "employer": "Acme inc.", "ats": "lever"?, "sectors": ["TI"], "cities": ["Montréal"], "group": "Acme"?, "no_workday": true?, "tenant"/"host"/"site" (workday)?} - sans "ats" : les 9 plateformes sont sondées dans l'ordre DEFAULT_ORDER (Workday en dernier — découverte automatique de l'hôte wdN et du site par le redirect de la racine du tenant). - les candidats partageant le même "group" (défaut : employer) sont des variantes du même employeur : on s'arrête à la première confirmation. - les slugs/tenants déjà connectés (sources.json + jobka/connectors/*.py) sont sautés automatiquement. Politesse : timeout 6 s, throttle 0,3 s par domaine d'API, 8 threads. """ from __future__ import annotations import argparse import json import re import sys import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from urllib.parse import urlparse import requests ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) from jobka.normalize import is_quebec_location # noqa: E402 UA = "JobKaBot/1.0 (+https://www.job-ka.com; contact@spboucher.ai)" TIMEOUT = 6 THROTTLE = 0.3 # s entre deux requêtes vers un même domaine d'API WD_HOSTS = ["wd3", "wd1", "wd5", "wd10", "wd12", "wd103", "wd502", "wd504"] # Noms de sites carrières Workday fréquents — essayés quand le site est # inconnu. L'API cxs répond 422 si le tenant/hôte est faux mais 404 si seul # le site l'est : on découvre d'abord l'hôte, puis on devine le site. WD_COMMON_SITES = [ "External", "Careers", "External_Careers", "ExternalCareers", "careers", "External_Career_Site", "CareerSite", "Career", "Global", "external", "Global_Careers", "GlobalCareers", "jobs", "Jobs", "External_Site", "ExternalSite", "Externe", "Carrieres", "carrieres", "Recruiting", "Recrutement", "Emplois", "External_Portal", "EXTERNAL", "Ext", "EXT", ] DEFAULT_ORDER = ["lever", "greenhouse", "ashby", "smartrecruiters", "workable", "recruitee", "breezy", "bamboohr", "dayforce", "workday"] # --- throttling par domaine d'API ------------------------------------------- _locks: dict[str, threading.Lock] = {} _last: dict[str, float] = {} _glock = threading.Lock() _nreq = [0] def _req(family: str, method: str, url: str, **kw) -> requests.Response: """Requête HTTP throttlée (verrou + espacement par famille/domaine).""" with _glock: lock = _locks.setdefault(family, threading.Lock()) _nreq[0] += 1 headers = {"User-Agent": UA} headers.update(kw.pop("headers", None) or {}) with lock: wait = THROTTLE - (time.time() - _last.get(family, 0.0)) if wait > 0: time.sleep(wait) try: return requests.request(method, url, timeout=TIMEOUT, headers=headers, **kw) finally: _last[family] = time.time() def _qc_stats(labels: list[list[str]]) -> tuple[int, list[str]]: """[[libellés d'une offre], …] -> (nb d'offres QC, libellés QC exemples).""" qc, ex = 0, [] for locs in labels: hit = next((l for l in locs if l and is_quebec_location(l)), None) if hit: qc += 1 if len(ex) < 5 and hit not in ex: ex.append(hit) return qc, ex # --- sondes par ATS ---------------------------------------------------------- def probe_lever(slug: str, cand: dict) -> dict | None: r = _req("api.lever.co", "GET", f"https://api.lever.co/v0/postings/{slug}?mode=json") if r.status_code != 200: return None data = r.json() if not isinstance(data, list) or not data: return None labels = [] for p in data: locs = [(p.get("categories") or {}).get("location") or ""] locs += [l for l in (p.get("allLocations") or []) if isinstance(l, str)] labels.append(locs) qc, ex = _qc_stats(labels) return {"found": len(data), "qc": qc, "qc_ex": ex, "api_name": ""} def probe_greenhouse(slug: str, cand: dict) -> dict | None: r = _req("boards-api.greenhouse.io", "GET", f"https://boards-api.greenhouse.io/v1/boards/{slug}/jobs") if r.status_code != 200: return None jobs = r.json().get("jobs") or [] if not jobs: return None labels = [[(j.get("location") or {}).get("name") or ""] for j in jobs] qc, ex = _qc_stats(labels) name = "" if qc: try: r2 = _req("boards-api.greenhouse.io", "GET", f"https://boards-api.greenhouse.io/v1/boards/{slug}") if r2.status_code == 200: name = r2.json().get("name") or "" except requests.RequestException: pass return {"found": len(jobs), "qc": qc, "qc_ex": ex, "api_name": name} def probe_ashby(slug: str, cand: dict) -> dict | None: r = _req("api.ashbyhq.com", "GET", f"https://api.ashbyhq.com/posting-api/job-board/{slug}") if r.status_code != 200: return None data = r.json() jobs = data.get("jobs") or [] if not jobs: return None labels = [] for j in jobs: locs = [j.get("location") or ""] locs += [(s.get("location") or "") for s in (j.get("secondaryLocations") or [])] labels.append(locs) qc, ex = _qc_stats(labels) return {"found": len(jobs), "qc": qc, "qc_ex": ex, "api_name": data.get("name") or ""} def probe_smartrecruiters(slug: str, cand: dict) -> dict | None: variants = [slug] if slug != slug.capitalize(): variants.append(slug.capitalize()) for v in variants: r = _req("api.smartrecruiters.com", "GET", f"https://api.smartrecruiters.com/v1/companies/{v}/postings?limit=100") if r.status_code != 200: continue data = r.json() content = data.get("content") or [] total = int(data.get("totalFound") or len(content)) if not content: continue labels, name = [], "" for p in content: loc = p.get("location") or {} country = (loc.get("country") or "").lower() lbl = ", ".join(x for x in (loc.get("city"), loc.get("region"), loc.get("country")) if x) reg = (loc.get("region") or "").lower() ok = reg in ("qc", "quebec", "québec") or ( country in ("ca", "canada") and is_quebec_location(lbl)) labels.append([lbl] if ok else [""]) name = name or (p.get("company") or {}).get("name") or "" qc, ex = _qc_stats(labels) return {"found": total, "qc": qc, "qc_ex": ex, "api_name": name, "org_override": v} return None def probe_workable(slug: str, cand: dict) -> dict | None: r = _req("apply.workable.com", "GET", f"https://apply.workable.com/api/v1/widget/accounts/{slug}?details=false") if r.status_code != 200: return None data = r.json() jobs = data.get("jobs") or [] if not jobs: return None labels = [[", ".join(x for x in (j.get("city"), j.get("state"), j.get("country")) if x)] for j in jobs] qc, ex = _qc_stats(labels) return {"found": len(jobs), "qc": qc, "qc_ex": ex, "api_name": data.get("name") or ""} def probe_recruitee(slug: str, cand: dict) -> dict | None: r = _req(f"{slug}.recruitee.com", "GET", f"https://{slug}.recruitee.com/api/offers/") if r.status_code != 200: return None offers = r.json().get("offers") or [] if not offers: return None labels = [] for o in offers: locs = [o.get("location") or "", ", ".join(x for x in (o.get("city"), o.get("country")) if x)] labels.append(locs) qc, ex = _qc_stats(labels) return {"found": len(offers), "qc": qc, "qc_ex": ex, "api_name": ""} def probe_breezy(slug: str, cand: dict) -> dict | None: r = _req(f"{slug}.breezy.hr", "GET", f"https://{slug}.breezy.hr/json") if r.status_code != 200: return None try: data = r.json() except ValueError: return None if not isinstance(data, list) or not data: return None labels, name = [], "" for p in data: loc = p.get("location") or {} labels.append([loc.get("name") or "", ", ".join(x for x in (loc.get("city"), (loc.get("state") or {}).get("name") if isinstance(loc.get("state"), dict) else loc.get("state"), loc.get("country", {}).get("name") if isinstance(loc.get("country"), dict) else loc.get("country")) if x)]) name = name or (p.get("company") or {}).get("name", "") \ if isinstance(p.get("company"), dict) else name qc, ex = _qc_stats(labels) return {"found": len(data), "qc": qc, "qc_ex": ex, "api_name": name} def probe_bamboohr(slug: str, cand: dict) -> dict | None: r = _req(f"{slug}.bamboohr.com", "GET", f"https://{slug}.bamboohr.com/careers/list", headers={"User-Agent": UA, "Accept": "application/json"}) if r.status_code != 200: return None try: data = r.json() except ValueError: return None jobs = data.get("result") or [] if not jobs: return None labels = [] for j in jobs: loc = j.get("location") or {} labels.append([", ".join(x for x in (loc.get("city"), loc.get("state")) if x)]) qc, ex = _qc_stats(labels) return {"found": len(jobs), "qc": qc, "qc_ex": ex, "api_name": (data.get("meta") or {}).get("companyName") or ""} def _wd_post(tenant: str, host: str, site: str, search: str = "") -> requests.Response: url = (f"https://{tenant}.{host}.myworkdayjobs.com" f"/wday/cxs/{tenant}/{site}/jobs") return _req(f"{tenant}.{host}.myworkdayjobs.com", "POST", url, json={"appliedFacets": {}, "limit": 20, "offset": 0, "searchText": search}) def _wd_eval(tenant: str, host: str, site: str) -> dict | None: """Site valide -> confirme s'il y a ≥1 offre dont ≥1 au Québec (page 1 brute, puis recherches « Québec »).""" total = 0 for st in ("", "Québec", "Quebec, Canada"): try: r = _wd_post(tenant, host, site, st) except requests.RequestException: return None if r.status_code != 200: return None try: data = r.json() except ValueError: return None items = data.get("jobPostings") or [] total = max(total, int(data.get("total") or 0)) if st == "" and not total: return None labels = [[i.get("locationsText") or ""] for i in items] qc, ex = _qc_stats(labels) if qc: return {"found": total, "qc": qc, "qc_ex": ex, "api_name": "", "tenant": tenant, "host": host, "site": site} return None def probe_workday(slug: str, cand: dict) -> dict | None: """La page racine myworkdayjobs répond 406 aux bots : on passe par l'API cxs — 422 = tenant absent de cet hôte, 404 = hôte bon mais site inconnu, 200 = tenant+site valides. Hôte scanné puis site deviné au besoin.""" tenant = (cand.get("tenant") or slug).lower() hosts = [cand["host"]] if cand.get("host") else WD_HOSTS for host in hosts: try: r = _wd_post(tenant, host, cand.get("site") or "__jobka_probe__") except requests.RequestException: continue if r.status_code == 200: # site fourni et valide return _wd_eval(tenant, host, cand["site"]) if r.status_code != 404: continue # 422/5xx : hôte suivant # hôte confirmé — deviner le nom du site t = tenant derived = [f"{t}careers", f"{t}_careers", f"{t.capitalize()}Careers", f"{t.capitalize()}_Careers", f"{t.upper()}_CAREERS", f"{t.capitalize()}_External_Careers", t.capitalize(), f"Carrieres{t.capitalize()}", f"{t}_carrieres"] seen: list[str] = [] for s in ([cand["site"]] if cand.get("site") else []) + derived \ + WD_COMMON_SITES: if s in seen: continue seen.append(s) if len(seen) > 34: break try: r2 = _wd_post(tenant, host, s) except requests.RequestException: continue if r2.status_code == 200: return _wd_eval(tenant, host, s) return None # hôte trouvé mais nom de site introuvable return None def probe_dayforce(slug: str, cand: dict) -> dict | None: """Dayforce HCM — même API que jobka/connectors/dayforce.py : csrf anonyme puis recherche paginée. Le namespace est sensible à la casse tel qu'écrit dans l'URL ; babillard par défaut CANDIDATEPORTAL.""" base = "https://jobs.dayforcehcm.com" boards = [b for b in (cand.get("board"), "CANDIDATEPORTAL") if b] sess = requests.Session() sess.headers["User-Agent"] = UA with _glock: lock = _locks.setdefault("jobs.dayforcehcm.com", threading.Lock()) for board in dict.fromkeys(boards): with lock: wait = THROTTLE - (time.time() - _last.get("dayforce", 0.0)) if wait > 0: time.sleep(wait) try: csrf = (sess.get(f"{base}/api/auth/csrf", timeout=TIMEOUT).json() or {}).get( "csrfToken") or "" r = sess.post( f"{base}/api/geo/{slug}/jobposting/search", timeout=TIMEOUT, headers={"X-CSRF-TOKEN": csrf, "Accept": "application/json"}, json={"clientNamespace": slug, "jobBoardCode": board, "cultureCode": "fr-CA", "paginationStart": 0}) except requests.RequestException: return None finally: _last["dayforce"] = time.time() _nreq[0] += 1 if r.status_code != 200: continue try: data = r.json() except ValueError: continue postings = data.get("jobPostings") or [] total = int(data.get("maxCount") or len(postings)) if not postings: continue labels = [] for p in postings: locs = [] for l in (p.get("postingLocations") or []): if (l.get("stateCode") or "").upper() == "QC" and \ (l.get("isoCountryCode") or "CA").upper() == "CA": locs.append(l.get("formattedAddress") or f"{l.get('cityName') or ''}, QC") else: locs.append(l.get("formattedAddress") or "") labels.append(locs) qc, ex = _qc_stats(labels) return {"found": total, "qc": qc, "qc_ex": ex, "api_name": "", "ns": slug, "board": board} return None PROBES = { "lever": probe_lever, "greenhouse": probe_greenhouse, "ashby": probe_ashby, "smartrecruiters": probe_smartrecruiters, "workable": probe_workable, "recruitee": probe_recruitee, "breezy": probe_breezy, "bamboohr": probe_bamboohr, "workday": probe_workday, "dayforce": probe_dayforce, } # --- registre existant (pour sauter les slugs déjà connectés) ---------------- def known_slugs() -> set[str]: known: set[str] = set() src = ROOT / "data" / "sources.json" if src.exists(): for s in json.loads(src.read_text())["sources"]: known.add(s["id"]) pat = re.compile(r"^\s*(?:ORG|BOARD|COMPANY|TENANT)\s*=\s*['\"]([^'\"]+)", re.M) for f in (ROOT / "jobka" / "connectors").glob("*.py"): for m in pat.finditer(f.read_text(encoding="utf-8")): known.add(m.group(1).lower()) return known # --- pipeline ----------------------------------------------------------------- def probe_group(group: str, cands: list[dict], stats: dict, slock: threading.Lock) -> tuple[dict | None, dict]: """Essaie chaque variante du groupe ; retourne (feed confirmé, rejet).""" best = None for cand in cands: slug = cand["slug"] order = [cand["ats"]] if cand.get("ats") else [ a for a in DEFAULT_ORDER if not (cand.get("no_workday") and a == "workday")] for ats in order: with slock: stats.setdefault(ats, [0, 0])[0] += 1 try: res = PROBES[ats](slug, cand) except Exception: # API hors-format = slug non confirmé res = None if not res: continue if res["qc"] >= 1: with slock: stats[ats][1] += 1 feed = {"ats": ats, "employer": cand.get("employer") or slug, "sectors": cand.get("sectors") or [], "cities": cand.get("cities") or [], "url": cand.get("url", ""), "_found": res["found"], "_qc": res["qc"], "_qc_locations": res["qc_ex"], "_api_name": res.get("api_name", "")} if ats == "workday": feed.update(tenant=res["tenant"], host=res["host"], site=res["site"]) elif ats == "dayforce": feed.update(ns=res["ns"], board=res["board"], source_id=res["ns"].lower()) else: feed["org"] = res.get("org_override", slug) return feed, {} if best is None or res["found"] > best.get("found", 0): best = {"ats": ats, "slug": slug, "found": res["found"]} reject = {"group": group, "slugs": [c["slug"] for c in cands], "reason": "no_qc" if best else "no_api", "best": best} return None, reject def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("candidates") ap.add_argument("--out", required=True) ap.add_argument("--rejected", default="") ap.add_argument("--workers", type=int, default=8) args = ap.parse_args() raw = Path(args.candidates).read_text(encoding="utf-8").strip() if raw.startswith("["): cands = json.loads(raw) else: cands = [json.loads(l) for l in raw.splitlines() if l.strip()] known = known_slugs() groups: dict[str, list[dict]] = {} skipped = 0 for c in cands: slug = (c.get("slug") or "").strip() if not slug: continue if slug.lower() in known or (c.get("tenant") or "").lower() in known: skipped += 1 continue groups.setdefault(c.get("group") or c.get("employer") or slug, []).append(c) n_slugs = sum(len(v) for v in groups.values()) print(f"[probe] {len(groups)} groupes / {n_slugs} slugs à sonder " f"({skipped} déjà connectés, sautés)", flush=True) stats: dict[str, list[int]] = {} slock = threading.Lock() confirmed, rejected = [], [] t0 = time.time() with ThreadPoolExecutor(max_workers=args.workers) as ex: futs = {ex.submit(probe_group, g, cs, stats, slock): g for g, cs in groups.items()} done = 0 for fut in as_completed(futs): done += 1 try: feed, reject = fut.result() except Exception as exc: print(f"[probe] {futs[fut]} : erreur {exc}", flush=True) continue if feed: confirmed.append(feed) print(f" ✓ {feed['employer']} -> {feed['ats']}" f" ({feed.get('org') or feed.get('tenant')}) " f"qc={feed['_qc']}/{feed['_found']}", flush=True) elif reject: rejected.append(reject) if done % 50 == 0: print(f"[probe] {done}/{len(groups)} groupes, " f"{len(confirmed)} confirmés, {_nreq[0]} requêtes, " f"{time.time()-t0:.0f}s", flush=True) confirmed.sort(key=lambda f: (f["ats"], f.get("org") or f.get("tenant", ""))) Path(args.out).write_text( json.dumps(confirmed, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") if args.rejected: Path(args.rejected).write_text( "\n".join(json.dumps(r, ensure_ascii=False) for r in rejected) + "\n", encoding="utf-8") print(f"\n[probe] terminé en {time.time()-t0:.0f}s — {_nreq[0]} requêtes") print(f"[probe] slugs sondés par ATS (sondés/confirmés) :") for ats in DEFAULT_ORDER: p, c = stats.get(ats, [0, 0]) print(f" {ats:16s} {p:5d} / {c}") print(f"[probe] {len(confirmed)} groupes confirmés, " f"{len(rejected)} rejetés -> {args.out}") if __name__ == "__main__": main()