# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/brossard.py : connecteur Ville de Brossard — calendrier des # événements municipaux (brossard.ca, Montérégie) # Source : API REST WordPress publique — type de contenu `city-event` # (~400 événements : bibliothèque, conférences, ateliers, # spectacles familiaux). Fraîcheur vérifiée 2026-08 (fiches # publiées en août 2026 pour des dates jusqu'en décembre 2026). # Extraction: 1) liste JSON /wp-json/wp/v2/city-event (paginée, champs # title/link/content/modified) ; 2) dates + heures par fiche : # bannière « Cet événement aura lieu le 6 décembre 2026, de # 15 h 00 à 16 h 00 » (fiches ville) ou JSON-LD Event # (fiches biblio.brossard.ca, redirection suivie) — cache disque # incrémental data/brossard_cache.json, plafond # BROSSARD_DETAIL_MAX fiches/sync, re-visite 7 jours. # Accès : API REST WordPress publique (aucune authentification), GET # throttlés, identification honnête. robots.txt : Allow. # Prix : non publié par la source → jamais inventé. # ----------------------------------------------------------------------------- from __future__ import annotations import json import os import re import time from ..normalize import clean_text, parse_date_fr, parse_time from ..schema import Event from .base import BaseConnector API = "https://brossard.ca/wp-json/wp/v2/city-event" PAGE_SIZE = 100 DETAIL_MAX = int(os.environ.get("BROSSARD_DETAIL_MAX", "80")) # fiches/sync REFRESH_AFTER = 7 * 86400 _LD_RE = re.compile(r'', re.S) _OG_IMG_RE = re.compile(r'property="og:image" content="([^"]*)"') # « Cet événement aura lieu le 6 décembre 2026, de 15 h 00 à 16 h 00 » # « … aura lieu le 6 décembre 2026, à 15 h 00 » # « … aura lieu du 3 juillet 2026 au 5 juillet 2026 » _BANNER_RE = re.compile( r"aura lieu\s+(?:le\s+(?P\d{1,2}(?:er)?\s+\S+\s+\d{4})" r"(?:,?\s*de\s+(?P\d{1,2}\s*h\s*(?:\d{2})?)" r"\s*(?:à|-|–)\s*(?P\d{1,2}\s*h\s*(?:\d{2})?)" r"|,?\s*à\s+(?P\d{1,2}\s*h\s*(?:\d{2})?))?" r"|du\s+(?P\d{1,2}(?:er)?\s+\S+\s+\d{4})\s+au\s+" r"(?P\d{1,2}(?:er)?\s+\S+\s+\d{4}))", re.I) class BrossardConnector(BaseConnector): source_id = "brossard" request_delay = 0.8 # -- liste REST --------------------------------------------------------- def _list(self) -> list[dict]: out, page = [], 1 while True: try: batch = self.get_json(API, params={ "per_page": PAGE_SIZE, "page": page, "orderby": "modified", "order": "desc", "_fields": "id,link,title,content,modified,type-event"}) except Exception: if page > 1: # « rest_post_invalid_page_number » = fin break raise if not isinstance(batch, list) or not batch: break out += batch if len(batch) < PAGE_SIZE: break page += 1 return out # -- taxonomie type-event : id → nom (1 requête, ~11 termes) --------------- def _type_event_map(self) -> dict[int, str]: """Vraies catégories WordPress (« Bibliothèque — adultes », « Événements de la Ville »…) — remplace l'heuristique par titre.""" try: terms = self.get_json( "https://brossard.ca/wp-json/wp/v2/type-event", params={"per_page": 100, "_fields": "id,name"}) return {int(t["id"]): clean_text(t.get("name") or "") for t in terms if t.get("id")} except Exception as exc: # taxonomie indisponible ≠ source cassée print(f"[sorti-ka] brossard : taxonomie type-event ignorée : {exc}") return {} # -- fiche : dates/heure/lieu (bannière ville OU JSON-LD biblio) ---------- def _parse_fiche(self, html: str) -> dict: info: dict = {} for block in _LD_RE.findall(html): if '"Event"' not in block: continue try: ld = json.loads(block) except ValueError: continue if isinstance(ld, list): ld = ld[0] if ld else {} if ld.get("@type") != "Event": continue start, end = str(ld.get("startDate") or ""), str(ld.get("endDate") or "") info["start_date"] = start[:10] or None info["end_date"] = end[:10] or None info["time"] = parse_time(start) info["end_time"] = parse_time(end) loc = ld.get("location") or [] loc = loc[0] if isinstance(loc, list) and loc else loc if isinstance(loc, dict): info["venue"] = clean_text(str(loc.get("name") or "")) addr = loc.get("address") or {} if isinstance(addr, dict): info["address"] = clean_text(str(addr.get("streetAddress") or "")) info["postal_code"] = clean_text(str(addr.get("postalCode") or "")) if ld.get("image"): info["image"] = str(ld["image"]) return info m = _BANNER_RE.search(clean_text(html[:200000])) if m: if m.group("jour"): info["start_date"] = parse_date_fr(m.group("jour")) info["end_date"] = info["start_date"] info["time"] = parse_time(m.group("heure") or m.group("heure2") or "") info["end_time"] = parse_time(m.group("fin") or "") else: info["start_date"] = parse_date_fr(m.group("du")) info["end_date"] = parse_date_fr(m.group("au")) img = _OG_IMG_RE.search(html) if img: info.setdefault("image", img.group(1)) return info def _enrich(self, records: list[dict]) -> dict: """Cache url → {fetched_at, start_date, …}, incrémental et plafonné. Fiches jamais vues d'abord (les plus récemment modifiées en premier — la liste REST est déjà triée), puis re-visite roulante après 7 jours.""" cache = self.load_cache() current = {r["link"] for r in records} cache = {u: c for u, c in cache.items() if u in current} now = time.time() candidates = [r["link"] for r in records if r["link"] not in cache or now - cache[r["link"]].get("fetched_at", 0) > REFRESH_AFTER] for url in candidates[:DETAIL_MAX]: try: info = self._parse_fiche(self.get(url).text) except Exception: # fiche cassée/404 : on ne bloque pas la source print(f"[sorti-ka] brossard : fiche ignorée {url}") info = {} info["fetched_at"] = now cache[url] = info self.save_cache(cache) return cache def fetch(self) -> list[Event]: records = [r for r in self._list() if r.get("link") and (r.get("title") or {}).get("rendered")] cache = self._enrich(records) type_names = self._type_event_map() events: list[Event] = [] for rec in records: info = cache.get(rec["link"]) or {} title = clean_text(rec["title"]["rendered"]) # repli : beaucoup de titres portent la date « (27 novembre 2026) » start = info.get("start_date") or parse_date_fr(title) if not start: continue # fiche pas encore visitée et titre sans date cats = [type_names[t] for t in rec.get("type-event") or [] if t in type_names] events.append(Event( source=self.source_id, external_id=str(rec["id"]), url=rec["link"], title=title, # description intégrale (la coupe à 600 caractères de la # vague 1 amputait 90 % des fiches — Phase 2) description=clean_text( (rec.get("content") or {}).get("rendered") or ""), raw_categories=cats or [title], venue=info.get("venue", ""), address=info.get("address", ""), postal_code=info.get("postal_code", ""), city="Brossard", region="Montérégie", start_date=start, start_time=info.get("time"), end_date=info.get("end_date") or start, end_time=info.get("end_time"), organizer="Ville de Brossard", image=info.get("image", ""), )) return events