Idealista Scraper in Python: Extract Property Data to CSV

python -m venv .venv

.\.venv\Scripts\Activate.ps1

source .venv/bin/activate

python -m pip install requests==2.34.2 beautifulsoup4==4.15.0

python idealista_scraper.py --output idealista_listings

document.querySelectorAll("YOUR_CARD_SELECTOR").length
document.querySelector("YOUR_PRICE_SELECTOR")?.textContent.trim()

python idealista_scraper.py --write-demo saved_html

python idealista_scraper.py --url "https://www.idealista.com/YOUR_SEARCH_PATH/" --selectors saved_html/selectors.json --max-pages 1 --limit 3 --delay 2 --output property_check

python idealista_scraper.py --url "https://www.idealista.com/YOUR_SEARCH_PATH/" --selectors saved_html/selectors.json --max-pages 5 --limit 50 --delay 2 --output property_listings

python idealista_scraper.py --write-demo saved_html
python idealista_scraper.py --manifest saved_html/manifest.json --selectors saved_html/selectors.json --output saved_listings

{
  "start_url": "https://example.test/en/venta-viviendas/madrid/",
  "search_pages": {
    "https://example.test/en/venta-viviendas/madrid/": "search-1.html",
    "https://example.test/en/venta-viviendas/madrid/pagina-2.html": "search-2.html"
  },
  "details": {
    "1001": "detail-1001.html",
    "1002": "detail-1002.html",
    "1003": "detail-1003.html"
  }
}

"""Request or read property HTML, parse fields, and export CSV/JSON."""
import argparse
import csv
import json
import os
import time
import re
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup

BASE = "https://example.test/en/venta-viviendas/madrid/"
PAGES = {
    BASE: '''
      <article class="listing-card">
        <a class="listing-link" href="/en/inmueble/1001/">Piso en Chamberi</a>
        <span class="price">455.000 €</span><span class="location">Chamberi, Madrid</span>
      </article>
      <article class="listing-card">
        <a class="listing-link" href="/en/inmueble/1002/">Piso en Retiro</a>
        <span class="price">389.000 €</span><span class="location">Retiro, Madrid</span>
      </article>
      <a rel="next" href="pagina-2.html">Next</a>
    ''',
    urljoin(BASE, "pagina-2.html"): '''
      <article class="listing-card">
        <a class="listing-link" href="/en/inmueble/1003/">Atico en Tetuan</a>
        <span class="price">520.000 €</span><span class="location">Tetuan, Madrid</span>
      </article>
    ''',
}
DETAILS = {
    "1001": '''<h1>Piso en Chamberi</h1><span class="price">455.000 €</span>
      <ul><li class="feature">82 m²</li><li class="feature">2 rooms</li></ul>
      <div class="description">Bright apartment near the metro.</div>''',
    "1002": '''<h1>Piso en Retiro</h1><span class="price">389.000 €</span>
      <ul><li class="feature">67 m²</li><li class="feature">2 rooms</li></ul>
      <div class="description">Two-bedroom flat close to the park.</div>''',
    "1003": '''<h1>Atico en Tetuan</h1><span class="price">520.000 €</span>
      <ul><li class="feature">91 m²</li><li class="feature">3 rooms</li></ul>
      <div class="description">Top-floor flat with a terrace.</div>''',
}
FIELDS = ["listing_id", "title", "price_eur", "location", "size_m2", "rooms",
          "description", "property_url", "scraped_at_utc"]
SELECTORS = {
    "card": "article.listing-card",
    "link": 'a.listing-link[href*="/inmueble/"]',
    "card_price": ".price", "location": ".location",
    "next": 'a[rel="next"][href]', "title": "h1",
    "detail_price": ".price", "features": ".feature",
    "description": ".description",
    "decimal_separator": ",", "thousands_separator": ".",
}

def text(node):
    return node.get_text(" ", strip=True) if node else ""

def number(value, config):
    # One explicit locale for the input batch; reject unexpected formats.
    value = re.sub(r"[\s\u00a0\u202f]", "", value)
    decimal = config["decimal_separator"]
    thousands = config["thousands_separator"]
    grouped = rf"(?:[0-9]+|[0-9]{{1,3}}(?:{re.escape(thousands)}[0-9]{{3}})+)"
    pattern = grouped + rf"(?:{re.escape(decimal)}[0-9]+)?"
    if not re.fullmatch(pattern, value):
        raise ValueError(f"Unexpected number format: {value!r}; check locale settings")
    try:
        result = Decimal(value.replace(thousands, "").replace(decimal, "."))
    except InvalidOperation as exc:
        raise ValueError(f"Invalid number: {value!r}") from exc
    return int(result) if result == result.to_integral_value() else float(result)

def euro_amount(value, config):
    if not value:
        return None
    if "€" not in value and "EUR" not in value.upper():
        raise ValueError(f"Expected an EUR price: {value!r}")
    return number(re.sub(r"EUR|€", "", value, flags=re.I).strip(), config)

def parse_search(html, page_url, config=SELECTORS):
    soup = BeautifulSoup(html, "html.parser")
    items = []
    for card in soup.select(config["card"]):
        link = card.select_one(config["link"])
        if not link or not link.get("href"):
            continue
        url = urljoin(page_url, link["href"])
        match = re.search(r"/inmueble/([0-9]+)(?:/|$)", urlparse(url).path)
        if not match:
            continue
        # Initialize every field before optional detail enrichment.
        item = dict.fromkeys(FIELDS)
        item.update({
            "listing_id": match.group(1), "title": text(link),
            "price_eur": euro_amount(text(card.select_one(config["card_price"])), config),
            "location": text(card.select_one(config["location"])), "property_url": url,
        })
        items.append(item)
    nxt = soup.select_one(config["next"])
    return items, urljoin(page_url, nxt["href"]) if nxt and nxt.get("href") else None

def parse_detail(html, config=SELECTORS):
    soup = BeautifulSoup(html, "html.parser")
    features = " | ".join(text(x) for x in soup.select(config["features"]))
    size = re.search(r"([0-9][0-9., \u00a0\u202f]*)\s*m[²2]", features, re.I)
    rooms = re.search(r"([0-9]+)\s*(?:rooms?|habitaci[oó]n(?:es)?|bedrooms?|beds?|bed)\b", features, re.I)
    return {
        "title": text(soup.select_one(config["title"])),
        "price_eur": euro_amount(text(soup.select_one(config["detail_price"])), config),
        "size_m2": number(size.group(1).strip(), config) if size else None,
        "rooms": int(rooms.group(1)) if rooms else None,
        "description": text(soup.select_one(config["description"])),
    }

def build_authorized_session():
    session = requests.Session()
    session.headers["User-Agent"] = "AuthorizedRealEstateResearch/1.0"
    proxy_url = os.getenv("ROLA_PROXY_URL")
    if proxy_url:
        session.proxies.update({"http": proxy_url, "https": proxy_url})
    return session

def fetch_authorized_html(session, url, max_attempts=2):
    '''Fetch an explicitly authorized page with bounded retry behavior.'''
    host = (urlparse(url).hostname or "").lower()
    if host not in {"idealista.com", "www.idealista.com"}:
        raise ValueError("Use only an explicitly authorized Idealista URL")
    if urlparse(url).scheme != "https" or not 1 <= max_attempts <= 2:
        raise ValueError("Use HTTPS and 1 or 2 attempts")
    for attempt in range(1, max_attempts + 1):
        try:
            response = session.get(url, timeout=(10, 25), allow_redirects=False)
        except requests.Timeout as exc:
            if attempt == max_attempts:
                raise RuntimeError("Request timed out after bounded attempts") from exc
            time.sleep(1)
            continue
        except requests.exceptions.ProxyError as exc:
            raise RuntimeError("Proxy connection or authentication failed; check endpoint and credentials") from exc
        except requests.ConnectionError as exc:
            raise RuntimeError("Connection failed; verify DNS, host, port and the authorized network path") from exc
        except requests.RequestException as exc:
            raise RuntimeError(f"Network error; stop and inspect access policy: {exc}") from exc
        if response.status_code == 407:
            raise RuntimeError("HTTP 407; check proxy authentication")
        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After")
            raise RuntimeError(f"HTTP 429; stop this run. Retry-After={retry_after!r}")
        if response.status_code in (401, 403):
            raise RuntimeError(f"HTTP {response.status_code}; stop and confirm authorization")
        if 300 <= response.status_code < 400:
            raise RuntimeError("Redirect returned; inspect the permitted workflow")
        response.raise_for_status()
        if "captcha" in response.url.lower() or "captcha" in response.text.lower():
            raise RuntimeError("Verification page returned; stop the run")
        return response.text
    raise RuntimeError("Request stopped after bounded attempts")

def scrape_pages(pages, details, start_url, max_pages=2, limit=10, config=SELECTORS, fetch_html=None):
    rows, seen, visited, page_url = [], set(), set(), start_url
    for page_no in range(1, max_pages + 1):
        if page_url in visited:
            break
        visited.add(page_url)
        if fetch_html:
            html = fetch_html(page_url)
        else:
            if page_url not in pages:
                raise ValueError(f"No saved search HTML mapped for {page_url}")
            html = pages[page_url]
        listings, next_url = parse_search(html, page_url, config)
        print(f"search page {page_no}: {len(listings)} cards; next={'yes' if next_url else 'no'}")
        for item in listings:
            lid = item["listing_id"]
            if lid in seen:
                print(f"  duplicate skipped: {lid}")
                continue
            seen.add(lid)
            detail_html = fetch_html(item["property_url"]) if fetch_html else details.get(lid, "")
            detail = parse_detail(detail_html, config)
            for key, value in detail.items():
                if value not in (None, ""):
                    item[key] = value
            item["scraped_at_utc"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
            rows.append(item)
            print(f"  parsed {lid}: EUR {item['price_eur']}, {item['size_m2']} m2, {item['rooms']} rooms")
            if len(rows) >= limit:
                return rows
        if not next_url:
            break
        page_url = next_url
    return rows

def validate(rows):
    if not rows:
        raise ValueError("No records parsed; inspect the HTML and selectors")
    ids = [row["listing_id"] for row in rows]
    if any(not lid for lid in ids) or len(ids) != len(set(ids)):
        raise ValueError("Missing or duplicate listing IDs")
    for row in rows:
        if row["price_eur"] is not None and row["price_eur"] <= 0:
            raise ValueError("Invalid non-positive price")
        if row["size_m2"] is not None and row["size_m2"] <= 0:
            raise ValueError("Invalid non-positive area")
        if urlparse(row["property_url"]).scheme not in {"http", "https"}:
            raise ValueError("Expected an absolute HTTP(S) listing URL")
    missing = {field: sum(row.get(field) in (None, "") for row in rows) for field in FIELDS}
    print(f"quality check: {len(rows)} rows; unique IDs={len(set(ids))}")
    print("missing values:", json.dumps(missing, ensure_ascii=False))

def export(rows, basename):
    with open(f"{basename}.csv", "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(rows)
    with open(f"{basename}.json", "w", encoding="utf-8") as f:
        json.dump(rows, f, ensure_ascii=False, indent=2)
    print(f"saved {len(rows)} rows to {basename}.csv and {basename}.json")

def load_saved(manifest_path):
    path = Path(manifest_path).resolve()
    manifest = json.loads(path.read_text(encoding="utf-8"))
    def read_map(mapping):
        return {key: (path.parent / filename).read_text(encoding="utf-8")
                for key, filename in mapping.items()}
    return read_map(manifest["search_pages"]), read_map(manifest.get("details", {})), manifest["start_url"]

def write_demo(directory):
    directory = Path(directory)
    directory.mkdir(parents=True, exist_ok=True)
    manifest = {"start_url": BASE, "search_pages": {}, "details": {}}
    for i, (url, html) in enumerate(PAGES.items(), 1):
        filename = f"search-{i}.html"
        (directory / filename).write_text(html, encoding="utf-8")
        manifest["search_pages"][url] = filename
    for lid, html in DETAILS.items():
        filename = f"detail-{lid}.html"
        (directory / filename).write_text(html, encoding="utf-8")
        manifest["details"][lid] = filename
    (directory / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    (directory / "selectors.json").write_text(json.dumps(SELECTORS, indent=2), encoding="utf-8")
    print(f"wrote demo HTML, manifest.json and selectors.json to {directory}")

def main():
    parser = argparse.ArgumentParser(description="Parse bundled or saved property HTML")
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--url", help="HTTPS search-page URL")
    mode.add_argument("--manifest", help="JSON mapping original page URLs and listing IDs to local files")
    mode.add_argument("--write-demo", metavar="DIR", help="Write sample HTML and configuration files, then exit")
    parser.add_argument("--selectors", help="JSON overrides for CSS selectors and number separators")
    parser.add_argument("--max-pages", type=int, default=2)
    parser.add_argument("--delay", type=float, default=2, help="Seconds between page requests")
    parser.add_argument("--limit", type=int, default=10)
    parser.add_argument("--output", default="idealista_listings")
    args = parser.parse_args()
    if args.max_pages < 1 or args.limit < 1 or args.delay < 0:
        parser.error("max-pages and limit must be positive")
    if args.write_demo:
        write_demo(args.write_demo)
        return
    config = dict(SELECTORS)
    if args.selectors:
        overrides = json.loads(Path(args.selectors).read_text(encoding="utf-8"))
        if not isinstance(overrides, dict) or set(overrides) - set(SELECTORS):
            parser.error("selectors must be a JSON object containing known configuration keys")
        config.update(overrides)
    if ({config["decimal_separator"], config["thousands_separator"]} != {".", ","}):
        parser.error("number separators must be distinct dot/comma characters")
    session = None
    fetch_html = None
    if args.url:
        pages, details, start_url = {}, {}, args.url
        session = build_authorized_session()
        last_request = None
        def fetch_html(url):
            nonlocal last_request
            if last_request is not None:
                time.sleep(max(0, args.delay - (time.monotonic() - last_request)))
            last_request = time.monotonic()
            return fetch_authorized_html(session, url)
        print("mode: HTTP search and detail pages")
    elif args.manifest:
        pages, details, start_url = load_saved(args.manifest)
        print("mode: saved HTML; no HTTP requests")
    else:
        pages, details, start_url = PAGES, DETAILS, BASE
        print("mode: bundled fictional HTML; no HTTP requests")
    try:
        rows = scrape_pages(pages, details, start_url, args.max_pages, args.limit, config, fetch_html)
        validate(rows)
        export(rows, args.output)
    finally:
        if session is not None:
            session.close()

if __name__ == "__main__":
    main()

python -m json.tool idealista_listings.json
python -c "import csv; r=list(csv.DictReader(open('idealista_listings.csv', encoding='utf-8-sig'))); print(len(r), len({x['listing_id'] for x in r}))"

# Windows PowerShell
$env:ROLA_PROXY_URL = "http://USERNAME:PASSWORD@HOST:PORT"

# macOS or Linux
export ROLA_PROXY_URL="http://USERNAME:PASSWORD@HOST:PORT"