# ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : jobka/connectors/dayforce.py # Rôle : Classe de plateforme Dayforce HCM (jobs.dayforcehcm.com) — API JSON # publique du frontal Next.js, un employeur = une sous-classe # Créé : 2026-08-22 Modifié : 2026-08-26 # ============================================================================= """Plateforme Dayforce HCM (ex-Ceridian) — jobs.dayforcehcm.com. API JSON publique du frontal (celle que le navigateur appelle) : - jeton : GET /api/auth/csrf -> {csrfToken} (+ cookie de session next-auth) ; toute requête POST doit porter l'en-tête X-CSRF-TOKEN, sinon 403. - liste : POST /api/geo//jobposting/search body {clientNamespace, jobBoardCode, cultureCode, paginationStart} (pages FIXES de 25, champ découvert dans le chunk Next.js : paginationStart=(page-1)*25) -> {jobPostings:[{jobPostingId, jobTitle, jobDescription (HTML COMPLET dès la liste), jobBoardId, postingStartTimestampUTC, postingExpiryTimestampUTC, isEvergreen, postingLocations:[{formattedAddress, isoCountryCode, stateCode, cityName, coordinates "lat:…;lng:…"}]}], maxCount} - détail : GET /api/geo//jobposting//// -> jobPostingAttributes [{name: "PayType"/"PayClass", value}] + jobPostingContent (header/description/footer). Visité avec cache BD + budget par cycle (n'apporte que les attributs type/paie). cultureCode « fr-CA » : Dayforce sert la traduction française quand elle existe (public québécois), sinon le texte original. """ from __future__ import annotations import os import re import time import requests from ..schema import JobPosting, clean_html, is_quebec_location from .base import BaseConnector BASE = "https://jobs.dayforcehcm.com" PAGE_SIZE = 25 # taille fixe côté serveur (paginationStart) MAX_DETAILS = int(os.environ.get("JOBKA_DAYFORCE_DETAIL_LIMIT", "80")) _COORD_RE = re.compile(r"lat:(-?\d+(?:\.\d+)?);lng:(-?\d+(?:\.\d+)?)") class DayforceConnector(BaseConnector): """Base Dayforce — sous-classes : définir source_id, EMPLOYER, NS (clientNamespace) et au besoin BOARD (jobBoardCode, défaut CANDIDATEPORTAL).""" ats = "dayforce" request_delay = 0.8 EMPLOYER = "" NS = "" # clientNamespace (ex. « lavieenrose ») BOARD = "CANDIDATEPORTAL" # jobBoardCode du site carrières public CULTURE = "fr-CA" quebec_only = True max_pages = 40 fetch_details = True # attributs PayClass/PayType (cache + budget) # -- lieux ------------------------------------------------------------ @staticmethod def _is_qc(loc: dict) -> bool: if (loc.get("isoCountryCode") or "").upper() not in ("CA", ""): return False if (loc.get("stateCode") or "").upper() == "QC": return True return is_quebec_location( f"{loc.get('cityName') or ''}, {loc.get('formattedAddress') or ''}") def _keep(self, locations: list[dict], virtual: bool) -> bool: if not self.quebec_only: return True return any(self._is_qc(l) for l in locations) # -- détail (attributs type d'emploi / paie) --------------------------- def _fetch_detail(self, board_id: int, jid: str) -> dict: data = self.get( f"{BASE}/api/geo/{self.NS}/jobposting/{self.NS}/{self.CULTURE}/" f"{board_id}/{jid}").json() attrs = {a.get("name"): a.get("value") for a in (data.get("jobPostingAttributes") or []) if a.get("name")} content = data.get("jobPostingContent") or {} desc = "".join(content.get(k) or "" for k in ("jobDescriptionHeader", "jobDescription", "jobDescriptionFooter")) return {"attributes": attrs, "description": clean_html(desc)} # -- recherche (garde-fou 5xx transitoire) ------------------------------ def _search_page(self, search_url: str, csrf: str, start: int) -> dict: """POST de recherche avec garde-fou : le frontal Dayforce renvoie parfois un 5xx passager (502 Bad Gateway observé le 2026-08-26 sur groupemaurice) qui avortait tout le sync — on retente une fois après pause, puis on lève (sync en échec, jamais un faux 0).""" for attempt in range(2): try: return self.post( search_url, json={"clientNamespace": self.NS, "jobBoardCode": self.BOARD, "cultureCode": self.CULTURE, "paginationStart": start}, headers={"X-CSRF-TOKEN": csrf, "Accept": "application/json"}).json() except requests.HTTPError as e: status = getattr(e.response, "status_code", 0) or 0 if attempt == 0 and status >= 500: time.sleep(10) continue raise raise RuntimeError("unreachable") # -- contrat ------------------------------------------------------------ def fetch(self) -> list[JobPosting]: csrf = (self.get(f"{BASE}/api/auth/csrf").json() or {}).get( "csrfToken") or "" search_url = f"{BASE}/api/geo/{self.NS}/jobposting/search" out: list[JobPosting] = [] seen: set[str] = set() details_used = 0 start = 0 for _ in range(self.max_pages): data = self._search_page(search_url, csrf, start) postings = data.get("jobPostings") or [] if not postings: break for p in postings: jid = str(p.get("jobPostingId") or "") if not jid or jid in seen: continue seen.add(jid) locations = p.get("postingLocations") or [] if not self._keep(locations, p.get("hasVirtualLocation")): continue loc = next((l for l in locations if self._is_qc(l)), locations[0] if locations else {}) lat = lng = None if m := _COORD_RE.search(loc.get("coordinates") or ""): lat, lng = float(m.group(1)), float(m.group(2)) job = JobPosting( source=self.source_id, external_id=jid, url=f"{BASE}/{self.CULTURE}/{self.NS}/{self.BOARD}" f"/jobs/{jid}", employer=self.EMPLOYER, title=p.get("jobTitle") or "", description=clean_html(p.get("jobDescription") or ""), address=loc.get("formattedAddress") or "", city=loc.get("cityName") or "", location_label=loc.get("formattedAddress") or (loc.get("cityName") or ""), date_posted=p.get("postingStartTimestampUTC"), date_deadline=p.get("postingExpiryTimestampUTC"), lat=lat, lng=lng, ats=self.ats, ) if p.get("isEvergreen"): job.details["evergreen"] = True board_id = p.get("jobBoardId") or 1 if self.fetch_details: key = ((p.get("postingStartTimestampUTC") or "") + (p.get("jobTitle") or "")[:40]) if details_used < MAX_DETAILS: fresh = [False] def _fn(b=board_id, i=jid, fresh=fresh): fresh[0] = True return self._fetch_detail(b, i) d = self.detail(jid, key, _fn) if fresh[0]: details_used += 1 else: d = self.stale_detail(jid) attrs = d.get("attributes") or {} if attrs.get("PayClass"): job.details["employment_label"] = attrs["PayClass"] if attrs.get("PayType"): job.details["pay_type_label"] = attrs["PayType"] if len(d.get("description") or "") > len(job.description): job.description = d["description"] out.append(job) total = int(data.get("maxCount") or 0) start += PAGE_SIZE if total and start >= total: break return out