spb/forma-ka Public

1# -----------------------------------------------------------------------------2# Forma-Ka — Agrégateur de formations (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/hec_dirigeants.py : connecteur École des dirigeant(e)s HEC Montréal5#   (ecole-dirigeants.hec.ca — remplace l'ancien ecoledesdirigeants.hec.ca,6#   dont le domaine ne résout plus). ~90 séminaires et certifications pour7#   cadres et gestionnaires (leadership, finance, IA, stratégie…).8#   Boutique Shopify : l'API publique /products.json fournit tout le catalogue9#   (titre, type, prix, variantes avec mode + dates de séance, résumé HTML,10#   images, formateurs en étiquettes). La fiche produit ajoute les sections11#   riches (OBJECTIFS, EST-CE POUR VOUS?, MÉTHODE PÉDAGOGIQUE, Programme,12#   Animé par, durée, fil d'Ariane -> catégorie). Fiches en cache,13#   rafraîchies chaque semaine (clé ISO « AAAA-WSS »).14# -----------------------------------------------------------------------------15from __future__ import annotations1617import datetime18import re1920from bs4 import BeautifulSoup2122from ..schema import Formation, clean_text2324from .base import BaseConnector2526BASE = "https://ecole-dirigeants.hec.ca"27PRODUCTS_URL = f"{BASE}/products.json"2829# les produits « Ticket »/« Event » sont des événements promotionnels gratuits30# (conférences, webinaires d'information) — pas des formations du catalogue31_SKIP_TYPES = {"ticket", "event"}3233# product_type Shopify -> type de formation Forma-Ka34_TYPE_MAP = {35    "certification": "Certification",36    "séminaire": "Séminaire",37}3839_DUREE_VAL_RE = re.compile(r"\d(?:[.,]\d)?\s*(?:jours?|journ[ée]es?|heures?|h\b|"40                           r"semaines?|mois)", re.I)4142# sections h3 du corps « Présentation » de la fiche43_SECTION_FIELDS = [44    (re.compile(r"^objectifs?", re.I), "objectives"),45    (re.compile(r"est-ce pour vous", re.I), "audience"),46    (re.compile(r"avantages distinctifs", re.I), "_avantages"),47    (re.compile(r"m[ée]thodes? p[ée]dagogiques?", re.I), "_methode"),48]495051def _strip_html(html: str) -> str:52    """HTML de résumé Shopify -> texte propre en paragraphes."""53    soup = BeautifulSoup(html or "", "html.parser")54    paras = [clean_text(p.get_text(" ")) for p in soup.find_all(["p", "li"])]55    paras = [p for p in paras if p]56    return "\n\n".join(paras) or clean_text(soup.get_text(" "))575859def _rich_sections(container) -> dict:60    """Découpe un bloc riche (métachamp Shopify) selon ses titres h3/strong."""61    out: dict[str, list[str]] = {}62    current: str | None = None63    for el in container.find_all(["h3", "h4", "p", "ul", "ol"]):64        if el.name in ("h3", "h4"):65            heading = clean_text(el.get_text(" "))66            current = None67            for rx, field in _SECTION_FIELDS:68                if rx.search(heading):69                    current = field70                    out.setdefault(current, [])71                    break72            continue73        if current is None:74            continue75        if el.name in ("ul", "ol"):76            out[current] += [clean_text(li.get_text(" "))77                             for li in el.find_all("li") if clean_text(li.get_text(" "))]78        else:79            txt = clean_text(el.get_text(" "))80            if txt:81                out[current].append(txt)82    return out838485class HecDirigeantsConnector(BaseConnector):86    source_id = "hec_dirigeants"87    request_delay = 0.58889    def fetch(self) -> list[Formation]:90        # 1) Catalogue complet via l'API Shopify /products.json (paginée)91        products: list[dict] = []92        page = 193        while True:94            resp = self.get(PRODUCTS_URL, params={"limit": 250, "page": page})95            batch = resp.json().get("products", [])96            products += batch97            if len(batch) < 250:98                break99            page += 1100101        week = datetime.date.today().strftime("%G-W%V")102        out: list[Formation] = []103        for prod in products:104            ptype = (prod.get("product_type") or "").strip().lower()105            if ptype in _SKIP_TYPES:106                continue107            handle = prod.get("handle", "")108            url = f"{BASE}/products/{handle}"109110            f = Formation(111                source=self.source_id,112                external_id=handle,113                url=url,114                title=clean_text(prod.get("title", "")),115                training_type=_TYPE_MAP.get(ptype, "Séminaire"),116                language="fr",117                description=_strip_html(prod.get("body_html", "")),118                images=[img["src"] for img in prod.get("images", [])119                        if img.get("src")][:3],120            )121122            # variantes : prix + options (Localisation, Date)123            prices = []124            for var in prod.get("variants", []):125                try:126                    p = float(var.get("price") or 0)127                except (TypeError, ValueError):128                    continue129                if p > 0:130                    prices.append(p)131            if prices:132                f.price = min(prices)133                f.price_label = f"{f.price:g} $ + tx"134                if len(set(prices)) > 1:135                    f.details["price_from"] = True136137            modes, dates = [], []138            for opt in prod.get("options", []):139                name = (opt.get("name") or "").lower()140                values = [v for v in opt.get("values", []) if v]141                if "localisation" in name or "format" in name:142                    modes += values143                elif "date" in name:144                    dates += values145            if modes:146                f.mode = modes[0] if len(set(modes)) == 1 else "hybride"147                f.details["modes_offerts"] = sorted(set(modes))148            if dates:149                f.schedule_label = " ; ".join(dates)150            if re.search(r"présentiel|campus|montr[ée]al", f.mode, re.I):151                f.city = "Montréal"152153            # étiquettes : catégorie + formateurs (affinées par la fiche)154            if prod.get("tags"):155                f.tags = [t for t in prod["tags"] if isinstance(t, str)]156157            # 2) Fiche produit — sections riches, cache hebdomadaire158            key = f"{week}:{prod.get('updated_at', '')}"159            payload = self.detail(handle, key, lambda u=url: self._fetch_detail(u))160            for k, v in (payload or {}).items():161                if k == "details":162                    f.details = {**f.details, **v}163                elif hasattr(f, k) and v not in (None, "", []):164                    setattr(f, k, v)165            out.append(f)166        return out167168    # -- fiche ----------------------------------------------------------------169    def _fetch_detail(self, url: str) -> dict:170        html = self.fetch_html(url)171        soup = BeautifulSoup(html, "html.parser")172        payload: dict = {}173        details: dict = {}174175        # fil d'Ariane : « Accueil > Leadership > … » -> catégorie176        bc = soup.select_one(".product-main__breadcrumbs")177        if bc:178            crumbs = [clean_text(a.get_text(" ")) for a in bc.find_all("a")]179            crumbs = [c for c in crumbs if c and c.lower() != "accueil"]180            if crumbs:181                payload["category"] = crumbs[0]182183        # contenus d'onglets : deux gabarits selon le type de produit —184        # séminaires (.product-main__presentation-body / __programme) ou185        # certifications (onglets dynamiques « hec-tab-* » appariés par ordre186        # aux boutons Présentation / Programme / Animé par)187        tabs: dict[str, "BeautifulSoup"] = {}188        btns = soup.select('[class*="hec-tab-btn"]')189        conts = soup.select('[class*="hec-tab-content"]')190        for btn, cont in zip(btns, conts):191            tabs[clean_text(btn.get_text(" ")).lower()] = cont192193        # corps « Présentation » : OBJECTIFS / EST-CE POUR VOUS? / MÉTHODE…194        body = soup.select_one(".product-main__presentation-body") or \195            tabs.get("présentation")196        if body:197            sections = _rich_sections(body)198            if sections.get("objectives"):199                # on écarte les phrases d'amorce (« Ce programme vous permettra de : »)200                objs = [o for o in sections["objectives"] if not o.endswith(":")]201                payload["objectives"] = objs or sections["objectives"]202            if sections.get("audience"):203                payload["audience"] = " ".join(sections["audience"])204            if sections.get("_avantages"):205                details["avantages_distinctifs"] = sections["_avantages"]206            if sections.get("_methode"):207                details["methode_pedagogique"] = " ".join(sections["_methode"])208209        # onglet « Programme » : plan de la formation210        prog = soup.select_one(".product-main__programme") or tabs.get("programme")211        if prog:212            items = [clean_text(li.get_text(" ")) for li in prog.find_all("li")]213            items = [i for i in items if i]214            if not items:215                items = [clean_text(p.get_text(" ")) for p in prog.find_all("p")216                         if clean_text(p.get_text(" "))]217            if items:218                payload["program"] = items219220        # « Animé par » : formateurs et formatrices221        names = [clean_text(n.get_text(" "))222                 for n in soup.select(".product-main__animateurs-expert-name")]223        titles = [clean_text(t.get_text(" "))224                  for t in soup.select(".product-main__animateurs-expert-title")]225        experts = list(dict.fromkeys(226            f"{n} ({t})" if t else n227            for n, t in zip(names, titles + [""] * len(names)) if n))228        if not experts and tabs.get("animé par") is not None:229            # gabarit certification : noms en <strong>, titres en <p> suivants230            experts = list(dict.fromkeys(231                clean_text(s.get_text(" "))232                for s in tabs["animé par"].find_all("strong")233                if clean_text(s.get_text(" "))))234        if experts:235            payload["instructor"] = ", ".join(experts)236237        # encadré d'inscription : durée (« Durée » suivi de « 2 jours »…)238        lbl = soup.find(string=re.compile(r"^\s*Durée\s*$"))239        if lbl:240            for nxt in lbl.find_all_next(string=True, limit=8):241                val = clean_text(str(nxt))242                if _DUREE_VAL_RE.search(val):243                    payload["duration"] = val244                    break245                if re.search(r"Tarif|Langue|Format", val):246                    break247248        if details:249            payload["details"] = details250        return payload251