# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/urban_services.py : connecteur Urban Services (urbanservices.ca) # Gestionnaire de Gatineau (Urban Tenant Solutions) ; portail de location # dédié rent.urbanservices.ca sur la plateforme Building Stack — même # famille que le portail edifialocation.com du connecteur edifia.py : # - /Listing/Listings embarque `var units = [...]` (une entrée par # appartement annoncé : prix, pi², chambres, sdb, adresse complète, # GPS, contact de l'immeuble, photo) ; # - les pages immeuble /b/ listent TOUTES les unités disponibles # (onglets : numéro, prix, pi², chambres, sdb, date de disponibilité) # + les commodités de l'immeuble — le JSON de la liste est parfois # tronqué (immeubles à fort inventaire), la page /b/ fait donc foi # pour l'inventaire et les dates ; le JSON enrichit (GPS, contact…). # Une annonce Lou-Ka = une unité disponible. Le parc ontarien (Ottawa) # est exclu (province != QC). ~17 requêtes par sync (1 + 16 immeubles). # ----------------------------------------------------------------------------- from __future__ import annotations import html as _html import json import re from bs4 import BeautifulSoup from ..schema import Listing, parse_price from .base import BaseConnector BASE = "https://rent.urbanservices.ca" LIST_URL = f"{BASE}/Listing/Listings" def _extract_json(html: str, marker: str): """Décode la structure JSON qui suit `marker` dans un script inline.""" i = html.find(marker) if i < 0: return None try: data, _ = json.JSONDecoder().raw_decode(html[i + len(marker):]) except Exception: return None return data def _sqft(raw) -> float | None: try: v = float(str(raw).strip().replace(" ", "").replace(" ", "") .replace(",", ".")) except (TypeError, ValueError): return None return v if 80 <= v <= 20000 else None class UrbanServicesConnector(BaseConnector): source_id = "urban_services" request_delay = 0.6 max_buildings = 40 # garde-fou pages immeuble (16 au 2026-08) max_images = 12 def fetch(self) -> list[Listing]: html = self.get(LIST_URL).text bs_units = _extract_json(html, "var units = ") or [] # regrouper le JSON par immeuble (parc québécois seulement) by_pub: dict[str, list[dict]] = {} for u in bs_units: addr = u.get("Address") or {} prov = ((addr.get("Province") or {}).get("ProvinceCode") or "") if prov.upper() != "QC": continue # Ottawa (Ontario) : exclu pub = str((u.get("Building") or {}) .get("PublicListBuildingName") or "") if pub: by_pub.setdefault(pub, []).append(u) listings: list[Listing] = [] for pub, units in sorted(by_pub.items())[: self.max_buildings]: try: listings.extend(self._building(pub, units)) except Exception: continue # un immeuble ne bloque pas le reste # dédup par external_id (sécurité) uniq: dict[str, Listing] = {} for lst in listings: uniq.setdefault(lst.external_id, lst) return list(uniq.values()) # -- page immeuble /b/ : inventaire complet + commodités ------------------- def _building_page(self, pub: str) -> tuple[list[dict], list[str]]: """(lignes d'unités des onglets, commodités de l'immeuble).""" html = self.get(f"{BASE}/b/{pub}").text soup = BeautifulSoup(html, "html.parser") rows: list[dict] = [] for a in soup.select("ul.wpb-tabs-menu a.apartment-view-potential"): spans = [s.get_text(" ", strip=True) for s in a.find_all("span")] if len(spans) >= 6 and spans[0]: rows.append({ # Unité, Prix, pi², chambres, sdb, dispo "unit": spans[0], "price": spans[1], "sqft": spans[2], "beds": spans[3], "baths": spans[4], "dispo": spans[5]}) amenities: list[str] = [] for li in soup.select(".facilities ul li label"): t = re.sub(r"\s+", " ", li.get_text(" ", strip=True)) if 3 <= len(t) <= 90 and t not in amenities: amenities.append(t) return rows, amenities def _building(self, pub: str, units: list[dict]) -> list[Listing]: rows, amenities = self._building_page(pub) # infos communes de l'immeuble depuis le JSON de la liste ref = units[0] addr = ref.get("Address") or {} # le JSON Razor du portail encode les apostrophes (') : décoder bname = _html.unescape(str(ref.get("BuildingName") or "")).strip() try: lat = float(addr.get("Latitude")) lng = float(addr.get("Longitude")) except (TypeError, ValueError): lat = lng = None contact: dict = {} contacts = ((ref.get("Building") or {}) .get("ListingEmployeesContacts") or []) if contacts: c = contacts[0] contact = {k: v for k, v in [ ("name", c.get("FullName")), ("phone", c.get("FormattedPhoneNumber")), ("email", c.get("Email"))] if v} b_img = str(ref.get("BuildingPreviewUrl") or "") by_name = {str((u.get("Apartment") or {}).get("UnitName") or ""): u for u in units} out: list[Listing] = [] seen: set[str] = set() for row in rows: num = row["unit"].strip() if not num or num in seen: continue seen.add(num) js = by_name.get(num) or {} out.append(self._listing(pub, num, row, js, bname, addr, amenities, contact, b_img, lat, lng)) # unités du JSON absentes des onglets (prudence : ne rien perdre) for num, js in by_name.items(): if num and num not in seen: seen.add(num) out.append(self._listing(pub, num, None, js, bname, addr, amenities, contact, b_img, lat, lng)) return out # -- une annonce par unité ------------------------------------------------------- def _listing(self, pub: str, num: str, row: dict | None, js: dict, bname: str, addr: dict, amenities: list[str], contact: dict, b_img: str, lat, lng) -> Listing: apt = js.get("Apartment") or {} # chambres / sdb : onglet de la page immeuble, sinon JSON de la liste def _int(v) -> int | None: try: return int(str(v).strip()) except (TypeError, ValueError): return None beds = _int(row["beds"]) if row else None if beds is None: beds = _int(apt.get("NumberOfBedrooms")) baths = _int(row["baths"]) if row else None if baths is None: baths = _int(apt.get("NumberOfBathrooms")) # convention Building Stack (comme edifia) : 0 chambre = studio unit_type = "" if beds is None else ("Studio" if beds == 0 else f"{beds + 2}½") # prix : JSON structuré prioritaire, sinon texte de l'onglet price = float(apt.get("Price") or 0) or None price_label = str(apt.get("PriceFormatted") or "") if row and not price_label: price_label = row["price"] if price is None and row: price = parse_price(row["price"]) area = _sqft(apt.get("Area")) if area is None and row: area = _sqft(row["sqft"]) # disponibilité : texte de l'onglet (« Disponible dès maintenant! », # « oct. 01, 2027 »…) — préfixé quand c'est une date nue availability = "" if row: dispo = row["dispo"].strip() if dispo and re.search(r"\d", dispo) and "maintenant" not in dispo.lower(): availability = f"Libre {dispo}" else: availability = dispo desc = " — ".join(x for x in [ f"{area:g} pi²" if area else "", f"{beds} chambre(s)" if beds else "", f"{baths} salle(s) de bain" if baths else ""] if x) details: dict = {} if bname: details["building"] = bname if contact: details["contact"] = dict(contact) if beds is not None: details["bedrooms"] = beds if baths is not None: details["bathrooms"] = baths images: list[str] = [] prev = str(js.get("PreviewUrl") or "") if prev.startswith("http"): images.append(prev) if b_img.startswith("http") and b_img not in images: images.append(b_img) apt_id = js.get("ApartmentId") url = (f"{BASE}/b/{pub}/{apt_id}" if apt_id else f"{BASE}/b/{pub}") title = (f"{bname} — unité {num}" if bname else f"Urban Services — unité {num}") return Listing( source=self.source_id, external_id=f"{pub}-{num}", url=url, title=title, address=_html.unescape(str(addr.get("Full") or "")), sector="", city=_html.unescape(str(addr.get("City") or "")), unit_type=unit_type, price=price, price_label=price_label, availability=availability, area_sqft=area, description=desc, amenities=list(amenities), details=details, images=images[: self.max_images], lat=lat, lng=lng, )