spb/ora-ka Public

1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (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 les connecteurs de portails répètent pour capter TOUTES les6#   infos de la fiche source : description JSON-LD, coordonnées, application7#   au Listing avec cache BD et budget de requêtes par synchronisation.8# -----------------------------------------------------------------------------9from __future__ import annotations1011import html as _html12import json13import re1415from ..schema import Listing1617_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 de fiches."""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 (fiches de courtiers) cherchés dans un texte aplati71# « valeur | libellé » OU « libellé | valeur » — version location72_LABELS = [73    "Type de propriété", "Genre de propriété", "Style de bâtiment",74    "Année de construction", "Superficie habitable", "Superficie du terrain",75    "Nombre de pièces", "Nombre d'unités", "Stationnement (total)",76    "Stationnement", "Garage", "Système de chauffage",77    "Énergie pour le chauffage", "Piscine", "Déménagement", "Date d'emménagement",78    "Disponibilité", "Bail", "Durée du bail", "Meublé", "Animaux",79    "Inclus dans le loyer", "Cuisine",80]818283def centris_details(text: str) -> dict:84    """Extrait les caractéristiques Centris d'un texte aplati (les deux ordres)."""85    out: dict = {}86    for label in _LABELS:87        lab = re.escape(label)88        m = (re.search(r"([^|]{1,55})\s*\|\s*" + lab + r"\b", text)89             or re.search(lab + r"\b\s*\|\s*([^|]{1,55})", text))90        if m:91            val = m.group(1).strip(" |")92            if val and 1 <= len(val) <= 55 and val.lower() != label.lower():93                out[label] = val94    return out959697def enrich(connector, listings, limit, parse_fn, key="v1", fetch_html=None):98    """Enrichit `listings` via leur page détail, avec cache BD + plafond `limit`.99100    - `parse_fn(html) -> dict` : extrait les champs riches d'une page détail.101    - `key` : versionne le cache (changer pour forcer un rafraîchissement).102    - `fetch_html(url) -> str` : par défaut connector.get(url).text ; passer103      connector.get_rendered / get_scrapfly pour les sites derrière anti-bot.104    Le budget `limit` fait que chaque sync n'enrichit qu'un quota de fiches ;105    le parc se complète sur plusieurs cycles (le cache est permanent tant que106    la clé ne change pas).107    """108    if limit <= 0:109        return110    from .. import db111    fetch_html = fetch_html or (lambda u: connector.get(u).text)112    con = db.connect()113    budget = limit114    try:115        for lst in listings:116            cached = db.get_cached_detail(con, connector.source_id, lst.external_id, key)117            if cached is None:118                if budget <= 0:119                    continue120                try:121                    cached = parse_fn(fetch_html(lst.url))122                except Exception:123                    cached = {}124                db.put_cached_detail(con, connector.source_id, lst.external_id, key, cached)125                budget -= 1126            apply_detail(lst, cached)127    finally:128        con.close()129130131def apply_detail(lst: Listing, d: dict) -> None:132    """Applique un payload détail au Listing sans écraser les valeurs déjà133    présentes (sauf images : on garde la plus grande galerie)."""134    if not d:135        return136    imgs = d.get("images")137    if imgs and len(imgs) > len(lst.images):138        lst.images = imgs139    if d.get("amenities"):140        seen = {a.lower() for a in lst.amenities}141        for a in d["amenities"]:142            if a.lower() not in seen:143                lst.amenities.append(a)144                seen.add(a.lower())145    if d.get("details"):146        merged = dict(d["details"])147        merged.update(lst.details)      # les valeurs du connecteur priment148        lst.details = merged149    # description : on garde la plus riche (la fiche détail bat le résumé liste)150    if d.get("description") and len(d["description"]) > len(lst.description or ""):151        lst.description = d["description"]152    for f in ("price_label", "address", "city", "sector", "unit_type",153              "availability", "title"):154        if d.get(f) and not getattr(lst, f, ""):155            setattr(lst, f, d[f])156    for f in ("price", "area_sqft", "lat", "lng", "availability_date",157              "pets", "furnished"):158        if d.get(f) is not None and getattr(lst, f, None) is None:159            setattr(lst, f, d[f])160