# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (Québec + expansion Ontario) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/medallion.py : connecteur Medallion Corporation (medallioncorp.com) # Gestionnaire torontois (~25 000 unités est., Toronto/GTA + Brampton, Ajax, # Whitby, Oshawa, Richmond Hill, Hamilton, London). Site WordPress (thème # RealHomes/inspiry + CPT UI) SANS anti-bot : le CPT `property` est exposé # par l'API REST standard — `/wp-json/wp/v2/properties?per_page=100` # (84 propriétés au moment de l'écriture, x-wp-total 84, une seule page). # Chaque entrée porte dans `class_list` la ville (property-city-), le # type (property-type-residential/commercial…) et le statut (property-status- # for-rent vs commercial-rent) : on ne garde que le résidentiel à louer # (56 immeubles). `property_meta` fournit l'adresse civique complète # (REAL_HOMES_property_address), les coordonnées GPS et la galerie photo # (URLs wp-content/uploads, variantes redimensionnées). # Les prix/types d'unités ne sont PAS dans le flux : la fiche propriété # (rendu serveur) affiche un accordéon « Floor Plans » (nom, N Bedrooms, # N Bathrooms, prix plancher, « Call for Availability ») — visitée via le # cache BD self.detail() (clé = date `modified` du flux). Une annonce par # plan d'étage ; repli « une annonce par propriété » (sans prix inventé) # quand l'accordéon est absent. # # Expansion Ontario — GATÉE par LOUKA_ONTARIO=1 : sans la variable, le # connecteur est `disabled` et exclu du registre (zéro impact prod QC). # ----------------------------------------------------------------------------- from __future__ import annotations import os import re from bs4 import BeautifulSoup from ..schema import Listing, normalize_unit_type, parse_price from .base import BaseConnector BASE = "https://medallioncorp.com" FEED_URL = f"{BASE}/wp-json/wp/v2/properties?per_page=100" # Gate expansion Ontario : le connecteur reste hors registre tant que la # variable d'environnement LOUKA_ONTARIO=1 n'est pas posée. _ONTARIO = os.environ.get("LOUKA_ONTARIO") == "1" # villes du flux (slug de class_list `property-city-`) -> affichage ; # tout slug inconnu est titré tel quel (« richmond-hill » -> « Richmond Hill ») _CITY_LABEL = { "toronto": "Toronto", "brampton": "Brampton", "ajax": "Ajax", "whitby": "Whitby", "oshawa": "Oshawa", "richmond-hill": "Richmond Hill", "hamilton": "Hamilton", "london": "London", "milton": "Milton", } # lignes de l'accordéon plans d'étage : « 2 Bedrooms », « 1.5 Bathrooms » _BED_RE = re.compile(r"(\d+)\s*Bedroom", re.I) _BATH_RE = re.compile(r"([\d.]+)\s*Bathroom", re.I) _SQFT_RE = re.compile(r"([\d,]+)\s*(?:sq\.?\s*ft|pi2|ft2)", re.I) _CITY_CLS_RE = re.compile(r"^property-city-(.+)$") class MedallionConnector(BaseConnector): source_id = "medallion" request_delay = 1.0 disabled = not _ONTARIO # gate expansion Ontario (LOUKA_ONTARIO=1) max_properties = 80 # garde-fou (56 immeubles résidentiels aujourd'hui) max_images = 20 def fetch(self) -> list[Listing]: props = self.get(FEED_URL, headers={"Accept": "application/json"}).json() listings: list[Listing] = [] count = 0 for p in props: try: cls = set(p.get("class_list") or []) # résidentiel longue durée seulement : à louer, non commercial if "property-status-for-rent" not in cls: continue if "property-type-residential" not in cls: continue if count >= self.max_properties: break count += 1 listings.extend(self._property_listings(p)) except Exception: continue return listings # -- annonces d'une propriété (une par plan d'étage, repli propriété) ------- def _property_listings(self, p: dict) -> list[Listing]: pid = str(p.get("id")) url = p.get("link") or "" title = BeautifulSoup((p.get("title") or {}).get("rendered") or "", "html.parser").get_text(" ", strip=True) meta = p.get("property_meta") or {} # adresse civique complète du flux (« 25 Kitney Dr, Ajax, ON L1S 0G6, # Canada ») — suffixe « , Canada » retiré address = re.sub(r",\s*Canada$", "", (meta.get("REAL_HOMES_property_address") or "").strip()) # ville : slug property-city- de class_list city = "" for c in p.get("class_list") or []: m = _CITY_CLS_RE.match(c) if m: slug = m.group(1) city = _CITY_LABEL.get(slug, slug.replace("-", " ").title()) break # coordonnées GPS du flux (finalize() valide la bbox Ontario) loc = meta.get("REAL_HOMES_property_location") or {} try: lat = float(loc.get("latitude")) if loc.get("latitude") else None lng = float(loc.get("longitude")) if loc.get("longitude") else None except (TypeError, ValueError): lat = lng = None # galerie photo : variantes redimensionnées du flux (large ~1024px) images: list[str] = [] for img in meta.get("REAL_HOMES_property_images") or []: sizes = (img or {}).get("sizes") or {} u = "" for k in ("1536x1536", "large", "medium_large"): u = (sizes.get(k) or {}).get("url") or "" if u: break if not u and img.get("file"): u = f"{BASE}/wp-content/uploads/{img['file']}" if u and u not in images: images.append(u) images = images[: self.max_images] # description : contenu WordPress de la fiche (rendu texte) desc = BeautifulSoup((p.get("content") or {}).get("rendered") or "", "html.parser").get_text(" ", strip=True)[:600] # fiche propriété (plans d'étage + commodités) via le cache BD : # revisitée seulement quand le post WordPress est modifié feed_key = str(p.get("modified") or p.get("modified_gmt") or "") d = self.detail(pid, feed_key, lambda: self._fetch_detail(url)) amenities = d.get("amenities") or [] common = dict( source=self.source_id, url=url, address=address, city=city, province="ON", description=desc, amenities=amenities[:25], images=images, lat=lat, lng=lng, ) # une annonce par plan d'étage (type d'unité + prix plancher affiché) out: list[Listing] = [] for fp in d.get("floorplans") or []: name = fp.get("name") or "" slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") beds = fp.get("beds") price = fp.get("price") out.append(Listing( external_id=f"{pid}-{slug or 'u'}", title=f"{title} — {name}" if name else title, unit_type=("Studio" if beds == 0 else normalize_unit_type( f"{int(beds)} chambres") if beds is not None else normalize_unit_type(name)), bedrooms=beds, bathrooms=fp.get("baths"), price=price, price_label=(f"À partir de {price:.0f} $ /mois" if price is not None else ""), availability=fp.get("avail") or "", area_sqft=fp.get("sqft"), **common, )) if out: return out # repli : une annonce par propriété — aucun prix inventé (None = inconnu) return [Listing( external_id=pid, title=title, unit_type="", **common, )] # -- fiche propriété : plans d'étage + commodités ---------------------------- def _fetch_detail(self, url: str) -> dict: """Scrape l'accordéon « Floor Plans » (nom, chambres, sdb, prix, disponibilité) et la liste « Buildings Features » de la fiche.""" out: dict = {"floorplans": [], "amenities": []} if not url: return out try: page = self.get(url).text except Exception: return out soup = BeautifulSoup(page, "html.parser") # plans d'étage :
(nom h3, meta « N Bedrooms / # N Bathrooms / prix / Call for Availability ») seen: set[str] = set() for fp in soup.select(".floor-plans-accordions .floor-plan"): h = fp.select_one(".floor-plan-title .title h3") name = h.get_text(" ", strip=True) if h else "" if not name or name.lower() in seen: continue seen.add(name.lower()) meta_el = fp.select_one(".floor-plan-meta") meta_txt = meta_el.get_text(" ", strip=True) if meta_el else "" mb = _BED_RE.search(meta_txt) or _BED_RE.search(name) beds = (float(mb.group(1)) if mb else 0.0 if re.search(r"bachelor|studio", name, re.I) else None) mba = _BATH_RE.search(meta_txt) msq = _SQFT_RE.search(meta_txt) sqft = float(msq.group(1).replace(",", "")) if msq else None price_el = fp.select_one(".floor-price-value") price = parse_price(price_el.get_text(" ", strip=True)) \ if price_el else None # texte de dispo : ce qui suit le prix (« Call for Availability ») avail = "" fprice = fp.select_one(".floor-price") if fprice: avail = fprice.get_text(" ", strip=True) if price_el: avail = avail.replace( price_el.get_text(" ", strip=True), "").strip() out["floorplans"].append({ "name": name, "beds": beds, "baths": float(mba.group(1)) if mba else None, "sqft": sqft if sqft and 80 <= sqft <= 20000 else None, "price": price, "avail": avail, }) # commodités : « Buildings Features » (liste à puces de la fiche) h = soup.find(["h3", "h4"], string=re.compile( r"Buildings? Features", re.I)) if h: ul = h.find_next("ul") if ul: amenities = [] for li in ul.find_all("li"): t = li.get_text(" ", strip=True) if t and len(t) < 60 and t not in amenities: amenities.append(t) out["amenities"] = amenities[:25] return out