Скрейпинг недвижимости: Практическое руководство по данным о недвижимости

TL;DR

Сбор данных в сфере недвижимости собирает опубликованную информацию о недвижимости, в то время как MLS-ленты, зарегистрированные продажи и порталы объявлений предоставляют разные наборы данных и права.
Держите идентичность объявления отдельно от идентичности недвижимости и храните наблюдения, а не перезаписывайте историю.
Нормализуйте валюту, продажу против аренды, период аренды, единицу площади и временную метку перед сравнением цен.
Этот учебник включает в себя исполняемый парсер для двух локальных синтетических HTML-страниц; он не обращается к порталу недвижимости.
RESO определяет стандарты, а не бесплатную базу данных MLS. Доступ к производству предоставляется соответствующим поставщиком и лицензией.

Что такое сбор данных в сфере недвижимости и как это работает?

Сбор данных в сфере недвижимости извлекает поля, связанные с недвижимостью, из разрешенных источников в структурированный набор данных. Обычно рабочий процесс включает в себя поиск страниц объявлений, получение одобренного контента, разбор полей, валидацию и удаление дубликатов записей, а также хранение наблюдений для последующего сравнения.

Отличайте активные цены запроса, зарегистрированные транзакции, предложения по аренде, оценки и рыночные агрегаты. Запрашиваемая цена на портале не является зарегистрированной ценой продажи, а оценка может использовать другую дату и метод оценки.

Какие данные о недвижимости можно извлечь?

Доступные поля зависят от источника, региона и соглашения о доступе. Определите необходимый набор данных перед выбором краулера.

Поле	Пример значения	Является явным
ID объявления	Запись предложения одного источника	Пространственное пространство и правила повторной публикации
ID недвижимости	Идентичность физического актива	Уверенность в совпадении и идентификатор поставщика

Источник	Типичная сила	Ограничение
Разрешенные страницы портала	Отображаемые текущие предложения	Ограничения доступа, дублирование, изменения разметки
Лицензированная MLS/IDX лента	Структурированные данные о списках	Квалификация, цель и ограничения на перераспределение
Государственные записи	Зарегистрированные факты о недвижимости или транзакциях	Юрисдикция, задержка, схема и политика доступа
API коммерческой недвижимости	Стандартизированная интеграция	Объем продукта и лицензия
Веб-сайт вашего агентства	Контролируемая публикация и авторизованное тестирование	Ограниченное покрытие рынка

"""Извлечение синтетических объявлений с локальных HTML страниц; доступ к порталу не требуется."""
import argparse
import csv
import json
import sqlite3
from datetime import datetime, timezone
from decimal import Decimal
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import urljoin


class Listings(HTMLParser):
    def __init__(self):
        super().__init__()
        self.rows, self.next_page = [], None
        self.row, self.title = None, False

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "article" and "listing" in a.get("class", "").split():
            self.row = dict(listing_id=a["data-id"], title="",
                            amount=a["data-price"], currency=a["data-currency"],
                            area=a["data-area"], area_unit=a["data-area-unit"],
                            location=a["data-location"], offer_type=a["data-offer"])
        if tag == "h2" and self.row is not None:
            self.title = True
        if tag == "a" and self.row is not None and a.get("class") == "details":
            self.row["url"] = urljoin("https://property-demo.example/", a["href"])
        if tag == "a" and "next" in a.get("rel", "").split():
            self.next_page = a["href"]

    def handle_data(self, data):
        if self.title:
            self.row["title"] += data

    def handle_endtag(self, tag):
        if tag == "h2":
            self.title = False
        if tag == "article" and self.row is not None:
            self.rows.append(self.row)
            self.row = None


def normalize(row):
    price, area = Decimal(row.pop("amount")), Decimal(row.pop("area"))
    if not price.is_finite() or not area.is_finite() or price <= 0 or area <= 0:
        raise ValueError("Неверная цена или площадь")
    unit = row.pop("area_unit")
    if unit not in ("sqft", "sqm") or row["offer_type"] not in ("sale", "rent_monthly"):
        raise ValueError("Неизвестная единица или тип предложения")
    area_sqm = area * Decimal("0.09290304") if unit == "sqft" else area
    return {**row, "title": row["title"].strip(), "price": str(price),
            "area_sqm": str(area_sqm.quantize(Decimal("0.01")))}


def collect(root):
    root = root.resolve()
    current, seen_pages, rows = root / "page1.html", set(), {}
    for _ in range(5):
        current = current.resolve()
        if not current.is_relative_to(root) or current in seen_pages:
            raise ValueError("Небезопасный или повторяющийся путь страницы")
        seen_pages.add(current)
        parsed = Listings()
        parsed.feed(current.read_text())
        for raw in parsed.rows:
            row = normalize(raw)
            key = row["listing_id"]
            if key in rows and rows[key] != row:
                raise ValueError("Конфликтующее дублирующее объявление")
            rows[key] = row
if not parsed.next_page:
            return list(rows.values())
        current = current.parent / parsed.next_page
    raise ValueError("Превышен лимит пагинации")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--fixtures", default="property_fixture")
    parser.add_argument("--db", default="property.sqlite")
    args = parser.parse_args()
    rows = collect(Path(args.fixtures))
    observed = datetime.now(timezone.utc).isoformat()
    if not rows:
        raise SystemExit("Нет принятых объявлений")
    with sqlite3.connect(args.db) as db:
        db.execute("""CREATE TABLE IF NOT EXISTS observations(
            listing_id TEXT, observed_at TEXT, payload TEXT,
            PRIMARY KEY(listing_id,observed_at))""")
        db.executemany("INSERT INTO observations VALUES(?,?,?)",
                       [(row["listing_id"], observed, json.dumps(row)) for row in rows])
    Path("property.json").write_text(json.dumps(rows, indent=2))
    with open("property.csv", "w", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)
    print(f"accepted_unique_listings={len(rows)}")

Сбор данных о недвижимости: Руководство по сбору данных о государственной собственности

TL;DR

Что такое сбор данных в сфере недвижимости и как это работает?

Какие данные о недвижимости можно извлечь?

Популярные статьи

Сбор данных о недвижимости: Руководство по сбору данных о государственной собственности

Dify SERP Scraping: Как создать рабочий процесс данных поиска

Объявления о недвижимости и данные о ценах

Местоположение, особенности недвижимости и история объявления

Источники данных о недвижимости: Порталы, MLS и публичные записи

Методики сбора данных в сфере недвижимости по сравнению

Python Requests и Beautiful Soup

Playwright для рендеринга объявлений на JavaScript

Управляемые API веб-сканирования

Шаг за Шагом: Создание парсера недвижимости на Python

Шаг 1: Изучите структуру страницы объявления

Шаг 2: Извлеките объявления о недвижимости

Шаг 3: Обработка пагинации и дублирующихся объявлений

Шаг 4: Очистка и нормализация данных о недвижимости

Шаг 5: Хранение объявлений в CSV или базе данных

Проблемы и решения в области извлечения данных о недвижимости

Как построить конвейер мониторинга цен на недвижимость

Извлечение данных о недвижимости для анализа рынка

Масштабирование сбора данных о недвижимости

Расходы на сбор данных о недвижимости: API против саморазмещения

Правовые, конфиденциальные и лицензионные соображения по данным

Заключительный контрольный список для надежных конвейеров данных о недвижимости

Проверьте утвержденные страницы свойств

ЧаВо