spb/auto-ka

1# -----------------------------------------------------------------------------2# Auto-Ka — Agrégateur de voitures usagées à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# dealers.py : coordonnées des concessionnaires (adresse, téléphone, GPS)5#6#   1 source = 1 site de concessionnaire (sauf portails/marketplaces) : on7#   visite UNE page par site (l'accueil suffit — l'adresse et le téléphone8#   sont dans l'en-tête/le pied de page, et la plupart des plateformes (D2C,9#   SM360, Convertus…) émettent un JSON-LD AutoDealer/LocalBusiness complet).10#   Extraction : JSON-LD d'abord (address/telephone/geo), replis regex11#   (lien tel:, adresse avec code postal canadien). GPS par défaut : centre12#   de la ville de la source (data/villes_gps.json).13#14#   Budget : ~138 requêtes UNE fois (cache en base — on ne re-visite que les15#   sources absentes de la table), throttle 1 s.16#17#   One-shot :  python3 -m autoka.dealers [--refresh]18# -----------------------------------------------------------------------------19from __future__ import annotations2021import json22import re23import sqlite324import time25from pathlib import Path2627import requests2829from .normalize import city_gps3031SOURCES_PATH = Path(__file__).resolve().parent.parent / "data" / "sources.json"3233USER_AGENT = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "34              "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36 "35              "AutoKaBot/1.0 (+contact@spboucher.ai)")3637_SCHEMA = """38CREATE TABLE IF NOT EXISTS dealers (39    source     TEXT PRIMARY KEY,   -- id de la source (data/sources.json)40    name       TEXT,41    address    TEXT,42    phone      TEXT,43    website    TEXT,44    city       TEXT,45    lat        REAL,46    lng        REAL,47    fetched_at REAL48);49"""5051_LD_RE = re.compile(52    r'<script[^>]*type=["\']application/ld\+json["\'][^>]*>(.*?)</script>', re.S)53_TEL_RE = re.compile(r'href=["\']tel:([+\d][\d\s().\-]{6,18}\d)["\']', re.I)54# « 123 rue Untel, Ville (QC) J0X 1A0 » — ancré sur le code postal canadien55# (MAJUSCULES obligatoires et ; : # { } exclus : évite les couleurs hex et56# variables CSS inline des thèmes D2C qui imitent le motif postal)57_ADDR_RE = re.compile(58    r"(\d{1,5}[^<>{}|;:#\n]{5,90}?\b[A-Z]\d[A-Z]\s?\d[A-Z]\d\b)")59_STYLE_RE = re.compile(r"<(style|script)[^>]*>.*?</\1>", re.S | re.I)60_PHONE_TXT_RE = re.compile(r"(?<!\d)(\(?[2-9]\d{2}\)?[\s.\-]\d{3}[\s.\-]\d{4})(?!\d)")6162_DEALER_LD_TYPES = {"AutoDealer", "AutomotiveBusiness", "LocalBusiness",63                    "MotorcycleDealer", "Organization", "AutoRepair", "Store"}646566def ensure_schema(con: sqlite3.Connection) -> None:67    con.executescript(_SCHEMA)68    con.commit()697071def _iter_ld(html: str):72    for blob in _LD_RE.findall(html):73        try:74            data = json.loads(blob.strip())75        except ValueError:76            continue77        stack = data if isinstance(data, list) else [data]78        for item in stack:79            if isinstance(item, dict):80                yield from _flatten_ld(item)818283def _flatten_ld(item: dict):84    yield item85    for v in item.get("@graph", []) if isinstance(item.get("@graph"), list) else []:86        if isinstance(v, dict):87            yield v888990def _fmt_address(addr) -> str:91    if isinstance(addr, str):92        return " ".join(addr.split())93    if isinstance(addr, dict):94        parts = [addr.get("streetAddress"), addr.get("addressLocality"),95                 addr.get("addressRegion"), addr.get("postalCode")]96        return ", ".join(str(p).strip() for p in parts if p)97    return ""9899100def _clean_phone(raw: str) -> str:101    digits = re.sub(r"\D", "", raw or "")102    if digits.startswith("1") and len(digits) == 11:103        digits = digits[1:]104    if len(digits) != 10:105        return (raw or "").strip()106    return f"({digits[0:3]}) {digits[3:6]}-{digits[6:]}"107108109def _extract(html: str) -> dict:110    """address / phone / lat / lng depuis une page HTML de concessionnaire."""111    out: dict = {}112    for item in _iter_ld(html):113        types = item.get("@type") or ""114        types = {types} if isinstance(types, str) else set(types)115        if not (types & _DEALER_LD_TYPES):116            continue117        if not out.get("address") and item.get("address"):118            out["address"] = _fmt_address(item["address"])119        if not out.get("phone") and item.get("telephone"):120            out["phone"] = _clean_phone(str(item["telephone"]))121        geo = item.get("geo") or {}122        if isinstance(geo, dict) and not out.get("lat"):123            try:124                lat, lng = float(geo.get("latitude")), float(geo.get("longitude"))125                if 44.0 <= lat <= 63.0 and -80.0 <= lng <= -57.0:   # Québec126                    out["lat"], out["lng"] = lat, lng127            except (TypeError, ValueError):128                pass129        if out.get("address") and out.get("phone"):130            break131    if not out.get("phone"):132        m = _TEL_RE.search(html) or _PHONE_TXT_RE.search(html)133        if m:134            out["phone"] = _clean_phone(m.group(1))135    if not out.get("address"):136        text = re.sub(r"<[^>]+>", " ", _STYLE_RE.sub(" ", html))137        m = _ADDR_RE.search(text)138        # garde-fou : une vraie adresse contient un mot (rue, boul., ville…)139        if m and re.search(r"[A-Za-zÀ-ÿ]{3,}", m.group(1)):140            out["address"] = " ".join(m.group(1).split())141    return out142143144def populate(con: sqlite3.Connection, refresh: bool = False,145             delay: float = 1.0) -> dict:146    """Peuple la table dealers — 1 requête par source manquante, throttle 1 s."""147    ensure_schema(con)148    registry = json.loads(SOURCES_PATH.read_text(encoding="utf-8"))["sources"]149    have = {r["source"] for r in con.execute("SELECT source FROM dealers")}150    session = requests.Session()151    session.headers["User-Agent"] = USER_AGENT152    fetched = filled_addr = filled_phone = errors = 0153    for s in registry:154        sid = s["id"]155        if s.get("type") == "marketplace":     # Kijiji : pas un concessionnaire156            continue157        if sid in have and not refresh:158            continue159        info: dict = {}160        try:161            resp = session.get(s["url"], timeout=25)162            resp.raise_for_status()163            info = _extract(resp.text)164            fetched += 1165        except Exception as exc:166            errors += 1167            print(f"[auto-ka] dealers : {sid} — {exc}")168        gps = (info.get("lat"), info.get("lng"))169        if gps[0] is None or gps[1] is None:170            hit = city_gps(s.get("city"))171            gps = hit if hit else (None, None)172        con.execute(173            """INSERT INTO dealers (source, name, address, phone, website,174               city, lat, lng, fetched_at) VALUES (?,?,?,?,?,?,?,?,?)175               ON CONFLICT(source) DO UPDATE SET name=excluded.name,176               address=excluded.address, phone=excluded.phone,177               website=excluded.website, city=excluded.city,178               lat=excluded.lat, lng=excluded.lng,179               fetched_at=excluded.fetched_at""",180            (sid, s.get("name", sid), info.get("address", ""),181             info.get("phone", ""), s["url"], s.get("city", ""),182             gps[0], gps[1], time.time()))183        con.commit()184        if info.get("address"):185            filled_addr += 1186        if info.get("phone"):187            filled_phone += 1188        time.sleep(delay)189    total = con.execute("SELECT COUNT(*) c FROM dealers").fetchone()["c"]190    out = {"fetched": fetched, "errors": errors, "with_address": filled_addr,191           "with_phone": filled_phone, "dealers_total": total}192    print(f"[auto-ka] dealers : {out}")193    return out194195196def list_dealers(con: sqlite3.Connection) -> list[dict]:197    ensure_schema(con)198    counts = {r["source"]: r["n"] for r in con.execute(199        "SELECT source, COUNT(*) n FROM vehicles WHERE active=1 GROUP BY source")}200    rows = [dict(r) for r in con.execute(201        "SELECT * FROM dealers ORDER BY name COLLATE NOCASE")]202    for r in rows:203        r["active_listings"] = counts.get(r["source"], 0)204    return rows205206207if __name__ == "__main__":208    import sys209    from . import db210    con = db.connect()211    populate(con, refresh="--refresh" in sys.argv)212    con.close()213