')
# fiche = 1er lien de la carte finissant par -
/ ; le chemin varie selon
# le site (/listing/, /listings/, /all-regional-listings/…)
_LINK_RE = re.compile(r'href="(https?://[^"]+?-(\d{6,10})/?)"')
_RIBBON_RE = re.compile(r'rps-ribbon[^>]*>\s*([^<]+?)\s*<')
_PRICE_RE = re.compile(r'rps-price[^>]*>\s*\$\s*([\d,]+)')
_H4_RE = re.compile(r"\s*(.*?)\s*
", re.S)
# avec ou sans selon le thème du site
_CITY_RE = re.compile(r'city-province-postalcode[^>]*>\s*(?:\s*)?([^<]+?)\s*<', re.S)
_FEAT_RE = re.compile(r'rps-result-feature-label[^>]*>\s*([^<]+?)\s*<')
_CARD_BROKER_RE = re.compile(r'text-muted[^>]*>\s*\s*([^<]+?)\s*(?:)', re.S)
_DDFIMG_RE = re.compile(r'https://ddfcdn\.realtor\.ca/[^")\'\s\\]+')
_ROW_RE = re.compile(r"]*>\s*([^<]{2,45})\s*\s*"
r"]*>(.*?)", re.S)
_DESC_RE = re.compile(r'\s*]*>(.*?)', re.S)
_DESC_RE2 = re.compile(r']*>(.*?)
', re.S)
# ville depuis « 3383 Romeo Street, Greater Sudbury (Valley East), Ontario … »
_TITLE_CITY_RE = re.compile(r",\s*([^,<>|]{2,60}),\s*Ontario\b")
_PRICING_RE = re.compile(r'rps-pricing[^>]*>\s*\$\s*([\d,]+)')
_ID_TAIL_RE = re.compile(r"\s*\(id:\d{4,9}\)\s*$")
_TAG_RE = re.compile(r"<[^>]+>")
_NUM_RE = re.compile(r"[\d,]+(?:\.\d+)?")
_YEAR_RE = re.compile(r"\b(1[6-9]\d{2}|20\d{2})\b")
# territoire couvert (Québec + Ontario) — même boîte que schema.finalize()
_BBOX = (41.6, 63.0, -95.5, -56.0)
def _num(s: str) -> float | None:
m = _NUM_RE.search(s or "")
if not m:
return None
try:
return float(m.group(0).replace(",", ""))
except ValueError:
return None
class _RealtyPress(BaseConnector):
"""Connecteur générique de site RealtyPress (voir data/ontario_agencies.json)."""
agency_name = ""
site_url = ""
archive = "listing" # chemin de l'archive (revelrealty: "listings",
# codygroup: "all-regional-listings")
max_pages = 150 # 100 cartes/page → jusqu'à 15 000 fiches par site
request_delay = 0.6
def fetch(self) -> list[PropertyListing]:
# mur CREA des fiches détail : le cookie suffit (posé pour tout domaine,
# les redirections www/apex restent couvertes)
self.session.cookies.set("disclaimer", "accepted")
by_id: dict[str, PropertyListing] = {}
base = self.site_url.rstrip("/")
dry = 0
for page in range(1, self.max_pages + 1):
url = f"{base}/{self.archive}/page/{page}/?posts_per_page=100"
try:
body = self.get(url).text
except Exception:
break
cards = self._cards(body)
if not cards:
break
before = len(by_id)
for card in cards:
self._parse_card(card, by_id)
dry = dry + 1 if len(by_id) == before else 0
if dry >= 2:
break
listings = list(by_id.values())
du.enrich(self, listings, DETAIL_LIMIT, parse_rp_detail, key="v1")
for lst in listings:
# n° MLS du board (fiche détail) — utile à la dédup inter-plateformes
if not lst.mls and lst.details.get("MLS® Number"):
lst.mls = str(lst.details["MLS® Number"])
if not lst.title:
lst.title = ", ".join(filter(None, (lst.address, lst.city))) \
or "Propriété à vendre"
return listings
def _cards(self, body: str) -> list[str]:
marks = list(_CARD_RE.finditer(body))
return [body[m.start():(marks[i + 1].start() if i + 1 < len(marks)
else m.start() + 6000)]
for i, m in enumerate(marks)]
def _parse_card(self, card: str, by_id: dict) -> None:
ml = _LINK_RE.search(card)
if not ml:
return
url, ddf = ml.group(1), ml.group(2)
eid = f"ddf{ddf}"
if eid in by_id:
return
mr = _RIBBON_RE.search(card)
ribbon = (mr.group(1) if mr else "").strip().lower()
if "rent" in ribbon or "lease" in ribbon:
return # locations : hors périmètre
lst = PropertyListing(source=self.source_id, external_id=eid, url=url,
region="Ontario", agency=self.agency_name,
broker_name=self.agency_name)
ma = _H4_RE.search(card)
if ma:
lst.address = _html.unescape(_TAG_RE.sub(" ", ma.group(1))).strip()
mc = _CITY_RE.search(card)
if mc:
city = _html.unescape(mc.group(1)).strip().rstrip(",")
city = re.sub(r",?\s*Ontario\b.*$", "", city, flags=re.I)
lst.city = city.split("(")[0].strip()
mp = _PRICE_RE.search(card)
if mp:
lst.price = _num(mp.group(1))
lst.price_label = f"{mp.group(1)} $"
for feat in _FEAT_RE.findall(card):
f = _html.unescape(feat).strip()
low = f.lower()
n = _num(f)
if not n:
continue
if "bedroom" in low:
lst.bedrooms = int(n)
elif "bathroom" in low:
lst.bathrooms = int(n)
elif "sqft" in low or "sq ft" in low or "ft" in low:
lst.area_sqft = n # plage « 1,100 - 1,500 ft² » : borne basse
mbk = _CARD_BROKER_RE.search(card)
if mbk:
lst.broker_name = _html.unescape(mbk.group(1)).strip()[:120]
mi = _DDFIMG_RE.search(card)
if mi:
lst.images = [mi.group(0)]
by_id[lst.external_id] = lst
def parse_rp_detail(html: str) -> dict:
"""Fiche RealtyPress : tableaux DDF, description, GPS, galerie, courtier."""
out: dict = {}
details: dict = {}
for lab, val in _ROW_RE.findall(html):
label = _html.unescape(lab).strip().rstrip(":")
value = re.sub(r"\s+", " ", _html.unescape(_TAG_RE.sub(" ", val))).strip()
if label and value and len(value) <= 300:
details.setdefault(label, value)
def dv(*labels: str) -> str:
for lb in labels:
if details.get(lb):
return details[lb]
return ""
b = _num(dv("Bedrooms Total", "Bedrooms", "Bedrooms Above Ground"))
if b is not None and 0 < b <= 30:
out["bedrooms"] = int(b)
b = _num(dv("Bathroom Total", "Bathrooms"))
if b is not None and 0 < b <= 30:
out["bathrooms"] = int(b)
b = _num(dv("Half Bath Total"))
if b is not None and 0 < b <= 10:
out["powder_rooms"] = int(b)
my = _YEAR_RE.search(dv("Constructed Date", "Construction Year", "Age"))
if my:
out["year_built"] = int(my.group(1))
si = dv("Size Interior")
if si and "sqft" in si.lower().replace(" ", ""):
a = _num(si) # « 7,901 Sqft » / « 1200 - 1399 sqft »
if a and a >= 100:
out["area_sqft"] = a
pt = dv("Property Type", "Building Type", "Type")
if pt:
out["property_type"] = pt # anglais DDF — normalisé par finalize()
sec = dv("Neigbourhood", "Neighbourhood", "Community Name")
if sec:
out["sector"] = sec
mp = _PRICING_RE.search(html)
if mp:
out["price"] = _num(mp.group(1))
out["price_label"] = f"{mp.group(1)} $"
md = _DESC_RE.search(html) or _DESC_RE2.search(html)
if md:
desc = _html.unescape(_TAG_RE.sub(" ", md.group(1)))
desc = re.sub(r"\s+", " ", desc).strip()
out["description"] = _ID_TAIL_RE.sub("", desc)[:6000]
mt = re.search(r"<title>(.*?)", html, re.S)
if mt:
mc = _TITLE_CITY_RE.search(_html.unescape(mt.group(1)))
if mc:
# « Greater Sudbury (Valley East) » : le secteur part dans sector
city = mc.group(1).split("(")[0].strip()
if city and not any(c.isdigit() for c in city):
out["city"] = city
msec = re.search(r"\(([^)]{2,45})\)", mc.group(1))
if msec and "sector" not in out:
out["sector"] = msec.group(1).strip()
for n in du.ld_nodes(html):
t = n.get("@type")
types = set(t if isinstance(t, list) else [t])
geo = n.get("geo") or {}
if isinstance(geo, dict) and "lat" not in out:
try:
lat, lng = float(geo["latitude"]), float(geo["longitude"])
if _BBOX[0] <= lat <= _BBOX[1] and _BBOX[2] <= lng <= _BBOX[3]:
out["lat"], out["lng"] = lat, lng
except (KeyError, TypeError, ValueError):
pass
if types & {"RealEstateAgent", "Organization"}:
name = str(n.get("name") or "").strip()
if name and "broker_name" not in out:
out["broker_name"] = name[:120]
tel = str(n.get("telephone") or "").strip()
if tel and "broker_phone" not in out:
out["broker_phone"] = tel[:40]
if "lat" not in out:
m = re.search(r'"latitude"\s*:\s*"?(-?\d{1,2}\.\d{3,})"?\s*,\s*'
r'"longitude"\s*:\s*"?(-?\d{2,3}\.\d{3,})"?', html)
if m:
lat, lng = float(m.group(1)), float(m.group(2))
if _BBOX[0] <= lat <= _BBOX[1] and _BBOX[2] <= lng <= _BBOX[3]:
out["lat"], out["lng"] = lat, lng
gal = [u for u in dict.fromkeys(_DDFIMG_RE.findall(html))
if "/listings/" in u.lower()]
if gal:
out["images"] = gal[:60]
if details:
out["details"] = details
return out
def _load() -> list[dict]:
try:
return json.loads(REGISTRY.read_text(encoding="utf-8"))
except Exception:
return []
# PAUSE ONTARIO (2026-08-27) : l'expansion ON est suspendue — les classes ne
# s'enregistrent dans le registre CONNECTORS que si IMMOKA_ONTARIO=1 est posé
# (.env). Les annonces déjà en base sont conservées mais dépubliées par
# quality.refresh (raison « pause-ontario »). Rien n'est effacé.
_ONTARIO = os.environ.get("IMMOKA_ONTARIO") == "1"
# Génère une classe par site du registre.
for _ag in (_load() if _ONTARIO else []):
if not all(_ag.get(k) for k in ("id", "site")):
continue
_sid = _ag["id"]
globals()[f"REALTYPRESS_{_sid.upper()}"] = type(
"RealtyPress" + "".join(p.title() for p in _sid.split("_")),
(_RealtyPress,),
{
"source_id": _sid,
"site_url": _ag["site"],
"agency_name": _ag.get("name", _sid),
"archive": _ag.get("archive", "listing"),
"max_pages": int(_ag.get("max_pages", 150)),
},
)