Realtor.com - For-sale listings and listing facts

import json, os, tempfile, time, urllib.parse
 
LOCATION = "Austin_TX"   # "City_ST" (spaces as hyphens, e.g. "San-Francisco_CA") or a 5-digit ZIP, e.g. "78704"
MAX_PAGES = 3            # safety cap; the block says so when it stops here
PAGE_SIZE = 42           # rows per full search page
PAUSE_S = 3              # polite pause between page loads
TRIES = 3                # loads per page before giving up
 
SRP_JS = r"""(() => {
  const r = window.__reactRouterDataRouter;
  const srp = r && r.state && r.state.loaderData && r.state.loaderData.srp;
  const v = srp && srp.search && srp.search.value;
  if (!v) return {error: 'no search data yet', title: document.title, url: location.href};
  return {
    url: location.href,
    total: v.total,
    rows: (v.properties || []).map(p => {
      const a = (p.location && p.location.address) || {};
      const d = p.description || {};
      const f = p.flags || {};
      return {
        property_id: p.property_id, listing_id: p.listing_id,
        address: a.line, unit: a.unit,   // line already includes the unit
        city: a.city, state: a.state_code, zip: a.postal_code,
        price: p.list_price, beds: d.beds, baths: d.baths_consolidated,
        sqft: d.sqft, lot_sqft: d.lot_sqft, type: d.type,
        status: p.status, status_text: p.statusText,
        is_pending: !!f.is_pending, is_contingent: !!f.is_contingent,
        is_foreclosure: !!f.is_foreclosure, is_new_listing: !!f.is_new_listing,
        list_date: p.list_date, lat: p.lat, lng: p.lng, broker: p.brokerageName,
        url: p.ldpSlug ? 'https://www.realtor.com/realestateandhomes-detail/' + p.ldpSlug : null
      };
    })
  };
})()"""
 
def search_url(location, page=1):
    base = "https://www.realtor.com/realestateandhomes-search/" + urllib.parse.quote(location, safe="")
    return base if page == 1 else f"{base}/pg-{page}"
 
def same_page(href, url):
    a, b = urllib.parse.urlsplit(href or ""), urllib.parse.urlsplit(url)
    norm = lambda u: urllib.parse.unquote(u.path).rstrip("/").lower()
    return a.netloc == b.netloc and norm(a) == norm(b)
 
def read_srp(url, wait_s=20):
    """Poll until the tab shows `url` and its loader data is ready. Returns (data or None, last seen)."""
    last, deadline = None, time.time() + wait_s
    while time.time() < deadline:
        try:
            d = js(SRP_JS) or {}
        except (TimeoutError, RuntimeError) as e:  # js() timeouts and mid-navigation errors; poll again
            d = {"error": str(e)[:200]}
        if not d.get("error") and same_page(d.get("url"), url) and isinstance(d.get("rows"), list):
            return d, d
        last = d
        time.sleep(1)
    return None, last
 
def load_page(url, first):
    last = None
    for attempt in range(TRIES):
        try:
            if first and attempt == 0:
                new_tab(url)
            else:
                goto_url(url)
        except TimeoutError:  # Page.navigate can outlast the IPC wait; read_srp checks the URL either way
            pass
        d, last = read_srp(url)
        if d:
            return d
        if attempt + 1 < TRIES:      # no sleep after the last load
            time.sleep(5 * 2 ** attempt)
    raise RuntimeError(f"no search data for {url} after {TRIES} loads; last seen: {last}")
 
listings, seen, total, complete = [], set(), None, False
for page in range(1, MAX_PAGES + 1):
    d = load_page(search_url(LOCATION, page), first=(page == 1))
    rows = d["rows"]
    if isinstance(d.get("total"), (int, float)) and not isinstance(d.get("total"), bool):
        total = int(d["total"])
    for r in rows:
        pid = r.get("property_id")
        if pid is None:
            listings.append(r)       # no id to dedupe on; keep it
        elif pid not in seen:
            seen.add(pid)
            listings.append(r)
    print(f"page {page}: {len(rows)} rows, total={total}")
    if len(rows) < PAGE_SIZE or (total is not None and page * PAGE_SIZE >= total):
        complete = True
        break
    time.sleep(PAUSE_S)
 
if not complete:
    print(f"STOPPED AT MAX_PAGES={MAX_PAGES}: {len(listings)} of {total} collected; raise MAX_PAGES for the rest")
print(len(listings), "listings of", total, "(complete)" if complete else "(partial)")
for r in listings[:3]:
    print(r["address"], r["zip"], r["price"], r["beds"], r["baths"], r["sqft"], r["status_text"], r["url"])
OUT = os.path.join(tempfile.gettempdir(), "realtor_listings.json")
json.dump(listings, open(OUT, "w"), indent=1)
print("saved", OUT)

import email.utils, json, os, tempfile, time, urllib.parse
 
N = 5          # how many listings to enrich
PAUSE_S = 2
TRIES = 4      # fetches per listing before it counts as failed
MAX_WAIT_S = 120  # longest Retry-After or backoff sleep; a longer ask stops the batch
 
class WaitTooLong(Exception):
    pass
 
START_JS = r"""((url, key) => {
  window.__rdcFacts = window.__rdcFacts || {};
  window.__rdcFacts[key] = null;
  const ctl = new AbortController();
  const timer = setTimeout(() => ctl.abort(), 30000);
  fetch(url, {credentials: 'same-origin', signal: ctl.signal}).then(async res => {
    const html = await res.text();
    const out = {http_status: res.status, final_url: res.url, retry_after: res.headers.get('retry-after')};
    const m = html.match(/<script id="__NEXT_DATA__"[^>]*>([\s\S]*?)<\/script>/);
    if (!m) return Object.assign(out, {error: 'no __NEXT_DATA__'});
    let p;
    try {
      const s = ((JSON.parse(m[1]).props || {}).pageProps || {}).initialReduxState || {};
      p = s.propertyDetails;
    } catch (e) { return Object.assign(out, {error: 'bad __NEXT_DATA__: ' + e}); }
    if (!p) return Object.assign(out, {error: 'no propertyDetails'});
    const d = p.description || {}, loc = p.location || {}, a = loc.address || {};
    const best = ((p.estimates && p.estimates.current_values) || []).find(e => e && e.isbest_homevalue);
    const seller = (p.advertisers || []).find(x => x && x.type === 'seller');
    out.facts = {
      url: p.href,
      property_id: p.property_id, listing_id: p.listing_id,
      status: p.status,
      flags: Object.entries(p.flags || {}).filter(([k, v]) => v === true).map(([k]) => k),
      list_price: p.list_price, price_per_sqft: p.price_per_sqft,
      list_date: p.list_date, last_update: p.source && p.source.last_update_date,
      days_on_mls: p.source && p.source.days_on_mls,
      address: a.line, unit: a.unit, city: a.city, state: a.state_code, zip: a.postal_code,
      county: loc.county && loc.county.name,
      neighborhoods: (loc.neighborhoods || []).map(n => n && n.name).filter(Boolean),
      type: d.type, sub_type: d.sub_type,
      beds: d.beds, baths: d.baths_consolidated, baths_full: d.baths_full, baths_half: d.baths_half,
      sqft: d.sqft, lot_sqft: d.lot_sqft, year_built: d.year_built,
      stories: d.stories, garage: d.garage,
      hoa_fee: p.hoa ? p.hoa.fee : null,
      estimate: best ? best.estimate : null,
      property_tax_rate: p.mortgage && p.mortgage.property_tax_rate,
      photo_count: p.photo_count,
      broker: seller && seller.broker ? seller.broker.name : null,
      remarks: d.text,
      details: Object.fromEntries((p.details || []).filter(c => c && c.category).map(c => [c.category, c.text]))
    };
    return out;
  }).catch(e => ({error: String(e)})).then(r => { clearTimeout(timer); window.__rdcFacts[key] = r; });
  return true;
})"""
 
def on_realtor():
    try:
        return urllib.parse.urlsplit(page_info().get("url", "")).hostname == "www.realtor.com"
    except (TimeoutError, RuntimeError):
        return False
 
def retry_after_s(value, default):
    if not value:
        return default
    try:
        return max(0.0, float(value))
    except ValueError:
        pass
    try:
        return max(0.0, email.utils.parsedate_to_datetime(value).timestamp() - time.time())
    except (TypeError, ValueError):
        return default
 
def fetch_once(url, key, wait_s=40):
    js(f"{START_JS}({json.dumps(url)}, {json.dumps(key)})")
    deadline = time.time() + wait_s
    while time.time() < deadline:
        time.sleep(1)
        try:
            r = js(f"window.__rdcFacts && window.__rdcFacts[{json.dumps(key)}]")
        except (TimeoutError, RuntimeError):  # a slow js() round trip; poll again
            continue
        if r:
            return r
    return {"error": f"no result after {wait_s}s"}
 
def get_facts(row):
    last = None
    for attempt in range(TRIES):
        try:
            r = fetch_once(row["url"], f"{row.get('property_id')}-{attempt}-{time.time()}")
        except (TimeoutError, RuntimeError) as e:
            r = {"error": str(e)[:200]}
        status, f = r.get("http_status"), r.get("facts")
        if status == 200 and f:
            if row.get("property_id") is not None and f.get("property_id") != row["property_id"]:
                raise RuntimeError(f"{row['url']} returned property {f.get('property_id')}, wanted {row['property_id']}")
            return f
        last = {k: r.get(k) for k in ("http_status", "final_url", "error")}
        if isinstance(status, int) and 400 <= status < 500 and status != 429:
            break                           # 404 and friends will not fix themselves
        if status == 200 and "/realestateandhomes-detail/" not in urllib.parse.urlsplit(r.get("final_url") or "").path:
            break                           # redirected away from the listing (seen for an unknown slug)
        if attempt + 1 == TRIES:
            break                           # no retry left, so no sleep
        wait = retry_after_s(r.get("retry_after"), 5 * 2 ** attempt)
        if wait > MAX_WAIT_S:
            raise WaitTooLong(f"{row['url']}: server asked to wait {wait:.0f}s (> {MAX_WAIT_S}s); stopping. last: {last}")
        time.sleep(wait)
    raise RuntimeError(f"{row['url']}: {last}")
 
if not on_realtor():
    try:
        goto_url("https://www.realtor.com/")
    except TimeoutError:
        pass
    for _ in range(30):
        if on_realtor():
            break
        time.sleep(1)
    else:
        raise RuntimeError("need a www.realtor.com tab for same-origin fetch()")
 
listings = json.load(open(os.path.join(tempfile.gettempdir(), "realtor_listings.json")))
todo = [r for r in listings if r.get("url")][:N]
print(f"{len(todo)} listings to enrich ({sum(1 for r in listings if not r.get('url'))} saved rows have no URL)")
facts, failed, stop = [], [], None
for r in todo:
    try:
        f = get_facts(r)
    except WaitTooLong as e:
        stop = e
        print("STOPPED", e)
        break
    except RuntimeError as e:
        failed.append(str(e))
        print("FAILED", e)
    else:
        facts.append(f)
        print(f["address"], f["list_price"], f["beds"], f["baths"], f["sqft"], f["year_built"],
              "hoa", f["hoa_fee"], "dom", f["days_on_mls"], f["status"], f["flags"])
    time.sleep(PAUSE_S)
json.dump(facts, open(os.path.join(tempfile.gettempdir(), "realtor_listing_facts.json"), "w"), indent=1)
print(len(facts), "of", len(todo), "listings enriched")
if stop:
    raise stop
if failed:
    raise RuntimeError(f"{len(failed)} of {len(todo)} listings failed after retries: {failed}")

import json, os, tempfile, time, urllib.parse
 
listings = json.load(open(os.path.join(tempfile.gettempdir(), "realtor_listings.json")))
url = next((r["url"] for r in listings if r.get("url")), None)
if not url:
    raise RuntimeError("no saved listing has a URL; run block 1 first")
 
LD_JS = r"""({href: location.href, ready: document.readyState,
  lds: [...document.querySelectorAll('script[type="application/ld+json"]')].map(s => s.textContent)})"""
 
def same_page(href, want):
    a, b = urllib.parse.urlsplit(href or ""), urllib.parse.urlsplit(want)
    norm = lambda u: urllib.parse.unquote(u.path).rstrip("/").lower()
    return a.netloc == b.netloc and norm(a) == norm(b)
 
def find_listing(lds, want):
    for raw in lds or []:
        try:
            data = json.loads(raw)
        except (TypeError, ValueError):
            continue
        for item in (data if isinstance(data, list) else [data]):
            if not isinstance(item, dict):
                continue
            t = item.get("@type")
            if t == "RealEstateListing" or (isinstance(t, list) and "RealEstateListing" in t):
                ref = item.get("url") or str(item.get("@id") or "").split("#")[0]
                if same_page(ref, want):
                    return item
    return None                     # no listing for this URL; never fall back to another one
 
def load_listing(want, tries=3, wait_s=20):
    last = None
    for attempt in range(tries):
        try:
            goto_url(want)
        except TimeoutError:        # Page.navigate can outlast the IPC wait; the URL check decides
            pass
        deadline = time.time() + wait_s
        while time.time() < deadline:
            try:
                s = js(LD_JS) or {}
            except (TimeoutError, RuntimeError) as e:
                s = {"error": str(e)[:200]}
            if same_page(s.get("href"), want) and s.get("ready") == "complete":
                item = find_listing(s.get("lds"), want)
                if item:
                    return item
            last = {k: s.get(k) for k in ("href", "ready", "error")}
            time.sleep(1)
        if attempt + 1 < tries:     # no sleep after the last load
            time.sleep(5 * 2 ** attempt)
    raise RuntimeError(f"no RealEstateListing JSON-LD for {want}; last seen: {last}")
 
def first_dict(v):
    if isinstance(v, list):
        v = next((x for x in v if isinstance(x, dict)), None)
    return v if isinstance(v, dict) else {}
 
def num(v):
    if isinstance(v, bool):
        return None
    if isinstance(v, (int, float)):
        return v
    if isinstance(v, str):
        try:
            f = float(v.replace(",", ""))
            return int(f) if f.is_integer() else f
        except ValueError:
            return None
    return None
 
def image_urls(v):
    out = []
    for x in (v if isinstance(v, list) else [v]):
        if isinstance(x, str):
            out.append(x)
        elif isinstance(x, dict) and isinstance(x.get("url") or x.get("contentUrl"), str):
            out.append(x.get("url") or x.get("contentUrl"))
    return out
 
listing = load_listing(url)
home, offer = first_dict(listing.get("mainEntity")), first_dict(listing.get("offers"))
size = home.get("floorSize")
print({
    "name": listing.get("name"),
    "price": num(offer.get("price")),
    "availability": offer.get("availability"),
    "beds": num(home.get("numberOfBedrooms")),
    "baths": num(home.get("numberOfBathroomsTotal")),
    "sqft": num(size.get("value") if isinstance(size, dict) else size),
    "year_built": num(home.get("yearBuilt")),
    "type": home.get("@type"),
    "date_posted": listing.get("datePosted"),
    "photos": len(image_urls(home.get("photo"))),
    "summary": listing.get("description"),
})