Rightmove - Scrape for-sale and to-rent listings

import re, json, time, gzip, email.utils, urllib.parse, urllib.request, urllib.error
from datetime import timezone
 
HEADERS = {
    "User-Agent": "browser-harness-rightmove-skill/1.0 (+https://github.com/browser-use/browser-harness)",  # truthful client UA
    "Accept-Language": "en-GB,en;q=0.9",
    "Accept-Encoding": "gzip",
}
RETRY_STATUS = {429, 500, 502, 503, 504}
MAX_WAIT = 120  # seconds; a longer Retry-After stops instead of sleeping
PER_PAGE, MAX_INDEX = 24, 984  # Rightmove serves at most 42 pages per search
CHANNEL_PATHS = {"buy": "property-for-sale", "rent": "property-to-rent"}
 
def _retry_after(value):
    """Retry-After is either seconds or an HTTP-date. Returns seconds or None."""
    if not value:
        return None
    value = value.strip()
    if value.isdigit():
        return float(value)
    try:
        dt = email.utils.parsedate_to_datetime(value)
    except (TypeError, ValueError):
        return None
    if dt is None:
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return max(0.0, dt.timestamp() - time.time())
 
def _decode(body, headers):
    if headers.get("Content-Encoding") == "gzip":
        body = gzip.decompress(body)
    return body.decode("utf-8", "replace")
 
def fetch(url, accept="text/html", tries=4):
    """GET -> (status, final_url, text). Retries 429/5xx/network errors with backoff, then raises.
    Raises at once on a 403/429 "Rightmove Group Limited only" access block or a Retry-After over MAX_WAIT.
    Other HTTP errors (400, 404, 410) are returned for the caller to handle."""
    last = None
    for attempt in range(tries):
        wait = None
        try:
            req = urllib.request.Request(url, headers={**HEADERS, "Accept": accept})
            with urllib.request.urlopen(req, timeout=30) as r:
                return r.status, r.geturl(), _decode(r.read(), r.headers)
        except urllib.error.HTTPError as e:
            body = _decode(e.read(), e.headers)
            if e.code in (403, 429) and "for the use of Rightmove Group Limited only" in body:
                raise RuntimeError(f"HTTP {e.code} access block on {url}; stop and back off, do not retry")
            if e.code not in RETRY_STATUS:
                return e.code, url, body
            last, wait = f"HTTP {e.code}", _retry_after(e.headers.get("Retry-After"))
        except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
            last = repr(e)
        if attempt == tries - 1:
            break  # no sleep after the final attempt
        delay = wait if wait is not None else 2 * 2 ** attempt
        if delay > MAX_WAIT:
            raise RuntimeError(f"{url}: server asked to wait {delay:.0f}s (over {MAX_WAIT}s); stopping")
        time.sleep(delay)
    raise RuntimeError(f"giving up on {url} after {tries} tries: {last}")
 
def _norm(s):
    return " ".join(re.sub(r"[^\w\s]", " ", s or "").lower().split())
 
def find_location(query):
    """Free text -> list of (locationIdentifier, displayName), best first. Punctuation breaks the endpoint (HTTP 400)."""
    url = "https://los.rightmove.co.uk/typeahead?" + urllib.parse.urlencode({"query": _norm(query), "limit": 10, "exclude": "STREET"})
    status, _, text = fetch(url, accept="application/json")  # without this Accept you get XML
    if status != 200:
        raise RuntimeError(f"typeahead HTTP {status} for {query!r}")
    return [(f"{m['type']}^{m['id']}", m.get("displayName")) for m in json.loads(text).get("matches") or []]
 
def pick_location(query):
    """First typeahead match whose displayName starts with the query. Raises instead of guessing."""
    matches = find_location(query)
    for loc_id, name in matches:
        if _norm(name).startswith(_norm(query)):
            return loc_id, name
    raise LookupError(f"no Rightmove location starting with {query!r}; candidates: {matches}")
 
def search_page(location_id, channel="buy", index=0, **filters):
    """One results page. None if Rightmove redirected to /page-not-found (unknown location or index past 984)."""
    if channel not in CHANNEL_PATHS:
        raise ValueError(f"channel must be 'buy' or 'rent', got {channel!r}")
    path = CHANNEL_PATHS[channel]
    params = {"locationIdentifier": location_id, "index": index, **filters}
    url = f"https://www.rightmove.co.uk/{path}/find.html?" + urllib.parse.urlencode(params)  # ^ -> %5E
    status, final_url, html = fetch(url)
    if status != 200:
        raise RuntimeError(f"HTTP {status} for {url}")
    if urllib.parse.urlparse(final_url).path == "/page-not-found":
        return None
    m = re.search(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', html, re.S)
    if not m:
        raise RuntimeError(f"no __NEXT_DATA__ on {final_url} (blocked or layout changed)")
    sr = json.loads(m.group(1))["props"]["pageProps"]["searchResults"]
    sp = sr.get("searchParameters") or {}
    if sp.get("index") != str(index) or sp.get("locationIdentifier") != location_id:
        raise RuntimeError(f"asked for {location_id} index {index}, got {sp.get('locationIdentifier')} index {sp.get('index')}")
    return sr
 
def _d(x):
    return x if isinstance(x, dict) else {}
 
def _num(x):
    return x if isinstance(x, (int, float)) and not isinstance(x, bool) else None
 
def _text(x):
    return " ".join(x.split()) if isinstance(x, str) else None
 
def to_row(p):
    price = _d(p.get("price"))
    dps = price.get("displayPrices") if isinstance(price.get("displayPrices"), list) else []
    dp = _d(dps[0]) if dps else {}
    pid = p.get("id")
    return {
        "id": pid,
        "address": _text(p.get("displayAddress")),
        "price": _num(price.get("amount")),
        "price_text": " ".join(x for x in [dp.get("displayPriceQualifier"), dp.get("displayPrice")] if isinstance(x, str) and x),
        "frequency": price.get("frequency"),
        "beds": _num(p.get("bedrooms")),
        "baths": _num(p.get("bathrooms")),
        "type": p.get("propertySubType"),
        "agent": _d(p.get("customer")).get("branchDisplayName"),
        "added_or_reduced": p.get("addedOrReduced"),
        "first_visible": p.get("firstVisibleDate"),
        "featured": p.get("featuredProperty"),
        "url": f"https://www.rightmove.co.uk/properties/{pid}" if pid is not None else None,
    }
 
def _count(x):
    s = str(x or "").replace(",", "")
    return int(s) if s.isdigit() else None
 
def scrape(query, channel="buy", max_pages=2, **filters):
    loc_id, name = pick_location(query)
    rows, seen, total, pages, stopped = [], set(), None, 0, "max_pages"
    for n in range(max_pages):
        index = n * PER_PAGE
        if index > MAX_INDEX:
            stopped = "rightmove_page_cap"
            break
        sr = search_page(loc_id, channel, index=index, **filters)
        if sr is None:
            raise RuntimeError(f"{loc_id} index {index} redirected to /page-not-found")
        pages += 1
        total = _count(sr.get("resultCount"))
        props = sr.get("properties") or []
        for p in props:
            pid = p.get("id")
            if pid is not None and pid in seen:  # featured listings also show up organically on other pages
                continue
            if pid is not None:
                seen.add(pid)
            rows.append(to_row(p))
        has_next = bool(_d(sr.get("pagination")).get("next"))
        if not has_next:
            capped = index == MAX_INDEX and total is not None and total > MAX_INDEX + PER_PAGE
            stopped = "rightmove_page_cap" if capped else "last_page"
            break
        if not props:
            raise RuntimeError(f"empty page at index {index} but pagination says there is more")
        time.sleep(1.5)  # polite pacing
    return {"location": name, "location_id": loc_id, "result_count": total,
            "pages": pages, "stopped": stopped, "rows": rows}
 
for query, channel, max_pages, filters in [
    ("Bristol", "buy", 2, {"sortType": 6, "minBedrooms": 2, "maxPrice": 400000}),
    ("Camden Town", "rent", 20, {}),
]:
    res = scrape(query, channel=channel, max_pages=max_pages, **filters)
    print(res["location"], res["location_id"], "results:", res["result_count"], "pages:", res["pages"],
          "stopped:", res["stopped"], "rows:", len(res["rows"]))
    for r in res["rows"][:2]:
        print(r)

import json, time, html as htmllib, urllib.parse
 
def _is_timeout(e):
    return isinstance(e, TimeoutError) or "timed out" in str(e)
 
def retry_browser(fn, *args, tries=3):
    """Browser calls can time out on a slow cloud browser: TimeoutError, or RuntimeError('... timed out ...')."""
    for attempt in range(tries):
        try:
            return fn(*args)
        except (TimeoutError, RuntimeError) as e:
            if not _is_timeout(e) or attempt == tries - 1:
                raise
            time.sleep(2)
 
def open_and_confirm(url, check_js, first=False, timeout=30):
    """Navigate, then poll check_js until it returns 'ok' (True) or 'not_found' (False)."""
    retry_browser(new_tab if first else goto_url, url)
    if first:
        activate_tab(current_tab())  # a hidden background tab stalled in the cloud browser
    deadline = time.time() + timeout
    while time.time() < deadline:
        state = retry_browser(js, check_js)
        if state in ("ok", "not_found"):
            return state == "ok"
        time.sleep(0.5)
    raise TimeoutError(f"{url} not confirmed after {timeout}s; tab is at {retry_browser(js, 'location.href')}")
 
def search_check(location_id, index):
    return f"""(() => {{
      if (location.pathname === '/page-not-found') return 'not_found';
      const s = document.getElementById('__NEXT_DATA__');
      if (!s) return 'wait';
      const sp = JSON.parse(s.textContent)?.props?.pageProps?.searchResults?.searchParameters || {{}};
      return sp.index === {json.dumps(str(index))} && sp.locationIdentifier === {json.dumps(location_id)} ? 'ok' : 'wait';
    }})()"""
 
def property_check(property_id):
    return f"""(() => {{
      if (location.pathname !== {json.dumps(f'/properties/{property_id}')}) return 'wait';
      const nav = performance.getEntriesByType('navigation')[0];
      if (nav && [404, 410].includes(nav.responseStatus)) return 'not_found';
      return window.__PAGE_MODEL ? 'ok' : 'wait';
    }})()"""
 
def decode_page_model(pm):
    """window.__PAGE_MODEL = {"data": "<json array>", "encoding": "on"}.
    Every value inside an object/array is an index into that array; resolve recursively."""
    arr = json.loads(pm["data"])
    memo = {}
    def r(i):
        if not isinstance(i, int) or isinstance(i, bool) or not 0 <= i < len(arr):
            return None
        if i in memo:
            return memo[i]
        v = arr[i]
        if isinstance(v, dict):
            out = memo[i] = {}
            out.update({k: r(x) for k, x in v.items()})
        elif isinstance(v, list):
            out = memo[i] = []
            out.extend(r(x) for x in v)
        else:
            out = v
        return out
    return r(0)
 
def _d(x):
    return x if isinstance(x, dict) else {}
 
def _station(s):
    s = _d(s)
    dist = s.get("distance")
    return (s.get("name"), round(dist, 2) if isinstance(dist, (int, float)) and not isinstance(dist, bool) else None, s.get("unit"))
 
def get_property(property_id, first=False):
    """Decoded property page, or None if the listing is gone (404/410)."""
    url = f"https://www.rightmove.co.uk/properties/{property_id}"
    if not open_and_confirm(url, property_check(property_id), first=first):
        return None
    pm = json.loads(retry_browser(js, "JSON.stringify(window.__PAGE_MODEL)"))
    pd = _d(_d(decode_page_model(pm)).get("propertyData"))
    if str(pd.get("id")) != str(property_id):
        raise RuntimeError(f"asked for {property_id}, page model is for {pd.get('id')}")
    addr = _d(pd.get("address"))
    return {
        "id": pd.get("id"),
        "address": " ".join((addr.get("displayAddress") or "").split()),
        "postcode": " ".join(x for x in [addr.get("outcode"), addr.get("incode")] if isinstance(x, str) and x),
        "price": _d(pd.get("prices")).get("primaryPrice"),
        "beds": pd.get("bedrooms"),
        "baths": pd.get("bathrooms"),
        "type": pd.get("propertySubType"),
        "tenure": _d(pd.get("tenure")).get("tenureType"),
        "council_tax_band": _d(pd.get("livingCosts")).get("councilTaxBand"),
        "key_features": [htmllib.unescape(k) for k in pd.get("keyFeatures") or [] if isinstance(k, str)],  # may be empty
        "added": _d(pd.get("listingHistory")).get("listingUpdateReason"),
        "agent": _d(pd.get("customer")).get("branchDisplayName"),
        "stations": [_station(s) for s in pd.get("nearestStations") or []],
        "description_html": _d(pd.get("text")).get("description"),
        "lettings": pd.get("lettings"),  # deposit / let type / furnishing for rentals, None for sales
    }
 
def first_ids(path, location_id, n=2, first=False):
    url = f"https://www.rightmove.co.uk/{path}/find.html?" + urllib.parse.urlencode({"locationIdentifier": location_id, "index": 0})
    if not open_and_confirm(url, search_check(location_id, 0), first=first):
        raise RuntimeError(f"{url} redirected to /page-not-found")
    sr = json.loads(retry_browser(js, "document.getElementById('__NEXT_DATA__').textContent"))["props"]["pageProps"]["searchResults"]
    return [p["id"] for p in sr.get("properties") or [] if p.get("id") is not None and not p.get("featuredProperty")][:n]
 
ids = first_ids("property-for-sale", "REGION^219", first=True)
time.sleep(1.5)
ids += first_ids("property-to-rent", "REGION^85262") + [1]  # 1 = a removed id
for pid in ids:
    time.sleep(1.5)
    prop = get_property(pid)
    if prop is None:
        print(pid, "listing not available (404/410)")
        continue
    print({k: v for k, v in prop.items() if k != "description_html"})
    print(len(prop["description_html"] or ""), "chars of description HTML")

import json, time, urllib.parse
 
def _is_timeout(e):
    return isinstance(e, TimeoutError) or "timed out" in str(e)
 
def retry_browser(fn, *args, tries=3):
    """Browser calls can time out on a slow cloud browser: TimeoutError, or RuntimeError('... timed out ...')."""
    for attempt in range(tries):
        try:
            return fn(*args)
        except (TimeoutError, RuntimeError) as e:
            if not _is_timeout(e) or attempt == tries - 1:
                raise
            time.sleep(2)
 
def open_and_confirm(url, check_js, first=False, timeout=30):
    """Navigate, then poll check_js until it returns 'ok' (True) or 'not_found' (False)."""
    retry_browser(new_tab if first else goto_url, url)
    if first:
        activate_tab(current_tab())  # a hidden background tab stalled in the cloud browser
    deadline = time.time() + timeout
    while time.time() < deadline:
        state = retry_browser(js, check_js)
        if state in ("ok", "not_found"):
            return state == "ok"
        time.sleep(0.5)
    raise TimeoutError(f"{url} not confirmed after {timeout}s; tab is at {retry_browser(js, 'location.href')}")
 
def search_check(location_id, index):
    return f"""(() => {{
      if (location.pathname === '/page-not-found') return 'not_found';
      const s = document.getElementById('__NEXT_DATA__');
      if (!s) return 'wait';
      const sp = JSON.parse(s.textContent)?.props?.pageProps?.searchResults?.searchParameters || {{}};
      return sp.index === {json.dumps(str(index))} && sp.locationIdentifier === {json.dumps(location_id)} ? 'ok' : 'wait';
    }})()"""
 
def property_check(property_id):
    return f"""(() => {{
      if (location.pathname !== {json.dumps(f'/properties/{property_id}')}) return 'wait';
      const nav = performance.getEntriesByType('navigation')[0];
      if (nav && [404, 410].includes(nav.responseStatus)) return 'not_found';
      return window.__PAGE_MODEL ? 'ok' : 'wait';
    }})()"""
 
def decode_page_model(pm):
    arr = json.loads(pm["data"])
    memo = {}
    def r(i):
        if not isinstance(i, int) or isinstance(i, bool) or not 0 <= i < len(arr):
            return None
        if i in memo:
            return memo[i]
        v = arr[i]
        if isinstance(v, dict):
            out = memo[i] = {}
            out.update({k: r(x) for k, x in v.items()})
        elif isinstance(v, list):
            out = memo[i] = []
            out.extend(r(x) for x in v)
        else:
            out = v
        return out
    return r(0)
 
def _d(x):
    return x if isinstance(x, dict) else {}
 
base = "https://www.rightmove.co.uk/property-to-rent/find.html?"
location_id = "REGION^219"
params = {"locationIdentifier": location_id, "maxBedrooms": 2}
rows, seen, stopped = [], set(), "max_pages"
for n in range(2):
    index = n * 24
    if not open_and_confirm(base + urllib.parse.urlencode({**params, "index": index}), search_check(location_id, index), first=(n == 0)):
        raise RuntimeError(f"index {index} redirected to /page-not-found")
    sr = json.loads(retry_browser(js, "document.getElementById('__NEXT_DATA__').textContent"))["props"]["pageProps"]["searchResults"]
    for p in sr.get("properties") or []:
        pid = p.get("id")
        if pid is not None and pid in seen:
            continue
        if pid is not None:
            seen.add(pid)
        dps = _d(p.get("price")).get("displayPrices")
        dp = _d(dps[0]) if isinstance(dps, list) and dps else {}
        rows.append((pid, " ".join((p.get("displayAddress") or "").split()), dp.get("displayPrice"),
                     p.get("bedrooms"), p.get("propertySubType"), _d(p.get("customer")).get("branchDisplayName"), p.get("addedOrReduced")))
    pg = _d(sr.get("pagination"))
    print("page", pg.get("page"), "of", pg.get("total"), "| results", sr.get("resultCount"))
    if not pg.get("next"):
        stopped = "last_page"
        break
    time.sleep(1.5)
print(len(rows), "rows, stopped:", stopped)
for r in rows[:3]:
    print(r)
 
if rows and rows[0][0] is not None:
    pid = rows[0][0]
    time.sleep(1.5)
    if not open_and_confirm(f"https://www.rightmove.co.uk/properties/{pid}", property_check(pid)):
        raise RuntimeError(f"property {pid} answered 404/410")
    pd = _d(decode_page_model(json.loads(retry_browser(js, "JSON.stringify(window.__PAGE_MODEL)"))).get("propertyData"))
    if str(pd.get("id")) != str(pid):
        raise RuntimeError(f"asked for {pid}, page model is for {pd.get('id')}")
    print(pd.get("id"), _d(pd.get("prices")).get("primaryPrice"), pd.get("keyFeatures"), pd.get("lettings"))