# ----------------------------------------------------------------------------- # Immo-Ka — Agrégateur de maisons à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/profusion.py : Profusion Immobilier (luxe, affilié Forbes/Christie's) # Énumération : la page « nos-inscriptions » liste toutes les fiches en HTML # rendu serveur, sous forme /propriete/{no-centris}/. Chaque fiche publie un # JSON-LD RealEstateListing (adresse, description, prix, courtier) + un bloc # de specs « Libellé: valeur » + une galerie profusionimmo.ca. # ----------------------------------------------------------------------------- from __future__ import annotations import html as _html import json import os import re from .base import BaseConnector from . import _detailutil as du from ..schema import PropertyListing SITE = "https://www.profusion.global" LISTING_PAGE = f"{SITE}/acheter/nos-inscriptions/" DETAIL_LIMIT = int(os.environ.get("IMMOKA_PROFUSION_DETAIL_LIMIT", "500")) _PROP_RE = re.compile(r'/propriete/(\d{5,})/') _IMG_RE = re.compile(r'https://profusionimmo\.ca/public/user_files/proprietes/fiche/\d+-\d+\.jpg', re.I) # nom JSON-LD : « {Type} à vendre à {Ville (Secteur)} – {Adresse civique} » _NAME_RE = re.compile(r'^(.*?)\s+à vendre à\s+(.+?)\s*$', re.I) _SPEC_RE = re.compile(r'([A-Za-zÀ-ÿ\'’()/ ]{3,40})\s*:\s*([^<|]{1,60})') class ProfusionConnector(BaseConnector): source_id = "profusion" request_delay = 0.5 use_detail_cache = True def fetch(self) -> list[PropertyListing]: try: html = self.get(LISTING_PAGE).text except Exception: return [] ids: list[str] = [] seen = set() for m in _PROP_RE.finditer(html): if m.group(1) not in seen: seen.add(m.group(1)) ids.append(m.group(1)) listings = [PropertyListing(source=self.source_id, external_id=cid, url=f"{SITE}/propriete/{cid}/", mls=cid, broker_name="Profusion Immobilier") for cid in ids] self._enrich(listings) return listings def _enrich(self, listings: list[PropertyListing]) -> None: from .. import db con = db.connect() budget = DETAIL_LIMIT try: for lst in listings: d = db.get_cached_detail(con, self.source_id, lst.external_id, "v1") if d is None: if budget <= 0: continue try: d = parse_profusion_detail(self.get(lst.url).text) except Exception: d = {} db.put_cached_detail(con, self.source_id, lst.external_id, "v1", d) budget -= 1 _apply(lst, d) finally: con.close() def parse_profusion_detail(html: str) -> dict: out: dict = {} title = "" for n in du.ld_nodes(html): t = n.get("@type") types = t if isinstance(t, list) else [t] if "RealEstateListing" in types: title = _html.unescape(n.get("name", "")).strip() if n.get("description"): out["description"] = _html.unescape(str(n["description"])).strip() offer = n.get("offers") or {} price = offer.get("price") try: price = float(price) if price >= 10_000: out["price"] = price except (TypeError, ValueError): pass seller = offer.get("seller") or {} if seller.get("name"): out["broker_name"] = seller["name"] if seller.get("telephone"): out["broker_phone"] = str(seller["telephone"]) break if title: out["title"] = title m = _NAME_RE.search(title) if m: type_phrase, loc = m.group(1).strip(), m.group(2).strip() out["property_type"] = type_phrase # « Maison », « Terrain », « Condo »… # séparer « Ville (Secteur) » de l'adresse civique (après le tiret) parts = re.split(r"\s+[–—-]\s+", loc, maxsplit=1) city_block = parts[0].strip() if len(parts) > 1: out["address"] = parts[1].strip() ps = re.search(r"\(([^)]+)\)", city_block) out["city"] = re.sub(r"\s*\([^)]*\)", "", city_block).strip() if ps: out["sector"] = ps.group(1).strip() # galerie complète imgs, seen = [], set() for u in _IMG_RE.findall(html): if u not in seen: seen.add(u) imgs.append(u) if imgs: out["images"] = imgs # specs « Libellé: valeur » text = re.sub(r"\s+", " ", _html.unescape(re.sub(r"<[^>]+>", " ", html))) details = {} for label, rx in (("Type de propriété", r"Type de propri[ée]t[ée]\s*:\s*([^:]{2,40}?)(?:\s{2,}|Chambres|Ann[ée]e|Superficie|$)"), ("Chambres", r"Chambres\s*:\s*(\d+)"), ("Salles de bain", r"Salles? de bain\s*:\s*(\d+)"), ("Année de construction", r"Ann[ée]e\s*:\s*(\d{4})"), ("Superficie habitable", r"Superficie\s*:\s*([\d ,]+\s*(?:pi²|pi2|m²)?)")): m = re.search(rx, text, re.I) if m: details[label] = m.group(1).strip() ptype = details.get("Type de propriété", "") if ptype: out["property_type"] = ptype if details.get("Chambres"): out["bedrooms"] = int(re.search(r"\d+", details["Chambres"]).group()) if details.get("Salles de bain"): out["bathrooms"] = int(re.search(r"\d+", details["Salles de bain"]).group()) if details.get("Année de construction") and details["Année de construction"] != "0": out["year_built"] = int(details["Année de construction"]) if details: out["details"] = {k: v for k, v in details.items() if v and v != "0"} return out def _apply(lst: PropertyListing, d: dict) -> None: if not d: return # le vrai courtier (fiche) remplace le placeholder « Profusion Immobilier » if d.get("broker_name"): lst.broker_name = d["broker_name"] for f in ("title", "address", "city", "sector", "property_type", "description", "broker_phone"): if d.get(f) and not getattr(lst, f, ""): setattr(lst, f, d[f]) if not lst.address and d.get("title"): lst.address = d["title"] for f in ("price", "bedrooms", "bathrooms", "year_built"): if d.get(f) is not None and getattr(lst, f, None) in (None, 0): setattr(lst, f, d[f]) if d.get("images") and len(d["images"]) > len(lst.images): lst.images = d["images"] if d.get("details"): lst.details.update(d["details"])