spb/forma-ka Public
Python 65.1%
TypeScript 17.9%
CSS 16.4%
HTML 0.5%
1# -----------------------------------------------------------------------------2# Forma-Ka — Agrégateur de formations (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/ets_formation.py : connecteur ÉTS Formation (perf.etsmtl.ca)5# Formation continue de l'École de technologie supérieure — ~600 formations6# professionnelles (technologie, gestion, construction, RH…).7# Site ASP.NET rendu serveur, très bien balisé schema.org :8# - liste : ItemList JSON-LD (toutes les formations, dédupliquées par sigle)9# - fiche : Course JSON-LD (description, UEC, séances datées avec prix,10# mode Onsite/Online, charge PTxxH, formateur, lieu) + sections HTML11# (objectifs pédagogiques, contenu, clientèle visée) + catégorie du12# fil d'Ariane. Fiches en cache, rafraîchies chaque semaine (clé ISO).13# -----------------------------------------------------------------------------14from __future__ import annotations1516import datetime17import re1819from bs4 import BeautifulSoup2021from ..schema import Formation, clean_text22from .base import BaseConnector, ldjson_objects2324BASE = "https://www.perf.etsmtl.ca"25LIST_URL = f"{BASE}/Formations"2627_WORKLOAD_RE = re.compile(r"PT(\d+(?:\.\d+)?)H", re.I)2829# eventAttendanceMode / courseMode -> mode canonique Forma-Ka30_MODE_MAP = {31 "onsite": "présentiel",32 "online": "en ligne",33 "offlineeventattendancemode": "présentiel",34 "onlineeventattendancemode": "en ligne",35 "mixedeventattendancemode": "hybride",36}373839def _instance_mode(inst: dict) -> str:40 for raw in (str(inst.get("eventAttendanceMode", "")),41 str(inst.get("courseMode", ""))):42 key = raw.rsplit("/", 1)[-1].lower()43 if key in _MODE_MAP:44 return _MODE_MAP[key]45 return ""464748def _flatten(seq) -> list[str]:49 """educationalCredentialAwarded arrive parfois en listes imbriquées."""50 out: list[str] = []51 if isinstance(seq, str):52 return [seq]53 for item in seq or []:54 out.extend(_flatten(item) if isinstance(item, (list, tuple)) else [str(item)])55 return out565758def _section_items(soup: BeautifulSoup, heading_rx: str) -> list[str]:59 """Items de liste (<li>) qui suivent un titre de section donné."""60 h = soup.find(["h2", "h3", "h4"],61 string=re.compile(heading_rx, re.I))62 if h is None:63 return []64 items: list[str] = []65 for sib in h.find_all_next(["ul", "ol", "h2", "h3", "h4"], limit=8):66 if sib.name in ("h2", "h3", "h4"):67 break68 items += [clean_text(li.get_text(" ")) for li in sib.find_all("li")]69 if items:70 break71 return [i for i in items if i]727374class EtsFormationConnector(BaseConnector):75 source_id = "ets_formation"76 request_delay = 0.57778 def fetch(self) -> list[Formation]:79 html = self.fetch_html(LIST_URL)8081 # 1) Liste complète depuis l'ItemList JSON-LD (contient des doublons)82 courses: dict[str, dict] = {}83 for obj in ldjson_objects(html):84 if obj.get("@type") != "ItemList":85 continue86 for li in obj.get("itemListElement", []):87 item = li.get("item") or {}88 code = item.get("courseCode") or ""89 if item.get("@type") == "Course" and code and code not in courses:90 courses[code] = item9192 # 2) Fiche détaillée par formation — cache hebdomadaire93 week = datetime.date.today().strftime("%G-W%V")94 out: list[Formation] = []95 for code, item in courses.items():96 url = item.get("url") or item.get("@id") or ""97 key = f"{week}:{item.get('name', '')}"98 payload = self.detail(code, key, lambda u=url: self._fetch_detail(u))99 f = Formation(100 source=self.source_id,101 external_id=code,102 url=url,103 title=item.get("name", ""),104 training_type="Formation continue",105 language="fr",106 code=code,107 )108 for k, v in (payload or {}).items():109 if hasattr(f, k) and v not in (None, "", []):110 setattr(f, k, v)111 out.append(f)112 return out113114 # -- fiche ----------------------------------------------------------------115 def _fetch_detail(self, url: str) -> dict:116 if not url:117 return {}118 html = self.fetch_html(url)119 soup = BeautifulSoup(html, "html.parser")120 payload: dict = {}121122 course = next((o for o in ldjson_objects(html)123 if o.get("@type") == "Course"), None)124 if course:125 payload["description"] = clean_text(course.get("description", ""))126 creds = _flatten(course.get("educationalCredentialAwarded"))127 payload["credential"] = " · ".join(dict.fromkeys(creds))128129 sessions, cities, modes, instructors = [], [], [], []130 price = None131 hours = None132 for inst in course.get("hasCourseInstance", []) or []:133 if not isinstance(inst, dict):134 continue135 start = str(inst.get("startDate", ""))[:10]136 if re.match(r"20\d{2}-\d{2}-\d{2}", start):137 sessions.append(start)138 mode = _instance_mode(inst)139 if mode:140 modes.append(mode)141 loc = ((inst.get("location") or {}).get("address") or {})142 city = loc.get("addressLocality", "")143 if city:144 cities.append(city)145 offer = inst.get("offers") or {}146 if isinstance(offer, list):147 offer = offer[0] if offer else {}148 if price is None and offer.get("price"):149 try:150 price = float(str(offer["price"]).replace(",", "."))151 except ValueError:152 pass153 m = _WORKLOAD_RE.search(str(inst.get("courseWorkload", "")))154 if m and hours is None:155 hours = float(m.group(1))156 for pers in inst.get("instructor", []) or []:157 if isinstance(pers, dict) and pers.get("name"):158 instructors.append(pers["name"])159160 sessions = sorted(set(sessions))161 if sessions:162 payload["sessions"] = sessions163 payload["start_date"] = sessions[0]164 modes = list(dict.fromkeys(modes))165 if modes:166 payload["mode"] = modes[0] if len(modes) == 1 else "hybride"167 payload["details"] = {"modes_offerts": modes}168 if cities:169 payload["city"] = cities[0]170 if price is not None:171 payload["price"] = price172 payload["price_label"] = f"{price:g} $ + tx"173 if hours is not None:174 payload["duration_hours"] = hours175 payload["duration"] = f"{hours:g} h"176 if instructors:177 payload["instructor"] = ", ".join(dict.fromkeys(instructors))178179 # catégorie : lien du fil d'Ariane (« Technologies de l'information… »)180 cat = soup.find(id="ContentPlaceHolder1_LinkFilArianeCat")181 if cat:182 payload["category"] = clean_text(cat.get_text(" "))183184 # sections riches de la fiche185 objectives = _section_items(soup, r"objectifs? p[ée]dagogiques?")186 if objectives:187 payload["objectives"] = objectives188 program = []189 for h in soup.find_all("h3"):190 txt = clean_text(h.get_text(" "))191 if ((txt.endswith(":") or txt.endswith(" :"))192 and not re.search(r"clients qui ont suivi", txt, re.I)):193 program.append(txt.rstrip(" :"))194 if program:195 payload["program"] = program196 aud = soup.find(["h2", "h3"], string=re.compile(r"client[èe]le", re.I))197 if aud:198 nxt = aud.find_next(["p", "ul"])199 if nxt:200 payload["audience"] = clean_text(nxt.get_text(" "))201 return payload202