# ----------------------------------------------------------------------------- # Rent-Ka — Rental listings aggregator (Canada, outside Québec) # Author: Simon-Pierre Boucher — contact@spboucher.ai # connectors/fb_marketplace.py : Facebook Marketplace — catégorie location # (propertyrentals), TOUTE la province. Annonces de particuliers, fort # complément aux gestionnaires. Depuis 2026-08-22 le scraping est délégué à # l'ACTEUR APIFY MAISON gorgeous_thistle/ka-fb-marketplace (source : # actors/ka-fb-marketplace de ce repo) : HTML public déconnecté SANS rendu # JS (les JSON de recherche, la fiche PDP et la galerie MediaViewer # préchargée vivent dans le HTML brut) via proxy résidentiel CA — ~100× # moins cher que l'ancienne chaîne Scrapfly ASP+render_js, et couverture # élargie de 10 à 33 villes (IDs Marketplace des villes régionales validés ; # les slugs inventés retombent sur des villes aléatoires, ne pas en ajouter # sans les valider). # Le scroll infini reste bloqué hors connexion (~25 annonces/URL) : on # pagine par TRANCHES DE PRIX et on accumule sur plusieurs synchronisations. # Le flux public TOURNE : une phase de RATTRAPAGE (extraDetailIds) visite # les fiches actives jamais enrichies (budget RENTKA_FBMP_BACKFILL). # Le connecteur garde le cache détail BD (clé v2, payloads compatibles) et # passe à l'acteur la liste des fiches fraîches à NE PAS revisiter. # ⚠️ Données personnelles (Loi 25) : ne jamais republier nom/téléphone du # vendeur, garder le lien sortant. # ----------------------------------------------------------------------------- from __future__ import annotations import json import os import re import time import requests as _requests from ..schema import Listing, normalize_unit_type from .base import BaseConnector from . import _detailutil as du BASE = "https://www.facebook.com/marketplace" APIFY_API = "https://api.apify.com/v2" ACTOR = os.environ.get("RENTKA_FBMP_ACTOR", "gorgeous_thistle~ka-fb-marketplace") # Canadian cities outside Québec — large inventories: Marketplace vanity # slug -> display name. Only UNAMBIGUOUS slugs (e.g. «london» resolves to # London UK on Facebook — use a verified numeric REGIONS id instead). CITIES = { "toronto": "Toronto", "ottawa": "Ottawa", "mississauga": "Mississauga", "brampton": "Brampton", "vancouver": "Vancouver", "calgary": "Calgary", "edmonton": "Edmonton", "winnipeg": "Winnipeg", "saskatoon": "Saskatoon", "regina": "Regina", "halifax": "Halifax", } # city -> province for the searched cities (province of each listing) _CITY_PROV = { "Toronto": "ON", "Ottawa": "ON", "Mississauga": "ON", "Brampton": "ON", "Vancouver": "BC", "Calgary": "AB", "Edmonton": "AB", "Winnipeg": "MB", "Saskatoon": "SK", "Regina": "SK", "Halifax": "NS", } # Facebook state labels -> province code _STATE_PROV = { "on": "ON", "ontario": "ON", "bc": "BC", "british columbia": "BC", "ab": "AB", "alberta": "AB", "sk": "SK", "saskatchewan": "SK", "mb": "MB", "manitoba": "MB", "ns": "NS", "nova scotia": "NS", "nb": "NB", "new brunswick": "NB", "pe": "PE", "prince edward island": "PE", "nl": "NL", "newfoundland": "NL", "newfoundland and labrador": "NL", "yt": "YT", "nt": "NT", "nu": "NU", } # regional cities: Marketplace numeric ID -> name (to be filled with # VERIFIED ids for London ON, Kitchener, Hamilton, Victoria, Moncton… # — vanity slugs are ambiguous for those) REGIONS: dict[str, str] = {} # tranches de prix (bornes en $) : le flux déconnecté sert un sous-ensemble # tournant, chaque tranche renvoie un lot quasi disjoint PRICE_BANDS = [(0, 800), (800, 1100), (1100, 1400), (1400, 1700), (1700, 2100), (2100, 2800), (2800, 6000)] REGIONAL_BANDS = [(0, 1200), (1200, 6000)] # parcs plus petits : 2 tranches # garde-fous de prix mensuel (rejette « $90 » = /nuit, et les valeurs à vendre) PRICE_MIN, PRICE_MAX = 300, 12000 DETAIL_LIMIT = int(os.environ.get("RENTKA_FBMP_DETAIL_LIMIT", "120")) # rattrapage : fiches ACTIVES en BD encore pauvres (sans description, sans # GPS ou avec ≤ 1 image) qui ne repassent plus dans la recherche publique BACKFILL_LIMIT = int(os.environ.get("RENTKA_FBMP_BACKFILL", "80")) # les annonces FB changent peu après publication : TTL long = le budget détail # sert surtout aux NOUVELLES annonces plutôt qu'à re-visiter les connues TTL_DAYS = float(os.environ.get("RENTKA_FBMP_TTL_DAYS", "30")) CITY_LIMIT = os.environ.get("RENTKA_FBMP_CITIES", "") # ex. "montreal,laval" CONCURRENCY = int(os.environ.get("RENTKA_FBMP_CONCURRENCY", "8")) RUN_TIMEOUT = int(os.environ.get("RENTKA_FBMP_RUN_TIMEOUT", "2400")) # s # clé du cache détail — v2 : payloads identiques à l'ère Scrapfly (galerie, # ville/province, statut) ; l'acteur produit la même forme, cache réutilisé DETAIL_KEY = "v2" # accepted scope: Canada outside Québec — the public feed sometimes slips # out-of-area ads (US border cities, Québec) into a city's results _QC_STATES = {"qc", "quebec", "québec"} _CANADA_BBOX = (41.6, 83.2, -141.0, -52.5) _QC_BBOX = (44.9, 62.8, -79.6, -56.9) def _unit_type(title: str, desc: str = "") -> str: blob = f"{title} {desc}" ut = normalize_unit_type(title) if re.match(r"^\d bedrooms?$|^5\+ bedrooms$|^Studio$|^Loft$", ut or ""): return ut m = re.search(r"(\d+)\s*(?:bed|bedroom|chambre|cc|br)\b", blob, re.I) if m: n = int(m.group(1)) if n >= 5: return "5+ bedrooms" return f"{n} bedroom" + ("s" if n > 1 else "") if re.search(r"\bstudio|bachelor\b", blob, re.I): return "Studio" if re.search(r"\b(?:private\s+)?room\b|chambre", blob, re.I): return "Room" return "" def _usable(payload: dict | None) -> bool: """Payload détail exploitable (ni vide, ni marqueur « fiche sans objet »).""" return bool(payload) and not payload.get("nopdp") def _non_qc(payload: dict) -> bool: """True when the ad is OUT of Rent-Ka's scope (Québec or outside Canada). Name kept for the call sites.""" state = (payload.get("state") or "").strip().lower() if state: if state in _QC_STATES: return True return state not in _STATE_PROV # US states etc. # no state: geographic net — some PDPs only provide GPS lat, lng = payload.get("lat"), payload.get("lng") if lat is not None and lng is not None: in_canada = (_CANADA_BBOX[0] <= lat <= _CANADA_BBOX[1] and _CANADA_BBOX[2] <= lng <= _CANADA_BBOX[3]) in_qc = (_QC_BBOX[0] <= lat <= _QC_BBOX[1] and _QC_BBOX[2] <= lng <= _QC_BBOX[3]) return (not in_canada) or in_qc return False # version des payloads détail : bump quand parse_detail (acteur) apprend de # nouveaux champs, pour re-visiter progressivement les fiches déjà en cache PAYLOAD_VERSION = 3 _TYPE_LABELS = {"apartment", "house", "townhouse", "condo", "room", "flat", "appartement", "maison", "maison de ville", "chambre", "loft"} _BEDS_RX = re.compile(r"(\d+(?:[.,]\d+)?)\s*(?:beds?\b|chambres?\b|lits?\b)", re.I) _BATHS_RX = re.compile(r"(\d+(?:[.,]\d+)?)\s*(?:baths?\b|salles?\b)", re.I) def _std_detail(d: dict) -> dict: """Payload acteur -> dict pour du.apply_detail (labels PDP en/fr).""" std = {k: d[k] for k in ("description", "images", "lat", "lng", "city", "address") if d.get(k)} amen: list[str] = [] for lbl in d.get("unit_fields") or []: mb = _BEDS_RX.search(lbl) if mb: # « 2 beds · 1 bath » std["bedrooms"] = float(mb.group(1).replace(",", ".")) ms = _BATHS_RX.search(lbl) if ms: std["bathrooms"] = float(ms.group(1).replace(",", ".")) continue low = lbl.casefold() if low in _TYPE_LABELS: continue # type générique : _unit_type (titre/desc) fait mieux if "furnish" in low or "meublé" in low: std["furnished"] = not ("unfurnish" in low or "non meublé" in low) if "pet" in low or "animaux" in low: std["pets"] = ("non" if "no pet" in low or "pas d" in low else "oui") amen.append(lbl) if amen: std["amenities"] = amen for part in (d.get("listed_text") or "").split("·"): p = part.strip() if p.casefold().startswith(("available", "disponible")): std["availability"] = p extras = {k: d[k] for k in ("walk_score", "transit_score", "bike_score") if d.get(k) is not None} if d.get("virtual_tour_url"): extras["virtual_tour"] = d["virtual_tour_url"] if extras: std["details"] = extras return std class FacebookMarketplaceConnector(BaseConnector): source_id = "fb_marketplace" request_delay = 1.0 # -- orchestration de l'acteur Apify -------------------------------------- def _search_urls(self) -> tuple[list[str], dict[str, str]]: """URLs de recherche (ville × tranche) + mapping URL -> nom de ville.""" cities, regions = dict(CITIES), dict(REGIONS) if CITY_LIMIT: wanted = {c.strip().casefold() for c in CITY_LIMIT.split(",")} cities = {k: v for k, v in cities.items() if k in wanted or v.casefold() in wanted} regions = {k: v for k, v in regions.items() if v.casefold() in wanted} urls: list[str] = [] by_url: dict[str, str] = {} for slug, city in cities.items(): for lo, hi in PRICE_BANDS: u = (f"{BASE}/{slug}/propertyrentals" f"?minPrice={lo}&maxPrice={hi}&sortBy=creation_time_descend") urls.append(u) by_url[u] = city for cid, city in regions.items(): for lo, hi in REGIONAL_BANDS: u = (f"{BASE}/{cid}/propertyrentals" f"?minPrice={lo}&maxPrice={hi}&sortBy=creation_time_descend") urls.append(u) by_url[u] = city return urls, by_url def _run_actor(self, payload: dict, token: str) -> list[dict]: """Lance l'acteur, attend la fin, retourne les items du dataset.""" r = _requests.post( f"{APIFY_API}/acts/{ACTOR}/runs?waitForFinish=120", json=payload, timeout=180, headers={"Authorization": f"Bearer {token}"}) r.raise_for_status() run = r.json()["data"] deadline = time.time() + RUN_TIMEOUT while run["status"] in ("READY", "RUNNING") and time.time() < deadline: time.sleep(10) run = _requests.get( f"{APIFY_API}/actor-runs/{run['id']}", timeout=60, headers={"Authorization": f"Bearer {token}"}).json()["data"] if run["status"] != "SUCCEEDED": raise RuntimeError(f"acteur {ACTOR} : run {run['id']} " f"terminé en {run['status']}") items: list[dict] = [] offset = 0 while True: batch = _requests.get( f"{APIFY_API}/datasets/{run['defaultDatasetId']}/items" f"?limit=1000&offset={offset}", timeout=120, headers={"Authorization": f"Bearer {token}"}).json() items.extend(batch) if len(batch) < 1000: return items offset += 1000 # -- caches / BD ----------------------------------------------------------- def _fresh_ids(self, cache: du.TtlDetailCache) -> list[str]: """IDs dont le payload détail en cache est encore frais (clé + TTL) : l'acteur ne les revisitera pas.""" rows = cache.con.execute( "SELECT external_id FROM detail_cache" " WHERE source=? AND key=? AND fetched_at > ?" " AND (json_extract(payload,'$.pv') >= ?" " OR json_extract(payload,'$.nopdp') IS NOT NULL" " OR json_extract(payload,'$.gone') IS NOT NULL)", (self.source_id, DETAIL_KEY, time.time() - TTL_DAYS * 86400, PAYLOAD_VERSION)).fetchall() return [r["external_id"] for r in rows] def _poor_active_rows(self, cache: du.TtlDetailCache) -> list: """Fiches ACTIVES en BD encore pauvres (candidates au rattrapage).""" rows = cache.con.execute( "SELECT external_id, url, title, address, sector, city, unit_type," " bedrooms, bathrooms, price, price_label, availability," " availability_date, area_sqft, pets, furnished, description," " amenities, details, images, lat, lng" " FROM listings WHERE source=? AND active=1" " ORDER BY last_seen DESC", (self.source_id,)).fetchall() def poor(r) -> bool: try: n_img = len(json.loads(r["images"] or "[]")) except ValueError: n_img = 0 return (not (r["description"] or "").strip() or r["lat"] is None or n_img <= 1 or not (r["address"] or "").strip() or r["bedrooms"] is None) return [r for r in rows if poor(r)] def _listing_from_row(self, r) -> Listing: """Reconstruit le Listing depuis sa ligne BD (phase de rattrapage).""" def js(s, default): try: return json.loads(s) if s else default except ValueError: return default return Listing( source=self.source_id, external_id=r["external_id"], url=r["url"], title=r["title"] or "", address=r["address"] or "", sector=r["sector"] or "", city=r["city"] or "", unit_type=r["unit_type"] or "", bedrooms=r["bedrooms"], bathrooms=r["bathrooms"], price=r["price"], price_label=r["price_label"] or "", availability=r["availability"] or "", availability_date=r["availability_date"], area_sqft=r["area_sqft"], pets=r["pets"], furnished=(None if r["furnished"] is None else bool(r["furnished"])), description=r["description"] or "", amenities=js(r["amenities"], []), details=js(r["details"], {}), images=js(r["images"], []), lat=r["lat"], lng=r["lng"], ) def _backfill(self, cache: du.TtlDetailCache, poor_rows: list, done: set[str]) -> list[Listing]: """Rattrapage : ré-émet les fiches actives pauvres dont le cache (rempli par l'acteur via extraDetailIds) apporte du neuf. Les annonces mortes suivent le cycle normal miss_count → retrait.""" def adds(r, s: dict) -> bool: if s.get("description") and len(s["description"]) > \ len(r["description"] or ""): return True if s.get("lat") is not None and r["lat"] is None: return True try: n_img = len(json.loads(r["images"] or "[]")) except ValueError: n_img = 0 if len(s.get("images") or []) > n_img: return True if s.get("address") and not (r["address"] or "").strip(): return True if s.get("bedrooms") is not None and r["bedrooms"] is None: return True return bool(s.get("city")) and not (r["city"] or "") out: list[Listing] = [] for r in poor_rows: if r["external_id"] in done: continue d, _fresh = cache.peek(r["external_id"]) if not _usable(d) or d.get("gone") or _non_qc(d): continue # morte / hors QC / rien de neuf std = _std_detail(d) if not adds(r, std): continue # rien à apporter : laisser vivre son cycle lst = self._listing_from_row(r) du.apply_detail(lst, std) if d.get("city"): # reverse geocode FB : autoritaire lst.city = d["city"] if not lst.unit_type: lst.unit_type = _unit_type(lst.title, lst.description) out.append(lst) return out # -- pipeline principal ---------------------------------------------------- def fetch(self) -> list[Listing]: token = os.environ.get("APIFY_TOKEN") if not token: raise RuntimeError("APIFY_TOKEN manquant (voir .env)") urls, city_by_url = self._search_urls() cache = du.TtlDetailCache(self, budget=0, ttl_days=TTL_DAYS, key=DETAIL_KEY, fetch_html=lambda _u: "") try: fresh = set(self._fresh_ids(cache)) poor_rows = self._poor_active_rows(cache) extra = [r["external_id"] for r in poor_rows if r["external_id"] not in fresh][:BACKFILL_LIMIT] actor_input: dict = { "searchUrls": urls, "getDetails": True, "maxDetails": DETAIL_LIMIT + BACKFILL_LIMIT, "skipDetailIds": sorted(fresh), "extraDetailIds": extra, "concurrency": CONCURRENCY, "requestDelay": 1, } # abonnement résidentiel Oxylabs (déjà payé) plutôt que le proxy # Apify facturé au Go ; l'acteur retombe sur Apify RESIDENTIAL # si le gabarit est absent ox_u = os.environ.get("OXYLABS_PROXY_USER") ox_p = os.environ.get("OXYLABS_PROXY_PASS") if ox_u and ox_p: actor_input["proxyUrlTemplate"] = ( f"http://{ox_u}-cc-CA-sessid-{{session}}:{ox_p}" f"@pr.oxylabs.io:7777") items = self._run_actor(actor_input, token) # payloads détail -> cache BD (mêmes formes/clés que l'ère Scrapfly) for it in items: if it.get("kind") != "detail": continue if not it.get("ok"): continue # échec réseau acteur : pas de cache payload = {k: v for k, v in it.items() if k not in ("kind", "id", "ok") and v is not None} if payload and not payload.get("nopdp"): payload["pv"] = PAYLOAD_VERSION cache.put(str(it["id"]), payload or {"nopdp": True}) out: dict[str, Listing] = {} for it in items: if it.get("kind") != "listing": continue lid = str(it["id"]) price = it.get("price") if price is None or not (PRICE_MIN <= price <= PRICE_MAX): continue # sans prix mensuel valable : ignorer if it.get("is_sold") or it.get("is_pending"): continue # déjà loué / en attente if _non_qc(it): continue # suggestion FB hors Québec dès la recherche detail, _f = cache.peek(lid) detail = detail if _usable(detail) else {} if detail.get("gone"): continue # la fiche dit : loué / retiré if detail and _non_qc(detail): continue # annonce hors Québec glissée dans le flux title = it.get("title") or "" desc = detail.get("description") or "" photo = it.get("primary_photo") or "" images = detail.get("images") or ([photo] if photo else []) city = (detail.get("city") or it.get("city") or city_by_url.get(it.get("source_url") or "", "")) prov = (_STATE_PROV.get( (detail.get("state") or "").strip().lower()) or _CITY_PROV.get(city, "ON")) lst = Listing( source=self.source_id, external_id=lid, url=f"{BASE}/item/{lid}/", title=title[:200], city=city, province=prov, unit_type=_unit_type(title, desc), price=float(price), price_label=f"${price:,.0f}/month", description=desc, images=images, lat=detail.get("lat"), lng=detail.get("lng"), ) # champs structurés du PDP : adresse, chambres/sdb, commodités, # disponibilité, scores de marche/transport du.apply_detail(lst, _std_detail(detail)) out[lid] = lst listings = list(out.values()) # rattrapage des fiches actives pauvres sorties de la recherche listings.extend(self._backfill(cache, poor_rows, set(out))) finally: cache.close() return listings