spb/lou-ka

1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/chartwell.py : Chartwell (chartwell.com/fr) — résidences pour5#   retraités. Chaîne PANCANADIENNE : on ne garde QUE le Québec, garanti deux6#   fois — par le préfixe /fr/qc/ des URLs du sitemap ET par le champ7#   Province Abbreviation == "QC" des données de la page.8#   Site Next.js/Sitecore rendu serveur : chaque page « plans des9#   appartements » (/fr/qc/<ville>/<résidence>/plans-des-appartements)10#   embarque dans __NEXT_DATA__ la fiche complète de la résidence (nom,11#   adresse civique, ville, lat/lng, téléphone) et la liste12#   « Property Suit Plans » : type d'unité (Studio, 2 1/2 … 5 1/2), prix13#   « À partir de » (Regular SuitePrice), niveau de soins (Autonome…),14#   inclusions (Key Features) et visuel. Une annonce Lou-Ka par résidence ×15#   type d'unité, prix plancher « à partir de ».16# -----------------------------------------------------------------------------17from __future__ import annotations1819import json20import re2122from ..schema import Listing, normalize_unit_type23from .base import BaseConnector2425SITEMAP = "https://chartwell.com/sitemap.xml"2627_PLANS_URL_RE = re.compile(28    r"https://chartwell\.com/fr/qc/[a-z0-9-]+/[a-z0-9-]+/plans-des-appartements$")29_NEXT_DATA_RE = re.compile(30    r'<script id="__NEXT_DATA__"[^>]*>([\s\S]*?)</script>')313233def _jv(fields: dict, name: str):34    """Valeur `jsonValue` d'un champ Sitecore (liste de {name, jsonValue})."""35    f = fields.get(name)36    return f.get("jsonValue") if isinstance(f, dict) else None373839def _scalar(fields: dict, name: str) -> str:40    v = _jv(fields, name)41    if isinstance(v, dict):42        return str(v.get("value") or "").strip()43    return ""444546class ChartwellConnector(BaseConnector):47    source_id = "chartwell"48    request_delay = 0.849    timeout = 455051    # -- fiche résidence (champs Sitecore du parent de la page plans) ----------------52    @staticmethod53    def _property_fields(next_data: dict) -> dict | None:54        """Trouve le nœud « fields » du bien (celui qui porte55        Property Suit Plans) n'importe où dans le layout Sitecore."""56        stack = [next_data]57        while stack:58            node = stack.pop()59            if isinstance(node, dict):60                if isinstance(node.get("fields"), list):61                    names = {f.get("name") for f in node["fields"]62                             if isinstance(f, dict)}63                    if "Property Suit Plans" in names and "Latitude" in names:64                        return {f["name"]: f for f in node["fields"]65                                if isinstance(f, dict) and f.get("name")}66                stack.extend(node.values())67            elif isinstance(node, list):68                stack.extend(node)69        return None7071    def _parse_page(self, url: str, html: str) -> list[Listing]:72        m = _NEXT_DATA_RE.search(html)73        if not m:74            return []75        try:76            data = json.loads(m.group(1))77        except ValueError:78            return []79        fields = self._property_fields(data)80        if not fields:81            return []8283        # garde-fou province : Chartwell est pancanadien, Québec seulement84        prov = _jv(fields, "Province") or []85        abbr = ""86        if isinstance(prov, list) and prov:87            abbr = str(((prov[0].get("fields") or {})88                        .get("Province Abbreviation") or {}).get("value") or "")89        if abbr and abbr.upper() != "QC":90            return []9192        name = (_scalar(fields, "NavigationTitle")93                or _scalar(fields, "Short Property Name")94                or _scalar(fields, "Property Name"))95        address = _scalar(fields, "StreetNameAndNumber")96        postal = _scalar(fields, "Postal code")97        phone = _scalar(fields, "Contact Number")98        city = ""99        cv = _jv(fields, "City") or []100        if isinstance(cv, list) and cv:101            # « City Name » porte les accents (Lévis), displayName non (Levis)102            city = str(((cv[0].get("fields") or {}).get("City Name") or {})103                       .get("value")104                       or cv[0].get("displayName") or cv[0].get("name") or "")105        lat = lng = None106        try:107            lat = float(_scalar(fields, "Latitude"))108            lng = float(_scalar(fields, "Longitude"))109        except ValueError:110            lat = lng = None111112        thumb = ""113        tv = _jv(fields, "Thumbnail Photo")114        if isinstance(tv, dict):115            thumb = str(((tv.get("value") or {}).get("src")) or "")116117        res_slug = url.rstrip("/").rsplit("/", 2)[-2]118        city_slug = url.rstrip("/").rsplit("/", 3)[-3]119120        def val(d, *keys):121            cur = d122            for k in keys:123                if not isinstance(cur, dict):124                    return None125                cur = cur.get(k)126            return cur127128        out: list[Listing] = []129        plans = _jv(fields, "Property Suit Plans") or []130        for plan in plans:131            pf = plan.get("fields") if isinstance(plan, dict) else None132            if not isinstance(pf, dict):133                continue134            suite = val(pf, "SuiteName", "fields", "suiteType", "value") or ""135            price_raw = val(pf, "Regular SuitePrice", "value")136            care = val(pf, "Care Level", "fields",137                       "Suite Care Level", "value") or ""138            try:139                price = float(str(price_raw).replace(" ", "").replace(",", ""))140            except (TypeError, ValueError):141                continue142            if not suite or not 400 <= price <= 20000:143                continue144            feats = []145            for kf in (val(pf, "Key Features") or []):146                t = val(kf, "fields", "Key Feature Name", "value")147                if t and t not in feats:148                    feats.append(str(t))149            images = []150            bg = val(pf, "background Image", "value", "src")151            if bg:152                images.append(str(bg))153            if thumb:154                images.append(thumb)155            details = {"Résidence pour aînés": "oui"}156            if care:157                details["Niveau de soins"] = care158            if postal:159                details["Code postal"] = postal160            if phone:161                details["contact"] = {"phone": phone}162            eid = (f"{city_slug}-{res_slug}-"163                   f"{re.sub(r'[^a-z0-9]+', '-', suite.lower()).strip('-')}")164            out.append(Listing(165                source=self.source_id,166                external_id=eid,167                url=url,168                title=f"{suite} — {name}",169                address=address,170                city=city,171                unit_type=normalize_unit_type(suite),172                price=price,173                price_label=f"À partir de {price:,.0f} $ par mois"174                            .replace(",", " "),175                amenities=feats,176                details=details,177                images=images,178                lat=lat,179                lng=lng,180            ))181        return out182183    def fetch(self) -> list[Listing]:184        out: list[Listing] = []185        xml = self.get(SITEMAP).text186        urls = sorted({u for u in re.findall(r"<loc>([^<]+)</loc>", xml)187                       if _PLANS_URL_RE.match(u)})188        for url in urls:189            try:190                out.extend(self._parse_page(url, self.get(url).text))191            except Exception:192                continue193        return out194