SPB Git

spb/forma-ka Public

Python 65.1% TypeScript 17.9% CSS 16.4% HTML 0.5%
9.2 KB · 215 lines python
Raw Blame History
1# -----------------------------------------------------------------------------2# Forma-Ka — Agrégateur de formations (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/lewagon_mtl.py : connecteur Le Wagon — campus de Montréal5#   (lewagon.com/fr/montreal) — bootcamps intensifs en développement web /6#   IA / data (Web Development, Data Analytics, Data Science, Data Engineering).7#   Site Rails rendu serveur (Hotwire/Turbo) :8#   - liste : liens /fr/montreal/<slug>-course sur la page campus9#   - fiche : HTML riche (h1, og:description, modules du cursus #curriculum,10#     pré-requis de la section Admission, « X semaines »)11#   - séances : turbo-frame /fr/local_courses/<id>/sections/upcoming_sessions12#     (requiert l'en-tête « Turbo-Frame ») — rythme, dates, format campus/en13#     ligne, prix en CAD et langue (onglets Anglais/Français) par cohorte.14#   Fiches en cache, rafraîchies chaque semaine (clé ISO AAAA-WSS).15# -----------------------------------------------------------------------------16from __future__ import annotations1718import datetime19import json20import re2122from bs4 import BeautifulSoup2324from ..schema import Formation, clean_text25from .base import BaseConnector2627BASE = "https://www.lewagon.com"28LIST_URL = f"{BASE}/fr/montreal"2930# secours si la page campus change : les 4 formations connues de Montréal31_KNOWN_SLUGS = ["web-development-course", "data-analytics-course",32                "data-science-course", "data-engineering-course"]3334_FRAME_RE = re.compile(r"/fr/local_courses/(\d+)/sections/upcoming_sessions")35_PRICE_RE = re.compile(r"([\d][\d\s  ]*)\s*CAD")36_WEEKS_RE = re.compile(r"(\d+)\s*semaines", re.I)37# dates du tableau des séances : « oct. 12, 2026 » (mois abrégé français)38_SESSION_DATE_RE = re.compile(r"([a-zûé]+)\.?\s+(\d{1,2}),\s*(20\d{2})", re.I)3940_MONTHS_FR = {"jan": 1, "fév": 2, "fev": 2, "mar": 3, "avr": 4, "mai": 5,41              "jui": 6, "juin": 6, "juil": 7, "aoû": 8, "aou": 8, "sep": 9,42              "oct": 10, "nov": 11, "déc": 12, "dec": 12}434445def _parse_session_date(text: str) -> str | None:46    """« oct. 12, 2026 » -> « 2026-10-12 » (attention : juin/juil)."""47    m = _SESSION_DATE_RE.search(text or "")48    if not m:49        return None50    raw = m.group(1).lower()51    mo = _MONTHS_FR.get(raw[:4]) or _MONTHS_FR.get(raw[:3])52    if not mo:53        return None54    return f"{int(m.group(3)):04d}-{mo:02d}-{int(m.group(2)):02d}"555657class LeWagonMtlConnector(BaseConnector):58    source_id = "lewagon_mtl"59    request_delay = 0.56061    def fetch(self) -> list[Formation]:62        # 1) liste des formations offertes au campus de Montréal63        html = self.fetch_html(LIST_URL)64        slugs = list(dict.fromkeys(65            m.rsplit("/", 1)[-1]66            for m in re.findall(r'href="/fr/montreal/([\w-]+-course)"', html)))67        if not slugs:68            slugs = _KNOWN_SLUGS6970        # 2) fiche + séances par formation — cache hebdomadaire71        week = datetime.date.today().strftime("%G-W%V")72        out: list[Formation] = []73        for slug in slugs:74            url = f"{BASE}/fr/montreal/{slug}"75            payload = self.detail(slug, week, lambda u=url: self._fetch_detail(u))76            f = Formation(77                source=self.source_id,78                external_id=slug,79                url=url,80                training_type="Bootcamp",81                category="Informatique",82                city="Montréal",83                credential="Certificat Le Wagon",84            )85            for k, v in (payload or {}).items():86                if hasattr(f, k) and v not in (None, "", []):87                    setattr(f, k, v)88            out.append(f)89        return out9091    # -- fiche ----------------------------------------------------------------92    def _fetch_detail(self, url: str) -> dict:93        html = self.fetch_html(url)94        soup = BeautifulSoup(html, "html.parser")95        payload: dict = {}9697        # titre : og:title (« Bootcamp Développement Web à Montréal | Le Wagon »)98        og = soup.find("meta", property="og:title")99        title = (og.get("content", "") if og else "").split("|")[0]100        if not title:101            h1 = soup.find("h1")102            title = h1.get_text(" ", strip=True) if h1 else ""103        payload["title"] = clean_text(title)104105        # description : og:description + phrase d'accroche du héros106        ogd = soup.find("meta", property="og:description")107        if ogd and ogd.get("content"):108            payload["description"] = clean_text(ogd["content"])109110        # durée : « Formez-vous en 9 semaines… »111        m = _WEEKS_RE.search(html)112        if m:113            payload["duration"] = f"{m.group(1)} semaines (temps plein)"114115        # plan de cours : modules de la section #curriculum116        cur = soup.find(id="curriculum")117        if cur:118            program = []119            for h in cur.find_all("h3"):120                txt = clean_text(h.get_text(" "))121                if (txt and not txt.endswith(":") and txt not in program122                        and not _WEEKS_RE.search(txt)):   # titre de section123                    program.append(txt)124            if program:125                payload["program"] = program126127        # pré-requis : bloc « Pré-requis » de la section Admission128        pre = soup.find(string=lambda s: s and s.strip().lower()129                        in ("pré-requis", "prérequis"))130        if pre:131            for p in pre.find_parent().find_all_next("p", limit=5):132                txt = clean_text(p.get_text(" "))133                if (len(txt) < 60 or txt.lower().startswith("en france")134                        or txt.lower().startswith("découvrez")):135                    continue                     # bruit / mention hors Québec136                payload["prerequisites"] = txt137                break138139        # séances : turbo-frame des prochaines sessions140        mf = _FRAME_RE.search(html)141        if mf:142            payload.update(self._fetch_sessions(mf.group(0)))143        return payload144145    # -- séances (turbo-frame) --------------------------------------------------146    def _fetch_sessions(self, frame_path: str) -> dict:147        try:148            resp = self.get(f"{BASE}{frame_path}",149                            headers={"Turbo-Frame": "local_course_upcoming_sessions"})150        except Exception:151            return {}152        soup = BeautifulSoup(resp.text, "html.parser")153154        sessions: list[dict] = []155        seen: set = set()156        for pane_id, lang in (("en-tab-pane", "en"), ("fr-tab-pane", "fr")):157            pane = soup.find(id=pane_id)158            if pane is None:159                continue160            for art in pane.find_all("article", class_="upcoming-sessions-grid"):161                cells = [clean_text(d.get_text(" "))162                         for d in art.find_all("div", recursive=False)]163                if len(cells) < 4:164                    continue165                pace, dates, fmt, price_txt = cells[0], cells[1], cells[2], cells[3]166                # métadonnées GTM du bouton (batch_id, format, campus/en ligne)167                gtm = {}168                btn = art.find("a", attrs={"data-gtm-data-layer-attributes-value": True})169                if btn:170                    try:171                        gtm = json.loads(btn["data-gtm-data-layer-attributes-value"])172                    except (ValueError, KeyError):173                        gtm = {}174                key = (gtm.get("batch_id"), lang) if gtm.get("batch_id") \175                    else (pace, dates, lang)176                if key in seen:177                    continue178                seen.add(key)179                found = _SESSION_DATE_RE.findall(dates)180                start = _parse_session_date(dates)181                end = _parse_session_date(dates[dates.find("->") + 2:]) \182                    if "->" in dates else (183                        _parse_session_date(" ".join(184                            f"{a} {b}, {c}" for a, b, c in found[1:2])) or None)185                mprice = _PRICE_RE.search(price_txt)186                sessions.append({187                    "rythme": pace,188                    "debut": start,189                    "fin": end,190                    "format": fmt,191                    "langue": lang,192                    "prix": float(re.sub(r"[\s  ]", "", mprice.group(1)))193                    if mprice else None,194                    "batch_id": gtm.get("batch_id"),195                })196197        if not sessions:198            return {}199        payload: dict = {"details": {"cohortes": sessions}}200        starts = sorted({s["debut"] for s in sessions if s["debut"]})201        if starts:202            payload["sessions"] = starts203            payload["start_date"] = starts[0]204            payload["schedule_label"] = "Rentrées : " + ", ".join(starts)205        modes = {("en ligne" if "ligne" in s["format"].lower() else "présentiel")206                 for s in sessions}207        payload["mode"] = modes.pop() if len(modes) == 1 else "hybride"208        langs = {s["langue"] for s in sessions}209        payload["language"] = "fr/en" if len(langs) > 1 else langs.pop()210        prices = [s["prix"] for s in sessions if s["prix"]]211        if prices:212            payload["price"] = min(prices)213            payload["price_label"] = f"{min(prices):,.0f} CAD".replace(",", " ")214        return payload215