# ----------------------------------------------------------------------------- # Rent-Ka — Rental listings aggregator (Canada, outside Québec) # Author: Simon-Pierre Boucher — contact@spboucher.ai # connectors/cogir.py : Cogir Real Estate (cogir.net) # Server-rendered «gestion-immeubles-residentiels» list page. Rent-Ka keeps # the buildings whose URL slug is OUTSIDE Québec (verified live 2026-08-27: # Toronto 22 + boroughs, London 7, Guelph 3, Ottawa 2, Waterloo, Timmins, # Mississauga in ON; Halifax in NS). Each building page has a # «Modèles disponibles» table (type + starting price) and a photo gallery # (DATA/PHOTO). One listing per unit model, else per building. # Buildings that redirect to an external microsite (608church.com, etc.) # are skipped — only pages served by cogir.net (same French template) are # parsed. Retirement/student residences are excluded (FR + EN patterns). # ----------------------------------------------------------------------------- from __future__ import annotations import os import re from urllib.parse import unquote, urlparse from bs4 import BeautifulSoup from ..schema import Listing, normalize_unit_type, parse_price from .base import BaseConnector def _dedup_address(raw: str) -> str: """Retire les segments dupliqués des adresses Cogir (« 1769 Rue Careau, Québec, QC, Québec, Québec G1M 0E6 »).""" parts, seen = [], set() for seg in (s.strip() for s in (raw or "").split(",")): key = seg.lower() if not seg or key in seen: continue seen.add(key) parts.append(seg) return ", ".join(parts) BASE = "https://www.cogir.net" LIST_URL = f"{BASE}/gestion-immeubles-residentiels.html" # URL slugs outside Québec -> (sector, city, province). Slugs absent from # this map (Montréal, Québec, Laval… and any new QC city) are ignored. _ROC_SLUGS = { # City of Toronto and former boroughs "toronto": ("", "Toronto", "ON"), "scarborough": ("Scarborough", "Toronto", "ON"), "north-york": ("North York", "Toronto", "ON"), "east-york": ("East York", "Toronto", "ON"), "etobicoke": ("Etobicoke", "Toronto", "ON"), # Rest of Ontario "ottawa": ("", "Ottawa", "ON"), "london": ("", "London", "ON"), "guelph": ("", "Guelph", "ON"), "waterloo": ("", "Waterloo", "ON"), "kitchener": ("", "Kitchener", "ON"), "mississauga": ("", "Mississauga", "ON"), "hamilton": ("", "Hamilton", "ON"), "timmins": ("", "Timmins", "ON"), # Atlantic "halifax": ("", "Halifax", "NS"), "dartmouth": ("Dartmouth", "Halifax", "NS"), } # Generic slugs: replaced by a more precise slug when available («toronto»: # some buildings also appear under their borough slug) _GENERIC_SLUGS = {"toronto"} _BUILDING_RE = re.compile( r'href="(immeuble-residentiel-([a-z0-9\-]+)/(\d+)-[^"]+\.html)"') _EXCLUDE_RE = re.compile( r"résidence[s]? pour (aîné|retrait)|résidence[s]? étudiante|" r"stationnement|commercial|rangement|entreposage", re.I) # English variant (pages of ON/NS buildings) _EXCLUDE_EN_RE = re.compile( r"retirement (?:residence|home|living|community)|" r"senior[s']*\s+(?:residence|living|housing)|" r"student (?:residence|housing)|parking|storage", re.I) class CogirConnector(BaseConnector): source_id = "cogir" request_delay = 0.6 max_buildings = 80 # safety cap (~45 non-QC buildings today) max_images = 25 def fetch(self) -> list[Listing]: html = self.get(LIST_URL).text # 1) Buildings outside Québec (deduplicated by numeric id; the most # precise slug wins, e.g. scarborough > toronto) buildings: dict[str, dict] = {} for m in _BUILDING_RE.finditer(html): path, slug, bid = m.group(1), m.group(2), m.group(3) if slug not in _ROC_SLUGS: continue cur = buildings.get(bid) if cur is None or (cur["slug"] in _GENERIC_SLUGS and slug not in _GENERIC_SLUGS): buildings[bid] = {"path": path, "slug": slug, "id": bid} listings: list[Listing] = [] for i, b in enumerate(buildings.values()): if i >= self.max_buildings: break try: prov = _ROC_SLUGS[b["slug"]][2] self._ingest_building(b, prov, listings) except Exception: continue return listings def _ingest_building(self, b: dict, province: str, listings: list[Listing]) -> None: """Parse une fiche immeuble et ajoute ses annonces à `listings`.""" try: # Do not follow redirects — pages redirected to an external # microsite (608church.com, 107redpath.com…) use a different # template and are often 403: skip them. resp = self.get(f"{BASE}/{b['path']}", allow_redirects=False) if (300 <= resp.status_code < 400 or "cogir.net" not in (urlparse(resp.url).hostname or "")): return page = resp.text except Exception: return soup = BeautifulSoup(page, "html.parser") h1 = soup.select_one("h1") name = h1.get_text(strip=True) if h1 else f"Immeuble {b['id']}" addr_el = soup.select_one("p.adresse") address = "" if addr_el: address = addr_el.get_text(" ", strip=True) address = re.sub(r"Coordonnées complètes.*|Visite virtuelle.*", "", address).strip(" »") address = _dedup_address(address) # suffix the province code unless it is already in the address if address and not re.search(rf",?\s+(?:{province})\b", address): address = f"{address}, {province}" # description (+ éventuelle section « Promotion » en tête) desc = "" d_h2 = soup.find("h2", string=re.compile("Description", re.I)) if d_h2 and d_h2.find_parent(): desc = d_h2.find_parent().get_text(" ", strip=True) desc = re.sub(r"^Description\s*", "", desc)[:600] p_h2 = soup.find("h2", string=re.compile(r"^\s*Promotion", re.I)) if p_h2 and p_h2.find_parent(): promo = p_h2.find_parent().get_text(" ", strip=True) promo = re.sub(r"^Promotion\s*", "", promo).strip()[:200] if promo: desc = f"Promotion : {promo}. {desc}".strip()[:700] # exclude retirement/student residences and commercial (FR + EN) if _EXCLUDE_RE.search(name) or _EXCLUDE_RE.search(desc[:200]) \ or _EXCLUDE_EN_RE.search(name) \ or _EXCLUDE_EN_RE.search(desc[:200]): return # commodités amenities: list[str] = [] s_h2 = soup.find("h2", string=re.compile( "Services dans l'immeuble", re.I)) if s_h2: ul = s_h2.find_next("ul") if ul: amenities = [li.get_text(strip=True) for li in ul.select("li")][:20] # photos (relatives DATA/PHOTO/... -> absolues) images: list[str] = [] for im in soup.select("img[src]"): src = im["src"] if "DATA/PHOTO" not in src: continue absu = src if src.startswith("http") else f"{BASE}/{src.lstrip('/')}" if absu not in images: images.append(absu) images = images[: self.max_images] # contact de l'immeuble (liens tel:/mailto:, hors pied de page # corporatif info@cogir.net / 1-866-671-6381) contact: dict = {} for a in soup.select('a[href^="tel:"]'): digits = re.sub(r"\D", "", a.get("href", ""))[-10:] if len(digits) == 10 and not digits.startswith("866"): contact["phone"] = (f"{digits[:3]}-{digits[3:6]}-" f"{digits[6:]}") break for a in soup.select('a[href^="mailto:"]'): email = unquote(a["href"].removeprefix("mailto:")).strip() if email and email.lower() != "info@cogir.net": contact["email"] = email break details_extra = {"contact": contact} if contact else {} sector, city, _prov = _ROC_SLUGS[b["slug"]] url = f"{BASE}/{b['path']}" # 2) Une annonce par modèle du tableau « Modèles disponibles » # (#tableModele : colType / colModele / colPrix). Les modèles # marqués « Pas disponible » ou « Liste d'attente » sont exclus. rows = [] table = soup.select_one("table#tableModele") or soup.find("table") for tr in table.select("tbody tr") if table else []: typ_el = tr.select_one("td.colType") mod_el = tr.select_one("td.colModele") prix_el = tr.select_one("td.colPrix") tds = tr.select("td") typ = (typ_el.get_text(" ", strip=True) if typ_el else (tds[0].get_text(" ", strip=True) if tds else "")) modele = mod_el.get_text(" ", strip=True) if mod_el else "" # certains immeubles mettent « À partir de » dans colModele if re.fullmatch(r"à partir de\s*", modele, re.I): modele = "" if prix_el is not None: price_txt = prix_el.get_text(" ", strip=True) else: price_txt = next((td.get_text(" ", strip=True) for td in tds[1:] if "$" in td.get_text()), "") if not typ or not re.match( r"^\d\s*1/2|^\d\s*½|^Studio|^Loft|^Penthouse", typ, re.I): continue if re.search(r"pas disponible|liste d'attente", price_txt, re.I): continue # modèle affiché mais non offert price_txt = re.sub(r"à partir de", "", price_txt, flags=re.I) rows.append((typ, modele, price_txt.strip())) if rows: seen_ext: set[str] = set() for typ, modele, price_txt in rows: ut = normalize_unit_type(typ) ext = f"{b['id']}-{re.sub(r'[^a-z0-9]+', '', ut.lower()) or 'u'}" if ext in seen_ext: # 2e modèle du même type (ex. « 3 1/2 + den ») : # suffixe du nom de modèle pour un uid unique suffix = re.sub(r"[^a-z0-9]+", "-", modele.lower()).strip("-") ext = f"{ext}-{suffix or len(seen_ext)}" if ext in seen_ext: continue seen_ext.add(ext) label = (f"{ut} ({modele})" if modele and modele.lower() != typ.lower() else ut) listings.append(Listing( source=self.source_id, external_id=ext, url=url, title=f"{name} — {label}", address=address, sector=sector, city=city, province=province, unit_type=ut, price=parse_price(price_txt), price_label=(f"À partir de {price_txt}" if price_txt else ""), availability="Disponible" if price_txt else "", description=desc, amenities=amenities, details=dict(details_extra), images=images, )) else: listings.append(Listing( source=self.source_id, external_id=b["id"], url=url, title=name, address=address, sector=sector, city=city, province=province, unit_type="", price=None, price_label="", availability="", description=desc, amenities=amenities, details=dict(details_extra), images=images, ))