#!/usr/bin/env python3 # ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : scripts/gen_connectors.py # Rôle : Génération en série des connecteurs employeurs à partir d'un # fichier de flux vérifiés (data/feeds.json) + MAJ de sources.json # Créé : 2026-08-17 Modifié : 2026-08-22 # ============================================================================= """Génère un connecteur (~15 lignes) par flux ATS vérifié. Usage : python3 scripts/gen_connectors.py data/feeds.json Format d'entrée (liste JSON) — champs selon l'ATS : {"ats":"workday","tenant":"x","host":"wd3","site":"Site","employer":"Nom", "qc":N, "cities":[...]} {"ats":"lever"|"greenhouse"|"smartrecruiters"|"ashby"|"workable"| "recruitee"|"breezy"|"bamboohr","org":"slug","employer":"Nom", ...} Idempotent : un connecteur existant n'est PAS réécrit (préserve les ajustements manuels) ; sources.json est fusionné par id. """ from __future__ import annotations import datetime import json import re import sys import unicodedata from pathlib import Path ROOT = Path(__file__).resolve().parent.parent CONN_DIR = ROOT / "jobka" / "connectors" SOURCES = ROOT / "data" / "sources.json" TODAY = datetime.date.today().isoformat() # ats -> (module de plateforme, classe de base, attribut(s) d'org) PLATFORMS = { "workday": ("workday", "WorkdayConnector"), "lever": ("lever", "LeverConnector"), "greenhouse": ("greenhouse", "GreenhouseConnector"), "smartrecruiters": ("smartrecruiters", "SmartRecruitersConnector"), "ashby": ("ashby", "AshbyConnector"), "workable": ("workable", "WorkableConnector"), "recruitee": ("recruitee", "RecruiteeConnector"), "breezy": ("breezy", "BreezyConnector"), "bamboohr": ("bamboohr", "BambooHRConnector"), } ORG_ATTR = {"lever": "ORG", "greenhouse": "BOARD", "smartrecruiters": "COMPANY", "ashby": "ORG", "workable": "ORG", "recruitee": "ORG", "breezy": "ORG", "bamboohr": "ORG"} # ATS à attributs multiples (secteur public/parapublic et grands employeurs) : # feed[clé] -> attribut de classe. Voir data/feeds-public.json. EXTRA_ATTRS = { "taleo": [("host", "HOST"), ("section", "SECTION"), ("portal", "PORTAL")], "njoyn": [("base", "BASE"), ("cl", "CL"), ("clid", "CLID"), ("use_scrapfly", "USE_SCRAPFLY")], "ultipro": [("org", "ORG"), ("board", "BOARD")], "icims": [("sub", "SUB")], "successfactors": [("base", "BASE")], "adp": [("cid", "CID"), ("ccid", "CCID")], "digitalrecruiters": [("base", "BASE")], "workland": [("list_url", "LIST_URL")], "dayforce": [("ns", "NS"), ("board", "BOARD")], } CAREERS_URL = { "workday": "https://{tenant}.{host}.myworkdayjobs.com/{site}", "lever": "https://jobs.lever.co/{org}", "greenhouse": "https://job-boards.greenhouse.io/{org}", "smartrecruiters": "https://jobs.smartrecruiters.com/company/{org}", "ashby": "https://jobs.ashbyhq.com/{org}", "workable": "https://apply.workable.com/{org}", "recruitee": "https://{org}.recruitee.com", "breezy": "https://{org}.breezy.hr", "bamboohr": "https://{org}.bamboohr.com/careers", "taleo": "https://{host}.taleo.net/careersection/{section}/jobsearch.ftl?lang=fr", "njoyn": "{base}/{cl}/xweb/xweb.asp?clid={clid}&page=joblisting&lang=2", "ultipro": "https://recruiting.ultipro.com/{org}/JobBoard/{board}", "icims": "https://{sub}.icims.com/jobs/search?ss=1", "successfactors": "{base}", "adp": ("https://workforcenow.adp.com/mascsr/default/mdf/recruitment/" "recruitment.html?cid={cid}&ccId={ccid}&lang=fr_CA"), "digitalrecruiters": "{base}", "workland": "{list_url}", "dayforce": "https://jobs.dayforcehcm.com/fr-CA/{ns}/{board}", } PLATFORMS.update({ "taleo": ("taleo", "TaleoConnector"), "njoyn": ("njoyn", "NjoynConnector"), "ultipro": ("ultipro", "UltiProConnector"), "icims": ("icims", "ICIMSConnector"), "successfactors": ("successfactors", "SuccessFactorsConnector"), "adp": ("adp", "ADPWorkforceNowConnector"), "digitalrecruiters": ("digitalrecruiters", "DigitalRecruitersConnector"), "workland": ("workland", "WorklandConnector"), "dayforce": ("dayforce", "DayforceConnector"), }) def slugify(s: str) -> str: s = unicodedata.normalize("NFD", s) s = "".join(c for c in s if unicodedata.category(c) != "Mn") s = re.sub(r"[^A-Za-z0-9]+", "_", s.lower()).strip("_") s = re.sub(r"_+", "_", s) # un module/identifiant Python ne peut pas commencer par un chiffre return f"x{s}" if s and s[0].isdigit() else s def camel(s: str) -> str: return "".join(p.capitalize() for p in s.split("_")) HEADER = """\ # ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : jobka/connectors/{fname} # Rôle : Connecteur {employer} — {ats} ({detail}) # [généré par scripts/gen_connectors.py, flux vérifié le {today}] # Créé : {today} Modifié : {today} # ============================================================================= """ def gen_one(feed: dict) -> tuple[str, str] | None: ats = feed["ats"] module, base = PLATFORMS[ats] sid = feed.get("source_id") or slugify(feed.get("org") or feed.get("tenant") or feed["employer"]) fname = f"{sid}.py" path = CONN_DIR / fname if path.exists(): return None # idempotent : ne pas écraser un connecteur existant cls = camel(sid) + "Connector" if ats == "workday": detail = f"tenant {feed['tenant']}.{feed['host']}, site {feed['site']}" body = (f" TENANT = {feed['tenant']!r}\n" f" HOST = {feed['host']!r}\n" f" SITE = {feed['site']!r}\n") elif ats in EXTRA_ATTRS: pairs = [(k, a) for k, a in EXTRA_ATTRS[ats] if feed.get(k) not in (None, "")] detail = ", ".join(f"{a.lower()}={feed[k]!r}" for k, a in pairs[:2]) body = "".join(f" {a} = {feed[k]!r}\n" for k, a in pairs) else: attr = ORG_ATTR[ats] detail = f"org « {feed['org']} »" body = f" {attr} = {feed['org']!r}\n" src = (HEADER.format(fname=fname, employer=feed["employer"], ats=ats, detail=detail, today=TODAY) + f"from .{module} import {base}\n\n\n" + f"class {cls}({base}):\n" + f" source_id = {sid!r}\n" + f" EMPLOYER = {feed['employer']!r}\n" + body) path.write_text(src, encoding="utf-8") return sid, fname def careers_url(feed: dict) -> str: return CAREERS_URL[feed["ats"]].format(**feed) def main() -> None: feeds = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) reg = json.loads(SOURCES.read_text(encoding="utf-8")) known = {s["id"] for s in reg["sources"]} created = skipped = 0 for feed in feeds: sid = feed.get("source_id") or slugify(feed.get("org") or feed.get("tenant") or feed["employer"]) feed["source_id"] = sid res = gen_one(feed) if res: created += 1 print(f" + {res[1]}") else: skipped += 1 if sid not in known: reg["sources"].append({ "id": sid, "name": feed["employer"], "url": feed.get("url", ""), "careers_url": careers_url(feed), "sectors": feed.get("sectors", []), "connector": feed["ats"], "status": "actif", "region": ", ".join((feed.get("cities") or [])[:3]), }) known.add(sid) reg["_meta"]["updated"] = TODAY SOURCES.write_text(json.dumps(reg, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(f"[gen] {created} connecteurs créés, {skipped} déjà présents, " f"{len(reg['sources'])} sources au registre") if __name__ == "__main__": main()