# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/akelius.py : connecteur Akelius Residential (rent.akelius.com) # Page de recherche Canada rendue côté serveur (Angular SSR) : l'état # TransferState (', re.S) # Villes admissibles (Grand Montréal) -> (ville normalisée, secteur imposé) _GM_CITIES = { "montreal": ("Montréal", None), # secteur = borough du flux "westmount": ("Westmount", "Westmount"), "mont-royal": ("Mont-Royal", "Mont-Royal"), "saint-lambert": ("Saint-Lambert", "Saint-Lambert"), "greenfield park": ("Longueuil", "Greenfield Park"), } _BED_TYPES = {0: "Studio", 1: "3½", 2: "4½", 3: "5½", 4: "6½"} # keyfacts booléens du JSON détail -> libellé de commodité (si vrai) _KF_AMENITIES = { "has-air-conditioning": "Climatisation", "has-central-air-conditioning": "Climatisation centrale", "has-balcony": "Balcon", "has-terrace": "Terrasse", "has-bicycle-racks": "Supports à vélos", "has-blinds": "Stores", "has-bosch-built-in-appliances": "Électroménagers encastrés Bosch", "has-canada-post-parcel-locker": "Casier à colis Postes Canada", "has-central-vacuuming-system": "Aspirateur central", "has-concierge": "Concierge", "has-dishwasher": "Lave-vaisselle", "has-dryer": "Sécheuse", "has-washing-machine": "Laveuse", "has-washer-dryer": "Laveuse-sécheuse", "has-elevator": "Ascenseur", "has-fitness-centre": "Salle d'entraînement", "has-heated-bathroom-floor": "Plancher de salle de bain chauffant", "has-indoor-pool": "Piscine intérieure", "has-outdoor-pool": "Piscine extérieure", "has-laundry-room": "Buanderie", "has-microwave": "Micro-ondes", "has-openplan-kitchen": "Cuisine à aire ouverte", "has-sauna": "Sauna", "has-wine-fridge": "Cellier à vin", "is-broadband-included-in-rent": "Internet inclus", "is-electricity-included-in-rent": "Électricité incluse", "is-gas-included-in-rent": "Gaz inclus", "is-heating-included-in-rent": "Chauffage inclus", "is-hot-water-included-in-rent": "Eau chaude incluse", "is-water-included-in-rent": "Eau incluse", "is-refurbished": "Rénové", "is-smart-home": "Logement intelligent", } class _DetailBudget(Exception): """Budget de nouvelles pages détail épuisé pour cette synchronisation.""" class AkeliusConnector(BaseConnector): source_id = "akelius" request_delay = 0.6 max_units = 400 # garde-fou max_details = 150 # nouveaux JSON détail max par synchronisation def fetch(self) -> list[Listing]: html = self.get(SEARCH_URL).text m = STATE_RE.search(html) if not m: return [] # TransferState Angular : les guillemets sont encodés « &q; » state = json.loads(m.group(1).replace("&q;", '"')) # La clé du cache API est un hash variable : on repère la liste d'unités units: list[dict] = [] for val in state.values(): body = val.get("b") if isinstance(val, dict) else None if (isinstance(body, list) and body and isinstance(body[0], dict) and "keyfacts" in body[0]): units = body break self._detail_fetches = 0 listings: list[Listing] = [] for u in units[: self.max_units]: try: lst = self._unit_listing(u) if lst: self._enrich(lst, u) listings.append(lst) except Exception: continue return listings # -- JSON détail (keyfacts complets) --------------------------------------- def _fetch_detail(self, uid: str) -> dict: try: resp = self.get(DETAIL_URL.format(uid=uid)) data = json.loads(resp.content.decode("utf-8-sig")) except Exception: return {} if not isinstance(data, dict): return {} docs = data.get("documents") or [] images = [d.get("mediumUrl") or d.get("originalImageUrl") for d in docs if isinstance(d, dict) and not d.get("isExample")] return { "keyfacts": data.get("keyfacts") or {}, "contact": data.get("contactDetails") or {}, "images": [i for i in images if isinstance(i, str)][:30], } def _enrich(self, lst: Listing, u: dict) -> None: """Complète l'annonce avec le JSON détail (via cache self.detail).""" key = hashlib.sha1(json.dumps( {"rent": (u.get("keyfacts") or {}).get("total-rent"), "avail": (u.get("keyfacts") or {}).get("available-from-date"), "pub": u.get("lastPublishedDate")}, sort_keys=True).encode("utf-8")).hexdigest() def fetch_fn(): if self._detail_fetches >= self.max_details: raise _DetailBudget() self._detail_fetches += 1 return self._fetch_detail(lst.external_id) try: payload = self.detail(lst.external_id, key, fetch_fn) except _DetailBudget: return kf = payload.get("keyfacts") or {} if not kf: return # commodités (libellés français, seulement les keyfacts vrais) for k, label in _KF_AMENITIES.items(): if kf.get(k) and label not in lst.amenities: lst.amenities.append(label) # animaux / meublé (valeurs structurées de la source) pets = str(kf.get("pets-allowed") or "").strip().lower() if pets == "yes": lst.pets = "oui" elif pets == "no": lst.pets = "non" elif pets: lst.pets = "conditions" furn = str(kf.get("furnished-state") or "").strip().lower() if furn == "furnished": lst.furnished = True elif furn == "unfurnished": lst.furnished = False # details structurés (booléens explicites du JSON — jamais devinés) details: dict = {} inclusions = {} for src, dst in (("is-heating-included-in-rent", "heating"), ("is-electricity-included-in-rent", "electricity"), ("is-hot-water-included-in-rent", "hot_water"), ("is-broadband-included-in-rent", "internet")): if isinstance(kf.get(src), bool): inclusions[dst] = kf[src] if inclusions: details["inclusions"] = inclusions appliances = {} if isinstance(kf.get("has-dishwasher"), bool): appliances["dishwasher"] = kf["has-dishwasher"] if isinstance(kf.get("has-washer-dryer"), bool): wd = kf["has-washer-dryer"] or ( bool(kf.get("has-washing-machine")) and bool(kf.get("has-dryer"))) appliances["washer_dryer"] = wd if appliances: details["appliances"] = appliances if isinstance(kf.get("has-air-conditioning"), bool): details["ac"] = (kf["has-air-conditioning"] or bool(kf.get("has-central-air-conditioning"))) if isinstance(kf.get("has-elevator"), bool): details["elevator"] = kf["has-elevator"] if isinstance(kf.get("has-balcony"), bool): details["balcony"] = kf["has-balcony"] or bool(kf.get("has-terrace")) if isinstance(kf.get("has-indoor-pool"), bool) or \ isinstance(kf.get("has-outdoor-pool"), bool): details["pool"] = bool(kf.get("has-indoor-pool")) or \ bool(kf.get("has-outdoor-pool")) if isinstance(kf.get("has-fitness-centre"), bool): details["gym"] = kf["has-fitness-centre"] if isinstance(kf.get("has-laundry-room"), bool): details["laundry"] = kf["has-laundry-room"] year = kf.get("construction-year") if isinstance(year, int): details["construction_year"] = year phone = ((payload.get("contact") or {}).get("phoneNumber") or "").strip() if phone: details["contact"] = {"phone": phone} if details: lst.details = details # description libre éventuelle (keyfact « free-text ») free = str(kf.get("free-text") or "").strip() if free: lst.description = (lst.description + " — " + free)[:600] \ if lst.description else free[:600] # photos pleine résolution du détail (600 px au lieu de 400 px) if payload.get("images"): lst.images = payload["images"][:30] def _unit_listing(self, u: dict) -> Listing | None: addr = u.get("address") or {} kf = u.get("keyfacts") or {} if (addr.get("province") or "").upper() != "QC": return None city_key = strip_accents((addr.get("city") or "").strip().lower()) if city_key not in _GM_CITIES: return None city, forced_sector = _GM_CITIES[city_key] sector = forced_sector or (addr.get("borough") or "").strip() uid = str(u.get("id") or "").strip() if not uid: return None street = (addr.get("streetName") or "").strip() postal = (addr.get("postalCode") or "").strip() beds = kf.get("number-of-bedrooms") unit_type = _BED_TYPES.get(beds, "") if isinstance(beds, int) else "" apt_type = (kf.get("apartment-type") or "").strip() if not unit_type and apt_type == "loft": unit_type = "Loft" rent = kf.get("total-rent") price = float(rent) if isinstance(rent, (int, float)) and rent else None if kf.get("is-available-from-now-on"): availability = "Libre maintenant" else: availability = (kf.get("available-from-date") or "")[:10] if availability: availability = f"Disponible le {availability}" size = kf.get("unit-size") baths = kf.get("number-of-bathrooms") floor = kf.get("floor") desc_bits = [] if apt_type: desc_bits.append(f"Type : {apt_type}") if size: desc_bits.append(f"{size} pi²") if baths: desc_bits.append(f"{baths} salle(s) de bain") if floor is not None: desc_bits.append(f"étage {floor}") if kf.get("free-rent"): desc_bits.append(f"promotion : {kf['free-rent']}") amenities = [] if size: amenities.append(f"{size} pi²") if baths: amenities.append(f"{baths} sdb") images = [i for i in (u.get("imageUrls") or []) if isinstance(i, str) and i.startswith("http")][:30] title = f"{street} — unité {uid.split('-')[-1]}" if street else uid return Listing( source=self.source_id, external_id=uid, url=f"{BASE}/en/search/canada/detail/{uid}", title=title, address=", ".join(x for x in [street, city, postal] if x), sector=sector, city=city, unit_type=unit_type, price=price, price_label=f"{int(rent)} $/mois" if price else "", availability=availability, description=" — ".join(desc_bits)[:600], amenities=amenities, images=images, lat=addr.get("latitude"), lng=addr.get("longitude"), )