spb/job-ka

1# =============================================================================2# Job·Ka — Groupe KA3# Auteur  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : jobka/connectors/wpjobmanager.py6# Rôle    : Classe de plateforme WordPress + WP Job Manager — flux job_feed7#           (contenu complet) + JSON-LD des pages poste (lieu, date limite).8#           Utilisée par la plateforme « carrieresante.gouv.qc.ca » du réseau9#           de la santé (un site WordPress par établissement Santé Québec).10# Créé    : 2026-08-19   Modifié : 2026-08-1911# =============================================================================12"""Plateforme WordPress / WP Job Manager.1314- liste  : GET <BASE>/?feed=job_feed&posts_per_page=50015           -> RSS : <item> title / link (/poste/<slug>/) / pubDate /16              guid (?post_type=job_listing&p=<id>) / content:encoded (HTML17              complet de l'offre) — tout en une requête.18- détail : la page /poste/<slug>/ embarque un JSON-LD schema.org/JobPosting19           (validThrough, jobLocation, employmentType) — cache BD + budget,20           uniquement pour compléter lieu et date limite.21"""22from __future__ import annotations2324import html as _html25import os26import re2728from ..schema import JobPosting, clean_html29from . import _jsonld30from .base import BaseConnector3132MAX_DETAILS = int(os.environ.get("JOBKA_WPJM_DETAIL_LIMIT", "60"))3334_ITEM_RE = re.compile(r"<item>(.*?)</item>", re.S | re.I)35_TAG_RE = {t: re.compile(rf"<{t}>(.*?)</{t}>", re.S | re.I)36           for t in ("title", "link", "pubDate", "guid")}37_CONTENT_RE = re.compile(38    r"<content:encoded><!\[CDATA\[(.*?)\]\]></content:encoded>", re.S | re.I)39_PID_RE = re.compile(r"[?&](?:amp;)?p=(\d+)")404142class WPJobManagerConnector(BaseConnector):43    """Base WP Job Manager — sous-classes : définir source_id, EMPLOYER,44    BASE (racine du site WordPress) et DEFAULT_CITY au besoin."""4546    ats = "wpjobmanager"47    request_delay = 1.04849    EMPLOYER = ""50    BASE = ""51    DEFAULT_CITY = ""52    PER_PAGE = 5005354    def _cdata(self, s: str) -> str:55        s = (s or "").strip()56        if s.startswith("<![CDATA["):57            s = s[9:]58            if s.endswith("]]>"):59                s = s[:-3]60        return _html.unescape(s.strip())6162    def _fetch_detail(self, url: str) -> dict:63        html = self.get(url).text64        node = _jsonld.extract_jobposting(html)65        return _jsonld.jobposting_fields(node) if node else {}6667    def fetch(self) -> list[JobPosting]:68        xml = self.get(f"{self.BASE}/",69                       params={"feed": "job_feed",70                               "posts_per_page": str(self.PER_PAGE)}).text71        out: list[JobPosting] = []72        details_used = 073        seen: set[str] = set()74        for raw in _ITEM_RE.findall(xml):75            def g(tag: str, raw=raw) -> str:76                m = _TAG_RE[tag].search(raw)77                return self._cdata(m.group(1)) if m else ""78            link, guid = g("link"), g("guid")79            m_id = _PID_RE.search(_html.unescape(guid))80            eid = m_id.group(1) if m_id else ""81            title = g("title")82            if not link or not title:83                continue84            if not eid:85                m_slug = re.search(r"/poste/([^/]+)/?$", link)86                eid = m_slug.group(1) if m_slug else ""87            if not eid or eid in seen:88                continue89            seen.add(eid)90            m_c = _CONTENT_RE.search(raw)91            job = JobPosting(92                source=self.source_id, external_id=eid, url=link,93                employer=self.EMPLOYER, title=title,94                description=clean_html(_html.unescape(m_c.group(1)))95                if m_c else "",96                city=self.DEFAULT_CITY,97                date_posted=g("pubDate") or None,98                ats=self.ats,99            )100            key = f"{title}|{job.date_posted or ''}"101            if details_used < MAX_DETAILS:102                fresh = [False]103104                def _fn(u=link, fresh=fresh):105                    fresh[0] = True106                    return self._fetch_detail(u)107108                fields = self.detail(eid, key, _fn)109                if fresh[0]:110                    details_used += 1111            else:112                fields = self.stale_detail(eid)113            if fields:114                if fields.get("date_deadline"):115                    job.date_deadline = fields["date_deadline"]116                if fields.get("city"):117                    job.city = fields["city"]118                if fields.get("postal_code"):119                    job.postal_code = fields["postal_code"]120                if fields.get("employment_label"):121                    job.details["employment_label"] = fields["employment_label"]122                if fields.get("lat") is not None:123                    job.lat, job.lng = fields["lat"], fields["lng"]124            out.append(job)125        return out126