SPB Git

spb/forma-ka Public

Python 65.1% TypeScript 17.9% CSS 16.4% HTML 0.5%
11.7 KB · 270 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/mcgill_scs.py : connecteur McGill School of Continuing Studies5#   (continuingstudies.mcgill.ca). Formation continue de l'Université McGill :6#   cours de certificats de perfectionnement professionnel (PDC), ateliers7#   publics, cours en ligne, webinaires… (~200-300 activités).8#   Le site vitrine (www.mcgill.ca/continuingstudies) est derrière un WAF9#   Azure, mais le portail d'inscription Destiny One est librement accessible :10#   - liste : recherche avancée /search/publicCourseAdvancedSearch.do11#     interrogée catégorie par catégorie (avec pagination de session12#     « displaytag ») — titre, sigle, lieu, formats de diffusion13#   - fiche : /search/publicCourseSearchDetails.do?courseId=… — Description,14#     Topics Covered, Learning Outcomes, Who Should Attend?, Course Fee,15#     Duration (hours), CEU et programmes auxquels le cours contribue.16#     Fiches en cache, rafraîchies chaque semaine (clé ISO « AAAA-WSS »).17# -----------------------------------------------------------------------------18from __future__ import annotations1920import datetime21import re22from urllib.parse import urlencode2324from bs4 import BeautifulSoup2526from ..schema import Formation, clean_text2728from .base import BaseConnector2930BASE = "https://continuingstudies.mcgill.ca"31SEARCH_URL = f"{BASE}/search/publicCourseAdvancedSearch.do"32DETAIL_URL = f"{BASE}/search/publicCourseSearchDetails.do"3334# catégories Destiny One de type « formation » -> type de formation Forma-Ka35# (les catégories de tests de classement, services, séances d'information…36#  sont volontairement exclues)37CATEGORIES = {38    "21539211": ("Transcript PDC Course", "Formation continue"),39    "21539210": ("Transcript Non-PDC Course", "Formation continue"),40    "21539205": ("Public Workshop", "Atelier"),41    "22158260": ("Short Program Course", "Formation continue"),42    "21539087": ("Asynchronous Online Course", "Cours en ligne"),43    "17116": ("Courses", "Formation continue"),44    "17117": ("Online courses", "Cours en ligne"),45    "17114": ("Workshops", "Atelier"),46    "17120": ("Intensive courses", "Formation continue"),47    "17119": ("Summer courses", "Formation continue"),48    "17122": ("Lecture Series", "Conférence"),49    "17124": ("Webinars", "Webinaire"),50}5152_FEE_RE = re.compile(r"Course Fee\s*:?\s*\$?\s*([\d,]+(?:\.\d{2})?)", re.I)53_HOURS_RE = re.compile(r"Duration\s*\(hours\)\s*:?\s*(\d+(?:\.\d+)?)", re.I)54_CEU_RE = re.compile(r"(\d+(?:\.\d+)?)\s*Continuing Education Units?", re.I)5556# formats de diffusion Destiny -> mode canonique Forma-Ka57_MODE_MAP = [58    (re.compile(r"blended", re.I), "hybride"),59    (re.compile(r"asynchronous|self[- ]paced", re.I), "asynchrone"),60    (re.compile(r"online|remote|virtual", re.I), "en ligne"),61    (re.compile(r"in[- ]class|in[- ]person|classroom", re.I), "présentiel"),62]636465def _mode_from(formats: list[str]) -> str:66    modes = []67    for fmt in formats:68        for rx, canon in _MODE_MAP:69            if rx.search(fmt):70                modes.append(canon)71                break72    modes = list(dict.fromkeys(modes))73    if not modes:74        return ""75    if len(modes) == 1:76        return modes[0]77    return "hybride"        # plusieurs formats offerts (en classe et en ligne)787980def _section_content(h2) -> tuple[list[str], list[str]]:81    """(paragraphes, items) qui suivent un titre h2 de la fiche Destiny."""82    paras: list[str] = []83    items: list[str] = []84    for sib in h2.find_next_siblings():85        if sib.name in ("h1", "h2"):86            break87        if sib.name in ("ul", "ol"):88            items += [clean_text(li.get_text(" ")) for li in sib.find_all("li")]89        elif sib.name == "p":90            paras.append(clean_text(sib.get_text(" ")))91        elif sib.name == "div":92            items += [clean_text(li.get_text(" ")) for li in sib.find_all("li")]93            paras += [clean_text(p.get_text(" ")) for p in sib.find_all("p")]94    if not paras and not items:95        # certains gabarits mettent le texte directement dans le conteneur du h296        # (ex. div#courseProfileRecommendedPrerequisites)97        parent = h2.parent98        if parent is not None:99            items = [clean_text(li.get_text(" ")) for li in parent.find_all("li")]100            txt = clean_text(parent.get_text(" "))101            txt = re.sub(r"^" + re.escape(clean_text(h2.get_text(" "))) + r"\s*",102                         "", txt)103            for i in items:104                txt = txt.replace(i, "")105            txt = clean_text(txt)106            if txt:107                paras = [txt]108    return [p for p in paras if p], [i for i in items if i]109110111class McgillScsConnector(BaseConnector):112    source_id = "mcgill_scs"113    request_delay = 0.6114115    def fetch(self) -> list[Formation]:116        # amorce de session Destiny (cookie JSESSIONID pour la pagination)117        self.get(f"{SEARCH_URL}?method=load")118119        # 1) Liste : une recherche par catégorie, pagination de session120        found: dict[str, dict] = {}          # courseId -> infos de la liste121        for cat_id, (cat_label, ttype) in CATEGORIES.items():122            params = {123                "method": "doPaginatedSearch",124                "showInternal": "false",125                "courseSearch.courseCategoryStringArray": cat_id,126                "courseSearch.filterString": "all",127            }128            html = self.get(f"{SEARCH_URL}?{urlencode(params)}").text129            pages = [html]130            # liens « displaytag » de pagination (d-XXXX-p=N), propres à la session131            page_links = sorted(set(re.findall(132                r'href="(/search/publicCourseAdvancedSearch\.do\?method=doPagination'133                r'[^"]*?-p=(\d+))"', html)), key=lambda t: int(t[1]))134            for link, _n in page_links:135                pages.append(self.get(BASE + link.replace("&", "&")).text)136137            for page_html in pages:138                for row in self._parse_rows(page_html):139                    cid = row["course_id"]140                    if cid not in found:141                        row["category_label"] = cat_label142                        row["training_type"] = ttype143                        found[cid] = row144145        # 2) Fiche détaillée par cours — cache hebdomadaire146        week = datetime.date.today().strftime("%G-W%V")147        out: list[Formation] = []148        for cid, row in found.items():149            url = f"{DETAIL_URL}?method=load&courseId={cid}"150            key = f"{week}:{row['title']}"151            payload = self.detail(cid, key, lambda u=url: self._fetch_detail(u))152            f = Formation(153                source=self.source_id,154                external_id=cid,155                url=url,156                title=row["title"],157                training_type=row["training_type"],158                language="en",159                code=row.get("code", ""),160                mode=_mode_from(row.get("formats", [])),161            )162            if row.get("formats"):163                f.details["formats_offerts"] = row["formats"]164            f.details["categorie_scs"] = row["category_label"]165            if row.get("location"):166                f.details["lieu"] = row["location"]167                if "montreal" in row["location"].lower():168                    f.city = "Montréal"169            if row.get("availability"):170                f.details["disponibilite"] = row["availability"]171            for k, v in (payload or {}).items():172                if k == "details":173                    f.details = {**f.details, **v}174                elif hasattr(f, k) and v not in (None, "", []):175                    setattr(f, k, v)176            out.append(f)177        return out178179    # -- liste ------------------------------------------------------------------180    @staticmethod181    def _parse_rows(html: str) -> list[dict]:182        soup = BeautifulSoup(html, "html.parser")183        rows: list[dict] = []184        for td in soup.select("td.firstColumn"):185            a = td.select_one(".courseName a[href]")186            if a is None:187                continue188            m = re.search(r"courseId=(\d+)", a["href"])189            if not m:190                continue191            tr = td.find_parent("tr")192            code_el = td.select_one(".courseCode")193            tds = tr.find_all("td") if tr else []194            location = clean_text(tds[1].get_text(" ")) if len(tds) > 1 else ""195            formats = [clean_text(x.get_text(" "))196                       for x in tr.select(".courseDelivery a")] if tr else []197            avail_el = tr.select_one(".courseAvailability") if tr else None198            rows.append({199                "course_id": m.group(1),200                "title": clean_text(a.get_text(" ")),201                "code": clean_text(code_el.get_text(" ")) if code_el else "",202                "location": re.sub(r"\s*,\s*", ", ", location),203                "formats": [f for f in formats if f],204                "availability": clean_text(avail_el.get_text(" ")) if avail_el else "",205            })206        return rows207208    # -- fiche ------------------------------------------------------------------209    def _fetch_detail(self, url: str) -> dict:210        html = self.fetch_html(url)211        soup = BeautifulSoup(html, "html.parser")212        payload: dict = {}213        details: dict = {}214215        for h2 in soup.find_all("h2"):216            heading = clean_text(h2.get_text(" ")).lower()217            if heading == "description":218                paras, items = _section_content(h2)219                if paras:220                    desc = "\n\n".join(paras)221                    payload["description"] = re.sub(222                        r"^Official Description\s*", "", desc)223                ceu = _CEU_RE.search(" ".join(paras))224                if ceu:225                    payload["credits"] = f"{ceu.group(1)} CEU"226            elif heading.startswith("topics covered"):227                _, items = _section_content(h2)228                if items:229                    payload["program"] = items230            elif heading.startswith("learning outcomes"):231                paras, items = _section_content(h2)232                objs = [o for o in (items or paras) if not o.endswith(":")]233                if objs:234                    payload["objectives"] = objs235            elif heading.startswith("who should attend"):236                paras, items = _section_content(h2)237                aud = " ".join(paras) or "; ".join(items)238                if aud:239                    payload["audience"] = aud240            elif heading.startswith("applies towards"):241                ul = h2.find_next("ul")242                progs = [clean_text(a.get_text(" "))243                         for a in (ul.find_all("a") if ul else [])]244                progs = list(dict.fromkeys(p for p in progs if p))245                if progs:246                    details["programmes_lies"] = progs247                    # catégorie lisible : « PDC in Business Analysis » -> sujet248                    cat = re.sub(r"^(Professional Development |Graduate )?"249                                 r"Certificate in ", "", progs[0]).strip(" :")250                    if cat:251                        payload["category"] = cat252253        body_txt = clean_text(soup.get_text(" "))254        m = _FEE_RE.search(body_txt)255        if m:256            payload["price"] = float(m.group(1).replace(",", ""))257            payload["price_label"] = f"Course Fee: ${m.group(1)}"258        m = _HOURS_RE.search(body_txt)259        if m:260            payload["duration_hours"] = float(m.group(1))261            payload["duration"] = f"{m.group(1)} hours"262        m = re.search(r"Discounts?\s*:\s*(.{10,220}?)(?:Thank you|Required fields|$)",263                      body_txt)264        if m:265            details["rabais"] = clean_text(m.group(1))266267        if details:268            payload["details"] = details269        return payload270