spb/job-ka

1# =============================================================================2# Job·Ka — Groupe KA3# Auteur  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : jobka/connectors/ashby.py6# Rôle    : Classe de plateforme Ashby (api.ashbyhq.com/posting-api) — un7#           employeur = une sous-classe de ~8 lignes (ORG, EMPLOYER)8# Créé    : 2026-08-17   Modifié : 2026-08-179# =============================================================================10"""Plateforme Ashby.1112API publique : GET https://api.ashbyhq.com/posting-api/job-board/<org>13                   ?includeCompensation=true14-> {jobs:[{id, title, location, secondaryLocations, department, team,15   employmentType(FullTime|PartTime|Intern|Contract|Temporary), isRemote,16   workplaceType, publishedAt, jobUrl, applyUrl, descriptionHtml,17   descriptionPlain, compensation:{compensationTierSummary,18   scrapeableCompensationSalarySummary, ...}}]}.1920Une seule requête liste TOUT (descriptions + salaires inclus) — l'ATS le plus21transparent sur les salaires.22"""23from __future__ import annotations2425from ..schema import JobPosting, is_quebec_location26from .base import BaseConnector2728_EMPLOYMENT = {"fulltime": "temps_plein", "parttime": "temps_partiel",29               "intern": "stage", "contract": "contractuel",30               "temporary": "contractuel"}313233class AshbyConnector(BaseConnector):34    """Base Ashby — sous-classes : définir source_id, EMPLOYER, ORG."""3536    ats = "ashby"37    request_delay = 1.03839    EMPLOYER = ""40    ORG = ""41    quebec_only = True4243    @staticmethod44    def _locations(item: dict) -> list[str]:45        locs = [item.get("location") or ""]46        locs += [s.get("location") or "" for s in item.get("secondaryLocations") or []]47        addr = ((item.get("address") or {}).get("postalAddress") or {})48        if addr.get("addressLocality"):49            locs.append(f"{addr['addressLocality']}, {addr.get('addressRegion') or ''}")50        return [l for l in locs if l]5152    def _keep(self, item: dict) -> bool:53        if not self.quebec_only:54            return True55        return any(is_quebec_location(l) for l in self._locations(item))5657    def fetch(self) -> list[JobPosting]:58        data = self.get(59            f"https://api.ashbyhq.com/posting-api/job-board/{self.ORG}",60            params={"includeCompensation": "true"}).json()61        out: list[JobPosting] = []62        for item in data.get("jobs") or []:63            if not item.get("isListed", True) or not self._keep(item):64                continue65            comp = item.get("compensation") or {}66            salary_label = (comp.get("scrapeableCompensationSalarySummary")67                            or comp.get("compensationTierSummary") or "")68            # salaires structurés (summaryComponents) : plus fiables que le libellé69            s_min = s_max = s_unit = None70            extra_comp: list[dict] = []71            for c in comp.get("summaryComponents") or []:72                ctype = c.get("compensationType") or ""73                if ctype == "Salary" and s_min is None:74                    s_min, s_max = c.get("minValue"), c.get("maxValue")75                    interval = (c.get("interval") or "").upper()76                    s_unit = ("hour" if "HOUR" in interval77                              else "month" if "MONTH" in interval78                              else "year" if "YEAR" in interval else None)79                elif ctype and ctype != "Salary":80                    # composantes Bonus / Equity / Commission (montants si publiés)81                    entry = {"type": ctype}82                    if c.get("minValue") is not None:83                        entry["min"] = c["minValue"]84                    if c.get("maxValue") is not None:85                        entry["max"] = c["maxValue"]86                    if c.get("interval") and (c.get("interval") or "").upper() != "NONE":87                        entry["interval"] = c["interval"]88                    extra_comp.append(entry)89            workplace = (item.get("workplaceType") or "").lower()90            work_mode = ("hybride" if workplace == "hybrid"91                         else "teletravail" if item.get("isRemote")92                         else "presentiel" if workplace == "onsite" else None)93            job = JobPosting(94                source=self.source_id, external_id=str(item.get("id") or ""),95                url=item.get("jobUrl") or "",96                employer=self.EMPLOYER,97                title=item.get("title") or "",98                description=item.get("descriptionPlain") or "",99                location_label=self._locations(item)[0] if self._locations(item) else "",100                work_mode=work_mode,101                employment_type=_EMPLOYMENT.get(102                    (item.get("employmentType") or "").lower()),103                date_posted=item.get("publishedAt"),104                salary_min=s_min, salary_max=s_max,105                salary_unit=s_unit if s_min is not None else None,106                salary_label=salary_label,107                apply_url=item.get("applyUrl") or "",108                ats=self.ats,109            )110            if item.get("team"):111                job.details["team"] = item["team"]112            if extra_comp:113                job.details["compensation_components"] = extra_comp114            out.append(job)115        return out116