# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/ticketmaster.py : connecteur Ticketmaster Canada (Québec) — SANS clé # Mode principal (sans clé) : scraping léger des pages découverte publiques # www.ticketmaster.ca/discover/?page=N (montreal, quebec-city, # gatineau). Chaque page embarque son état Next.js (__NEXT_DATA__) : la # requête serveur `cityEvents` y est déjà résolue avec 20 événements/page # (titre, date UTC + fuseau, salle géolocalisée lat/lng, adresse, code # postal, ville, URL billetterie, catégorie majeure, seatmap). Pagination # par ?page=N (vérifié 2026-08-18 : montreal total=1165, quebec-city=136). # Aucun prix publié dans ce flux → is_free/price_* inconnus (jamais # inventés). Artistes : patrons sûrs des titres (artists_from_title). # robots.txt : /discover/ non interdit (seuls /api/ismds/* le sont). # Anti-bot : GET direct d'abord ; si bloqué → Scrapfly ASP SANS render_js. # Plafonds : TM_MAX_EVENTS (défaut 360), TM_PAGES_PER_CITY (défaut 8). # Mode repli automatique (si TICKETMASTER_API_KEY apparaît dans .env) : # Discovery API officielle v2 — plus riche (priceRanges, attractions → # artistes, images 16:9). Les identifiants d'événements sont IDENTIQUES # dans les deux modes → aucun doublon lors d'un changement de mode. # ----------------------------------------------------------------------------- from __future__ import annotations import json import os from ..normalize import artists_from_title, utc_to_local from ..schema import Event, normalize_status from .base import BaseConnector # --- mode API officielle (avec clé) ------------------------------------------ API = "https://app.ticketmaster.com/discovery/v2/events.json" PAGE_SIZE = 200 DEEP_PAGE_LIMIT = 1000 # l'API refuse page*size >= 1000 (pagination profonde) # --- mode scraping (sans clé) ------------------------------------------------- DISCOVER = "https://www.ticketmaster.ca/discover/{city}?page={page}" CITIES = ("montreal", "quebec-city", "gatineau") # villes découverte QC publiées TM_MAX_EVENTS = int(os.environ.get("TM_MAX_EVENTS", "360")) TM_PAGES_PER_CITY = int(os.environ.get("TM_PAGES_PER_CITY", "8")) SCRAPE_PAGE_SIZE = 20 # taille fixe du flux cityEvents des pages découverte # catégories majeures Ticketmaster (segments Discovery) → libellés source _CAT_NAMES = { "KZFzniwnSyZfZ7v7nJ": "Concerts", "KZFzniwnSyZfZ7v7na": "Arts & Theatre", "KZFzniwnSyZfZ7v7nE": "Sports", "KZFzniwnSyZfZ7v7n1": "Family", "KZFzniwnSyZfZ7v7n7": "Film", } def _best_image(images: list[dict]) -> str: """Plus grande image 16:9 (sinon la plus large disponible).""" if not images: return "" ranked = sorted(images, key=lambda i: ((i.get("ratio") == "16_9"), i.get("width") or 0), reverse=True) return ranked[0].get("url") or "" class TicketmasterConnector(BaseConnector): source_id = "ticketmaster" request_delay = 1.0 # politesse scraping ; l'API tolère 5 req/s def __init__(self) -> None: super().__init__() self._force_scrapfly = False self.session.headers["Accept-Language"] = "fr-CA,fr;q=0.9,en;q=0.5" # ========================================================================== # Mode scraping SANS clé (pages découverte publiques) # ========================================================================== def _discover_html(self, url: str) -> str: """HTML d'une page découverte : GET direct, Scrapfly ASP en secours.""" if not self._force_scrapfly: try: html = self.get(url).text if "__NEXT_DATA__" in html: return html except Exception: pass if not os.environ.get("SCRAPFLY_API_KEY"): raise RuntimeError("GET direct bloqué et SCRAPFLY_API_KEY absent") self._force_scrapfly = True print("[sorti-ka] ticketmaster : GET direct bloqué → " "Scrapfly ASP (sans render_js)") return self.get_rendered(url, render_js=False) @staticmethod def _next_data(html: str) -> dict: i = html.find("__NEXT_DATA__") if i < 0: raise ValueError("__NEXT_DATA__ introuvable (anti-bot ?)") start = html.find(">", i) + 1 data, _ = json.JSONDecoder().raw_decode(html[start:].lstrip()) return data def _city_page(self, city: str, page: int) -> tuple[list[dict], int]: """Événements (hits) + total annoncé d'une page découverte.""" data = self._next_data(self._discover_html( DISCOVER.format(city=city, page=page))) queries = (((data.get("props") or {}).get("pageProps") or {}) .get("initialReduxState") or {}).get("api", {}).get("queries", {}) hits: list[dict] = [] total = 0 for key, q in queries.items(): if key.startswith("cityEvents(") and f'"page":{page}' in key: d = q.get("data") or {} hits.extend(d.get("events") or []) total = max(total, int(d.get("total") or 0)) return hits, total def _scrape_hit_to_event(self, hit: dict) -> Event | None: ext_id = hit.get("id") or "" title = hit.get("title") or "" if not ext_id or not title: return None if hit.get("virtual"): return None # Phase 2 : un événement annulé est CONSERVÉ avec status=cancelled # (bandeau côté frontend + eventStatus schema.org) au lieu d'être jeté. status = "cancelled" if hit.get("cancelled") else "" v = hit.get("venue") or {} dates = hit.get("dates") or {} # startDate publié en UTC + fuseau → date/heure locales Québec ; # heure retenue seulement si la source l'affiche (showDateTime, pas TBA) start_date, start_time = utc_to_local(dates.get("startDate")) if hit.get("tba") or dates.get("dateDisplay") != "showDateTime": start_time = None cat_id = (hit.get("majorCategory") or {}).get("id") or "" venue_name = v.get("name") or "" return Event( source=self.source_id, external_id=ext_id, url=hit.get("url") or "", title=title, raw_categories=[_CAT_NAMES[cat_id]] if cat_id in _CAT_NAMES else [], venue=venue_name, address=v.get("addressLineOne") or "", city=v.get("city") or "", postal_code=v.get("code") or "", lat=v.get("latitude"), lng=v.get("longitude"), start_date=start_date, start_time=start_time, end_date=start_date, status=status, artists=artists_from_title(title, venue_name), # seul visuel publié dans ce flux : le plan de salle officiel image=hit.get("seatmapUrl") or "", ) def _fetch_scrape(self) -> list[Event]: by_id: dict[str, Event] = {} for city in CITIES: for page in range(TM_PAGES_PER_CITY): if len(by_id) >= TM_MAX_EVENTS: break try: hits, total = self._city_page(city, page) except Exception as exc: print(f"[sorti-ka] ticketmaster : {city} page {page} " f"ignorée : {exc}") break if not hits: break for hit in hits: ev = self._scrape_hit_to_event(hit) if ev is not None: by_id.setdefault(ev.external_id, ev) if total and (page + 1) * SCRAPE_PAGE_SIZE >= total: break if len(by_id) >= TM_MAX_EVENTS: break return list(by_id.values()) # ========================================================================== # Mode API officielle (repli automatique dès qu'une clé est fournie) # ========================================================================== def _page(self, key: str, page: int) -> dict: return self.get_json(API, params={ "apikey": key, "countryCode": "CA", "stateCode": "QC", "locale": "fr-ca,en", "size": PAGE_SIZE, "page": page, "sort": "date,asc"}) def _hit_to_event(self, hit: dict) -> Event | None: ext_id = hit.get("id") or "" title = hit.get("name") or "" if not ext_id or not title: return None emb = hit.get("_embedded") or {} venues = emb.get("venues") or [{}] v = venues[0] loc = v.get("location") or {} dates = (hit.get("dates") or {}).get("start") or {} end_obj = (hit.get("dates") or {}).get("end") or {} end = end_obj.get("localDate") # statut officiel Discovery : onsale/offsale/cancelled/postponed/rescheduled status = normalize_status( ((hit.get("dates") or {}).get("status") or {}).get("code")) # classifications → libellés source (segment + genre + sous-genre) cats: list[str] = [] for c in hit.get("classifications") or []: for k in ("segment", "genre", "subGenre"): name = (c.get(k) or {}).get("name") or "" if name and name.lower() != "undefined" and name not in cats: cats.append(name) # fourchette de prix officielle (si publiée par la billetterie) price_min, price_label = None, "" ranges = [r for r in hit.get("priceRanges") or [] if r.get("min") is not None] if ranges: lo = min(float(r["min"]) for r in ranges) hi = max(float(r.get("max") or r["min"]) for r in ranges) price_min = lo if lo > 0 else None price_label = (f"{lo:g} $ à {hi:g} $ (Ticketmaster)" if hi > lo else f"{lo:g} $ (Ticketmaster)") artists = [a.get("name") for a in emb.get("attractions") or [] if a.get("name")] return Event( source=self.source_id, external_id=ext_id, url=hit.get("url") or "", title=title, description=hit.get("info") or hit.get("pleaseNote") or "", raw_categories=cats, venue=v.get("name") or "", address=((v.get("address") or {}).get("line1") or ""), city=((v.get("city") or {}).get("name") or ""), postal_code=v.get("postalCode") or "", lat=loc.get("latitude"), lng=loc.get("longitude"), start_date=dates.get("localDate"), start_time=(None if dates.get("timeTBA") or dates.get("noSpecificTime") else dates.get("localTime")), end_date=end or dates.get("localDate"), end_time=end_obj.get("localTime"), status=status, artists=artists, price_min=price_min, price_label=price_label, is_free=(False if price_min else None), organizer=((hit.get("promoter") or {}).get("name") or ""), image=_best_image(hit.get("images") or []), ) def _fetch_api(self, key: str) -> list[Event]: self.request_delay = 0.25 # quota officiel : 5 req/s events: list[Event] = [] page, total_pages = 0, 1 while page < total_pages and (page + 1) * PAGE_SIZE <= DEEP_PAGE_LIMIT: data = self._page(key, page) total_pages = int((data.get("page") or {}).get("totalPages") or 0) for hit in (data.get("_embedded") or {}).get("events") or []: ev = self._hit_to_event(hit) if ev is not None: events.append(ev) page += 1 return events # ========================================================================== def fetch(self) -> list[Event]: key = os.environ.get("TICKETMASTER_API_KEY", "").strip() if key: return self._fetch_api(key) # mode riche dès qu'une clé existe return self._fetch_scrape() # mode sans clé (par défaut)