spb/immo-ka Public

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/_detailutil.py : utilitaires partagés d'enrichissement « page détail »5#   Mutualise ce que chaque connecteur d'agence répète pour capter TOUTES les6#   infos de la fiche source (comme remax_quebec.py) : description JSON-LD,7#   coordonnées, aplatissement HTML, application au PropertyListing.8# -----------------------------------------------------------------------------9from __future__ import annotations1011import html as _html12import json13import re1415from ..schema import PropertyListing1617_LD_RE = re.compile(r'<script[^>]+application/ld\+json[^>]*>(.*?)</script>', re.S | re.I)18_COORD_RE = re.compile(r'(?:google\.[^"\']*?[?&](?:q|query|ll|center)=|maps/@)'19                       r'(-?\d{1,2}\.\d+)[ ,%+A-Za-z]+?(-?\d{2,3}\.\d+)')20_LD_PROP_TYPES = {"RealEstateListing", "Residence", "SingleFamilyResidence",21                  "House", "Apartment", "Product", "Offer", "Place", "Accommodation"}222324def ld_nodes(html: str):25    """Itère les objets JSON-LD (aplatis depuis @graph)."""26    for block in _LD_RE.findall(html):27        try:28            data = json.loads(block)29        except ValueError:30            continue31        graph = data.get("@graph", [data]) if isinstance(data, dict) else data32        for node in (graph if isinstance(graph, list) else [graph]):33            if isinstance(node, dict):34                yield node353637def ld_description(html: str) -> str:38    """Description depuis un nœud JSON-LD de type propriété (le plus long trouvé)."""39    best = ""40    for n in ld_nodes(html):41        t = n.get("@type")42        types = t if isinstance(t, list) else [t]43        if any(x in _LD_PROP_TYPES for x in types) and n.get("description"):44            d = _html.unescape(str(n["description"])).strip()45            if len(d) > len(best):46                best = d47    return best484950def gmaps_coords(html: str) -> tuple[float, float] | None:51    m = _COORD_RE.search(html)52    if not m:53        return None54    try:55        lat, lng = float(m.group(1)), float(m.group(2))56    except ValueError:57        return None58    if 44.5 <= lat <= 63.0 and -80.0 <= lng <= -56.0:59        return lat, lng60    return None616263def flatten(html: str) -> str:64    """HTML -> texte « valeur | libellé » pour extraire les tableaux Centris."""65    t = _html.unescape(re.sub(r"<[^>]+>", " | ", html))66    t = re.sub(r"[ \t\r\n]*\|[ \t\r\n|]*", " | ", t)67    return re.sub(r"[ \t]+", " ", t)686970# libellés Centris standard cherchés dans « valeur | libellé » OU « libellé | valeur »71_LABELS = [72    "Type de propriété", "Genre de propriété", "Style de bâtiment",73    "Année de construction", "Superficie habitable", "Superficie du terrain",74    "Superficie du bâtiment (au sol)", "Nombre de pièces", "Nombre d'unités",75    "Stationnement (total)", "Stationnement", "Garage", "Système de chauffage",76    "Énergie pour le chauffage", "Type de fenestration", "Fenêtres", "Toiture",77    "Revêtement", "Sous-sol", "Piscine", "Zonage", "Système d'égouts",78    "Approvisionnement en eau", "Déménagement", "Taxes municipales",79    "Taxes scolaires", "Évaluation municipale (terrain)",80    "Évaluation municipale (bâtiment)", "Cuisine",81]828384def centris_details(text: str) -> dict:85    """Extrait les caractéristiques Centris d'un texte aplati (les deux ordres)."""86    out: dict = {}87    for label in _LABELS:88        lab = re.escape(label)89        m = (re.search(r"([^|]{1,55})\s*\|\s*" + lab + r"\b", text)90             or re.search(lab + r"\b\s*\|\s*([^|]{1,55})", text))91        if m:92            val = m.group(1).strip(" |")93            if val and 1 <= len(val) <= 55 and val.lower() != label.lower():94                out[label] = val95    return out969798_INT_RE = re.compile(r"\d+")99100101def _int(v):102    if v is None:103        return None104    if isinstance(v, (int, float)):105        return int(v) or None106    m = _INT_RE.search(str(v))107    return int(m.group()) if m else None108109110def enrich(connector, listings, limit, parse_fn, key="v1", fetch_html=None):111    """Enrichit `listings` via leur page détail, avec cache BD + plafond `limit`.112113    - `parse_fn(html) -> dict` : extrait les champs riches d'une page détail.114    - `key` : versionne le cache (changer pour forcer un rafraîchissement).115    - `fetch_html(url) -> str` : par défaut connector.get(url).text ; passer116      connector.get_rendered pour les sites derrière Firecrawl/anti-bot.117    """118    if limit <= 0:119        return120    from .. import db121    fetch_html = fetch_html or (lambda u: connector.get(u).text)122    con = db.connect()123    budget = limit124    try:125        for lst in listings:126            cached = db.get_cached_detail(con, connector.source_id, lst.external_id, key)127            if cached is None:128                if budget <= 0:129                    continue130                try:131                    cached = parse_fn(fetch_html(lst.url))132                except Exception:133                    cached = {}134                db.put_cached_detail(con, connector.source_id, lst.external_id, key, cached)135                budget -= 1136            apply_detail(lst, cached)137    finally:138        con.close()139140141def apply_detail(lst: PropertyListing, d: dict) -> None:142    """Applique un payload détail au PropertyListing sans écraser les valeurs déjà143    présentes (sauf images : on garde la plus grande galerie)."""144    if not d:145        return146    imgs = d.get("images")147    if imgs and len(imgs) > len(lst.images):148        lst.images = imgs149    if d.get("features"):150        # fusionne en dédoublonnant151        seen = {f.lower() for f in lst.features}152        for f in d["features"]:153            if f.lower() not in seen:154                lst.features.append(f)155                seen.add(f.lower())156    if d.get("details"):157        lst.details.update(d["details"])158    if d.get("broker_name"):159        lst.broker_name = d["broker_name"]160    for f in ("description", "price_label"):161        if d.get(f) and not getattr(lst, f, ""):162            setattr(lst, f, d[f])163    for f in ("bedrooms", "bathrooms", "powder_rooms", "year_built",164              "area_sqft", "lot_sqft", "lat", "lng", "broker_phone", "price"):165        if d.get(f) is not None and getattr(lst, f, None) in (None, "", 0):166            setattr(lst, f, d[f])167