# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/magog.py : connecteur Ville de Magog — événements et activités # Source : https://www.ville.magog.qc.ca/calendrier/evenements-et-activites/ # — la page municipale embarque un calendrier Google public # (plugin « Simple Google Calendar », id découvert dans le HTML). # Extraction: flux iCal PUBLIC du calendrier Google # fr7c02sjcaibul422352d811oc@group.calendar.google.com # (~1 230 VEVENT dont l'historique depuis 2017) : SUMMARY, # DTSTART/DTEND UTC → date+heure locales (utc_to_local), # LOCATION (« Salle, adresse, ville... »), DESCRIPTION (contient # l'URL de la fiche ville.magog.qc.ca), STATUS. Récurrences # Google observées (2026-08-25) : FREQ=DAILY/WEEKLY avec COUNT/ # UNTIL/BYDAY → mini-expanseur déterministe (aucune invention), # occurrences plafonnées à HORIZON_DAYS. 1 requête par sync. # Accès : flux iCal publié publiquement par la Ville (bouton « + Google # Agenda » de sa page) ; conditions Google : accès lecture aux # calendriers publics. GET direct, identification honnête. # Prix : non publié par le flux → jamais inventé. # ----------------------------------------------------------------------------- from __future__ import annotations import re from datetime import date, timedelta from ..normalize import clean_text, utc_to_local from ..schema import Event from .base import BaseConnector CAL_ID = "fr7c02sjcaibul422352d811oc@group.calendar.google.com" FEED = ("https://calendar.google.com/calendar/ical/" + CAL_ID.replace("@", "%40") + "/public/basic.ics") PAGE = "https://www.ville.magog.qc.ca/calendrier/evenements-et-activites/" HORIZON_DAYS = 365 # occurrences futures émises au plus à 1 an GRACE_DAYS = 7 # on garde les événements finis depuis < 7 jours # URL de fiche dans DESCRIPTION — parfois enrobée d'un redirect Google # (« ...&sa=D&source=... ») : on coupe au premier « & » _URL_RE = re.compile(r"https?://(?:www\.)?ville\.magog\.qc\.ca[^\s\"'<>&]*") _BYDAY = {"MO": 0, "TU": 1, "WE": 2, "TH": 3, "FR": 4, "SA": 5, "SU": 6} _DATE_ONLY_RE = re.compile(r"^(\d{4})(\d{2})(\d{2})$") def _unfold(ics: str) -> str: """Déplie les lignes continuées du format iCalendar (RFC 5545).""" return re.sub(r"\r?\n[ \t]", "", ics.replace("\r\n", "\n")) def _unescape(v: str) -> str: return (v.replace("\\n", "\n").replace("\\,", ",") .replace("\\;", ";").replace("\\\\", "\\")) def _prop(block: str, name: str) -> str: m = re.search(rf"^{name}(?:;[^:\n]*)?:(.*)$", block, re.M) return _unescape(m.group(1).strip()) if m else "" def _dt(block: str, name: str) -> tuple[str | None, str | None]: """DTSTART/DTEND iCal → (date, heure) locales QC. Trois formes Google : « ...Z » (UTC → conversion), « ;TZID=America/Toronto:... » (déjà locale), « ;VALUE=DATE:AAAAMMJJ » (journée entière, heure None).""" m = re.search(rf"^{name}(;[^:\n]*)?:(.*)$", block, re.M) if not m: return None, None raw = m.group(2).strip() if _DATE_ONLY_RE.match(raw): return f"{raw[:4]}-{raw[4:6]}-{raw[6:8]}", None if raw.endswith("Z"): return utc_to_local(raw) dm = re.match(r"(\d{4})(\d{2})(\d{2})T(\d{2})(\d{2})", raw) if not dm: return None, None t = f"{dm.group(4)}:{dm.group(5)}" # TZID/flottant = heure locale return (f"{dm.group(1)}-{dm.group(2)}-{dm.group(3)}", None if t == "00:00" else t) def _expand(start_iso: str, rrule: str, horizon: "date") -> list["date"]: """Occurrences d'une RRULE Google simple (DAILY/WEEKLY, COUNT/UNTIL/BYDAY/ INTERVAL) — expansion déterministe, plafonnée à `horizon`. Les fréquences non gérées retournent la seule date de départ (jamais d'invention).""" try: d0 = date.fromisoformat(start_iso) except ValueError: return [] parts = dict(p.split("=", 1) for p in rrule.split(";") if "=" in p) freq = parts.get("FREQ", "") if freq not in ("DAILY", "WEEKLY"): return [d0] interval = max(int(parts.get("INTERVAL", "1") or 1), 1) count = int(parts["COUNT"]) if parts.get("COUNT", "").isdigit() else None until = None m = re.match(r"(\d{4})(\d{2})(\d{2})", parts.get("UNTIL", "")) if m: until = date(int(m.group(1)), int(m.group(2)), int(m.group(3))) out: list[date] = [] if freq == "DAILY": d, n = d0, 0 while d <= horizon and (count is None or n < count) \ and (until is None or d <= until): out.append(d) n += 1 d += timedelta(days=interval) return out # WEEKLY : jours de la semaine de BYDAY (défaut : jour de départ) days = sorted({_BYDAY[t] for t in parts.get("BYDAY", "").split(",") if t in _BYDAY}) or [d0.weekday()] week0 = d0 - timedelta(days=d0.weekday()) # lundi de la 1re semaine n, w = 0, 0 while True: base = week0 + timedelta(weeks=w * interval) if base > horizon: break for wd in days: d = base + timedelta(days=wd) if d < d0 or d > horizon or (until and d > until): continue if count is not None and n >= count: return out out.append(d) n += 1 w += 1 if count is not None and n >= count: break return out class MagogConnector(BaseConnector): source_id = "magog" def _parse_feed(self, ics: str, today: str | None = None) -> list[dict]: today = today or date.today().isoformat() floor = (date.fromisoformat(today) - timedelta(days=GRACE_DAYS)).isoformat() horizon = date.fromisoformat(today) + timedelta(days=HORIZON_DAYS) rows: list[dict] = [] for block in re.findall(r"BEGIN:VEVENT\n(.*?)\nEND:VEVENT", _unfold(ics), re.S): title = clean_text(_prop(block, "SUMMARY")) if not title: continue start, start_time = _dt(block, "DTSTART") end, end_time = _dt(block, "DTEND") if not start: continue # DTEND iCal exclusif sur les événements « journée entière » if end and end_time is None and start_time is None and end > start: end = (date.fromisoformat(end) - timedelta(days=1)).isoformat() desc = _prop(block, "DESCRIPTION") m = _URL_RE.search(desc) url = m.group(0).rstrip(").,&") if m else PAGE base = { "uid": _prop(block, "UID").split("@")[0], "title": title, "url": url, "location": clean_text(_prop(block, "LOCATION")), "status": _prop(block, "STATUS"), "start_time": start_time, "end_time": end_time, } duration = 0 if end and end >= start: duration = (date.fromisoformat(end) - date.fromisoformat(start)).days rrule = _prop(block, "RRULE") starts = (_expand(start, rrule, horizon) if rrule else [date.fromisoformat(start)]) for d in starts: e = (d + timedelta(days=duration)).isoformat() if e < floor or d.isoformat() > horizon.isoformat(): continue rows.append(dict(base, start=d.isoformat(), end=e)) return rows def fetch(self) -> list[Event]: rows = self._parse_feed(self.get(FEED).text) events: list[Event] = [] seen: set[str] = set() for r in rows: ext = f"{r['uid']}-{r['start']}" if ext in seen: continue seen.add(ext) # LOCATION Google : « Lieu, 100 Rue X, Magog, QC J1X..., Canada » venue, address = r["location"], "" if "," in r["location"]: venue, address = [p.strip() for p in r["location"].split(",", 1)] events.append(Event( source=self.source_id, external_id=ext, url=r["url"], title=r["title"], raw_categories=[r["title"]], venue=venue, address=address, city="Magog", region="Estrie", start_date=r["start"], start_time=r["start_time"], end_date=r["end"], end_time=r["end_time"], status="" if r["status"] == "CONFIRMED" else r["status"], organizer="Ville de Magog", )) return events