# ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : jobka/connectors/zoho_recruit.py # Rôle : Classe de plateforme Zoho Recruit (.zohorecruit.com) — flux # RSS public du site carrières (contenu complet en une requête) # Créé : 2026-08-25 Modifié : 2026-09-10 # ============================================================================= """Plateforme Zoho Recruit (sites carrières .zohorecruit.com). - flux : GET https://.zohorecruit.com/jobs//rss -> items avec titre, lien (id numérique), description HTML complète précédée de « Catégorie: X » et « Lieu: Ville Région Pays ». - repli : le RSS ne reflète que les offres « publiées aux job boards » par le tenant — il peut être VIDE alors que le site carrières affiche des dizaines d'offres (cas gus, 2026-09-10 : RSS 0 item, 51 offres QC publiées). La page /jobs/ est une SPA, mais elle embarque le listing complet dans un (JSON : Posting_Title, City/State/Country, Job_Description, Job_Type, Date_Opened, Publish…) : quand le RSS n'a aucun , on parse ce JSON embarqué. """ from __future__ import annotations import html as _html import json as _json import re from ..schema import JobPosting, clean_html, is_quebec_location from .base import BaseConnector _ITEM_RE = re.compile(r"(.*?)", re.S | re.I) _TAG_RE = {t: re.compile(rf"<{t}>(.*?)", re.S | re.I) for t in ("title", "link", "description", "pubDate")} _ID_RE = re.compile(r"/jobs/[^/]+/(\d+)/") _LIEU_RE = re.compile(r"Lieu\s*:\s*([^<]+)", re.I) _CAT_RE = re.compile(r"Cat[ée]gorie\s*:\s*([^<]+)", re.I) _QC_RE = re.compile(r"\b(qu[ée]bec|qc)\b", re.I) _INPUT_RE = re.compile( r'', re.S) # suffixes de quartier des villes Zoho (« Gatineau Northeast », # « Cap-de-la-Madeleine Central and southeast ») _DISTRICT_WORDS = {"north", "south", "east", "west", "northeast", "northwest", "southeast", "southwest", "central", "and"} class ZohoRecruitConnector(BaseConnector): """Base Zoho Recruit — sous-classes : définir source_id, EMPLOYER, ORG, PAGE (nom du site carrières dans l'URL, ex. « Careers »).""" ats = "zoho_recruit" request_delay = 1.0 EMPLOYER = "" ORG = "" # .zohorecruit.com PAGE = "Careers" # /jobs//rss quebec_only = True def _cdata(self, s: str) -> str: s = (s or "").strip() if s.startswith(""): s = s[:-3] return s.strip() def _is_qc(self, lieu: str) -> bool: if _QC_RE.search(lieu): return True # « Lieu: Ville Région Pays » sans virgule : tester la 1re moitié return bool(lieu.strip()) and is_quebec_location(lieu) def _clean_city(self, city: str) -> str: """« Vaudreuil-Soulanges (Coteau-du-Lac) » -> la localité entre parenthèses ; sinon retirer les suffixes de quartier anglais de Zoho (« Gatineau Northeast » -> « Gatineau »).""" city = (city or "").strip() m = re.search(r"\(([^)]+)\)\s*$", city) if m: return m.group(1).strip() words = city.split() while words and words[-1].lower() in _DISTRICT_WORDS: words.pop() return " ".join(words) or city def _fetch_career_page(self) -> list[JobPosting]: """Repli : listing JSON embarqué () de la page carrières — utilisé quand le RSS ne contient aucun .""" page = self.get( f"https://{self.ORG}.zohorecruit.com/jobs/{self.PAGE}").text rows = [] for m in _INPUT_RE.finditer(page): if m.group(2) == "jobs": rows = _json.loads(_html.unescape(m.group(1))) break out: list[JobPosting] = [] for row in rows: jid = str(row.get("id") or "") title = (row.get("Posting_Title") or row.get("Job_Opening_Name") or "").strip() if not jid or not title or not row.get("Publish", True): continue lieu = " ".join(s for s in (row.get("City"), row.get("State"), row.get("Country")) if s).strip() if self.quebec_only and not self._is_qc(lieu): continue job = JobPosting( source=self.source_id, external_id=jid, url=(f"https://{self.ORG}.zohorecruit.com" f"/jobs/{self.PAGE}/{jid}/"), employer=self.EMPLOYER, title=title, description=clean_html(row.get("Job_Description") or ""), city=self._clean_city(row.get("City") or ""), location_label=lieu, date_posted=row.get("Date_Opened") or None, work_mode="teletravail" if row.get("Remote_Job") else None, ats=self.ats, ) if row.get("Job_Type"): job.details["employment_label"] = str(row["Job_Type"]).strip() if row.get("Industry"): job.details["category_label"] = str(row["Industry"]).strip() out.append(job) return out def fetch(self) -> list[JobPosting]: xml = self.get( f"https://{self.ORG}.zohorecruit.com/jobs/{self.PAGE}/rss").text items = _ITEM_RE.findall(xml) if not items: # RSS vide (le tenant ne publie pas/plus aux job boards) alors que # le site carrières peut afficher des offres : repli page embarquée return self._fetch_career_page() out: list[JobPosting] = [] for raw in items: def g(tag: str, raw=raw) -> str: m = _TAG_RE[tag].search(raw) return self._cdata(m.group(1)) if m else "" link = _html.unescape(g("link")) m_id = _ID_RE.search(link) if not link or not m_id: continue desc = _html.unescape(g("description")) m_lieu = _LIEU_RE.search(desc) lieu = _html.unescape(m_lieu.group(1)).strip() if m_lieu else "" if self.quebec_only and not self._is_qc(lieu): continue m_cat = _CAT_RE.search(desc) # ville = premier segment du lieu (« Quebec City Quebec Canada ») city = re.sub(r"\s+(Quebec|Québec|QC|Canada)\b.*$", "", lieu, flags=re.I).strip() job = JobPosting( source=self.source_id, external_id=m_id.group(1), url=link, employer=self.EMPLOYER, title=_html.unescape(g("title")), description=clean_html(desc), city=city, location_label=lieu, date_posted=g("pubDate") or None, ats=self.ats, ) if m_cat: job.details["category_label"] = \ _html.unescape(m_cat.group(1)).strip() out.append(job) return out