spb/auto-ka Public

1# -----------------------------------------------------------------------------2# Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/megacentre.py : connecteur Méga Centre de liquidation5#   (megacentredeliquidation.com), place de marché d'occasion du Groupe6#   Laplante — Québec, Trois-Rivières, Rimouski, Matane, Mont-Joli, Amqui,7#   Rawdon (Lanaudière), etc.8#9#   Plateforme : Gatsby (site statique sur Fastly), inventaire EvalAuto.10#   Le sitemap (/sitemap-index.xml -> /sitemap-0.xml) recense chaque page11#   véhicule /auto-usage/<marque-modele-annee-stock>/. Pour chaque slug, le12#   JSON Gatsby /page-data/auto-usage/<slug>/page-data.json expose l'objet13#   `usedCar` complet : prix, odomètre, NIV, stock, transmission, rouage,14#   carburant, carrosserie, couleurs, moteur, portes, places, options,15#   description, photos, CarProof ET la concession réelle du véhicule16#   (nameDealer + cityDealer — la ville par véhicule).17#18#   Les page-data sont mises en cache BD (clé hebdomadaire, même logique que19#   les connecteurs D2C) : seuls les nouveaux véhicules et la revalidation20#   hebdomadaire génèrent de vraies requêtes. AUTOKA_MAX_DETAILS (env)21#   plafonne les vraies requêtes détail par synchronisation.22# -----------------------------------------------------------------------------23from __future__ import annotations2425import datetime26import html as htmllib27import os28import re2930from ..schema import Vehicle31from .base import BaseConnector3233# pages véhicule : /auto-usage/<slug>-<année>-<stock>/ (un seul segment)34_VEH_URL_RE = re.compile(35    r"^https://www\.megacentredeliquidation\.com/auto-usage/"36    r"([a-z0-9-]+-(?:19|20)\d{2}-[a-z0-9-]+)/$")3738# le groupe liste aussi quelques non-automobiles (Can-Am Ryker, VR…)39_EXCLUDE_RE = re.compile(40    r"can-?am|spyder|ryker|harley|ski-?doo|sea-?doo|polaris|kawasaki|"41    r"gulf ?stream|jayco|keystone|forest river|motoris|roulotte|remorque|"42    r"motorcycle|snowmobile|atv|\brv\b|trailer|motorhome", re.I)434445def _clean_html(text: str | None) -> str:46    """HTML avec entités (&eacute;…) -> texte brut compact."""47    if not text:48        return ""49    out = htmllib.unescape(str(text))50    out = re.sub(r"<[^>]+>", " ", out)51    return re.sub(r"\s{2,}", " ", out).strip()525354class MegaCentreDeLiquidation(BaseConnector):55    """Méga Centre de liquidation — sitemap + page-data Gatsby (usedCar)."""5657    source_id = "megacentredeliquidation"58    base_url = "https://www.megacentredeliquidation.com"59    dealer_name = "Méga Centre de liquidation"60    request_delay = 1.061    max_details = 600                 # plafond de vraies requêtes par sync6263    # -- sitemap -------------------------------------------------------------64    def _vehicle_slugs(self) -> list[str]:65        index = self.get(f"{self.base_url}/sitemap-index.xml").text66        maps = re.findall(r"<loc>([^<]+)</loc>", index) \67            or [f"{self.base_url}/sitemap-0.xml"]68        slugs: list[str] = []69        for sitemap_url in maps:70            xml = self.get(sitemap_url.strip()).text71            for loc in re.findall(r"<loc>([^<]+)</loc>", xml):72                m = _VEH_URL_RE.match(loc.strip())73                if m:74                    slugs.append(m.group(1))75        return sorted(set(slugs))7677    # -- page-data Gatsby -> payload usedCar ----------------------------------78    def _fetch_detail(self, slug: str) -> dict:79        url = f"{self.base_url}/page-data/auto-usage/{slug}/page-data.json"80        try:81            data = self.get(url).json()82        except ValueError:83            return {}84        result = data.get("result") or {}85        used_car = (result.get("data") or {}).get("usedCar")86        if not isinstance(used_car, dict):87            return {}88        used_car.pop("coordinates", None)          # gros bloc inutile89        used_car.pop("loanPaiement", None)90        return {"usedCar": used_car,91                "latestModification":92                    (result.get("pageContext") or {}).get("latestModification")}9394    # -- contrat ---------------------------------------------------------------95    def fetch(self) -> list[Vehicle]:96        slugs = self._vehicle_slugs()97        week = datetime.date.today().isocalendar()98        cache_key = f"v1:{week.year}w{week.week}"    # revalidation hebdomadaire99        cap = int(os.environ.get("AUTOKA_MAX_DETAILS", self.max_details))100101        vehicles: list[Vehicle] = []102        real_fetches = 0103        for slug in slugs:104            def _fetch(s=slug):105                return self._fetch_detail(s)106107            # plafond atteint : ne servir que le cache (fetch réel interdit)108            if real_fetches >= cap:109                from .. import db110                if self._detail_con is None:111                    self._detail_con = db.connect()112                data = db.get_cached_detail(self._detail_con, self.source_id,113                                            slug, cache_key)114                if data is None:115                    continue116            else:117                before = self._last_request118                try:119                    data = self.detail(slug, cache_key, _fetch)120                except Exception:121                    # page disparue (vendu) ou erreur ponctuelle : ne jamais122                    # bloquer la source entière123                    if self._last_request != before:124                        real_fetches += 1125                    continue126                if self._last_request != before:   # une vraie requête a eu lieu127                    real_fetches += 1128            if not data or not data.get("usedCar"):129                continue130            veh = self._to_vehicle(slug, data["usedCar"])131            if veh is not None:132                vehicles.append(veh)133        return vehicles134135    # -- usedCar -> Vehicle ------------------------------------------------------136    def _to_vehicle(self, slug: str, d: dict) -> Vehicle | None:137        make = str(d.get("brand") or "").strip()138        model = str(d.get("labelModel_fr") or d.get("model") or "").strip()139        body = str(d.get("labelBodytype_fr") or d.get("bodytype") or "").strip()140        if _EXCLUDE_RE.search(f"{make} {model} {body}"):    # non-automobile141            return None142143        year = d.get("year")144        trim = str(d.get("trim") or "").strip()145        # certains labels de modèle incluent déjà la version (X1 XDRIVE28I)146        title_trim = "" if trim.lower() in model.lower() else trim147        title = " ".join(str(x)148                         for x in (make, model, title_trim, year or "") if x)149        if not title:150            return None151152        price = d.get("price")153        try:154            price = float(price) if price else None155        except (TypeError, ValueError):156            price = None157158        km = d.get("odometer")159        try:160            km = float(km) if km not in (None, "") else None161        except (TypeError, ValueError):162            km = None163        unit = str(d.get("odometer_Type") or "km").lower()164        if km is not None and unit.startswith("mi"):165            km = round(km * 1.609344)166167        images = d.get("images") or []168        if isinstance(images, str):169            images = [images]170        images = [u for u in images if isinstance(u, str)]171        if not images and d.get("image"):172            images = [str(d["image"])]173174        features = [f.strip() for f in str(d.get("options") or "").split("|")175                    if f.strip()]176177        doors = d.get("doors")178        seats = d.get("passengers")179180        details: dict = {"branch": str(d.get("nameDealer") or "")}181        for src, dst in (("accident", "accidented"),182                         ("certified", "certified"),183                         ("uniqueOwner", "unique_owner")):184            if d.get(src) is not None:185                details[dst] = bool(d[src])186        if d.get("cityConsumption"):187            details["consumption_city_l_100km"] = d["cityConsumption"]188        if d.get("hightwayConsumption"):189            details["consumption_hwy_l_100km"] = d["hightwayConsumption"]190        if d.get("inventoryID"):191            details["inventory_id"] = str(d["inventoryID"])192193        return Vehicle(194            source=self.source_id,195            external_id=slug,196            url=f"{self.base_url}/auto-usage/{slug}/",197            title=title,198            make=make,199            model=model,200            trim=trim,201            year=int(year) if year else None,202            price=price,203            price_label=f"{price:,.0f} $".replace(",", " ") if price else "",204            mileage_km=km,205            mileage_label=f"{km:,.0f} km".replace(",", " ") if km else "",206            transmission=str(d.get("transmission") or ""),207            fuel=str(d.get("labelEngine_fr") or d.get("engine") or ""),208            drivetrain=str(d.get("drivetrain") or ""),209            body_type=body,210            exterior_color=str(d.get("labelColorExt_fr") or ""),211            interior_color=str(d.get("labelColorInt_fr") or ""),212            engine=str(d.get("motor") or ""),213            doors=int(doors) if doors else None,214            seats=int(seats) if seats else None,215            vin=str(d.get("niv") or ""),216            stock_number=str(d.get("stockNO") or ""),217            dealer_name=str(d.get("nameDealer") or "").strip()218                        or self.dealer_name,219            city=str(d.get("cityDealer") or "").strip(),220            description=_clean_html(d.get("description"))[:4000],221            features=features[:60],222            details=details,223            images=images[:20],224            carfax_url=str(d.get("carproofURL") or ""),225        )226