# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/_planpoint.py : helper PARTAGÉ pour le widget Planpoint # (app.planpoint.io — même API JSON publique que cosoltec/devimco : # POST /api/{groups,projects}/find avec {namespace, hostName}). # Utilisé par espace_w, m3_laval, quartier_7 et lac_jerome (Rive-Nord). # Pas un connecteur : aucun source_id, ignoré par le registre auto-découvrant. # ----------------------------------------------------------------------------- from __future__ import annotations import time from ..schema import Listing, normalize_unit_type PLANPOINT_API = "https://app.planpoint.io/api" def planpoint_projects(conn, kind: str, namespace: str, host_name: str) -> list[dict]: """Projets d'un compte Planpoint (kind = « groups » ou « projects »). POST throttlé via la session du connecteur ; `groups/find` renvoie {projects: [...]}, `projects/find` renvoie un projet unique. """ wait = conn.request_delay - (time.time() - conn._last_request) if wait > 0: time.sleep(wait) resp = conn.session.post(f"{PLANPOINT_API}/{kind}/find", json={"namespace": namespace, "hostName": host_name}, timeout=conn.timeout) conn._last_request = time.time() resp.raise_for_status() data = resp.json() if kind == "groups": return data.get("projects") or [] return [data] if isinstance(data, dict) else [] def project_images(project: dict) -> list[str]: """Galerie du projet (photos communes) + image de couverture.""" imgs = [i for i in (project.get("images") or []) if isinstance(i, str) and i.startswith("http")] cover = project.get("projectImageUrl") if isinstance(cover, str) and cover.startswith("http") and cover not in imgs: imgs.append(cover) return imgs def unit_listing(source_id: str, project: dict, floor: dict, unit: dict, *, page_url: str, default_city: str, building: str = "", extra_amenities: list[str] | None = None, ) -> Listing | None: """Une unité Planpoint « Available » -> Listing standard (sinon None). Champs communs aux sites Planpoint : external_id = ObjectId de l'unité (stable), prix mensuel s'il est publié (>0, jamais inventé), pi², inclusions françaises (inclusionsArr), plans + photos d'unité puis galerie du projet, date de livraison -> disponibilité. """ if (unit.get("availability") or "").lower() != "available": return None uid = unit.get("_id") or "" if not uid: return None name = (project.get("name") or "").strip() raw_addr = (project.get("address") or "").strip() addr_parts = [p.strip() for p in raw_addr.split(",") if p.strip()] street = addr_parts[0] if addr_parts else "" city = addr_parts[1] if len(addr_parts) > 1 else default_city lat, lng = project.get("lat"), project.get("lon") price = unit.get("price") price = float(price) if isinstance(price, (int, float)) and price > 0 else None area = unit.get("squareFeet") area = float(area) if isinstance(area, (int, float)) and area > 0 else None # type d'unité : champ `type` s'il ressemble à « 4 1/2 » / « 4,5-G », # sinon dérivé du nombre de chambres (« 2 bedrooms » -> 4½) unit_type = normalize_unit_type(unit.get("type") or "") if not unit_type.endswith("½") and unit_type != "6½+": unit_type = normalize_unit_type(unit.get("bedrooms") or "") or unit_type images = [i for i in (unit.get("images") or []) if isinstance(i, str) and i.startswith("http")] images += [i for i in (unit.get("layoutGallery") or []) if isinstance(i, str) and i.startswith("http") and i not in images] images += [i for i in project_images(project) if i not in images] amenities: list[str] = [] for inc in unit.get("inclusionsArr") or []: lbl = ((inc.get("fr") or inc.get("en") or "").strip() if isinstance(inc, dict) else str(inc).strip()) if lbl and lbl not in amenities: amenities.append(lbl[0].upper() + lbl[1:]) for lbl in extra_amenities or []: if lbl not in amenities: amenities.append(lbl) details: dict = {} if isinstance(unit.get("bathrooms"), (int, float)): details["bathrooms"] = unit["bathrooms"] if floor.get("name"): details["floor"] = floor["name"] if unit.get("orientation"): details["orientation"] = unit["orientation"] if unit.get("type"): details["model"] = unit["type"] delivery = str(unit.get("deliveryDate") or "")[:10] availability = f"Disponible le {delivery}" if delivery else "Disponible" facts: list[str] = [] where = building or name if unit.get("name"): facts.append(f"Unité {unit['name']}" + (f" — {where}" if where else "") + f", {city}.") bits: list[str] = [] if unit.get("type"): bits.append(f"modèle {unit['type']}") if floor.get("name"): bits.append(f"étage {floor['name']}") if area: bits.append(f"{area:g} pi²") if isinstance(unit.get("bathrooms"), (int, float)): bits.append(f"{unit['bathrooms']:g} salle(s) de bain") if bits: facts.append(", ".join(bits).capitalize() + ".") if amenities: facts.append("Inclusions : " + ", ".join(amenities) + ".") return Listing( source=source_id, external_id=uid, # ObjectId Planpoint de l'unité url=page_url, # pas de page publique par unité title=f"{name} — Unité {unit.get('name', '')}".strip(" —"), address=street, city=city, unit_type=unit_type, price=price, price_label=f"{price:.0f} $ /mois" if price else "", availability=availability, availability_date=delivery or None, area_sqft=area, furnished=unit["furnished"] if isinstance(unit.get("furnished"), bool) else None, description=" ".join(facts)[:2000], amenities=amenities, details=details, images=images[:30], lat=lat if (lat is not None and lng is not None) else None, lng=lng if (lat is not None and lng is not None) else None, )