SPB Git forge

spb/job-ka

Public
229commits 1branches 0releases
38.1 MBsize
maindefault branch
5 h agolast push
HTML 82.1% Python 14.6% TypeScript 1.9% CSS 1% JavaScript 0.5%
6.7 KB · 164 lines python
Raw Blame History
1# =============================================================================2# Job·Ka — Groupe KA3# Auteur  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : jobka/connectors/ultipro.py6# Rôle    : Classe de plateforme UKG Pro Recruiting (UltiPro) — API JSON7#           publique LoadSearchResults, un employeur = une sous-classe8# Créé    : 2026-08-18   Modifié : 2026-08-189# =============================================================================10"""Plateforme UKG Pro Recruiting (recruiting.ultipro.com).1112API JSON publique (celle du frontal) :13- liste  : POST https://recruiting.ultipro.com/<ORG>/JobBoard/<BOARD>/14                JobBoardView/LoadSearchResults15           body {opportunitySearch:{Top, Skip, OrderBy…}}16           -> {opportunities:[{Id, Title, RequisitionNumber, BriefDescription,17               FullTime, JobCategoryName, PostedDate, Locations:[{Address}]}],18               totalCount}19- détail : POST .../JobBoardView/LoadOpportunity {opportunityId} -> Description20           complète (HTML). Visité avec cache BD + budget.21"""22from __future__ import annotations2324import os25import re2627from ..schema import JobPosting, clean_html, is_quebec_location28from .base import BaseConnector2930PAGE_SIZE = 5031MAX_DETAILS = int(os.environ.get("JOBKA_ULTIPRO_DETAIL_LIMIT", "80"))323334class UltiProConnector(BaseConnector):35    """Base UltiPro — sous-classes : définir source_id, EMPLOYER, ORG, BOARD."""3637    ats = "ultipro"38    request_delay = 0.83940    EMPLOYER = ""41    ORG = ""      # ex. MES1005MESR42    BOARD = ""    # GUID du JobBoard43    HOST = "recruiting.ultipro.com"   # tenants canadiens : recruiting.ultipro.ca44    quebec_only = True45    max_pages = 204647    @property48    def _board_url(self) -> str:49        return f"https://{self.HOST}/{self.ORG}/JobBoard/{self.BOARD}"5051    @staticmethod52    def _loc_fields(loc: dict) -> tuple[str, str, str, str]:53        """(ville, province, code postal, libellé) d'un objet Location."""54        addr = loc.get("Address") or {}55        city = addr.get("City") or ""56        state = ((addr.get("State") or {}).get("Code")57                 if isinstance(addr.get("State"), dict)58                 else addr.get("State")) or ""59        postal = addr.get("PostalCode") or ""60        label = loc.get("LocalizedDescription") or city61        return city, str(state), postal, label6263    def _keep(self, locations: list[dict]) -> bool:64        if not self.quebec_only:65            return True66        for loc in locations:67            city, state, postal, label = self._loc_fields(loc)68            if state.upper() == "QC" or re.match(r"^[GHJ]\d[A-Z]", postal.upper()):69                return True70            if is_quebec_location(f"{label} {city}"):71                return True72        return False7374    _DESC_RE = re.compile(r'"Description"\s*:\s*"((?:[^"\\]|\\.)*)"')7576    def _fetch_detail(self, opp_id: str) -> dict:77        """La page OpportunityDetail embarque l'objet JSON de l'offre —78        on en extrait la description complète (chaîne JSON échappée)."""79        html = self.get(f"{self._board_url}/OpportunityDetail",80                        params={"opportunityId": opp_id}).text81        best = ""82        for m in self._DESC_RE.finditer(html):83            try:84                import json as _json85                val = _json.loads(f'"{m.group(1)}"')86            except ValueError:87                continue88            if len(val) > len(best):89                best = val90        return {"description": clean_html(best)}9192    def fetch(self) -> list[JobPosting]:93        url = f"{self._board_url}/JobBoardView/LoadSearchResults"94        out: list[JobPosting] = []95        details_used = 096        skip = 097        for _ in range(self.max_pages):98            body = {99                "opportunitySearch": {100                    "Top": PAGE_SIZE, "Skip": skip, "QueryString": "",101                    "OrderBy": [{"Value": "postedDateDesc",102                                 "PropertyName": "PostedDate",103                                 "Ascending": False}],104                    "Filters": [],105                },106                "matchCriteria": {"PreferredJobs": [], "Educations": [],107                                  "LicenseAndCertifications": [], "Skills": [],108                                  "hasNoLicenses": False, "SkippedSkills": []},109            }110            data = self.post(url, json=body,111                             headers={"Accept": "application/json"}).json()112            opps = data.get("opportunities") or []113            if not opps:114                break115            for o in opps:116                locations = o.get("Locations") or []117                if not self._keep(locations):118                    continue119                oid = str(o.get("Id") or "")120                if not oid:121                    continue122                city = label = postal = ""123                for loc in locations:124                    c, s, p, lbl = self._loc_fields(loc)125                    if s.upper() == "QC" or is_quebec_location(f"{lbl} {c}"):126                        city, postal, label = c or lbl, p, lbl127                        break128                job = JobPosting(129                    source=self.source_id, external_id=oid,130                    url=f"{self._board_url}/OpportunityDetail?opportunityId={oid}",131                    employer=self.EMPLOYER,132                    title=o.get("Title") or "",133                    city=city, postal_code=postal, location_label=label,134                    date_posted=o.get("PostedDate") or None,135                    ats=self.ats,136                )137                job.details["requisition"] = o.get("RequisitionNumber") or ""138                if o.get("JobCategoryName"):139                    job.details["team"] = o["JobCategoryName"]140                if o.get("FullTime") is True:141                    job.details["employment_label"] = "Temps plein"142                elif o.get("FullTime") is False:143                    job.details["employment_label"] = "Temps partiel"144                key = (o.get("PostedDate") or "") + (o.get("Title") or "")[:40]145                if details_used < MAX_DETAILS:146                    fresh = [False]147148                    def _fn(i=oid, fresh=fresh):149                        fresh[0] = True150                        return self._fetch_detail(i)151152                    d = self.detail(oid, key, _fn)153                    if fresh[0]:154                        details_used += 1155                    job.description = d.get("description", "")156                if not job.description:157                    job.description = clean_html(o.get("BriefDescription") or "")158                out.append(job)159            skip += PAGE_SIZE160            total = int(data.get("totalCount") or 0)161            if total and skip >= total:162                break163        return out164