SPB Git

spb/auto-ka Public

Python 81.8% TypeScript 12.4% CSS 5.5%
6.0 KB · 161 lines python
Raw Blame History
1# -----------------------------------------------------------------------------2# Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/yannicklaberge.py : Automobiles Yannick Laberge (Beauport) —5#   WordPress sur la plateforme AMVOQ (thème « theme_central », inventaire6#   autousagee.ca, images image.autousagee.ca). API REST verrouillée (401).7#8#   Stratégie : le sitemap Yoast `/vehicules-sitemap.xml` recense les annonces9#   (/vehicule-usage/<slug>/<id>/). Chaque page détail expose un JSON-LD10#   schema.org @type Vehicle : prix, VIN, km, transmission, carburant,11#   couleur intérieure, description. Galerie complétée par les URLs12#   image.autousagee.ca de la page. Pages détail en cache BD (clé hebdo).13# -----------------------------------------------------------------------------14from __future__ import annotations1516import datetime17import json18import os19import re2021from ..schema import Vehicle22from .base import BaseConnector2324_VEH_URL_RE = re.compile(r"/vehicule-usage/[^/]+/(\d+)/?$")2526_LD_RE = re.compile(27    r'<script type=["\']application/ld\+json["\'][^>]*>(.*?)</script>', re.S)2829_IMG_RE = re.compile(r'https://image\.autousagee\.ca/[^"\'\\\s)]+\.jpe?g')303132class YannickLaberge(BaseConnector):33    source_id = "automobilesyannicklaberge"34    base_url = "https://www.automobilesyannicklaberge.com"35    dealer_name = "Automobiles Yannick Laberge"36    city = "Beauport"37    request_delay: float = 1.038    max_details = 2003940    # -- sitemap ----------------------------------------------------------------41    def _vehicle_urls(self) -> list[str]:42        xml = self.get(f"{self.base_url}/vehicules-sitemap.xml").text43        return [loc.strip() for loc in re.findall(r"<loc>([^<]+)</loc>", xml)44                if _VEH_URL_RE.search(loc.strip())]4546    # -- page détail -> JSON-LD Vehicle + galerie ---------------------------------47    def _fetch_detail(self, url: str) -> dict:48        html = self.get(url).text49        for block in _LD_RE.findall(html):50            try:51                data = json.loads(block.strip())52            except ValueError:53                continue54            nodes = data if isinstance(data, list) else [data]55            for node in nodes:56                types = node.get("@type")57                types = types if isinstance(types, list) else [types]58                if "Vehicle" in types or "Car" in types:59                    seen: set[str] = set()60                    node["_gallery"] = [61                        u for u in _IMG_RE.findall(html)62                        if not (u in seen or seen.add(u))][:20]63                    return node64        return {}6566    # -- contrat -----------------------------------------------------------------67    def fetch(self) -> list[Vehicle]:68        urls = self._vehicle_urls()69        week = datetime.date.today().isocalendar()70        cache_key = f"v1:{week.year}w{week.week}"71        cap = int(os.environ.get("AUTOKA_MAX_DETAILS", self.max_details))7273        vehicles: list[Vehicle] = []74        real_fetches = 075        for url in urls:76            ext_id = _VEH_URL_RE.search(url).group(1)7778            def _fetch(u=url):79                return self._fetch_detail(u)8081            if real_fetches >= cap:            # plafond : cache seulement82                from .. import db83                if self._detail_con is None:84                    self._detail_con = db.connect()85                data = db.get_cached_detail(self._detail_con, self.source_id,86                                            ext_id, cache_key)87                if data is None:88                    continue89            else:90                before = self._last_request91                data = self.detail(ext_id, cache_key, _fetch)92                if self._last_request != before:93                    real_fetches += 194            if not data:95                continue96            veh = self._to_vehicle(ext_id, url, data)97            if veh is not None:98                vehicles.append(veh)99        return vehicles100101    def _to_vehicle(self, ext_id: str, url: str, d: dict) -> Vehicle | None:102        title = " ".join(str(d.get("name") or "").split())103        if not title:104            return None105106        offers = d.get("offers") or {}107        price = None108        try:109            p = float(offers.get("price"))110            price = p if p > 0 else None111        except (TypeError, ValueError):112            pass113114        odo = d.get("mileageFromOdometer") or {}115        km = odo.get("value") if isinstance(odo, dict) else odo116        try:117            km = float(km)118            km = km if km > 0 else None119        except (TypeError, ValueError):120            km = None121122        brand = d.get("brand") or {}123        make = brand.get("name") if isinstance(brand, dict) else str(brand)124125        year = None126        try:127            year = int(str(d.get("productionDate"))[:4])128        except (TypeError, ValueError):129            pass130131        try:132            doors = int(str(d.get("numberOfDoors")))133        except (TypeError, ValueError):134            doors = None135136        images = d.get("_gallery") or []137        if not images and isinstance(d.get("image"), str):138            images = [d["image"]]139140        return Vehicle(141            source=self.source_id,142            external_id=ext_id,143            url=url,144            title=title,145            make=str(make or ""),146            year=year,147            price=price,148            price_label=(f"{price:,.0f} $".replace(",", " ") if price else ""),149            mileage_km=km,150            mileage_label=(f"{km:,.0f} km".replace(",", " ") if km else ""),151            transmission=str(d.get("vehicleTransmission") or ""),152            fuel=str(d.get("fuelType") or ""),153            interior_color=str(d.get("vehicleInteriorColor") or ""),154            doors=doors,155            vin=str(d.get("vehicleIdentificationNumber") or ""),156            dealer_name=self.dealer_name,157            city=self.city,158            description=" ".join(str(d.get("description") or "").split())[:4000],159            images=images,160        )161