# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/groupe_dallaire.py : connecteur Groupe Dallaire — Faubourg du # Moulin (faubourgdumoulin.ca). Complexes Alizé, Noroît (I-II), Sirocco, # Sonora, Zéphyr (I-II) — rue Lionel-Audet, Québec (Beauport). # La page /disponibilite charge les unités via JetEngine (admin-ajax.php) : # on rejoue l'appel `get_listing` (lazy-load) puis la pagination # `jet_smart_filters` pour récupérer toutes les unités disponibles # (numéro, étage, prix, date de disponibilité, plan). Les fiches d'unité # (via cache BD) ajoutent type d'unité, superficie nette (« Pieds carrés ») # et inclusions réelles (« Incluant »). # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import html as htmllib import json import re import time from bs4 import BeautifulSoup from ..schema import Listing, normalize_unit_type, parse_price from .base import BaseConnector BASE = "https://faubourgdumoulin.ca" AJAX = f"{BASE}/wp-admin/admin-ajax.php" DISPO_URL = f"{BASE}/disponibilite/" SECTOR = "Beauport" CITY = "Québec" ADDRESS = "rue Lionel-Audet, Québec (Faubourg du Moulin)" COMPLEXES = { "alize": "Alizé", "alize-2": "Alizé (Phase II)", "noroit": "Noroît", "noroit-2": "Noroît II", "sirocco": "Sirocco", "sonora": "Sonora", "zephyr": "Zéphyr", "zephyr-2": "Zéphyr II", } # paramètres du widget « dispo » de la page /disponibilite PAGE_SETTINGS = { "page_settings[post_id]": "105", "page_settings[queried_id]": "105|WP_Post", "page_settings[element_id]": "cb310a3", "page_settings[page]": "1", "listing_type": "elementor", "isEditMode": "false", } def _flatten(prefix: str, obj, out: dict) -> None: """Aplati un dict/list en champs de formulaire PHP (a[b][0]=x).""" if isinstance(obj, dict): for k, v in obj.items(): _flatten(f"{prefix}[{k}]", v, out) elif isinstance(obj, list): for i, v in enumerate(obj): _flatten(f"{prefix}[{i}]", v, out) elif isinstance(obj, bool): out[prefix] = "true" if obj else "false" else: out[prefix] = "" if obj is None else str(obj) class GroupeDallaireConnector(BaseConnector): source_id = "groupe_dallaire" request_delay = 0.6 max_pages = 25 # garde-fou pagination max_details = 150 # garde-fou fiches d'unité (vraies requêtes) # -- POST throttlé --------------------------------------------------- def _post(self, url: str, data: dict): wait = self.request_delay - (time.time() - self._last_request) if wait > 0: time.sleep(wait) resp = self.session.post(url, data=data, timeout=self.timeout) self._last_request = time.time() resp.raise_for_status() return resp # --------------------------------------------------------------------- def fetch(self) -> list[Listing]: # 1) Premier appel lazy-load : items page 1 + query/widget_settings first = self._post(AJAX, { "action": "jet_engine_ajax", "handler": "get_listing", **PAGE_SETTINGS, }).json() html1 = (first.get("data") or {}).get("html", "") nav = None m = re.search(r'data-nav="([^"]*)"', html1) if m: try: nav = json.loads(htmllib.unescape(m.group(1))) except Exception: nav = None listings: dict[str, Listing] = {} for lst in self._parse_items(html1): listings.setdefault(lst.external_id, lst) # 2) Pagination via jet_smart_filters (si le data-nav est disponible) if nav and nav.get("query") and nav.get("widget_settings"): page = 2 while page <= self.max_pages: data = { "action": "jet_smart_filters", "provider": "jet-engine/dispo", "paged": str(page), } _flatten("defaults", nav["query"], data) _flatten("settings", nav["widget_settings"], data) try: resp = self._post(AJAX, data).json() except Exception: break content = resp.get("content") or "" new = 0 for lst in self._parse_items(content): if lst.external_id not in listings: listings[lst.external_id] = lst new += 1 pag = resp.get("pagination") or {} max_pages = int(pag.get("max_num_pages") or 0) if new == 0 or (max_pages and page >= max_pages): break page += 1 # 3) Fiche de l'unité (cache BD) : type d'unité (« 3½ TYPE D'UNITÉ »), # superficie nette (« Pieds carrés »), inclusions réelles # (« Incluant ») et plans (unité + niveau) self._fetched = 0 for lst in listings.values(): card_key = hashlib.sha1( f"{lst.title}|{lst.price_label}|{lst.availability}" .encode("utf-8")).hexdigest() try: payload = self.detail(lst.external_id, card_key, lambda u=lst.url: self._fetch_detail(u)) except Exception: continue self._apply_detail(lst, payload) return list(listings.values()) # --------------------------------------------------------------------- def _fetch_detail(self, url: str) -> dict: """Fiche d'unité : type, superficie nette, inclusions, plans. Retour JSON-sérialisable (cache BD).""" if self._fetched >= self.max_details: raise RuntimeError("budget de fiches d'unité atteint") self._fetched += 1 dhtml = self.get(url).text out: dict = {} dtext = re.sub(r"", "", dhtml, flags=re.S) dtext = re.sub(r"", "", dtext, flags=re.S) dtext = re.sub(r"<[^>]+>", "\n", dtext) tm = re.search(r"(\d\s*½\s*\+?|Studio)\s*\n+\s*TYPE D['’]UNITÉ", dtext) if tm: out["unit_type"] = tm.group(1) sm = re.search(r"(\d{3,4})\s*\n+\s*Pieds carrés", dtext) if sm: out["area_sqft"] = float(sm.group(1)) # champ « Pieds carrés » # Inclusions réelles (« Incluant » -> répéteur JetEngine) dsoup = BeautifulSoup(dhtml, "html.parser") amen = [sp.get_text(" ", strip=True) for sp in dsoup.select(".jet-listing-dynamic-repeater__item span")] out["amenities"] = [a for a in dict.fromkeys(amen) if a][:15] # Plans : plan de l'unité (jpg) + plan du niveau (NIVEAU_x.png) plans = [] for u in dict.fromkeys(re.findall( r"https://faubourgdumoulin\.ca/wp-content/uploads/" r"[^\"'\s\\]+\.(?:jpe?g|webp|png)", dhtml, re.I)): if re.search(r"logo|_rgb_|descripteur|favicon|cropped-|dalcon|" r"alpha-|-\d+x\d+\.", u, re.I): continue if u.lower().endswith(".png") and "niveau" not in u.lower(): continue plans.append(u) out["images"] = plans[:10] return out def _apply_detail(self, lst: Listing, d: dict) -> None: """Reporte le payload (frais/cache) sur l'annonce.""" if not d: return if d.get("unit_type") and not lst.unit_type: lst.unit_type = normalize_unit_type(d["unit_type"]) if lst.unit_type and "½" not in lst.title \ and "Studio" not in lst.title: lst.title += f" ({lst.unit_type})" if d.get("area_sqft"): lst.area_sqft = d["area_sqft"] if d.get("amenities"): # inclusions réelles de la fiche (remplacent le repli générique) extra = [a for a in lst.amenities if a == "En promotion"] lst.amenities = d["amenities"] + extra if d.get("images"): lst.images = list(dict.fromkeys(lst.images + d["images"]))[:15] # --------------------------------------------------------------------- def _parse_items(self, content: str) -> list[Listing]: out: list[Listing] = [] if not content: return out soup = BeautifulSoup(content, "html.parser") for item in soup.select(".jet-listing-grid__item[data-post-id]"): try: a = item.select_one('a[href*="/unites/"]') if not a: continue url = a["href"] slug = url.rstrip("/").split("/")[-1] for st in item.select("style"): st.decompose() text = item.get_text(" ", strip=True) # Complexe depuis le slug (ex. sonora-101, noroit-2-151) cslug = re.match(r"([a-z]+(?:-2)?)-\d+", slug) complexe = COMPLEXES.get(cslug.group(1) if cslug else "", "") if not complexe and cslug: complexe = cslug.group(1).replace("-", " ").title() num_m = re.search(r"Unité\s+(\w+)\s*-?", text) unit_no = num_m.group(1) if num_m else slug.split("-")[-1] floor_m = re.search(r"Étage\s+(\d+)", text) price = None price_label = "" pm = re.search(r"(\d{3,5})\s*\$\s*(/mois\*?)", text) if pm: price = parse_price(pm.group(1) + "$") price_label = (pm.group(1) + "$" + pm.group(2)) if "*" in pm.group(2): price_label += (" (chauffé, éclairé, climatisé, " "stationnement int.)") avail_m = re.search(r"Disponibilité\s*:\s*([^|]+?)(?:Unité|$)", text) availability = (f"Disponibilité : {avail_m.group(1).strip()}" if avail_m else "Disponible") promo = "En promotion" in text # Images : logo du complexe exclu, plan de l'unité conservé images: list[str] = [] unit_type = "" for img in item.select("img[src]"): src = img["src"] if not src.startswith("http"): continue if re.search(r"logo|_rgb_|descripteur|\.svg$", src, re.I): continue full = re.sub(r"-\d+x\d+(\.(?:jpg|jpeg|png|webp))$", r"\1", src) images.append(full) # type d'unité encodé dans le nom du plan tm = (re.search(r"pieces[-_](\d)\.5", src, re.I) or re.search(r"_(\d)\.5\b", src)) if tm and not unit_type: unit_type = normalize_unit_type(f"{tm.group(1)} 1/2") images = list(dict.fromkeys(images)) desc_parts = [f"Complexe {complexe}" if complexe else ""] if floor_m: desc_parts.append(f"Étage {floor_m.group(1)}") if promo: desc_parts.append("En promotion") desc_parts = [d for d in desc_parts if d] # repli générique (remplacé par la liste « Incluant » réelle # de la fiche d'unité quand elle est disponible) amenities = ["Eau chaude incluse", "Internet haute vitesse inclus", "Stationnement extérieur ou intérieur", "Accès aux aires communes"] if promo: amenities.append("En promotion") details: dict = {} if floor_m: details["floor"] = int(floor_m.group(1)) out.append(Listing( source=self.source_id, external_id=slug, url=url, title=f"Faubourg du Moulin ({complexe or 'complexe'}) — " f"Unité {unit_no}" f"{' (' + unit_type + ')' if unit_type else ''}", address=ADDRESS, sector=SECTOR, city=CITY, unit_type=unit_type, price=price, price_label=price_label, availability=availability, description=" | ".join(desc_parts), amenities=amenities, details=details, images=images, )) except Exception: continue return out