\s*'
r'
None:
super().__init__()
self._host_last: dict[str, float] = {}
def _get_cp1252(self, url: str) -> str:
"""GET décodé cp1252 (ColdFusion), Crawl-delay: 120 honoré PAR HÔTE."""
host = url.split("/")[2]
wait = HOST_DELAY - (time.time() - self._host_last.get(host, 0.0))
if wait > 0:
time.sleep(wait)
resp = self.get(url)
self._host_last[host] = time.time()
return resp.content.decode("cp1252", errors="replace")
def _orgs_quebec(self) -> dict[str, dict]:
"""{sous-domaine: {nom, ville, region}} des organisations AU QUÉBEC."""
html = self._get_cp1252(PORTAL)
orgs: dict[str, dict] = {}
from ..regions import _k
for nom, ville, region, dom in _ORG_RE.findall(html):
if _k(region) in _EXCLUDE_REGIONS:
continue
orgs[dom] = {"nom": clean_text(nom), "ville": clean_text(ville),
"region": clean_text(region)}
return orgs
def _parse_detail(self, html: str) -> dict:
desc = _OG_DESC_RE.search(html)
# bloc coordonnées de la salle : « |La Chapelle| 620, avenue Plante| … »
plain = re.sub(r"<[^>]+>", "|", html)
plain = re.sub(r"[\t\r\n ]+", " ", plain)
addr = ""
for m in _ADDR_SEG_RE.finditer(plain):
seg = m.group(1)
if not _STREET_RE.search(seg):
continue # « 1h40 Récit… », « 5 BALLES… »
seg = clean_text(seg)
# la source suffixe souvent la ville (« …, Québec ») : on la
# retire si la fin n'a ni chiffre ni mot de voirie — l'Event
# porte déjà la ville, geocode.py préfère l'adresse nue
head, _, tail = seg.rpartition(",")
if head and not any(c.isdigit() for c in tail) \
and not _STREET_RE.search(tail):
seg = head.strip()
addr = seg
break
# « 0,00 $ » = ligne tarifaire vide (constaté live 2026-08-25, lsq) —
# jamais un prix plancher réel, on l'écarte (patron ticketpro c > 0)
prices = [float(p.replace(",", ".")) for p in _PRICE_RE.findall(plain)]
prices = [p for p in prices if p > 0]
return {
"description": clean_text(desc.group(1)) if desc else "",
"address": addr,
"price_min": min(prices) if prices else None,
}
def fetch(self) -> list[Event]:
cache = self.load_cache()
now = time.time()
meta = cache.get("_meta") or {}
if now - meta.get("ts", 0) <= TTL and cache.get("_rows"):
return self._emit(cache) # cache frais : aucun réseau
orgs = self._orgs_quebec()
if not orgs:
raise RuntimeError("ticketacces : annuaire du portail vide")
rows: dict[str, dict] = {} # clé de séance → ligne
for cat, label in CATEGORIES.items():
try:
html = self._get_cp1252(CALENDAR.format(cat=cat))
except Exception:
print(f"[sorti-ka] ticketacces : calendrier {label} inaccessible")
continue
images = dict(_IMG_RE.findall(html))
for org, evid, title, subtitle, date_raw, place in _ROW_RE.findall(html):
if org not in orgs: # hors Québec ou org inconnue
continue
date_iso = parse_date_fr(date_raw)
heure = parse_time(date_raw)
if not date_iso:
continue
venue, city = "", orgs[org]["ville"]
place = clean_text(place)
if "," in place:
venue, city = (p.strip() for p in place.rsplit(",", 1))
elif place:
venue = place
key = f"{org}-{evid}-{date_iso}-{heure or ''}"
if key in rows: # le HTML responsive duplique
continue
rows[key] = {
"external_id": key,
"org": org, "evid": evid,
"url": (f"https://{org}.ticketacces.net/fr/organisation/"
f"representations/index.cfm?EvenementID={evid}"),
"title": clean_text(title),
"subtitle": clean_text(subtitle or ""),
"raw_category": label,
"venue": venue, "city": city,
"region": orgs[org]["region"],
"organizer": orgs[org]["nom"],
"start_date": date_iso, "start_time": heure,
"image": images.get(evid, ""),
}
# fiches détail incrémentales (description, adresse, prix plancher) —
# Crawl-delay par hôte : au plus UNE fiche par organisation par passage
fiches = {f"{r['org']}-{r['evid']}": (r["org"], r["url"])
for r in rows.values()}
details = {k: v for k, v in cache.items()
if k not in ("_meta", "_rows") and k in fiches}
# jamais vue > parseur périmé (v < DETAIL_V, patron lavitrine) > plus
# ancienne — le bump de DETAIL_V re-visite progressivement le stock
stale = sorted(
(k for k in fiches
if k not in details
or details[k].get("v", 1) < DETAIL_V
or now - details[k].get("ts", 0) > REFRESH_AFTER),
key=lambda k: (details.get(k, {}).get("v", 1) if k in details else 0,
details.get(k, {}).get("ts", 0)))
seen_hosts: set[str] = set()
picked: list[str] = []
for k in stale:
org = fiches[k][0]
if org in seen_hosts:
continue
seen_hosts.add(org)
picked.append(k)
if len(picked) >= DETAIL_MAX:
break
for k in picked:
try:
detail = self._parse_detail(self._get_cp1252(fiches[k][1]))
except Exception:
detail = {}
details[k] = {"ts": now, "v": DETAIL_V, "detail": detail}
cache = dict(details)
cache["_meta"] = {"ts": now}
cache["_rows"] = list(rows.values())
self.save_cache(cache)
return self._emit(cache)
def _emit(self, cache: dict) -> list[Event]:
events: list[Event] = []
for r in cache.get("_rows", []):
detail = cache.get(f"{r['org']}-{r['evid']}", {}).get("detail") or {}
title = r["title"]
if r["subtitle"]:
title = f"{title} — {r['subtitle']}"
events.append(Event(
source=self.source_id,
external_id=r["external_id"],
url=r["url"],
title=title,
description=detail.get("description", ""),
raw_categories=[r["raw_category"]],
venue=r["venue"],
address=detail.get("address", ""),
city=r["city"],
region=r["region"], # gardée si déjà administrative
tourist_region=r["region"], # sinon Charlevoix/Bellechasse…
start_date=r["start_date"],
start_time=r["start_time"],
price_min=detail.get("price_min"),
organizer=r["organizer"],
image=r["image"],
))
return events