# ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : jobka/connectors/workland.py # Rôle : Classe de plateforme Workland / Atlas (ATS québécois — CSS # scolaires, cégeps, santé, sociétés d'État) — API JSON publique # jobs-prod.workland.com (celle du frontal atlas.workland.com) # Créé : 2026-08-18 Modifié : 2026-08-19 # ============================================================================= """Plateforme Workland (atlas.workland.com). Réécriture 2026-08-19 : l'ancienne version grattait les pages carrières des employeurs (liens atlas rendus serveur) et n'avait AUCUNE description (la SPA Atlas n'expose rien côté serveur). L'API publique du frontal fait tout en deux requêtes : - offres : POST https://jobs-prod.workland.com/api/v1/job/published-external body {"load_with": ["locations"], "account_ids": [...]?} -> liste complète (traductions fr/en avec description HTML entière, lieux géocodés avec ville/région/lat-lng, dates de publication et d'expiration, mode de travail) ; - noms : POST https://user-accounts-prod.workland.com/api/v1/accounts/ default-profiles body {"account_ids": [...]} -> nom et slug d'employeur par compte. Sans filtre `account_ids`, l'API renvoie tout le réseau Atlas (~1 600 offres, dont ~1 240 au Québec : CSS scolaires, cégeps, CIUSSS/CISSS, STQ, UQTR…) — c'est ce que fait le connecteur réseau `atlas_workland`. Une sous-classe peut cibler un employeur précis via ACCOUNT_IDS ou SLUG. """ from __future__ import annotations import re from ..schema import JobPosting, clean_html from .base import BaseConnector _JOBS_API = "https://jobs-prod.workland.com/api/v1/job/published-external" _PROFILES_API = ("https://user-accounts-prod.workland.com/api/v1/accounts/" "default-profiles") _BY_SLUG_API = ("https://user-accounts-prod.workland.com/api/v1/accounts/" "profiles/by-slug/{slug}") _WORK_MODE = {"ON-SITE": "presentiel", "REMOTE": "teletravail", "HYBRID": "hybride"} class WorklandConnector(BaseConnector): """Base Workland — sous-classes : définir source_id et, pour cibler un employeur, SLUG (page carrières atlas) ou ACCOUNT_IDS ; sans filtre, tout le réseau est lu et EMPLOYER vient du profil de chaque compte.""" ats = "workland" request_delay = 1.0 EMPLOYER = "" # sinon : nom du profil Workland du compte SLUG = "" # slug de la page carrières atlas (résolu en compte) ACCOUNT_IDS: list[int] = [] quebec_only = True def _account_ids(self) -> list[int]: if self.ACCOUNT_IDS: return list(self.ACCOUNT_IDS) if self.SLUG: data = self.get(_BY_SLUG_API.format(slug=self.SLUG), headers={"Accept": "application/json"}).json() aid = ((data.get("data") or {}).get("account_id")) return [int(aid)] if aid else [] return [] def _profiles(self, account_ids: list[int]) -> dict: """{account_id: nom d'employeur} via default-profiles (une requête).""" if not account_ids: return {} data = self.post(_PROFILES_API, json={"account_ids": account_ids}, headers={"Accept": "application/json"}).json() out = {} for k, v in (data.get("data") or {}).items(): name = (v or {}).get("name") or "" if name: out[int(k)] = re.sub(r"\s+", " ", name).strip() return out @staticmethod def _qc_locations(job: dict) -> list[dict]: return [l for l in (job.get("locations") or []) if (l.get("province_or_state_code") or "").upper() == "QC"] def fetch(self) -> list[JobPosting]: body: dict = {"load_with": ["locations"]} ids = self._account_ids() if ids: body["account_ids"] = ids jobs = self.post(_JOBS_API, json=body, headers={"Accept": "application/json"}).json() if not isinstance(jobs, list): raise RuntimeError(f"workland: réponse inattendue {str(jobs)[:120]}") need_names = sorted({j["account_id"] for j in jobs if j.get("account_id")}) if not self.EMPLOYER \ else [] names = self._profiles(need_names) out: list[JobPosting] = [] seen: set[str] = set() for j in jobs: jid = str(j.get("id") or "") if not jid or jid in seen: continue locs = self._qc_locations(j) if self.quebec_only and not locs: continue if not locs: locs = j.get("locations") or [] seen.add(jid) tr = j.get("translations") or {} fr = tr.get("fr") or {} en = tr.get("en") or {} title = (fr.get("title") or en.get("title") or j.get("title") or "").strip() desc = fr.get("description") or en.get("description") \ or j.get("description") or "" employer = self.EMPLOYER or names.get(j.get("account_id"), "") if not title or not employer: continue loc = locs[0] if locs else {} job_p = JobPosting( source=self.source_id, external_id=jid, url=(f"https://atlas.workland.com/work/{jid}/" f"{j.get('url') or ''}"), employer=employer, title=title, description=clean_html(desc), city=loc.get("city") or "", postal_code=loc.get("postal_code") or "", location_label=", ".join( sorted({l.get("city") for l in locs if l.get("city")})), date_posted=(j.get("published_externally_at") or "")[:10] or None, date_deadline=(j.get("external_expiry_date") or "")[:10] or None, work_mode=_WORK_MODE.get(j.get("work_location") or ""), ats=self.ats, ) try: if loc.get("latitude"): job_p.lat = float(loc["latitude"]) job_p.lng = float(loc["longitude"]) except (TypeError, ValueError): pass if j.get("is_full_time") is True: job_p.details["employment_label"] = "Temps plein" elif j.get("is_full_time") is False: job_p.details["employment_label"] = "Temps partiel" if j.get("number_of_positions"): job_p.details["positions"] = j["number_of_positions"] out.append(job_p) return out