# =============================================================================
# Job·Ka — Groupe KA
# Auteur : Simon-Pierre Boucher
# Contact : contact@spboucher.ai
# Fichier : jobka/connectors/njoyn.py
# Rôle : Classe de plateforme Njoyn (CGI) — très répandu au secteur public
# québécois (santé, villes, sociétés d'État). HTML serveur + jeton.
# Créé : 2026-08-18 Modifié : 2026-09-13
# =============================================================================
"""Plateforme Njoyn (CGI).
Les babillards Njoyn servent un HTML rendu côté serveur :
//xweb/xweb.asp?clid=&page=joblisting&lang=
Le serveur redirige en ajoutant un jeton de session (tbtoken) — il suffit de
suivre la redirection. Chaque offre est un (habillage moderne) ou
une ligne de tableau (habillage classique) pointant vers
``page=jobdetails&jobid=J####-####``.
⚠ Les domaines *.njoyn.com sont derrière Radware Bot Manager : pour ces
tenants, mettre ``USE_SCRAPFLY = True`` (contournement ASP). Depuis la
maintenance plateforme du 2026-09-05→07, Radware couvre aussi des domaines
personnalisés (ex. emplois.santemontreal.qc.ca depuis le 2026-09-08) ; le
fallback de ``_html`` bascule alors tout le sync sur Scrapfly.
"""
from __future__ import annotations
import hashlib
import html as _html
import os
import re
import time
import requests
from ..schema import JobPosting, clean_html
from .base import BaseConnector
MAX_DETAILS = int(os.environ.get("JOBKA_NJOYN_DETAIL_LIMIT", "40"))
PAGE_SIZE = 500 # le formulaire de recherche accepte Inp_ItemsPerPage
_ARTICLE_RE = re.compile(r"]*>(.*?)", re.S | re.I)
_H4_RE = re.compile(r"]*>(.*?)", re.S | re.I)
_TIME_RE = re.compile(r"]*>(.*?)", re.S | re.I)
_JOBID_RE = re.compile(r"jobid=(J\d{4}-\d+)", re.I)
_FIELD_RE = re.compile(
r"([^<:]+?)(?: )?\s*:?\s*\s*(.*?)",
re.S | re.I)
_APPLY_RE = re.compile(
r"]*class='btn btn-primary'[^>]*href='([^']+)'", re.I)
_ROW_LINK_RE = re.compile(
r''']*href=["'][^"']*jobid=(J\d{4}-\d+)[^"']*["'][^>]*>(.*?)''',
re.S | re.I)
_TR_RE = re.compile(r"]*>(.*?)", re.S | re.I)
_TD_RE = re.compile(r"]*>(.*?)", re.S | re.I)
_TOTAL_RE = re.compile(
r"(?:R[ée]sultats de (?:la )?recherche|Search results)\s*\((\d+)\)", re.I)
# Maintenance planifiée de la plateforme entière (constatée 2026-09-05 ~22h30
# EDT, typiquement vendredi soir → dimanche ~21h) : TOUS les tenants — domaines
# *.njoyn.com ET personnalisés — redirigent vers
# maintenance.njoyn.com/.../platform-maintenance.html (~29 ko).
_MAINT_RE = re.compile(
r"Platform maintenance \| Njoyn|platform-maintenance", re.I)
# État transitoire post-maintenance (constaté 2026-09-07 ~11h-15h EDT, fenêtre
# officielle jusqu'à dimanche ~21h) : TOUS les tenants répondent 200 avec un
# corps minuscule « Invalid request XWP10023-UA » — y compris
# pour un vrai navigateur (vérifié Chromium/Playwright depuis IP résidentielle).
# Ce n'est PAS de l'anti-bot : échec immédiat, sans retry ni 2e crédit Scrapfly.
_INVALID_RE = re.compile(r"Invalid request XWP\d+", re.I)
# Mur Radware Bot Manager (« Radware Block Page », CSS shieldsquare) : depuis
# la maintenance du 2026-09-05→07 il couvre AUSSI des domaines personnalisés
# (constaté 2026-09-08 sur emplois.santemontreal.qc.ca). ⚠ La page challenge
# embarque l'URL Xweb.asp dans ses paramètres : sans détection dédiée elle
# passait le garde-fou de _fetch_listing et donnait un faux « 0 offre, ok ».
_RADWARE_RE = re.compile(r"Radware Block Page|shieldsquare|perfdrive", re.I)
def _txt(s: str) -> str:
return re.sub(r"\s+", " ", _html.unescape(re.sub(r"<[^>]+>", " ", s or ""))).strip()
class NjoynConnector(BaseConnector):
"""Base Njoyn — sous-classes : définir source_id, EMPLOYER, BASE, CL, CLID."""
ats = "njoyn"
request_delay = 1.2
EMPLOYER = ""
BASE = "" # ex. https://emplois.santemontreal.qc.ca
CL = "CL3" # instance (CL2, CL3, CL4…)
CLID = "" # identifiant client Njoyn
LANG = "2" # 2 = français, 1 = anglais
USE_SCRAPFLY = False # requis pour les domaines *.njoyn.com (Radware)
max_pages = 6 # pagination du formulaire (PAGE_SIZE offres/page)
max_details = MAX_DETAILS # abaisser pour les babillards via Scrapfly
quebec_only = False # filtrer (table classique : colonnes ville/pays)
def _url(self, page: str, jobid: str = "") -> str:
# ⚠ Casse du chemin : « Xweb.asp » avec X majuscule OBLIGATOIRE.
# Depuis le 2026-09-02, le couple Njoyn/Radware répond une page
# malformée au chemin minuscule « xweb.asp » (Scrapfly ASP échoue en
# ERR::SCRAPE::UPSTREAM_WEBSITE_ERROR, 0 octet) — tous les tenants
# *.njoyn.com étaient tombés d'un coup pour cette seule raison.
u = (f"{self.BASE}/{self.CL}/xweb/Xweb.asp?clid={self.CLID}"
f"&page={page}&lang={self.LANG}")
if jobid:
u += f"&jobid={jobid}"
return u
def _sf_session(self) -> str:
"""Nom de session Scrapfly du sync courant : cookies + IP (proxy
collant) partagés entre le GET liste, les POST de pagination et les
pages détail. Sans session, le tbtoken du formulaire (lié aux cookies
ASP de la session du GET) est rejeté en 400 Bad Request au POST."""
if not getattr(self, "_sf_session_name", None):
self._sf_session_name = f"njoyn-{self.source_id}-{int(time.time())}"
return self._sf_session_name
def _html(self, url: str) -> str:
if self.USE_SCRAPFLY:
return self.get_scrapfly(url, render_js=False,
session=self._sf_session())
text = self.get(url).text
if _RADWARE_RE.search(text[:4000]): # mur anti-bot : bascule Scrapfly
# … pour TOUTE la suite du sync (pagination POST et pages détail
# passent par le même mur), pas seulement cette requête.
self.USE_SCRAPFLY = True
return self.get_scrapfly(url, render_js=False,
session=self._sf_session())
return text
def _fetch_detail(self, jobid: str) -> dict:
page = self._html(self._url("jobdetails", jobid))
page = re.sub(r"<(noscript|nav|script|style)[^>]*>.*?", " ", page,
flags=re.S | re.I)
# habillage moderne : sections Titre
corps
sections = re.findall(
r'\s*]*>(.*?)(.*?)
', page,
re.S | re.I)
if sections:
body = "\n\n".join(f"{clean_html(h)}\n{clean_html(b)}"
for h, b in sections)
return {"description": body[:20000]}
# habillage classique : contenu principal après le
m = re.search(r"]*>.*?
(.*?)(?: list[dict]:
items = []
for art in _ARTICLE_RE.findall(page):
m = _JOBID_RE.search(art)
if not m:
continue
fields = {_txt(k).lower(): _txt(v) for k, v in _FIELD_RE.findall(art)}
apply_m = _APPLY_RE.search(art)
h4 = _H4_RE.search(art)
t = _TIME_RE.search(art)
items.append({
"jobid": m.group(1),
"title": _txt(h4.group(1)) if h4 else "",
"date": _txt(t.group(1)).lstrip("Le ").strip() if t else "",
"fields": fields,
"apply_url": apply_m.group(1) if apply_m else "",
})
return items
def _parse_rows(self, page: str) -> list[dict]:
"""Habillage classique : simples liens de tableau vers jobdetails."""
items, seen = [], set()
for jobid, label in _ROW_LINK_RE.findall(page):
title = _txt(label)
if not title or title.lower().startswith(("en savoir", "postuler",
"apply", "more")):
title = ""
if jobid in seen:
# garder le premier libellé non vide
if title:
for it in items:
if it["jobid"] == jobid and not it["title"]:
it["title"] = title
continue
seen.add(jobid)
items.append({"jobid": jobid, "title": title, "date": "",
"fields": {}, "apply_url": ""})
return [it for it in items if it["title"]]
def _parse_table(self, page: str) -> list[dict]:
"""Habillage tableau multi-colonnes : N° | Titre | Catégorie | Ville |
Pays (ex. CGI). Signature stricte — le libellé du lien est le numéro
de poste lui-même et la ligne a ≥ 4 colonnes — pour ne JAMAIS
intercepter les habillages « titre dans le lien » (STL, RTC…)."""
items, seen = [], set()
for tr in _TR_RE.findall(page):
m = _ROW_LINK_RE.search(tr)
if not m or m.group(1) in seen:
continue
if _txt(m.group(2)) != m.group(1):
continue # le lien porte un titre : autre habillage
tds = [_txt(td) for td in _TD_RE.findall(tr)]
if len(tds) < 4:
continue
seen.add(m.group(1))
title = tds[1] if len(tds) > 1 else ""
items.append({
"jobid": m.group(1), "title": title, "date": "",
"fields": {}, "apply_url": "",
"category": tds[2] if len(tds) > 2 else "",
"city": tds[3] if len(tds) > 3 else "",
"country": tds[4] if len(tds) > 4 else "",
})
return [it for it in items if it["title"] and it["title"] != it["jobid"]]
def _post_form(self, action: str, data: dict) -> str:
if self.USE_SCRAPFLY:
from urllib.parse import urlencode
return self.scrapfly(
action, render_js=False, method="POST",
body=urlencode(data),
headers={"Content-Type": "application/x-www-form-urlencoded"},
session=self._sf_session(),
).get("content") or ""
return self.post(action, data=data).text
def _paginate(self, page: str, items: list[dict]) -> list[dict]:
"""Le rendu direct ne liste que ~50 offres : rejouer le formulaire de
recherche (POST, mêmes cookies de session) avec Inp_ItemsPerPage/pn
pour lire tout le babillard (constaté : Santé Montréal 762 offres).
En cas d'échec (habillage différent), la première page est conservée.
Le total affiché (« Résultats de recherche (N) ») évite de rejouer le
formulaire quand la première page contient déjà tout le babillard.
"""
m_tot = _TOTAL_RE.search(page)
total = int(m_tot.group(1)) if m_tot else None
if total is not None and total <= len(items):
return items # tout est déjà sur la page 1
if total is None and len(items) < 50:
return items # page visiblement complète
m = re.search(r']*action="([^"]+)"', page, re.I)
if not m:
return items
action = _html.unescape(m.group(1))
# Même gotcha de casse que _url : le formulaire annonce « xweb.asp »
# minuscule, que le couple Njoyn/Radware rejette depuis le 2026-09-02.
action = re.sub(r"xweb\.asp", "Xweb.asp", action)
if not action.startswith("http"):
action = f"{self.BASE}/{self.CL}/xweb/{action}"
m_sort = re.search(
r']*name="s"[^>]*value="([^"]*)"', page, re.I) or \
re.search(r']*value="([^"]*)"[^>]*name="s"', page, re.I)
form = {"s": m_sort.group(1) if m_sort else " PostDateFrom2 DESC ",
"hiddencategoryid": "", "Inp_Keywords": "",
"Inp_XWEB_JobListing_JobID": "", "inp_City": "",
"joblistingsearchbutton": "Recherche",
"Inp_ItemsPerPage": str(PAGE_SIZE)}
merged: list[dict] = []
seen: set[str] = set()
for pn in range(1, self.max_pages + 1):
try:
html_pn = self._post_form(action, {**form, "pn": str(pn)})
except Exception:
break
batch = (self._parse_articles(html_pn)
or self._parse_table(html_pn)
or self._parse_rows(html_pn))
new = [it for it in batch if it["jobid"] not in seen]
if not new:
break
for it in new:
seen.add(it["jobid"])
merged.append(it)
if len(batch) < PAGE_SIZE:
break
if total is not None and len(merged) >= total:
break
if not merged: # formulaire non joué : page 1
return items
for it in items: # filet : rien perdre de la p.1
if it["jobid"] not in seen:
merged.append(it)
return merged
def _keep(self, it: dict) -> bool:
"""Filtre Québec pour les babillards pancanadiens/mondiaux (colonnes
ville/pays de l'habillage tableau)."""
if not self.quebec_only:
return True
from ..schema import is_quebec_location
country = (it.get("country") or "").strip().lower()
if country and country not in ("canada", "ca"):
# certains habillages (ex. Groupe Master) mettent la DATE dans
# cette colonne : n'exclure que si c'est clairement un autre pays
if country in ("united states", "usa", "us", "états-unis",
"etats-unis", "france", "mexico", "mexique",
"united kingdom", "uk", "germany", "allemagne",
"india", "inde", "australia", "australie"):
return False
place = (it.get("city") or
it.get("fields", {}).get("villes et arrondissements") or "")
return bool(place) and is_quebec_location(place.replace("-", " "))
def _fetch_listing(self) -> tuple[str, list[dict]]:
"""Page liste + items parsés, avec garde-fou : une réponse vide ou
anti-bot (Scrapfly en échec renvoie "" — voir base.get_scrapfly) ne
doit JAMAIS passer pour un babillard vide. Toute page Njoyn légitime,
même sans offre, contient le formulaire xweb.asp — sinon on retente
(3 essais, backoff 10-20 s) puis on lève une erreur (sync en échec,
pas « 0 trouvé »). ⚠ Entre deux essais, ré-amorcer la session
Scrapfly : la session colle cookies + IP de proxy, donc retenter sur
la même session rejoue le même proxy défaillant (constaté 2026-09-09
12h26 : bellemare et groupe_master tombés ensemble, 0 octet aux deux
essais de la même session, repartis au sync suivant)."""
for attempt in range(3):
if attempt:
time.sleep(10 * attempt)
self._sf_session_name = None # nouvelle session = nouveau proxy
try:
page = self._html(self._url("joblisting"))
except requests.RequestException:
# Panne réseau transitoire vers api.scrapfly.io ou le site
# (constaté 2026-09-13 04:28 : ConnectTimeout 180 s vers l'API
# Scrapfly sur groupe_bellemare, seul échec de tout le cycle) :
# même traitement qu'une réponse vide — nouvelle session,
# backoff, retry — au lieu de laisser l'exception avorter le
# sync au premier essai.
if attempt == 2:
raise
continue
if _MAINT_RE.search(page):
# Maintenance plateforme : inutile de retenter dans 10 s (la
# fenêtre dure des heures) — échec immédiat et explicite, les
# offres en base restent actives.
raise RuntimeError(
f"njoyn {self.source_id}: plateforme Njoyn en maintenance "
f"planifiée (redirection maintenance.njoyn.com) — sync "
f"avorté, offres conservées")
if _INVALID_RE.search(page[:200]):
# Plateforme en état transitoire (« Invalid request XWP… ») :
# touche même un vrai navigateur — inutile de retenter.
raise RuntimeError(
f"njoyn {self.source_id}: plateforme Njoyn indisponible "
f"(réponse « Invalid request XWP10023 », état transitoire "
f"post-maintenance) — sync avorté, offres conservées")
items = (self._parse_articles(page) or self._parse_table(page)
or self._parse_rows(page))
if items or re.search(r"xweb\.asp", page, re.I):
return page, items
raise RuntimeError(
f"njoyn {self.source_id}: page liste illisible après 3 essais "
f"(vide ou anti-bot, {len(page)} octets) — sync avorté")
def fetch(self) -> list[JobPosting]:
page, items = self._fetch_listing()
items = self._paginate(page, items)
items = [it for it in items if self._keep(it)]
out: list[JobPosting] = []
details_used = 0
for it in items:
f = it["fields"]
salary = f.get("salaire", "")
if re.match(r"^0\s*\$\s*-\s*0\s*\$", salary):
salary = ""
job = JobPosting(
source=self.source_id, external_id=it["jobid"],
url=self._url("jobdetails", it["jobid"]),
employer=f.get("établissement") or f.get("etablissement")
or self.EMPLOYER,
title=it["title"],
city=(f.get("villes et arrondissements", "").split(",")[0]
or (it.get("city") or "").split(",")[0]),
location_label=(f.get("villes et arrondissements", "")
or it.get("city") or ""),
salary_label=salary,
date_posted=it["date"] or None,
ats=self.ats,
)
if f.get("statut de l'employé") or f.get("statut de l'employe"):
job.details["employment_label"] = (
f.get("statut de l'employé") or f.get("statut de l'employe"))
if it.get("category"):
job.details["team"] = it["category"]
if f.get("niveau de scolarité"):
job.requirements["scolarite"] = f["niveau de scolarité"]
if it["apply_url"]:
job.apply_url = it["apply_url"]
job.details["apply_url"] = it["apply_url"]
key = hashlib.sha1(
f"{it['title']}|{it['date']}".encode()).hexdigest()[:12]
if details_used < self.max_details:
fresh = [False]
def _fn(j=it["jobid"], fresh=fresh):
fresh[0] = True
return self._fetch_detail(j)
d = self.detail(it["jobid"], key, _fn)
if fresh[0]:
details_used += 1
job.description = d.get("description", "")
out.append(job)
return out