# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/remax_quebec.py : RE/MAX Québec (remax-quebec.com) — LOCATIONS # Le site interroge un index Meilisearch public (search-only key exposée dans # la config de la page). L'index « inscriptions » couvre toute la province et # mélange ventes et locations : on ne garde QUE les slugs « -a-louer » / # « -for-rent » (et on écarte les Loué/Vendu que l'index conserve). # Meilisearch plafonne à 1000 hits par requête (maxTotalHits) : on shard donc # par région de tri d'acheminement postale (FSA, 3 premiers caractères) et on # dédoublonne par numéro d'inscription. Adapté du connecteur « à vendre » # d'Immo-Ka (agent-courtage/immoka). # ----------------------------------------------------------------------------- from __future__ import annotations import html as _html import os import re from bs4 import BeautifulSoup from ..normalize import parse_area_sqft, parse_price from ..schema import Listing from .base import BaseConnector from . import _detailutil as du SEARCH_URL = "https://search.remax-quebec.com/indexes/inscriptions/search" # Clé de recherche (search-only) exposée publiquement dans la config du site. SEARCH_KEY = "b0b93998ab78573e8b937b528ad37d2ce3fbc97e07a9f2c909c4220db910c152" SITE = "https://www.remax-quebec.com" # Territoires postaux du Québec : préfixes G, H et J. FSA_LETTERS = ("G", "H", "J") # annonces conclues que l'index conserve encore quelque temps GONE_TOKENS = ("loué", "loue", "rented", "vendu", "sold") # Enrichissement page détail (photos + description + specs) : plafonné par # exécution pour garder chaque sync borné. Les fiches en cache sont réutilisées # gratis ; le parc se complète sur plusieurs cycles. DETAIL_LIMIT = int(os.environ.get("LOUKA_REMAX_DETAIL_LIMIT", "400")) # Réparation ciblée : re-visite des fiches déjà en cache SANS description # (payload d'avant le repli « Addenda », ou échec réseau passé) — le # marqueur desc_checked évite de re-visiter les fiches dont l'absence de # description est confirmée à la source. REPAIR_LIMIT = int(os.environ.get("LOUKA_REMAX_REPAIR_LIMIT", "400")) # slug « condo-a-louer-montreal/... » -> type affichable. Les appartements / # condos / plex restent sans unit_type figé : la fiche détail fournit le nombre # de chambres (converti en n+2½ par la normalisation lou-ka). _TYPE_SLUG = { "maison": "Maison", "house": "Maison", "chalet": "Chalet", "cottage": "Chalet", "loft": "Loft", "condo": "Condo", "appartement": "Appartement", "apartment": "Appartement", "plex": "Multiplex", "duplex": "Duplex", "triplex": "Triplex", } # ce qui n'est pas un logement (lou-ka = résidentiel locatif uniquement) _EXCLUDE_SLUG = ("commercial", "commerciale", "terrain", "land", "industriel", "bureau", "ferme", "fermette") _RENT_SLUG_RE = re.compile(r"-(?:a-louer|for-rent)\b") class RemaxQuebecConnector(BaseConnector): source_id = "remax_quebec" request_delay = 0.4 # partagé API de recherche / pages détail def fetch(self) -> list[Listing]: by_id: dict[int, dict] = {} for fsa in self._fsa_candidates(): for h in self._search(fsa): nid = h.get("no_inscription") if nid is not None: by_id[nid] = h # dédoublonnage inter-shards listings = [] for h in by_id.values(): lst = self._to_listing(h) if lst is not None: listings.append(lst) du.enrich(self, listings, DETAIL_LIMIT, parse_remax_detail, key="v1") self._repair_missing_desc(listings) for lst in listings: _unit_from_bedrooms(lst) return listings def _repair_missing_desc(self, listings: list[Listing]) -> None: """Re-visite les fiches en cache restées sans description : payloads d'avant le repli « Addenda » du parseur, ou payloads vides (échec réseau passé). Le nouveau parseur pose desc_checked=True, donc une fiche réellement sans addenda n'est re-visitée qu'une seule fois. Le reste du cache (fiches déjà riches) n'est pas invalidé.""" from .. import db budget = REPAIR_LIMIT if budget <= 0: return con = db.connect() try: for lst in listings: if budget <= 0: break if lst.description: continue cached = db.get_cached_detail(con, self.source_id, lst.external_id, "v1") if cached is None: # jamais visitée : du.enrich s'en charge continue if cached.get("description") or cached.get("desc_checked"): continue budget -= 1 try: payload = parse_remax_detail(self.get(lst.url).text) except Exception: continue # erreur réseau : on réessaiera db.put_cached_detail(con, self.source_id, lst.external_id, "v1", payload) du.apply_detail(lst, payload) finally: con.close() # -- sharding -------------------------------------------------------------- @staticmethod def _fsa_candidates(): for letter in FSA_LETTERS: for digit in "0123456789": for last in "ABCDEFGHIJKLMNOPQRSTUVWXYZ": yield f"{letter}{digit}{last}" def _search(self, q: str) -> list[dict]: try: resp = self.post( SEARCH_URL, headers={"Authorization": f"Bearer {SEARCH_KEY}", "Content-Type": "application/json"}, json={"q": q, "limit": 1000}, ) except Exception: return [] data = resp.json() return data.get("hits", []) if isinstance(data, dict) else [] # -- mapping --------------------------------------------------------------- def _to_listing(self, h: dict) -> Listing | None: nid = h.get("no_inscription") if nid is None: return None slug = (h.get("slug") or {}).get("fr", "") or "" slug_en = (h.get("slug") or {}).get("en", "") or "" # UNIQUEMENT les locations (l'index mélange ventes et locations) ; # l'URL de la fiche exige le slug français if not slug or not (_RENT_SLUG_RE.search(slug) or _RENT_SLUG_RE.search(slug_en)): return None price_label = (h.get("display_price") or {}).get("fr", "") or "" # exclure les logements déjà loués / retirés (l'index les conserve) if any(tok in price_label.lower() for tok in GONE_TOKENS): return None prop_type, region = _from_slug(slug) if prop_type is None: # commercial/terrain : pas un logement return None url = f"{SITE}/fr/proprietes/{slug}" full_addr = (h.get("full_address") or {}).get("fr", "") or "" address, sector, city = _split_address(full_addr) details: dict = {"MLS": str(nid)} if prop_type: details["Type"] = prop_type if region: details["Région"] = region return Listing( source=self.source_id, external_id=str(nid), url=url, title=address or full_addr, address=address, sector=sector, city=city, # appartement/condo/plex : la fiche détail donne les chambres unit_type=prop_type if prop_type in ("Maison", "Chalet", "Loft") else "", price=parse_price(price_label), # « 2 690$ par mois » -> 2690.0 price_label=price_label, details=details, ) def _unit_from_bedrooms(lst: Listing) -> None: """Condo/appartement : « Chambres » de la fiche détail -> unit_type « n chambres » (la normalisation lou-ka convertit en n+2½).""" if lst.unit_type not in ("", "Condo", "Appartement"): return m = re.match(r"\d+", str(lst.details.get("Chambres", ""))) if not m: if not lst.unit_type: lst.unit_type = lst.details.get("Type", "") return n = int(m.group(0)) lst.unit_type = "Studio" if n == 0 else f"{n} chambres" # --------------------------------------------------------------------------- # Analyse de l'adresse et du slug # --------------------------------------------------------------------------- _PAREN_RE = re.compile(r"\(([^)]*)\)") _POSTAL_RE = re.compile(r"[GHJ]\d[A-Z]\s?\d[A-Z]\d", re.I) def _split_address(full: str) -> tuple[str, str, str]: """« 9561 Boul. Perras, Montréal (Rivière-des-Prairies/…) (RDP), H1E4C4 » -> (adresse, secteur, ville).""" if not full: return "", "", "" parts = [p.strip() for p in full.split(",")] # retirer le code postal final if parts and _POSTAL_RE.search(parts[-1]): parts = parts[:-1] address = parts[0] if parts else "" city = sector = "" if len(parts) >= 2: muni = parts[1] parens = _PAREN_RE.findall(muni) city = _PAREN_RE.sub("", muni).strip() if parens: sector = parens[-1].strip() # secteur supplémentaire dans les champs suivants (avant le postal) for extra in parts[2:]: e = _PAREN_RE.sub("", extra).strip() or extra.strip() if e and not sector: sector = e return address, sector, city def _from_slug(slug: str) -> tuple[str | None, str]: """slug « condo-a-louer-montreal/9561-boul-perras-…-20357732 » -> (type affichable, région). Type None = pas un logement (commercial…).""" if not slug: return "", "" head = slug.split("/", 1)[0] # condo-a-louer-montreal if any(re.search(rf"\b{x}", head) for x in _EXCLUDE_SLUG): return None, "" prop_type = "" for key, canon in _TYPE_SLUG.items(): if re.search(rf"\b{key}", head): prop_type = canon break # région = ce qui suit « -a-louer-/-for-rent- » m = re.search(r"(?:a-louer|for-rent)-(.+)$", head) region = m.group(1).replace("-", " ").title() if m else "" return prop_type, region # --------------------------------------------------------------------------- # Page détail : photos + description + caractéristiques + courtier # --------------------------------------------------------------------------- # Les photos apparaissent dans plusieurs buckets de taille (www_full, _medium, # _small…) selon le lazy-load ; on les capte toutes et on les normalise en # pleine résolution, dédoublonnées par nom de fichier. _IMG_RE = re.compile( r'https://media\.remax-quebec\.com/img/www_[a-z]+/[^"\'\\ ]+\.(?:jpg|jpeg|png|webp)', re.I) _IMG_SIZE_RE = re.compile(r"/www_[a-z]+/", re.I) _COORD_RE = re.compile(r"query=(-?\d+\.\d+)%2C\+?(-?\d+\.\d+)") # en tête de fiche, la valeur précède le libellé : « 2 | Chambres », « 1 | Salle de bain » _BED_RE = re.compile(r"(\d+)(?:\s*\([^)]*\))?\s*\|\s*Chambres?\b", re.I) _BATH_RE = re.compile(r"(\d+)\s*\|\s*Salles? de bain", re.I) _WATER_RE = re.compile(r"(\d+)\s*\|\s*Salles? d'eau", re.I) _QSTAT_RE = re.compile( r"quick-stat-text[^>]*>\s*]*>\s*(.*?)\s*\s*]*>\s*(.*?)\s*", re.S) _INCL_RE = re.compile( r"inclusions-exclusions-section(.*?)(?:|realtor-section|financial-section)", re.S) _ROOM_RE = re.compile(r"rooms-details-section__vertical-table[^>]*>(.*?)\s*", re.S) # Sur les fiches LOCATION, le tableau « Particularités » est en ordre # libellé | valeur (l'inverse des fiches vente) : on privilégie donc cet ordre # et du.centris_details ne sert que de repli. _RENT_LABELS = [ "Type de propriété", "Genre de propriété", "Style de bâtiment", "Année de construction", "Superficie habitable", "Superficie du terrain", "Nombre de pièces", "Nombre d'unités", "Stationnement (total)", "Garage", "Mode de chauffage", "Système de chauffage", "Énergie pour le chauffage", "Approvisionnement en eau", "Système d'égouts", "Piscine", "Déménagement", "Date d'emménagement", "Disponibilité", "Bail", "Durée du bail", "Meublé", "Animaux", "Inclus dans le loyer", "Cuisine", ] # une « valeur » qui est en fait un titre de section = extraction décalée _SECTION_VAL_RE = re.compile( r"^(?:Particularit[ée]s|Caract[ée]ristiques|Inclusions|Exclusions|Addenda)", re.I) def _labels_first(text: str, labels: list[str]) -> dict: """Extraction « libellé | valeur » (ordre des fiches location RE/MAX).""" out: dict = {} for label in labels: boundary = r"\b" if label[-1].isalnum() else "" m = re.search(re.escape(label) + boundary + r"\s*\|\s*([^|]{1,55})", text) if m: val = m.group(1).strip(" |") if (val and 1 <= len(val) <= 55 and val.lower() != label.lower() and val not in labels and not _SECTION_VAL_RE.match(val)): out[label] = val return out def _drop_section_values(details: dict) -> None: """Purge les valeurs qui sont des titres de section (repli valeur|libellé de du.centris_details décalé sur les fiches en ordre libellé|valeur).""" for k in [k for k, v in details.items() if isinstance(v, str) and _SECTION_VAL_RE.match(v)]: del details[k] def _ld_agents(html: str): """Nœuds RealEstateAgent (dans l'ordre du document), même imbriqués.""" for n in du.ld_nodes(html): queue = [n] while queue: cur = queue.pop(0) if isinstance(cur, dict): if cur.get("@type") == "RealEstateAgent" and cur.get("name"): yield cur queue.extend(cur.values()) elif isinstance(cur, list): queue.extend(cur) def _addenda_description(html: str) -> str: """Repli quand le JSON-LD n'a pas de description : le texte du courtier vit alors dans les accordéons « Description » / « Addenda » de la fiche.""" soup = BeautifulSoup(html, "html.parser") parts: list[str] = [] for cls in ("description-section", "addenda-section"): for node in soup.select(f"div.{cls} .accordion__content"): txt = _html.unescape(node.get_text("\n", strip=True)) txt = re.sub(r"[ \t]+", " ", txt) txt = re.sub(r"\n{3,}", "\n\n", txt).strip() if len(txt) >= 40 and txt not in parts: parts.append(txt) return "\n\n".join(parts) def parse_remax_detail(html: str) -> dict: """Fiche RE/MAX location : description JSON-LD (repli : accordéons Description/Addenda), galerie pleine résolution, GPS, caractéristiques Centris, pièces, inclusions, courtier/agence.""" # desc_checked : marqueur de cache — la présence de description a été # vérifiée AVEC le repli Addenda (évite les re-visites infinies quand la # fiche n'a réellement aucun texte) out: dict = {"desc_checked": True} details: dict = {} # description via JSON-LD RealEstateListing, sinon accordéon Addenda desc = du.ld_description(html) or _addenda_description(html) if desc: out["description"] = desc[:6000] # photos : toutes tailles -> pleine résolution, dédoublonnées par fichier seen, images = set(), [] for u in _IMG_RE.findall(html): full = _IMG_SIZE_RE.sub("/www_full/", u) fn = full.rsplit("/", 1)[-1] if fn not in seen and "nophoto" not in full: seen.add(fn) images.append(full) if images: out["images"] = images # coordonnées GPS (lien Google Maps) m = _COORD_RE.search(html) if m: lat, lng = float(m.group(1)), float(m.group(2)) if 44.5 <= lat <= 63.0 and -80.0 <= lng <= -56.0: out["lat"], out["lng"] = lat, lng # courtier + agence via JSON-LD RealEstateAgent (souvent imbriqué) for n in _ld_agents(html): details.setdefault("Courtier", str(n["name"]).strip().title()) if n.get("telephone"): details.setdefault("Téléphone", str(n["telephone"]).strip()) org = n.get("parentOrganization") or {} org_name = org.get("name", "") if isinstance(org, dict) else "" if org_name: agence = str(org_name).strip().title() \ .replace("Re/Max", "RE/MAX").replace("Inc.", "inc.") details.setdefault("Agence", agence) text = du.flatten(html) # chambres / salles de bain (en tête : « 2 | Chambres », « 1 | Salle de bain ») m = _BED_RE.search(text) if m: details["Chambres"] = m.group(1) m = _BATH_RE.search(text) if m: details["Salles de bain"] = m.group(1) m = _WATER_RE.search(text) if m: details["Salles d'eau"] = m.group(1) # caractéristiques : repli valeur|libellé (du), écrasé par libellé|valeur details.update(du.centris_details(text)) details.update(_labels_first(text, _RENT_LABELS)) _drop_section_values(details) # « Cuisine » happé depuis le tableau des pièces (dimensions/niveau) : retirer if re.search(r"\d\s*[xX]\s*\d|Niveau|^\d+$", str(details.get("Cuisine", ""))): details.pop("Cuisine", None) # quick stats (Nb de pièces, Superficie habitable…) for label, val in _QSTAT_RE.findall(html): label = _html.unescape(re.sub(r"\s+", " ", label)).strip() val = _html.unescape(re.sub(r"\s+", " ", val)).strip() if label and val and label not in details: details[label] = val # pièces avec dimensions/niveau/revêtement rooms = [] for blk in _ROOM_RE.findall(html): rt = _html.unescape(re.sub(r"<[^>]+>", " ", blk)) rt = re.sub(r"\s+", " ", rt).strip() name = re.split(r"Niveau\s*:", rt)[0].strip() niveau = re.search(r"Niveau\s*:\s*([^:]+?)(?:Dimensions|Revêtement|$)", rt) dim = re.search(r"Dimensions\s*:\s*([0-9\'\".,X x×]+)", rt) rev = re.search(r"Revêtement\s*:\s*([A-Za-zÀ-ÿ ,-]+?)(?:\s*Détails|\s*$)", rt) if name: rooms.append({ "nom": name[:40], "niveau": (niveau.group(1).strip() if niveau else ""), "dimensions": (dim.group(1).strip() if dim else ""), "revetement": (rev.group(1).strip() if rev else ""), }) if rooms: details["pieces"] = rooms[:20] # inclusions / exclusions -> commodités affichables mi = _INCL_RE.search(html) feats = [] if mi: blk = _html.unescape(re.sub(r"<[^>]+>", "\n", mi.group(1))) for line in blk.split("\n"): line = line.strip(" \t•-") if 3 <= len(line) <= 120 and not line.lower().startswith( ("inclusion", "exclusion")): feats.append(line) if feats: out["amenities"] = feats[:25] # superficie habitable (quick stat ou tableau) -> pi² if details.get("Superficie habitable"): out["area_sqft"] = parse_area_sqft(details["Superficie habitable"]) # date d'emménagement / déménagement -> disponibilité for k in ("Date d'emménagement", "Déménagement", "Disponibilité"): if details.get(k): out["availability"] = details[k] break # chambres -> type d'unité (les fiches liste condo/appartement sont vides) if details.get("Chambres"): try: n = int(details["Chambres"]) out["unit_type"] = "Studio" if n == 0 else f"{n} chambres" except ValueError: pass if details: out["details"] = details return out