SPB Git

spb/forma-ka Public

Python 65.1% TypeScript 17.9% CSS 16.4% HTML 0.5%
10.6 KB · 248 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/institut_leadership.py : connecteur Institut de leadership5#   (institutleadership.ca, campus de Montréal) — certifications, programmes6#   et formations signatures en leadership et gestion (~30 programmes animés7#   par des dirigeants et personnalités québécoises).8#   Site WordPress multisite (sous-site /montreal/) :9#   - liste : liens du menu « Programmes de formation » de la page d'accueil10#     (slugs certification-*, programme-*, formation-signature-*, etc.)11#   - fiche : REST /montreal/wp-json/wp/v2/pages?slug=… -> content.rendered12#     propre (sans gabarit) : description, clientèle (« À QUI S'ADRESSE »),13#     modules (h4), objectifs, tarifs « 5595$ + taxes », cohortes datées,14#     mode en ligne / présentiel, formateurs (« animée par … »), image Yoast.15#   Fiches en cache, rafraîchies chaque semaine (clé ISO AAAA-WSS).16# -----------------------------------------------------------------------------17from __future__ import annotations1819import datetime20import json21import re2223from bs4 import BeautifulSoup2425from ..schema import Formation, clean_text, parse_date_fr26from .base import BaseConnector2728BASE = "https://www.institutleadership.ca"29LIST_URL = f"{BASE}/montreal/"30API_URL = f"{BASE}/montreal/wp-json/wp/v2/pages"3132# slugs de pages « programme » dans le menu (le reste = pages corporatives)33_PROGRAM_SLUG_RE = re.compile(34    r"^(certification-|programme-|formation-signature-|habiletes-politiques|"35    r"lessentiel-|communiquer-|c-lmh|experience-immersive)")3637_PRICE_RE = re.compile(r"(\d[\d\s  ]{2,9})\s*\$\s*(?:\*)?\s*\+\s*taxes", re.I)38_HOURS_RE = re.compile(r"de\s+(\d{1,2})\s*h\s*(\d{2})?\s*à\s+(\d{1,2})\s*h\s*(\d{2})?",39                       re.I)40_DAYS_RE = re.compile(r"(\d+)\s*(?:jours|journées)", re.I)41_INSTRUCTOR_RE = re.compile(42    r"animée?s? par\s+((?:[A-ZÀ-Ý][\wà-ÿ'’.-]+\s+){1,3}[A-ZÀ-Ý][\wà-ÿ'’.-]+)")43_DAY_MONTH_RE = re.compile(44    r"(\d{1,2}|1er)\s+(janvier|février|mars|avril|mai|juin|juillet|août|"45    r"septembre|octobre|novembre|décembre)", re.I)46_YEAR_RE = re.compile(r"\b(20\d{2})\b")4748# mot-clé du slug/titre -> catégorie Forma-Ka49_CATEGORY_MAP = [50    (re.compile(r"\bia\b|chatgpt|copilot|claude|numérique|agentique", re.I),51     "Intelligence artificielle"),52    (re.compile(r"gouvernance", re.I), "Gouvernance"),53    (re.compile(r"strat[ée]gi", re.I), "Stratégie"),54    (re.compile(r"gestion de projet|gestion de projets", re.I), "Gestion de projet"),55    (re.compile(r"n[ée]gociation", re.I), "Négociation"),56    (re.compile(r"finance", re.I), "Finance"),57    (re.compile(r"communiquer|parler en public|réunions|rétroaction", re.I),58     "Communication"),59    (re.compile(r"transfert d", re.I), "Transfert d'entreprise"),60]616263def _json_lenient(raw: str):64    """L'API WP du site ajoute parfois du bruit après le JSON — décodage tolérant."""65    return json.JSONDecoder().raw_decode(raw.lstrip(" \r\n"))[0]666768class InstitutLeadershipConnector(BaseConnector):69    source_id = "institut_leadership"70    request_delay = 0.57172    def fetch(self) -> list[Formation]:73        # 1) liste des programmes depuis le menu de la page d'accueil Montréal74        html = self.fetch_html(LIST_URL)75        slugs = list(dict.fromkeys(76            s for s in re.findall(77                r'href="https?://(?:www\.)?institutleadership\.ca/montreal/([\w-]+)/"',78                html)79            if _PROGRAM_SLUG_RE.match(s)))8081        # 2) fiche par programme via l'API REST — cache hebdomadaire82        week = datetime.date.today().strftime("%G-W%V")83        out: list[Formation] = []84        for slug in slugs:85            payload = self.detail(slug, week, lambda s=slug: self._fetch_detail(s))86            if not payload:87                continue88            f = Formation(89                source=self.source_id,90                external_id=slug,91                url=payload.get("url") or f"{BASE}/montreal/{slug}/",92                language="fr",93            )94            # type + accréditation selon la famille de programme95            if slug.startswith(("certification-", "c-lmh")):96                f.training_type = "Certification"97                f.credential = "Certification de l'Institut de leadership"98            elif slug.startswith(("programme-", "experience-immersive")):99                f.training_type = "Programme"100            else:101                f.training_type = "Formation continue"102            for k, v in payload.items():103                if k != "url" and hasattr(f, k) and v not in (None, "", []):104                    setattr(f, k, v)105            # catégorie par mots-clés (défaut : Leadership)106            probe = f"{slug} {f.title}"107            f.category = next((cat for rx, cat in _CATEGORY_MAP108                               if rx.search(probe)), "Leadership")109            f.tags = sorted({f.category, "Leadership"})110            out.append(f)111        return out112113    # -- fiche ----------------------------------------------------------------114    def _fetch_detail(self, slug: str) -> dict:115        resp = self.get(API_URL, params={116            "slug": slug,117            "_fields": "slug,link,title,content,excerpt,yoast_head_json.og_image",118        })119        pages = _json_lenient(resp.text)120        if not pages:121            return {}122        page = pages[0]123        payload: dict = {124            "title": clean_text(BeautifulSoup(125                page.get("title", {}).get("rendered", ""),126                "html.parser").get_text(" ")),127            "url": page.get("link", ""),128        }129        soup = BeautifulSoup(page.get("content", {}).get("rendered", ""),130                             "html.parser")131        text = soup.get_text("\n")132        lines = [clean_text(l) for l in text.split("\n") if clean_text(l)]133134        # description : paragraphes avant la première grande section (h3)135        first_h3 = soup.find("h3")136        desc_parts = []137        for p in soup.find_all("p"):138            if first_h3 and p.sourceline and first_h3.sourceline \139                    and p.sourceline >= first_h3.sourceline:140                break141            txt = clean_text(p.get_text(" "))142            if len(txt) > 60:143                desc_parts.append(txt)144            if len(" ".join(desc_parts)) > 900:145                break146        payload["description"] = "\n\n".join(desc_parts)147        if not payload["description"]:148            exc = BeautifulSoup(page.get("excerpt", {}).get("rendered", ""),149                                "html.parser").get_text(" ")150            payload["description"] = clean_text(exc)151152        # clientèle visée : section « À QUI S'ADRESSE … »153        aud = soup.find(["h2", "h3"], string=re.compile(r"à qui s['’]adresse", re.I))154        if aud:155            parts = []156            for sib in aud.find_all_next(["p", "ul", "h2", "h3"], limit=6):157                if sib.name in ("h2", "h3"):158                    break159                parts.append(clean_text(sib.get_text(" ")))160            payload["audience"] = " ".join(x for x in parts if x)[:600]161162        # objectifs : liste qui suit « Objectifs de la formation »163        obj = soup.find(string=re.compile(r"objectifs de la (formation|certification)",164                                          re.I))165        if obj:166            ul = obj.find_parent().find_next("ul")167            if ul:168                payload["objectives"] = [clean_text(li.get_text(" "))169                                         for li in ul.find_all("li")]170171        # plan : modules h4 (hors sections administratives)172        program = []173        for h in soup.find_all("h4"):174            txt = clean_text(h.get_text(" "))175            if txt and not re.search(r"à lire aussi|témoignages", txt, re.I):176                program.append(txt)177        if program:178            payload["program"] = program179180        # tarifs : « 5595$ + taxes » (minimum = tarif d'appel)181        prices = []182        labels = []183        for line in lines:184            for m in _PRICE_RE.finditer(line):185                val = float(re.sub(r"[\s  ]", "", m.group(1)))186                if val >= 100:                    # ignore les frais annexes187                    prices.append(val)188                    if line not in labels:189                        labels.append(line)190        if prices:191            payload["price"] = min(prices)192            payload["price_label"] = " | ".join(labels[:3])193194        # cohortes datées : « Automne 2026 – divisé : 30 septembre, … »195        sessions = []196        cohortes = []197        for line in lines:198            my = _YEAR_RE.search(line)199            md = _DAY_MONTH_RE.search(line)200            if md and (my or re.search(r"20\d{2}", line)):201                cohortes.append(line)202                year = my.group(1) if my else ""203                iso = parse_date_fr(f"{md.group(1)} {md.group(2)} {year}")204                if iso:205                    sessions.append(iso)206        if sessions:207            sessions = sorted(set(sessions))208            payload["sessions"] = sessions209            payload["start_date"] = sessions[0]210        if cohortes:211            payload["schedule_label"] = cohortes[0]212            payload["details"] = {"cohortes": cohortes[:12]}213214        # mode et ville215        online = re.search(r"en ligne", text, re.I)216        onsite = re.search(r"pr[ée]sentiel", text, re.I)217        if online and onsite:218            payload["mode"] = "hybride"219        elif online:220            payload["mode"] = "en ligne"221        elif onsite:222            payload["mode"] = "présentiel"223        if onsite and re.search(r"pr[ée]sentiel à Montréal|à Montréal", text, re.I):224            payload["city"] = "Montréal"225226        # durée : « de 8h30 à 13h00 » (demi-journée) ou « 6 jours »227        mh = _HOURS_RE.search(text)228        md = _DAYS_RE.search(text)229        if mh:230            start = int(mh.group(1)) + int(mh.group(2) or 0) / 60231            end = int(mh.group(3)) + int(mh.group(4) or 0) / 60232            if end > start:233                payload["duration_hours"] = round(end - start, 2)234                payload["duration"] = f"{end - start:g} h"235        elif md:236            payload["duration"] = f"{md.group(1)} jours"237238        # formateur(s) : « La formation est animée par … »239        mi = _INSTRUCTOR_RE.search(text)240        if mi:241            payload["instructor"] = clean_text(mi.group(1))242243        # image Yoast (og:image)244        ogs = (page.get("yoast_head_json") or {}).get("og_image") or []245        payload["images"] = [o["url"] for o in ogs if isinstance(o, dict)246                             and o.get("url")]247        return payload248