HTML 82.1%
Python 14.6%
TypeScript 1.9%
CSS 1%
JavaScript 0.5%
1# =============================================================================2# Job·Ka — Groupe KA3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : jobka/connectors/espresso_jobs.py6# Rôle : Connecteur portail — Espresso-Jobs (emplois TI / créatif / web au7# Québec). AGRÉGATEUR : moins autoritaire qu'une page carrière.8# Créé : 2026-08-18 Modifié : 2026-08-189# =============================================================================10"""Portail spécialisé Espresso-Jobs (espresso-jobs.com — TI et créatif QC).1112- liste : pages /emploi?page_no=N — chaque page embarque un JSON-LD13 schema.org/ItemList (URL + titre de ~21 offres) ;14- détail : chaque page offre embarque un JSON-LD schema.org/JobPosting15 (employeur, description, lieu, salaire) — cache BD + budget.16"""17from __future__ import annotations1819import json20import os21import re2223from ..schema import JobPosting, is_quebec_location24from .base import BaseConnector25from . import _jsonld2627BASE = "https://www.espresso-jobs.com"28MAX_PAGES = int(os.environ.get("JOBKA_ESPRESSO_PAGES", "15"))29MAX_DETAILS = int(os.environ.get("JOBKA_ESPRESSO_DETAIL_LIMIT", "100"))3031_SCRIPT_RE = re.compile(32 r'<script type="application/ld\+json">(.*?)</script>', re.S | re.I)33_ID_RE = re.compile(r"/emploi/(\d+)/")343536class EspressoJobsConnector(BaseConnector):37 """Portail spécialisé TI/créatif — offres majoritairement québécoises."""3839 source_id = "espresso_jobs"40 ats = "portail"41 request_delay = 1.04243 def _list_page(self, page_no: int) -> list[tuple[str, str, str]]:44 """[(id, url, titre)] extraits du JSON-LD ItemList de la page."""45 html = self.get(f"{BASE}/emploi",46 params={"page_no": str(page_no)}).text47 for m in _SCRIPT_RE.finditer(html):48 try:49 data = json.loads(m.group(1))50 except ValueError:51 continue52 if isinstance(data, dict) and data.get("@type") == "ItemList":53 out = []54 for el in data.get("itemListElement") or []:55 item = el.get("item") or {}56 url = item.get("url") or item.get("@id") or ""57 m_id = _ID_RE.search(url)58 if m_id:59 out.append((m_id.group(1), url,60 item.get("name") or ""))61 return out62 return []6364 def _fetch_detail(self, url: str) -> dict:65 html = self.get(url).text66 node = _jsonld.extract_jobposting(html)67 return _jsonld.jobposting_fields(node) if node else {}6869 def fetch(self) -> list[JobPosting]:70 seen: dict[str, tuple[str, str]] = {}71 for page in range(1, MAX_PAGES + 1):72 items = self._list_page(page)73 if not items:74 break75 new = 076 for eid, url, name in items:77 if eid not in seen:78 seen[eid] = (url, name)79 new += 180 if new == 0:81 break8283 out: list[JobPosting] = []84 details_used = 085 for eid, (url, name) in seen.items():86 fields: dict = {}87 if details_used < MAX_DETAILS:88 fresh = [False]8990 def _fn(u=url, fresh=fresh):91 fresh[0] = True92 return self._fetch_detail(u)9394 fields = self.detail(eid, eid, _fn)95 if fresh[0]:96 details_used += 197 else:98 from .. import db99 if self._detail_con is None:100 self._detail_con = db.connect()101 fields = db.get_cached_detail(self._detail_con, self.source_id,102 eid, eid) or {}103 job = JobPosting(source=self.source_id, external_id=eid, url=url,104 title=name, ats=self.ats)105 if fields:106 _jsonld.apply_fields(job, fields, override_employer=True)107 # filtre Québec indulgent : le portail est essentiellement QC —108 # on n'écarte que les lieux explicitement hors province109 region = (fields.get("region_code") or "").upper()110 if region and region not in ("QC", "QUÉBEC", "QUEBEC"):111 continue112 if not region and job.city and not is_quebec_location(job.city) \113 and "télétravail" not in job.city.lower():114 continue115 if job.title and job.employer:116 out.append(job)117 return out118