SPB Git forge

spb/sorti-ka

Public

Toutes les sorties et tous les événements du Québec, un seul endroit — 7 connecteurs, fiches SSR, design Groupe KA.

58commits 1branches 0releases
13.7 MBsize
maindefault branch
17 days agolast push
HTML 82.9% Python 15.2% TypeScript 0.9% JavaScript 0.7%
8.6 KB · 203 lines python
Raw Blame History
1# -----------------------------------------------------------------------------2# Sorti-Ka — Agrégateur de sorties & événements (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/chaudiereappalaches.py : Tourisme Chaudière-Appalaches —5#   festivals et événements (ATR, Phase 4). Région faiblement couverte6#   avant connexion (essentiellement SITQ).7#   Source    : https://chaudiereappalaches.com/planifier-votre-sejour/8#               ete-automne/festivals-et-evenements/ — CMS Woody, liste rendue9#               serveur paginée (?listpage=1..3, ~12 fiches/page, revérifié10#               2026-08-25 — liste saisonnière unique, pas d'équivalent hiver).11#               Chaque fiche embarque un bloc JSON-LD Event COMPLET (dates ISO,12#               description, adresse civique, ville, code postal, GPS, images)13#               ET un payload JS « const HwSheet = {...} » (fiche touristique14#               Raccourci/Hubo).15#   Extraction: pages liste (1 GET/page) → fiches détail incrémentales (cache16#               data/chaudiereappalaches_cache.json, CHA_FICHE_MAX/sync,17#               re-visite 7 j, version v=2) → JSON-LD Event (parseur générique18#               ld_events de sallesjsonld). ENRICHI (2026-08-25) : payload19#               HwSheet → site web officiel (websites/bookingUrl) et tarif20#               textuel publié (tariffFree/tariffComplement → price_label21#               brut, ex. « Accès gratuit au site… »). Pas d'heure structurée22#               (openingPeriods.hasHours=false partout, schedules vides) →23#               jamais inventée.24#   Accès     : site public de l'ATR, robots.txt sans interdiction sur25#               /planifier-votre-sejour/ — GET throttlés, UA honnête.26# -----------------------------------------------------------------------------27from __future__ import annotations2829import json30import os31import re32import time3334from ..normalize import clean_text, parse_date_iso35from ..schema import Event36from .base import BaseConnector37from .sallesjsonld import ld_events3839BASE = "https://chaudiereappalaches.com"40LIST_URL = BASE + "/planifier-votre-sejour/ete-automne/festivals-et-evenements/"41PAGE_MAX = 6                                   # 3 pages constatées en 2026-0842FICHE_MAX = int(os.environ.get("CHA_FICHE_MAX", "12"))43REFRESH_AFTER = 7 * 8640044CACHE_V = 2                 # v2 : + site web + tarif textuel (HwSheet),45#                             2026-08-254647_FICHE_RE = re.compile(48    r'href="(https://chaudiereappalaches\.com/planifier-votre-sejour/[^"]*'49    r'/festivals-et-evenements/[a-z0-9-]+-fr-(\d+)/)"')50_LISTPAGE_RE = re.compile(r"listpage=(\d+)")51_HWSHEET_RE = re.compile(r"const\s+HwSheet\s*=")525354def _hwsheet(html: str) -> dict:55    """Payload « const HwSheet = {...} » d'une fiche ({} si absent/cassé)."""56    m = _HWSHEET_RE.search(html)57    if not m:58        return {}59    try:60        d, _ = json.JSONDecoder().raw_decode(html[m.end():].lstrip())61        return d if isinstance(d, dict) else {}62    except Exception:63        return {}646566def _dig(obj, key):67    """Première valeur non-nulle de `key` dans un arbre dict/list (prudent)."""68    if isinstance(obj, dict):69        if obj.get(key) is not None:70            return obj[key]71        for v in obj.values():72            r = _dig(v, key)73            if r is not None:74                return r75    elif isinstance(obj, list):76        for v in obj:77            r = _dig(v, key)78            if r is not None:79                return r80    return None818283class ChaudiereAppalachesConnector(BaseConnector):84    source_id = "chaudiereappalaches"85    request_delay = 0.88687    def _list_fiches(self) -> dict[str, str]:88        """{url de fiche: id numérique} sur toutes les pages de la liste."""89        fiches: dict[str, str] = {}90        first = self.get(LIST_URL).text91        pages = {int(p) for p in _LISTPAGE_RE.findall(first) if p.isdigit()}92        last = min(max(pages or {1}), PAGE_MAX)93        for url, fid in _FICHE_RE.findall(first):94            fiches[url] = fid95        for page in range(2, last + 1):96            try:97                html = self.get(LIST_URL, params={"listpage": page}).text98            except Exception:99                break100            for url, fid in _FICHE_RE.findall(html):101                fiches[url] = fid102        return fiches103104    def _parse_fiche(self, url: str, fid: str, html: str) -> dict | None:105        for ld in ld_events(html):106            title = clean_text(str(ld.get("name") or ""))107            start = parse_date_iso(str(ld.get("startDate") or ""))108            if not title or not start:109                continue110            loc = ld.get("location")111            loc = (loc[0] if isinstance(loc, list) and loc else loc) or {}112            loc = loc if isinstance(loc, dict) else {}113            addr = loc.get("address")114            addr = addr if isinstance(addr, dict) else {}115            geo = loc.get("geo")116            geo = geo if isinstance(geo, dict) else {}117            image = ld.get("image")118            if isinstance(image, list):119                image = image[0] if image else ""120            venue = clean_text(str(loc.get("name") or ""))121            if venue.lower() in ("adresse", title.lower()):122                venue = ""                     # libellé générique du CMS123            # payload HwSheet : site web officiel + tarif textuel publié124            sheet = _hwsheet(html)125            websites = _dig(sheet, "websites") or []126            website = ""127            if isinstance(websites, list) and websites:128                website = str(websites[0]).strip()129            if not website:130                website = str(_dig(sheet, "bookingUrl") or "").strip()131            price_label = clean_text(str(132                _dig(sheet, "tariffFree")133                or _dig(sheet, "tariffComplement") or ""))[:300]134            return {135                "external_id": fid,136                "url": url,137                "title": title,138                "description": clean_text(str(ld.get("description") or "")),139                "venue": venue,140                "address": clean_text(str(addr.get("streetAddress") or "")),141                "city": clean_text(str(addr.get("addressLocality") or "")),142                "postal_code": str(addr.get("postalCode") or "").strip(),143                "lat": geo.get("latitude"), "lng": geo.get("longitude"),144                "start_date": start,145                "end_date": parse_date_iso(str(ld.get("endDate") or "")) or start,146                "website": website if website.startswith("http") else "",147                "price_label": price_label,148                "image": image if isinstance(image, str) else "",149            }150        return None151152    def fetch(self) -> list[Event]:153        fiches = self._list_fiches()154        cache = self.load_cache()155        now = time.time()156        current = set(fiches.values())157        cache = {k: v for k, v in cache.items() if k in current}158        # nouvelles fiches puis celles parsées avant v2 (sans site web/tarif),159        # puis re-visite roulante160        to_fetch = sorted(161            (u for u, fid in fiches.items()162             if fid not in cache or cache[fid].get("v", 1) < CACHE_V163             or now - cache[fid].get("ts", 0) > REFRESH_AFTER),164            key=lambda u: (cache.get(fiches[u], {}).get("v", 1)165                           if fiches[u] in cache else 0,166                           cache.get(fiches[u], {}).get("ts", 0)))[:FICHE_MAX]167        for url in to_fetch:168            fid = fiches[url]169            try:170                row = self._parse_fiche(url, fid, self.get(url).text)171            except Exception:172                row = None173            if row is None:174                print(f"[sorti-ka] chaudiereappalaches : fiche ignorée {url}")175            cache[fid] = {"ts": now, "row": row, "v": CACHE_V}176        self.save_cache(cache)177178        events: list[Event] = []179        for entry in cache.values():180            r = entry.get("row")181            if not r:182                continue183            events.append(Event(184                source=self.source_id,185                external_id=r["external_id"],186                url=r["url"],187                title=r["title"],188                description=r["description"],189                raw_categories=[r["title"]],190                venue=r["venue"],191                address=r["address"],192                city=r["city"],193                postal_code=r["postal_code"],194                tourist_region="Chaudière-Appalaches",195                lat=r["lat"], lng=r["lng"],196                start_date=r["start_date"],197                end_date=r["end_date"],198                price_label=r.get("price_label", ""),199                website=r.get("website", ""),200                image=r["image"],201            ))202        return events203