spb/immo-ka

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/fb_marketplace.py : Facebook Marketplace (propriétés à vendre)5#   via l'acteur Apify maison gorgeous_thistle/ka-fb-immo (jumeau « vente »6#   de ka-fb-marketplace utilisé par Lou-Ka pour les locations).7#8#   L'acteur scrape le Marketplace public DÉCONNECTÉ (JSON préchargé du HTML,9#   proxy résidentiel CA) : phase recherche (33 villes × tranches de prix)10#   puis fiches détail (GPS, adresse civique, description, galerie complète,11#   statut vendu/retiré). Particularité de la source : le flux de recherche est12#   trié par date de création (~25 annonces/URL), une annonce en disparaît donc13#   en quelques jours alors qu'elle est toujours en vente. Le connecteur14#   maintient donc les fiches actives par RE-VÉRIFICATION tournante15#   (extraDetailIds, rotation par ancienneté du cache) : toujours en vente →16#   conservée, vendue/retirée → disparaît (délai de grâce standard).17#18#   ⚠️ Loi 25 : l'acteur n'extrait PAS le vendeur (nom/téléphone) — les champs19#   broker_* restent vides pour cette source.20# -----------------------------------------------------------------------------21from __future__ import annotations2223import json24import re25import time26from urllib.parse import quote2728import requests2930from .base import BaseConnector31from ._resilient import _secret32from ..schema import PropertyListing3334APIFY_API = "https://api.apify.com/v2"35ACTOR = "gorgeous_thistle~ka-fb-immo"3637DETAIL_KEY = "v1"          # bump = re-parse de toutes les fiches38MAX_DETAILS = 600          # budget fiches détail par run (nouveautés + recheck)39RECHECK_BUDGET = 300       # dont : re-vérifications d'annonces actives40MIN_INTERVAL_S = 3 * 3600  # entre deux runs Apify : rejouer la BD (annonces FB41                           # peu volatiles ; épargne compute + bande passante)42RUN_TIMEOUT_S = 45 * 6043MIN_PRICE = 20000          # sous ce prix : appâts « contactez-moi » / loyers4445_MLS_RE = re.compile(r"(?:centris|mls)\D{0,12}(\d{7,10})", re.I)46_DIGIT_RE = re.compile(r"\d")474849class FbMarketplaceConnector(BaseConnector):50    source_id = "fb_marketplace"51    use_detail_cache = False    # cache géré manuellement (batch via l'acteur)5253    # -- API Apify (accès direct : pas la chaîne anti-bot des sites cibles) ---54    def _apify(self, method: str, path: str, token: str, **kw):55        r = requests.request(method, f"{APIFY_API}{path}",56                             headers={"Authorization": f"Bearer {token}"},57                             timeout=120, **kw)58        r.raise_for_status()59        return r.json()6061    def _proxy_template(self) -> str | None:62        """Gabarit Oxylabs résidentiel CA ({session} = session collante) —63        même mécanique que les runs de prod de ka-fb-marketplace (le proxy64        Apify se fait servir le mur de login sur les fiches détail)."""65        endpoint = _secret("OXYLABS_PROXY")66        user = _secret("OXYLABS_PROXY_USER")67        pwd = _secret("OXYLABS_PROXY_PASS")68        if not (endpoint and user and pwd):69            return None70        endpoint = endpoint.replace("http://", "")71        return (f"http://{user}-cc-CA-sessid-{{session}}:"72                f"{quote(pwd, safe='')}@{endpoint}")7374    def _run_actor(self, token: str, run_input: dict) -> list[dict]:75        """Démarre un run, attend la fin, retourne les items du dataset."""76        run = self._apify(77            "POST", f"/acts/{ACTOR}/runs?memory=2048&timeout=3000",78            token, json=run_input)["data"]79        run_id = run["id"]80        t0 = time.time()81        status = run["status"]82        while status in ("READY", "RUNNING"):83            if time.time() - t0 > RUN_TIMEOUT_S:84                self._apify("POST", f"/actor-runs/{run_id}/abort", token)85                raise RuntimeError(f"run Apify {run_id} : délai dépassé")86            time.sleep(20)87            run = self._apify("GET", f"/actor-runs/{run_id}", token)["data"]88            status = run["status"]89        if status != "SUCCEEDED":90            raise RuntimeError(f"run Apify {run_id} : statut {status}")91        ds = run["defaultDatasetId"]92        items, offset = [], 093        while True:94            page = self._apify(95                "GET", f"/datasets/{ds}/items?clean=true&format=json"96                       f"&limit=1000&offset={offset}", token)97            items.extend(page)98            if len(page) < 1000:99                return items100            offset += 1000101102    # -- reconstruction depuis la BD (replay/rechecks) -------------------------103    def _row_to_listing(self, row) -> PropertyListing:104        return PropertyListing(105            source=self.source_id, external_id=row["external_id"],106            url=row["url"] or "", title=row["title"] or "",107            address=row["address"] or "", sector=row["sector"] or "",108            city=row["city"] or "", region=row["region"] or "",109            property_type=row["property_type"] or "", price=row["price"],110            price_label=row["price_label"] or "", bedrooms=row["bedrooms"],111            bathrooms=row["bathrooms"], powder_rooms=row["powder_rooms"],112            area_sqft=row["area_sqft"], lot_sqft=row["lot_sqft"],113            year_built=row["year_built"], mls=row["mls"] or "",114            status=row["status"] or "a-vendre",115            description=row["description"] or "",116            features=json.loads(row["features"] or "[]"),117            details=json.loads(row["details"] or "{}"),118            images=json.loads(row["images"] or "[]"),119            lat=row["lat"], lng=row["lng"])120121    # -- mapping recherche + détail -> PropertyListing --------------------------122    def _make_listing(self, lid: str, obj: dict, det: dict) -> PropertyListing:123        unit = det.get("unit_fields") or []124        addr = (det.get("address") or "").strip()125        if not _DIGIT_RE.search(addr):126            addr = ""                    # « Sainte-Claire, QC » n'est pas civique127        desc = det.get("description") or ""128        mls = ""129        m = _MLS_RE.search(desc)130        if m:131            mls = m.group(1)132        details: dict = {}133        for k_src, k_dst in (("category_name", "Catégorie Marketplace"),134                             ("listed_text", "Publication"),135                             ("external_url", "Lien externe"),136                             ("virtual_tour_url", "Visite virtuelle"),137                             ("postal_code", "Code postal (RTA)"),138                             ("walk_score", "Walk Score"),139                             ("transit_score", "Transit Score"),140                             ("bike_score", "Bike Score")):141            if det.get(k_src) not in (None, ""):142                details[k_dst] = det[k_src]143        images = det.get("images") or []144        if not images and obj.get("primary_photo"):145            images = [obj["primary_photo"]]146        city = (det.get("city") or obj.get("city") or "").strip()147        return PropertyListing(148            source=self.source_id, external_id=lid,149            url=obj.get("url") or f"https://www.facebook.com/marketplace/item/{lid}/",150            title=obj.get("title") or "",151            address=addr, city=city,152            property_type=(unit[0] if unit else obj.get("title") or ""),153            price=obj.get("price"),154            price_label=(f"{int(obj['price']):,} $".replace(",", " ")155                         if obj.get("price") else ""),156            mls=mls, description=desc,157            features=[f.replace("\xa0", " ") for f in unit],158            details=details, images=images,159            lat=det.get("lat"), lng=det.get("lng"))160161    # -- contrat ----------------------------------------------------------------162    def fetch(self) -> list[PropertyListing]:163        from .. import db164        con = db.connect()165166        active = {r["external_id"]: r for r in con.execute(167            "SELECT * FROM listings WHERE source=? AND active=1",168            (self.source_id,)).fetchall()}169170        # throttle : entre deux runs Apify, rejouer l'état connu (aucun réseau).171        # La date du dernier VRAI run = MAX(fetched_at) du cache détail (les172        # replays n'y écrivent pas, contrairement à sync_log).173        last = con.execute(174            "SELECT MAX(fetched_at) ts FROM detail_cache WHERE source=?",175            (self.source_id,)).fetchone()176        if last and last["ts"] and time.time() - last["ts"] < MIN_INTERVAL_S \177                and active:178            print(f"[immo-ka]   fb_marketplace: replay BD "179                  f"({len(active)} actives, prochain run Apify dans "180                  f"{(MIN_INTERVAL_S - time.time() + last['ts']) / 60:.0f} min)")181            return [self._row_to_listing(r) for r in active.values()]182183        token = _secret("APIFY_TOKEN")184        if not token:185            raise RuntimeError("APIFY_TOKEN manquant (voir ~/.claude/.env)")186187        cache = {r["external_id"]: r for r in con.execute(188            "SELECT external_id, key, fetched_at FROM detail_cache WHERE source=?",189            (self.source_id,)).fetchall()}190191        # re-vérification tournante des annonces actives : les plus anciennes192        # au cache d'abord (fetched_at) — vivantes → conservées, parties → non193        recheck = sorted(194            (i for i in active if i in cache),195            key=lambda i: cache[i]["fetched_at"] or 0)[:RECHECK_BUDGET]196        recheck_set = set(recheck)197        skip = [i for i, r in cache.items()198                if r["key"] == DETAIL_KEY and i not in recheck_set]199200        run_input: dict = {201            "getDetails": True, "maxDetails": MAX_DETAILS,202            "skipDetailIds": skip, "extraDetailIds": recheck,203        }204        tpl = self._proxy_template()205        if tpl:206            run_input["proxyUrlTemplate"] = tpl207208        items = self._run_actor(token, run_input)209        search = {i["id"]: i for i in items if i.get("kind") == "listing"}210        fresh = {i["id"]: {k: v for k, v in i.items()211                           if k not in ("kind", "id", "ok")}212                 for i in items if i.get("kind") == "detail" and i.get("ok")}213        print(f"[immo-ka]   fb_marketplace: {len(search)} en recherche, "214              f"{len(fresh)} fiches détail, {len(recheck)} re-vérifications")215216        for lid, payload in fresh.items():    # mémoriser (nopdp inclus)217            db.put_cached_detail(con, self.source_id, lid, DETAIL_KEY, payload)218219        out: list[PropertyListing] = []220        emitted: set[str] = set()221222        # 1) flux de recherche (nouveautés + annonces récentes)223        for lid, obj in search.items():224            st = (obj.get("state") or "").strip().lower()225            if st and st not in ("qc", "quebec", "québec"):226                continue227            if obj.get("is_sold") or obj.get("is_pending"):228                continue229            price = obj.get("price")230            if not price or price < MIN_PRICE:231                continue232            det = fresh.get(lid)233            if det is None:234                det = db.get_stale_detail(con, self.source_id, lid) or {}235            elif det.get("gone"):236                continue                  # verdict frais : vendue/retirée237            out.append(self._make_listing(lid, obj, det))238            emitted.add(lid)239240        # 2) annonces actives sorties du flux de recherche (tri par création) :241        #    re-vérifiées ce run → verdict frais ; sinon rejouées telles quelles242        #    en attendant leur tour de rotation243        for lid, row in active.items():244            if lid in emitted:245                continue246            det = fresh.get(lid)247            if det is not None and det.get("gone"):248                continue                  # vendue/retirée : délai de grâce puis retrait249            if det is not None and det.get("nopdp"):250                continue                  # fiche supprimée (mur permanent)251            lst = self._row_to_listing(row)252            if det:                       # rafraîchir les champs riches253                if det.get("description"):254                    lst.description = det["description"]255                if det.get("images"):256                    lst.images = det["images"]257                if det.get("lat") and det.get("lng"):258                    lst.lat, lst.lng = det["lat"], det["lng"]259            out.append(lst)260            emitted.add(lid)261        return out262