# ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : jobka/schema.py # Rôle : Modèle de données standardisé (JobPosting) + normalisation centrale # Créé : 2026-08-17 Modifié : 2026-08-25 # ============================================================================= """Schéma standard d'une offre d'emploi (JobPosting) et normalisation. Chaque connecteur, peu importe l'employeur ou l'ATS source, doit produire des objets `JobPosting` conformes à ce schéma. La méthode `finalize()` applique ensuite la couche de normalisation commune (jobka/normalize.py) : dates ISO, salaires numériques + conversions $/h et $/an, mode de travail, type d'emploi, catégorie… — les connecteurs restent simples et remplissent les champs bruts. """ from __future__ import annotations import hashlib import json from dataclasses import dataclass, field, asdict from .normalize import ( # ré-exportés pour les connecteurs canonical_city, categorize, clean_html, clean_text, detect_language, extract_benefits, extract_requirements, is_quebec_location, parse_date, parse_employment_type, parse_location, parse_salary, parse_seniority, parse_work_mode, salary_from_text, salary_to_hourly, salary_to_yearly, sanitize_salary, strip_accents, ) __all__ = [ "JobPosting", "categorize", "clean_html", "clean_text", "is_quebec_location", "parse_date", "parse_employment_type", "parse_location", "parse_salary", "parse_work_mode", "strip_accents", ] @dataclass class JobPosting: """Offre d'emploi standardisée Job·Ka.""" source: str # id du connecteur (voir data/sources.json) external_id: str # identifiant chez l'employeur/ATS url: str # lien direct vers l'offre — zéro boîte noire employer: str = "" # nom de l'entreprise title: str = "" # titre du poste description: str = "" # texte complet (HTML nettoyé) address: str = "" # adresse du lieu de travail si disponible city: str = "" # Montréal, Québec, Sherbrooke… region: str = "" # région administrative / « Québec » postal_code: str = "" location_label: str = "" # libellé de lieu original (brut) work_mode: str | None = None # presentiel | hybride | teletravail employment_type: str | None = None # temps_plein | temps_partiel | # contractuel | stage | saisonnier salary_min: float | None = None # nombres, jamais des strings salary_max: float | None = None salary_unit: str | None = None # hour | day | week | biweek | month | year salary_label: str = "" # texte original (ex. « 25 $ à 30 $/h ») benefits: list[str] = field(default_factory=list) requirements: dict = field(default_factory=dict) # scolarité, expérience, langues date_posted: str | None = None # ISO 8601 date_deadline: str | None = None # ISO 8601 (date limite pour postuler) category: str = "" # taxonomie interne (TI, Santé, …) ats: str = "" # workday | lever | greenhouse | custom… details: dict = field(default_factory=dict) # champs structurés (sparse) lat: float | None = None lng: float | None = None company_logo: str = "" # URL du logo employeur (si exposé) language: str = "" # fr | en | bilingue ("" = indéterminée) apply_url: str = "" # lien de candidature directe (≠ url fiche) seniority: str | None = None # stage | junior | intermediaire | # senior | direction (None = indéterminée) @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() # Conversions dérivées — calculées à la volée, PAS dans content_hash # (elles découlent de salary_min/max/unit, les stocker dans le hash ne # ferait que dupliquer l'information). def salary_year_min(self) -> float | None: return salary_to_yearly(self.salary_min, self.salary_unit) def salary_year_max(self) -> float | None: return salary_to_yearly(self.salary_max, self.salary_unit) def salary_hour_min(self) -> float | None: return salary_to_hourly(self.salary_min, self.salary_unit) def salary_hour_max(self) -> float | None: return salary_to_hourly(self.salary_max, self.salary_unit) def finalize(self) -> "JobPosting": """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.employer = clean_text(self.employer) self.description = clean_text(self.description) if "\n" not in self.description \ else self.description.strip() # lieu : compléter ville/région/code postal depuis le libellé brut, # puis canonicaliser la ville (« Montreal » -> « Montréal ») if self.location_label and not (self.city and self.region): loc = parse_location(self.location_label) self.city = self.city or loc["city"] self.region = self.region or loc["region"] self.postal_code = self.postal_code or loc["postal_code"] self.city = canonical_city(self.city) # région administrative (17 régions, répertoire MAMH embarqué) ; # repli : « Québec » (province) quand seule l'appartenance QC est sûre if self.city and (not self.region or self.region == "Québec"): from .regions import region_for_city self.region = region_for_city(self.city) or self.region if self.city and not self.region and is_quebec_location(self.city): self.region = "Québec" # salaire : parser le libellé si le connecteur n'a pas de valeurs numériques if self.salary_min is None and self.salary_label: lo, hi, unit = parse_salary(self.salary_label) self.salary_min, self.salary_max, self.salary_unit = lo, hi, unit if self.salary_min is None and self.description: # mention salariale dans la description (fréquent au QC) — avec # garde-fou de contexte (financement, remboursements ≠ salaire) lo, hi, unit, label = salary_from_text(self.description) if lo is not None: self.salary_min, self.salary_max, self.salary_unit = lo, hi, unit self.salary_label = self.salary_label or label # bornes de plausibilité aussi sur les valeurs STRUCTURÉES des ATS # (unités mal étiquetées : « 75 000 $/mois » qui est un annuel…) self.salary_min, self.salary_max, self.salary_unit = sanitize_salary( self.salary_min, self.salary_max, self.salary_unit) if self.salary_min is None: self.salary_unit = None # dates : accepter epoch/textuel/relatif -> ISO if self.date_posted is not None: self.date_posted = parse_date(self.date_posted) if self.date_deadline is not None: self.date_deadline = parse_date(self.date_deadline) # mode de travail / type d'emploi : depuis le libellé de lieu, le titre # puis la description — première réponse trouvée, jamais de défaut if self.work_mode is None: for texte in (self.location_label, self.title, self.description[:500]): self.work_mode = parse_work_mode(texte) if self.work_mode: break if self.employment_type is None: for texte in (self.details.get("employment_label", ""), self.title, self.description[:500]): self.employment_type = parse_employment_type(str(texte)) if self.employment_type: break if not self.category: self.category = categorize(self.title, self.description) # langue de l'offre : valeur source si fournie, sinon détection légère if not self.language: self.language = detect_language(self.title, self.description) elif self.language not in ("fr", "en", "bilingue"): lang = str(self.language).lower()[:2] self.language = {"fr": "fr", "en": "en"}.get(lang, "") # avantages : blocs structurés « Avantages / Benefits » des descriptions if not self.benefits and self.description: self.benefits = extract_benefits(self.description) # exigences structurées (expérience, scolarité, langues) : extraites de # la description, sans jamais écraser une valeur fournie par l'ATS if self.description: for k, v in extract_requirements(self.title, self.description).items(): self.requirements.setdefault(k, v) # séniorité du poste : titre, libellé ATS, puis années d'expérience if self.seniority is None: self.seniority = parse_seniority( self.title, self.requirements, self.employment_type) # coordonnées fournies par la source : rejeter tout point hors de la # province (lat/lng inversés, 0/0, coquilles) — le géocodeur prendra # le relais sur l'adresse/la ville if self.lat is not None and self.lng is not None: if not (44.5 <= self.lat <= 63.0 and -80.0 <= self.lng <= -56.0): self.lat = self.lng = None return self