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/imvest.py : connecteur Société Imvest (societeimvest.ca)5#   Le Méribel, 195 logements en 2 phases au 1191, rue de Courchevel à Lévis6#   (secteur Saint-Romuald). WordPress (thème TeamPress) avec plans d'étage7#   interactifs : endpoints AJAX POST maison —8#     /ajax/recherche-etage/ (id=immeuble) -> liste des étages (li.Etage)9#     /ajax/load-etage/ (id=étage)         -> <area> par unité (no, type,10#         superficie dans data-tooltip) + JSON `areas` dont fillColor code le11#         statut : cbcbcb = loué, 940723 = réservé, autre = disponible12#     /ajax/recherche-unite/ (id=unité)    -> détail (disponibilité, plan,13#         caractéristiques)14#   Granularité : unité. Pas de prix publié (Tier B).15# -----------------------------------------------------------------------------16from __future__ import annotations1718import hashlib19import json20import re2122from ..schema import Listing, normalize_unit_type23from .base import BaseConnector2425BASE = "https://www.societeimvest.ca"26AJAX_ETAGES = f"{BASE}/ajax/recherche-etage/"27AJAX_PLAN = f"{BASE}/ajax/load-etage/"28AJAX_UNITE = f"{BASE}/ajax/recherche-unite/"2930# phases (slug de page « immeubles ») — l'id AJAX est lu sur la page31PHASES = [("le-meribel", "Le Méribel — Phase 1"),32          ("le-meribel-phase-2", "Le Méribel — Phase 2")]3334ADDRESS = "1191, rue de Courchevel, Lévis"35LAT, LNG = 46.732973, -71.2629413637IMMEUBLE_ID_RE = re.compile(r"immeuble_id = '(\d+)'")38FLOOR_RE = re.compile(r'<li id="(\d+)" class="Etage"><a class="BoutonEtage"'39                      r' href="#" title="([^"]*)"')40AREA_RE = re.compile(r'<area data-tooltip="([^"]*)" id="(\d+)"[^>]*'41                     r'data-key="(\d+)"')42AREAS_JSON_RE = re.compile(r"areas:\s*(\[.*?\])", re.S)43TYPE_RE = re.compile(r"<strong>([^<]*)</strong>")44SQFT_RE = re.compile(r"Superficie\s*:\s*([\d\s]+)\s*p\.?c\.?", re.I)45DISPO_RE = re.compile(r"Disponibilit[^:]*:\s*<span class=\"Spacing\">"46                      r"([^<]*)</span>", re.I)47PLAN_RE = re.compile(r'href="(https://www\.societeimvest\.ca/wp-content/'48                     r'uploads/[^"]+\.pdf)"')49IMG_RE = re.compile(r'<img[^>]+src="([^"]+phpThumb[^"]+)"')50CARAC_RE = re.compile(r"Caractéristiques du logement</h3>\s*<ul>(.*?)</ul>",51                      re.S)52LI_RE = re.compile(r"<li>(.*?)</li>", re.S)5354# statuts non offerts en location (gris = loué, rouge foncé = réservé)55TAKEN_COLORS = {"cbcbcb", "940723"}565758class ImvestConnector(BaseConnector):59    source_id = "imvest"60    request_delay = 0.66162    def fetch(self) -> list[Listing]:63        listings: list[Listing] = []64        for slug, phase_name in PHASES:65            page_url = f"{BASE}/immeubles/{slug}/"66            try:67                page = self.get(page_url).text68            except Exception:69                continue70            m = IMMEUBLE_ID_RE.search(page)71            if not m:72                continue73            try:74                floors_html = self.post(AJAX_ETAGES,75                                        data={"id": m.group(1)}).text76            except Exception:77                continue78            for floor_id, floor_name in FLOOR_RE.findall(floors_html):79                try:80                    listings.extend(self._parse_floor(81                        slug, phase_name, page_url, floor_id, floor_name))82                except Exception:83                    continue84        return listings8586    def _parse_floor(self, slug: str, phase_name: str, page_url: str,87                     floor_id: str, floor_name: str) -> list[Listing]:88        out: list[Listing] = []89        plan_html = self.post(AJAX_PLAN, data={"id": floor_id}).text9091        # statut de chaque unité (JSON mapster : key = no d'unité)92        status: dict[str, str] = {}93        m = AREAS_JSON_RE.search(plan_html)94        if m:95            try:96                for a in json.loads(m.group(1)):97                    status[str(a.get("key"))] = a.get("fillColor", "")98            except ValueError:99                pass100101        for tooltip, unit_id, unit_no in AREA_RE.findall(plan_html):102            color = status.get(unit_no, "")103            if not color or color.lower() in TAKEN_COLORS:104                continue           # loué / réservé / inconnu105106            tm = TYPE_RE.search(tooltip)107            unit_type = normalize_unit_type(tm.group(1)) if tm else ""108            sm = SQFT_RE.search(tooltip)109            area = float(sm.group(1).replace(" ", "").replace(" ", "")) \110                if sm else None111112            key = hashlib.sha1(f"{color}|{tooltip}".encode()).hexdigest()113            d = self.detail(f"{slug}-{unit_no}", key,114                            lambda uid=unit_id: self._fetch_unit(uid))115116            out.append(Listing(117                source=self.source_id,118                external_id=f"{slug}-{unit_no}",119                url=page_url,120                title=f"Unité {unit_no} — {phase_name}",121                address=ADDRESS,122                sector="Saint-Romuald",123                city="Lévis",124                unit_type=unit_type,125                area_sqft=area,126                price_label="Prix sur demande",127                availability=d.get("availability", ""),128                description=f"Unité {unit_no} ({unit_type}), {floor_name} — "129                            f"{phase_name}, Société Imvest, Lévis.",130                amenities=d.get("amenities") or [],131                details={"floor": floor_name,132                         **({"floor_plan": d["plan"]} if d.get("plan")133                            else {})},134                images=d.get("images") or [],135                lat=LAT,136                lng=LNG,137            ))138        return out139140    def _fetch_unit(self, unit_id: str) -> dict:141        """Détail d'une unité via /ajax/recherche-unite/ (POST id=…)."""142        out: dict = {}143        try:144            html = self.post(AJAX_UNITE, data={"id": unit_id}).text145        except Exception:146            return out147        m = DISPO_RE.search(html)148        if m:149            dispo = m.group(1).strip()150            out["availability"] = dispo if re.match(r"^\d|^libre|^immédiat",151                                                    dispo, re.I) \152                else f"Disponibilité : {dispo}"153        m = PLAN_RE.search(html)154        if m:155            out["plan"] = m.group(1)156        m = CARAC_RE.search(html)157        if m:158            out["amenities"] = [159                re.sub(r"<[^>]+>", "", li).strip()160                for li in LI_RE.findall(m.group(1))161                if re.sub(r"<[^>]+>", "", li).strip()][:20]162        imgs = [u.replace("&amp;", "&") for u in IMG_RE.findall(html)163                if "ajax_loading" not in u]164        if imgs:165            out["images"] = imgs[:5]166        return out167