SPB Git forge

spb/job-ka

Public
229commits 1branches 0releases
38.1 MBsize
maindefault branch
6 h agolast push
HTML 82.1% Python 14.6% TypeScript 1.9% CSS 1% JavaScript 0.5%
6.8 KB · 159 lines python
Raw Blame History
1# =============================================================================2# Job·Ka — Groupe KA3# Auteur  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : jobka/connectors/workland.py6# Rôle    : Classe de plateforme Workland / Atlas (ATS québécois — CSS7#           scolaires, cégeps, santé, sociétés d'État) — API JSON publique8#           jobs-prod.workland.com (celle du frontal atlas.workland.com)9# Créé    : 2026-08-18   Modifié : 2026-08-1910# =============================================================================11"""Plateforme Workland (atlas.workland.com).1213Réécriture 2026-08-19 : l'ancienne version grattait les pages carrières des14employeurs (liens atlas rendus serveur) et n'avait AUCUNE description (la SPA15Atlas n'expose rien côté serveur). L'API publique du frontal fait tout en16deux requêtes :1718- offres  : POST https://jobs-prod.workland.com/api/v1/job/published-external19            body {"load_with": ["locations"], "account_ids": [...]?}20            -> liste complète (traductions fr/en avec description HTML21            entière, lieux géocodés avec ville/région/lat-lng, dates de22            publication et d'expiration, mode de travail) ;23- noms    : POST https://user-accounts-prod.workland.com/api/v1/accounts/24            default-profiles body {"account_ids": [...]}25            -> nom et slug d'employeur par compte.2627Sans filtre `account_ids`, l'API renvoie tout le réseau Atlas (~1 600 offres,28dont ~1 240 au Québec : CSS scolaires, cégeps, CIUSSS/CISSS, STQ, UQTR…) —29c'est ce que fait le connecteur réseau `atlas_workland`. Une sous-classe30peut cibler un employeur précis via ACCOUNT_IDS ou SLUG.31"""32from __future__ import annotations3334import re3536from ..schema import JobPosting, clean_html37from .base import BaseConnector3839_JOBS_API = "https://jobs-prod.workland.com/api/v1/job/published-external"40_PROFILES_API = ("https://user-accounts-prod.workland.com/api/v1/accounts/"41                 "default-profiles")42_BY_SLUG_API = ("https://user-accounts-prod.workland.com/api/v1/accounts/"43                "profiles/by-slug/{slug}")4445_WORK_MODE = {"ON-SITE": "presentiel", "REMOTE": "teletravail",46              "HYBRID": "hybride"}474849class WorklandConnector(BaseConnector):50    """Base Workland — sous-classes : définir source_id et, pour cibler un51    employeur, SLUG (page carrières atlas) ou ACCOUNT_IDS ; sans filtre, tout52    le réseau est lu et EMPLOYER vient du profil de chaque compte."""5354    ats = "workland"55    request_delay = 1.05657    EMPLOYER = ""          # sinon : nom du profil Workland du compte58    SLUG = ""              # slug de la page carrières atlas (résolu en compte)59    ACCOUNT_IDS: list[int] = []60    quebec_only = True6162    def _account_ids(self) -> list[int]:63        if self.ACCOUNT_IDS:64            return list(self.ACCOUNT_IDS)65        if self.SLUG:66            data = self.get(_BY_SLUG_API.format(slug=self.SLUG),67                            headers={"Accept": "application/json"}).json()68            aid = ((data.get("data") or {}).get("account_id"))69            return [int(aid)] if aid else []70        return []7172    def _profiles(self, account_ids: list[int]) -> dict:73        """{account_id: nom d'employeur} via default-profiles (une requête)."""74        if not account_ids:75            return {}76        data = self.post(_PROFILES_API, json={"account_ids": account_ids},77                         headers={"Accept": "application/json"}).json()78        out = {}79        for k, v in (data.get("data") or {}).items():80            name = (v or {}).get("name") or ""81            if name:82                out[int(k)] = re.sub(r"\s+", " ", name).strip()83        return out8485    @staticmethod86    def _qc_locations(job: dict) -> list[dict]:87        return [l for l in (job.get("locations") or [])88                if (l.get("province_or_state_code") or "").upper() == "QC"]8990    def fetch(self) -> list[JobPosting]:91        body: dict = {"load_with": ["locations"]}92        ids = self._account_ids()93        if ids:94            body["account_ids"] = ids95        jobs = self.post(_JOBS_API, json=body,96                         headers={"Accept": "application/json"}).json()97        if not isinstance(jobs, list):98            raise RuntimeError(f"workland: réponse inattendue {str(jobs)[:120]}")99100        need_names = sorted({j["account_id"] for j in jobs101                             if j.get("account_id")}) if not self.EMPLOYER \102            else []103        names = self._profiles(need_names)104105        out: list[JobPosting] = []106        seen: set[str] = set()107        for j in jobs:108            jid = str(j.get("id") or "")109            if not jid or jid in seen:110                continue111            locs = self._qc_locations(j)112            if self.quebec_only and not locs:113                continue114            if not locs:115                locs = j.get("locations") or []116            seen.add(jid)117            tr = j.get("translations") or {}118            fr = tr.get("fr") or {}119            en = tr.get("en") or {}120            title = (fr.get("title") or en.get("title")121                     or j.get("title") or "").strip()122            desc = fr.get("description") or en.get("description") \123                or j.get("description") or ""124            employer = self.EMPLOYER or names.get(j.get("account_id"), "")125            if not title or not employer:126                continue127            loc = locs[0] if locs else {}128            job_p = JobPosting(129                source=self.source_id, external_id=jid,130                url=(f"https://atlas.workland.com/work/{jid}/"131                     f"{j.get('url') or ''}"),132                employer=employer, title=title,133                description=clean_html(desc),134                city=loc.get("city") or "",135                postal_code=loc.get("postal_code") or "",136                location_label=", ".join(137                    sorted({l.get("city") for l in locs if l.get("city")})),138                date_posted=(j.get("published_externally_at") or "")[:10]139                or None,140                date_deadline=(j.get("external_expiry_date") or "")[:10]141                or None,142                work_mode=_WORK_MODE.get(j.get("work_location") or ""),143                ats=self.ats,144            )145            try:146                if loc.get("latitude"):147                    job_p.lat = float(loc["latitude"])148                    job_p.lng = float(loc["longitude"])149            except (TypeError, ValueError):150                pass151            if j.get("is_full_time") is True:152                job_p.details["employment_label"] = "Temps plein"153            elif j.get("is_full_time") is False:154                job_p.details["employment_label"] = "Temps partiel"155            if j.get("number_of_positions"):156                job_p.details["positions"] = j["number_of_positions"]157            out.append(job_p)158        return out159