.*?

([^<]+)

', re.S | re.I) _CHAR_ROW_RE = re.compile(r']*>(.*?)\s*]*>(.*?)', re.S | re.I) _FEAT_RE = re.compile(r'

([^<]+)

\s*

([^<]*)

', re.I) _LATLNG_RE = re.compile(r'google\.maps\.LatLng$\s*(-?\d{1,2}\.\d+)\s*,\s*(-?\d{2,3}\.\d+)\s*$') _AGENT_RE = re.compile(r'class="agents">.*?

([^<]+)

(.*?)', re.S | re.I) _TEL_RE = re.compile(r'href="tel:(\+?\d{7,11})"') _IFRAME_RE = re.compile(r']+src="([^"]+)"', re.I) # libellés du bloc « features » promus en colonnes ; le reste va dans details _FEAT_FIELDS = { "Année de construction": "year_built", "Chambre(s) à coucher": "bedrooms", "Salle(s) de bain": "bathrooms", "Salle(s) d'eau": "powder_rooms", } class BhhsQuebecConnector(BaseConnector): source_id = "bhhs_ag_qc" request_delay = 0.4 max_pages = 40 def fetch(self) -> list[PropertyListing]: by_id: dict[str, PropertyListing] = {} dry = 0 for page in range(1, self.max_pages + 1): url = LISTING if page == 1 else f"{LISTING}?pages={page}" try: html = self.get(url).text except Exception: break before = len(by_id) parts = _CARD_SPLIT.split(html) # parts = [pre, url1, mls1, blk1, url2, mls2, blk2, …] for i in range(1, len(parts) - 2, 3): url_p, mls, blk = parts[i], parts[i + 1], parts[i + 2][:1800] lst = self._card(url_p, mls, blk) if lst: by_id.setdefault(lst.external_id, lst) 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_detail, key="v1") return listings def _card(self, url_p: str, mls: str, blk: str) -> PropertyListing | None: addr, city = "", "" am = _ADDR_RE.search(blk) if am: raw = _html.unescape(am.group(1)).strip() if "," in raw: addr, city = [x.strip() for x in raw.rsplit(",", 1)] else: addr = raw if not addr: um = re.search(r'/properties/([^/]+)/\d{6,9}', url_p) if um: addr = _html.unescape(urllib.parse.unquote_plus(um.group(1))).strip() rm = _REGION_RE.search(blk) region = _html.unescape(rm.group(1)).strip() if rm else "" pm = _PRICE_RE.search(blk) price_label = pm.group(1).strip() if pm else "" price = parse_price(price_label.replace("$", "").replace(",", "")) \ if price_label.startswith("$") else parse_price(price_label) amn = _AMEN_RE.search(blk) im = _IMG_RE.search(blk) return PropertyListing( source=self.source_id, external_id=mls, url=url_p, title=addr or "Propriété à vendre", address=addr, city=city, region=region, price=price, price_label=price_label, bedrooms=int(amn.group(1)) if amn else None, bathrooms=int(amn.group(2)) if amn else None, mls=mls, images=[im.group(1)] if im else [], agency=AGENCY, broker_name=AGENCY, ) def _txt(fragment: str) -> str: t = _html.unescape(re.sub(r"", "\n", fragment, flags=re.I)) t = re.sub(r"<[^>]+>", " ", t).replace("\xa0", " ") t = re.sub(r"[ \t]+", " ", t) return re.sub(r"\n{3,}", "\n\n", t).strip() def _parse_detail(html: str) -> dict: """Fiche détail BHHS (MW Properties) : description+addenda, features, caractéristiques, GPS, visite virtuelle, courtier, galerie complète.""" out: dict = {} # description :

du bloc description + addenda (collapse #showAddendum) i = html.find('class="description block') if i >= 0: j = html.find('id="addendum-btn"', i) if j < 0: j = html.find('class="char block"', i) seg = html[i:j if j > 0 else i + 30000] paras = [_txt(p) for p in _P_RE.findall(seg)] desc = "\n\n".join(p for p in paras if p) if desc: out["description"] = desc details: dict = {} # bloc « features » (année, pièces, chambres, sdb…) for lab, val in _FEAT_RE.findall(html): lab, val = _txt(lab), _txt(val) if not val: continue field = _FEAT_FIELDS.get(lab) if field: m = re.search(r"\d+", val) if m: out[field] = int(m.group()) elif len(val) <= 120: details[lab] = val # tableau « Caractéristiques » (th/td) i = html.find('id="charateristics"') if i >= 0: seg = html[i:html.find("", i)] for th, td in _CHAR_ROW_RE.findall(seg): lab, val = _txt(th), _txt(td) if lab and val and len(lab) <= 60 and len(val) <= 250: details[lab] = val # GPS inline m = _LATLNG_RE.search(html) if m: lat, lng = float(m.group(1)), float(m.group(2)) if 44.5 <= lat <= 63.0 and -80.0 <= lng <= -56.0: out["lat"], out["lng"] = lat, lng # visite virtuelle (iframe du collapse dédié) i = html.find('id="virtualCollapse"') if i >= 0: m = _IFRAME_RE.search(html[i:i + 3000]) if m and m.group(1).startswith("http"): details["Visite virtuelle"] = m.group(1) # courtier inscripteur (carte .agents) m = _AGENT_RE.search(html) if m: out["broker_name"] = _txt(m.group(1)) t = _TEL_RE.search(m.group(2)) if t: tel = t.group(1).lstrip("+") if len(tel) == 10: tel = f"({tel[:3]}) {tel[3:6]}-{tel[6:]}" out["broker_phone"] = tel if details: out["details"] = details # galerie pleine résolution (sans /thumbs-320/), ordre d'origine seen, uniq = set(), [] for u in _GALLERY_RE.findall(html): if u not in seen: seen.add(u) uniq.append(u) if not uniq: # toutes les fiches BHHS ont une galerie (flux Centris) ; une page sans # galerie = rendu transitoirement dégradé (l'API images du plugin MW a # flanché) → échec de parse : on ne met PAS en cache un payload sans # photos, la fiche sera réessayée au prochain cycle (du.enrich garde # l'éventuel payload périmé en attendant) return {} out["images"] = uniq return out