# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/celeste.py : connecteur CÉLESTE LaSalle (louerceleste.com) # 815, rue Gagné, LaSalle (Montréal) — deux immeubles (A et B), 3½ à 5½. # WordPress/Avada : chaque unité est un « avada_portfolio » exposé par la # REST API (/wp-json/wp/v2/avada_portfolio, ~452 items FR+EN). Taxonomie # portfolio_category : 62/64/66 = types FR (3½/4½/5½), 60 = loué, # 58 = disponible. Candidats = items FR marqués disponibles ou non marqués # loués ; la fiche fait foi (« Disponibilité: LOUÉ » -> exclu). La fiche # fournit grandeur, étage, salles de bain, « Loyer : à partir de $…/mois », # détails/équipements et la galerie. Inventaire souvent nul (immeubles # pleins) : le connecteur retourne alors une liste vide, ce qui est correct. # ----------------------------------------------------------------------------- from __future__ import annotations import re from bs4 import BeautifulSoup from ..schema import Listing from .base import BaseConnector BASE = "https://louerceleste.com" API = f"{BASE}/wp-json/wp/v2/avada_portfolio" ADDRESS = "815, rue Gagné, LaSalle, Montréal, QC" CAT_TYPES_FR = {62: "3½", 64: "4½", 66: "5½"} # types (taxonomie FR) CAT_LOUE = 60 # marqué loué CAT_DISPO = 58 # marqué disponible _PRICE_RE = re.compile(r"Loyer\s*:?\s*\|?\s*à partir de\s*\$?\s*([\d\s,]+)" r"\s*/\s*mois", re.I) _FLOOR_RE = re.compile(r"Étage\s*:\s*\|?\s*(\d+)", re.I) _BATH_RE = re.compile(r"salles? de bain\s*:\s*\|?\s*([\d.]+)", re.I) _SIZE_RE = re.compile(r"Grandeur\s*:\s*\|?\s*(\d\s*½|\d\s*1/2)", re.I) _LOUE_RE = re.compile(r"Disponibilité\s*:\s*\|?\s*LOU[ÉE]", re.I) _IMG_RE = re.compile( r'https://louerceleste\.com/wp-content/uploads/[^"\'\s\\)]+?' r'\.(?:jpg|jpeg|png|webp)', re.I) _SKIP_IMG = re.compile(r"logo|icon|favicon|-\d{2,4}x\d{2,4}\.", re.I) class CelesteConnector(BaseConnector): source_id = "celeste" request_delay = 0.6 max_candidates = 40 # garde-fou de crawl des fiches max_images = 20 # -- inventaire REST (toutes les unités, FR + EN) --------------------------- def _portfolio_items(self) -> list[dict]: items: list[dict] = [] for page in range(1, 12): try: batch = self.get(API, params={ "per_page": "100", "page": str(page), "_fields": "id,link,title,portfolio_category", }).json() except Exception: # HTTP 400 = fin de pagination WP break if not isinstance(batch, list) or not batch: break items.extend(batch) return items def fetch(self) -> list[Listing]: listings: list[Listing] = [] count = 0 for it in self._portfolio_items(): try: cats = set(it.get("portfolio_category") or []) unit_type = next((t for c, t in CAT_TYPES_FR.items() if c in cats), "") if not unit_type: continue # item EN ou hors taxonomie des types # candidat = marqué disponible, ou pas explicitement loué if CAT_DISPO not in cats and CAT_LOUE in cats: continue if count >= self.max_candidates: break count += 1 lst = self._parse_unit(it, unit_type) if lst is not None: listings.append(lst) except Exception: continue return listings # -- fiche de l'unité : la mention « LOUÉ » fait foi ------------------------- def _parse_unit(self, it: dict, unit_type: str) -> Listing | None: url = it.get("link") or "" title = re.sub(r"<[^>]+>", "", (it.get("title") or {}).get("rendered") or "").strip() if not url: return None html = self.get(url).text soup = BeautifulSoup(html, "html.parser") txt = soup.get_text(" | ", strip=True) if _LOUE_RE.search(txt): return None # loué malgré la taxonomie price = None price_label = "" pm = _PRICE_RE.search(txt) if pm: digits = re.sub(r"[^\d]", "", pm.group(1)) if digits: price = float(digits) price_label = f"À partir de {int(price)} $/mois" if price is not None and not (100 <= price <= 20000): price = None bathrooms = None bm = _BATH_RE.search(txt) if bm: try: bathrooms = float(bm.group(1)) except ValueError: bathrooms = None details: dict = {} fm = _FLOOR_RE.search(txt) if fm: details["floor"] = fm.group(1) sm = _SIZE_RE.search(txt) if sm: # ex. « 3 ½ » (confirme le type) unit_type = sm.group(1).replace(" ", "").replace("1/2", "½") # détails / équipements (listes de la fiche) amenities: list[str] = [] for li in soup.select(".fusion-text li, .post-content li"): t = li.get_text(" ", strip=True) if t and len(t) < 80 and t not in amenities: amenities.append(t) amenities = amenities[:25] images: list[str] = [] for u in _IMG_RE.findall(html): if _SKIP_IMG.search(u): continue if u not in images: images.append(u) images = images[: self.max_images] return Listing( source=self.source_id, external_id=str(it.get("id")), url=url, title=title or f"CÉLESTE — {unit_type}", address=ADDRESS, sector="LaSalle", city="Montréal", unit_type=unit_type, bathrooms=bathrooms, price=price, price_label=price_label, availability="Disponible", description="", amenities=amenities, details=details, images=images, )