# ----------------------------------------------------------------------------- # Home-Ka — US real-estate aggregator (Groupe KA) # Author: Simon-Pierre Boucher — contact@spboucher.ai # discovery.py : automatic brokerage pipeline # # discover brokerage → inspect website → detect IDX/feed/provider # → find public contact information → classify potential connector # → score brokerage → add to admin (/admin/brokerages) # # Crawling is used HERE — to discover brokerages and their technology — then # every interesting source should be converted into a DIRECT FEED (RESO Web # API first). Direct requests with polite throttling; the anti-bot fallback # chain (base.py) only fires when a site blocks. # ----------------------------------------------------------------------------- from __future__ import annotations import json import re import time import requests from . import db UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/126 Safari/537.36 " "HomeKaBot/1.0 (+https://www.home-ka.com/bot; contact@spboucher.ai)") # -------------------------------------------------------------------------- # IDX / website-platform fingerprints. Each entry: # (provider name, regex on the homepage HTML/headers, feed hint, feed prob.) # feed hint = the most realistic connector once a partnership exists. # Probabilities reflect how easily that platform's brokerages hand out a # usable feed (kvCORE/iHomefinder shops usually have MLS/RESO access; pure # website builders rarely do). # -------------------------------------------------------------------------- _FINGERPRINTS: list[tuple[str, str, str, int]] = [ ("kvCORE / BoldTrail", r"kvcore|boldtrail|insiderealestate|idxhome\.com", "reso", 70), ("iHomefinder", r"ihomefinder|idxhome|optimahomes|ihfkestrel", "reso", 65), ("IDX Broker", r"idxbroker\.com|idx-broker|IDX Broker", "reso", 60), ("Showcase IDX", r"showcaseidx", "reso", 55), ("Real Geeks", r"realgeeks", "json-api", 55), ("Sierra Interactive", r"sierrastatic|sierrainteractive", "json-api", 60), ("Ylopo", r"ylopo", "json-api", 55), ("CINC", r"cincpro|cinc\.com", "json-api", 50), ("Lofty (Chime)", r"chime\.me|lofty\.com|chimeroi", "json-api", 50), ("Luxury Presence", r"luxurypresence", "idx-partner", 45), ("AgentFire", r"agentfire", "idx-partner", 45), ("Placester", r"placester", "idx-partner", 40), ("Real Estate Webmasters", r"realestatewebmasters|rew\.ca|rew-", "json-api", 55), ("Delta Media Group", r"deltamediagroup|deltagroup", "xml", 55), ("Union Street Media", r"unionstreetmedia", "idx-partner", 45), ("Moxi Works", r"moxiworks|moxi-", "idx-partner", 50), ("Tribus", r"tribus", "json-api", 50), ("BoomTown", r"boomtownroi|boomtown", "json-api", 50), ("Homes.com / Homesnap", r"homesnap|homes\.com/widget", "idx-partner", 35), ("WordPress + IDX plugin", r"wp-content/plugins/(idx|impress|wpl|realtyna|optima)", "idx-partner", 50), ("WordPress", r"wp-content|wp-includes", "crawl", 30), ("Squarespace", r"squarespace", "crawl", 15), ("Wix", r"wix\.com|wixstatic", "crawl", 15), ] # Signals that a RESO/RETS pipeline exists behind the site _RESO_RE = re.compile( r"reso\b|api\.mlsgrid|trestle[.-]corelogic|api\.bridgedataoutput" r"|sparkapi\.com|spark\.paragonrels|retsly|rets\b|webapi.*odata|odata.*Property", re.I) _MLS_HINTS = [ ("ACTRIS", r"actris"), ("ARMLS", r"armls"), ("BRIGHT", r"bright\s*mls"), ("CRMLS", r"crmls"), ("Canopy", r"canopy\s*mls"), ("CharlestonMLS", r"charleston.*mls"), ("FMLS", r"\bfmls\b"), ("GAMLS", r"gamls"), ("HAR", r"\bhar\.com|houston association of realtors"), ("Heartland", r"heartland\s*mls"), ("MRED", r"\bmred\b|midwest real estate data"), ("MLSPIN", r"mls\s*pin|mlspin"), ("NTREIS", r"ntreis"), ("NWMLS", r"nwmls"), ("OneKey", r"onekey"), ("RMLS", r"\brmls\b"), ("SABOR", r"sabor"), ("Stellar", r"stellar\s*mls"), ("TRREB", r"trreb"), ("UtahRealEstate", r"utahrealestate"), ("Realtracs", r"realtracs"), ("BeachesMLS", r"beaches\s*mls"), ("MiamiMLS", r"miami.*(association|mls)"), ("Metrolist", r"metrolist"), ] _EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}") _CONTACT_HREF_RE = re.compile( r'href=["\']([^"\']*(?:contact|about|our-team|team|offices?)[^"\']*)["\']', re.I) _BAD_EMAIL = re.compile(r"example\.|sentry|wixpress|schema|\.png|\.jpg|\.gif|" r"godaddy|placeholder|yourdomain|@2x|email@domain|" r"^(user|name|test|john\.?doe|jane\.?doe)@", re.I) def _fetch_page(s: requests.Session, url: str, timeout: int = 25): """Direct GET; on an anti-bot block, escalate through the fallback chain (Oxylabs → Scrapfly → Bright Data) like the connectors do.""" from .connectors import _resilient as kar try: resp = s.get(url, timeout=timeout, allow_redirects=True) resp.raise_for_status() return resp except requests.HTTPError as exc: r = getattr(exc, "response", None) if r is not None and kar.is_blocked(r): better = kar.escalate_if_blocked(r, str(r.url) or url, timeout=timeout) if better is not None and getattr(better, "status_code", 0) == 200: return better raise except (requests.ConnectionError, requests.Timeout): better = kar.escalate(url, timeout=timeout) if better is not None and getattr(better, "status_code", 0) == 200: return better raise def inspect_website(website: str, session: requests.Session | None = None) -> dict: """Inspect one brokerage website: platform/IDX detection, RESO hints, MLS affiliations, contact discovery. Direct GET only (one page + the contact page); returns an evidence dict.""" s = session or requests.Session() s.headers["User-Agent"] = UA url = website if website.startswith("http") else f"https://{website}" out: dict = {"final_url": "", "idx_present": 0, "idx_provider": "", "reso_detected": 0, "possible_feed_type": "crawl", "feed_probability_score": 20, "mls_affiliations": [], "contact_page": "", "partnership_contact": "", "technical_contact": "", "evidence": {}} resp = _fetch_page(s, url) html = resp.text[:800_000] out["final_url"] = str(resp.url) server = " ".join(f"{k}: {v}" for k, v in resp.headers.items() if k.lower() in ("server", "x-powered-by", "x-generator")) haystack = html + " " + server hits = [] for name, pattern, feed, prob in _FINGERPRINTS: if re.search(pattern, haystack, re.I): hits.append({"provider": name, "feed": feed, "prob": prob}) if hits: best = max(hits, key=lambda h: h["prob"]) out["idx_present"] = 1 out["idx_provider"] = best["provider"] out["possible_feed_type"] = best["feed"] out["feed_probability_score"] = best["prob"] out["evidence"]["providers"] = [h["provider"] for h in hits] if _RESO_RE.search(haystack): out["reso_detected"] = 1 out["possible_feed_type"] = "reso" out["feed_probability_score"] = max(out["feed_probability_score"], 75) # any IDX search page at all (listings embedded) counts as IDX present if not out["idx_present"] and re.search( r"/idx/|idx-search|property-search|listing-search|mls-search", haystack, re.I): out["idx_present"] = 1 out["idx_provider"] = out["idx_provider"] or "unknown IDX" out["feed_probability_score"] = max(out["feed_probability_score"], 40) out["possible_feed_type"] = ("idx-partner" if out["possible_feed_type"] == "crawl" else out["possible_feed_type"]) for mls, pattern in _MLS_HINTS: if re.search(pattern, haystack, re.I): out["mls_affiliations"].append(mls) # contacts: emails on the homepage, then the contact page emails = [e for e in _EMAIL_RE.findall(html) if not _BAD_EMAIL.search(e)] m = _CONTACT_HREF_RE.search(html) if m: href = m.group(1) contact_url = href if href.startswith("http") else \ str(resp.url).rstrip("/") + "/" + href.lstrip("/") out["contact_page"] = contact_url try: time.sleep(0.5) c = _fetch_page(s, contact_url, timeout=20) emails += [e for e in _EMAIL_RE.findall(c.text) if not _BAD_EMAIL.search(e)] except requests.RequestException: pass emails = list(dict.fromkeys(e.lower() for e in emails)) if emails: tech = [e for e in emails if re.match( r"(it|tech|web|webmaster|dev|api|data|idx)[@.]", e)] biz = [e for e in emails if re.match( r"(info|contact|hello|admin|office|broker|partnerships?|media)[@.]", e)] out["technical_contact"] = tech[0] if tech else "" out["partnership_contact"] = biz[0] if biz else emails[0] out["evidence"]["emails"] = emails[:8] return out def priority(estimated_listings: int | None, estimated_agents: int | None, feed_prob: float, n_states: int) -> float: """0-100 priority: size × feed probability × geographic coverage.""" import math listings = estimated_listings or (estimated_agents or 0) * 8 size = min(60.0, 12 * math.log10(max(listings, 1) + 1)) return round(min(100.0, size * (0.4 + 0.6 * feed_prob / 100) + min(n_states, 5) * 3), 1) def inspect_one(con, brokerage_id: int) -> dict: """Run the inspection for one brokerage row and persist the results.""" row = con.execute("SELECT * FROM brokerages WHERE id=?", (brokerage_id,)).fetchone() if row is None: raise ValueError(f"brokerage {brokerage_id} not found") now = time.time() try: info = inspect_website(row["website"]) error = "" except Exception as exc: info = {} error = str(exc)[:300] if info: mls = list(dict.fromkeys( json.loads(row["mls_affiliations"] or "[]") + info["mls_affiliations"])) states = json.loads(row["states"] or "[]") prio = priority(row["estimated_listings"], row["estimated_agents"], info["feed_probability_score"], len(states)) con.execute( """UPDATE brokerages SET idx_present=?, idx_provider=?, reso_detected=?, possible_feed_type=?, feed_probability_score=?, priority_score=?, mls_affiliations=?, contact_page=COALESCE(NULLIF(?,''), contact_page), partnership_contact=COALESCE(NULLIF(?,''), partnership_contact), technical_contact=COALESCE(NULLIF(?,''), technical_contact), evidence=?, last_inspected=?, inspect_error='', updated=? WHERE id=?""", (info["idx_present"], info["idx_provider"], info["reso_detected"], info["possible_feed_type"], info["feed_probability_score"], prio, json.dumps(mls, ensure_ascii=False), info["contact_page"], info["partnership_contact"], info["technical_contact"], json.dumps(info["evidence"], ensure_ascii=False), now, now, brokerage_id)) else: con.execute( "UPDATE brokerages SET last_inspected=?, inspect_error=?, updated=?" " WHERE id=?", (now, error, now, brokerage_id)) con.commit() return {"id": brokerage_id, "ok": not error, "error": error, **(info or {})} def run_batch(limit: int = 25, reinspect_days: int = 90) -> dict: """Inspect the next never-inspected (then stalest) prospects — called after each sync cycle and by `run.py discover`.""" con = db.connect() cutoff = time.time() - reinspect_days * 86400 rows = con.execute( "SELECT id FROM brokerages WHERE website IS NOT NULL AND website<>''" " AND (last_inspected IS NULL OR" " (last_inspected < ? AND inspect_error=''))" " ORDER BY last_inspected IS NOT NULL, priority_score DESC, id" " LIMIT ?", (cutoff, limit)).fetchall() inspected = errors = 0 for r in rows: res = inspect_one(con, r["id"]) inspected += 1 errors += 0 if res["ok"] else 1 time.sleep(1.0) # politeness between brokerage sites con.close() return {"inspected": inspected, "errors": errors}