# ----------------------------------------------------------------------------- # Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/jlkauto.py : JLK Auto (Montréal) — plateforme EDealer # (edealer.ca ; WordPress + inventaire EDealer, CDN media.edealer.ca). # # Stratégie : le sitemap Yoast `/inventory-listing-sitemap.xml` recense les # pages véhicule (…-vdp/) en FR et EN. Chaque page détail (VDP) # expose un JSON-LD schema.org @graph avec un nœud @type [Product, Car] # complet : prix, VIN, km, transmission, carrosserie, couleurs, moteur. # Pages détail mises en cache BD (clé hebdomadaire), comme les sites D2C. # ----------------------------------------------------------------------------- from __future__ import annotations import datetime import json import os import re from ..schema import Vehicle from .base import BaseConnector # …/inventaire/2005-chrysler-crossfire-limitee-jGKAbGVgRXus0SSkyI35kgvdp/ # le token (id EDealer) est le dernier segment alphanumérique SANS tiret : # le même véhicule a un slug différent en FR et en EN, mais le même token _VDP_RE = re.compile(r"-([A-Za-z0-9]{12,})vdp/?$") _LD_RE = re.compile( r'', re.S) _DRIVE_MAP = { "FrontWheelDriveConfiguration": "Traction", "RearWheelDriveConfiguration": "Propulsion", "AllWheelDriveConfiguration": "Intégrale / 4x4", "FourWheelDriveConfiguration": "Intégrale / 4x4", } def _qty(node) -> float | None: """QuantitativeValue -> float ; accepte aussi un scalaire nu.""" value = node.get("value") if isinstance(node, dict) else node try: v = float(value) return v if v > 0 else None except (TypeError, ValueError): return None class JLKAuto(BaseConnector): source_id = "jlkauto" base_url = "https://www.jlkauto.ca" dealer_name = "JLK Auto" city = "Montréal" request_delay: float = 1.0 max_details = 400 # -- sitemap ---------------------------------------------------------------- def _vehicle_urls(self) -> list[str]: xml = self.get(f"{self.base_url}/inventory-listing-sitemap.xml").text by_token: dict[str, str] = {} for loc in re.findall(r"([^<]+)", xml): loc = loc.strip() m = _VDP_RE.search(loc) if not m: continue # pages listes (…vlp) et autres token = m.group(1) # préférer la page FR (/inventaire/) à la page EN (/inventory/) if token not in by_token or "/inventaire/" in loc: by_token[token] = loc return [by_token[t] for t in sorted(by_token)] # -- page détail -> nœud JSON-LD [Product, Car] ------------------------------ def _fetch_detail(self, url: str) -> dict: html = self.get(url).text for block in _LD_RE.findall(html): try: data = json.loads(block.strip()) except ValueError: continue nodes = data.get("@graph") if isinstance(data, dict) else None if nodes is None: nodes = data if isinstance(data, list) else [data] for node in nodes: types = node.get("@type") types = types if isinstance(types, list) else [types] if "Car" in types or "Vehicle" in types: # galerie complète depuis le CDN EDealer (JSON-LD n'a # qu'une photo) imgs = re.findall( r'https://media\.edealer\.ca/[^"\'\\\s]+\.(?:webp|jpe?g|png)', html) seen: set[str] = set() node["_gallery"] = [u for u in imgs if not (u in seen or seen.add(u))][:20] return node return {} # -- contrat ----------------------------------------------------------------- def fetch(self) -> list[Vehicle]: urls = self._vehicle_urls() week = datetime.date.today().isocalendar() cache_key = f"v1:{week.year}w{week.week}" cap = int(os.environ.get("AUTOKA_MAX_DETAILS", self.max_details)) vehicles: list[Vehicle] = [] real_fetches = 0 for url in urls: ext_id = _VDP_RE.search(url).group(1) def _fetch(u=url): return self._fetch_detail(u) if real_fetches >= cap: # plafond : cache seulement from .. import db if self._detail_con is None: self._detail_con = db.connect() data = db.get_cached_detail(self._detail_con, self.source_id, ext_id, cache_key) if data is None: continue else: before = self._last_request data = self.detail(ext_id, cache_key, _fetch) if self._last_request != before: real_fetches += 1 if not data: continue veh = self._to_vehicle(ext_id, url, data) if veh is not None: vehicles.append(veh) return vehicles def _to_vehicle(self, ext_id: str, url: str, d: dict) -> Vehicle | None: # véhicules usagés uniquement cond = str(d.get("itemCondition") or "") if cond and "Used" not in cond: return None title = " ".join(str(d.get("name") or "").split()) if not title: return None offers = d.get("offers") or {} price = None try: p = float(offers.get("price")) price = p if p > 0 else None except (TypeError, ValueError): pass km = _qty(d.get("mileageFromOdometer")) brand = d.get("brand") or {} make = brand.get("name") if isinstance(brand, dict) else str(brand) drivetrain = "" drive_raw = str(d.get("driveWheelConfiguration") or "") for key, canon in _DRIVE_MAP.items(): if key in drive_raw: drivetrain = canon break year = None for k in ("vehicleModelDate", "modelDate", "productionDate"): try: year = int(str(d.get(k))[:4]) break except (TypeError, ValueError): continue engine = d.get("vehicleEngine") or {} engine_name = engine.get("name") if isinstance(engine, dict) else "" images = d.get("_gallery") or [] if not images and isinstance(d.get("image"), str): images = [d["image"]] doors = d.get("numberOfDoors") doors = doors if isinstance(doors, int) else None try: seats = int(str(d.get("vehicleSeatingCapacity"))) except (TypeError, ValueError): seats = None details = {k: d[k] for k in ("numberOfForwardGears", "wheelbase", "fuelConsumption") if d.get(k)} return Vehicle( source=self.source_id, external_id=ext_id, url=url, title=title, make=str(make or ""), model=str(d.get("model") or ""), trim=str(d.get("vehicleConfiguration") or ""), year=year, price=price, price_label=(f"{price:,.0f} $".replace(",", " ") if price else ""), mileage_km=km, mileage_label=(f"{km:,.0f} km".replace(",", " ") if km else ""), transmission=str(d.get("vehicleTransmission") or ""), fuel=str(d.get("fuelType") or ""), drivetrain=drivetrain, body_type=str(d.get("bodyType") or ""), exterior_color=str(d.get("color") or ""), interior_color=str(d.get("vehicleInteriorColor") or ""), engine=str(engine_name or ""), doors=doors, seats=seats, vin=str(d.get("vehicleIdentificationNumber") or ""), stock_number=str(d.get("sku") or ""), dealer_name=self.dealer_name, city=self.city, description=" ".join(str(d.get("description") or "").split())[:4000], details=details, images=images, )