SPB Git

spb/forma-ka Public

Python 65.1% TypeScript 17.9% CSS 16.4% HTML 0.5%
8.2 KB · 209 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/lesaffaires.py : connecteur Événements Les Affaires5#   (evenements.lesaffaires.com) — conférences, formations et webinaires6#   d'affaires du média Les Affaires (Montréal/Québec/en ligne).7#   Boutique Shopify : l'API publique /products.json expose TOUT le catalogue8#   (un produit = un événement) avec des tags très structurés :9#   - Format_Conférence / Formation / Webinaire / Démos du midi…10#   - Thématique_Technologies de l'information / Finances / RH…11#   - start-date_/end-date_AAAA-MM-JJ HH:MM:SS, Emplacement_Montréal/En ligne,12#     Expert_Prénom Nom (conférenciers), edition-Gratuit, « En différé ».13#   Prix par variante (Présentiel / Virtuel), description dans body_html.14#   Aucune page détail nécessaire : la liste JSON contient tout.15# -----------------------------------------------------------------------------16from __future__ import annotations1718import re1920from bs4 import BeautifulSoup2122from ..schema import Formation, clean_text23from .base import BaseConnector2425BASE = "https://evenements.lesaffaires.com"26PRODUCTS_URL = f"{BASE}/products.json"2728# préfixe de date dans les titres : « 2026-11-19***Ambition »29# (parfois double : « 2026-12-09 & 2026-12-10***Masterclass Défense »)30_TITLE_DATE_RE = re.compile(r"^\s*20\d{2}-\d{2}-\d{2}[^*]*\*+\s*")31_DATETIME_RE = re.compile(r"(20\d{2}-\d{2}-\d{2})[ T](\d{2}:\d{2})")3233# tag Format_… -> type de formation canonique Forma-Ka34_FORMAT_MAP = {35    "conférence": "Conférence",36    "inédit : souper-conférence": "Conférence",37    "salon": "Conférence",38    "webinaire": "Webinaire",39    "démos du midi": "Webinaire",40    "formation": "Formation continue",41    "formation am": "Formation continue",42    "formation pm": "Formation continue",43}444546def _parse_tags(tags: list[str]) -> dict:47    """Décompose les tags Shopify structurés (Format_, Expert_, start-date_…)."""48    out: dict = {"experts": [], "formats": [], "themes": [], "flags": []}49    for tag in tags:50        if "_" in tag:51            prefix, value = tag.split("_", 1)52            prefix = prefix.strip().lower()53            value = value.strip()54            if prefix == "format":55                out["formats"].append(value)56            elif prefix in ("thématique", "thematique"):57                out["themes"].append(value)58            elif prefix == "expert":59                out["experts"].append(value)60            elif prefix == "emplacement":61                out["emplacement"] = value62            elif prefix == "start-date":63                out["start"] = value64            elif prefix == "end-date":65                out["end"] = value66            else:67                out["flags"].append(tag)68        else:69            out["flags"].append(tag)70    return out717273class LesAffairesConnector(BaseConnector):74    source_id = "lesaffaires"75    request_delay = 0.57677    def fetch(self) -> list[Formation]:78        # 1) catalogue complet via l'API Shopify (paginée par sécurité)79        products: list[dict] = []80        page = 181        while True:82            resp = self.get(PRODUCTS_URL, params={"limit": 250, "page": page})83            batch = resp.json().get("products", [])84            products.extend(batch)85            if len(batch) < 250:86                break87            page += 18889        out: list[Formation] = []90        for prod in products:91            tags = _parse_tags(prod.get("tags") or [])92            if "hide" in [f.lower() for f in tags["flags"]]:93                continue94            f = self._build(prod, tags)95            if f is not None:96                out.append(f)97        return out9899    # -- construction d'une formation depuis un produit Shopify ----------------100    def _build(self, prod: dict, tags: dict) -> Formation | None:101        handle = prod.get("handle") or ""102        raw_title = prod.get("title") or ""103        if not handle or not raw_title:104            return None105        title = _TITLE_DATE_RE.sub("", raw_title).strip()106107        f = Formation(108            source=self.source_id,109            external_id=handle,110            url=f"{BASE}/products/{handle}",111            title=title,112            language="fr",113        )114115        # type d'événement depuis le tag Format_116        fmt = (tags["formats"][0] if tags["formats"] else "").lower()117        f.training_type = _FORMAT_MAP.get(fmt, "Conférence")118        if tags["formats"]:119            f.details["format"] = tags["formats"][0]120121        # thématique -> catégorie + tags122        if tags["themes"]:123            f.category = tags["themes"][0]124        f.tags = tags["themes"] + tags["formats"]125126        # conférenciers / experts127        if tags["experts"]:128            f.instructor = ", ".join(dict.fromkeys(tags["experts"]))129130        # description : body_html du produit131        body = prod.get("body_html") or ""132        if body:133            f.description = BeautifulSoup(body, "html.parser").get_text("\n").strip()134135        # dates précises (tags start-date_/end-date_)136        start = tags.get("start")137        end = tags.get("end")138        ms = _DATETIME_RE.search(start or "")139        me = _DATETIME_RE.search(end or "")140        if ms:141            f.start_date = ms.group(1)142            f.sessions = [ms.group(1)]143            f.schedule_label = f"{ms.group(1)} {ms.group(2)}"144            f.details["start_datetime"] = f"{ms.group(1)} {ms.group(2)}"145        if me:146            f.details["end_datetime"] = f"{me.group(1)} {me.group(2)}"147        if ms and me:148            if ms.group(1) == me.group(1):     # même jour -> durée en heures149                h1, m1 = map(int, ms.group(2).split(":"))150                h2, m2 = map(int, me.group(2).split(":"))151                hours = (h2 * 60 + m2 - h1 * 60 - m1) / 60.0152                if hours > 0:153                    f.duration_hours = round(hours, 2)154                    f.duration = f"{hours:g} h"155            else:                              # plusieurs jours156                import datetime157                d1 = datetime.date.fromisoformat(ms.group(1))158                d2 = datetime.date.fromisoformat(me.group(1))159                days = (d2 - d1).days + 1160                f.duration = f"{days} jours"161162        # événement en différé (rediffusion à la demande)163        if any(fl.lower().startswith("en différé") for fl in tags["flags"]):164            f.tags.append("En différé")165            f.details["en_differe"] = True166167        # mode / ville : Emplacement_ + variantes Présentiel/Virtuel168        variants = [v for v in (prod.get("variants") or []) if isinstance(v, dict)]169        vtitles = {clean_text(v.get("title", "")).lower() for v in variants}170        emplacement = tags.get("emplacement", "")171        has_pres = "présentiel" in vtitles172        has_virt = "virtuel" in vtitles173        if has_pres and has_virt:174            f.mode = "hybride"175        elif emplacement:176            f.mode = "en ligne" if emplacement.lower() == "en ligne" else "présentiel"177        if emplacement and emplacement.lower() != "en ligne":178            f.city = emplacement179180        # prix : minimum des variantes (le 1 $ des gratuits est un artefact)181        prices = []182        labels = []183        for v in variants:184            try:185                p = float(v.get("price") or 0)186            except (TypeError, ValueError):187                continue188            prices.append(p)189            vt = clean_text(v.get("title", ""))190            if vt and vt.lower() != "default title":191                labels.append(f"{p:g} $ ({vt})")192        gratuit = any(fl.lower() == "edition-gratuit" for fl in tags["flags"]) \193            or "edition-Gratuit" in (prod.get("tags") or [])194        if gratuit or (prices and min(prices) == 0.0):195            f.price = 0.0196            f.is_free = True197            f.price_label = "Gratuit"198        elif prices and min(prices) > 1.0:      # 1 $ = placeholder Shopify199            f.price = min(prices)200            f.price_label = " / ".join(labels) if labels else f"{min(prices):g} $"201202        # images du produit203        for img in prod.get("images") or []:204            src = img.get("src") if isinstance(img, dict) else None205            if src:206                f.images.append(src)207208        return f209