# ----------------------------------------------------------------------------- # Immo-Ka — Agrégateur de maisons à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/avecuncourtier.py : portail « avec un courtier » de la Chambre # immobilière de l'Outaouais (avecuncourtier.com). Couvre TOUT l'Outaouais # (Gatineau → Maniwaki/Pontiac) d'un coup — c'est le portail MLS régional de # la chambre, pas une agence unique. Recherche rendue SERVEUR (plateforme # Concrete5, Tonik Web Studio), paginée par ?page=N (12 cartes/page, ~341 # pages). Chaque carte porte déjà : n° MLS (Centris), adresse, ville/secteur, # type, aire habitable, prix et photo. # # ENRICHISSEMENT DÉTAIL (v2) : /fr/proprietes/details/{mls}-{slug} est rendue # serveur : galerie complète (/centris/photos/{mls}/{n}.jpg, ordre d'origine), # description + Addendas, GPS (data-lat/data-lon), accordéons « Proximité », # « Dimension des pièces » (4 col.), « Taxes et dépenses » + « Bâtiment et # intérieur » (paires td/td : chambres, sdb, aire habitable en m², terrain, # année, évaluations…), visites libres, courtier(s) inscripteur(s) + agence # réelle (RE/MAX Vision, Exit…). Cache BD via _detailutil.enrich. # # source_id « cio_ag_outaouais » : l'infixe _ag_ active la dédup Centris # (db.refresh_dedup, sources %_ag_%). Comme le portail agrège les inscriptions # des bannières déjà couvertes (RE/MAX, Royal LePage, Via Capitale, Proprio # Direct…), la dédup masque les fiches dont le n° Centris est déjà porté par # une source canonique ; seules les inscriptions d'agences NON couvertes de # l'Outaouais restent visibles. Aucun double-comptage. # ----------------------------------------------------------------------------- from __future__ import annotations import html as _html import os import re from .base import BaseConnector from . import _detailutil as du from ..normalize import parse_price from ..schema import PropertyListing SITE = "https://avecuncourtier.com" LISTING = SITE + "/fr/proprietes" MAX_PAGES = int(os.environ.get("IMMOKA_AUC_MAX_PAGES", "400")) AGENCY = "Chambre immobilière de l'Outaouais" # fiches détail (re)lues au plus par cycle — cache BD (key v2). ~1 400 fiches # publiées → ~5 cycles de rattrapage, puis seules les nouvelles sont lues. DETAIL_LIMIT = int(os.environ.get("IMMOKA_AUC_DETAIL_LIMIT", os.environ.get("IMMOKA_DETAIL_LIMIT", "300"))) # une carte commence par
et court jusqu'à la fermeture du # bloc listing suivant ; on la borne au
suivant. _CARD_SPLIT = re.compile(r'
') _HREF_RE = re.compile(r'/fr/proprietes/details/(\d{6,9})-([a-z0-9-]+)', re.I) _ALT_RE = re.compile(r']+alt="([^"]*)"', re.I) _IMG_RE = re.compile(r']+src="([^"]+)"', re.I) _MLS_RE = re.compile(r'MLS[^0-9]{0,3}(\d{6,9})', re.I) _PRICE_RE = re.compile(r'class="price">\s*([\d\s ,.]+\$[^<]*)<', re.I) _ADDR_RE = re.compile(r'
\s*([^<]*)\s*(?:)?\s*([^<]*)
', re.S | re.I) _TYPE_RE = re.compile(r'header">\s*Type\s*
\s*
\s*([^<]+?)\s*
', re.S | re.I) class AvecUnCourtierConnector(BaseConnector): source_id = "cio_ag_outaouais" request_delay = 0.4 max_pages = MAX_PAGES def fetch(self) -> list[PropertyListing]: by_id: dict[str, PropertyListing] = {} dry = 0 for page in range(1, self.max_pages + 1): url = f"{LISTING}?page={page}" try: html = self.get(url).text except Exception: break before = len(by_id) for blk in _CARD_SPLIT.split(html)[1:]: blk = blk[:3000] lst = self._card(blk) if lst: by_id.setdefault(lst.external_id, lst) # fin de pagination : plus aucune carte neuve sur 2 pages de suite dry = dry + 1 if len(by_id) == before else 0 if dry >= 2: break listings = list(by_id.values()) # fiche détail (server-rendered) : galerie complète, description + # addendas, GPS, pièces, taxes/évaluations, courtier + agence réelle. du.enrich(self, listings, DETAIL_LIMIT, _parse_auc_detail, key="v2") for lst in listings: office = lst.details.pop("__agency", "") if office: lst.agency = office return listings def _card(self, blk: str) -> PropertyListing | None: hm = _HREF_RE.search(blk) if not hm: return None mls = hm.group(1) addr, city = "", "" am = _ADDR_RE.search(blk) if am: addr = _html.unescape(am.group(1)).strip() city = _html.unescape(re.sub(r'\s+', ' ', am.group(2))).strip() if not addr: alt = _ALT_RE.search(blk) addr = _html.unescape(alt.group(1)).strip() if alt else "" pm = _PRICE_RE.search(blk) price_label = _html.unescape(pm.group(1)).replace(" ", " ").strip() if pm else "" tm = _TYPE_RE.search(blk) prop_type = _html.unescape(tm.group(1)).strip() if tm else "" im = _IMG_RE.search(blk) img = im.group(1) if im else "" if img and img.startswith("/"): img = SITE + img return PropertyListing( source=self.source_id, external_id=mls, url=f"{SITE}/fr/proprietes/details/{mls}-{hm.group(2)}", title=addr or f"{prop_type} à vendre", address=addr, city=city, property_type=prop_type, price=parse_price(price_label), price_label=price_label, mls=mls, images=[img] if img else [], agency=AGENCY, ) # --- fiche détail (Concrete5 / Tonik, server-rendered) ------------------------ _GAL_RE = re.compile(r'data-lazy="(/centris/photos/\d+/[^"]+)"') _COORD_RE = re.compile(r'data-lat="(-?\d{1,2}\.\d+)"\s+data-lon="(-?\d{2,3}\.\d+)"') _DESC_RE = re.compile(r'class="property-description">\s*

[^<]*

(.*?)
', re.S) _OPEN_RE = re.compile(r'class="open-houses">\s*

[^<]*

(.*?)
', re.S) _ITEM_TITLE_RE = re.compile(r'>\s*([^<]{3,60}?)\s*([^<]+)<') _AGENT_TEL_RE = re.compile(r'href="tel:[^"]*"[^>]*>([^<]+)<') #
NOM INC.
RE/MAX VISION
Agence Immobilière
_AGENT_BANNER_RE = re.compile(r'class="agency">[^<]*
([^<]+)
', re.S) _TOUR_RE = re.compile(r'https?://(?:my\.matterport\.com/show/[^"\'\s<>]+' r'|(?:www\.)?youtube\.com/(?:embed/|watch\?v=)[^"\'\s<>]+' r'|youtu\.be/[^"\'\s<>]+' r'|(?:player\.)?vimeo\.com/(?:video/)?\d+[^"\'\s<>]*)', re.I) # libellés « Bâtiment et intérieur » promus en colonnes du schéma _INT_FIELDS = {"Chambres": "bedrooms", "Salles de bain": "bathrooms", "Salles d'eau": "powder_rooms"} def _clean(s: str) -> str: return re.sub(r'\s+', ' ', _html.unescape(s)).strip() def _to_sqft(val: str) -> float | None: """« 128.8 Mètres carrés » / « 1 200 Pieds carrés » / « 305.4 » (m² implicite, unités Centris métriques sur ce portail) -> pi².""" m = re.search(r'([\d][\d\s ,.]*)', val or "") if not m: return None raw = m.group(1).replace(" ", "").replace(" ", "").replace(" ", "") if raw.count(",") == 1 and "." not in raw: raw = raw.replace(",", ".") else: raw = raw.replace(",", "") try: v = float(raw) except ValueError: return None if v <= 0: return None if re.search(r'pied|pi2|pi²|pc\b', val, re.I): return round(v) return round(v * 10.7639) # métrique (défaut du portail) def _parse_auc_detail(html: str) -> dict: """Fiche avecuncourtier.com : galerie, description+addendas, GPS, accordéons (pièces, taxes, bâtiment), courtiers + agence réelle.""" out: dict = {} details: dict = {} feats: list[str] = [] # galerie : slider principal seulement (avant les vignettes dupliquées) main = html.split('gallery-slider__thumbnails')[0] imgs = list(dict.fromkeys(SITE + u for u in _GAL_RE.findall(main))) if len(imgs) > 1: out["images"] = imgs[:80] mc = _COORD_RE.search(html) if mc: lat, lng = float(mc.group(1)), float(mc.group(2)) if 44.5 <= lat <= 63.0 and -80.0 <= lng <= -56.0: out["lat"], out["lng"] = lat, lng desc, addendum = "", "" md = _DESC_RE.search(html) if md: desc = _clean(re.sub(r'<[^>]+>', ' ', md.group(1))) mo = _OPEN_RE.search(html) if mo: oh = _clean(re.sub(r'<[^>]+>', ' ', mo.group(1))) if oh: details["Visites libres"] = oh[:200] # accordéons « Détails » for item in re.split(r'
', html)[1:]: tm = _ITEM_TITLE_RE.search(item) title = _clean(tm.group(1)) if tm else "" if title == "Proximité": mp = _PROX_RE.search(item) if mp: prox = _clean(re.sub(r'<[^>]+>', ' ', mp.group(1))) if prox: details["Proximité"] = prox[:300] feats.append(f"Proximité : {prox}"[:300]) elif title == "Dimension des pièces": rooms = _ROOM_RE.findall(item) for name, lvl, dim, floor in rooms: name, lvl = _clean(name), _clean(lvl) dim, floor = _clean(dim), _clean(floor) line = name + (f" — {dim}" if dim else "") + (f", {lvl}" if lvl else "") if floor: line += f" ({floor})" feats.append(line) elif title == "Addendas": ma = _ADD_RE.search(item) if ma: # les sauts de ligne du texte Centris sont préservés dans le addendum = _html.unescape(ma.group(1)).strip() else: # Taxes et dépenses / Bâtiment… for label, val in _PAIR_RE.findall(item): label, val = _clean(label).rstrip(":"), _clean(val) if not label or not val or val in ("0", "-"): continue fld = _INT_FIELDS.get(label) if fld: mi = re.search(r'\d+', val) if mi: out[fld] = int(mi.group()) elif label == "Aire habitable": a = _to_sqft(val) if a: out["area_sqft"] = a elif label == "Superficie du terrain": t = _to_sqft(val) if t: out["lot_sqft"] = t val = f"{val} m²" if re.fullmatch(r'[\d\s,.]+', val) else val elif label == "Année de construction": my = re.search(r'\b((?:1[6-9]|20)\d{2})\b', val) if my: out["year_built"] = int(my.group(1)) if label in details and val not in details[label]: details[label] = f"{details[label]}, {val}"[:300] else: details[label] = val[:300] if desc or addendum: out["description"] = "\n\n".join(filter(None, (desc, addendum)))[:6000] if feats: out["features"] = feats[:40] # courtier(s) inscripteur(s) + bannière réelle (encadré agents-wrapper) : # premier agent = inscripteur principal i = html.find('class="agents-wrapper"') if i != -1: j = html.find('calculator-wrapper', i) seg = html[i:j if j != -1 else i + 5000] mn = _AGENT_NAME_RE.search(seg) if mn: out["broker_name"] = _clean(mn.group(1)) mt = _AGENT_TEL_RE.search(seg) if mt: out["broker_phone"] = _clean(mt.group(1)) mb = _AGENT_BANNER_RE.search(seg) if mb: banner = _clean(mb.group(1)) if banner and banner.lower() != "agence immobilière": details["__agency"] = banner.title() if banner.isupper() else banner # visite virtuelle / vidéo (hors liens de pied de page /channel/) mtou = _TOUR_RE.search(html) if mtou: details.setdefault("Visite virtuelle / vidéo", mtou.group(0)) if details: out["details"] = details return out