# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/lavitrine.py : connecteur La Vitrine culturelle (lavitrine.com) # Source : billetterie/calendrier culturel — spectacles, expositions, # concerts (Grand Montréal + Québec + tournées). # Extraction: sitemap.xml public (~2 000 fiches /fr/evenement/ + 185 # /fr/exposition/, lastmod par URL) puis pages fiche rendues # serveur (h1, bloc .event_calendar : jour/mois/année, HEURE # (« 15:00 HNE » dans la slide), salle, ville ; og:image ; # description = event_tabs_description-short + about-description). # GET directs — pas d'anti-bot. # INCRÉMENTAL avec cache disque (data/lavitrine_cache.json) : # nouvelles fiches d'abord, re-visite roulante après 7 jours, # plafonné à LV_MAX_FETCH fiches/sync. # Accès : robots.txt « User-Agent: * / Allow: / » + Content-Signal # « search=yes, use=reference » → usage index + lien vers la # fiche originale, conforme aux signaux publiés (CLAUDE.md §15). # Prix/GPS : les pages FICHE ne publient ni tarif, ni JSON-LD, ni adresse. # MAIS le sitemap contient aussi ~1 000 pages « représentation » # (/fr/evenement///) qui publient un # JSON-LD Event complet (vérifié live 2026-08-25, 14/14 pages : # Place name + PostalAddress + GeoCoordinates 100 %, offers.price # ~80 %) → parseur v3 : adresse civique, code postal, lat/lng et # prix billetterie attachés à LA représentation correspondante # (identifier = dateId de la slide), sans AUCUN GET de plus (ces # pages étaient déjà téléchargées et ignorées). Les autres # représentations restent au géocodage aval salle+ville # (sortika/venues.py + geocode.py). isAccessibleForFree vaut # false par défaut chez la source → jamais utilisé seul ; # is_free=False seulement si un prix > 0 est publié. # ----------------------------------------------------------------------------- from __future__ import annotations import re import time from ..normalize import clean_text from ..schema import Event from .base import BaseConnector SITEMAP = "https://www.lavitrine.com/sitemap.xml" LV_MAX_FETCH = int(__import__("os").environ.get("LV_MAX_FETCH", "400")) # fiches/sync (GET directs polis) REFRESH_AFTER = 7 * 86400 # re-visite roulante d'une fiche (secondes) PARSE_V = 3 # v2 : heure + description ; v3 : JSON-LD des pages représentation # (adresse/GPS/prix) → backfill prioritaire des entrées < v3 _LOC_RE = re.compile(r"([^<]+)\s*(?:([^<]+))?") _URL_BLOCK_RE = re.compile(r"(.*?)", re.S) _H1_RE = re.compile(r"]*>(.*?)", re.S) _OG_IMG_RE = re.compile(r'property="og:image" content="([^"]*)"') _MONTHS = {"jan": 1, "fév": 2, "fev": 2, "mar": 3, "avr": 4, "mai": 5, "juin": 6, "juil": 7, "aoû": 8, "aou": 8, "août": 8, "sep": 9, "oct": 10, "nov": 11, "déc": 12, "dec": 12} # une « slide » du calendrier : jour / mois / année, puis salle / ville _DATE_RE = re.compile( r'calender-date-text-large">(\d{1,2})' r'
([^<]+)
' r'
(\d{4})
') _PLACE_RE = re.compile(r'class="text-style-2lines">([^<]*)
([^<]*)
') # l'heure de la représentation, dans la même slide, entre la date et la salle : #
15:00 HNE
_TIME_RE = re.compile(r'class="text-size-small[^"]*">\s*(\d{1,2}[:h]\d{2})') # description de la fiche : accroche courte + début du texte long (onglet À propos) _DESC_SHORT_RE = re.compile(r'event_tabs_description-short">(.*?)', re.S) _DESC_LONG_RE = re.compile(r'id="about-description">(.*?)', re.S) _SLIDE_A_RE = re.compile( r']*href="([^"]*)"[^>]*class="event_calendar_slide-content(.*?)', re.S) _LD_SCRIPT_RE = re.compile( r'', re.S) def _ld_representations(html: str) -> dict[str, dict]: """JSON-LD Event des pages « représentation » du sitemap (…//) : {dateId → salle, adresse civique, code postal, lat/lng, prix billetterie}. `identifier` = le dateId de la slide correspondante du carrousel. Les pages fiche n'en publient pas (vérifié live 2026-08-25) → dict vide, aucun impact.""" import json out: dict[str, dict] = {} for m in _LD_SCRIPT_RE.finditer(html): raw = m.group(1) try: data = json.loads(raw) except Exception: try: # caractères de contrôle bruts dans les descriptions data = json.loads(re.sub(r"[\x00-\x1f]", " ", raw)) except Exception: continue for d in (data if isinstance(data, list) else [data]): if not isinstance(d, dict) or d.get("@type") != "Event": continue ident = str(d.get("identifier") or "") if not ident: continue loc = d.get("location") or {} adr = loc.get("address") or {} geo = loc.get("geo") or {} off = d.get("offers") or {} if isinstance(off, list): off = off[0] if off else {} try: price = float(off.get("price")) except (TypeError, ValueError): price = None out[ident] = { "venue": clean_text(loc.get("name") or ""), "address": clean_text(adr.get("streetAddress") or ""), "city": clean_text(adr.get("addressLocality") or ""), "postal_code": clean_text(adr.get("postalCode") or ""), "lat": geo.get("latitude"), "lng": geo.get("longitude"), # prix publié par la billetterie elle-même (souvent null tant # que la vente n'est pas ouverte : jamais inventé) "price": price if price and price > 0 else None, } return out # date + heure LOCALES de la représentation courante, publiées dans le fil # d'Ariane JSON-LD (« 17 sept 2026 | 20:00 HAE ») — le startDate de l'Event # est en UTC et peut donc changer de jour, on ne l'utilise JAMAIS. _CRUMB_DATE_RE = re.compile( r'"name":\s*"(\d{1,2})\s+([a-zA-Zûéà]+)\.?\s+(\d{4})' r'(?:\s*\|\s*(\d{1,2}:\d{2}))?[^"]*"') def _month_num(month_txt: str) -> int | None: t = clean_text(month_txt).lower() return _MONTHS.get(t[:4].rstrip(".")) or _MONTHS.get(t[:3]) def _slides(html: str) -> list[tuple]: """Slides du calendrier : (id de fiche de la slide, id de représentation « dateId », url canonique, jour, mois, année, salle, ville, heure). Le carrousel d'une fiche de tournée liste TOUTES les dates de la tournée, chacune avec l'id de SA fiche (/fr/evenement///). On garde tout mais on identifie chaque représentation par l'id de sa slide : la même date vue depuis plusieurs fiches → même identifiant → aucun doublon à l'ingestion, aucune date perdue. Le dateId (dernier segment) sert de clé de jointure vers le JSON-LD Event des pages représentation (identifier = dateId — v3, jamais utilisé dans external_id : zéro churn). """ out = [] for m in _SLIDE_A_RE.finditer(html): href, block = m.group(1), m.group(2) parts = href.strip("/").split("/") # …/evenement/// slide_id = parts[-2] if len(parts) >= 2 and parts[-2].isdigit() else "" date_id = parts[-1] if slide_id and parts[-1].isdigit() else "" canon = "https://www.lavitrine.com/" + "/".join(parts[:-1]) if slide_id else "" d = _DATE_RE.search(block) if not d: continue p = _PLACE_RE.search(block) t = _TIME_RE.search(block) out.append((slide_id, date_id, canon, d.group(1), d.group(2), d.group(3), p.group(1) if p else "", p.group(2) if p else "", t.group(1).replace("h", ":") if t else None)) return out class LaVitrineConnector(BaseConnector): source_id = "lavitrine" request_delay = 0.8 def _fiches(self) -> list[str]: xml = self.get(SITEMAP).text urls = [] for block in _URL_BLOCK_RE.findall(xml): m = _LOC_RE.search(block) if not m: continue url = m.group(1) if "/fr/evenement/" in url or "/fr/exposition/" in url: urls.append(url) return urls def _parse_fiche(self, url: str, html: str) -> list[dict]: h1 = _H1_RE.search(html) title = clean_text(h1.group(1)) if h1 else "" if not title: return [] img = _OG_IMG_RE.search(html) image = img.group(1) if img else "" # description : accroche courte de la fiche + début du texte long # (déjà dans le HTML téléchargé — Phase 2, gisement n° 1 de l'audit) d_short = _DESC_SHORT_RE.search(html) d_long = _DESC_LONG_RE.search(html) description = " ".join(p for p in ( clean_text(d_short.group(1)) if d_short else "", clean_text(d_long.group(1)) if d_long else "") if p)[:2000] ext = url.rstrip("/").rsplit("/", 1)[-1] # id numérique stable parts = url.rstrip("/").split("/") # dernier segment non numérique = slug lisible (les URLs représentation # …/// donnaient « 22040 » en catégorie brute) slug = next((p for p in reversed(parts) if not p.isdigit()), "") rows = [] for slide_id, date_id, canon, day, month_txt, year, venue, city, hhmm \ in _slides(html)[:60]: mo = _month_num(month_txt) if not mo: continue iso = f"{int(year):04d}-{mo:02d}-{int(day):02d}" rows.append({ # id de la slide (fiche propre à cette représentation) : la # même date vue depuis plusieurs fiches → même uid, zéro doublon "external_id": f"{slide_id or ext}-{iso}", "url": canon or url, "title": title, "image": image, "description": description, "raw_categories": [title, slug.replace("-", " ")], "venue": clean_text(venue or ""), "city": clean_text(city or ""), "start_date": iso, "end_date": iso, "start_time": hhmm, # heure de la slide (« 15:00 HNE ») }) # page « représentation » du sitemap (…//) : son # JSON-LD Event décrit la représentation COURANTE — jamais listée # dans son propre carrousel (vérifié live 2026-08-25) → on émet son # occurrence à part : adresse/GPS/prix du JSON-LD, date+heure LOCALES # du fil d'Ariane. À l'émission, cette version enrichie remplace la # slide nue vue depuis les autres fiches (même external_id). info = (_ld_representations(html).get(parts[-1]) if len(parts) >= 2 and parts[-1].isdigit() and parts[-2].isdigit() else None) crumb = _CRUMB_DATE_RE.search(html) if info else None mo = _month_num(crumb.group(2)) if crumb else None if info and crumb and mo: iso = f"{int(crumb.group(3)):04d}-{mo:02d}-{int(crumb.group(1)):02d}" price = info.get("price") rows.append({ "external_id": f"{parts[-2]}-{iso}", # même schéma que les slides "url": "/".join(parts[:-1]), # fiche canonique "title": title, "image": image, "description": description, "raw_categories": [title, slug.replace("-", " ")], "venue": info.get("venue", ""), "city": info.get("city", ""), "address": info.get("address", ""), "postal_code": info.get("postal_code", ""), "lat": info.get("lat"), "lng": info.get("lng"), "start_date": iso, "end_date": iso, "start_time": crumb.group(4), "price_min": price, "price_label": (f"{price:.2f} $ (billetterie lavitrine.com)" if price else ""), }) return rows def fetch(self) -> list[Event]: cache = self.load_cache() urls = self._fiches() current = set(urls) cache = {u: c for u, c in cache.items() if u in current} now = time.time() # nouvelles fiches d'abord, puis celles parsées avant PARSE_V # (v2 sans adresse/GPS/prix — montée progressive), puis les # plus anciennes visites (roulant) candidates = sorted( (u for u in urls if u not in cache or cache[u].get("v", 1) < PARSE_V or now - cache[u].get("fetched_at", 0) > REFRESH_AFTER), key=lambda u: (cache.get(u, {}).get("v", 1) if u in cache else 0, # pages représentation d'abord : seules à livrer # adresse/GPS/prix (JSON-LD v3) 0 if re.search(r"/\d+/\d+/?$", u) else 1, cache.get(u, {}).get("fetched_at", 0)))[:LV_MAX_FETCH] def worker(url, session): return self._parse_fiche(url, session.get(url, timeout=20).text) for url, rows in self.fetch_many(candidates, worker, max_workers=4).items(): if rows is None: # fiche cassée : on ne bloque pas la source print(f"[sorti-ka] lavitrine : fiche ignorée {url}") continue cache[url] = {"fetched_at": now, "rows": rows, "v": PARSE_V} self.save_cache(cache) # même slide vue depuis plusieurs pages (fiche de tournée + page # représentation) : la version enrichie JSON-LD (adresse) l'emporte best: dict[str, dict] = {} for entry in cache.values(): for row in entry.get("rows", []): cur = best.get(row["external_id"]) if cur is None or (not cur.get("address") and row.get("address")): best[row["external_id"]] = row events: list[Event] = [] for row in best.values(): events.append(Event( source=self.source_id, external_id=row["external_id"], url=row["url"], title=row["title"], description=row.get("description", ""), raw_categories=row.get("raw_categories") or [], venue=row.get("venue", ""), city=row.get("city", ""), address=row.get("address", ""), postal_code=row.get("postal_code", ""), lat=row.get("lat"), lng=row.get("lng"), start_date=row.get("start_date"), start_time=row.get("start_time"), end_date=row.get("end_date"), price_min=row.get("price_min"), price_label=row.get("price_label", ""), is_free=False if row.get("price_min") else None, image=row.get("image", ""), )) return events