# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/cantonsdelest.py : connecteur Tourisme Cantons-de-l'Est (ATR Estrie) # Source : cantonsdelest.com/evenements — calendrier officiel de # l'association touristique régionale (données Tourinsoft, la # même plateforme que le SIT Québec : médias mto.tourinsoft.eu). # Extraction: la page liste ne rend que ~13 cartes mais publie ~596 # occurrences (data-found) derrière un bouton « Voir plus » → # PAGINATION AJAX : POST /event/more {ids, offset} avec le # jeton CSRF Yii de la page (vérifié 2026-08-25). Chaque carte # donne lien /evenements//[/], catégorie, # titre, image et « Ville | date FR ». Fiches détail # incrémentales (cache disque data/cantonsdelest_cache.json, # plafond CDE_FICHE_MAX/sync, re-visite 7 j) : JSON-LD Event # complet (description riche, adresse civique, images) + GPS # exact (attributs data-lat/data-lng de la carte). # Accès : pages publiques de l'ATR, rendu serveur — GET/POST directs # throttlés, identification honnête, lien vers la fiche # originale. Recoupement avec le flux SITQ géré par dedup_key. # Prix : non publié de façon structurée → jamais inventé. # ----------------------------------------------------------------------------- from __future__ import annotations import json import os import re import time from ..normalize import clean_text, parse_date_range_fr from ..schema import Event from .base import BaseConnector BASE = "https://www.cantonsdelest.com" LIST_URL = BASE + "/evenements" MORE_URL = BASE + "/event/more" CDE_FICHE_MAX = int(os.environ.get("CDE_FICHE_MAX", "40")) # fiches/sync CDE_PAGE_MAX = int(os.environ.get("CDE_PAGE_MAX", "120")) # pages AJAX/sync REFRESH_AFTER = 7 * 86400 _LD_JSON_RE = re.compile(r"application/ld\+json[^>]*>") _GPS_RE = re.compile(r"data-lat=\"(-?\d+\.?\d*)\" data-lng=\"(-?\d+\.?\d*)\"") _CSRF_PARAM_RE = re.compile(r'\s*' r'(?:]*>)?\s*.*?' r'container-article-category">\s*([^<]*).*?' r'container-article-title[^"]*">(.*?).*?' r'listing-city""?>([^<]*)

', re.S) # mois abrégés du gabarit (« 21 janv. 2027 », « 5 févr. 2027 ») que le parseur # central ne connaît pas — constaté live 2026-08-25 (45/596 cartes) _MONTH_ABBR_RE = re.compile(r"\b(janv|f[ée]vr)\.?", re.I) _MONTH_ABBR = {"janv": "janvier", "fevr": "février", "févr": "février"} def _dates_fr(raw: str) -> tuple[str | None, str | None]: """parse_date_range_fr avec les abréviations propres au site déployées.""" raw = _MONTH_ABBR_RE.sub( lambda m: _MONTH_ABBR[m.group(1).lower()], raw) return parse_date_range_fr(raw) def _cards(html: str) -> list[dict]: """Cartes d'une page liste : path, image, catégorie, titre, ville, dates.""" out = [] for m in _CARD_RE.finditer(html): path, img, cat, title, city_date = m.groups() city, date_raw = "", city_date or "" if "|" in city_date: city, date_raw = city_date.split("|", 1) start, end = _dates_fr(date_raw) parts = path.strip("/").split("/") # evenements//[/] out.append({ "path": path, "external_id": parts[1] + (f"-{parts[3]}" if len(parts) > 3 else ""), "url": BASE + path, "title": clean_text(title), "raw_categories": [clean_text(cat), clean_text(title)], "city": clean_text(city), "start_date": start, "end_date": end, "image": (img or "").strip(), }) return out def _ld_event(html: str) -> dict | None: """Premier bloc JSON-LD de type Event d'une fiche.""" for m in _LD_JSON_RE.finditer(html): try: d, _ = json.JSONDecoder().raw_decode(html[m.end():].lstrip()) except Exception: continue for doc in (d if isinstance(d, list) else [d]): if isinstance(doc, dict) and doc.get("@type") == "Event": return doc return None class CantonsDeLEstConnector(BaseConnector): source_id = "cantonsdelest" request_delay = 0.8 def _post_more(self, data: dict) -> str: """POST throttlé vers /event/more (pagination AJAX Yii).""" self._throttle() resp = self.session.post(MORE_URL, data=data, timeout=self.timeout) self._last_request = time.time() resp.raise_for_status() return resp.text def _list_cards(self) -> list[dict]: """Toutes les cartes du calendrier : page 1 rendue serveur puis pagination AJAX (POST {ids, offset, _csrf}) jusqu'à épuisement.""" first = self.get(LIST_URL).text cards = _cards(first) seen = {c["path"] for c in cards} param = _CSRF_PARAM_RE.search(first) token = _CSRF_TOKEN_RE.search(first) ids = _IDS_RE.search(first) if not (param and token and ids): # structure changée : page 1 seule print("[sorti-ka] cantonsdelest : pagination AJAX introuvable") return cards base = {param.group(1): token.group(1), "ids": ids.group(1)} for offset in range(2, CDE_PAGE_MAX + 1): try: more = _cards(self._post_more({**base, "offset": offset})) except Exception: break new = [c for c in more if c["path"] not in seen] if not new: # page vide ou répétée → fin break cards.extend(new) seen.update(c["path"] for c in new) return cards def _parse_fiche(self, path: str, html: str) -> dict | None: ld = _ld_event(html) if not ld or not ld.get("name"): return None addr = ((ld.get("location") or {}).get("address") or {}) gps = _GPS_RE.search(html) images = ld.get("image") or [] if isinstance(images, str): images = [images] parts = path.strip("/").split("/") # evenements//[/] ext = parts[1] + (f"-{parts[3]}" if len(parts) > 3 else "") return { "external_id": ext, "url": BASE + path, "title": clean_text(ld.get("name") or ""), "description": clean_text(ld.get("description") or ""), "raw_categories": [clean_text(ld.get("name") or ""), parts[2].replace("-", " ")], "address": clean_text(addr.get("streetAddress") or ""), "city": clean_text(addr.get("addressLocality") or ""), "postal_code": (addr.get("postalCode") or "").strip(), "lat": float(gps.group(1)) if gps else None, "lng": float(gps.group(2)) if gps else None, "start_date": ld.get("startDate"), "end_date": ld.get("endDate") or ld.get("startDate"), "image": images[0] if images else "", } def fetch(self) -> list[Event]: cards = self._list_cards() paths = [c["path"] for c in cards] cache = self.load_cache() now = time.time() current = set(paths) cache = {p: c for p, c in cache.items() if p in current} to_fetch = sorted( (p for p in paths if p not in cache or now - cache[p].get("fetched_at", 0) > REFRESH_AFTER), key=lambda p: cache.get(p, {}).get("fetched_at", 0))[:CDE_FICHE_MAX] def worker(path, session): return self._parse_fiche( path, session.get(BASE + path, timeout=20).text) for path, row in self.fetch_many( to_fetch, worker, max_workers=4).items(): if row is None: # fiche cassée : on ne bloque pas la source print(f"[sorti-ka] cantonsdelest : fiche ignorée {path}") continue cache[path] = {"fetched_at": now, "row": row} self.save_cache(cache) # émission : la carte (titre, catégorie, ville, date, image) porte # chaque occurrence ; la fiche (cache) enrichit description, adresse, # code postal, GPS et précise les dates (JSON-LD de l'occurrence) events: list[Event] = [] seen: set[str] = set() for card in cards: if card["external_id"] in seen: continue seen.add(card["external_id"]) fiche = (cache.get(card["path"]) or {}).get("row") or {} events.append(Event( source=self.source_id, external_id=card["external_id"], url=card["url"], title=card["title"] or fiche.get("title", ""), description=fiche.get("description", ""), raw_categories=[c for c in card["raw_categories"] if c] or fiche.get("raw_categories") or [], address=fiche.get("address", ""), city=card["city"] or fiche.get("city", ""), postal_code=fiche.get("postal_code", ""), tourist_region="Cantons-de-l'Est", lat=fiche.get("lat"), lng=fiche.get("lng"), start_date=fiche.get("start_date") or card["start_date"], end_date=fiche.get("end_date") or card["end_date"], image=card["image"] or fiche.get("image", ""), )) return events