# ----------------------------------------------------------------------------- # Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/desmeules.py : connecteur Desmeules Chrysler (desmeules.ca), # gros concessionnaire Chrysler/Dodge/Jeep/RAM de Laval — ~220 véhicules # d'occasion. # # Plateforme : ColdFusion maison + inventaire AutoJini (photos # images.autojini.net). La page occasion expose un flux RSS complet : # /fr/vehicules-doccasion/?fuseaction=inventory.feedRSS&ajaxCall=1 # -> un par véhicule : titre (année marque modèle + prix), lien # détail (…-v.html), no de stock, VIN, photo, longue description. # # La page détail complète : kilométrage (detailsMileageData), moteur, # transmission/entraînement (microdata) et galerie photos. Elle porte aussi # la liste d'équipements AutoJini (« Équipement standard », divs .column : # h3 catégorie + ul d'items) dont la section « Transmission & Moteur » # contient le carburant (« Essence », « Plug-in électric / gas »…) et la # motricité (« Traction Intégrale », « Quatre roues motrices »…), ainsi # qu'une boîte « Catégorie » (Car/VUS/SUV/Truck) pour la carrosserie. # Ces pages sont mises en cache BD à vie (clé stable — les specs ne # changent pas ; le prix frais vient du RSS). AUTOKA_MAX_DETAILS (env) # plafonne les vraies requêtes détail par synchronisation. # ----------------------------------------------------------------------------- from __future__ import annotations import html as htmllib import os import re from ..schema import (Vehicle, normalize_body, normalize_drivetrain, normalize_fuel, strip_accents) from .base import BaseConnector _ITEM_RE = re.compile(r"(.*?)", re.S) _TITLE_RE = re.compile(r"\s*<!\[CDATA\[(.*?)\]\]>", re.S) _LINK_RE = re.compile(r'href="(https?://[^"]+-v(\d+)\.html)"') _STOCK_RE = re.compile(r"Stock #:\s*([^<]+)</div>") _VIN_RE = re.compile(r"VIN #:\s*([^<]+)</div>") _IMG_RE = re.compile(r'img[^>]*src="([^"]+)"') _PRICE_RE = re.compile(r"-\s*\$([\d,]+)\s*$") # page détail (la transmission/couleur vient de la meta description : # « … - Engine: V8 4.0L Cylinder Engine - Color : . - Transmission: Automatique ») _KM_RE = re.compile(r'detailsMileageData">\s*([\d,\s]+)\s*km', re.I) _ENGINE_RE = re.compile(r"Moteur:\s*</div>\s*<div[^>]*>\s*([^<]+)", re.S) _TRANS_RE = re.compile(r"Transmission:\s*([A-Za-zÀ-ÿ][A-Za-zÀ-ÿ ]*)") _DRIVE_RE = re.compile(r"Entra[îi]nement\s*:\s*<span>\s*([^<]+)", re.I) _COLOR_RE = re.compile(r"Color\s*:\s*([A-Za-zÀ-ÿ][^.<\"]*)[.<\"]") _GALLERY_RE = re.compile( r'https?://images\.autojini\.[a-z]+/[^"\s\\<>]+?\.jpg(?!/)', re.I) # liste d'équipements AutoJini : divs .column (h3 catégorie + ul d'items) _FEATURE_SECTION_RE = re.compile( r'<div class="column[^"]*">\s*<h3>([^<]+)</h3>\s*<ul>(.*?)</ul>', re.S) _LI_RE = re.compile(r"<li>(.*?)</li>", re.S) # boîte « Catégorie » (Car / VUS / SUV / Truck…) -> carrosserie _CATEGORY_RE = re.compile( r'colorBoxHead">\s*Cat(?:é|é)gorie\s*</div>' r'.*?colorBoxText">(.*?)</div>', re.S) _EXCLUDE_RE = re.compile( r"moto(?:cyclette|neige|marine)?\b|\bvtt\b|scooter|spyder|ryker|" r"can-?am|ski-?doo|sea-?doo|roulotte|remorque|\bvr\b|campeur", re.I) def _clean(text) -> str: if not text: return "" out = htmllib.unescape(str(text)) out = re.sub(r"<[^>]+>", " ", out) return re.sub(r"\s+", " ", out).strip() # clé de cache détail stable (bump si le parsing change) # v3 : + équipements, carburant/motricité (Transmission & Moteur), carrosserie _DETAIL_KEY = "v3" class DesmeulesChrysler(BaseConnector): """Desmeules Chrysler — flux RSS inventaire + pages détail (km, specs).""" source_id = "desmeules" base_url = "https://www.desmeules.ca" dealer_name = "Desmeules Chrysler" city = "Laval" request_delay = 1.0 max_details = 300 # plafond de vraies requêtes par sync # -- flux RSS ---------------------------------------------------------------- def _rss_items(self) -> list[dict]: xml = self.get( f"{self.base_url}/fr/vehicules-doccasion/" "?fuseaction=inventory.feedRSS&ajaxCall=1").text items: list[dict] = [] for chunk in _ITEM_RE.findall(xml): link = _LINK_RE.search(chunk) if not link: continue entry: dict = {"url": link.group(1).replace("http://", "https://"), "id": link.group(2)} m = _TITLE_RE.search(chunk) title = _clean(m.group(1)) if m else "" m = _PRICE_RE.search(title) if m: entry["price"] = float(m.group(1).replace(",", "")) title = title[:m.start()].strip() title = re.sub(r"^For Sale\s+", "", title, flags=re.I) entry["title"] = title m = _STOCK_RE.search(chunk) if m: entry["stock"] = _clean(m.group(1)) m = _VIN_RE.search(chunk) if m: entry["vin"] = _clean(m.group(1)) m = _IMG_RE.search(chunk) if m: entry["photo"] = m.group(1).split("/Resize")[0] # description longue : le 2e bloc CDATA de la description blocks = re.findall(r"<!\[CDATA\[(.*?)\]\]>", chunk, re.S) if len(blocks) >= 3: entry["description"] = _clean(blocks[2])[:4000] items.append(entry) return items # -- page détail -> payload --------------------------------------------------- def _fetch_detail(self, url: str) -> dict: html = self.get(url).text payload: dict = {} m = _KM_RE.search(html) if m: payload["km"] = float(re.sub(r"[^\d]", "", m.group(1))) m = _ENGINE_RE.search(html) if m: payload["engine"] = _clean(m.group(1)) m = _TRANS_RE.search(html) if m: payload["transmission"] = _clean(m.group(1)) m = _DRIVE_RE.search(html) if m: payload["drivetrain"] = _clean(m.group(1)) m = _COLOR_RE.search(html) if m: payload["color"] = _clean(m.group(1)) # boîte « Catégorie » -> carrosserie canonique (Car reste ambigu) m = _CATEGORY_RE.search(html) if m: body = normalize_body(_clean(m.group(1))) if body: payload["body"] = body # équipements par catégorie ; « Transmission & Moteur » porte le # carburant et la motricité, absents partout ailleurs features: list[str] = [] for head, ul in _FEATURE_SECTION_RE.findall(html): items = [_clean(li) for li in _LI_RE.findall(ul)] items = [it for it in items if it] features.extend(items) if not strip_accents(head).lower().startswith("transmission"): continue for it in items: if "fuel" not in payload: fuel = normalize_fuel(it) if fuel: payload["fuel"] = fuel if "drivetrain" not in payload: low = strip_accents(it).lower() if "quatre roues" in low or "4 roues" in low: payload["drivetrain"] = "Intégrale / 4x4" elif "controle" not in low: # « Contrôle de traction » drive = normalize_drivetrain(it) if drive: payload["drivetrain"] = drive if features: payload["features"] = features[:60] images, seen = [], set() for u in _GALLERY_RE.findall(html): base = u.replace("_x.jpg", ".jpg") if base not in seen: seen.add(base) images.append(base) payload["images"] = images[:20] return payload # -- contrat --------------------------------------------------------------- def fetch(self) -> list[Vehicle]: cap = int(os.environ.get("AUTOKA_MAX_DETAILS", self.max_details)) vehicles: list[Vehicle] = [] real_fetches = 0 for item in self._rss_items(): if _EXCLUDE_RE.search(item.get("title", "")): continue detail: dict = {} if real_fetches >= cap: # plafond : cache seulement from .. import db if self._detail_con is None: self._detail_con = db.connect() detail = db.get_cached_detail(self._detail_con, self.source_id, item["id"], _DETAIL_KEY) or {} else: def _fetch(u=item["url"]): return self._fetch_detail(u) before = self._last_request try: detail = self.detail(item["id"], _DETAIL_KEY, _fetch) except Exception: detail = {} # page disparue : données RSS if self._last_request != before: real_fetches += 1 veh = self._to_vehicle(item, detail or {}) if veh is not None: vehicles.append(veh) return vehicles # -- item RSS + détail -> Vehicle ---------------------------------------------- def _to_vehicle(self, item: dict, detail: dict) -> Vehicle | None: title = item.get("title") or "" if not title: return None price = item.get("price") km = detail.get("km") images = list(detail.get("images") or []) if not images and item.get("photo"): images = [item["photo"]] return Vehicle( source=self.source_id, external_id=item["id"], url=item["url"], title=title, # « 2021 Aston Martin DBX … » 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=detail.get("transmission", ""), fuel=detail.get("fuel", ""), drivetrain=detail.get("drivetrain", ""), body_type=detail.get("body", ""), exterior_color=detail.get("color", ""), engine=detail.get("engine", ""), vin=item.get("vin", ""), stock_number=item.get("stock", ""), dealer_name=self.dealer_name, city=self.city, description=item.get("description", ""), features=list(detail.get("features") or []), images=images, )