# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/beaudoin.py : connecteur Société Beaudoin immobilier # (beaudoinimmobilier.ca — Longueuil, Boucherville, Montréal : # Anjou, Lachine, Saint-Léonard). Thème WordPress Houzez : les logements # sont des posts « property » exposés par l'API REST # (/wp-json/wp/v2/properties) avec prix, adresse géocodée, lat/lng, # taxonomies (type d'unité, ville, disponibilité, commodités) et # galerie d'images (IDs de médias résolus via /wp-json/wp/v2/media). # Le site couvre d'autres régions : seul le Grand Montréal est conservé. # ----------------------------------------------------------------------------- from __future__ import annotations import html as htmllib import re from ..schema import Listing, normalize_unit_type, strip_accents from .base import BaseConnector BASE = "https://www.beaudoinimmobilier.ca" API = f"{BASE}/index.php/wp-json/wp/v2" # Villes admissibles (Grand Montréal) — comparées sans accents, en minuscules. _ALLOWED_CITIES = ( "montreal", "longueuil", "boucherville", "brossard", "saint-lambert", "laval", "lachine", "anjou", "saint-leonard", "saint-hubert", ) def _clean(txt: str) -> str: txt = re.sub(r"<[^>]+>", " ", txt or "") return re.sub(r"\s+", " ", htmllib.unescape(txt)).strip() def _additional_features(serialized: str) -> list[str]: """Champ Houzez « additional_features » (PHP sérialisé) -> libellés. Paires titre/valeur (« Eau chaude » / « incluse ») -> « Eau chaude incluse ». """ toks = re.findall(r's:\d+:"((?:[^"\\]|\\.)*)"', serialized or "") out: list[str] = [] i = 0 while True: try: ti = toks.index("fave_additional_feature_title", i) except ValueError: break title = toks[ti + 1].strip() if ti + 1 < len(toks) else "" value = "" if ti + 3 < len(toks) and toks[ti + 2] == "fave_additional_feature_value": value = toks[ti + 3].strip() label = f"{title} {value}".strip() if title and label not in out: out.append(label) i = ti + 2 return out class BeaudoinConnector(BaseConnector): source_id = "beaudoin" request_delay = 0.5 max_pages = 5 # garde-fou (5 x 100 propriétés) max_media_ids = 30 # images max par annonce # -- helpers REST ---------------------------------------------------------- def _terms(self, taxonomy: str) -> dict[int, str]: try: data = self.get(f"{API}/{taxonomy}?per_page=100").json() return {t["id"]: htmllib.unescape(t["name"]) for t in data} except Exception: return {} def _media_urls(self, ids: list[str]) -> list[str]: ids = [i for i in ids if str(i).isdigit()][: self.max_media_ids] if not ids: return [] try: url = (f"{API}/media?include={','.join(map(str, ids))}" f"&per_page=100&_fields=id,source_url") data = self.get(url).json() by_id = {str(m["id"]): m.get("source_url", "") for m in data} return [by_id[str(i)] for i in ids if by_id.get(str(i))] except Exception: return [] @staticmethod def _split_city(name: str) -> tuple[str, str]: """'Montréal Anjou' -> ('Montréal', 'Anjou') ; 'Longueuil' -> (ville, '').""" name = name.strip() if strip_accents(name.lower()).startswith("montreal"): sector = name[len("Montréal"):].strip(" -–") return "Montréal", sector return name, "" @staticmethod def _city_allowed(name: str) -> bool: key = strip_accents(name.lower()).replace(" ", "-") return any(tok in key for tok in _ALLOWED_CITIES) # -- fetch ------------------------------------------------------------------ def fetch(self) -> list[Listing]: types = self._terms("property_type") statuses = self._terms("property_status") cities = self._terms("property_city") features = self._terms("property_feature") props: list[dict] = [] for page in range(1, self.max_pages + 1): try: batch = self.get( f"{API}/properties?per_page=100&page={page}").json() except Exception: break if not isinstance(batch, list) or not batch: break props.extend(batch) if len(batch) < 100: break listings: list[Listing] = [] for p in props: try: lst = self._parse_property(p, types, statuses, cities, features) except Exception: continue if lst: listings.append(lst) return listings def _parse_property(self, p: dict, types: dict, statuses: dict, cities: dict, features: dict) -> Listing | None: meta = p.get("property_meta") or {} def m1(key: str) -> str: v = meta.get(key) or [] return str(v[0]).strip() if v and v[0] is not None else "" # Ville / secteur (taxonomie property_city, ex. « Montréal Anjou ») city_name = next((cities[i] for i in (p.get("property_city") or []) if i in cities), "") if not city_name or not self._city_allowed(city_name): return None # hors Grand Montréal city, sector = self._split_city(city_name) title = _clean((p.get("title") or {}).get("rendered") or "") # Adresse géocodée : « 7340, Avenue Guy, Anjou, Montréal, ... Canada » map_addr = m1("fave_property_map_address") address = ", ".join(s.strip() for s in map_addr.split(",")[:2]) if map_addr else "" address = re.sub(r"\s+", " ", address).strip() # Type d'unité (taxonomie « 3 1/2 », « 4 1/2 Penthouse », ...) type_name = next((types[i] for i in (p.get("property_type") or []) if i in types), "") unit_type = normalize_unit_type(type_name) # Prix mensuel (champ Houzez) price = None price_label = "" raw_price = m1("fave_property_price") if raw_price: try: price = float(re.sub(r"[^\d.]", "", raw_price)) except ValueError: price = None postfix = m1("fave_property_price_postfix") or "mois" price_label = f"{raw_price}$ / {postfix}" if price is not None and not (100 <= price <= 20000): price = None # Disponibilité (taxonomie property_status, « 08 - Disponible pour août ») avail = next((statuses[i] for i in (p.get("property_status") or []) if i in statuses), "") availability = re.sub(r"^\d+\s*-\s*", "", avail) amenities = [features[i] for i in (p.get("property_feature") or []) if i in features] # « Caractéristiques additionnelles » Houzez (PHP sérialisé) : # Eau chaude incluse, Lave-vaisselle inclus, Chiens et chats acceptés… for label in _additional_features(m1("additional_features")): if label not in amenities: amenities.append(label) # Superficie structurée (fave_property_size, préfixe « Pieds carrés ») area_sqft = None raw_size = m1("fave_property_size") size_prefix = m1("fave_property_size_prefix").lower() if raw_size: try: val = float(raw_size.replace(",", ".")) except ValueError: val = None if val and 80 <= val <= 20000 and ( not size_prefix or "pied" in size_prefix or "pi" in size_prefix or "sq" in size_prefix): area_sqft = val # Chambres / salles de bain (champs Houzez) -> commodités affichables beds = m1("fave_property_bedrooms") baths = m1("fave_property_bathrooms") if beds.isdigit() and int(beds) > 0: amenities.append(f"{beds} chambre{'s' if int(beds) > 1 else ''}") if baths.isdigit() and int(baths) > 0: amenities.append( f"{baths} salle{'s' if int(baths) > 1 else ''} de bain") # Coordonnées lat = lng = None try: lat = float(m1("houzez_geolocation_lat")) lng = float(m1("houzez_geolocation_long")) except (ValueError, TypeError): lat = lng = None # Galerie d'images (IDs de médias -> URLs) images = self._media_urls(meta.get("fave_property_images") or []) if not images: thumb = m1("_thumbnail_id") if thumb: images = self._media_urls([thumb]) description = _clean((p.get("content") or {}).get("rendered") or "")[:600] return Listing( source=self.source_id, external_id=str(p.get("id")), url=p.get("link") or "", title=title or f"Logement {p.get('id')}", address=address, sector=sector, city=city, unit_type=unit_type, price=price, price_label=price_label, availability=availability, area_sqft=area_sqft, description=description, amenities=amenities, images=images, lat=lat, lng=lng, )