# -----------------------------------------------------------------------------
# Sorti-Ka — Agrégateur de sorties & événements (province de Québec)
# Auteur : Simon-Pierre Boucher — contact@spboucher.ai
# connectors/shawinigan.py : connecteur Ville de Shawinigan — événements
# Source : https://www.shawinigan.ca/loisirs-et-culture/evenements/
# calendrier/ (WordPress + plugin maison « tri-evenements »).
# Extraction: archive rendue serveur (cartes « tri-evenement-single » :
# date FR, titre, extrait, image, url fiche datée
# /evenements/AAAA/MM/JJ//), pagination /calendrier/
# page/N. Enrichissement par fiche (date de fin « Date de
# fin : ... », lieu « Lieu : ... », heure si publiée) — CACHE
# INCRÉMENTAL disque, plafond SHA_FICHES par sync.
# Accès : site municipal public, robots.txt permissif (wp-admin
# seulement). GET directs throttlés, identification honnête.
# Prix : non publié par la source → jamais inventé.
# -----------------------------------------------------------------------------
from __future__ import annotations
import html as _html
import os
import re
from ..normalize import clean_text, parse_date_fr, parse_time
from ..schema import Event
from .base import BaseConnector
LIST_URL = "https://www.shawinigan.ca/loisirs-et-culture/evenements/calendrier/"
PAGES_MAX = int(os.environ.get("SHA_PAGES", "6"))
FICHES_MAX = int(os.environ.get("SHA_FICHES", "40"))
# cartes : « -single » (prochain événement) et « -list » (liste « À venir »)
_CARD_RE = re.compile(
r']*class="tri-evenement-evenement tri-evenement-(?:single|list)"[^>]*>(.*?)',
re.S)
_TITLE_RE = re.compile(
r'tri-evenement-titre">]*>(.*?)', re.S)
_DATE_RE = re.compile(r'tri-evenement-date">.*?\s*(.*?)\s*
', re.S)
_DESC_RE = re.compile(r'tri-evenement-desc">(.*?)', re.S)
_IMG_RE = re.compile(r'
]+src="([^"]+)"')
_URL_DATE_RE = re.compile(r"/evenements/(\d{4})/(\d{2})/(\d{2})/")
# fiche
_F_FIN_RE = re.compile(r"Date de fin\s*:\s*(.*?)\s*", re.S)
_F_LIEU_RE = re.compile(r"Lieu\s*:\s*(.*?)\s*", re.S)
_F_HEURE_RE = re.compile(r"Heure\s*:\s*(.*?)\s*", re.S)
class ShawiniganConnector(BaseConnector):
source_id = "shawinigan"
request_delay = 0.8
def _parse_page(self, html: str) -> list[dict]:
rows: list[dict] = []
for block in _CARD_RE.findall(html):
t = _TITLE_RE.search(block)
if not t:
continue
url = t.group(1)
start = None
m = _URL_DATE_RE.search(url) # URL datée = source fiable
if m:
start = f"{m.group(1)}-{m.group(2)}-{m.group(3)}"
else:
d = _DATE_RE.search(block)
start = parse_date_fr(clean_text(d.group(1))) if d else None
if not start:
continue
desc = _DESC_RE.search(block)
img = _IMG_RE.search(block)
rows.append({
"url": url,
"title": clean_text(_html.unescape(
re.sub(r"<[^>]+>", " ", t.group(2)))),
"start": start,
"description": clean_text(_html.unescape(re.sub(
r"<[^>]+>", " ", desc.group(1)))) if desc else "",
"image": img.group(1) if img else "",
})
return rows
def _fiche(self, url: str) -> dict:
html = self.get(url).text
out: dict = {}
m = _F_FIN_RE.search(html)
if m:
end = parse_date_fr(clean_text(re.sub(r"<[^>]+>", " ", m.group(1))))
if end:
out["end"] = end
m = _F_LIEU_RE.search(html)
if m:
out["venue"] = clean_text(_html.unescape(
re.sub(r"<[^>]+>", " ", m.group(1))))
m = _F_HEURE_RE.search(html)
if m:
t = parse_time(clean_text(re.sub(r"<[^>]+>", " ", m.group(1))))
if t:
out["start_time"] = t
return out
def fetch(self) -> list[Event]:
rows: list[dict] = []
seen_urls: set[str] = set()
for page in range(1, PAGES_MAX + 1):
url = LIST_URL if page == 1 else f"{LIST_URL}page/{page}/"
try:
batch = self._parse_page(self.get(url).text)
except Exception:
if page > 1: # fin de pagination (404)
break
raise
new = [r for r in batch if r["url"] not in seen_urls]
if not new:
break
seen_urls.update(r["url"] for r in new)
rows += new
cache = self.load_cache()
fetched = 0
for r in rows:
# « 2026/09/05/ » : la date de l'URL fait partie de la clé
key = "/".join(r["url"].rstrip("/").split("/")[-4:])
r["key"] = key
if key not in cache and fetched < FICHES_MAX:
try:
cache[key] = self._fiche(r["url"])
fetched += 1
except Exception as exc: # fiche cassée ≠ source cassée
print(f"[sorti-ka] shawinigan : fiche {key} ignorée : {exc}")
if fetched:
self.save_cache(cache)
events: list[Event] = []
seen: set[str] = set()
for r in rows:
ext = r["key"].replace("/", "-")
if ext in seen:
continue
seen.add(ext)
extra = cache.get(r["key"]) or {}
events.append(Event(
source=self.source_id,
external_id=ext,
url=r["url"],
title=r["title"],
description=r["description"],
raw_categories=[r["title"]],
venue=extra.get("venue", ""),
city="Shawinigan",
region="Mauricie",
start_date=r["start"],
start_time=extra.get("start_time"),
end_date=extra.get("end") or r["start"],
organizer="Ville de Shawinigan",
image=r["image"],
))
return events