# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/eventbrite.py : connecteur Eventbrite (Québec) — SANS clé # Source : eventbrite.ca — billetterie libre-service (communautaire, # ateliers, festivals, affaires, vie nocturne...). # Extraction: pages de découverte publiques /d//all-events/?page=N — # chaque page embarque window.__SERVER_DATA__ avec les résultats # de recherche COMPLETS (20 événements/page) : nom, résumé, # date + heure locales, salle géocodée (lat/lng, adresse, code # postal), image, catégories (tags Eventbrite*), URL. Vérifié # 2026-08-18 : Montréal 3 341 événements (49 pages exposées). # Prix : fiches détail /e/ incrémentales (JSON-LD # AggregateOffer lowPrice/highPrice CAD) avec cache disque # data/eventbrite_cache.json, plafond EB_DETAIL_MAX fiches/sync # (défaut 150), re-visite 7 jours — patron La Vitrine. # Accès : robots.txt « User-agent: * » n'interdit ni /d/ ni /e/ (seuls # /directory/, checkout et endpoints privés le sont) — GET # directs throttlés, identification honnête, lien direct vers # chaque fiche Eventbrite. Filtre region=QC ; en ligne exclus. # Prix : uniquement l'AggregateOffer publié par Eventbrite — jamais # inventé (gratuit seulement si high == 0). # ----------------------------------------------------------------------------- from __future__ import annotations import json import os import re import time from ..schema import Event from .base import BaseConnector # lieux QC crawlés (slug découverte, budget de pages de liste par sync) — # pondération par taille de marché ; ~30 GET de liste/sync ≈ 600 fiches PLACES: tuple[tuple[str, int], ...] = ( ("canada--montreal", 10), ("canada--qu%C3%A9bec", 6), # ville de Québec ("canada--laval", 3), ("canada--gatineau", 3), ("canada--sherbrooke", 3), ("canada--longueuil", 2), ("canada--trois-rivi%C3%A8res", 2), ) LIST_URL = "https://www.eventbrite.ca/d/{place}/all-events/?page={page}" EB_DETAIL_MAX = int(os.environ.get("EB_DETAIL_MAX", "150")) # fiches prix/sync REFRESH_AFTER = 7 * 86400 # re-visite d'une fiche prix (secondes) _LD_JSON_RE = re.compile(r"application/ld\+json[^>]*>") def _server_data(html: str) -> dict: i = html.find("__SERVER_DATA__") if i < 0: raise ValueError("__SERVER_DATA__ introuvable") start = html.find("{", i) data, _ = json.JSONDecoder().raw_decode(html[start:]) return data def _parse_detail(html: str) -> dict: """JSON-LD d'une fiche : AggregateOffer (lowPrice/highPrice/priceCurrency) + organisateur (organizer.name) + description complète — même requête.""" price, organizer, description = None, "", "" 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 not isinstance(doc, dict): continue org = doc.get("organizer") if isinstance(org, dict) and org.get("name") and not organizer: organizer = str(org["name"]) desc = doc.get("description") if isinstance(desc, str) and len(desc) > len(description): description = desc[:2000] offers = doc.get("offers") if not offers or price is not None: continue for off in (offers if isinstance(offers, list) else [offers]): if not isinstance(off, dict) or off.get("@type") != "AggregateOffer": continue try: low = float(off.get("lowPrice")) high = float(off.get("highPrice") or off.get("lowPrice")) except (TypeError, ValueError): continue price = {"low": low, "high": high, "cur": off.get("priceCurrency") or ""} break return {"price": price, "organizer": organizer, "description": description} class EventbriteConnector(BaseConnector): source_id = "eventbrite" request_delay = 0.8 # -- pages de liste -------------------------------------------------------- def _list_results(self) -> list[dict]: results: list[dict] = [] for place, budget in PLACES: page_count = budget for page in range(1, budget + 1): if page > page_count: break try: data = _server_data(self.get( LIST_URL.format(place=place, page=page)).text) ev = ((data.get("search_data") or {}).get("events") or {}) page_count = min(budget, int( (ev.get("pagination") or {}).get("page_count") or 0)) results.extend(ev.get("results") or []) except Exception as exc: # un lieu cassé ne bloque pas la source print(f"[sorti-ka] eventbrite : {place} page {page} " f"ignorée : {exc}") break return results # -- fiche → Event --------------------------------------------------------- def _result_to_event(self, r: dict, detail: dict | None) -> Event | None: ext_id = str(r.get("id") or r.get("eventbrite_event_id") or "") title = r.get("name") or "" if not ext_id or not title: return None if r.get("is_online_event"): return None venue = r.get("primary_venue") or {} addr = venue.get("address") or {} if (addr.get("region") or "") != "QC": # filtre Québec strict return None # statut billetterie (Phase 2) : annulé (is_cancelled — l'événement est # conservé et affiché barré/bandeau) ; complet via urgency_signals # (catégorie/message « sold out » explicite seulement, jamais déduit). status = "cancelled" if r.get("is_cancelled") else "" if not status: us = r.get("urgency_signals") or {} signals = " ".join([*(us.get("messages") or []), *(us.get("categories") or [])]).lower() if "sold" in signals and "out" in signals: status = "soldout" price = (detail or {}).get("price") organizer = (detail or {}).get("organizer") or "" # description : full_description (vide dans le flux de liste, constaté # 2026-08-19) > JSON-LD de la fiche détail > summary (140 caractères) description = (r.get("full_description") or (detail or {}).get("description") or r.get("summary") or "") cats: list[str] = [] for t in r.get("tags") or []: if (t.get("prefix") or "").startswith("Eventbrite"): dn = ((t.get("localized") or {}).get("display_name") or t.get("display_name") or "") if dn and dn.lower() != "other" and dn not in cats: cats.append(dn) img = (((r.get("image") or {}).get("original") or {}).get("url") or (r.get("image") or {}).get("url") or "") is_free, price_min, label = None, None, "" if price and price.get("cur") in ("CAD", ""): low, high = price["low"], price["high"] if high <= 0: is_free, label = True, "Gratuit (Eventbrite)" elif low <= 0: # options gratuites ET payantes label = f"0 $ à {high:g} $ (Eventbrite)" else: is_free, price_min = False, low label = (f"{low:g} $ à {high:g} $ (Eventbrite)" if high > low else f"{low:g} $ (Eventbrite)") return Event( source=self.source_id, external_id=ext_id, url=r.get("url") or "", title=title, description=description, raw_categories=cats, venue=venue.get("name") or "", address=addr.get("address_1") or "", city=addr.get("city") or "", postal_code=addr.get("postal_code") or "", lat=addr.get("latitude"), lng=addr.get("longitude"), start_date=r.get("start_date"), start_time=r.get("start_time"), end_date=r.get("end_date"), end_time=r.get("end_time"), status=status, is_free=is_free, price_min=price_min, price_label=label, organizer=organizer, image=img, ) # -- enrichissement prix + organisateur (incrémental, plafonné) ------------ def _enrich_details(self, rows: list[dict]) -> dict: """Cache {event_id: {fetched_at, price|None, organizer}} ; fiches jamais visitées d'abord, puis celles d'avant la Phase 2 (sans clé organizer), puis re-visite roulante après 7 jours.""" cache = self.load_cache() now = time.time() current = {str(r.get("id")): r for r in rows if r.get("id")} cache = {k: v for k, v in cache.items() if k in current} candidates = sorted( (eid for eid, r in current.items() if r.get("url") and (eid not in cache or "organizer" not in cache[eid] or "description" not in cache[eid] or now - cache[eid].get("fetched_at", 0) > REFRESH_AFTER)), key=lambda eid: cache.get(eid, {}).get("fetched_at", 0))[:EB_DETAIL_MAX] url_to_id = {current[eid]["url"]: eid for eid in candidates} def worker(url, session): return _parse_detail(session.get(url, timeout=20).text) for url, detail in self.fetch_many(list(url_to_id), worker, max_workers=4).items(): entry = {"fetched_at": now} | (detail or { "price": None, "organizer": "", "description": ""}) cache[url_to_id[url]] = entry self.save_cache(cache) return cache # -- interface -------------------------------------------------------------- def fetch(self) -> list[Event]: rows = self._list_results() # dédoublonnage inter-lieux (un même événement sort dans plusieurs villes) uniq: dict[str, dict] = {} for r in rows: eid = str(r.get("id") or "") if eid: uniq.setdefault(eid, r) details = self._enrich_details(list(uniq.values())) events: list[Event] = [] for eid, r in uniq.items(): ev = self._result_to_event(r, details.get(eid)) if ev is not None: events.append(ev) return events