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/ithq.py : connecteur ITHQ (ithq.qc.ca)5#   Ateliers grand public (cours SAQ par ITHQ : vins, spiritueux, dégustation)6#   et formations continues professionnelles (tourisme, hôtellerie,7#   restauration) de l'Institut de tourisme et d'hôtellerie du Québec.8#   Site WordPress : l'API REST publique expose deux types de contenus9#   (« workshop » et « formation ») avec le HTML Gutenberg complet de chaque10#   fiche + taxonomies (_embed) — AUCUNE page à visiter individuellement :11#   - caractéristiques : Durée, Formule (en présence/en ligne), Langue, Coût12#   - blocs libres : présentation, « Ce que vous découvrirez/apprendrez »13#     (objectifs), Contenu, Préalable, Coût, Modalités (attestation)14#   - bloc de dates : séances datées par ville (Montréal, Québec, Brossard…)15#   - taxonomies : catégorie, langue, niveau, domaine, clientèle, formule.16# -----------------------------------------------------------------------------17from __future__ import annotations1819import json20import re2122from bs4 import BeautifulSoup2324from ..schema import Formation, clean_text, parse_date_fr25from .base import BaseConnector2627BASE = "https://www.ithq.qc.ca"28API_URL = f"{BASE}/wp-json/wp/v2/{{cpt}}?per_page=100&page={{page}}&_embed=1"2930# type de contenu WordPress -> type de formation Forma-Ka31_CPT_TYPES = {"workshop": "Atelier", "formation": "Formation continue"}3233# titres de sections (h2/h3/h4 Gutenberg) -> rôle dans le schéma34_SECTION_ROLES = [35    (re.compile(r"d[ée]couvrirez|apprendrez|objectifs?", re.I), "objectives"),36    (re.compile(r"^contenu|programme|au menu|d[ée]roulement", re.I), "program"),37    (re.compile(r"pr[ée]alable", re.I), "prerequisites"),38    (re.compile(r"^co[ûu]t", re.I), "price"),39    (re.compile(r"modalit[ée]s?", re.I), "modalites"),40    (re.compile(r"s[’']adresse|client[èe]le|à qui", re.I), "audience"),41    (re.compile(r"^dur[ée]e", re.I), "duration"),42    (re.compile(r"^pr[ée]sentation", re.I), ""),43    (re.compile(r"^dates?$|information compl[ée]mentaire", re.I), "skip"),44]4546# « Formule » / taxonomie activity-type -> mode canonique Forma-Ka47_FORMULE_MAP = {"en présence": "présentiel", "en ligne": "en ligne",48                "hybride": "hybride"}495051def _terms_by_taxonomy(item: dict) -> dict[str, list[str]]:52    """Noms des termes de taxonomie ({'workshop-category': ['Cours vins']…})."""53    out: dict[str, list[str]] = {}54    for group in (item.get("_embedded", {}).get("wp:term") or []):55        for term in group or []:56            if isinstance(term, dict) and term.get("name"):57                out.setdefault(term.get("taxonomy", ""), []).append(term["name"])58    return out596061def _section_role(title: str) -> str | None:62    """Rôle d'un titre de section, ou None si ce n'est pas un titre connu."""63    for rx, role in _SECTION_ROLES:64        if rx.search(title):65            return role66    return None676869class IthqConnector(BaseConnector):70    source_id = "ithq"71    request_delay = 0.57273    def fetch(self) -> list[Formation]:74        out: list[Formation] = []75        for cpt, training_type in _CPT_TYPES.items():76            for item in self._fetch_items(cpt):77                out.append(self._build(item, cpt, training_type))78        return out7980    # -- API REST -----------------------------------------------------------------81    def _get_json(self, url: str) -> list:82        """JSON de l'API avec repli : requests direct -> Scrapfly (le site83        refuse certains agents utilisateurs, mais l'API reste publique)."""84        try:85            return self.get(url).json()86        except Exception:87            pass88        text = self.get_scrapfly(url).strip()89        if text.startswith("<"):             # réponse enrobée de HTML90            text = BeautifulSoup(text, "html.parser").get_text()91        return json.loads(text)9293    def _fetch_items(self, cpt: str) -> list[dict]:94        """Tous les contenus publiés d'un type donné (pagination REST)."""95        items: list[dict] = []96        page = 197        while True:98            batch = self._get_json(API_URL.format(cpt=cpt, page=page))99            items.extend(batch)100            if len(batch) < 100:101                break102            page += 1103        return items104105    # -- construction ---------------------------------------------------------------106    def _build(self, item: dict, cpt: str, training_type: str) -> Formation:107        taxes = _terms_by_taxonomy(item)108        title = BeautifulSoup(item.get("title", {}).get("rendered", ""),109                              "html.parser").get_text()110        f = Formation(111            source=self.source_id,112            external_id=f"{cpt}:{item.get('slug', item.get('id'))}",113            url=item.get("link", ""),114            title=clean_text(title),115            training_type=training_type,116            category=(taxes.get(f"{cpt}-category") or [""])[0],117            mode=_FORMULE_MAP.get(118                (taxes.get("activity-type") or [""])[0].lower(),119                (taxes.get("activity-type") or [""])[0]),120            language=(taxes.get(f"{cpt}-language") or [""])[0],121            level=(taxes.get("workshop-level") or [""])[0],122            audience=", ".join(taxes.get("formation-clientele") or []),123            tags=(taxes.get("formation-domain") or [])124                 + (taxes.get("formation-family") or []),125        )126        media = item.get("_embedded", {}).get("wp:featuredmedia") or []127        if media and isinstance(media[0], dict) and media[0].get("source_url"):128            f.images = [media[0]["source_url"]]129        self._parse_content(f, item.get("content", {}).get("rendered", ""))130        return f131132    # -- HTML Gutenberg de la fiche ---------------------------------------------------133    def _parse_content(self, f: Formation, html: str) -> None:134        soup = BeautifulSoup(html, "html.parser")135136        # 1) bloc de caractéristiques (Durée / Formule / Langue / Coût / Domaine)137        for it in soup.find_all(class_="activity-characteristics__item"):138            label = clean_text((it.find(class_="label") or it).get_text(" "))139            content = it.find(class_="content")140            value = clean_text(content.get_text(" ")) if content else ""141            if not value:142                continue143            low = label.lower()144            if low.startswith("dur"):145                f.duration = value146            elif low.startswith("formule") and not f.mode:147                f.mode = _FORMULE_MAP.get(value.lower(), value)148            elif low.startswith("langue") and not f.language:149                f.language = value150            elif low.startswith("co"):151                f.price_label = value152            elif low.startswith("domaine") and not f.category:153                f.category = value154155        # 2) blocs libres au premier niveau : description + sections titrées156        #    (les titres de section connus — Présentation, Objectifs, Contenu,157        #    Coût, Modalités… — routent les blocs qui les suivent ; les autres158        #    titres sont des sous-titres qui restent dans la section courante)159        desc_parts: list[str] = []160        role = ""161        for el in soup.find_all(recursive=False):162            cls = " ".join(el.get("class") or [])163            if el.name in ("h2", "h3", "h4") and "wp-block-heading" in cls:164                title = clean_text(el.get_text(" "))165                known = _section_role(title)166                if known is not None:167                    role = known168                else:                           # sous-titre descriptif169                    role = ""170                    desc_parts.append(title)171                continue172            if "activity-dates-block" in cls or "activity-related-block" in cls:173                role = "skip"174            if role == "skip" or el.name in ("section", "hr", "figure"):175                continue176            if el.name not in ("p", "ul", "ol", "div"):177                continue178            items = [clean_text(li.get_text(" ")) for li in el.find_all("li")]179            text = clean_text(el.get_text(" "))180            if not text:181                continue182            if role in ("objectives", "program"):183                target = f.objectives if role == "objectives" else f.program184                if items:185                    target += items186                elif not text.endswith(":"):    # phrase d'amorce ignorée187                    target.append(text)188            elif role == "prerequisites":189                f.prerequisites = (f.prerequisites + " " + text).strip()190            elif role == "price":191                if not f.price_label:192                    f.price_label = text193                f.details.setdefault("cout_details", text)194            elif role == "duration":195                m = re.search(r"\d+\s*h(?:\s*\d{1,2})?\b|\d+\s*min\w*", text)196                if m and not f.duration:197                    f.duration = m.group(0)198            elif role == "modalites":199                f.details.setdefault("modalites", []).extend(items or [text])200                if re.search(r"attestation", text, re.I):201                    f.credential = "Attestation de participation"202            elif role == "audience":203                f.audience = clean_text(f"{f.audience} {text}").strip()204            elif role == "":205                # « Cette formation s'adresse : … » (paragraphe ou bloc dédié)206                m = re.search(r"s[’']adresse\s*:?", text)207                if m:208                    rest = clean_text(text[m.end():])209                    if rest:210                        f.audience = clean_text(f"{f.audience} {rest}").strip()211                    else:212                        role = "audience"213                    continue214                desc_parts += items or [text]215        f.description = "\n\n".join(desc_parts)216217        # 3) séances datées par ville218        sessions, cities = [], []219        for it in soup.find_all(class_="activity-dates-block__date-item"):220            t = it.find(class_="activity-dates-block__date-item__title")221            iso = parse_date_fr(clean_text(t.get_text(" "))) if t else None222            if iso:223                sessions.append(iso)224            city = clean_text(it.get("data-location", ""))225            if city and city not in cities:226                cities.append(city)227        if sessions:228            f.sessions = sorted(set(sessions))229            f.start_date = f.sessions[0]230        if cities:231            f.city = cities[0]232            if len(cities) > 1:233                f.details["villes"] = cities234