# -----------------------------------------------------------------------------
# Lou-Ka — Agrégateur de logements à louer (province de Québec)
# Auteur : Simon-Pierre Boucher — contact@spboucher.ai
# connectors/quartier_sila.py : connecteur Quartier Sila (silalevis.ca)
# Complexe de condos locatifs du Groupe Damco / Développement Beaubourg à
# Saint-Romuald (Lévis) — 4 phases (Guillaume-Couture, J.-B.-Demers,
# d'Anticosti), 3½ à 5½. Site WordPress/Elementor derrière Cloudflare ; la
# page « Condos locatifs Sila » embarque le module de plans interactifs
# Livya (app.livya.com, client « damco », projet « sila »). La page Next.js
# du module est rendue côté serveur : son flux RSC (self.__next_f.push)
# contient l'inventaire JSON complet — numéro d'unité, immeuble/étage,
# statut, loyer, pièces, superficie, adresse civique, GPS et plans. Les
# unités de la phase 4 (prélocation) n'ont pas encore de prix publié.
# 2 requêtes par sync : la page WordPress (id d'entité Livya + services et
# commodités + bannières de phases) + la page du module.
# -----------------------------------------------------------------------------
from __future__ import annotations
import codecs
import json
import re
from bs4 import BeautifulSoup
from ..schema import Listing, infer_city
from .base import BaseConnector
BASE = "https://silalevis.ca"
LIST_URL = f"{BASE}/condos-locatifs-sila/"
LIVYA = "https://app.livya.com"
SECTOR = "Saint-Romuald"
# fragments RSC de Next.js : self.__next_f.push([1,"...payload échappé..."])
_NEXT_F_RE = re.compile(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)')
# bannières de la page : « Phase 3 en construction pour livraison à l'été
# 2026! », « Phase 4 en prélocation à l'automne 2026! »
_PHASE_BANNER_RE = re.compile(
r"Phase\s*(\d)\s*(en\s+(?:construction|pr[ée]location)[^!.<]{0,80})", re.I)
# item de commodité réservé à certaines phases : « (Phases 2 et 3) »
_AMEN_PHASE_RE = re.compile(r"phases?\s*([\d\s,et]+)", re.I)
def _flight_blob(html: str) -> str:
"""Concatène et désérialise les fragments RSC d'une page Livya.
`unicode_escape` interprète les octets en latin-1 : on ré-encode pour
retrouver l'UTF-8 d'origine (sinon « Étage » devient « Ãtage »).
"""
blob = "".join(codecs.decode(c, "unicode_escape")
for c in _NEXT_F_RE.findall(html))
return blob.encode("latin-1", "ignore").decode("utf-8", "ignore")
def _json_arrays(blob: str, key: str) -> list[list]:
"""Toutes les valeurs de `"key":[...]` du flux (équilibrage de crochets)."""
out: list[list] = []
for m in re.finditer(re.escape(f'"{key}":['), blob):
j = m.end() - 1
depth, in_str, esc = 0, False, False
for k in range(j, len(blob)):
c = blob[k]
if in_str:
if esc:
esc = False
elif c == "\\":
esc = True
elif c == '"':
in_str = False
elif c == '"':
in_str = True
elif c == "[":
depth += 1
elif c == "]":
depth -= 1
if depth == 0:
try:
out.append(json.loads(blob[j:k + 1]))
except ValueError:
pass
break
return out
def _unit_type(rooms: float | None) -> str:
"""3.5 -> « 3½ » ; 0.5 -> « Studio » (finalize gère 6½+)."""
if not rooms:
return ""
if rooms < 1:
return "Studio"
return f"{int(rooms)}½"
class QuartierSilaConnector(BaseConnector):
source_id = "quartier_sila"
request_delay = 0.8
# -- page WordPress : commodités listées sous « Services et commodités » ----
@staticmethod
def _amenities(soup: BeautifulSoup) -> list[str]:
"""Items `
` entre le titre « Services et commodités » et le titre
suivant (le pied de page et le bandeau de cookies ont aussi des puces)."""
items: list[str] = []
collecting = False
for el in soup.find_all(["h1", "h2", "h3", "h4", "li"]):
text = re.sub(r"\s+", " ", el.get_text(" ", strip=True))
if el.name != "li":
if collecting:
break
collecting = "services et commodit" in text.lower()
continue
if collecting and 3 <= len(text) <= 120:
items.append(text)
return list(dict.fromkeys(items))[:20]
@staticmethod
def _amenities_for_phase(amenities: list[str], phase: str) -> list[str]:
"""Écarte les items réservés à d'autres phases (« (Phases 2 et 3) »)."""
out = []
for item in amenities:
m = _AMEN_PHASE_RE.search(item)
if m and phase and phase not in re.findall(r"\d", m.group(1)):
continue
out.append(item)
return out
def fetch(self) -> list[Listing]:
# 1) Page « Condos locatifs Sila » : id d'entité Livya + commodités +
# bannières d'état des phases + téléphone du bureau de location
wp = self.get(LIST_URL).text
m = re.search(r"<[^>]*livya-module-container-plans[^>]*>", wp)
if not m:
raise RuntimeError("module Livya introuvable sur la page condos")
tag = m.group(0)
client_m = re.search(r'data-client="([^"]+)"', wp)
project = re.search(r'data-project="([^"]+)"', tag)
entity = re.search(r'data-entity="([^"]+)"', tag)
if not (client_m and project and entity):
raise RuntimeError("attributs data-client/project/entity manquants")
soup = BeautifulSoup(wp, "html.parser")
amenities = self._amenities(soup)
contact: dict = {}
tel = re.search(r'href="tel:([\d\s\-.]{10,14})"', wp)
if tel:
d = re.sub(r"\D", "", tel.group(1))[-10:]
if len(d) == 10:
contact["phone"] = f"{d[:3]}-{d[3:6]}-{d[6:]}"
# état publié par phase (construction/prélocation + horizon de livraison)
banners = {n: f"Phase {n} {re.sub(r'[ ]+', ' ', rest).strip()}"
for n, rest in _PHASE_BANNER_RE.findall(
soup.get_text(" ", strip=True))}
# 2) Page du module Livya (rendue serveur) -> inventaire JSON complet
livya_url = (f"{LIVYA}/fr/{client_m.group(1)}/projects/"
f"{project.group(1)}/plans/{entity.group(1)}?noLayout=1")
blob = _flight_blob(self.get(livya_url).text)
phase_names: dict[str, str] = {}
for arr in _json_arrays(blob, "phases"):
for p in arr:
if isinstance(p, dict) and p.get("phaseId") and p.get("name"):
phase_names[p["phaseId"]] = str(p["name"])
units, seen = [], set()
for arr in _json_arrays(blob, "units"):
for u in arr:
if isinstance(u, dict) and u.get("unitId") and u["unitId"] not in seen:
seen.add(u["unitId"])
units.append(u)
listings: list[Listing] = []
for u in units:
if u.get("availability") != "AVAILABLE" or not u.get("rental", True):
continue
num = str(u.get("number") or "").strip()
price = u.get("rentalPrice")
price = float(price) if isinstance(price, (int, float)) and price > 0 else None
area = u.get("unitSize")
area = float(area) if isinstance(area, (int, float)) and area > 0 else None
phase = phase_names.get(u.get("phaseId") or "", "")
# immeuble + étage depuis « Sila 3 - Étage 6 »
floor_disp = str(u.get("floorDisplayName") or "").strip()
building, _, floor_label = (x.strip() for x in
floor_disp.partition(" - "))
desc: list[str] = []
if floor_label:
desc.append(floor_label.capitalize())
elif u.get("floorNumber"):
desc.append(f"Étage {u['floorNumber']}")
if phase:
desc.append(f"Phase {phase}")
if u.get("typeName"):
desc.append(f"Modèle {u['typeName']}")
if u.get("roomsBed"):
desc.append(f"{u['roomsBed']} chambre(s)")
if u.get("roomsBath"):
desc.append(f"{u['roomsBath']} salle(s) de bain")
if u.get("balconySize"):
desc.append(f"Balcon de {u['balconySize']} pi²")
if u.get("floorPlanUrl"):
desc.append(f"Plan : {u['floorPlanUrl']}")
address = ", ".join(x for x in (
u.get("address") or "", u.get("city") or "",
u.get("postalCode") or "") if x)
images = [img.get("fullUrl") for img in (u.get("typeImages") or [])
if isinstance(img, dict) and img.get("fullUrl")]
if u.get("floorPlanImageUrl"):
images.append(u["floorPlanImageUrl"])
# disponibilité : date future publiée > bannière de phase
# (« Phase 4 en prélocation à l'automne 2026 ») > statut du plan
future = u.get("futureAvailability")
availability = (str(future) if future
else banners.get(phase, "Disponible"))
details: dict = {}
if contact:
details["contact"] = dict(contact)
floor_no = str(u.get("floorNumber") or "")
if floor_no.isdigit():
details["floor"] = int(floor_no)
titre = (f"Quartier Sila — {building}, unité {num}" if building
else f"Quartier Sila — unité {num}")
lat, lng = u.get("latitude"), u.get("longitude")
listings.append(Listing(
source=self.source_id,
external_id=str(u["unitId"]),
url=LIST_URL,
title=f"{titre} ({_unit_type(u.get('rooms'))})",
address=address,
sector=SECTOR,
city=infer_city(SECTOR),
unit_type=_unit_type(u.get("rooms")),
price=price,
price_label=f"{price:.0f} $ /mois" if price else "",
availability=availability,
area_sqft=area,
description=" | ".join(desc),
amenities=self._amenities_for_phase(amenities, phase),
details=details,
images=images[:12],
lat=float(lat) if lat else None,
lng=float(lng) if lng else None,
))
return listings