)
_SUPERFICIE_RE = re.compile(
r"id=['\"]superficie_encadre['\"][^>]*>\s*([\d\s]{2,7})\s*pi", re.I)
# Fiche : coordonnées de la carte Google (new google.maps.LatLng(46.84,-71.26))
_LATLNG_RE = re.compile(r"google\.maps\.LatLng\((-?\d{1,3}\.\d+),\s*(-?\d{1,3}\.\d+)\)")
# Fiche : adresse complète avec no d'unité (« 4855, 6e Avenue Ouest # 403 »)
_ADDR_UNIT_RE = re.compile(r'
(.*?)
', re.S | re.I)
def _strip_tags(s: str) -> str:
return re.sub(r"\s+", " ", htmllib.unescape(re.sub(r"<[^>]+>", " ", s))).strip()
def _fix_sup(street: str) -> str:
"""« 6 e Avenue » (issu de 6
e) -> « 6e Avenue »."""
return re.sub(r"(\d)\s+(e|er|re)(\s+|$)", r"\1\2 ", street).strip()
class _DetailBudget(Exception):
"""Plafond de vraies requêtes de fiches atteint pour cette synchro."""
class LogisbourgConnector(BaseConnector):
source_id = "logisbourg"
request_delay = 0.6
max_details = 80 # plafond de VRAIES requêtes de fiches par synchro
def __init__(self) -> None:
super().__init__()
self._detail_fetches = 0
# le serveur n'annonce pas de charset : forcer l'UTF-8 déclaré dans le HTML
def _get_text(self, url: str) -> str:
resp = self.get(url)
resp.encoding = "utf-8"
return resp.text
def fetch(self) -> list[Listing]:
html = self._get_text(LIST_URL)
# 1) Position des en-têtes d'immeuble (secteur : nom + adresse de rue)
complexes: list[tuple[int, str, str, str]] = [] # (pos, secteur, nom, rue)
for m in _COMPLEX_RE.finditer(html):
header = _strip_tags(m.group(1)) # "Secteur : Nom"
street = _strip_tags(m.group(2)) # "6e Avenue Ouest"
sector, _, name = header.partition(":")
complexes.append((m.start(), sector.strip(), name.strip(), street))
# 2) Lignes de disponibilités
listings: dict[str, Listing] = {}
row_keys: dict[str, str] = {} # lid -> hash de la ligne
for row_m in _ROW_RE.finditer(html):
row = row_m.group(0)
link = _LINK_RE.search(row)
if not link:
continue
sid, cid, lid, sector_q, complex_q = link.groups()
if lid in listings:
continue
# immeuble courant = dernier en-tête avant cette ligne
sector = name = street = ""
for pos, sec, nom, rue in complexes:
if pos < row_m.start():
sector, name, street = sec, nom, rue
else:
break
sector = sector or sector_q
name = name or complex_q
def cell(cls: str) -> str:
c = re.search(r'class="%s"[^>]*>(.*?)' % cls, row, re.S | re.I)
return _strip_tags(c.group(1)) if c else ""
def cell_title(cls: str) -> str:
c = re.search(r'class="%s"\s+TITLE="([^"]*)"' % cls, row, re.I)
return htmllib.unescape(c.group(1)).replace("\xa0", " ") if c else ""
unit_raw = cell("Grandeur")
avail = cell_title("DateDispo") or cell("DateDispo")
floor = cell_title("Etage") or cell("Etage")
price_label = cell("Prix")
# inclusions (icônes avec ALT dans la colonne "Inclus")
included = [htmllib.unescape(a) for a in
re.findall(r'ALT="([^"]+)"', row, re.I)
if "favoris" not in a.lower()]
url = (f"{BASE}/appartement.asp?SID={sid}&CID={cid}&LID={lid}"
f"&Secteur={quote(sector_q)}&Complexe={quote(complex_q)}")
amenities = list(dict.fromkeys(included))
if floor:
amenities.append(f"Étage : {floor}")
# clé de cache des fiches : contenu de la ligne du tableau
# (préfixe v2 = version de l'extraction, pour invalider le cache
# quand le connecteur change)
row_keys[lid] = "v3:" + hashlib.sha1(
" | ".join([unit_raw, avail, floor, price_label] + included)
.encode("utf-8")).hexdigest()
listings[lid] = Listing(
source=self.source_id,
external_id=lid,
url=url,
title=f"{name} — {unit_raw}".strip(" —"),
sector=sector,
city=infer_city(sector, default="Québec"),
unit_type=normalize_unit_type(unit_raw),
price=parse_price(price_label),
price_label=price_label,
availability=avail,
amenities=amenities,
)
# 3) Fiches (avec cache BD) : adresse + no d'unité, lat/lng, superficie,
# images, caractéristiques, services optionnels, téléphone
for lid, lst in listings.items():
url = lst.url
try:
payload = self.detail(
lid, row_keys.get(lid, ""),
lambda u=url: self._fetch_detail(u))
except _DetailBudget:
continue
except Exception:
continue
self._apply_detail(lst, payload)
return list(listings.values())
# -- fiche appartement.asp -------------------------------------------------
def _fetch_detail(self, url: str) -> dict:
if self._detail_fetches >= self.max_details:
raise _DetailBudget(url)
self._detail_fetches += 1
detail = self._get_text(url)
out: dict = {}
# adresse complète avec no d'unité (« 4855, 6e Avenue Ouest # 403 »)
m = _ADDR_UNIT_RE.search(detail)
if m:
addr = _fix_sup(_strip_tags(m.group(1)))
out["address"] = re.sub(r"#\s*", "#", addr)
if not out.get("address"):
# repli : variables JS adr1/adr2 de la carte Google
m = re.search(r'var adr1 = "([^"]*)";\s*var adr2 = "([^"]*)"', detail)
if m and m.group(1):
out["address"] = f"{m.group(1)}, {_fix_sup(_strip_tags(m.group(2)))}".strip(" ,")
# coordonnées de la carte Google (structurées)
m = _LATLNG_RE.search(detail)
if m:
out["lat"], out["lng"] = float(m.group(1)), float(m.group(2))
# superficie : encadré dédié de la fiche (structuré)
m = _SUPERFICIE_RE.search(detail)
if m:
try:
val = float(m.group(1).replace(" ", ""))
if 80 <= val <= 20000:
out["area_sqft"] = val
except ValueError:
pass
# téléphone du bureau de location (lien tel: du formulaire de visite)
m = re.search(r"Pour une visite.*?tel:\+?1?(\d{10})", detail, re.S | re.I)
if m:
d = m.group(1)
out["phone"] = f"{d[0:3]}-{d[3:6]}-{d[6:10]}"
# retirer commentaires et scripts avant les extractions texte
detail = re.sub(r"", " ", detail, flags=re.S)
detail = re.sub(r"
", " ", detail, flags=re.S | re.I)
# images de l'immeuble : .///images/xxx.jpg
imgs = re.findall(
r'(?:\./)?([a-z0-9_\-]+/[a-z0-9_\-]+/images/[^"\'\s]+'
r'\.(?:jpg|jpeg|png|webp))', detail, re.I)
out["images"] = [f"{BASE}/{u}" for u in dict.fromkeys(imgs)][:25]
# caractéristiques de l'appartement : ALT des icônes (« Entrées
# laveuse/sécheuse incluses ») + texte visible (« Balcon 60 pi² »,
# « Un stationnement extérieur »…)
amen: list[str] = []
for c in _CARAC_RE.findall(detail):
amen += [htmllib.unescape(a or b) for a, b in
re.findall(r"(?:ALT|alt)=(?:\"([^\"]{3,60})\"|'([^']{3,60})')", c)]
amen.append(_strip_tags(c))
# caractéristiques de l'immeuble (liste : Bell Fibe, Intercom…)
m = re.search(r"Caract[ée]ristiques immeuble(.*?)(?:Services optionnels|"
r"Les informations)", detail, re.S | re.I)
immeuble_txt = ""
if m:
immeuble_txt = _strip_tags(m.group(1))
amen += [_strip_tags(li) for li in
re.findall(r"(.*?)", m.group(1), re.S | re.I)]
# revêtements de sol (« Revêtements : Bois flottant, céramique… »)
m = re.search(r"Rev[êe]tements?\s*:\s*([^<]{3,120})", detail, re.I)
if m:
amen.append("Revêtements : " + _strip_tags(m.group(1)))
out["amenities"] = [a for a in dict.fromkeys(amen)
if len(a) >= 3 and a.lower() != "superficie"][:25]
# description : présentation du complexe + immeuble + services optionnels
desc_bits = []
m = re.search(r'(.*?)
', detail,
re.S | re.I)
if m and _strip_tags(m.group(1)):
desc_bits.append(_strip_tags(m.group(1)))
if immeuble_txt:
desc_bits.append("Caractéristiques de l'immeuble : " + immeuble_txt)
m = re.search(r"Services optionnels(.*?)Les informations", detail,
re.S | re.I)
if m and _strip_tags(m.group(1)):
desc_bits.append("Services optionnels : " + _strip_tags(m.group(1)))
out["description"] = " — ".join(desc_bits)[:900]
return out
def _apply_detail(self, lst: Listing, payload: dict) -> None:
if not payload:
return
if payload.get("address"):
lst.address = payload["address"]
lst.title = f"{payload['address']} — {lst.unit_type}".strip(" —")
if payload.get("lat") is not None and payload.get("lng") is not None:
lst.lat, lst.lng = payload["lat"], payload["lng"]
if payload.get("area_sqft"):
lst.area_sqft = payload["area_sqft"]
if payload.get("phone"):
lst.details.setdefault("contact", {})["phone"] = payload["phone"]
for a in payload.get("amenities", []):
if a not in lst.amenities:
lst.amenities.append(a)
if payload.get("images"):
lst.images = payload["images"]
if payload.get("description"):
lst.description = payload["description"]