# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/chaudiereappalaches.py : Tourisme Chaudière-Appalaches — # festivals et événements (ATR, Phase 4). Région faiblement couverte # avant connexion (essentiellement SITQ). # Source : https://chaudiereappalaches.com/planifier-votre-sejour/ # ete-automne/festivals-et-evenements/ — CMS Woody, liste rendue # serveur paginée (?listpage=1..3, ~12 fiches/page, revérifié # 2026-08-25 — liste saisonnière unique, pas d'équivalent hiver). # Chaque fiche embarque un bloc JSON-LD Event COMPLET (dates ISO, # description, adresse civique, ville, code postal, GPS, images) # ET un payload JS « const HwSheet = {...} » (fiche touristique # Raccourci/Hubo). # Extraction: pages liste (1 GET/page) → fiches détail incrémentales (cache # data/chaudiereappalaches_cache.json, CHA_FICHE_MAX/sync, # re-visite 7 j, version v=2) → JSON-LD Event (parseur générique # ld_events de sallesjsonld). ENRICHI (2026-08-25) : payload # HwSheet → site web officiel (websites/bookingUrl) et tarif # textuel publié (tariffFree/tariffComplement → price_label # brut, ex. « Accès gratuit au site… »). Pas d'heure structurée # (openingPeriods.hasHours=false partout, schedules vides) → # jamais inventée. # Accès : site public de l'ATR, robots.txt sans interdiction sur # /planifier-votre-sejour/ — GET throttlés, UA honnête. # ----------------------------------------------------------------------------- from __future__ import annotations import json import os import re import time from ..normalize import clean_text, parse_date_iso from ..schema import Event from .base import BaseConnector from .sallesjsonld import ld_events BASE = "https://chaudiereappalaches.com" LIST_URL = BASE + "/planifier-votre-sejour/ete-automne/festivals-et-evenements/" PAGE_MAX = 6 # 3 pages constatées en 2026-08 FICHE_MAX = int(os.environ.get("CHA_FICHE_MAX", "12")) REFRESH_AFTER = 7 * 86400 CACHE_V = 2 # v2 : + site web + tarif textuel (HwSheet), # 2026-08-25 _FICHE_RE = re.compile( r'href="(https://chaudiereappalaches\.com/planifier-votre-sejour/[^"]*' r'/festivals-et-evenements/[a-z0-9-]+-fr-(\d+)/)"') _LISTPAGE_RE = re.compile(r"listpage=(\d+)") _HWSHEET_RE = re.compile(r"const\s+HwSheet\s*=") def _hwsheet(html: str) -> dict: """Payload « const HwSheet = {...} » d'une fiche ({} si absent/cassé).""" m = _HWSHEET_RE.search(html) if not m: return {} try: d, _ = json.JSONDecoder().raw_decode(html[m.end():].lstrip()) return d if isinstance(d, dict) else {} except Exception: return {} def _dig(obj, key): """Première valeur non-nulle de `key` dans un arbre dict/list (prudent).""" if isinstance(obj, dict): if obj.get(key) is not None: return obj[key] for v in obj.values(): r = _dig(v, key) if r is not None: return r elif isinstance(obj, list): for v in obj: r = _dig(v, key) if r is not None: return r return None class ChaudiereAppalachesConnector(BaseConnector): source_id = "chaudiereappalaches" request_delay = 0.8 def _list_fiches(self) -> dict[str, str]: """{url de fiche: id numérique} sur toutes les pages de la liste.""" fiches: dict[str, str] = {} first = self.get(LIST_URL).text pages = {int(p) for p in _LISTPAGE_RE.findall(first) if p.isdigit()} last = min(max(pages or {1}), PAGE_MAX) for url, fid in _FICHE_RE.findall(first): fiches[url] = fid for page in range(2, last + 1): try: html = self.get(LIST_URL, params={"listpage": page}).text except Exception: break for url, fid in _FICHE_RE.findall(html): fiches[url] = fid return fiches def _parse_fiche(self, url: str, fid: str, html: str) -> dict | None: for ld in ld_events(html): title = clean_text(str(ld.get("name") or "")) start = parse_date_iso(str(ld.get("startDate") or "")) if not title or not start: continue loc = ld.get("location") loc = (loc[0] if isinstance(loc, list) and loc else loc) or {} loc = loc if isinstance(loc, dict) else {} addr = loc.get("address") addr = addr if isinstance(addr, dict) else {} geo = loc.get("geo") geo = geo if isinstance(geo, dict) else {} image = ld.get("image") if isinstance(image, list): image = image[0] if image else "" venue = clean_text(str(loc.get("name") or "")) if venue.lower() in ("adresse", title.lower()): venue = "" # libellé générique du CMS # payload HwSheet : site web officiel + tarif textuel publié sheet = _hwsheet(html) websites = _dig(sheet, "websites") or [] website = "" if isinstance(websites, list) and websites: website = str(websites[0]).strip() if not website: website = str(_dig(sheet, "bookingUrl") or "").strip() price_label = clean_text(str( _dig(sheet, "tariffFree") or _dig(sheet, "tariffComplement") or ""))[:300] return { "external_id": fid, "url": url, "title": title, "description": clean_text(str(ld.get("description") or "")), "venue": venue, "address": clean_text(str(addr.get("streetAddress") or "")), "city": clean_text(str(addr.get("addressLocality") or "")), "postal_code": str(addr.get("postalCode") or "").strip(), "lat": geo.get("latitude"), "lng": geo.get("longitude"), "start_date": start, "end_date": parse_date_iso(str(ld.get("endDate") or "")) or start, "website": website if website.startswith("http") else "", "price_label": price_label, "image": image if isinstance(image, str) else "", } return None def fetch(self) -> list[Event]: fiches = self._list_fiches() cache = self.load_cache() now = time.time() current = set(fiches.values()) cache = {k: v for k, v in cache.items() if k in current} # nouvelles fiches puis celles parsées avant v2 (sans site web/tarif), # puis re-visite roulante to_fetch = sorted( (u for u, fid in fiches.items() if fid not in cache or cache[fid].get("v", 1) < CACHE_V or now - cache[fid].get("ts", 0) > REFRESH_AFTER), key=lambda u: (cache.get(fiches[u], {}).get("v", 1) if fiches[u] in cache else 0, cache.get(fiches[u], {}).get("ts", 0)))[:FICHE_MAX] for url in to_fetch: fid = fiches[url] try: row = self._parse_fiche(url, fid, self.get(url).text) except Exception: row = None if row is None: print(f"[sorti-ka] chaudiereappalaches : fiche ignorée {url}") cache[fid] = {"ts": now, "row": row, "v": CACHE_V} self.save_cache(cache) events: list[Event] = [] for entry in cache.values(): r = entry.get("row") if not r: continue events.append(Event( source=self.source_id, external_id=r["external_id"], url=r["url"], title=r["title"], description=r["description"], raw_categories=[r["title"]], venue=r["venue"], address=r["address"], city=r["city"], postal_code=r["postal_code"], tourist_region="Chaudière-Appalaches", lat=r["lat"], lng=r["lng"], start_date=r["start_date"], end_date=r["end_date"], price_label=r.get("price_label", ""), website=r.get("website", ""), image=r["image"], )) return events