# ----------------------------------------------------------------------------- # Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/autodurocher.py : connecteur Entrepôt Auto Durocher # (autodurocher.com), gros marchand indépendant de Mirabel (~200 véhicules, # occasion seulement). # # Plateforme : WordPress + Elementor + JetEngine (CPT « vehicles »). # Le sitemap /fr/sitemap_index.xml expose des vehicles-sitemapN.xml qui ne # recensent QUE l'inventaire courant (/fr/vehicles//). Chaque page # détail embarque un JSON-LD (script class="fute-custom-schema-vehicles", # Organization -> makesOffer -> itemOffered @type Car) : prix, km, # transmission, couleur, carrosserie, rouage, année, marque, modèle — et la # galerie photos (CDN AutoTrader 1s-photomanager). Le no de stock est le # dernier segment du slug. Le JSON contient des sauts de ligne bruts -> # json.loads(strict=False). # # Pages détail en cache BD (clé hebdomadaire) ; AUTOKA_MAX_DETAILS (env) # plafonne les vraies requêtes détail par synchronisation. # ----------------------------------------------------------------------------- from __future__ import annotations import datetime import html as htmllib import json import os import re from ..schema import Vehicle from .base import BaseConnector _SITEMAP_RE = re.compile(r"vehicles-sitemap\d+\.xml$") _VEH_URL_RE = re.compile(r"/fr/vehicles/([a-z0-9-]+)/?$") _STOCK_RE = re.compile(r"-(?:19|20)\d{2}-([a-z0-9]+)$") _SCHEMA_RE = re.compile( r'', re.S) _LD_RE = re.compile( r'', re.S) _IMG_RE = re.compile(r'https://1s-photomanager[^"\'\s\\<>]+', re.I) _FUEL_HINT_RE = re.compile( r"(hybride rechargeable|plug-?in|phev|hybride|hybrid|électrique|" r"electrique|electric|\bev\b|diesel|tdi)", re.I) _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() def _num(value): try: v = float(str(value).replace(",", "").replace(" ", "")) except (TypeError, ValueError): return None return v if v else None class AutoDurocher(BaseConnector): """Entrepôt Auto Durocher — sitemaps vehicles + JSON-LD des pages détail.""" source_id = "autodurocher" base_url = "https://www.autodurocher.com" dealer_name = "Entrepôt Auto Durocher" city = "Mirabel" request_delay = 1.0 max_details = 300 # plafond de vraies requêtes par sync # -- sitemaps ---------------------------------------------------------------- def _vehicle_slugs(self) -> list[str]: index = self.get(f"{self.base_url}/fr/sitemap_index.xml").text maps = [loc.strip() for loc in re.findall(r"([^<]+)", index) if _SITEMAP_RE.search(loc.strip())] slugs: list[str] = [] for sitemap_url in maps: try: xml = self.get(sitemap_url).text except Exception: continue # un sitemap HS ne bloque pas for loc in re.findall(r"([^<]+)", xml): m = _VEH_URL_RE.search(loc.strip()) if m: slugs.append(m.group(1)) return sorted(set(slugs)) # -- page détail -> payload --------------------------------------------------- def _fetch_detail(self, url: str) -> dict: html = self.get(url).text payload: dict = {} blocks = _SCHEMA_RE.findall(html) or _LD_RE.findall(html) for block in blocks: try: data = json.loads(block.strip(), strict=False) except ValueError: continue offer = data.get("makesOffer") or {} car = offer.get("itemOffered") or {} if not isinstance(car, dict) or car.get("@type") != "Car": continue spec = offer.get("priceSpecification") or {} payload["price"] = spec.get("price") payload["car"] = {k: v for k, v in car.items() if not str(k).startswith("@")} break images, seen = [], set() for u in _IMG_RE.findall(html): if u not in seen: seen.add(u) images.append(u) payload["images"] = images[:20] return payload # -- contrat --------------------------------------------------------------- def fetch(self) -> list[Vehicle]: slugs = self._vehicle_slugs() week = datetime.date.today().isocalendar() cache_key = f"v1:{week.year}w{week.week}" # revalidation hebdomadaire cap = int(os.environ.get("AUTOKA_MAX_DETAILS", self.max_details)) vehicles: list[Vehicle] = [] real_fetches = 0 for slug in slugs: url = f"{self.base_url}/fr/vehicles/{slug}/" 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, slug, cache_key) if data is None: continue else: before = self._last_request try: data = self.detail(slug, cache_key, _fetch) except Exception: # page disparue (vendu) ou erreur ponctuelle : continuer if self._last_request != before: real_fetches += 1 continue if self._last_request != before: real_fetches += 1 if not data or not data.get("car"): continue veh = self._to_vehicle(slug, url, data) if veh is not None: vehicles.append(veh) return vehicles # -- payload -> Vehicle ------------------------------------------------------ def _to_vehicle(self, slug: str, url: str, d: dict) -> Vehicle | None: car = d.get("car") or {} make = _clean(car.get("manufacturer") or car.get("brand")) model = _clean(car.get("model")) name = _clean(str(car.get("name") or "").split("\n")[0]) if not (make or model or name): return None body = _clean(car.get("bodyType")) if _EXCLUDE_RE.search(f"{make} {model} {body} {name}"): return None year = None m = re.search(r"(19|20)\d{2}", str(car.get("vehicleModelDate") or car.get("modelDate") or "")) if m: year = int(m.group(0)) # « 2018 - Audi A3 » -> titre standard title = re.sub(r"^(?:19|20)\d{2}\s*-\s*", "", name) title = " ".join(x for x in (title or f"{make} {model}", str(year or "")) if x) price = _num(d.get("price")) km = _num(car.get("mileageFromOdometer")) stock = "" m = _STOCK_RE.search(slug) if m: stock = m.group(1).upper() engine = "" eng = car.get("vehicleEngine") if isinstance(eng, dict): engine = _clean(eng.get("name")) description = _clean(car.get("description"))[:4000] # indice carburant : le NOM seulement (la description parle de # « sièges électriques » etc. -> faux positifs) fuel = "" m = _FUEL_HINT_RE.search(name) if m: fuel = m.group(1) images = d.get("images") or [] ld_img = car.get("image") if not images and isinstance(ld_img, str): images = [ld_img] return Vehicle( source=self.source_id, external_id=slug, url=url, title=title, make=make, model=model, 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=_clean(car.get("vehicleTransmission")), fuel=fuel, drivetrain=_clean(car.get("driveWheelConfiguration")), body_type=body, exterior_color=_clean(car.get("color")), engine=engine, stock_number=stock, dealer_name=self.dealer_name, city=self.city, description=description, images=images, )