# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/logis_experts.py : connecteur Logis-Experts (logis-experts.com) # Gestionnaire-constructeur, centre-ville de Québec + Lévis (~190 immeubles). # Site Laravel/Alpine avec API JSON propre : # GET /api/search -> {"buildings": [{address{lat,lng,zipcode}, all_images, # apartments: [{number, type, price, available_type}]}]} # La page détail d'une unité (/immeuble//appartement/) ajoute # superficie, date de disponibilité, frais d'énergie et commodités — visitée # via le cache self.detail() (seulement si l'unité est nouvelle/modifiée). # ----------------------------------------------------------------------------- from __future__ import annotations import re from ..schema import Listing, infer_city from .base import BaseConnector BASE = "https://logis-experts.com" API_SEARCH = f"{BASE}/api/search" # Ville renvoyée par l'API (en minuscules) -> nom d'affichage CITY_MAP = { "quebec": "Québec", "québec": "Québec", "levis": "Lévis", "lévis": "Lévis", "saint-augustin-de-desmaures": "Saint-Augustin-de-Desmaures", } # « 3.5+office » -> « 3½ », « maisondeville2chambres » -> « Maison », etc. _TYPE_DEMI_RE = re.compile(r"^(\d)\.5") _TAG_RE = re.compile(r"<[^>]+>") _SEP = r"[§\s]+" # les balises deviennent « § » ; tolérer § et espaces mêlés _AREA_RE = re.compile(rf"Superficie{_SEP}([\d\s]+?)\s*pi{_SEP}2", re.I) _AVAIL_RE = re.compile(r"Disponible à partir du\s+([^§<]{3,40})", re.I) _ENERGY_RE = re.compile(rf"Frais d'énergie{_SEP}([^§]{{3,60}})", re.I) # sections « Services », « Commodités », « Télécommunications » : puces « - item » _AMEN_ITEM_RE = re.compile(r"- ([^§]{2,60}?) §") def _unit_type(raw: str) -> str: s = (raw or "").strip().lower() m = _TYPE_DEMI_RE.match(s) if m: return f"{m.group(1)}½" if "maisondeville" in s: return "Maison" if "loft" in s: return "Loft" if "studio" in s or "bachelor" in s: return "Studio" if "chambre" in s: return "Chambre" return raw.strip() class _DetailBudgetExceeded(Exception): """Plafond de pages détail atteint pour cette sync (repris à la suivante).""" class LogisExpertsConnector(BaseConnector): source_id = "logis_experts" request_delay = 0.8 max_details = 150 # garde-fou : vraies visites détail par sync def __init__(self) -> None: super().__init__() self._n_details = 0 # -- page détail d'une unité ---------------------------------------------- def _fetch_detail(self, url: str) -> dict: """Extrait superficie / disponibilité / énergie / commodités du HTML. Seules les vraies requêtes comptent contre max_details : les unités déjà en cache (self.detail) ne passent pas par ici. """ if self._n_details >= self.max_details: raise _DetailBudgetExceeded(url) self._n_details += 1 import html as _entities html = _entities.unescape(self.get(url).text) # texte aplati : les balises deviennent des « § » pour garder les # frontières de champs sans dépendre de la structure exacte flat = html.replace(" ", " ") flat = re.sub(r"\s+", " ", _TAG_RE.sub(" § ", flat)) out: dict = {} m = _AREA_RE.search(flat) if m: try: out["area_sqft"] = float(m.group(1).replace(" ", "")) except ValueError: pass m = _AVAIL_RE.search(flat) if m: out["availability"] = m.group(1).strip(" .§") m = _ENERGY_RE.search(flat) if m: out["energy"] = m.group(1).strip(" .§") # ex. « Chauffé & Éclairé » # commodités : entre le premier titre de section et « Suivez-nous » start = None for titre in ("Services", "Commodités", "Télécommunications"): j = flat.find(titre) if j >= 0 and (start is None or j < start): start = j if start is not None: window = flat[start:start + 2500] fin = window.find("Suivez-nous") if fin > 0: window = window[:fin] items = [it.strip() for it in _AMEN_ITEM_RE.findall(window)] if items: out["amenities"] = list(dict.fromkeys(items)) return out def fetch(self) -> list[Listing]: data = self.get(API_SEARCH, headers={"X-Requested-With": "XMLHttpRequest"}).json() listings: list[Listing] = [] self._n_details = 0 for b in data.get("buildings") or []: addr = b.get("address") or {} handle = b.get("handle") or "" images = [] for img in b.get("all_images") or []: u = (img or {}).get("url") if u: images.append(u) city_raw = (addr.get("city") or "").strip().lower() city = CITY_MAP.get(city_raw, (addr.get("city") or "").title()) sector = addr.get("district_name") or "" civic = addr.get("address") or "" zipcode = addr.get("zipcode") or "" full_addr = ", ".join(x for x in (civic, city, zipcode) if x) lat = addr.get("latitude") lng = addr.get("longitude") b_name = b.get("name") or civic for a in b.get("apartments") or []: if (a.get("available_type") or "").upper() != "AVAILABLE": continue num = str(a.get("number") or "").strip() ext_id = str(a.get("id") or f"{handle}-{num}") url = f"{BASE}/immeuble/{handle}/appartement/{num}" price = a.get("price") price = float(price) if isinstance(price, (int, float)) and price else None # page détail (superficie, dispo, énergie, commodités) — via # cache : re-visitée seulement si le contenu liste a changé key = f"{price}|{a.get('type')}|{a.get('available_type')}" try: info = self.detail(ext_id, key, lambda u=url: self._fetch_detail(u)) except Exception: # budget épuisé ou page détail en erreur info = {} amenities = list(info.get("amenities") or []) if info.get("energy"): amenities.insert(0, info["energy"]) listings.append(Listing( source=self.source_id, external_id=ext_id, url=url, title=f"{b_name} — app. {num}" if num else b_name, address=full_addr, sector=sector, city=infer_city(sector, default=city or "Québec"), unit_type=_unit_type(a.get("type") or ""), price=price, price_label=f"{price:.0f} $ /mois" if price else "", availability=info.get("availability", ""), area_sqft=info.get("area_sqft"), description="", amenities=amenities, images=images[:15], lat=float(lat) if lat else None, lng=float(lng) if lng else None, )) return listings