1# =============================================================================2# Job·Ka — Groupe KA3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : jobka/db.py6# Rôle : Persistance SQLite — upsert avec détection de changements, cycle de7# vie avec délai de grâce, anti-dérive, expiration sur date limite8# Créé : 2026-08-17 Modifié : 2026-08-259# =============================================================================10from __future__ import annotations1112import datetime as _dt13import json14import sqlite315import statistics16import time17from pathlib import Path1819from .schema import JobPosting2021DB_PATH = Path(__file__).resolve().parent.parent / "data" / "jobka.db"2223# Nombre d'exécutions consécutives où une offre doit être absente de la24# source avant d'être désactivée (délai de grâce contre les ratés ponctuels).25MISS_GRACE = 22627# Dérive : si une source retourne <= DRIFT_RATIO × sa médiane historique28# (médiane >= DRIFT_MIN_BASE offres), on alerte et on suspend les retraits.29DRIFT_RATIO = 0.2530DRIFT_MIN_BASE = 831DRIFT_HISTORY = 53233_SCHEMA = """34CREATE TABLE IF NOT EXISTS jobs (35 uid TEXT PRIMARY KEY,36 source TEXT NOT NULL,37 external_id TEXT NOT NULL,38 url TEXT,39 employer TEXT,40 title TEXT,41 description TEXT,42 address TEXT,43 city TEXT,44 region TEXT,45 postal_code TEXT,46 location_label TEXT,47 work_mode TEXT,48 employment_type TEXT,49 salary_min REAL,50 salary_max REAL,51 salary_unit TEXT,52 salary_label TEXT,53 salary_year_min REAL, -- conversions dérivées ($/an, $/h) pour filtres/tri54 salary_year_max REAL,55 salary_hour_min REAL,56 salary_hour_max REAL,57 benefits TEXT, -- JSON (liste de textes source)58 requirements TEXT, -- JSON (scolarité, expérience, langues)59 date_posted TEXT,60 date_deadline TEXT,61 category TEXT,62 ats TEXT,63 details TEXT, -- JSON (champs structurés, sparse)64 lat REAL,65 lng REAL,66 geocode_failed INTEGER DEFAULT 0,67 content_hash TEXT,68 first_seen REAL,69 last_seen REAL,70 updated_at REAL,71 miss_count INTEGER DEFAULT 0,72 active INTEGER DEFAULT 1,73 dup_of TEXT, -- uid de l'offre canonique si doublon inter-sources74 dup_sources TEXT -- JSON : autres sources où l'offre est publiée75);76CREATE INDEX IF NOT EXISTS idx_jobs_source ON jobs(source);77CREATE INDEX IF NOT EXISTS idx_jobs_city ON jobs(city);78CREATE INDEX IF NOT EXISTS idx_jobs_active ON jobs(active);79CREATE INDEX IF NOT EXISTS idx_jobs_category ON jobs(category);80CREATE INDEX IF NOT EXISTS idx_jobs_geo ON jobs(lat, lng);8182CREATE TABLE IF NOT EXISTS sync_log (83 id INTEGER PRIMARY KEY AUTOINCREMENT,84 source TEXT,85 ts REAL,86 found INTEGER,87 added INTEGER,88 updated INTEGER,89 removed INTEGER,90 ok INTEGER,91 message TEXT,92 stats TEXT -- JSON : taux de champs null, missed, alerte…93);9495CREATE TABLE IF NOT EXISTS detail_cache (96 source TEXT NOT NULL,97 external_id TEXT NOT NULL,98 key TEXT, -- hash du contenu « liste » de l'offre99 payload TEXT, -- JSON opaque propre au connecteur100 fetched_at REAL,101 PRIMARY KEY (source, external_id)102);103104CREATE TABLE IF NOT EXISTS salary_log (105 uid TEXT NOT NULL,106 ts REAL NOT NULL,107 s_min REAL,108 s_max REAL,109 unit TEXT110);111CREATE INDEX IF NOT EXISTS idx_salary_log_uid ON salary_log(uid);112113CREATE TABLE IF NOT EXISTS users (114 id INTEGER PRIMARY KEY AUTOINCREMENT,115 ka_sub TEXT UNIQUE, -- « ka:{sub} » identifiant stable du hub KA116 ka_id TEXT, -- identifiant membre « ka-0123456789 » (hub)117 email TEXT,118 name TEXT,119 picture TEXT, -- URL de l'avatar (hub / Google)120 created_at REAL,121 last_login REAL122);123CREATE UNIQUE INDEX IF NOT EXISTS users_ka_id ON users(ka_id);124125CREATE TABLE IF NOT EXISTS geocode_cache (126 address TEXT PRIMARY KEY, -- adresse/ville normalisée (clé de cache)127 lat REAL,128 lng REAL,129 provider TEXT,130 failed INTEGER DEFAULT 0,131 ts REAL132);133"""134135# Colonnes ajoutées après la v1 — migration automatique des bases existantes.136_MIGRATIONS: dict[str, dict[str, str]] = {137 "jobs": {138 # Phase 2 (2026-08-19) — enrichissement des connecteurs139 "company_logo": "TEXT", # URL du logo employeur (breezy, JSON-LD…)140 "language": "TEXT", # fr | en | bilingue (source ou heuristique)141 "apply_url": "TEXT", # lien de candidature directe142 "title_clean": "TEXT", # titre d'affichage normalisé (source intacte)143 "quarantine": "TEXT", # JSON des motifs qualité (NULL = publiable)144 # Phase 3 (2026-08-25) — enrichissement : séniorité du poste145 "seniority": "TEXT", # stage | junior | intermediaire | senior | direction146 },147 "sync_log": {},148}149150151def connect() -> sqlite3.Connection:152 DB_PATH.parent.mkdir(parents=True, exist_ok=True)153 con = sqlite3.connect(DB_PATH, timeout=30.0)154 con.row_factory = sqlite3.Row155 con.execute("PRAGMA busy_timeout=30000") # BD vivante (sync horaire)156 con.executescript(_SCHEMA)157 for table, cols in _MIGRATIONS.items():158 existing = {r["name"] for r in con.execute(f"PRAGMA table_info({table})")}159 for col, decl in cols.items():160 if col not in existing:161 con.execute(f"ALTER TABLE {table} ADD COLUMN {col} {decl}")162 con.commit()163 return con164165166# ---------------------------------------------------------------------------167# Synchronisation d'une source168# ---------------------------------------------------------------------------169170def _drift_alert(con: sqlite3.Connection, source: str, found: int,171 null_salary_rate: float) -> str | None:172 """Détecte une dérive du connecteur (chute du volume ou des salaires extraits).173174 Retourne un message d'alerte, ou None si tout est normal.175 """176 hist = con.execute(177 "SELECT found, stats FROM sync_log WHERE source=? AND ok=1"178 " ORDER BY ts DESC LIMIT ?", (source, DRIFT_HISTORY)).fetchall()179 if len(hist) < 3:180 return None181 med_found = statistics.median(r["found"] for r in hist)182 if med_found >= DRIFT_MIN_BASE and found <= DRIFT_RATIO * med_found:183 return (f"dérive: {found} offre(s) trouvée(s) contre une médiane de "184 f"{med_found:.0f} — retraits suspendus, vérifier le connecteur")185 if found >= DRIFT_MIN_BASE and null_salary_rate >= 0.95:186 rates = []187 for r in hist:188 try:189 rates.append(json.loads(r["stats"] or "{}")["null_salary_rate"])190 except (KeyError, ValueError, TypeError):191 continue192 if rates and statistics.median(rates) <= 0.5:193 return (f"dérive: {null_salary_rate:.0%} des offres sans salaire "194 f"(habituellement {statistics.median(rates):.0%}) — "195 "le format de la source a probablement changé")196 return None197198199def sync_source(con: sqlite3.Connection, source: str,200 postings: list[JobPosting]) -> dict:201 """Synchronise les offres d'une source.202203 - nouvelle offre -> insertion204 - offre modifiée -> mise à jour (comparaison de content_hash)205 - offre disparue -> miss_count += 1, puis active=0 après MISS_GRACE206 exécutions consécutives (délai de grâce)207 - dérive détectée -> alerte consignée, retraits suspendus208 """209 now = time.time()210 added = updated = 0211 seen_uids = set()212213 n = len(postings)214 null_salary = sum(1 for p in postings if p.salary_min is None)215 null_city = sum(1 for p in postings if not p.city)216 null_salary_rate = round(null_salary / n, 3) if n else 0.0217218 alert = _drift_alert(con, source, n, null_salary_rate)219220 from .normalize import clean_title221 from .quality import issues_for_posting222223 for job in postings:224 seen_uids.add(job.uid)225 h = job.content_hash()226 row = con.execute(227 "SELECT content_hash, salary_min, salary_max FROM jobs WHERE uid=?",228 (job.uid,)).fetchone()229 defects = issues_for_posting(job)230 params = dict(231 uid=job.uid, source=job.source, external_id=job.external_id,232 url=job.url, employer=job.employer, title=job.title,233 description=job.description, address=job.address, city=job.city,234 region=job.region, postal_code=job.postal_code,235 location_label=job.location_label, work_mode=job.work_mode,236 employment_type=job.employment_type,237 salary_min=job.salary_min, salary_max=job.salary_max,238 salary_unit=job.salary_unit, salary_label=job.salary_label,239 salary_year_min=job.salary_year_min(),240 salary_year_max=job.salary_year_max(),241 salary_hour_min=job.salary_hour_min(),242 salary_hour_max=job.salary_hour_max(),243 benefits=json.dumps(job.benefits, ensure_ascii=False),244 requirements=json.dumps(job.requirements, ensure_ascii=False),245 date_posted=job.date_posted, date_deadline=job.date_deadline,246 category=job.category, ats=job.ats,247 details=json.dumps(job.details, ensure_ascii=False),248 lat=job.lat, lng=job.lng, content_hash=h, now=now,249 company_logo=job.company_logo, language=job.language,250 apply_url=job.apply_url, seniority=job.seniority,251 title_clean=clean_title(job.title, job.city),252 quarantine=json.dumps(defects, ensure_ascii=False)253 if defects else None,254 )255 if row is None:256 con.execute(257 """INSERT INTO jobs (uid, source, external_id, url, employer,258 title, description, address, city, region, postal_code,259 location_label, work_mode, employment_type, salary_min,260 salary_max, salary_unit, salary_label, salary_year_min,261 salary_year_max, salary_hour_min, salary_hour_max, benefits,262 requirements, date_posted, date_deadline, category, ats,263 details, lat, lng, content_hash, first_seen, last_seen,264 updated_at, miss_count, active, company_logo, language,265 apply_url, title_clean, quarantine, seniority)266 VALUES (:uid,:source,:external_id,:url,:employer,:title,267 :description,:address,:city,:region,:postal_code,268 :location_label,:work_mode,:employment_type,:salary_min,269 :salary_max,:salary_unit,:salary_label,:salary_year_min,270 :salary_year_max,:salary_hour_min,:salary_hour_max,271 :benefits,:requirements,:date_posted,:date_deadline,272 :category,:ats,:details,:lat,:lng,:content_hash,273 :now,:now,:now,0,1,:company_logo,:language,:apply_url,274 :title_clean,:quarantine,:seniority)""", params)275 if job.salary_min is not None: # point de départ de l'historique276 con.execute(277 "INSERT INTO salary_log (uid, ts, s_min, s_max, unit)"278 " VALUES (?,?,?,?,?)",279 (job.uid, now, job.salary_min, job.salary_max, job.salary_unit))280 added += 1281 elif row["content_hash"] != h:282 # COALESCE : ne jamais écraser des coordonnées géocodées par null.283 # Si l'adresse change, geocode_failed est remis à 0 (correction du284 # bogue relevé chez Lou·Ka) pour permettre un re-géocodage.285 con.execute(286 """UPDATE jobs SET url=:url, employer=:employer, title=:title,287 description=:description,288 geocode_failed=CASE WHEN address<>:address OR city<>:city289 THEN 0 ELSE geocode_failed END,290 address=:address, city=:city, region=:region,291 postal_code=:postal_code, location_label=:location_label,292 work_mode=:work_mode, employment_type=:employment_type,293 salary_min=:salary_min, salary_max=:salary_max,294 salary_unit=:salary_unit, salary_label=:salary_label,295 salary_year_min=:salary_year_min,296 salary_year_max=:salary_year_max,297 salary_hour_min=:salary_hour_min,298 salary_hour_max=:salary_hour_max,299 benefits=:benefits, requirements=:requirements,300 date_posted=:date_posted, date_deadline=:date_deadline,301 category=:category, ats=:ats, details=:details,302 lat=COALESCE(:lat, lat), lng=COALESCE(:lng, lng),303 company_logo=:company_logo, language=:language,304 apply_url=:apply_url, title_clean=:title_clean,305 quarantine=:quarantine, seniority=:seniority,306 content_hash=:content_hash, last_seen=:now,307 updated_at=:now, miss_count=0, active=1308 WHERE uid=:uid""", params)309 if (job.salary_min, job.salary_max) != (row["salary_min"], row["salary_max"]):310 con.execute(311 "INSERT INTO salary_log (uid, ts, s_min, s_max, unit)"312 " VALUES (?,?,?,?,?)",313 (job.uid, now, job.salary_min, job.salary_max, job.salary_unit))314 updated += 1315 else:316 con.execute(317 "UPDATE jobs SET last_seen=?, miss_count=0, active=1 WHERE uid=?",318 (now, job.uid))319320 # Offres de cette source qui n'apparaissent plus : délai de grâce,321 # puis désactivation. Suspendu si une dérive est détectée.322 removed = missed = 0323 if not alert:324 for r in con.execute(325 "SELECT uid, miss_count FROM jobs WHERE source=? AND active=1",326 (source,)).fetchall():327 if r["uid"] in seen_uids:328 continue329 missed += 1330 if r["miss_count"] + 1 >= MISS_GRACE:331 con.execute(332 "UPDATE jobs SET active=0, miss_count=?, updated_at=?"333 " WHERE uid=?", (r["miss_count"] + 1, now, r["uid"]))334 removed += 1335 else:336 con.execute("UPDATE jobs SET miss_count=miss_count+1 WHERE uid=?",337 (r["uid"],))338339 stats = {340 "null_salary_rate": null_salary_rate,341 "null_city_rate": round(null_city / n, 3) if n else 0.0,342 "missed": missed,343 }344 if alert:345 stats["alert"] = alert346 con.execute(347 "INSERT INTO sync_log (source, ts, found, added, updated, removed, ok,"348 " message, stats) VALUES (?,?,?,?,?,?,1,?,?)",349 (source, now, n, added, updated, removed, alert or "ok",350 json.dumps(stats, ensure_ascii=False)))351 con.commit()352 out = {"source": source, "found": n, "added": added,353 "updated": updated, "removed": removed}354 if alert:355 out["alert"] = alert356 return out357358359def expire_past_deadline(con: sqlite3.Connection) -> int:360 """Désactive les offres dont la date limite est dépassée (offres zombies :361 encore affichées sur la page carrière mais plus ouvertes aux candidatures)."""362 today = _dt.date.today().isoformat()363 cur = con.execute(364 "UPDATE jobs SET active=0, updated_at=? WHERE active=1"365 " AND date_deadline IS NOT NULL AND date_deadline < ?",366 (time.time(), today))367 con.commit()368 return cur.rowcount369370371STALE_DAYS = 7 # offres plus jamais revues (connecteur retiré/muet) : archivées372373374def expire_stale(con: sqlite3.Connection, days: int = STALE_DAYS) -> int:375 """Archive (active=0) les offres actives non revues depuis `days` jours.376377 Filet de sécurité complémentaire au délai de grâce MISS_GRACE : couvre le378 cas d'un connecteur SUPPRIMÉ du code (ses offres ne passent plus jamais379 par sync_source et resteraient actives pour toujours — cas « valero »).380 Rien n'est supprimé : les offres sont archivées, réactivables si revues.381 """382 cutoff = time.time() - days * 86400383 cur = con.execute(384 "UPDATE jobs SET active=0, updated_at=? WHERE active=1 AND last_seen<?"385 " AND source<>'depot-direct'", # le dépôt direct n'est jamais resynchronisé386 (time.time(), cutoff))387 con.commit()388 return cur.rowcount389390391def log_failure(con: sqlite3.Connection, source: str, message: str) -> None:392 con.execute(393 "INSERT INTO sync_log (source, ts, found, added, updated, removed, ok, message)"394 " VALUES (?,?,0,0,0,0,0,?)", (source, time.time(), message))395 con.commit()396397398# ---------------------------------------------------------------------------399# Cache des pages détail (« détail si nouveau/modifié »)400# ---------------------------------------------------------------------------401402def get_cached_detail(con: sqlite3.Connection, source: str,403 external_id: str, key: str) -> dict | None:404 """Payload détail mis en cache si la clé (hash liste) n'a pas changé."""405 row = con.execute(406 "SELECT key, payload FROM detail_cache WHERE source=? AND external_id=?",407 (source, external_id)).fetchone()408 if row and row["key"] == key and row["payload"]:409 try:410 return json.loads(row["payload"])411 except ValueError:412 return None413 return None414415416def get_stale_detail(con: sqlite3.Connection, source: str,417 external_id: str) -> dict | None:418 """Payload détail en cache MÊME si la clé a changé — secours quand le419 budget de re-visites est épuisé : un détail périmé vaut mieux qu'une420 description vide (la re-visite aura lieu à un prochain cycle)."""421 row = con.execute(422 "SELECT payload FROM detail_cache WHERE source=? AND external_id=?",423 (source, external_id)).fetchone()424 if row and row["payload"]:425 try:426 return json.loads(row["payload"])427 except ValueError:428 return None429 return None430431432def put_cached_detail(con: sqlite3.Connection, source: str,433 external_id: str, key: str, payload: dict) -> None:434 con.execute(435 "INSERT INTO detail_cache (source, external_id, key, payload, fetched_at)"436 " VALUES (?,?,?,?,?)"437 " ON CONFLICT(source, external_id) DO UPDATE SET"438 " key=excluded.key, payload=excluded.payload, fetched_at=excluded.fetched_at",439 (source, external_id, key, json.dumps(payload, ensure_ascii=False),440 time.time()))441 con.commit()442