# ----------------------------------------------------------------------------- # Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # recalls.py : rappels de sécurité — Base de données des rappels de véhicules # de Transports Canada (données ouvertes, gratuit) # # Source : dump mensuel complet CSV du jeu de données « Vehicle Recalls # Database » (ouvert.canada.ca, jeu 1ec92326-47ef-4110-b7ca-959fab03f96d) : # https://opendatatc.tc.canada.ca/vrdb_full_monthly.csv (~200 Mo) # 1 ligne = 1 rappel × marque × modèle × année-modèle. L'API REST officielle # (tc.api.canada.ca) exige une clé délivrée manuellement — le CSV ouvert est # la voie sans clé, rafraîchi mensuellement par TC. # # Table `recalls` (make/model/year normalisés en MAJUSCULES sans accents) + # croisement par (make, model, year) avec les annonces : le modèle TC est # souvent plus court que le nôtre (« CIVIC » vs « Civic Sport Touring ») — # on matche si l'un préfixe l'autre au premier mot près, année exacte. # # Peuplement one-shot / mensuel : python3 -m autoka.recalls [--limit-years N] # ----------------------------------------------------------------------------- from __future__ import annotations import csv import io import sqlite3 import sys import time import unicodedata import requests CSV_URL = "https://opendatatc.tc.canada.ca/vrdb_full_monthly.csv" _SCHEMA = """ CREATE TABLE IF NOT EXISTS recalls ( recall_number TEXT NOT NULL, make TEXT NOT NULL, -- MAJUSCULES sans accents (norme TC) model TEXT NOT NULL, -- MAJUSCULES sans accents (norme TC) year INTEGER, -- année-modèle visée date TEXT, -- date du rappel (YYYY-MM-DD) component TEXT, -- système visé (FR, repli EN) description TEXT, -- description (FR, repli EN) category TEXT, -- Car / Truck / Motorcycle... (norme TC) units_affected INTEGER, PRIMARY KEY (recall_number, make, model, year) ); CREATE INDEX IF NOT EXISTS idx_recalls_mmy ON recalls(make, model, year); """ # alias marques Auto-Ka -> marque TC quand l'orthographe diffère _MAKE_ALIASES = { "MERCEDES-BENZ": ("MERCEDES-BENZ", "MERCEDES"), "LAND ROVER": ("LAND ROVER", "LANDROVER"), "VOLKSWAGEN": ("VOLKSWAGEN", "VW"), "CHEVROLET": ("CHEVROLET", "CHEV"), "MINI": ("MINI", "BMW MINI"), } def _norm(text: str | None) -> str: """MAJUSCULES sans accents, espaces réduits — clé de croisement.""" if not text: return "" t = unicodedata.normalize("NFKD", text) t = "".join(c for c in t if not unicodedata.combining(c)) return " ".join(t.upper().split()) def ensure_schema(con: sqlite3.Connection) -> None: con.executescript(_SCHEMA) con.commit() # --------------------------------------------------------------------------- # Peuplement depuis le dump CSV mensuel de Transports Canada # --------------------------------------------------------------------------- def _fr_or_en(fr: str, en: str) -> str: fr = (fr or "").strip() if fr and fr.lower() not in ("translation not available", "non saisie"): return fr return (en or "").strip() def populate(con: sqlite3.Connection, csv_path: str | None = None, min_year: int = 1980) -> dict: """(Re)charge la table depuis le dump complet (téléchargé si besoin). `min_year` écarte les années-modèles antérieures au parc plausible d'Auto-Ka (schema.py borne les annonces à 1980+). """ ensure_schema(con) t0 = time.time() if csv_path: fh = open(csv_path, encoding="utf-8", errors="replace", newline="") else: print(f"[auto-ka] rappels TC : téléchargement {CSV_URL} ...") resp = requests.get(CSV_URL, timeout=600) resp.raise_for_status() fh = io.StringIO(resp.content.decode("utf-8", errors="replace")) con.execute("DELETE FROM recalls") n = kept = 0 with fh: for row in csv.DictReader(fh): n += 1 try: year = int(float(row.get("YEAR") or 0)) or None except ValueError: year = None if year is not None and year < min_year: continue make = _norm(row.get("MAKE_NAME_NM")) model = _norm(row.get("MODEL_NAME_NM")) if not make or not row.get("RECALL_NUMBER_NUM"): continue try: units = int(float((row.get("UNIT_AFFECTED_NBR") or "0") .replace(",", ""))) except ValueError: units = 0 con.execute( """INSERT OR REPLACE INTO recalls (recall_number, make, model, year, date, component, description, category, units_affected) VALUES (?,?,?,?,?,?,?,?,?)""", (row["RECALL_NUMBER_NUM"], make, model, year, (row.get("RECALL_DATE_DTE") or "")[:10], _fr_or_en(row.get("SYSTEM_TYPE_FTXT"), row.get("SYSTEM_TYPE_ETXT")), _fr_or_en(row.get("COMMENT_FTXT"), row.get("COMMENT_ETXT")), row.get("CATEGORY_ETXT") or "", units)) kept += 1 con.commit() total = con.execute("SELECT COUNT(*) c FROM recalls").fetchone()["c"] out = {"csv_rows": n, "kept": kept, "recalls_rows": total, "seconds": round(time.time() - t0, 1)} print(f"[auto-ka] rappels TC : {out}") return out # --------------------------------------------------------------------------- # Croisement avec une annonce (make/model/year normalisés) # --------------------------------------------------------------------------- def for_vehicle(con: sqlite3.Connection, make: str, model: str, year: int | None) -> list[dict]: """Rappels TC visant ce (marque, modèle, année) — année exacte requise. Le modèle TC est générique (« CIVIC », « F-150 ») alors que le nôtre porte parfois la version ; on matche par préfixe dans les deux sens, ancré sur le premier mot pour éviter les faux positifs. """ if not make or not model or year is None: return [] ensure_schema(con) n_make, n_model = _norm(make), _norm(model) makes = _MAKE_ALIASES.get(n_make, (n_make,)) first_word = n_model.split()[0] q = f"""SELECT recall_number, make, model, year, date, component, description, category, units_affected FROM recalls WHERE make IN ({','.join('?' * len(makes))}) AND year=? AND (model=? OR model LIKE ? OR ? LIKE model || ' %' OR (length(model) >= 3 AND ? LIKE model || '%')) ORDER BY date DESC""" rows = con.execute(q, (*makes, year, n_model, f"{first_word} %", n_model, first_word)).fetchall() seen: set[str] = set() out = [] for r in rows: if r["recall_number"] in seen: continue seen.add(r["recall_number"]) out.append(dict(r)) return out if __name__ == "__main__": from . import db con = db.connect() path = sys.argv[1] if len(sys.argv) > 1 else None populate(con, csv_path=path) con.close()