# ----------------------------------------------------------------------------- # Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/otogo.py : PORTAIL — Otogo.ca (LC Media, 100 % Québec) # # Otogo expose une API JSON publique, sans clé ni rendu JS : # GET https://www.otogo.ca/api-search/?latlng=0,0&locale=fr&no_facets= # &rpp=100&page=N&status=Usagé,Certifié,Démo[&price=lo-hi] # Chaque annonce de la liste est déjà riche : titre/version FR, année, # prix, kilométrage, VIN (≈100 %), nom du concessionnaire et surtout le # GPS EXACT du vendeur (latitude/longitude) — parfait pour la recherche # par rayon et la dédup VIN avec les concessionnaires connectés. # (Paramètres découverts dans les bundles Nuxt : constantes c.b du module # 5731047.js — latlng/locale/rpp/page/price/status/no_facets/no_results.) # # Profondeur : l'API plafonne à ~1 000 résultats par requête (rpp=100 → # meta.pages ≤ 10) pour ~24 000 annonces d'occasion. On PARTITIONNE donc # par tranches de prix (bissection récursive + tranche ouverte « lo- ») : # chaque annonce n'est téléchargée qu'une fois, ~250 requêtes directes # par sync complet, zéro crédit d'API de scraping externe. # # Ville : absente de la liste — déduite du GPS exact via nearest_city() # (data/villes_gps.json) pour l'affichage et la région administrative. # Carburant/boîte : absents de la liste ; repli commun finalize() (jetons # du titre/de la version). Page détail : /auto/{slug_fr} (non visitée — # économie ; le VIN et le GPS sont déjà dans la liste). # # Plafond : AUTOKA_OTOGO_MAX (défaut 0 = inventaire complet). # ----------------------------------------------------------------------------- from __future__ import annotations import os import re from ..normalize import nearest_city from ..schema import Vehicle from .base import BaseConnector BASE = "https://www.otogo.ca" API = BASE + "/api-search/" RPP = 100 # résultats par page (max utile observé) DEPTH = 1000 # profondeur max de pagination par requête STATUS = "Usagé,Certifié,Démo" # occasion seulement (exclut Neuf) PRICE_SPLIT_FLOOR = 250 # $ : largeur minimale d'une tranche de prix _VIN_RE = re.compile(r"^[A-HJ-NPR-Z0-9]{17}$") def _max_listings() -> int: try: return max(0, int(os.environ.get("AUTOKA_OTOGO_MAX", "0"))) except ValueError: return 0 class OtogoQuebec(BaseConnector): """Portail Otogo.ca — inventaire d'occasion des concessionnaires du Québec.""" source_id = "otogo" request_delay = 0.8 def __init__(self) -> None: super().__init__() self.session.headers.update({ "Accept": "application/json", "Accept-Language": "fr-CA,fr;q=0.9,en;q=0.5", }) # -- API -------------------------------------------------------------------- def _query(self, page: int, lo: int, hi: int | None) -> tuple[dict, list[dict]]: """Une page de résultats pour la tranche de prix [lo, hi] (hi=None = ouvert).""" params = { "latlng": "0,0", "locale": "fr", "no_facets": "", "rpp": str(RPP), "status": STATUS, "price": f"{lo}-{'' if hi is None else hi}", } if page > 1: params["page"] = str(page) data = self.get(API, params=params).json() meta = data.get("meta") or {} results = ((data.get("data") or {}).get("results")) or [] return meta, results def _to_vehicle(self, r: dict) -> Vehicle | None: ext_id = str(r.get("id") or "") slug = r.get("slug_fr") or r.get("slug_en") or "" if not ext_id or not slug: return None price = r.get("price") price = float(price) if price and price >= 500 else None mileage = r.get("mileage") mileage = float(mileage) if mileage is not None else None try: lat = float(r.get("latitude")) lng = float(r.get("longitude")) except (TypeError, ValueError): lat = lng = None vin = (r.get("vin") or "").strip().upper() if not _VIN_RE.fullmatch(vin): vin = "" details: dict = {} if r.get("status_fr"): details["otogo_status"] = r["status_fr"] # Usagé | Certifié | Démo return Vehicle( source=self.source_id, external_id=ext_id, url=f"{BASE}/auto/{slug}", kind="auto", title=r.get("title_fr") or r.get("title_en") or "", trim=(r.get("trim_fr") or r.get("trim_en") or "").strip(" |"), year=r.get("year"), price=price, price_label=(f"{price:,.0f} $".replace(",", " ") if price else ""), mileage_km=mileage, vin=vin, dealer_name=(r.get("dealer_name") or "").strip(), city=nearest_city(lat, lng), lat=lat, lng=lng, details=details, images=[u for u in [r.get("thumbnail")] if u], ) # -- contrat ----------------------------------------------------------------- def fetch(self) -> list[Vehicle]: cap = _max_listings() vehicles: dict[str, Vehicle] = {} # partition par prix : bissection récursive tant qu'une tranche # dépasse la profondeur maximale de l'API (~1 000 annonces) stack: list[tuple[int, int | None]] = [(0, 100_000), (100_000, None)] while stack: if cap and len(vehicles) >= cap: break lo, hi = stack.pop() meta, results = self._query(1, lo, hi) total = int(meta.get("total") or 0) if total == 0: continue if total > DEPTH: if hi is None: # tranche ouverte : décaler stack.append((lo + 100_000, None)) stack.append((lo, lo + 100_000)) continue if hi - lo > PRICE_SPLIT_FLOOR: # bissection mid = (lo + hi) // 2 stack.append((mid, hi)) stack.append((lo, mid)) continue # tranche déjà minimale : on prend ce que la profondeur permet for r in results: v = self._to_vehicle(r) if v: vehicles[v.external_id] = v pages = int(meta.get("pages") or 1) for page in range(2, pages + 1): if cap and len(vehicles) >= cap: break try: _, results = self._query(page, lo, hi) except Exception: break # fin de pagination if not results: break for r in results: v = self._to_vehicle(r) if v: vehicles[v.external_id] = v return list(vehicles.values())