# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/bandsintown.py : connecteur Bandsintown (API publique par artiste) # Source : rest.bandsintown.com/artists/{artiste}/events — API publique # (app_id d'identification ; le format « js_ » des # widgets officiels est accepté, testé 2026-08 — un app_id # arbitraire simple est refusé « not authorized »). # Approche : l'API ne permet pas de recherche géographique → on interroge # les artistes déjà connus de la base (colonne `artists` extraite # d'evenko / lepointdevente / ticketmaster, événements à venir) # et on ne garde que leurs dates au QUÉBEC (venue.region = QC). # Volume : plafond BIT_ARTIST_MAX artistes/sync (défaut 100), rotation par # ancienneté via cache disque (data/bandsintown_cache.json) ; les # événements des artistes déjà sondés sont ré-émis depuis le # cache entre deux visites (re-visite après 3 jours). # Champs : datetime local (date + heure), salle géolocalisée (lat/lng), # lineup → artistes, `free` → is_free, offres/billets → url, # image de l'artiste. L'objet `artist` n'est renseigné que sur # CERTAINS événements de la réponse (dict vide sur d'autres — # constaté 2026-08-25, cause de l'image à 8 %) : l'image est # reprise de n'importe quel événement de la réponse, sinon du # profil /artists/{artiste} (1 GET de plus, seulement si # l'artiste a des dates QC sans image). # ----------------------------------------------------------------------------- from __future__ import annotations import os import time import urllib.parse from ..schema import Event from .base import BaseConnector API = "https://rest.bandsintown.com/artists/{artist}/events" APP_ID = os.environ.get("BANDSINTOWN_APP_ID", "js_www.sorti-ka.com") ARTIST_MAX = int(os.environ.get("BIT_ARTIST_MAX", "100")) # artistes/sync REFRESH_AFTER = 3 * 86400 # re-visite d'un artiste (tournées ajoutées) _QC_REGIONS = {"qc", "quebec", "québec"} def _known_artists() -> list[str]: """Artistes des événements à venir déjà en base (sources billetterie).""" import json from datetime import date from .. import db today = date.today().isoformat() con = db.connect() rows = con.execute( "SELECT artists FROM events WHERE active=1 AND artists IS NOT NULL " "AND artists != '[]' AND (end_date >= ? OR start_date >= ?) " "AND source IN ('evenko','lepointdevente','ticketmaster')", (today, today)).fetchall() con.close() seen, out = set(), [] for r in rows: for a in json.loads(r["artists"] or "[]"): k = a.strip().lower() if k and k not in seen: seen.add(k) out.append(a.strip()) return out class BandsintownConnector(BaseConnector): source_id = "bandsintown" request_delay = 0.5 def _artist_events(self, artist: str) -> list[dict]: """Dates québécoises à venir d'un artiste (champs utiles seulement).""" url = API.format(artist=urllib.parse.quote(artist, safe="")) data = self.get_json(url, params={"app_id": APP_ID, "date": "upcoming"}) if not isinstance(data, list): # {"Message": …} = refus/erreur raise RuntimeError(f"réponse inattendue pour {artist!r} : {data}") # l'objet `artist` (image_url) n'est renseigné que sur certains # événements de la réponse — on prend la 1ʳᵉ image trouvée n'importe où artist_img = next( (((e.get("artist") or {}).get("image_url") or "") for e in data if (e.get("artist") or {}).get("image_url")), "") rows = [] for e in data: venue = e.get("venue") or {} region = (venue.get("region") or "").strip().lower() country = (venue.get("country") or "").strip().lower() if country != "canada" or region not in _QC_REGIONS: continue dt = e.get("datetime") or "" # heure locale de la salle art = e.get("artist") or {} # offres billetterie : lien billets direct + statut de vente # (type Tickets ; status "available" / "sold out" chez la source) ticket_url, offer_status = "", "" for off in e.get("offers") or []: if (off.get("type") or "").lower() == "tickets" and off.get("url"): ticket_url = off["url"] offer_status = (off.get("status") or "").lower() break rows.append({ "id": str(e.get("id") or ""), "url": ticket_url or e.get("url") or "", "offer_status": offer_status, "title": e.get("title") or "", "description": e.get("description") or "", "artist_name": art.get("name") or artist, "lineup": [a for a in e.get("lineup") or [] if a], "venue": venue.get("name") or "", "address": venue.get("street_address") or "", "city": venue.get("city") or "", "postal_code": venue.get("postal_code") or "", "lat": venue.get("latitude"), "lng": venue.get("longitude"), "date": dt[:10] or None, "time": dt[11:16] or None, "free": e.get("free"), "image": art.get("image_url") or "", }) # aucune image dans toute la réponse mais des dates QC à émettre : # 1 GET de repli sur le profil artiste (image_url canonique) if rows and not artist_img and not any(r["image"] for r in rows): try: prof = self.get_json( "https://rest.bandsintown.com/artists/" + urllib.parse.quote(artist, safe=""), params={"app_id": APP_ID}) if isinstance(prof, dict): artist_img = prof.get("image_url") or "" except Exception: # profil indisponible : on n'invente rien artist_img = "" for r in rows: if not r["image"]: r["image"] = artist_img return rows def fetch(self) -> list[Event]: artists = _known_artists() cache = self.load_cache() # rotation : jamais sondés d'abord, puis les plus anciens ; on garde en # cache les artistes toujours présents en base (sinon purge douce) keyed = {a.lower(): a for a in artists} cache = {k: v for k, v in cache.items() if k in keyed} now = time.time() todo = sorted( (k for k in keyed if now - (cache.get(k) or {}).get("fetched_at", 0) > REFRESH_AFTER), key=lambda k: (cache.get(k) or {}).get("fetched_at", 0))[:ARTIST_MAX] for k in todo: try: cache[k] = {"fetched_at": now, "events": self._artist_events(keyed[k])} except Exception as exc: # un artiste raté ne bloque pas la source print(f"[sorti-ka] bandsintown : artiste ignoré {keyed[k]!r} : {exc}") cache.setdefault(k, {"fetched_at": 0, "events": []}) cache[k]["fetched_at"] = now # ne pas boucler dessus self.save_cache(cache) events, seen = [], set() for entry in cache.values(): for row in entry.get("events") or []: if not row.get("id") or row["id"] in seen: continue seen.add(row["id"]) title = row["title"] or row["artist_name"] events.append(Event( source=self.source_id, external_id=row["id"], url=row["url"] or "https://www.bandsintown.com", title=title, description=row.get("description") or "", raw_categories=["Concert"], venue=row.get("venue", ""), address=row.get("address", ""), city=row.get("city", ""), postal_code=row.get("postal_code", ""), lat=row.get("lat"), lng=row.get("lng"), start_date=row.get("date"), start_time=(None if row.get("time") == "00:00" else row.get("time")), end_date=row.get("date"), # statut de vente de l'offre (« sold out » explicite) status=("soldout" if "sold" in (row.get("offer_status") or "") else ""), artists=row.get("lineup") or [row["artist_name"]], # `free` vaut false par défaut chez Bandsintown : seul un # true explicite est significatif (on n'invente rien) is_free=(True if row.get("free") is True else None), image=row.get("image", ""), )) return events