# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/evenementsattractions.py : connecteur Événements Attractions Québec # Source : répertoire consommateur des festivals/événements membres d'EAQ, # publié sur quebecvacances.com (~350 fiches fr, toute la province). # (Le site corporatif evenementsattractions.quebec ne liste que # des formations aux membres — sondé 2026-08-25, sans intérêt.) # Extraction: sitemap paginé (/sitemap.xml?page=1..N) → fiches # /fr/evenements/, chacune avec JSON-LD Event complet # (nom, description, dates ISO de la prochaine édition, lieu, # adresse, image Cloudinary). Le site est derrière un anti-bot # (403 en GET direct) → chaque GET passe par la chaîne de secours # du BaseConnector (Oxylabs → Scrapfly ASP → Bright Data), # d'où un cache INCRÉMENTAL agressif (data/ # evenementsattractions_cache.json) : nouvelles fiches d'abord, # re-visite après 14 jours (festivals annuels stables), plafond # EAQ_DETAIL_MAX fiches/sync. # Accès : quebecvacances.com/robots.txt « User-agent: * / Allow: / » → # crawl permis ; rate-limiting poli, identification honnête. # Prix : non publié dans le JSON-LD → is_free inconnu (jamais inventé). # ----------------------------------------------------------------------------- from __future__ import annotations import json import os import re import time from ..normalize import clean_text from ..schema import Event from .base import BaseConnector BASE = "https://www.quebecvacances.com" SITEMAP_PAGES = 4 # /sitemap.xml?page=1..4 (sondé 2026-08-25) DETAIL_MAX = int(os.environ.get("EAQ_DETAIL_MAX", "80")) REFRESH_AFTER = 14 * 86400 # festivals annuels : fiches stables _LOC_RE = re.compile(r"([^<]+)") _LD_RE = re.compile(r'', re.S) # « Montréal (ville de) » / « Québec (ville de) » → nom de ville nu _CITY_SUFFIX_RE = re.compile(r"\s*\((?:ville|municipalité|village|paroisse|" r"canton|mrc)[^)]*\)\s*$", re.I) def _utf8_text(resp) -> str: """Texte de la réponse décodé en UTF-8 réel. Le site sert son HTML UTF-8 sans charset dans Content-Type ; certains backends de la chaîne de secours anti-bot décodent alors en latin-1 → mojibake (« siège » au lieu de « siège », constaté live 2026-08-25). On répare de façon déterministe : latin-1 est bijectif octet↔caractère, donc ré-encoder puis décoder en UTF-8 restaure le texte d'origine. """ text = resp.text if "Ã" in text: # marqueur sûr de mojibake UTF-8/latin-1 try: return text.encode("latin-1").decode("utf-8") except (UnicodeEncodeError, UnicodeDecodeError): pass return text class EvenementsAttractionsConnector(BaseConnector): source_id = "evenementsattractions" request_delay = 1.0 # -- découverte (sitemap paginé) ------------------------------------------- def _fiches(self) -> list[str]: urls: list[str] = [] for page in range(1, SITEMAP_PAGES + 1): try: xml = self.get(f"{BASE}/sitemap.xml", params={"page": page}).text except Exception as exc: # une page cassée ne bloque pas le reste print(f"[sorti-ka] evenementsattractions : sitemap p{page} " f"ignorée : {exc}") continue urls += [u for u in _LOC_RE.findall(xml) if "/fr/evenements/" in u] return sorted(set(urls)) # -- fiche -------------------------------------------------------------------- def _parse_fiche(self, url: str, html: str) -> dict | None: data = None for blob in _LD_RE.findall(html): try: d = json.loads(blob.strip()) except ValueError: continue if d.get("@type") == "Event" and d.get("startDate"): data = d break if not data: return None title = clean_text(data.get("name") or "") if not title: return None loc = data.get("location") or {} addr = loc.get("address") or {} city = _CITY_SUFFIX_RE.sub("", clean_text(addr.get("addressLocality") or "")) venue = clean_text(loc.get("name") or "") if venue == title: # la source répète le nom de l'événement venue = "" return { "title": title, "description": clean_text(data.get("description") or "")[:2000], "raw_categories": [title], "venue": venue, "address": clean_text(addr.get("streetAddress") or ""), "city": city, "postal_code": addr.get("postalCode") or "", "start_date": str(data.get("startDate") or "")[:10] or None, "end_date": str(data.get("endDate") or "")[:10] or None, "website": data.get("url") or "", "image": data.get("image") or "", } # -- pipeline ------------------------------------------------------------------- def fetch(self) -> list[Event]: cache = self.load_cache() urls = self._fiches() if urls: # sitemap indisponible → on garde le cache current = set(urls) cache = {u: c for u, c in cache.items() if u in current} else: urls = list(cache) now = time.time() candidates = sorted( (u for u in urls if u not in cache or now - cache[u].get("fetched_at", 0) > REFRESH_AFTER), key=lambda u: cache.get(u, {}).get("fetched_at", 0))[:DETAIL_MAX] for url in candidates: # séquentiel : chaque GET peut escalader try: row = self._parse_fiche(url, _utf8_text(self.get(url))) except Exception as exc: # fiche cassée : on ne bloque pas la source print(f"[sorti-ka] evenementsattractions : fiche ignorée " f"{url} : {exc}") row = cache.get(url, {}).get("row") cache[url] = {"fetched_at": now, "row": row} self.save_cache(cache) events: list[Event] = [] seen: set[str] = set() for url, entry in cache.items(): row = entry.get("row") if not row: continue slug = url.rstrip("/").rsplit("/", 1)[-1] if slug in seen: continue seen.add(slug) events.append(Event( source=self.source_id, external_id=slug, url=url, title=row["title"], description=row.get("description", ""), raw_categories=row.get("raw_categories") or [], venue=row.get("venue", ""), address=row.get("address", ""), city=row.get("city", ""), postal_code=row.get("postal_code", ""), start_date=row.get("start_date"), end_date=row.get("end_date"), website=row.get("website", ""), image=row.get("image", ""), )) return events