spb/auto-ka Public
Python 81.9%
TypeScript 12.4%
CSS 5.5%
1# -----------------------------------------------------------------------------2# Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/occasionbeaucage.py : connecteur Occasion Beaucage5# (occasionbeaucage.com), groupe de mégacentres d'occasion de l'Estrie /6# Centre-du-Québec / Montérégie (Sherbrooke, Magog, Granby, Drummondville,7# Victoriaville, Saint-Hyacinthe, Cowansville) — ~1 200 véhicules.8#9# Plateforme : Next.js App Router (autoroot.io) sur Vercel, inventaire10# EvalAuto. Chaque page liste /auto-usage/<n>/ (24 véhicules/page) embarque11# dans son flux RSC (self.__next_f.push) un objet12# {"vehicles":[...],"count":N,"page":n} où chaque véhicule est COMPLET :13# prix, km, VIN, stock, transmission, carburant, carrosserie, couleurs,14# moteur, portes, places, carfax, photos ET la concession du véhicule15# (display_dealer_name / display_city). AUCUNE page détail nécessaire :16# ~52 requêtes pour tout l'inventaire, ville exacte par véhicule.17#18# Les refs de flux («$313») pointant vers d'autres lignes RSC (les listes19# de photos) sont résolues localement. AUTOKA_MAX_PAGES (env) limite la20# pagination pour les tests.21#22# Enrichissement détail : la description complète et la liste d'équipements23# ne vivent PAS dans le flux liste — seulement dans le flux RSC de la page24# détail /auto-usage/<slug>/, sous forme de refs ($21…) vers des lignes25# longues «T<hex>,» (longueur en OCTETS UTF-8) : description = HTML brut,26# options = JSON {"fr": "équip1|équip2|…"}. Ces payloads sont mis en cache27# BD à clé stable (les équipements ne changent pas ; le prix frais vient de28# la liste) et plafonnés par AUTOKA_MAX_DETAILS / max_details par sync —29# l'inventaire complet se couvre en quelques synchronisations.30# -----------------------------------------------------------------------------31from __future__ import annotations3233import html as htmllib34import json35import os36import re3738from ..schema import Vehicle39from .base import BaseConnector4041# chunks du flux RSC : self.__next_f.push([1,"<chaîne JS>"])42_FLIGHT_RE = re.compile(r'self\.__next_f\.push\(\[1,("(?:\\.|[^"\\])*")\]\)')4344# non-automobiles éventuels (garde-fou ; le groupe ne vend que des autos)45_EXCLUDE_RE = re.compile(46 r"can-?am|harley|ski-?doo|sea-?doo|spyder|ryker|motoneige|vtt|"47 r"motomarine|roulotte|remorque", re.I)484950def _js_str(literal: str) -> str:51 """Chaîne littérale JS (avec guillemets) -> texte. Les chunks RSC sont52 encodés comme du JSON ; repli tolérant si un escape non-JSON traîne."""53 try:54 return json.loads(literal)55 except ValueError:56 out = literal[1:-1].encode("utf-8").decode("unicode_escape")57 try: # unicode_escape mange l'UTF-8 -> réparer58 return out.encode("latin-1").decode("utf-8")59 except (UnicodeEncodeError, UnicodeDecodeError):60 return out616263def _s(value) -> str:64 """Scalaire du flux RSC -> str propre ('$undefined', '$D...' -> '')."""65 if value is None:66 return ""67 text = str(value).strip()68 return "" if text.startswith("$") else text697071def _strip_html(text: str | None) -> str:72 """HTML (avec entités) -> texte brut compact."""73 if not text:74 return ""75 out = htmllib.unescape(str(text))76 out = re.sub(r"<[^>]+>", " ", out)77 return re.sub(r"\s{2,}", " ", out).strip()787980# clé de cache détail stable : description/équipements ne changent pas pour81# une annonce donnée (bump en cas de changement de parsing)82_DETAIL_KEY = "v1"838485class OccasionBeaucage(BaseConnector):86 """Occasion Beaucage — inventaire complet via les pages liste (RSC)."""8788 source_id = "occasionbeaucage"89 base_url = "https://www.occasionbeaucage.com"90 dealer_name = "Occasion Beaucage"91 request_delay = 1.092 max_pages = 120 # garde-fou dur (~52 pages réelles)93 max_details = 200 # plafond de vraies requêtes détail/sync9495 # -- flux RSC ----------------------------------------------------------96 def _flight_blob(self, html: str) -> str:97 return "".join(_js_str(m) for m in _FLIGHT_RE.findall(html))9899 def _page_payload(self, page: int) -> dict:100 """{"vehicles": [...], "count": N, "page": n} de la page liste."""101 url = f"{self.base_url}/auto-usage/"102 if page > 1:103 url += f"{page}/"104 blob = self._flight_blob(self.get(url).text)105 decoder = json.JSONDecoder()106 for m in re.finditer(r'"vehicles":\[', blob):107 start = blob.rfind("{", 0, m.start())108 if start < 0:109 continue110 try:111 obj, _ = decoder.raw_decode(blob, start)112 except ValueError:113 continue114 if isinstance(obj, dict) and "count" in obj \115 and isinstance(obj.get("vehicles"), list):116 obj["_blob"] = blob # pour résoudre les refs $117 return obj118 return {}119120 @staticmethod121 def _resolve_ref(blob: str, ref: str):122 """«$313» -> payload JSON de la ligne RSC 313 (ex. liste de photos).123124 Les lignes à préfixe de longueur (T<hex>,) ne sont pas suivies d'un125 saut de ligne — on repère donc «<id>:» n'importe où et on valide en126 décodant le JSON qui suit (raw_decode s'arrête à la fin du tableau).127 """128 decoder = json.JSONDecoder()129 pattern = re.compile(re.escape(ref[1:]) + r':(?:T[0-9a-f]+,)?(?=[\["])')130 for row in pattern.finditer(blob):131 try:132 value, _ = decoder.raw_decode(blob, row.end())133 except ValueError:134 continue135 if isinstance(value, (list, str)):136 return value137 return None138139 @staticmethod140 def _flight_row(blob: str, ref: str):141 """«$21» -> contenu de la ligne RSC 21, y compris les lignes longues142 «21:T6bc,<texte>» dont la longueur T<hex> est en OCTETS UTF-8 et dont143 le contenu est du texte brut (HTML ou JSON sérialisé), sans saut de144 ligne final. Les lignes JSON classiques sont décodées directement."""145 rid = ref[1:]146 if not rid or not re.fullmatch(r"[0-9a-f]+", rid):147 return None # $undefined, $D…, etc.148 decoder = json.JSONDecoder()149 pattern = re.compile(150 r"(?<![0-9a-zA-Z_$])" + re.escape(rid) + r":(?:T([0-9a-f]+),)?")151 for m in pattern.finditer(blob):152 if m.group(1): # ligne longue à préfixe T<hex>153 length = int(m.group(1), 16)154 raw = blob[m.end():].encode("utf-8")[:length]155 return raw.decode("utf-8", errors="ignore")156 if blob[m.end():m.end() + 1] not in '["{':157 continue158 try:159 value, _ = decoder.raw_decode(blob, m.end())160 except ValueError:161 continue162 return value163 return None164165 # -- page détail -> payload (description complète + équipements) --------166 def _fetch_detail(self, slug: str) -> dict:167 blob = self._flight_blob(168 self.get(f"{self.base_url}/auto-usage/{slug}/").text)169 decoder = json.JSONDecoder()170 veh: dict | None = None171 for m in re.finditer(r'"vehicle":\{', blob):172 try:173 obj, _ = decoder.raw_decode(blob, m.start() + len('"vehicle":'))174 except ValueError:175 continue176 if isinstance(obj, dict) and obj.get("slug"):177 veh = obj178 break179 if not veh:180 return {}181 payload: dict = {}182183 desc = veh.get("description")184 if isinstance(desc, str) and desc.startswith("$"):185 desc = self._flight_row(blob, desc)186 desc = _strip_html(desc if isinstance(desc, str) else "")187 if desc:188 payload["description"] = desc[:4000]189190 opts = veh.get("options")191 if isinstance(opts, str) and opts.startswith("$"):192 opts = self._flight_row(blob, opts)193 if isinstance(opts, str) and opts.lstrip().startswith("{"):194 try: # ligne longue = JSON sérialisé195 opts = json.loads(opts)196 except ValueError:197 opts = None198 if isinstance(opts, dict):199 opts = opts.get("fr") or opts.get("en") or ""200 if isinstance(opts, str) and opts:201 features = [f.strip() for f in opts.split("|") if f.strip()]202 if features:203 payload["features"] = features[:60]204 return payload205206 # -- contrat -----------------------------------------------------------207 def fetch(self) -> list[Vehicle]:208 cap = int(os.environ.get("AUTOKA_MAX_PAGES", self.max_pages))209 vehicles: list[Vehicle] = []210 seen: set[str] = set()211212 page, last_page = 1, 1213 while page <= min(last_page, cap, self.max_pages):214 payload = self._page_payload(page)215 batch = payload.get("vehicles") or []216 if not batch:217 break218 per_page = max(len(batch), 1)219 count = int(payload.get("count") or 0)220 last_page = max(1, -(-count // per_page)) # ceil221 new = 0222 for d in batch:223 veh = self._to_vehicle(d, payload.get("_blob", ""))224 if veh is not None and veh.external_id not in seen:225 seen.add(veh.external_id)226 vehicles.append(veh)227 new += 1228 if new == 0: # page recyclée229 break230 page += 1231232 self._enrich_details(vehicles)233 return vehicles234235 # -- enrichissement détail plafonné (cache stable) -----------------------236 def _enrich_details(self, vehicles: list[Vehicle]) -> None:237 cap = int(os.environ.get("AUTOKA_MAX_DETAILS", self.max_details))238 real_fetches = 0239 for veh in vehicles:240 slug = veh.url.rstrip("/").rsplit("/", 1)[-1]241 detail: dict = {}242 if real_fetches >= cap: # plafond : cache seulement243 from .. import db244 if self._detail_con is None:245 self._detail_con = db.connect()246 detail = db.get_cached_detail(247 self._detail_con, self.source_id,248 str(veh.external_id), _DETAIL_KEY) or {}249 else:250 def _fetch(s=slug):251 return self._fetch_detail(s)252 before = self._last_request253 try:254 detail = self.detail(veh.external_id, _DETAIL_KEY, _fetch)255 except Exception:256 detail = {} # page disparue : données liste257 if self._last_request != before:258 real_fetches += 1259 if not detail:260 continue261 desc = detail.get("description") or ""262 if len(desc) > len(veh.description or ""):263 veh.description = desc264 if detail.get("features") and not veh.features:265 veh.features = list(detail["features"])266267 # -- enregistrement liste -> Vehicle -------------------------------------268 def _to_vehicle(self, d: dict, blob: str) -> Vehicle | None:269 if d.get("is_sold") or d.get("is_active") is False:270 return None271272 slug = _s(d.get("slug"))273 ext_id = _s(d.get("inventory_provider_vehicle_id")) or slug274 if not ext_id or not slug:275 return None276277 make = _s(d.get("make"))278 model = _s(d.get("model"))279 body = _s(d.get("body_type"))280 if _EXCLUDE_RE.search(f"{make} {model} {body}"): # non-automobile281 return None282283 year = d.get("year")284 trim = _s(d.get("trim"))285 title = " ".join(x for x in (make, model, trim, _s(year)) if x)286287 price = d.get("price")288 try:289 price = float(price) if price else None290 except (TypeError, ValueError):291 price = None292293 km = d.get("mileage_from_odometer")294 try:295 km = float(km) if km not in (None, "") else None296 except (TypeError, ValueError):297 km = None298299 images = d.get("images")300 if isinstance(images, str) and images.startswith("$"):301 images = self._resolve_ref(blob, images)302 if isinstance(images, str) and images.lstrip().startswith("["):303 try: # parfois une liste JSON sérialisée304 images = json.loads(images)305 except ValueError:306 images = []307 if isinstance(images, str):308 images = [images]309 images = [u for u in (images or []) if isinstance(u, str)][:20]310311 doors = d.get("doors")312 seats = d.get("seating_capacity")313314 details: dict = {}315 for src, dst in (("is_certified", "certified"),316 ("is_accidented", "accidented"),317 ("is_unique_owner", "unique_owner")):318 if d.get(src) is not None:319 details[dst] = bool(d[src])320 if _s(d.get("fuel_consumption_city")):321 details["consumption_city_l_100km"] = _s(d["fuel_consumption_city"])322 if _s(d.get("fuel_consumption_highway")):323 details["consumption_hwy_l_100km"] = _s(d["fuel_consumption_highway"])324 promo = _s(d.get("promotional_price"))325 if promo and promo not in ("0", "0.0"):326 details["promotional_price"] = promo327 if _s(d.get("dealer_id")):328 details["branch_slug"] = _s(d["dealer_id"])329330 return Vehicle(331 source=self.source_id,332 external_id=ext_id,333 url=f"{self.base_url}/auto-usage/{slug}/",334 title=title,335 make=make,336 model=model,337 trim=trim,338 year=int(year) if year else None,339 price=price,340 price_label=f"{price:,.0f} $".replace(",", " ") if price else "",341 mileage_km=km,342 mileage_label=f"{km:,.0f} km".replace(",", " ") if km else "",343 transmission=_s(d.get("transmission")),344 fuel=_s(d.get("fuel_type")),345 drivetrain=_s(d.get("drive_wheel_configuration")),346 body_type=body,347 exterior_color=_s(d.get("exterior_color")),348 interior_color=_s(d.get("interior_color")),349 engine=_s(d.get("engine_type")),350 doors=int(doors) if doors else None,351 seats=int(seats) if seats else None,352 vin=_s(d.get("vin")),353 stock_number=_s(d.get("stock_number")),354 dealer_name=_s(d.get("display_dealer_name")) or355 _s(d.get("dealer_name")) or self.dealer_name,356 city=_s(d.get("display_city")) or _s(d.get("city")),357 description=_s(d.get("short_description"))[:4000],358 details=details,359 images=images,360 carfax_url=_s(d.get("carfax_url")),361 )362