# ----------------------------------------------------------------------------- # Forma-Ka — Agrégateur de formations (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # schema.py : modèle de données standardisé (Formation) + normalisation finale # ----------------------------------------------------------------------------- """Schéma standard d'une formation et normalisation des champs. Chaque connecteur, peu importe le site source (université, cégep, firme de formation, plateforme…), produit des objets `Formation` conformes à ce schéma. `finalize()` applique la couche commune (formaka/normalize.py). Philosophie : le PRIX EST OPTIONNEL (un cours universitaire n'affiche pas de prix) — ce qui compte, ce sont les DÉTAILS de la formation : description, objectifs, préalables, plan de cours, durée, crédits/UEC, mode, clientèle… """ from __future__ import annotations import hashlib import json from dataclasses import dataclass, field, asdict from .normalize import ( # ré-exportés pour les connecteurs clean_text, extract_details, merge_details, normalize_language, normalize_mode, normalize_type, parse_date_fr, parse_duration_hours, parse_price, price_is_from, strip_accents, ) __all__ = [ "Formation", "clean_text", "normalize_type", "normalize_mode", "normalize_language", "parse_price", "parse_duration_hours", "parse_date_fr", "strip_accents", ] @dataclass class Formation: """Formation standardisée Forma-Ka.""" source: str # id de la source (voir data/sources.json) external_id: str # identifiant chez la source url: str # page de la formation chez la source title: str = "" # ex. « Gestion de projet agile » training_type: str = "" # Cours universitaire, Séminaire, Atelier… category: str = "" # domaine : Informatique, Gestion, RH… mode: str = "" # en ligne | présentiel | hybride | asynchrone city: str = "" # ville (si présentiel/hybride) language: str = "" # fr | en | fr/en price: float | None = None # $ CAD — None = non affiché (normal !) price_label: str = "" # texte original (« 1 295 $ + tx ») is_free: bool | None = None # gratuit (None = inconnu) duration: str = "" # texte original (« 2 jours », « 45 h ») duration_hours: float | None = None start_date: str | None = None # ISO — prochaine séance/session schedule_label: str = "" # texte original des dates/horaires sessions: list[str] = field(default_factory=list) # toutes les dates offertes level: str = "" # débutant | intermédiaire | avancé credits: str = "" # « 3 crédits », « 1,4 UEC » credential: str = "" # attestation, certificat, diplôme, UEC… instructor: str = "" # formateur / professeur code: str = "" # sigle du cours (ex. « GSF-1020 ») description: str = "" # description complète objectives: list[str] = field(default_factory=list) # objectifs d'apprentissage prerequisites: str = "" # préalables / conditions d'admission audience: str = "" # clientèle visée program: list[str] = field(default_factory=list) # plan / contenu détaillé tags: list[str] = field(default_factory=list) details: dict = field(default_factory=dict) # champs structurés (JSON) images: list[str] = field(default_factory=list) # URLs absolues @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 finalize(self) -> "Formation": """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 = (self.description or "").strip() self.training_type = normalize_type(self.training_type) self.mode = normalize_mode(self.mode) self.language = normalize_language(self.language) self.objectives = [clean_text(o) for o in self.objectives if clean_text(o)] self.program = [clean_text(p) for p in self.program if clean_text(p)] self.tags = sorted({clean_text(t) for t in self.tags if clean_text(t)}) if self.price is None: self.price = parse_price(self.price_label) if self.is_free is None and self.price is not None: self.is_free = self.price == 0.0 if self.duration_hours is None: self.duration_hours = parse_duration_hours(self.duration) if self.start_date is None: self.start_date = parse_date_fr(self.schedule_label) if not self.sessions and self.start_date: self.sessions = [self.start_date] # détails dérivés des textes libres (le connecteur garde priorité) derived = extract_details(self.duration, self.description, self.credits, self.schedule_label) if self.price_label and price_is_from(self.price_label): derived["price_from"] = True self.details = merge_details(self.details, derived) if not self.level: self.level = self.details.get("level", "") if self.duration_hours is None: dh = self.details.get("duration_hours") self.duration_hours = float(dh) if dh is not None else None if not self.credits: if self.details.get("uec") is not None: self.credits = f"{self.details['uec']:g} UEC".replace(".", ",") elif self.details.get("credits") is not None: self.credits = f"{self.details['credits']:g} crédits" return self