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/profusion.py : Profusion Immobilier (luxe, affilié Forbes/Christie's)5#   Énumération : la page « nos-inscriptions » liste toutes les fiches en HTML6#   rendu serveur, sous forme /propriete/{no-centris}/. Chaque fiche publie un7#   JSON-LD RealEstateListing (adresse, description, prix, courtier) + un bloc8#   de specs « Libellé: valeur » + une galerie profusionimmo.ca.9# -----------------------------------------------------------------------------10from __future__ import annotations1112import html as _html13import json14import os15import re1617from .base import BaseConnector18from . import _detailutil as du19from ..schema import PropertyListing2021SITE = "https://www.profusion.global"22LISTING_PAGE = f"{SITE}/acheter/nos-inscriptions/"23DETAIL_LIMIT = int(os.environ.get("IMMOKA_PROFUSION_DETAIL_LIMIT", "500"))2425_PROP_RE = re.compile(r'/propriete/(\d{5,})/')26_IMG_RE = re.compile(r'https://profusionimmo\.ca/public/user_files/proprietes/fiche/\d+-\d+\.jpg', re.I)27# nom JSON-LD : « {Type} à vendre à {Ville (Secteur)} – {Adresse civique} »28_NAME_RE = re.compile(r'^(.*?)\s+à vendre à\s+(.+?)\s*$', re.I)29_SPEC_RE = re.compile(r'([A-Za-zÀ-ÿ\'’()/ ]{3,40})\s*:\s*([^<|]{1,60})')303132class ProfusionConnector(BaseConnector):33    source_id = "profusion"34    request_delay = 0.535    use_detail_cache = True3637    def fetch(self) -> list[PropertyListing]:38        try:39            html = self.get(LISTING_PAGE).text40        except Exception:41            return []42        ids: list[str] = []43        seen = set()44        for m in _PROP_RE.finditer(html):45            if m.group(1) not in seen:46                seen.add(m.group(1))47                ids.append(m.group(1))48        listings = [PropertyListing(source=self.source_id, external_id=cid,49                                    url=f"{SITE}/propriete/{cid}/", mls=cid,50                                    broker_name="Profusion Immobilier")51                    for cid in ids]52        self._enrich(listings)53        return listings5455    def _enrich(self, listings: list[PropertyListing]) -> None:56        from .. import db57        con = db.connect()58        budget = DETAIL_LIMIT59        try:60            for lst in listings:61                d = db.get_cached_detail(con, self.source_id, lst.external_id, "v1")62                if d is None:63                    if budget <= 0:64                        continue65                    try:66                        d = parse_profusion_detail(self.get(lst.url).text)67                    except Exception:68                        d = {}69                    db.put_cached_detail(con, self.source_id, lst.external_id, "v1", d)70                    budget -= 171                _apply(lst, d)72        finally:73            con.close()747576def parse_profusion_detail(html: str) -> dict:77    out: dict = {}78    title = ""79    for n in du.ld_nodes(html):80        t = n.get("@type")81        types = t if isinstance(t, list) else [t]82        if "RealEstateListing" in types:83            title = _html.unescape(n.get("name", "")).strip()84            if n.get("description"):85                out["description"] = _html.unescape(str(n["description"])).strip()86            offer = n.get("offers") or {}87            price = offer.get("price")88            try:89                price = float(price)90                if price >= 10_000:91                    out["price"] = price92            except (TypeError, ValueError):93                pass94            seller = offer.get("seller") or {}95            if seller.get("name"):96                out["broker_name"] = seller["name"]97            if seller.get("telephone"):98                out["broker_phone"] = str(seller["telephone"])99            break100    if title:101        out["title"] = title102        m = _NAME_RE.search(title)103        if m:104            type_phrase, loc = m.group(1).strip(), m.group(2).strip()105            out["property_type"] = type_phrase          # « Maison », « Terrain », « Condo »…106            # séparer « Ville (Secteur) » de l'adresse civique (après le tiret)107            parts = re.split(r"\s+[–—-]\s+", loc, maxsplit=1)108            city_block = parts[0].strip()109            if len(parts) > 1:110                out["address"] = parts[1].strip()111            ps = re.search(r"\(([^)]+)\)", city_block)112            out["city"] = re.sub(r"\s*\([^)]*\)", "", city_block).strip()113            if ps:114                out["sector"] = ps.group(1).strip()115116    # galerie complète117    imgs, seen = [], set()118    for u in _IMG_RE.findall(html):119        if u not in seen:120            seen.add(u)121            imgs.append(u)122    if imgs:123        out["images"] = imgs124125    # specs « Libellé: valeur »126    text = re.sub(r"\s+", " ", _html.unescape(re.sub(r"<[^>]+>", " ", html)))127    details = {}128    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|$)"),129                      ("Chambres", r"Chambres\s*:\s*(\d+)"),130                      ("Salles de bain", r"Salles? de bain\s*:\s*(\d+)"),131                      ("Année de construction", r"Ann[ée]e\s*:\s*(\d{4})"),132                      ("Superficie habitable", r"Superficie\s*:\s*([\d  ,]+\s*(?:pi²|pi2|m²)?)")):133        m = re.search(rx, text, re.I)134        if m:135            details[label] = m.group(1).strip()136    ptype = details.get("Type de propriété", "")137    if ptype:138        out["property_type"] = ptype139    if details.get("Chambres"):140        out["bedrooms"] = int(re.search(r"\d+", details["Chambres"]).group())141    if details.get("Salles de bain"):142        out["bathrooms"] = int(re.search(r"\d+", details["Salles de bain"]).group())143    if details.get("Année de construction") and details["Année de construction"] != "0":144        out["year_built"] = int(details["Année de construction"])145    if details:146        out["details"] = {k: v for k, v in details.items() if v and v != "0"}147    return out148149150def _apply(lst: PropertyListing, d: dict) -> None:151    if not d:152        return153    # le vrai courtier (fiche) remplace le placeholder « Profusion Immobilier »154    if d.get("broker_name"):155        lst.broker_name = d["broker_name"]156    for f in ("title", "address", "city", "sector", "property_type",157              "description", "broker_phone"):158        if d.get(f) and not getattr(lst, f, ""):159            setattr(lst, f, d[f])160    if not lst.address and d.get("title"):161        lst.address = d["title"]162    for f in ("price", "bedrooms", "bathrooms", "year_built"):163        if d.get(f) is not None and getattr(lst, f, None) in (None, 0):164            setattr(lst, f, d[f])165    if d.get("images") and len(d["images"]) > len(lst.images):166        lst.images = d["images"]167    if d.get("details"):168        lst.details.update(d["details"])169