# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/dma_locago.py : connecteur DMA / Locago (Douville, Moffet & Associés) # locago.ca présente les complexes; les unités (avec prix/disponibilité) sont # embarquées dans les pages MapSVG de v1.dma.immo (JSON `data_db.objects` : # superficie, étage, orientation, mois/année de disponibilité, prix). # La Tour Frontenac a son propre sélecteur JetEngine sur dma.immo # (/frontenac-selecteur-pub/ : cartes avec prix, étage, pi², orientation). # Une annonce par unité disponible; une annonce « projet » pour les complexes # sans sélecteur d'unités (IDOLA, Le Pivot). # Luxo Place (Ottawa) est exclu — hors région Québec/Lévis. # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from bs4 import BeautifulSoup from ..schema import Listing, infer_city, normalize_unit_type, parse_price from .base import BaseConnector V1 = "https://www.v1.dma.immo" DMA = "https://dma.immo" LOCAGO = "https://locago.ca" FRONTENAC_URL = f"{DMA}/frontenac-selecteur-pub/" FRONTENAC_ADDRESS = "1175, avenue de Germain-des-Prés, Québec" _MONTHS = { "01": "janvier", "02": "février", "03": "mars", "04": "avril", "05": "mai", "06": "juin", "07": "juillet", "08": "août", "09": "septembre", "10": "octobre", "11": "novembre", "12": "décembre", } # Complexes avec sélecteur d'unités MapSVG sur v1.dma.immo # (slug, nom, secteur, mot-clé image locago, site vitrine pour photos, adresse) # L'adresse n'est renseignée que si le site du projet publie une adresse # civique unique et sans ambiguïté (complexes multi-immeubles -> vide). _UNIT_PAGES = [ ("le-wow-unites", "Le WOW", "Sainte-Foy", "WOW", "https://lewow.ca", "930, rue Samuel-King, Québec"), ("la-suite-unites", "La Suite", "Sainte-Foy", "SUITE", "https://lasuite.ca", ""), ("domaine-des-meandres-unites", "Domaine des Méandres", "Lebourgneuf", "DDM", "https://info.domainedesmeandres.com", ""), ("vc-unites", "Villas Cortina", "Charlesbourg", "CORTINA", "https://info.villascortina.com", ""), ("le-divin-unites", "Le Divin", "Beauport", "DIVIN", "https://ledivin.ca", ""), ("le-divin-2-unites", "Le Divin (phase 2)", "Beauport", "DIVIN", "https://ledivin.ca", ""), ("ar-unites-01-02", "L'Aristocrate", "Lebourgneuf", "ARISTOCRATE", "https://info.laristocrate.ca", ""), ("ar-unites-03", "L'Aristocrate", "Lebourgneuf", "ARISTOCRATE", "https://info.laristocrate.ca", ""), ("ar-unites-04", "L'Aristocrate", "Lebourgneuf", "ARISTOCRATE", "https://info.laristocrate.ca", ""), ] # Complexes présentés sur locago.ca sans sélecteur d'unités public # (id, nom, secteur, mot-clé image locago, url projet, adresse) _PROJECT_ONLY = [ ("idola", "IDOLA", "Sainte-Foy", "IDOLA", "https://idola.ca/", "2689, boulevard Hochelaga, Québec"), ("le-pivot", "Le Pivot", "Vanier", "PIVOT", "https://pivotqc.com/", "1695, rue Monseigneur-Plessis, Québec"), ] _IMG_RE = re.compile( r'https?://[^"\'\s\)]+/wp-content/uploads/[^"\'\s\)]+\.(?:jpg|jpeg|png|webp)', re.I) _SKIP_IMG = re.compile(r"logo|icon|favicon|dma-|-\d{2,3}x\d{2,3}\.", re.I) class DMALocagoConnector(BaseConnector): source_id = "dma_locago" request_delay = 0.6 max_site_images = 12 # -- helpers --------------------------------------------------------------- def _page_images(self, url: str) -> list[str]: """Photos (wp-content/uploads) d'une page, sans logos ni vignettes.""" try: html = self.get(url).text except Exception: return [] imgs = [u for u in dict.fromkeys(_IMG_RE.findall(html)) if not _SKIP_IMG.search(u)] return imgs[: self.max_site_images] @staticmethod def _mapsvg_objects(html: str) -> list[dict]: """Extrait les unités des blocs `mapsvg_options = {...};` (data_db).""" objects: list[dict] = [] dec = json.JSONDecoder() for m in re.finditer(r"mapsvg_options\s*=\s*", html): try: opts, _ = dec.raw_decode(html[m.end():]) except Exception: continue data_db = opts.get("data_db") or {} if isinstance(data_db, dict): objects.extend(data_db.get("objects") or []) return objects @staticmethod def _unit_type(obj: dict) -> str: """'3.5' -> '3½' (avec repli sur typlog/descript).""" nb = str(obj.get("nbpiece") or "").strip().replace(",", ".") m = re.match(r"^(\d+)(\.5)?$", nb) if m: return f"{m.group(1)}½" raw = obj.get("descript") or obj.get("typlog") or "" if re.search(r"studio", str(raw), re.I): return "Studio" return normalize_unit_type(str(raw)) @staticmethod def _availability(obj: dict) -> str: # Le mois/année de disponibilité est renseigné pour toutes les unités # offertes (statuts « 1 » et « 2 ») ; il prime sur le statut brut. month = _MONTHS.get(str(obj.get("available_month") or ""), "") year = str(obj.get("available_year") or "").strip() if month and year: return f"{month} {year}" if obj.get("available") == "1": return "Libre immédiatement" return "Disponible prochainement" # -- fetch ----------------------------------------------------------------- def fetch(self) -> list[Listing]: # 1) Images « héro » des complexes depuis locago.ca img_keys = {row[3] for row in _UNIT_PAGES + _PROJECT_ONLY} locago_imgs: dict[str, str] = {} try: home = self.get(LOCAGO).text for u in _IMG_RE.findall(home): if _SKIP_IMG.search(u): continue for key in img_keys: if key.lower() in u.lower() and key not in locago_imgs: locago_imgs[key] = u except Exception: pass # Photos supplémentaires par site vitrine (1 requête par site) site_imgs: dict[str, list[str]] = {} listings: dict[str, Listing] = {} # 2) Unités disponibles des pages MapSVG (v1.dma.immo) for slug, name, sector, img_key, site, address in _UNIT_PAGES: try: html = self.get(f"{V1}/{slug}/").text except Exception: continue if site not in site_imgs: site_imgs[site] = self._page_images(site) images = ([locago_imgs[img_key]] if img_key in locago_imgs else []) images += [u for u in site_imgs[site] if u not in images] for obj in self._mapsvg_objects(html): if str(obj.get("available")) not in ("1", "2"): continue # unité louée uid = str(obj.get("title") or "").strip() unite = str(obj.get("unite") or "").strip() if not uid or uid in listings: continue # superficie structurée (pi²) -> area_sqft explicite sup = str(obj.get("superficie") or "").strip() area_sqft = None m = re.match(r"^(\d{2,5}(?:[.,]\d+)?)$", sup) if m: area_sqft = float(m.group(1).replace(",", ".")) desc_parts = [str(obj.get("descript") or "").strip()] if sup: desc_parts.append(f"{sup} pi²") elif obj.get("sup_range"): desc_parts.append(str(obj["sup_range"])) if obj.get("orientation"): desc_parts.append(f"Orientation : {obj['orientation']}") # étage structuré (« 0 » = non renseigné chez la source) details: dict = {} etage = str(obj.get("etage") or "").strip() if etage.isdigit() and int(etage) >= 1: details["floor"] = int(etage) startfrom = str(obj.get("startfrom") or "").strip() listings[uid] = Listing( source=self.source_id, external_id=uid, url=f"{V1}/{slug}/#unite-{unite or uid}", title=f"{name} — Unité {unite or uid}", address=address, sector=sector, city=infer_city(sector), unit_type=self._unit_type(obj), price=parse_price(startfrom), price_label=f"À partir de {startfrom}" if startfrom else "", availability=self._availability(obj), area_sqft=area_sqft, description=" — ".join(p for p in desc_parts if p)[:600], details=details, images=images, ) # 3) Tour Frontenac : sélecteur JetEngine sur dma.immo (cartes d'unités) try: self._fetch_frontenac(listings, locago_imgs) except Exception: pass # 4) Annonces « projet » pour les complexes sans sélecteur d'unités for pid, name, sector, img_key, url, address in _PROJECT_ONLY: try: images = ([locago_imgs[img_key]] if img_key in locago_imgs else []) images += [u for u in self._page_images(url) if u not in images] listings[f"projet-{pid}"] = Listing( source=self.source_id, external_id=f"projet-{pid}", url=url, title=name, address=address, sector=sector, city=infer_city(sector), unit_type="", price=None, price_label="", availability="", description=(f"Complexe locatif {name} ({sector}) géré par " "DMA / Locago — voir le site du projet pour " "les unités disponibles."), images=images, ) except Exception: continue return list(listings.values()) # -- Tour Frontenac (sélecteur JetEngine sur dma.immo) ---------------------- def _fetch_frontenac(self, listings: dict[str, Listing], locago_imgs: dict[str, str]) -> None: """Cartes d'unités disponibles : no, type, étage, pi², orientation, prix « À partir de » — pas de date de disponibilité affichée.""" html = self.get(FRONTENAC_URL).text soup = BeautifulSoup(html, "html.parser") images = ([locago_imgs["frontenac"]] if "frontenac" in locago_imgs else []) images += [u for u in self._page_images("https://appartsfrontenac.ca/") if u not in images] for card in soup.select(".jet-listing-grid__item"): text = card.get_text(" ", strip=True) m = re.search(r"Unité\s+(\w+)", text) if not m: continue unit = m.group(1) uid = f"FRONTENAC-{unit}" if uid in listings: continue type_el = card.select_one("span.subtitle") type_raw = type_el.get_text(" ", strip=True) if type_el else "" # types en toutes lettres chez la source (« Une chambre »…) key = type_raw.strip().lower() unit_type = { "studio": "Studio", "une chambre": "3½", "deux chambres": "4½", "trois chambres": "5½", }.get(key) or normalize_unit_type(type_raw) details: dict = {} fm = re.search(r"[ÉE]tage\s+(\d{1,2})\b", text) if fm: details["floor"] = int(fm.group(1)) area_sqft = None am = re.search(r"(\d[\d\s]{1,6})\s*pi", text) if am: area_sqft = float(am.group(1).replace(" ", "")) desc_parts = [type_raw] if am: desc_parts.append(f"{am.group(1).strip()} pi²") om = re.search(r"Orientation\s*:\s*([A-Za-zÀ-ÿ -]+?)(?:\s{2,}|$|T[ÉE]L)", text) if om: desc_parts.append(f"Orientation : {om.group(1).strip()}") pm = re.search(r"\$\s*([\d\s]{3,10})", text) price_label = f"À partir de {pm.group(1).strip()} $" if pm else "" listings[uid] = Listing( source=self.source_id, external_id=uid, url=FRONTENAC_URL, title=f"Tour Frontenac — Unité {unit}", address=FRONTENAC_ADDRESS, sector="Sainte-Foy", city=infer_city("Sainte-Foy"), unit_type=unit_type, price=parse_price(price_label), price_label=price_label, availability="", area_sqft=area_sqft, description=" — ".join(p for p in desc_parts if p)[:600], details=details, images=images, )