spb/lou-ka Public

1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/cosoltec.py : connecteur Cosoltec (cosoltec.com — projets locatifs5#   Evado à Sainte-Thérèse, Le Monroe à Blainville, Natür à Saint-Jérôme).6#   Les sites vitrines (evado.ca, lemonroe.ca, naturcondos.ca) affichent leurs7#   unités via le widget Planpoint : on interroge la même API JSON publique8#   (app.planpoint.io/api/{groups,projects}/find, POST namespace+hostName,9#   3 requêtes/sync) — prix mensuel, pi², chambres, statut, plans et photos10#   par unité ; seules les unités « Available » sont retenues.11# -----------------------------------------------------------------------------12from __future__ import annotations1314import time1516from ..schema import Listing, normalize_unit_type17from .base import BaseConnector1819PLANPOINT_API = "https://app.planpoint.io/api"2021# les trois vitrines locatives de Cosoltec branchées sur Planpoint22_SITES = [23    {"kind": "groups", "namespace": "evado", "hostName": "evado",24     "page": "https://www.evado.ca/plans", "city": "Sainte-Thérèse"},25    {"kind": "groups", "namespace": "monroe", "hostName": "monroe",26     "page": "https://www.lemonroe.ca/plans", "city": "Blainville"},27    {"kind": "projects", "namespace": "cosolte", "hostName": "Natur",28     "page": "https://www.naturcondos.ca/", "city": "Saint-Jérôme"},29]303132class CosoltecConnector(BaseConnector):33    source_id = "cosoltec"34    request_delay = 0.835    max_pages = 3        # 1 appel API par vitrine (evado, monroe, natur)3637    # -- POST poli (même throttling/session que get(), donc record/replay ok) --38    def _post_json(self, url: str, payload: dict) -> dict:39        wait = self.request_delay - (time.time() - self._last_request)40        if wait > 0:41            time.sleep(wait)42        resp = self.session.post(url, json=payload, timeout=self.timeout)43        self._last_request = time.time()44        resp.raise_for_status()45        return resp.json()4647    def fetch(self) -> list[Listing]:48        listings: list[Listing] = []49        for site in _SITES:50            try:51                data = self._post_json(52                    f"{PLANPOINT_API}/{site['kind']}/find",53                    {"namespace": site["namespace"], "hostName": site["hostName"]})54            except Exception:55                continue56            # groups/find -> {projects: [...]} ; projects/find -> un seul projet57            projects = data.get("projects") if site["kind"] == "groups" else [data]58            for project in projects or []:59                listings.extend(self._parse_project(project, site))60        return listings6162    # -- un projet Planpoint = un immeuble ---------------------------------------63    def _parse_project(self, project: dict, site: dict) -> list[Listing]:64        name = (project.get("name") or "").strip()65        # adresse « 350 Place Fabien-Drapeau, Sainte-Thérèse, Quebec J7E 0C4 »66        addr_parts = [p.strip() for p in (project.get("address") or "").split(",")67                      if p.strip()]68        street = addr_parts[0] if addr_parts else ""69        city = addr_parts[1] if len(addr_parts) > 1 else site["city"]70        lat = project.get("lat")71        lng = project.get("lon")7273        res: list[Listing] = []74        for floor in project.get("floors") or []:75            for u in floor.get("units") or []:76                # seules les unités affichées « Disponible » chez la source77                if (u.get("availability") or "").lower() != "available":78                    continue79                uid = u.get("_id") or ""80                if not uid:81                    continue82                price = u.get("price")83                price = float(price) if isinstance(price, (int, float)) and price > 0 else None84                area = u.get("squareFeet")85                area = float(area) if isinstance(area, (int, float)) and area > 0 else None86                images = [i for i in (u.get("images") or []) if isinstance(i, str)]87                images += [i for i in (u.get("layoutGallery") or [])88                           if isinstance(i, str) and i not in images]89                details: dict = {}90                if isinstance(u.get("bathrooms"), (int, float)):91                    details["bathrooms"] = u["bathrooms"]92                if floor.get("name"):93                    details["floor"] = floor["name"]94                res.append(Listing(95                    source=self.source_id,96                    external_id=uid,             # ObjectId Planpoint de l'unité97                    url=site["page"],            # pas de page par unité publiée98                    title=f"{name} — Unité {u.get('name', '')}".strip(" —"),99                    address=street,100                    city=city,101                    unit_type=normalize_unit_type(u.get("bedrooms") or ""),102                    price=price,103                    price_label=f"{price:.0f} $ /mois" if price else "",104                    availability=u.get("availability") or "",105                    area_sqft=area,106                    furnished=u["furnished"] if isinstance(u.get("furnished"), bool) else None,107                    details=details,108                    images=images[:30],109                    lat=lat if (lat is not None and lng is not None) else None,110                    lng=lng if (lat is not None and lng is not None) else None,111                ))112        return res113