# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (Québec + expansion Ontario) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/qresidential.py : connecteur Q Residential (qresidential.ca) # Gestionnaire ontarien (~7 000 unités — Toronto, St. Catharines, Hamilton, # Oshawa/GTA, Brampton, Barrie). « Corporate site » Yardi RentCafe derrière # un Cloudflare STRICT (curl direct = 403) : tout passe par Scrapfly ASP # (self.get_scrapfly, asp=true, SANS render_js — le contenu est rendu # serveur, vérifié sur la recherche et les fiches). # Contrairement à killam.py (même famille RentCafe), il n'y a PAS de blob # #available_prop : la page /searchlisting.aspx rend 25 cartes propriétés # côté serveur (nom, adresse, ville, province, vignette, URL de fiche). # La fiche propriété fournit un JSON-LD ApartmentComplex (description, # adresse postale, GPS, commodités, téléphone) et un tableau « Floor Plans » # (tbody.floorplan-details : nom, Bed/Bath, pi², loyer — souvent « Call for # pricing » —, disponibilité). Une annonce PAR PLAN D'ÉTAGE quand au moins # un plan affiche un prix ; sinon repli « une annonce par propriété » sans # prix (rien d'inventé). Fiches visitées via le cache BD self.detail(), # clé datée du jour : au plus UNE visite Scrapfly par propriété par jour, # peu importe le nombre de synchronisations. # # 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 datetime as _dt import json import os import re from bs4 import BeautifulSoup from ..schema import Listing, normalize_unit_type, parse_price from .base import BaseConnector BASE = "https://www.qresidential.ca" SEARCH_PAGE = f"{BASE}/searchlisting.aspx" # 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" # cellules du tableau plans d'étage : « Bed/Bath Studio / 1 », « 2 / 1.5 » _BEDBATH_RE = re.compile( r"Bed/Bath\s+(Studio|\d+)\s*/\s*([\d.]+)", re.I) _SQFT_RE = re.compile(r"([\d,]+)\s*Sq\.?\s*Ft", re.I) # identifiant stable du plan d'étage (carrousel photo du plan) _FPID_RE = re.compile(r"fp-myCarousel(\d+)") class QResidentialConnector(BaseConnector): source_id = "qresidential" request_delay = 1.5 # Cloudflare strict : Scrapfly ASP, rester poli disabled = not _ONTARIO # gate expansion Ontario (LOUKA_ONTARIO=1) max_properties = 35 # garde-fou (25 propriétés aujourd'hui) max_images = 15 def fetch(self) -> list[Listing]: page = self.get_scrapfly(SEARCH_PAGE, asp=True, render_js=False) if not page: raise RuntimeError("Q Residential : Scrapfly ASP n'a pas " "retourné la page de recherche") soup = BeautifulSoup(page, "html.parser") cards = soup.select("div.searchResult") listings: list[Listing] = [] count = 0 seen: set[str] = set() for card in cards: try: a = card.select_one("a.propertyUrl") if a is None: continue url = (a.get("href") or "").strip() if not url or url in seen: continue seen.add(url) # résidentiel longue durée seulement (structure « Apartment ») st = card.select_one(".structure-type") if st and st.get_text(" ", strip=True) and \ "apartment" not in st.get_text(" ", strip=True).lower(): continue state = card.select_one(".propertyState") if state and state.get_text(strip=True).upper() != "ON": continue if count >= self.max_properties: break count += 1 listings.extend(self._property_listings(card, url)) except Exception: continue return listings # -- annonces d'une propriété (une par plan d'étage tarifé, repli) ---------- def _property_listings(self, card, url: str) -> list[Listing]: # identifiant : slug de la fiche (« queenston-manor ») slug = url.rstrip("/").split("/")[-2] if url.endswith("default.aspx") \ else re.sub(r"[^a-z0-9-]", "", url.rstrip("/").split("/")[-1]) a = card.select_one("a.propertyUrl") name = a.get_text(" ", strip=True) city = (card.select_one(".propertyCity") or a) \ .get_text(" ", strip=True) if card.select_one(".propertyCity") \ else "" # adresse complète de la carte (« 382 Queenston Street St. Catharines # ON L2P 3V5 ») — on la reconstruit avec des virgules street = "" spans = card.select(".propertyAddress") if spans: street = spans[-1].get_text(" ", strip=True) postal = "" m = re.search(r"[A-Z]\d[A-Z]\s?\d[A-Z]\d", card.get_text(" ", strip=True)) if m: postal = m.group(0) address = ", ".join(x for x in (street, city) if x) if address: address += f", ON {postal}".rstrip() thumb = "" img = card.select_one("img.propertyThumb") if img is not None: thumb = (img.get("src") or "").strip() # fiche propriété via le cache BD — clé datée : au plus une visite # Scrapfly par propriété par jour (les prix « Call for pricing » # changent rarement, et jamais entre deux syncs du même jour) feed_key = f"{_dt.date.today().isoformat()}|{name}|{thumb}" d = self.detail(slug, feed_key, lambda: self._fetch_detail(url)) desc = d.get("description") or "" amenities = d.get("amenities") or [] images = [u for u in ([thumb] + (d.get("images") or [])) if u] lat, lng = d.get("lat"), d.get("lng") details: dict = {} if d.get("phone"): details["contact"] = {"phone": d["phone"]} common = dict( source=self.source_id, url=url, address=address, city=city, province="ON", description=desc, amenities=amenities[:25], images=images[: self.max_images], lat=lat, lng=lng, ) # une annonce par plan d'étage TARIFÉ (prix affiché) ; les plans # « Call for pricing » ne deviennent pas des annonces individuelles out: list[Listing] = [] priced = [fp for fp in d.get("floorplans") or [] if fp.get("price") is not None] for fp in priced: beds = fp.get("beds") price = fp["price"] out.append(Listing( external_id=f"{slug}-{fp['fpid']}", title=f"{name} — {fp['name']}" if fp.get("name") else name, unit_type=("Studio" if beds == 0 else normalize_unit_type( f"{int(beds)} chambres") if beds is not None else ""), bedrooms=beds, bathrooms=fp.get("baths"), price=price, price_label=f"À partir de {price:.0f} $ /mois", availability=fp.get("avail") or "", area_sqft=fp.get("sqft"), details=dict(details), **common, )) if out: return out # repli : une annonce par propriété — gamme de types connue seulement # via les plans d'étage non tarifés ; aucun prix inventé fps = d.get("floorplans") or [] beds_set = {fp.get("beds") for fp in fps if fp.get("beds") is not None} unit_type = "" if len(beds_set) == 1: b = beds_set.pop() unit_type = "Studio" if b == 0 else normalize_unit_type( f"{int(b)} chambres") return [Listing( external_id=slug, title=name, unit_type=unit_type, availability="Sur demande (contacter la gestion)" if fps else "", details=details, **common, )] # -- fiche propriété : JSON-LD + tableau plans d'étage ---------------------- def _fetch_detail(self, url: str) -> dict: out: dict = {"description": "", "amenities": [], "images": [], "floorplans": [], "lat": None, "lng": None, "phone": ""} page = self.get_scrapfly(url, asp=True, render_js=False) if not page: return out soup = BeautifulSoup(page, "html.parser") # JSON-LD ApartmentComplex : description, GPS, commodités, téléphone for tag in soup.find_all("script", type="application/ld+json"): try: ld = json.loads(tag.string or "") except (TypeError, ValueError): continue if not isinstance(ld, dict) or \ ld.get("@type") != "ApartmentComplex": continue out["description"] = BeautifulSoup( ld.get("description") or "", "html.parser" ).get_text(" ", strip=True)[:600] geo = ld.get("geo") or {} try: out["lat"] = float(geo.get("latitude")) out["lng"] = float(geo.get("longitude")) except (TypeError, ValueError): pass out["phone"] = ((ld.get("address") or {}) .get("telephone") or "").strip() ams = [] for af in ld.get("amenityFeature") or []: t = (af.get("name") or "").strip() if isinstance(af, dict) \ else "" if t and t not in ams: ams.append(t) out["amenities"] = ams[:25] break # visuels de la fiche (bannière dmslivecafe) — HORS plans d'étage/logos images: list[str] = [] for img in soup.find_all("img"): src = (img.get("src") or "").strip() if "cdngeneralcf.rentcafe.com/dmslivecafe" not in src: continue if re.search(r"logo|icon|floor\s?plan|FloorPlan", src, re.I): continue if img.find_parent(class_=re.compile("floorplan")): continue if src not in images: images.append(src) out["images"] = images[: self.max_images] # tableau plans d'étage : une ligne par variation (nom, Bed/Bath, # pi² éventuel, loyer, disponibilité) seen: set[str] = set() for tr in soup.select("tbody.floorplan-details tr"): txt = tr.get_text(" ", strip=True) if "Bed/Bath" not in txt: continue mid = _FPID_RE.search(str(tr)) name_td = tr.find(string=re.compile(r"Floor Plan\s")) name = "" if name_td: name = re.sub(r"^Floor Plan\s+", "", str(name_td).strip()).strip() fpid = mid.group(1) if mid else \ re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") if not fpid or fpid in seen: continue seen.add(fpid) mbb = _BEDBATH_RE.search(txt) beds = baths = None if mbb: beds = 0.0 if mbb.group(1).lower() == "studio" \ else float(mbb.group(1)) try: baths = float(mbb.group(2)) except ValueError: baths = None msq = _SQFT_RE.search(txt) sqft = float(msq.group(1).replace(",", "")) if msq else None # loyer : montant affiché seulement (« Call for pricing » -> None) price = None mrent = re.search(r"Rent\s+([^D]*?)(?:Deposit|$)", txt) if mrent: price = parse_price(mrent.group(1)) avail = "" mav = re.search(r"Deposit\s*(.*?)(?:Read More|$)", txt) if mav: avail = mav.group(1).strip() if avail.lower() in ("contact us", ""): avail = "" out["floorplans"].append({ "fpid": fpid, "name": name, "beds": beds, "baths": baths, "sqft": sqft if sqft and 80 <= sqft <= 20000 else None, "price": price, "avail": avail, }) return out