# ----------------------------------------------------------------------------- # Sorti-Ka — Agrégateur de sorties & événements (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # schema.py : modèle de données standardisé (Event) + normalisation centrale # (patron Lou-Ka : louka/schema.py — Listing → ici Event) # ----------------------------------------------------------------------------- """Schéma standard d'un événement (Event) et normalisation des champs. Chaque connecteur, peu importe la source, doit produire des objets `Event` conformes à ce schéma. La méthode `finalize()` applique ensuite la couche de normalisation commune (sortika/normalize.py + sortika/regions.py) : dates ISO, catégories canoniques, gratuité/prix, région administrative — les connecteurs peuvent donc rester simples et remplir les champs bruts. """ from __future__ import annotations import hashlib import json from dataclasses import asdict, dataclass, field from .normalize import ( clean_city, clean_text, map_categories, parse_date_iso, parse_latlng, parse_price, ) from .regions import is_valid_region, region_from_city, region_from_tourist __all__ = ["Event"] @dataclass class Event: """Événement standardisé Sorti-Ka.""" source: str # id de la source (voir data/sources.json) external_id: str # identifiant chez la source url: str # fiche de l'événement chez la source title: str = "" # ex. "Festival d'été de Québec" description: str = "" categories: list[str] = field(default_factory=list) # taxonomie canonique raw_categories: list[str] = field(default_factory=list) # libellés source audience: str = "" # ex. "Pour tous", "Adultes, Adolescents" venue: str = "" # nom du lieu (ex. "Quai Jacques-Cartier") address: str = "" # adresse civique city: str = "" # Montréal, Québec, ... region: str = "" # 1 des 17 régions administratives tourist_region: str = "" # région touristique source (SITQ/ATR) postal_code: str = "" lat: float | None = None lng: float | None = None start_date: str | None = None # ISO "2026-09-11" end_date: str | None = None # ISO "2026-09-13" is_free: bool | None = None # gratuit (None = inconnu) price_min: float | None = None # plus bas prix ($ CAD) si payant price_label: str = "" # texte original du coût (traçabilité) organizer: str = "" # organisme / structure responsable website: str = "" # site officiel de l'événement image: str = "" # URL d'une image (si la source en offre) @property def uid(self) -> str: return f"{self.source}:{self.external_id}" def content_hash(self) -> str: """Hash du contenu pour la détection de changements (pseudo-webhook).""" payload = asdict(self) blob = json.dumps(payload, sort_keys=True, ensure_ascii=False) return hashlib.sha256(blob.encode("utf-8")).hexdigest() def dedup_key(self) -> str: """Empreinte inter-sources : titre normalisé + ville + date de début. Un même événement publié par deux sources (ex. SITQ et la ville) doit produire la même clé pour être fusionné (patron Lou-Ka / section 12). """ from .normalize import strip_accents t = strip_accents(self.title).lower() t = "".join(c for c in t if c.isalnum()) c = strip_accents(self.city).lower().replace(" ", "") return hashlib.sha1(f"{t}|{c}|{self.start_date or ''}".encode()).hexdigest() def finalize(self) -> "Event": """Applique la normalisation commune. Appelé par le pipeline d'ingestion. Idempotent ; ne remplace jamais une valeur explicite du connecteur. """ self.title = clean_text(self.title) self.description = clean_text(self.description) self.venue = clean_text(self.venue) self.address = clean_text(self.address) self.city = clean_city(self.city) self.audience = clean_text(self.audience) self.organizer = clean_text(self.organizer) self.start_date = parse_date_iso(self.start_date) self.end_date = parse_date_iso(self.end_date) or self.start_date if not self.categories: self.categories = map_categories(self.raw_categories) if self.is_free is None and self.price_min is None and self.price_label: self.is_free, self.price_min, self.price_label = parse_price(self.price_label) self.lat, self.lng = parse_latlng(self.lat, self.lng) # Région administrative : source explicite > région touristique > ville. if not is_valid_region(self.region): self.region = (region_from_tourist(self.tourist_region) or region_from_city(self.city) or "") if not self.title: raise ValueError(f"événement sans titre : {self.uid}") return self