SPB Git forge

spb/sorti-ka

Public

Toutes les sorties et tous les événements du Québec, un seul endroit — 7 connecteurs, fiches SSR, design Groupe KA.

58commits 1branches 0releases
13.7 MBsize
maindefault branch
17 days agolast push
HTML 82.9% Python 15.2% TypeScript 0.9% JavaScript 0.7%
8.9 KB · 209 lines python
Raw Blame History
1# -----------------------------------------------------------------------------2# Sorti-Ka — Agrégateur de sorties & événements (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/magog.py : connecteur Ville de Magog — événements et activités5#   Source    : https://www.ville.magog.qc.ca/calendrier/evenements-et-activites/6#               — la page municipale embarque un calendrier Google public7#               (plugin « Simple Google Calendar », id découvert dans le HTML).8#   Extraction: flux iCal PUBLIC du calendrier Google9#               fr7c02sjcaibul422352d811oc@group.calendar.google.com10#               (~1 230 VEVENT dont l'historique depuis 2017) : SUMMARY,11#               DTSTART/DTEND UTC → date+heure locales (utc_to_local),12#               LOCATION (« Salle, adresse, ville... »), DESCRIPTION (contient13#               l'URL de la fiche ville.magog.qc.ca), STATUS. Récurrences14#               Google observées (2026-08-25) : FREQ=DAILY/WEEKLY avec COUNT/15#               UNTIL/BYDAY → mini-expanseur déterministe (aucune invention),16#               occurrences plafonnées à HORIZON_DAYS. 1 requête par sync.17#   Accès     : flux iCal publié publiquement par la Ville (bouton « + Google18#               Agenda » de sa page) ; conditions Google : accès lecture aux19#               calendriers publics. GET direct, identification honnête.20#   Prix      : non publié par le flux → jamais inventé.21# -----------------------------------------------------------------------------22from __future__ import annotations2324import re25from datetime import date, timedelta2627from ..normalize import clean_text, utc_to_local28from ..schema import Event29from .base import BaseConnector3031CAL_ID = "fr7c02sjcaibul422352d811oc@group.calendar.google.com"32FEED = ("https://calendar.google.com/calendar/ical/"33        + CAL_ID.replace("@", "%40") + "/public/basic.ics")34PAGE = "https://www.ville.magog.qc.ca/calendrier/evenements-et-activites/"35HORIZON_DAYS = 365          # occurrences futures émises au plus à 1 an36GRACE_DAYS = 7              # on garde les événements finis depuis < 7 jours3738# URL de fiche dans DESCRIPTION — parfois enrobée d'un redirect Google39# (« ...&sa=D&source=... ») : on coupe au premier « & »40_URL_RE = re.compile(r"https?://(?:www\.)?ville\.magog\.qc\.ca[^\s\"'<>&]*")41_BYDAY = {"MO": 0, "TU": 1, "WE": 2, "TH": 3, "FR": 4, "SA": 5, "SU": 6}42_DATE_ONLY_RE = re.compile(r"^(\d{4})(\d{2})(\d{2})$")434445def _unfold(ics: str) -> str:46    """Déplie les lignes continuées du format iCalendar (RFC 5545)."""47    return re.sub(r"\r?\n[ \t]", "", ics.replace("\r\n", "\n"))484950def _unescape(v: str) -> str:51    return (v.replace("\\n", "\n").replace("\\,", ",")52            .replace("\\;", ";").replace("\\\\", "\\"))535455def _prop(block: str, name: str) -> str:56    m = re.search(rf"^{name}(?:;[^:\n]*)?:(.*)$", block, re.M)57    return _unescape(m.group(1).strip()) if m else ""585960def _dt(block: str, name: str) -> tuple[str | None, str | None]:61    """DTSTART/DTEND iCal → (date, heure) locales QC. Trois formes Google :62    « ...Z » (UTC → conversion), « ;TZID=America/Toronto:... » (déjà locale),63    « ;VALUE=DATE:AAAAMMJJ » (journée entière, heure None)."""64    m = re.search(rf"^{name}(;[^:\n]*)?:(.*)$", block, re.M)65    if not m:66        return None, None67    raw = m.group(2).strip()68    if _DATE_ONLY_RE.match(raw):69        return f"{raw[:4]}-{raw[4:6]}-{raw[6:8]}", None70    if raw.endswith("Z"):71        return utc_to_local(raw)72    dm = re.match(r"(\d{4})(\d{2})(\d{2})T(\d{2})(\d{2})", raw)73    if not dm:74        return None, None75    t = f"{dm.group(4)}:{dm.group(5)}"           # TZID/flottant = heure locale76    return (f"{dm.group(1)}-{dm.group(2)}-{dm.group(3)}",77            None if t == "00:00" else t)787980def _expand(start_iso: str, rrule: str, horizon: "date") -> list["date"]:81    """Occurrences d'une RRULE Google simple (DAILY/WEEKLY, COUNT/UNTIL/BYDAY/82    INTERVAL) — expansion déterministe, plafonnée à `horizon`. Les fréquences83    non gérées retournent la seule date de départ (jamais d'invention)."""84    try:85        d0 = date.fromisoformat(start_iso)86    except ValueError:87        return []88    parts = dict(p.split("=", 1) for p in rrule.split(";") if "=" in p)89    freq = parts.get("FREQ", "")90    if freq not in ("DAILY", "WEEKLY"):91        return [d0]92    interval = max(int(parts.get("INTERVAL", "1") or 1), 1)93    count = int(parts["COUNT"]) if parts.get("COUNT", "").isdigit() else None94    until = None95    m = re.match(r"(\d{4})(\d{2})(\d{2})", parts.get("UNTIL", ""))96    if m:97        until = date(int(m.group(1)), int(m.group(2)), int(m.group(3)))98    out: list[date] = []99    if freq == "DAILY":100        d, n = d0, 0101        while d <= horizon and (count is None or n < count) \102                and (until is None or d <= until):103            out.append(d)104            n += 1105            d += timedelta(days=interval)106        return out107    # WEEKLY : jours de la semaine de BYDAY (défaut : jour de départ)108    days = sorted({_BYDAY[t] for t in parts.get("BYDAY", "").split(",")109                   if t in _BYDAY}) or [d0.weekday()]110    week0 = d0 - timedelta(days=d0.weekday())        # lundi de la 1re semaine111    n, w = 0, 0112    while True:113        base = week0 + timedelta(weeks=w * interval)114        if base > horizon:115            break116        for wd in days:117            d = base + timedelta(days=wd)118            if d < d0 or d > horizon or (until and d > until):119                continue120            if count is not None and n >= count:121                return out122            out.append(d)123            n += 1124        w += 1125        if count is not None and n >= count:126            break127    return out128129130class MagogConnector(BaseConnector):131    source_id = "magog"132133    def _parse_feed(self, ics: str, today: str | None = None) -> list[dict]:134        today = today or date.today().isoformat()135        floor = (date.fromisoformat(today)136                 - timedelta(days=GRACE_DAYS)).isoformat()137        horizon = date.fromisoformat(today) + timedelta(days=HORIZON_DAYS)138        rows: list[dict] = []139        for block in re.findall(r"BEGIN:VEVENT\n(.*?)\nEND:VEVENT",140                                _unfold(ics), re.S):141            title = clean_text(_prop(block, "SUMMARY"))142            if not title:143                continue144            start, start_time = _dt(block, "DTSTART")145            end, end_time = _dt(block, "DTEND")146            if not start:147                continue148            # DTEND iCal exclusif sur les événements « journée entière »149            if end and end_time is None and start_time is None and end > start:150                end = (date.fromisoformat(end) - timedelta(days=1)).isoformat()151            desc = _prop(block, "DESCRIPTION")152            m = _URL_RE.search(desc)153            url = m.group(0).rstrip(").,&") if m else PAGE154            base = {155                "uid": _prop(block, "UID").split("@")[0],156                "title": title,157                "url": url,158                "location": clean_text(_prop(block, "LOCATION")),159                "status": _prop(block, "STATUS"),160                "start_time": start_time,161                "end_time": end_time,162            }163            duration = 0164            if end and end >= start:165                duration = (date.fromisoformat(end)166                            - date.fromisoformat(start)).days167            rrule = _prop(block, "RRULE")168            starts = (_expand(start, rrule, horizon) if rrule169                      else [date.fromisoformat(start)])170            for d in starts:171                e = (d + timedelta(days=duration)).isoformat()172                if e < floor or d.isoformat() > horizon.isoformat():173                    continue174                rows.append(dict(base, start=d.isoformat(), end=e))175        return rows176177    def fetch(self) -> list[Event]:178        rows = self._parse_feed(self.get(FEED).text)179        events: list[Event] = []180        seen: set[str] = set()181        for r in rows:182            ext = f"{r['uid']}-{r['start']}"183            if ext in seen:184                continue185            seen.add(ext)186            # LOCATION Google : « Lieu, 100 Rue X, Magog, QC J1X..., Canada »187            venue, address = r["location"], ""188            if "," in r["location"]:189                venue, address = [p.strip() for p in190                                  r["location"].split(",", 1)]191            events.append(Event(192                source=self.source_id,193                external_id=ext,194                url=r["url"],195                title=r["title"],196                raw_categories=[r["title"]],197                venue=venue,198                address=address,199                city="Magog",200                region="Estrie",201                start_date=r["start"],202                start_time=r["start_time"],203                end_date=r["end"],204                end_time=r["end_time"],205                status="" if r["status"] == "CONFIRMED" else r["status"],206                organizer="Ville de Magog",207            ))208        return events209