# ----------------------------------------------------------------------------- # Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/megacentre.py : connecteur Méga Centre de liquidation # (megacentredeliquidation.com), place de marché d'occasion du Groupe # Laplante — Québec, Trois-Rivières, Rimouski, Matane, Mont-Joli, Amqui, # Rawdon (Lanaudière), etc. # # Plateforme : Gatsby (site statique sur Fastly), inventaire EvalAuto. # Le sitemap (/sitemap-index.xml -> /sitemap-0.xml) recense chaque page # véhicule /auto-usage//. Pour chaque slug, le # JSON Gatsby /page-data/auto-usage//page-data.json expose l'objet # `usedCar` complet : prix, odomètre, NIV, stock, transmission, rouage, # carburant, carrosserie, couleurs, moteur, portes, places, options, # description, photos, CarProof ET la concession réelle du véhicule # (nameDealer + cityDealer — la ville par véhicule). # # Les page-data sont mises en cache BD (clé hebdomadaire, même logique que # les connecteurs D2C) : seuls les nouveaux véhicules et la revalidation # hebdomadaire génèrent de vraies requêtes. AUTOKA_MAX_DETAILS (env) # plafonne les vraies requêtes détail par synchronisation. # ----------------------------------------------------------------------------- from __future__ import annotations import datetime import html as htmllib import os import re from ..schema import Vehicle from .base import BaseConnector # pages véhicule : /auto-usage/--/ (un seul segment) _VEH_URL_RE = re.compile( r"^https://www\.megacentredeliquidation\.com/auto-usage/" r"([a-z0-9-]+-(?:19|20)\d{2}-[a-z0-9-]+)/$") # le groupe liste aussi quelques non-automobiles (Can-Am Ryker, VR…) _EXCLUDE_RE = re.compile( r"can-?am|spyder|ryker|harley|ski-?doo|sea-?doo|polaris|kawasaki|" r"gulf ?stream|jayco|keystone|forest river|motoris|roulotte|remorque|" r"motorcycle|snowmobile|atv|\brv\b|trailer|motorhome", re.I) def _clean_html(text: str | None) -> str: """HTML avec entités (é…) -> texte brut compact.""" if not text: return "" out = htmllib.unescape(str(text)) out = re.sub(r"<[^>]+>", " ", out) return re.sub(r"\s{2,}", " ", out).strip() class MegaCentreDeLiquidation(BaseConnector): """Méga Centre de liquidation — sitemap + page-data Gatsby (usedCar).""" source_id = "megacentredeliquidation" base_url = "https://www.megacentredeliquidation.com" dealer_name = "Méga Centre de liquidation" request_delay = 1.0 max_details = 600 # plafond de vraies requêtes par sync # -- sitemap ------------------------------------------------------------- def _vehicle_slugs(self) -> list[str]: index = self.get(f"{self.base_url}/sitemap-index.xml").text maps = re.findall(r"([^<]+)", index) \ or [f"{self.base_url}/sitemap-0.xml"] slugs: list[str] = [] for sitemap_url in maps: xml = self.get(sitemap_url.strip()).text for loc in re.findall(r"([^<]+)", xml): m = _VEH_URL_RE.match(loc.strip()) if m: slugs.append(m.group(1)) return sorted(set(slugs)) # -- page-data Gatsby -> payload usedCar ---------------------------------- def _fetch_detail(self, slug: str) -> dict: url = f"{self.base_url}/page-data/auto-usage/{slug}/page-data.json" try: data = self.get(url).json() except ValueError: return {} result = data.get("result") or {} used_car = (result.get("data") or {}).get("usedCar") if not isinstance(used_car, dict): return {} used_car.pop("coordinates", None) # gros bloc inutile used_car.pop("loanPaiement", None) return {"usedCar": used_car, "latestModification": (result.get("pageContext") or {}).get("latestModification")} # -- 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: def _fetch(s=slug): return self._fetch_detail(s) # plafond atteint : ne servir que le cache (fetch réel interdit) if real_fetches >= cap: 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 : ne jamais # bloquer la source entière if self._last_request != before: real_fetches += 1 continue if self._last_request != before: # une vraie requête a eu lieu real_fetches += 1 if not data or not data.get("usedCar"): continue veh = self._to_vehicle(slug, data["usedCar"]) if veh is not None: vehicles.append(veh) return vehicles # -- usedCar -> Vehicle ------------------------------------------------------ def _to_vehicle(self, slug: str, d: dict) -> Vehicle | None: make = str(d.get("brand") or "").strip() model = str(d.get("labelModel_fr") or d.get("model") or "").strip() body = str(d.get("labelBodytype_fr") or d.get("bodytype") or "").strip() if _EXCLUDE_RE.search(f"{make} {model} {body}"): # non-automobile return None year = d.get("year") trim = str(d.get("trim") or "").strip() # certains labels de modèle incluent déjà la version (X1 XDRIVE28I) title_trim = "" if trim.lower() in model.lower() else trim title = " ".join(str(x) for x in (make, model, title_trim, year or "") if x) if not title: return None price = d.get("price") try: price = float(price) if price else None except (TypeError, ValueError): price = None km = d.get("odometer") try: km = float(km) if km not in (None, "") else None except (TypeError, ValueError): km = None unit = str(d.get("odometer_Type") or "km").lower() if km is not None and unit.startswith("mi"): km = round(km * 1.609344) images = d.get("images") or [] if isinstance(images, str): images = [images] images = [u for u in images if isinstance(u, str)] if not images and d.get("image"): images = [str(d["image"])] features = [f.strip() for f in str(d.get("options") or "").split("|") if f.strip()] doors = d.get("doors") seats = d.get("passengers") details: dict = {"branch": str(d.get("nameDealer") or "")} for src, dst in (("accident", "accidented"), ("certified", "certified"), ("uniqueOwner", "unique_owner")): if d.get(src) is not None: details[dst] = bool(d[src]) if d.get("cityConsumption"): details["consumption_city_l_100km"] = d["cityConsumption"] if d.get("hightwayConsumption"): details["consumption_hwy_l_100km"] = d["hightwayConsumption"] if d.get("inventoryID"): details["inventory_id"] = str(d["inventoryID"]) return Vehicle( source=self.source_id, external_id=slug, url=f"{self.base_url}/auto-usage/{slug}/", title=title, make=make, model=model, trim=trim, year=int(year) if year else None, 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("transmission") or ""), fuel=str(d.get("labelEngine_fr") or d.get("engine") or ""), drivetrain=str(d.get("drivetrain") or ""), body_type=body, exterior_color=str(d.get("labelColorExt_fr") or ""), interior_color=str(d.get("labelColorInt_fr") or ""), engine=str(d.get("motor") or ""), doors=int(doors) if doors else None, seats=int(seats) if seats else None, vin=str(d.get("niv") or ""), stock_number=str(d.get("stockNO") or ""), dealer_name=str(d.get("nameDealer") or "").strip() or self.dealer_name, city=str(d.get("cityDealer") or "").strip(), description=_clean_html(d.get("description"))[:4000], features=features[:60], details=details, images=images[:20], carfax_url=str(d.get("carproofURL") or ""), )