SPB Git

spb/forma-ka Public

Python 65.1% TypeScript 17.9% CSS 16.4% HTML 0.5%
9.8 KB · 229 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/technologia.py : connecteur Technologia (technologia.com)5#   Grande firme de formation professionnelle (Montréal/Québec) — ~6006#   formations en TI, IA, gestion de projets, leadership, manufacturier…7#   Site Umbraco + Vue, mais très bien outillé côté données :8#   - liste : API JSON interne /Api/Catalog/Browse (code, titre, thématique,9#     sous-thématique, durée en jours ou en heures, prix régulier et10#     préférentiel, formats ClasseVirtuelle/ELearning/EnClasse)11#   - fiche : JSON complet embarqué (<script id="training-data"> : sommaire,12#     objectif, plan de cours « syllabusHtml », clientèle visée, préalables,13#     méthode pédagogique, formateurs) + EducationEvent JSON-LD (séances14#     datées avec villes). Fiches en cache, rafraîchies chaque semaine.15# -----------------------------------------------------------------------------16from __future__ import annotations1718import datetime19import json20import re2122from bs4 import BeautifulSoup2324from ..schema import Formation, clean_text25from .base import BaseConnector2627BASE = "https://www.technologia.com"28API_BROWSE = f"{BASE}/Api/Catalog/Browse"29LIST_URL = f"{BASE}/formations"3031# le JSON riche embarqué dans chaque fiche32_TRAINING_DATA_RE = re.compile(33    r'<script id="training-data" type="application/json">(.*?)</script>', re.S)34# les scripts JSON-LD ont un attribut type HTML-encodé (« ld&#x2B;json »)35_LDJSON_RE = re.compile(36    r'<script[^>]*type="application/ld(?:\+|&#x2B;)json"[^>]*>(.*?)</script>',37    re.S | re.I)3839# formats Technologia -> mode canonique Forma-Ka40_FORMAT_MAP = {41    "classevirtuelle": "en ligne",42    "elearning": "asynchrone",43    "enclasse": "présentiel",44}454647def _strip_html(fragment: str) -> str:48    """Texte propre d'un fragment HTML (paragraphes séparés)."""49    if not fragment:50        return ""51    soup = BeautifulSoup(fragment, "html.parser")52    parts = [clean_text(p.get_text(" ")) for p in soup.find_all(["p", "li"])]53    parts = [p for p in parts if p]54    return "\n".join(dict.fromkeys(parts)) or clean_text(soup.get_text(" "))555657def _item_mode(formats: list[str] | None) -> tuple[str, list[str]]:58    """(mode canonique, tous les modes offerts) depuis les formats de l'API."""59    modes = [_FORMAT_MAP[f.lower()] for f in (formats or [])60             if f.lower() in _FORMAT_MAP]61    modes = list(dict.fromkeys(modes))62    if not modes:63        return "", []64    return (modes[0] if len(modes) == 1 else "hybride"), modes656667class TechnologiaConnector(BaseConnector):68    source_id = "technologia"69    request_delay = 0.570    limit: int | None = None          # borne optionnelle (tests/débogage)7172    def fetch(self) -> list[Formation]:73        # 1) Catalogue complet via l'API JSON interne74        resp = self.get(API_BROWSE,75                        params={"Count": 5000, "Page": 1, "Culture": "fr-CA"})76        items = resp.json().get("Items") or []77        if self.limit:78            items = items[: self.limit]7980        # 2) Fiche détaillée par formation — cache hebdomadaire81        week = datetime.date.today().strftime("%G-W%V")82        out: list[Formation] = []83        for item in items:84            code = item.get("Code") or item.get("ID") or ""85            url = BASE + (item.get("Uri") or "")86            key = f"{week}:{item.get('Title', '')}:{item.get('RegularPrice')}"87            payload = self.detail(code, key, lambda u=url: self._fetch_detail(u))8889            days = item.get("Duration")90            hours = item.get("DurationInHours")91            duration = (f"{hours:g} h" if hours92                        else f"{days:g} jour{'s' if days and days > 1 else ''}"93                        if days else "")94            price = item.get("RegularPrice")95            tags = [t for t in (item.get("SubThematic"),) if t]96            if "NouveauCours" in (item.get("Tags") or []):97                tags.append("Nouveau")9899            f = Formation(100                source=self.source_id,101                external_id=str(code),102                url=url,103                title=clean_text(item.get("Title", "")),104                training_type="Formation continue",105                category=item.get("Thematic", ""),106                language="fr",107                code=item.get("Code", ""),108                duration=duration,109                duration_hours=float(hours) if hours else None,110                price=float(price) if price is not None else None,111                price_label=f"{price:g} $ + tx" if price is not None else "",112                tags=tags,113            )114            mode, modes = _item_mode(item.get("Formats"))115            if mode:116                f.mode = mode117                f.details["modes_offerts"] = modes118            if item.get("IsOnDemand"):119                f.mode = f.mode or "asynchrone"120            pref = item.get("PreferentialPrice")121            if pref is not None:122                f.details["prix_preferentiel"] = pref123            for k, v in (payload or {}).items():124                if k == "details":125                    f.details.update(v or {})126                elif hasattr(f, k) and v not in (None, "", []):127                    setattr(f, k, v)128            out.append(f)129        return out130131    # -- fiche ----------------------------------------------------------------132    def _fetch_detail(self, url: str) -> dict:133        if not url:134            return {}135        html = self.fetch_html(url)136        payload: dict = {}137        details: dict = {}138139        m = _TRAINING_DATA_RE.search(html)140        if m:141            try:142                data = json.loads(m.group(1))143            except ValueError:144                data = {}145            # description : sommaire + corps de la fiche146            desc = [clean_text(data.get("summary", ""))]147            body = BeautifulSoup(data.get("content") or "", "html.parser")148            desc += [clean_text(p.get_text(" ")) for p in body.find_all("p")]149            payload["description"] = "\n\n".join(150                dict.fromkeys(d for d in desc if d and d != "\xa0"))151152            # objectifs : objectif général + « Ce que vous saurez faire »153            objectives = []154            if data.get("goal"):155                objectives.append(clean_text(data["goal"]))156            arch = BeautifulSoup(data.get("courseArchitecture") or "",157                                 "html.parser")158            objectives += [clean_text(p.get_text(" "))159                           for p in arch.find_all(["p", "li"])]160            payload["objectives"] = [o for o in dict.fromkeys(objectives)161                                     if o and o != "\xa0"]162163            if data.get("prerequisite"):164                payload["prerequisites"] = _strip_html(str(data["prerequisite"]))165            if data.get("targetCustomer"):166                payload["audience"] = clean_text(str(data["targetCustomer"]))167            if data.get("subTitle"):168                details["sous_titre"] = clean_text(data["subTitle"])169            if data.get("teachingMethod"):170                details["methode_pedagogique"] = clean_text(data["teachingMethod"])171            if data.get("satisfaction"):172                details["satisfaction_pct"] = data["satisfaction"]173            if data.get("pduCount"):174                details["pdu"] = data["pduCount"]175176            # plan de cours : section « Contenu de la formation » du syllabus177            syllabus = BeautifulSoup(data.get("syllabusHtml") or "",178                                     "html.parser")179            program, in_content = [], False180            for h in syllabus.find_all(["h2", "h3"]):181                txt = clean_text(h.get_text(" "))182                if h.name == "h2":183                    in_content = bool(re.search(r"contenu", txt, re.I))184                elif in_content and txt:185                    program.append(re.sub(r"^\d+\s*", "", txt))186            if not program:      # repli : puces du corps de la fiche187                program = [clean_text(li.get_text(" "))188                           for li in body.find_all("li")]189            payload["program"] = [p for p in dict.fromkeys(program) if p]190191            teachers = [t.get("name") for t in data.get("teachers") or []192                        if isinstance(t, dict) and t.get("name")]193            if teachers:194                payload["instructor"] = ", ".join(dict.fromkeys(teachers))195            images = [t.get("image") for t in data.get("teachers") or []196                      if isinstance(t, dict) and t.get("image")]197            if images:198                payload["images"] = list(dict.fromkeys(images))199200        # séances datées + villes via les EducationEvent JSON-LD201        sessions, cities = [], []202        for mm in _LDJSON_RE.finditer(html):203            try:204                obj = json.loads(mm.group(1))205            except ValueError:206                continue207            events = obj if isinstance(obj, list) else [obj]208            for ev in events:209                if not isinstance(ev, dict) or ev.get("@type") != "EducationEvent":210                    continue211                start = str(ev.get("startDate", ""))[:10]212                if re.match(r"20\d{2}-\d{2}-\d{2}", start):213                    sessions.append(start)214                loc = ev.get("location") or {}215                addr = loc.get("address") if isinstance(loc, dict) else {}216                city = (addr.get("addressLocality", "")217                        if isinstance(addr, dict) else "")218                if city and city.lower() != "virtuelle":219                    cities.append(city)220        sessions = sorted(set(sessions))221        if sessions:222            payload["sessions"] = sessions223            payload["start_date"] = sessions[0]224        if cities:225            payload["city"] = cities[0]226        if details:227            payload["details"] = details228        return payload229