# ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : jobka/connectors/adp.py # Rôle : Classe de plateforme ADP Workforce Now (centre de carrières) — # API JSON publique job-requisitions, un employeur = une sous-classe # Créé : 2026-08-18 Modifié : 2026-08-30 # ============================================================================= """Plateforme ADP Workforce Now (workforcenow.adp.com). API JSON publique du centre de carrières : - liste : GET /mascsr/default/careercenter/public/events/staffing/v1/ job-requisitions?cid=&ccId=&lang=&$top=&$skip= -> {jobRequisitions:[{itemID, requisitionTitle, postDate, payGradeRange, workLevelCode, requisitionLocations, customFieldGroup}]} - détail : GET .../job-requisitions/?cid=… -> requisitionDescription (HTML). Visité avec cache BD + budget. """ from __future__ import annotations import os import time from ..schema import JobPosting, clean_html, is_quebec_location from .base import BaseConnector PAGE_SIZE = 100 MAX_DETAILS = int(os.environ.get("JOBKA_ADP_DETAIL_LIMIT", "80")) _API = ("https://workforcenow.adp.com/mascsr/default/careercenter/public/" "events/staffing/v1/job-requisitions") _SALARY_UNIT = {"AN": "year", "ANNUEL": "year", "YR": "year", "YEAR": "year", "HR": "hour", "HO": "hour", "HORAIRE": "hour", "HOUR": "hour"} class ADPWorkforceNowConnector(BaseConnector): """Base ADP WFN — sous-classes : définir source_id, EMPLOYER, CID, CCID.""" ats = "adp" request_delay = 1.0 EMPLOYER = "" CID = "" # GUID du centre de carrières CCID = "19000101_000001" # identifiant du « career center » LANG = "fr_CA" quebec_only = True max_pages = 10 FORCE_CITY = "" # employeurs mono-site QC dont le babillard n'expose # aucune requisitionLocation : ville imposée def _get_json(self, url: str, params: dict) -> dict: # ADP renvoie parfois 200 avec un corps vide/HTML (fenêtres de # maintenance nocturnes constatées les 29-30 août 2026) : on retente # avec backoff avant d'abandonner, et l'erreur finale nomme le statut # HTTP + content-type pour que sync_log soit diagnostiquable. last: ValueError | None = None for attempt in range(3): resp = self.get(url, params=params, headers={"Accept": "application/json"}) try: return resp.json() except ValueError as exc: last = exc time.sleep(10 * (attempt + 1)) raise RuntimeError( f"ADP non-JSON après 3 essais (HTTP {resp.status_code}, " f"{resp.headers.get('content-type')}) : {last}") def _params(self, extra: dict | None = None) -> dict: p = {"cid": self.CID, "ccId": self.CCID, "lang": self.LANG, "locale": self.LANG} p.update(extra or {}) return p @staticmethod def _locations(req: dict) -> list[dict]: out = [] for loc in req.get("requisitionLocations") or []: addr = loc.get("address") or {} out.append({ "city": addr.get("cityName") or "", "prov": ((addr.get("countrySubdivisionLevel1") or {}) .get("codeValue") or ""), "postal": addr.get("postalCode") or "", "label": ((loc.get("nameCode") or {}).get("shortName") or "").strip(), }) return out def _keep(self, locations: list[dict]) -> bool: if not self.quebec_only: return True if not locations and self.FORCE_CITY: return True return any(l["prov"].upper() == "QC" or is_quebec_location(f"{l['label']} {l['city']}") for l in locations) def _fetch_detail(self, item_id: str) -> dict: data = self._get_json(f"{_API}/{item_id}", params=self._params()) return {"description": clean_html( data.get("requisitionDescription") or "")} def fetch(self) -> list[JobPosting]: out: list[JobPosting] = [] details_used = 0 skip = 0 for _ in range(self.max_pages): data = self._get_json(_API, params=self._params( {"$top": str(PAGE_SIZE), "$skip": str(skip)})) reqs = data.get("jobRequisitions") or [] if not reqs and skip == 0 and self.LANG != "en_US": # certains centres de carrières ne répondent qu'en anglais self.LANG = "en_US" continue if not reqs: break for r in reqs: locations = self._locations(r) if not self._keep(locations): continue iid = str(r.get("itemID") or "") if not iid: continue qc = next((l for l in locations if l["prov"].upper() == "QC" or is_quebec_location(f"{l['label']} {l['city']}")), locations[0] if locations else {"city": self.FORCE_CITY, "postal": "", "label": self.FORCE_CITY}) job = JobPosting( source=self.source_id, external_id=iid, url=("https://workforcenow.adp.com/mascsr/default/mdf/" f"recruitment/recruitment.html?cid={self.CID}" f"&ccId={self.CCID}&lang={self.LANG}&jobId={iid}"), employer=self.EMPLOYER, title=r.get("requisitionTitle") or "", city=qc["city"], postal_code=qc["postal"], location_label=qc["label"], date_posted=r.get("postDate") or None, ats=self.ats, ) pay = r.get("payGradeRange") or {} lo = ((pay.get("minimumRate") or {}).get("amountValue")) hi = ((pay.get("maximumRate") or {}).get("amountValue")) if lo: unit = None for c in ((r.get("customFieldGroup") or {}) .get("codeFields") or []): if ((c.get("nameCode") or {}) .get("codeValue")) == "SalaryType": unit = _SALARY_UNIT.get( str(c.get("codeValue") or "").upper()) \ or _SALARY_UNIT.get( str(c.get("shortName") or "").upper()) job.salary_min = float(lo) job.salary_max = float(hi) if hi else None job.salary_unit = unit or ("year" if float(lo) > 5000 else "hour") wl = (r.get("workLevelCode") or {}).get("shortName") if wl: job.details["employment_label"] = wl key = (r.get("postDate") or "") + (job.title or "")[:40] if details_used < MAX_DETAILS: fresh = [False] def _fn(i=iid, fresh=fresh): fresh[0] = True return self._fetch_detail(i) d = self.detail(iid, key, _fn) if fresh[0]: details_used += 1 job.description = d.get("description", "") out.append(job) if len(reqs) < PAGE_SIZE: break skip += PAGE_SIZE return out