# ----------------------------------------------------------------------------- # Immo-Ka — Agrégateur de maisons à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/fb_marketplace.py : Facebook Marketplace (propriétés à vendre) # via l'acteur Apify maison gorgeous_thistle/ka-fb-immo (jumeau « vente » # de ka-fb-marketplace utilisé par Lou-Ka pour les locations). # # L'acteur scrape le Marketplace public DÉCONNECTÉ (JSON préchargé du HTML, # proxy résidentiel CA) : phase recherche (33 villes × tranches de prix) # puis fiches détail (GPS, adresse civique, description, galerie complète, # statut vendu/retiré). Particularité de la source : le flux de recherche est # trié par date de création (~25 annonces/URL), une annonce en disparaît donc # en quelques jours alors qu'elle est toujours en vente. Le connecteur # maintient donc les fiches actives par RE-VÉRIFICATION tournante # (extraDetailIds, rotation par ancienneté du cache) : toujours en vente → # conservée, vendue/retirée → disparaît (délai de grâce standard). # # ⚠️ Loi 25 : l'acteur n'extrait PAS le vendeur (nom/téléphone) — les champs # broker_* restent vides pour cette source. # ----------------------------------------------------------------------------- from __future__ import annotations import json import re import time from urllib.parse import quote import requests from .base import BaseConnector from ._resilient import _secret from ..schema import PropertyListing APIFY_API = "https://api.apify.com/v2" ACTOR = "gorgeous_thistle~ka-fb-immo" DETAIL_KEY = "v1" # bump = re-parse de toutes les fiches MAX_DETAILS = 600 # budget fiches détail par run (nouveautés + recheck) RECHECK_BUDGET = 300 # dont : re-vérifications d'annonces actives MIN_INTERVAL_S = 3 * 3600 # entre deux runs Apify : rejouer la BD (annonces FB # peu volatiles ; épargne compute + bande passante) RUN_TIMEOUT_S = 45 * 60 MIN_PRICE = 20000 # sous ce prix : appâts « contactez-moi » / loyers _MLS_RE = re.compile(r"(?:centris|mls)\D{0,12}(\d{7,10})", re.I) _DIGIT_RE = re.compile(r"\d") class FbMarketplaceConnector(BaseConnector): source_id = "fb_marketplace" use_detail_cache = False # cache géré manuellement (batch via l'acteur) # -- API Apify (accès direct : pas la chaîne anti-bot des sites cibles) --- def _apify(self, method: str, path: str, token: str, **kw): r = requests.request(method, f"{APIFY_API}{path}", headers={"Authorization": f"Bearer {token}"}, timeout=120, **kw) r.raise_for_status() return r.json() def _proxy_template(self) -> str | None: """Gabarit Oxylabs résidentiel CA ({session} = session collante) — même mécanique que les runs de prod de ka-fb-marketplace (le proxy Apify se fait servir le mur de login sur les fiches détail).""" endpoint = _secret("OXYLABS_PROXY") user = _secret("OXYLABS_PROXY_USER") pwd = _secret("OXYLABS_PROXY_PASS") if not (endpoint and user and pwd): return None endpoint = endpoint.replace("http://", "") return (f"http://{user}-cc-CA-sessid-{{session}}:" f"{quote(pwd, safe='')}@{endpoint}") def _run_actor(self, token: str, run_input: dict) -> list[dict]: """Démarre un run, attend la fin, retourne les items du dataset.""" run = self._apify( "POST", f"/acts/{ACTOR}/runs?memory=2048&timeout=3000", token, json=run_input)["data"] run_id = run["id"] t0 = time.time() status = run["status"] while status in ("READY", "RUNNING"): if time.time() - t0 > RUN_TIMEOUT_S: self._apify("POST", f"/actor-runs/{run_id}/abort", token) raise RuntimeError(f"run Apify {run_id} : délai dépassé") time.sleep(20) run = self._apify("GET", f"/actor-runs/{run_id}", token)["data"] status = run["status"] if status != "SUCCEEDED": raise RuntimeError(f"run Apify {run_id} : statut {status}") ds = run["defaultDatasetId"] items, offset = [], 0 while True: page = self._apify( "GET", f"/datasets/{ds}/items?clean=true&format=json" f"&limit=1000&offset={offset}", token) items.extend(page) if len(page) < 1000: return items offset += 1000 # -- reconstruction depuis la BD (replay/rechecks) ------------------------- def _row_to_listing(self, row) -> PropertyListing: return PropertyListing( source=self.source_id, external_id=row["external_id"], url=row["url"] or "", title=row["title"] or "", address=row["address"] or "", sector=row["sector"] or "", city=row["city"] or "", region=row["region"] or "", property_type=row["property_type"] or "", price=row["price"], price_label=row["price_label"] or "", bedrooms=row["bedrooms"], bathrooms=row["bathrooms"], powder_rooms=row["powder_rooms"], area_sqft=row["area_sqft"], lot_sqft=row["lot_sqft"], year_built=row["year_built"], mls=row["mls"] or "", status=row["status"] or "a-vendre", description=row["description"] or "", features=json.loads(row["features"] or "[]"), details=json.loads(row["details"] or "{}"), images=json.loads(row["images"] or "[]"), lat=row["lat"], lng=row["lng"]) # -- mapping recherche + détail -> PropertyListing -------------------------- def _make_listing(self, lid: str, obj: dict, det: dict) -> PropertyListing: unit = det.get("unit_fields") or [] addr = (det.get("address") or "").strip() if not _DIGIT_RE.search(addr): addr = "" # « Sainte-Claire, QC » n'est pas civique desc = det.get("description") or "" mls = "" m = _MLS_RE.search(desc) if m: mls = m.group(1) details: dict = {} for k_src, k_dst in (("category_name", "Catégorie Marketplace"), ("listed_text", "Publication"), ("external_url", "Lien externe"), ("virtual_tour_url", "Visite virtuelle"), ("postal_code", "Code postal (RTA)"), ("walk_score", "Walk Score"), ("transit_score", "Transit Score"), ("bike_score", "Bike Score")): if det.get(k_src) not in (None, ""): details[k_dst] = det[k_src] images = det.get("images") or [] if not images and obj.get("primary_photo"): images = [obj["primary_photo"]] city = (det.get("city") or obj.get("city") or "").strip() return PropertyListing( source=self.source_id, external_id=lid, url=obj.get("url") or f"https://www.facebook.com/marketplace/item/{lid}/", title=obj.get("title") or "", address=addr, city=city, property_type=(unit[0] if unit else obj.get("title") or ""), price=obj.get("price"), price_label=(f"{int(obj['price']):,} $".replace(",", " ") if obj.get("price") else ""), mls=mls, description=desc, features=[f.replace("\xa0", " ") for f in unit], details=details, images=images, lat=det.get("lat"), lng=det.get("lng")) # -- contrat ---------------------------------------------------------------- def fetch(self) -> list[PropertyListing]: from .. import db con = db.connect() active = {r["external_id"]: r for r in con.execute( "SELECT * FROM listings WHERE source=? AND active=1", (self.source_id,)).fetchall()} # throttle : entre deux runs Apify, rejouer l'état connu (aucun réseau). # La date du dernier VRAI run = MAX(fetched_at) du cache détail (les # replays n'y écrivent pas, contrairement à sync_log). last = con.execute( "SELECT MAX(fetched_at) ts FROM detail_cache WHERE source=?", (self.source_id,)).fetchone() if last and last["ts"] and time.time() - last["ts"] < MIN_INTERVAL_S \ and active: print(f"[immo-ka] fb_marketplace: replay BD " f"({len(active)} actives, prochain run Apify dans " f"{(MIN_INTERVAL_S - time.time() + last['ts']) / 60:.0f} min)") return [self._row_to_listing(r) for r in active.values()] token = _secret("APIFY_TOKEN") if not token: raise RuntimeError("APIFY_TOKEN manquant (voir ~/.claude/.env)") cache = {r["external_id"]: r for r in con.execute( "SELECT external_id, key, fetched_at FROM detail_cache WHERE source=?", (self.source_id,)).fetchall()} # re-vérification tournante des annonces actives : les plus anciennes # au cache d'abord (fetched_at) — vivantes → conservées, parties → non recheck = sorted( (i for i in active if i in cache), key=lambda i: cache[i]["fetched_at"] or 0)[:RECHECK_BUDGET] recheck_set = set(recheck) skip = [i for i, r in cache.items() if r["key"] == DETAIL_KEY and i not in recheck_set] run_input: dict = { "getDetails": True, "maxDetails": MAX_DETAILS, "skipDetailIds": skip, "extraDetailIds": recheck, } tpl = self._proxy_template() if tpl: run_input["proxyUrlTemplate"] = tpl items = self._run_actor(token, run_input) search = {i["id"]: i for i in items if i.get("kind") == "listing"} fresh = {i["id"]: {k: v for k, v in i.items() if k not in ("kind", "id", "ok")} for i in items if i.get("kind") == "detail" and i.get("ok")} print(f"[immo-ka] fb_marketplace: {len(search)} en recherche, " f"{len(fresh)} fiches détail, {len(recheck)} re-vérifications") for lid, payload in fresh.items(): # mémoriser (nopdp inclus) db.put_cached_detail(con, self.source_id, lid, DETAIL_KEY, payload) out: list[PropertyListing] = [] emitted: set[str] = set() # 1) flux de recherche (nouveautés + annonces récentes) for lid, obj in search.items(): st = (obj.get("state") or "").strip().lower() if st and st not in ("qc", "quebec", "québec"): continue if obj.get("is_sold") or obj.get("is_pending"): continue price = obj.get("price") if not price or price < MIN_PRICE: continue det = fresh.get(lid) if det is None: det = db.get_stale_detail(con, self.source_id, lid) or {} elif det.get("gone"): continue # verdict frais : vendue/retirée out.append(self._make_listing(lid, obj, det)) emitted.add(lid) # 2) annonces actives sorties du flux de recherche (tri par création) : # re-vérifiées ce run → verdict frais ; sinon rejouées telles quelles # en attendant leur tour de rotation for lid, row in active.items(): if lid in emitted: continue det = fresh.get(lid) if det is not None and det.get("gone"): continue # vendue/retirée : délai de grâce puis retrait if det is not None and det.get("nopdp"): continue # fiche supprimée (mur permanent) lst = self._row_to_listing(row) if det: # rafraîchir les champs riches if det.get("description"): lst.description = det["description"] if det.get("images"): lst.images = det["images"] if det.get("lat") and det.get("lng"): lst.lat, lst.lng = det["lat"], det["lng"] out.append(lst) emitted.add(lid) return out