diff --git a/tradein-mvp/backend/tests/test_scraper_kit_base_golden_parity.py b/tradein-mvp/backend/tests/test_scraper_kit_base_golden_parity.py new file mode 100644 index 00000000..7c70152c --- /dev/null +++ b/tradein-mvp/backend/tests/test_scraper_kit_base_golden_parity.py @@ -0,0 +1,363 @@ +"""Golden-parity: `scraper_kit.base.save_listings` ≡ `app.services.scrapers.base.save_listings`. + +Strangler-инвариант (#2132): новая scraper_kit-копия базового слоя должна давать +БАЙТ-ИДЕНТИЧНЫЙ эффект старому боевому коду на одинаковом входе. Единственное +намеренное расхождение — развязка `region_code`: старый хардкодит литерал `66` в +INSERT SQL, новый получает его параметром `:region_code` (kit не знает про регион). +При region_code=66 итоговый INSERT идентичен — что этот тест и доказывает, +нормализуя `:region_code` → `66` на новой стороне перед сравнением. + +Метод: гоняем ОБА модуля на одном списке `ScrapedLot`, перехватывая через +recording-mock всю последовательность `db.execute(sql, params)` + аргументы hook'а +матчинга. Сравниваем: + - последовательность (SQL, params) вызовов execute (INSERT / SELECT / reconcile + UPDATE / house_id_fk UPDATE) — после нормализации region_code; + - kwargs каждого вызова matcher.match_or_create_house / upsert_listing_source; + - compute_dedup_hash / compute_card_hash / compute_price_per_m2 старого и нового + класса ScrapedLot на идентичном входе. + +Фикстуры покрывают: avito (без адреса), cian (адрес+coords → house resolve → FK +UPDATE), студию rooms=0, yandex без source_id (ext_id := dedup_hash fallback) и +dedup-коллизию (INSERT → UniqueViolation → reconcile UPDATE by source,source_id). +""" + +from __future__ import annotations + +import contextlib +from typing import Any +from unittest.mock import MagicMock, patch + +import psycopg.errors +from scraper_kit.base import ScrapedLot as NewLot +from scraper_kit.base import save_listings as new_save_listings +from sqlalchemy.exc import IntegrityError + +# СТАРЫЙ (боевой) и НОВЫЙ (scraper_kit) модули — импортируем оба в одном тесте. +from app.services.scrapers.base import ScrapedLot as OldLot +from app.services.scrapers.base import save_listings as old_save_listings + +_REGION_CODE = 66 + + +# ── recording session ──────────────────────────────────────────────────────── + + +class _Row: + def __init__(self, **kw: Any) -> None: + self.__dict__.update(kw) + + +class _Result: + def __init__(self, row: _Row | None) -> None: + self._row = row + + def fetchone(self) -> _Row | None: + return self._row + + +@contextlib.contextmanager +def _nested_cm() -> Any: + yield MagicMock() + + +class RecordingSession: + """Мок Session: пишет (sql, params) всех execute, детерминированно отвечает. + + - SELECT card_hash … WHERE dedup_hash → None (prior_row отсутствует, fresh) + - INSERT INTO listings … → row(id, inserted); если source_id в + drift_ids → raise IntegrityError(UniqueViolation) (dedup-коллизия) + - reconcile UPDATE listings SET dedup_hash … → row(id=reconcile_id) + - UPDATE listings … house_id_fk / snapshot EXISTS → None + """ + + def __init__( + self, + *, + drift_ids: frozenset[str] = frozenset(), + update_ids: frozenset[str] = frozenset(), + insert_id: int = 42, + reconcile_id: int = 99, + ) -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + self._drift = drift_ids + self._update = update_ids + self._insert_id = insert_id + self._reconcile_id = reconcile_id + + def execute(self, sql: Any, params: dict[str, Any] | None = None) -> _Result: + s = str(sql) + p = dict(params) if params else {} + self.calls.append((s, p)) + if "INSERT INTO listings (" in s: + sid = p.get("source_id") + if sid in self._drift: + raise IntegrityError("INSERT", p, psycopg.errors.UniqueViolation("dup")) + inserted = sid not in self._update + return _Result(_Row(id=self._insert_id, inserted=inserted)) + if "UPDATE listings" in s and "SET dedup_hash" in s: + return _Result(_Row(id=self._reconcile_id)) + return _Result(None) + + def begin_nested(self) -> Any: + return _nested_cm() + + def commit(self) -> None: + pass + + +# ── fixtures ───────────────────────────────────────────────────────────────── + + +def _fixture_data() -> list[dict[str, Any]]: + """Реалистичные входные dict'ы (общие для old/new ScrapedLot).""" + return [ + # 1. avito без адреса → house resolve пропускается (нет FK UPDATE) + { + "source": "avito", + "source_url": "https://www.avito.ru/ekaterinburg/kvartiry/1-k_987654?context=AB", + "source_id": "987654", + "price_rub": 4_500_000, + "area_m2": 38.0, + "rooms": 1, + "floor": 5, + "total_floors": 9, + }, + # 2. cian full: адрес + coords → house resolve → house_id_fk UPDATE + { + "source": "cian", + "source_url": "https://ekb.cian.ru/sale/flat/123456/", + "source_id": "123456", + "price_rub": 6_200_000, + "address": "Екатеринбург, улица Постовского, 17А", + "lat": 56.8385, + "lon": 60.5940, + "rooms": 2, + "area_m2": 52.0, + "floor": 7, + "total_floors": 16, + "living_area_m2": 35.0, + "bedrooms_count": 2, + "cadastral_number": "66:41:0204016:1234", + "building_cadastral_number": "66:41:0204016:100", + "phones": [{"countryCode": "7", "number": "9012345678"}], + "is_homeowner": True, + "metro_stations": [{"name": "Геологическая", "time": 10}], + "house_source": "cian", + "listing_segment": "vtorichka", + }, + # 3. студия rooms=0 (avito) с адресом → house resolve + { + "source": "avito", + "source_url": "https://www.avito.ru/ekaterinburg/kvartiry/studiya_555", + "source_id": "555", + "price_rub": 3_100_000, + "address": "Екатеринбург, Малышева 51", + "lat": 56.84, + "lon": 60.60, + "rooms": 0, + "area_m2": 24.5, + "floor": 12, + "total_floors": 25, + }, + # 4. yandex без source_id → ext_id := dedup_hash fallback + { + "source": "yandex", + "source_url": "https://realty.yandex.ru/offer/5500000000000000001/?from=serp", + "source_id": None, + "price_rub": 7_200_000, + "address": "Екатеринбург, улица Мира, 10", + "lat": 56.83, + "lon": 60.65, + "rooms": 2, + "area_m2": 48.0, + "yandex_offer_id": "5500000000000000001", + "predicted_price_rub": 7_554_000, + "price_trend": "DECREASED", + "price_previous_rub": 7_500_000, + }, + # 5. dedup-коллизия: INSERT → UniqueViolation → reconcile UPDATE + { + "source": "cian", + "source_url": "https://ekb.cian.ru/sale/flat/DRIFT-1/", + "source_id": "DRIFT-1", + "price_rub": 5_900_000, + "address": "Екатеринбург, Ленина 1", + "lat": 56.838, + "lon": 60.597, + "rooms": 3, + "area_m2": 72.0, + }, + ] + + +_DRIFT_IDS = frozenset({"DRIFT-1"}) + + +# ── run helpers ────────────────────────────────────────────────────────────── + + +def _run_old(lots: list[Any]) -> tuple[RecordingSession, MagicMock, MagicMock]: + db = RecordingSession(drift_ids=_DRIFT_IDS) + with ( + patch("app.services.scrapers.base.match_or_create_house") as m_house, + patch("app.services.scrapers.base.upsert_listing_source") as m_link, + patch("app.services.scrapers.base.upsert_listing_snapshot"), + ): + m_house.return_value = (101, 1.0, "new") + m_link.return_value = None + old_save_listings(db, lots) + return db, m_house, m_link + + +def _run_new(lots: list[Any]) -> tuple[RecordingSession, MagicMock]: + db = RecordingSession(drift_ids=_DRIFT_IDS) + matcher = MagicMock() + matcher.match_or_create_house.return_value = (101, 1.0, "new") + matcher.upsert_listing_source.return_value = None + with patch("scraper_kit.base.upsert_listing_snapshot"): + new_save_listings(db, lots, matcher=matcher, region_code=_REGION_CODE) + return db, matcher + + +def _normalize_new_calls( + calls: list[tuple[str, dict[str, Any]]], +) -> list[tuple[str, dict[str, Any]]]: + """Нормализуем расхождение region_code: `:region_code` → `66`, убираем ключ. + + Это единственное намеренное отличие новой копии; после нормализации при + region_code=66 SQL и params должны быть байт-идентичны старому боевому коду. + """ + out: list[tuple[str, dict[str, Any]]] = [] + for sql, params in calls: + norm_sql = sql.replace(":region_code", str(_REGION_CODE)) + norm_params = {k: v for k, v in params.items() if k != "region_code"} + out.append((norm_sql, norm_params)) + return out + + +def _kwargs_list(mock_or_method: Any) -> list[dict[str, Any]]: + return [c.kwargs for c in mock_or_method.call_args_list] + + +# ── tests ──────────────────────────────────────────────────────────────────── + + +def test_save_listings_execute_sequence_parity() -> None: + """Последовательность (SQL, params) execute идентична (после region_code-нормализации).""" + lots = [NewLot(**d) for d in _fixture_data()] + + old_db, _oh, _ol = _run_old(lots) + new_db, _nm = _run_new(lots) + + old_calls = old_db.calls + new_calls = _normalize_new_calls(new_db.calls) + + assert len(new_calls) == len( + old_calls + ), f"разное число execute: old={len(old_calls)} new={len(new_calls)}" + for i, (old_c, new_c) in enumerate(zip(old_calls, new_calls, strict=True)): + assert old_c[0] == new_c[0], f"SQL расходится на вызове #{i}" + assert old_c[1] == new_c[1], f"params расходятся на вызове #{i}" + + +def test_insert_sql_region_code_injection() -> None: + """Старый хардкодит `66` в VALUES; новый — параметр `:region_code`, params[region_code]=66.""" + lots = [NewLot(**_fixture_data()[0])] + + old_db, _oh, _ol = _run_old(lots) + new_db, _nm = _run_new(lots) + + old_insert = next(s for s, _ in old_db.calls if "INSERT INTO listings (" in s) + new_insert, new_params = next((s, p) for s, p in new_db.calls if "INSERT INTO listings (" in s) + + assert ", 66," in old_insert and ":region_code" not in old_insert + assert ":region_code," in new_insert and ", 66," not in new_insert + assert new_params["region_code"] == 66 + # После нормализации → идентичны + assert new_insert.replace(":region_code", "66") == old_insert + + +def test_matching_hook_kwargs_parity() -> None: + """kwargs match_or_create_house / upsert_listing_source идентичны old ↔ new.""" + lots = [NewLot(**d) for d in _fixture_data()] + + _old_db, m_house, m_link = _run_old(lots) + _new_db, matcher = _run_new(lots) + + assert _kwargs_list(m_house) == _kwargs_list(matcher.match_or_create_house) + assert _kwargs_list(m_link) == _kwargs_list(matcher.upsert_listing_source) + + # sanity: hook действительно сработал (не пустой прогон) + assert m_link.call_count == len(lots) + # house resolve только когда есть address/coords (4 из 5 фикстур) + assert m_house.call_count == 4 + assert matcher.match_or_create_house.call_count == 4 + + +def test_dedup_collision_triggers_reconcile_update_in_both() -> None: + """Drift-лот → INSERT UniqueViolation → reconcile UPDATE by (source, source_id) в обоих.""" + lots = [NewLot(**_fixture_data()[4])] # DRIFT-1 + + old_db, _oh, _ol = _run_old(lots) + new_db, _nm = _run_new(lots) + + old_reconcile = [s for s, _ in old_db.calls if "UPDATE listings" in s and "SET dedup_hash" in s] + new_reconcile = [s for s, _ in new_db.calls if "UPDATE listings" in s and "SET dedup_hash" in s] + assert len(old_reconcile) == 1 + assert len(new_reconcile) == 1 + assert old_reconcile[0] == new_reconcile[0] + + +def test_return_counts_parity() -> None: + """(inserted, updated) идентичны old ↔ new на одном входе.""" + lots = [NewLot(**d) for d in _fixture_data()] + + old_db = RecordingSession(drift_ids=_DRIFT_IDS) + with ( + patch("app.services.scrapers.base.match_or_create_house", return_value=(101, 1.0, "new")), + patch("app.services.scrapers.base.upsert_listing_source", return_value=None), + patch("app.services.scrapers.base.upsert_listing_snapshot"), + ): + old_result = old_save_listings(old_db, lots) + + new_db = RecordingSession(drift_ids=_DRIFT_IDS) + matcher = MagicMock() + matcher.match_or_create_house.return_value = (101, 1.0, "new") + matcher.upsert_listing_source.return_value = None + with patch("scraper_kit.base.upsert_listing_snapshot"): + new_result = new_save_listings(new_db, lots, matcher=matcher, region_code=_REGION_CODE) + + assert old_result == new_result + + +def test_hash_parity_old_vs_new_scraped_lot() -> None: + """compute_dedup_hash / compute_card_hash / compute_price_per_m2 идентичны у обоих классов.""" + for data in _fixture_data(): + old_lot = OldLot(**data) + new_lot = NewLot(**data) + assert old_lot.compute_dedup_hash() == new_lot.compute_dedup_hash() + assert old_lot.compute_card_hash() == new_lot.compute_card_hash() + assert old_lot.compute_price_per_m2() == new_lot.compute_price_per_m2() + + +def test_kit_base_public_api_importable() -> None: + """Acceptance: from scraper_kit.base import ScrapedLot, save_listings, BaseScraper.""" + from scraper_kit.base import BaseScraper, ScrapedLot, save_listings + + assert ScrapedLot is not None + assert callable(save_listings) + assert BaseScraper is not None + + +def test_kit_base_has_no_app_imports() -> None: + """scraper_kit.base не должен импортировать app.* (strangler-развязка).""" + import inspect + + import scraper_kit.base as kit_base + + source = inspect.getsource(kit_base) + # Игнорируем docstring-упоминания: проверяем именно import-стейтменты. + for line in source.splitlines(): + stripped = line.strip() + assert not stripped.startswith("from app"), f"app import найден: {line!r}" + assert not stripped.startswith("import app"), f"app import найден: {line!r}" diff --git a/tradein-mvp/packages/scraper-kit/pyproject.toml b/tradein-mvp/packages/scraper-kit/pyproject.toml index 5b8344bf..8f670152 100644 --- a/tradein-mvp/packages/scraper-kit/pyproject.toml +++ b/tradein-mvp/packages/scraper-kit/pyproject.toml @@ -3,7 +3,14 @@ name = "scraper-kit" version = "0.1.0" description = "Internal scraper toolkit — общие утилиты для tradein-mvp скрапперов" requires-python = ">=3.12" -dependencies = [] +dependencies = [ + "sqlalchemy>=2.0.30", + "pydantic>=2.7.0", + "psycopg[binary]>=3.2.0", + "httpx[socks]>=0.27.0", + "tenacity>=9.0.0", + "selectolax>=0.3.0", +] [build-system] requires = ["hatchling"] diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/__init__.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/__init__.py index 7781d442..490cfcb6 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/__init__.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/__init__.py @@ -1,7 +1,10 @@ """scraper-kit — общие утилиты для скрапперов tradein-mvp. -Пока пустой скаффолд: пакет установлен как uv workspace-member, -но код скрапперов ещё не перенесён сюда (см. issue #2128). +Strangler-копия core-слоя `app.services.scrapers.*` (#2132). Развязана от `app.*` +через Protocol-инжекцию (`scraper_kit.contracts`): здесь живёт независимая копия +базового слоя (`base.save_listings`, `ScrapedLot`, `BaseScraper`) + транспортных +утилит. Боевые скрапперы пока импортируют старые `app.services.scrapers.*`; +переключение боевых вызовов — отдельный поздний шаг (не здесь). """ __version__ = "0.1.0" diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/avito_exceptions.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/avito_exceptions.py new file mode 100644 index 00000000..3895df76 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/avito_exceptions.py @@ -0,0 +1,36 @@ +"""Avito-specific exceptions для anti-bot detection.""" + + +class AvitoError(Exception): + """Base для всех Avito scraper exceptions.""" + + +class AvitoBlockedError(AvitoError): + """HTTP 403 от Avito — IP-level block detected.""" + + +class AvitoRateLimitedError(AvitoError): + """HTTP 429 от Avito — rate limit triggered.""" + + +class AvitoListingGoneError(AvitoError): + """Объявление удалено / вернуло 404 (страница «Ошибка 404» вместо item-view). + + Отдельный сигнал от AvitoBlockedError/AvitoRateLimitedError: 404 — НЕ анти-бот + deflect, а мёртвый листинг (его незачем ротировать/ретраить). Наследует общий + AvitoError, но НЕ AvitoBlockedError — чтобы backfill-ловушка + `except (AvitoBlockedError, AvitoRateLimitedError)` его НЕ перехватывала и не крутила + consecutive-block breaker на мёртвых координатных дырах (run 458: 5 dead → abort). + """ + + +class AvitoContentBlockedError(AvitoBlockedError): + """HTTP 200 от Avito, но 0 карточек на первой странице — content-block / captcha. + + Наследует AvitoBlockedError, чтобы существующие except-блоки на AvitoBlockedError + продолжали ловить этот сигнал (mark_banned, pipeline abort и т.д.). + """ + + +class AvitoParseError(AvitoError): + """Cannot parse Avito HTML structure (selector changes).""" diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py new file mode 100644 index 00000000..51169263 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py @@ -0,0 +1,844 @@ +"""Базовый framework для парсеров объявлений (Avito / Cian / DomKlik / ...). + +scraper_kit-копия (strangler #2132) базового слоя `app.services.scrapers.base`. +Развязана от `app.*` через Protocol-инжекцию (см. `scraper_kit.contracts`): +`save_listings` принимает `matcher: HouseMatcher` вместо прямого импорта +`app.services.matching`, а `region_code` передаётся параметром (хардкод `66` +убран — дефолт задаёт вызывающая продуктовая сторона). Пока НИКТО из боевого +кода это не импортирует; старый `app.services.scrapers.base` остаётся боевым. + +Общие задачи: +- httpx.AsyncClient с realistic browser headers + UA rotation +- tenacity retry с exp backoff +- Sleep между запросами (anti-ban) +- ScrapedLot — единая Pydantic схема всех источников +- Запись в `listings` Postgres с дедупом по dedup_hash +- После INSERT/UPDATE — hook в matching service (инжектируемый `HouseMatcher`) + для регистрации в `listing_sources` и привязки к каноническому `houses`. + +Каждый конкретный парсер (avito.py, cian.py, ...) наследуется от BaseScraper +и реализует `_fetch_page()` и `_parse_listings()`. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import logging +import random +from abc import ABC, abstractmethod +from datetime import date, datetime +from typing import TYPE_CHECKING, Any +from zoneinfo import ZoneInfo + +import httpx +import psycopg.errors +from pydantic import BaseModel, Field +from sqlalchemy import text +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session +from tenacity import retry, stop_after_attempt, wait_exponential + +from scraper_kit.snapshot_writer import upsert_listing_snapshot + +if TYPE_CHECKING: + from scraper_kit.contracts import HouseMatcher + +logger = logging.getLogger(__name__) + +_MSK = ZoneInfo("Europe/Moscow") + + +# ── Realistic browser User-Agents (Apr 2026 versions) ────────────────────── + +_USER_AGENTS: list[str] = [ + # Chrome on macOS + "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36", # noqa: E501 + # Chrome on Windows + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36", # noqa: E501 + # Firefox on macOS + "Mozilla/5.0 (Macintosh; Intel Mac OS X 14.5; rv:127.0) Gecko/20100101 Firefox/127.0", + # Safari on macOS + "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15", # noqa: E501 + # Edge on Windows + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36 Edg/132.0.0.0", # noqa: E501 +] + + +def random_user_agent() -> str: + return random.choice(_USER_AGENTS) + + +# ── Унифицированная схема результата ──────────────────────────────────────── +class ScrapedLot(BaseModel): + """Результат парсинга одного объявления. + + Все источники приводятся к этому виду перед записью в Postgres. + Поля optional если источник их не отдаёт. + """ + + source: str # 'avito' / 'cian' / 'domklik' / ... + source_url: str # ссылка на оригинал — кликается из UI + source_id: str | None = None # внутренний id источника (для update) + + # Локация — должна быть хотя бы address ИЛИ (lat, lon) + address: str | None = None + lat: float | None = None + lon: float | None = None + + # Параметры квартиры + rooms: int | None = None # 0 = студия + area_m2: float | None = None + floor: int | None = None + total_floors: int | None = None + year_built: int | None = None + house_type: str | None = None + repair_state: str | None = None + has_balcony: bool | None = None + kadastr_num: str | None = None + + # Detail-поля кухни/потолка (#2007). Обычно из detail-страницы (avito_detail, + # cian_detail, yandex_detail), но yandex SERP отдаёт их прямо в карточке — + # промоутим из raw_payload в колонки для дашборда покрытия и matching. + kitchen_area_m2: float | None = None + ceiling_height_m: float | None = None + + # Class B promote-поля (#2008, cian SERP). Для дашборда покрытия / matching, + # НЕ для valuation — estimator их пока НЕ читает (follow-up #2012). cian отдаёт + # их прямо в карточке; раньше лежали только в raw_payload. + mortgage_available: bool | None = None # продавец допускает ипотеку + is_apartments: bool | None = None # апартаменты (НЕ квартира — иной правовой статус) + is_rosreestr_checked: bool | None = None # объявление сверено с ЕГРН + + # Avito house linking (Stage 2a — search → houses) + house_source: str | None = None # 'avito' / 'cian' + house_ext_id: str | None = None # Avito's '3171365' + house_url: str | None = None # full URL of /catalog/houses/... + listing_segment: str | None = None # 'vtorichka' / 'novostroyki' + + # Newbuilding (ЖК) linking — slug-id + canonical ЖК-subdomain URL + newbuilding_id: str | None = None # e.g. 'federatsiya-ekaterinburg' + newbuilding_url: str | None = None # e.g. https://zhk-federatsiya-ekaterinburg.avito.ru + + # Цена (обязательно) + price_rub: int = Field(gt=0) + price_per_m2: int | None = None + + # Геокодинг + # None → точность неизвестна (координаты пришли из скрейпера напрямую). + # 'city' → геокодер вернул city-centroid (нет номера дома в адресе) — + # листинг исключается из radius-аналогов estimator'а. + geo_precision: str | None = None + + # Метаданные + listing_date: date | None = None + publish_date: date | None = None + days_on_market: int | None = None + description: str | None = None + photo_urls: list[str] = Field(default_factory=list) + raw_payload: dict[str, Any] | None = None + + # Cian-specific extensions (Stage 2 of CianScraper v1) + living_area_m2: float | None = None + bedrooms_count: int | None = None + balconies_count: int | None = None + loggias_count: int | None = None + description_minhash: str | None = None + cadastral_number: str | None = None + building_cadastral_number: str | None = None + phones: list[dict] = Field(default_factory=list) + is_homeowner: bool | None = None + is_pro_seller: bool | None = None + bargain_allowed: bool | None = None + sale_type: str | None = None + metro_stations: list[dict] = Field(default_factory=list) + agency_name: str | None = None + + # Yandex gate-API rich fields + yandex_offer_id: str | None = None # numeric string offerId (kept as text) + predicted_price_rub: int | None = None # Yandex per-offer valuation (predictedPrice.value) + predicted_price_min: int | None = None + predicted_price_max: int | None = None + price_trend: str | None = None # INCREASED / DECREASED / UNCHANGED / ... + price_previous_rub: int | None = None # price.previous + + def compute_dedup_hash(self) -> str: + """SHA256(source + stable_key) — стабильный uniqueness key. + + stable_key = source_id (Cian offer_id 330200428, Avito data-item-id) если + есть; иначе source_url БЕЗ query-string. Cian/Avito URL несут волатильный + `?context=...` токен — без strip каждый ре-скрейп давал бы новый hash. + + #753: price_rub в ключ НЕ входит. Смена цены того же объявления — это UPDATE + существующей строки (`ON CONFLICT (dedup_hash) DO UPDATE`), а не новый объект. + Раньше цена была частью ключа → каждое изменение цены порождало дубль + активного listing → искажение asking_to_sold_ratio / sales-vs-listings / + активного инвентаря. Смена формулы разово даёт новый hash существующим + строкам — следующий скрейп upsert'нет их по новому ключу (всплеск разовый). + """ + h = hashlib.sha256() + h.update(self.source.encode()) + h.update(b"|") + key = self.source_id if self.source_id else (self.source_url or "").split("?")[0] + h.update((key or "").encode()) + return h.hexdigest() + + def compute_card_hash(self) -> str: + """SHA-256 over the volatile SERP-card fields. Used to detect whether a + listing's card content changed between scrapes (snapshot-dedup, detail-refresh gate). + Excludes last_seen/scrape timestamps and detail-only fields. + + Хэшируем детерминированную нормализованную репрезентацию полей карточки: + price_rub, area_m2, floor, total_floors, rooms, address (strip+lower, None→""), + len(photo_urls). Порядок частей фиксирован (нет зависимости от порядка + итерации dict) → стабильный hash при неизменной карточке. + """ + address = (self.address or "").strip().lower() + parts = [ + str(self.price_rub), + str(self.area_m2), + str(self.floor), + str(self.total_floors), + str(self.rooms), + address, + str(len(self.photo_urls)), + ] + return hashlib.sha256("|".join(parts).encode()).hexdigest() + + def compute_price_per_m2(self) -> int | None: + """price_per_m2 = price_rub / area_m2 если area задана.""" + if self.area_m2 and self.area_m2 > 0: + return int(self.price_rub / self.area_m2) + return None + + +# ── BaseScraper ───────────────────────────────────────────────────────────── +class BaseScraper(ABC): + """Базовый класс — наследовать для каждого источника. + + Subclass должен реализовать: + - `name` (class attr) — 'avito' / 'cian' / ... + - `_fetch_page(client, ...)` — async GET к источнику + - `_parse_listings(html_or_json, ...)` — html → list[ScrapedLot] + """ + + name: str = "base" + base_url: str = "" + request_delay_sec: float = 5.0 # sleep между запросами (anti-ban) + + def __init__(self) -> None: + self._client: httpx.AsyncClient | None = None + + async def __aenter__(self) -> BaseScraper: + self._client = httpx.AsyncClient( + timeout=20.0, + follow_redirects=True, + headers={ + "User-Agent": random_user_agent(), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", + "Accept-Encoding": "gzip, deflate, br", + "Connection": "keep-alive", + "Upgrade-Insecure-Requests": "1", + }, + ) + return self + + async def __aexit__(self, *args: Any) -> None: + if self._client is not None: + await self._client.aclose() + + @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=2, min=2, max=30)) + async def _http_get(self, url: str, **kwargs: Any) -> httpx.Response: + """GET с retry. Subclass-ы используют это вместо raw httpx.""" + assert self._client is not None, "Use scraper as async context manager" + response = await self._client.get(url, **kwargs) + # 403/429 — ретрай поможет (рандомный UA в новом контексте), 5xx тоже + if response.status_code in {403, 429, 502, 503, 504}: + response.raise_for_status() + return response + + async def sleep_between_requests(self) -> None: + """Случайный jitter в районе self.request_delay_sec — чтобы парсер + не выглядел как detstvo-bot.""" + jitter = random.uniform(0.7, 1.5) + await asyncio.sleep(self.request_delay_sec * jitter) + + @abstractmethod + async def fetch_around(self, lat: float, lon: float, radius_m: int = 1000) -> list[ScrapedLot]: + """Главный метод — собрать объявления вокруг (lat, lon) в радиусе radius_m. + + Returns: + Список ScrapedLot. Может быть пустым. + """ + ... + + +# ── Запись пачки результатов в Postgres ───────────────────────────────────── +def save_listings( + db: Session, + lots: list[ScrapedLot], + *, + matcher: HouseMatcher, + region_code: int, + run_id: int | None = None, + skip_seen_today: bool = False, +) -> tuple[int, int]: + """Пишем list[ScrapedLot] в `listings` с upsert по dedup_hash. + + Дополнительно пишет point-in-time snapshot в listings_snapshots (ON CONFLICT DO UPDATE + — берётся последний за день). Ошибка записи snapshot НЕ прерывает основной INSERT. + + Args: + db: SQLAlchemy session. + lots: список распарсенных объявлений. + matcher: инжектируемый `HouseMatcher` (scraper_kit.contracts) — кросс- + источниковый матчинг домов/листингов. Заменяет прямой импорт + `app.services.matching` (strangler-развязка). Продуктовая сторона + передаёт `RealMatcherAdapter` (см. app.services.scraper_adapters). + region_code: код региона для колонки `listings.region_code` (ранее хардкод + `66` в SQL). Дефолт задаёт вызывающая сторона, не kit. + run_id: FK scrape_runs.id текущего run'а (опционально; None при ad-hoc вызовах). + skip_seen_today: если True и листинг уже обновлялся сегодня по МСК + (last_seen_at >= начало сегодняшнего дня), то пропустить upsert и snapshot. + Используется full_load для экономии redundant upsert + price-trigger churn + при повторном прогоне в тот же день. Новые листинги (prior_row=None) всегда + вставляются. False = старое поведение (всегда upsert). + + Returns: + (inserted, updated) — counters для логов. + """ + if not lots: + return 0, 0 + + inserted = 0 + updated = 0 + skipped = 0 + reconciled = 0 # UPDATE by (source,source_id) при dedup_hash-дрейфе + matched = 0 + match_failures = 0 + + today_msk = datetime.now(_MSK).date() + + for lot in lots: + ppm2 = lot.price_per_m2 or lot.compute_price_per_m2() + dedup = lot.compute_dedup_hash() + card_hash = lot.compute_card_hash() + + # Pre-read the existing row's card_hash and last_seen_at (keyed by + # dedup_hash) BEFORE the upsert — needed to know the *prior* card + # content and to implement skip_seen_today logic. + # The upsert's RETURNING can only expose the NEW values (EXCLUDED.*), + # never the old ones, so a lightweight SELECT is the clean way. + # None → row did not exist yet (fresh insert) → always process. + prior_row = db.execute( + text("SELECT card_hash, last_seen_at FROM listings WHERE dedup_hash = :dedup"), + {"dedup": dedup}, + ).fetchone() + prior_card_hash = prior_row.card_hash if prior_row is not None else None + + # skip_seen_today: пропускаем листинги уже обновлённые сегодня по МСК. + # Только для existing строк (prior_row is not None); новые всегда вставляются. + if skip_seen_today and prior_row is not None and prior_row.last_seen_at is not None: + ls_msk = prior_row.last_seen_at.astimezone(_MSK).date() + if ls_msk == today_msk: + skipped += 1 + continue + + # ── Per-lot SAVEPOINT: изолируем listings-upsert ──────────────────── + # Когда контент существующего ad дрейфит → новый dedup_hash → INSERT + # находит prior_row=None (SELECT по старому dedup_hash пуст) → INSERT + # летит в UniqueViolation по (source, source_id) при source_id IS NOT NULL + # (constraint 133_listings_uq_source_source_id). Без SAVEPOINT это роняет + # весь прогон. С SAVEPOINT ловим IntegrityError, rollback SAVEPOINT'а, + # и делаем прямой UPDATE по (source, source_id) — reconcile drifted row. + # source_id IS NULL (yandex) → constraint не фаерит, обычный путь. + _upsert_params = { + "source": lot.source, + "source_url": lot.source_url, + "source_id": lot.source_id, + "dedup": dedup, + "region_code": region_code, + "address": lot.address, + "lat": lot.lat, + "lon": lot.lon, + "rooms": lot.rooms, + "area_m2": lot.area_m2, + "floor": lot.floor, + "total_floors": lot.total_floors, + "year_built": lot.year_built, + "house_type": lot.house_type, + "repair_state": lot.repair_state, + "has_balcony": lot.has_balcony, + "kitchen_area_m2": lot.kitchen_area_m2, + "ceiling_height_m": lot.ceiling_height_m, + "mortgage_available": lot.mortgage_available, + "is_apartments": lot.is_apartments, + "is_rosreestr_checked": lot.is_rosreestr_checked, + "house_source": lot.house_source, + "house_ext_id": lot.house_ext_id, + "house_url": lot.house_url, + "listing_segment": lot.listing_segment, + "newbuilding_id": lot.newbuilding_id, + "newbuilding_url": lot.newbuilding_url, + "price_rub": lot.price_rub, + "ppm2": ppm2, + "listing_date": lot.listing_date, + "publish_date": lot.publish_date, + "days_on_market": lot.days_on_market, + "description": lot.description, + "photos": _to_json(lot.photo_urls), + "raw": _to_json(lot.raw_payload) if lot.raw_payload else None, + # Cian-specific columns — None for non-Cian scrapers (→ SQL NULL) + "living_area_m2": lot.living_area_m2, + "bedrooms_count": lot.bedrooms_count, + "balconies_count": lot.balconies_count, + "loggias_count": lot.loggias_count, + "description_minhash": lot.description_minhash, + "cadastral_number": lot.cadastral_number, + "building_cadastral_number": lot.building_cadastral_number, + "phones": _to_json(lot.phones) if lot.phones else None, + "is_homeowner": lot.is_homeowner, + "is_pro_seller": lot.is_pro_seller, + "bargain_allowed": lot.bargain_allowed, + "sale_type": lot.sale_type, + "metro_stations": _to_json(lot.metro_stations) if lot.metro_stations else None, + "agency_name": lot.agency_name, + "yandex_offer_id": lot.yandex_offer_id, + "predicted_price_rub": lot.predicted_price_rub, + "predicted_price_min": lot.predicted_price_min, + "predicted_price_max": lot.predicted_price_max, + "price_trend": lot.price_trend, + "price_previous_rub": lot.price_previous_rub, + "geo_precision": lot.geo_precision, + "card_hash": card_hash, + } + + _upsert_sql = text( + """ + INSERT INTO listings ( + source, source_url, source_id, dedup_hash, + address, lat, lon, region_code, + rooms, area_m2, floor, total_floors, year_built, + house_type, repair_state, has_balcony, + kitchen_area_m2, ceiling_height, ceiling_height_m, + mortgage_available, is_apartments, is_rosreestr_checked, + house_source, house_ext_id, house_url, listing_segment, + newbuilding_id, newbuilding_url, + price_rub, price_per_m2, + listing_date, publish_date, days_on_market, description, + photo_urls, raw_payload, + living_area_m2, bedrooms_count, balconies_count, loggias_count, + description_minhash, cadastral_number, building_cadastral_number, + phones, is_homeowner, is_pro_seller, + bargain_allowed, sale_type, metro_stations, agency_name, + yandex_offer_id, predicted_price_rub, + predicted_price_min, predicted_price_max, + price_trend, price_previous_rub, + geo_precision, card_hash, + scraped_at, last_seen_at + ) VALUES ( + :source, :source_url, :source_id, :dedup, + :address, :lat, :lon, :region_code, + :rooms, :area_m2, :floor, :total_floors, :year_built, + :house_type, :repair_state, :has_balcony, + -- ceiling: один param :ceiling_height_m пишем в ОБЕ колонки — + -- ceiling_height (019, читает coverage-дашборд + yandex_detail/cian_detail) + -- и ceiling_height_m (111, живая avito-колонка). См. #2007. + :kitchen_area_m2, :ceiling_height_m, :ceiling_height_m, + :mortgage_available, :is_apartments, :is_rosreestr_checked, + :house_source, :house_ext_id, :house_url, :listing_segment, + :newbuilding_id, :newbuilding_url, + :price_rub, :ppm2, + :listing_date, :publish_date, :days_on_market, :description, + CAST(:photos AS jsonb), + CAST(:raw AS jsonb), + :living_area_m2, :bedrooms_count, :balconies_count, :loggias_count, + :description_minhash, :cadastral_number, :building_cadastral_number, + CAST(:phones AS jsonb), :is_homeowner, :is_pro_seller, + :bargain_allowed, :sale_type, CAST(:metro_stations AS jsonb), :agency_name, + :yandex_offer_id, :predicted_price_rub, + :predicted_price_min, :predicted_price_max, + :price_trend, :price_previous_rub, + :geo_precision, :card_hash, + NOW(), NOW() + ) + -- Конфликт-арбитр — dedup_hash (sha256(source|source_id)). + -- Для одного (source, source_id) формула даёт ОДИН dedup_hash, + -- поэтому этот апсерт уже не вставит второй раз → согласован с + -- частичным UNIQUE (source, source_id) WHERE source_id IS NOT NULL + -- (133_listings_uq_source_source_id.sql). Менять арбитр на + -- (source, source_id) НЕЛЬЗЯ: yandex/url-only дают source_id=NULL, + -- а NULL не годится как conflict-target → дубли вернулись бы (#1773). + ON CONFLICT (dedup_hash) DO UPDATE + SET last_seen_at = NOW(), + is_active = true, + -- если цена изменилась — обновляем + price_rub = EXCLUDED.price_rub, + price_per_m2 = EXCLUDED.price_per_m2, + -- Cian-specific: обновляем при каждом re-scrape + living_area_m2 = EXCLUDED.living_area_m2, + bedrooms_count = EXCLUDED.bedrooms_count, + balconies_count = EXCLUDED.balconies_count, + loggias_count = EXCLUDED.loggias_count, + description_minhash = EXCLUDED.description_minhash, + cadastral_number = EXCLUDED.cadastral_number, + building_cadastral_number = EXCLUDED.building_cadastral_number, + phones = EXCLUDED.phones, + is_homeowner = EXCLUDED.is_homeowner, + is_pro_seller = EXCLUDED.is_pro_seller, + bargain_allowed = EXCLUDED.bargain_allowed, + sale_type = EXCLUDED.sale_type, + metro_stations = EXCLUDED.metro_stations, + listing_date = COALESCE(EXCLUDED.listing_date, listings.listing_date), + area_m2 = COALESCE(EXCLUDED.area_m2, listings.area_m2), + -- kitchen/ceiling (#2007): COALESCE — SERP re-scrape источника без + -- этих полей (avito SERP → NULL) НЕ затирает detail-enriched значение + -- (avito_detail пишет ceiling_height_m отдельным UPDATE). + kitchen_area_m2 = COALESCE( + EXCLUDED.kitchen_area_m2, listings.kitchen_area_m2 + ), + ceiling_height = COALESCE( + EXCLUDED.ceiling_height, listings.ceiling_height + ), + ceiling_height_m = COALESCE( + EXCLUDED.ceiling_height_m, listings.ceiling_height_m + ), + -- Class B promote (#2008, cian SERP): COALESCE — re-scrape + -- источника без поля (avito/yandex SERP → NULL) НЕ затирает + -- значение, ранее записанное cian-ом / detail-ом. + mortgage_available = COALESCE( + EXCLUDED.mortgage_available, listings.mortgage_available + ), + is_apartments = COALESCE( + EXCLUDED.is_apartments, listings.is_apartments + ), + is_rosreestr_checked = COALESCE( + EXCLUDED.is_rosreestr_checked, listings.is_rosreestr_checked + ), + -- Yandex rich fields (+ shared description/agency/publish_date): + -- COALESCE so a source that does not provide them never wipes + -- a value previously written by another source / scrape. + publish_date = COALESCE(EXCLUDED.publish_date, listings.publish_date), + days_on_market = COALESCE( + EXCLUDED.days_on_market, listings.days_on_market + ), + description = COALESCE(EXCLUDED.description, listings.description), + agency_name = COALESCE(EXCLUDED.agency_name, listings.agency_name), + yandex_offer_id = COALESCE( + EXCLUDED.yandex_offer_id, listings.yandex_offer_id + ), + predicted_price_rub = COALESCE( + EXCLUDED.predicted_price_rub, listings.predicted_price_rub + ), + predicted_price_min = COALESCE( + EXCLUDED.predicted_price_min, listings.predicted_price_min + ), + predicted_price_max = COALESCE( + EXCLUDED.predicted_price_max, listings.predicted_price_max + ), + price_trend = COALESCE(EXCLUDED.price_trend, listings.price_trend), + price_previous_rub = COALESCE( + EXCLUDED.price_previous_rub, listings.price_previous_rub + ), + newbuilding_id = COALESCE( + EXCLUDED.newbuilding_id, listings.newbuilding_id + ), + newbuilding_url = COALESCE( + EXCLUDED.newbuilding_url, listings.newbuilding_url + ), + card_hash = EXCLUDED.card_hash + RETURNING id, (xmax = 0) AS inserted + """ + ) + + # ── Per-lot SAVEPOINT вокруг listings-upsert ──────────────────────── + # Когда контент существующего ad дрейфит → новый dedup_hash → INSERT + # не находит existing row по dedup_hash → пробует fresh INSERT → летит + # UniqueViolation по (source, source_id) WHERE source_id IS NOT NULL + # (constraint listings_source_source_id_uq). Без SAVEPOINT это переводит + # весь Session в InFailedSqlTransaction → весь прогон падает. + # С SAVEPOINT ловим IntegrityError/UniqueViolation, делаем rollback, + # и применяем UPDATE by (source, source_id) — reconcile drifted row. + # source_id IS NULL (yandex) → constraint не фаерит, обычный upsert путь. + result = None + listing_id: int | None = None + try: + with db.begin_nested(): + result = db.execute(_upsert_sql, _upsert_params).fetchone() + except IntegrityError as ie: + # Проверяем что это UniqueViolation именно по (source, source_id). + if lot.source_id is not None and isinstance(ie.orig, psycopg.errors.UniqueViolation): + logger.warning( + "save_listings:dedup_hash_drift source=%s source_id=%s " + "— reconcile UPDATE by (source, source_id)", + lot.source, + lot.source_id, + ) + # SAVEPOINT уже откачен; session чистая — применяем прямой UPDATE. + rec_row = db.execute( + text( + """ + UPDATE listings + SET dedup_hash = :dedup, + last_seen_at = NOW(), + is_active = true, + price_rub = :price_rub, + price_per_m2 = :ppm2, + living_area_m2 = :living_area_m2, + bedrooms_count = :bedrooms_count, + balconies_count = :balconies_count, + loggias_count = :loggias_count, + description_minhash = :description_minhash, + cadastral_number = :cadastral_number, + building_cadastral_number = :building_cadastral_number, + phones = CAST(:phones AS jsonb), + is_homeowner = :is_homeowner, + is_pro_seller = :is_pro_seller, + bargain_allowed = :bargain_allowed, + sale_type = :sale_type, + metro_stations = CAST(:metro_stations AS jsonb), + listing_date = COALESCE(:listing_date, listing_date), + area_m2 = COALESCE(:area_m2, area_m2), + kitchen_area_m2 = COALESCE(:kitchen_area_m2, kitchen_area_m2), + ceiling_height = COALESCE(:ceiling_height_m, ceiling_height), + ceiling_height_m = COALESCE(:ceiling_height_m, ceiling_height_m), + mortgage_available = COALESCE( + :mortgage_available, mortgage_available + ), + is_apartments = COALESCE(:is_apartments, is_apartments), + is_rosreestr_checked = COALESCE( + :is_rosreestr_checked, is_rosreestr_checked + ), + publish_date = COALESCE(:publish_date, publish_date), + days_on_market = COALESCE(:days_on_market, days_on_market), + description = COALESCE(:description, description), + agency_name = COALESCE(:agency_name, agency_name), + yandex_offer_id = COALESCE(:yandex_offer_id, yandex_offer_id), + predicted_price_rub = COALESCE( + :predicted_price_rub, predicted_price_rub + ), + predicted_price_min = COALESCE( + :predicted_price_min, predicted_price_min + ), + predicted_price_max = COALESCE( + :predicted_price_max, predicted_price_max + ), + price_trend = COALESCE(:price_trend, price_trend), + price_previous_rub = COALESCE( + :price_previous_rub, price_previous_rub + ), + newbuilding_id = COALESCE(:newbuilding_id, newbuilding_id), + newbuilding_url = COALESCE(:newbuilding_url, newbuilding_url), + card_hash = :card_hash + WHERE source = :source + AND source_id = :source_id + RETURNING id + """ + ), + _upsert_params, + ).fetchone() + if rec_row is not None: + listing_id = int(rec_row.id) + reconciled += 1 + else: + logger.error( + "save_listings:reconcile_failed source=%s source_id=%s " + "— UPDATE returned no row (unexpected)", + lot.source, + lot.source_id, + ) + else: + logger.error( + "save_listings:integrity_error source=%s source_id=%s: %s", + lot.source, + lot.source_id, + ie, + ) + raise + else: + if result is not None: + listing_id = int(result.id) + if result.inserted: + inserted += 1 + else: + updated += 1 + + # ── Snapshot: point-in-time observation in listings_snapshots ─── + # Fault-tolerant: failure here MUST NOT abort the listings batch. + # ON CONFLICT DO UPDATE — самый последний snapshot за день. + # + # Snapshot-dedup: пропускаем запись только когда МЫ УВЕРЕНЫ что карточка + # не изменилась (prior_card_hash == card_hash, row существовал) И snapshot + # за сегодня уже есть. Принцип "when in doubt — write": лишний snapshot + # неизменной карточки безвреден, ПРОПУЩЕННЫЙ snapshot реального изменения — + # недопустим. Поэтому equality-проверки мало (первый скрейп за день мог + # ещё не записать сегодняшний snapshot) — дополнительно подтверждаем + # наличие сегодняшней строки лёгким EXISTS. + if listing_id is not None: + skip_snapshot = False + if prior_card_hash is not None and prior_card_hash == card_hash: + today_row = db.execute( + text( + "SELECT 1 FROM listings_snapshots " + "WHERE listing_id = CAST(:lid AS bigint) " + " AND snapshot_date = CURRENT_DATE" + ), + {"lid": listing_id}, + ).fetchone() + skip_snapshot = today_row is not None + if skip_snapshot: + logger.debug( + "save_listings:snapshot_skipped (card unchanged) " "source=%s listing_id=%s", + lot.source, + listing_id, + ) + if listing_id is not None and not skip_snapshot: + try: + with db.begin_nested(): + upsert_listing_snapshot( + db, + listing_id=listing_id, + price_rub=lot.price_rub, + price_per_m2=ppm2, + run_id=run_id, + status="active", + ) + except Exception as e: + logger.warning( + "save_listings:snapshot_failed source=%s listing_id=%s: %s", + lot.source, + listing_id, + e, + ) + + # ── Hook: link listing → house via matching service ───────────── + # Idempotent (UPSERT on listing_sources UNIQUE(ext_source, ext_id)) + # and fault-tolerant: failure here MUST NOT abort the listings batch. + # Wrapped in SAVEPOINT so a failed match only rolls back its own work, + # not the surrounding INSERT (see .claude/rules/backend.md SAVEPOINT). + if listing_id is not None: + try: + with db.begin_nested(): + _link_listing_to_house(db, listing_id, lot, matcher) + matched += 1 + except Exception as e: + # Best-effort hook: log and continue so the listings batch isn't aborted. + match_failures += 1 + logger.warning( + "save_listings:match_failed source=%s source_id=%s listing_id=%s: %s", + lot.source, + lot.source_id, + listing_id, + e, + ) + + db.commit() + logger.info( + "save_listings: source=%s inserted=%d updated=%d reconciled=%d " + "skipped_seen_today=%d matched=%d match_failures=%d (total %d)", + lots[0].source if lots else "?", + inserted, + updated, + reconciled, + skipped, + matched, + match_failures, + len(lots), + ) + return inserted, updated + + +def _to_json(value: Any) -> str: + """JSON serialization helper — для jsonb колонок.""" + import json + + return json.dumps(value, ensure_ascii=False, default=str) + + +def _link_listing_to_house( + db: Session, listing_id: int, lot: ScrapedLot, matcher: HouseMatcher +) -> None: + """Hook scraped listing into matching service: resolve house, upsert listing_sources. + + Steps: + 1. matcher.match_or_create_house() — find or create canonical houses row for this + listing's address/coords. Uses Tier 0-3 matching (cadastr → source_exact → + fingerprint → geo_proximity → new). + 2. matcher.upsert_listing_source() — register (ext_source, ext_id) → listing_id in + listing_sources. Idempotent via UNIQUE (ext_source, ext_id). + + The matching method recorded is 'source_link': scraper already deduped by + `dedup_hash` (INSERT … ON CONFLICT DO UPDATE), so this is a direct registration + rather than a fuzzy listing-level match. + + ext_id source: lot.source_id if present, else dedup_hash (Yandex without + stable source_id falls back to URL-based dedup_hash — same hash on re-scrape). + + Skips silently if: + - lot has no source_id AND no address/lat/lon (cannot match house anyway) + + Raises on DB errors — caller wraps in try/except + SAVEPOINT. + """ + ext_id = lot.source_id or lot.compute_dedup_hash() + + # House resolution: needs at least address or (lat, lon). Skip otherwise — + # listing_sources requires listing_id but not house linkage, so still upsert. + house_id: int | None = None + if lot.address or (lot.lat is not None and lot.lon is not None): + # Use house_source/house_ext_id when scraper extracted them (Avito Houses + # catalog link, Cian newbuilding). Falls back to per-listing identity + # so each unique listing creates its own row if no canonical house exists. + h_src = lot.house_source or lot.source + h_ext = lot.house_ext_id or ext_id + # No inner begin_nested here — the caller (save_listings) already wraps this + # entire function in a per-row SAVEPOINT (backend.md SAVEPOINT pattern). + # A second nested SAVEPOINT inside would be redundant: if match_or_create_house + # raises a DB error, that error propagates out of _link_listing_to_house, + # the caller's SAVEPOINT rolls back only the hook work (not the listings INSERT), + # and the caller logs a match_failed warning. One SAVEPOINT level is sufficient. + house_id, _conf, _method = matcher.match_or_create_house( + db, + ext_source=h_src, + ext_id=h_ext, + address=lot.address, + lat=lot.lat, + lon=lot.lon, + year_built=lot.year_built, + building_cadastral_number=lot.building_cadastral_number, + cadastral_number=lot.cadastral_number or lot.kadastr_num, + source_url=lot.house_url or lot.source_url, + ) + + # Mirror the resolved house into listings.house_id_fk so direct + # `JOIN houses ON house_id_fk = id` keeps working on freshly scraped / + # re-scraped active rows. Without this the FK is populated only by the + # offline backfill (063_*.sql / backfill_listing_sources.py), leaving + # new active listings unlinked (avito 0/444, cian ~7%). Mirrors the + # backfill script's column/param style. IS DISTINCT FROM avoids a no-op + # write when the linkage is already correct. + if house_id is not None: + db.execute( + text( + "UPDATE listings " + " SET house_id_fk = CAST(:hid AS bigint) " + " WHERE id = CAST(:lid AS bigint) " + " AND house_id_fk IS DISTINCT FROM CAST(:hid AS bigint)" + ), + {"hid": house_id, "lid": listing_id}, + ) + + matcher.upsert_listing_source( + db, + listing_id=listing_id, + ext_source=lot.source, + ext_id=str(ext_id), + method="source_link", + confidence=1.0, + price_rub=lot.price_rub, + area_m2=lot.area_m2, + floor=lot.floor, + rooms_count=lot.rooms, + source_url=lot.source_url, + source_data={"house_id": house_id} if house_id else None, + ) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py new file mode 100644 index 00000000..f134a886 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py @@ -0,0 +1,269 @@ +"""browser_fetcher.py — HTTP-клиент к tradein-browser сервису (#884/#905). + +Тонкий клиент над tradein-browser контейнером, который запускает AsyncCamoufox +локально и экспонирует HTTP API (POST /fetch). + +Публичный интерфейс не изменился: + + async with BrowserFetcher() as fetcher: + html = await fetcher.fetch("https://example.com") + +Внутреннее устройство: httpx.AsyncClient + POST к settings.browser_http_endpoint. +Recycle, crash-recovery и управление браузером живут на стороне сервера (server.py). +При HTTPError / ConnectError делает одну повторную попытку после короткой паузы, +затем пробрасывает исключение. + +Архитектура выбрана потому, что playwright WS-сервер (launch_server) несовместим +с playwright >=1.45: ``browserServerImpl.js`` отсутствует → MODULE_NOT_FOUND. +Локальный AsyncCamoufox + HTTP — работающая альтернатива. +""" + +from __future__ import annotations + +import asyncio +import logging + +import httpx + +logger = logging.getLogger(__name__) + +_RETRY_SLEEP_S: float = 1.0 +_HTTP_TIMEOUT_S: float = 120.0 # навигация медленная → щедрый таймаут + + +class BrowserFetcher: + """Async context manager: HTTP-клиент к tradein-browser HTTP-сервису. + + Использование:: + + async with BrowserFetcher() as fetcher: + html = await fetcher.fetch("https://example.com") + """ + + def __init__( + self, + source: str = "avito", + fetch_timeout_s: float = _HTTP_TIMEOUT_S, + *, + endpoint: str, + ) -> None: + # source — логический источник ("avito"/"cian"/"yandex"/"domclick"). Сервер + # роутит /fetch по нему на отдельный браузер+прокси, когда включён + # FEATURE_BROWSER_POOL_ENABLED (Phase 1). При выключенном флаге source + # игнорируется — поведение не меняется. + # fetch_timeout_s — таймаут httpx-клиента для POST /fetch. Yandex-путь передаёт + # 30s чтобы один проблемный combo занимал ≤30s×retries вместо 120s×retries. + # endpoint — HTTP-эндпоинт tradein-browser сервиса. Инжектируется вызывающей + # стороной (продуктовый код передаёт settings.browser_http_endpoint / + # ScraperConfig.browser_http_endpoint) — kit НЕ импортирует app.core.config. + self._source = source + self._fetch_timeout_s = fetch_timeout_s + self._client: httpx.AsyncClient | None = None + self._endpoint: str | None = endpoint + + # ── lifecycle ────────────────────────────────────────────────────────────── + + async def __aenter__(self) -> BrowserFetcher: + assert self._endpoint is not None, "BrowserFetcher: endpoint не задан" + self._client = httpx.AsyncClient(timeout=self._fetch_timeout_s) + logger.info("BrowserFetcher: клиент создан, endpoint=%s", self._endpoint) + return self + + async def __aexit__(self, *_: object) -> None: + if self._client is not None: + await self._client.aclose() + self._client = None + logger.debug("BrowserFetcher: клиент закрыт") + + # ── public API ───────────────────────────────────────────────────────────── + + async def fetch(self, url: str) -> str: + """Запрашивает HTML страницы через tradein-browser HTTP-сервис. + + При HTTPError или ConnectError делает одну повторную попытку после + короткой паузы. Остальные исключения всплывают к вызывающему коду. + + Returns: + Полный HTML-контент страницы. + """ + assert self._client is not None, "BrowserFetcher: используй как async context manager" + assert self._endpoint is not None, "BrowserFetcher: endpoint не задан" + + try: + return await self._post_fetch(url) + except (httpx.HTTPError, httpx.TransportError) as exc: + logger.warning( + "BrowserFetcher: ошибка запроса (%s), retry через %.1fs: %s", + type(exc).__name__, + _RETRY_SLEEP_S, + url, + ) + await asyncio.sleep(_RETRY_SLEEP_S) + return await self._post_fetch(url) + + async def fetch_json( + self, + url: str, + *, + method: str = "GET", + headers: dict[str, str] | None = None, + body: str | None = None, + origin: str | None = None, + ) -> dict: + """In-page fetch() через sidecar /fetch-json. Возвращает {"status": int, "body": str}. + + body — уже сериализованная строка (caller делает json.dumps для POST). + origin — same-origin страница, на которую перейдёт камуфокс перед fetch. + + При HTTPError / TransportError делает одну повторную попытку после короткой + паузы. Остальные исключения всплывают к вызывающему коду. + """ + assert self._client is not None, "BrowserFetcher: используй как async context manager" + assert self._endpoint is not None, "BrowserFetcher: endpoint не задан" + + try: + return await self._post_fetch_json(url, method, headers, body, origin) + except (httpx.HTTPError, httpx.TransportError) as exc: + logger.warning( + "BrowserFetcher: ошибка fetch-json запроса (%s), retry через %.1fs: %s", + type(exc).__name__, + _RETRY_SLEEP_S, + url, + ) + await asyncio.sleep(_RETRY_SLEEP_S) + return await self._post_fetch_json(url, method, headers, body, origin) + + async def login( + self, + *, + url: str, + email: str, + password: str, + email_selector: str, + password_selector: str, + submit_selector: str, + success_cookie: str, + pre_click_selectors: list[str] | None = None, + wait_ms: int | None = None, + ) -> dict[str, str]: + """Логинится через tradein-browser /login и возвращает cookies как dict name→value. + + Использует увеличенный таймаут (60s) — логин медленнее обычного fetch. + При HTTPError / TransportError делает одну повторную попытку. + + Returns: + Плоский словарь {cookie_name: cookie_value}. + """ + assert self._client is not None, "BrowserFetcher: используй как async context manager" + assert self._endpoint is not None, "BrowserFetcher: endpoint не задан" + + body: dict[str, object] = { + "url": url, + "email": email, + "password": password, + "email_selector": email_selector, + "password_selector": password_selector, + "submit_selector": submit_selector, + "success_cookie": success_cookie, + "pre_click_selectors": pre_click_selectors or [], + } + if wait_ms is not None: + body["wait_ms"] = wait_ms + + try: + return await self._post_login(body) + except (httpx.HTTPError, httpx.TransportError) as exc: + logger.warning( + "BrowserFetcher: ошибка login-запроса (%s), retry через %.1fs", + type(exc).__name__, + _RETRY_SLEEP_S, + ) + await asyncio.sleep(_RETRY_SLEEP_S) + return await self._post_login(body) + + # ── internal ─────────────────────────────────────────────────────────────── + + async def _post_fetch(self, url: str) -> str: + """Один HTTP POST к /fetch эндпоинту сервиса.""" + assert self._client is not None + assert self._endpoint is not None + + resp = await self._client.post( + f"{self._endpoint}/fetch", + json={"url": url, "source": self._source}, + ) + resp.raise_for_status() + data: dict[str, str] = resp.json() + html = data["html"] + logger.debug("BrowserFetcher: fetch OK url=%r html_len=%d", url, len(html)) + return html + + async def _post_fetch_json( + self, + url: str, + method: str, + headers: dict[str, str] | None, + body: str | None, + origin: str | None, + ) -> dict: + """Один HTTP POST к /fetch-json эндпоинту сервиса.""" + assert self._client is not None + assert self._endpoint is not None + + resp = await self._client.post( + f"{self._endpoint}/fetch-json", + json={ + "url": url, + "source": self._source, + "method": method, + "headers": headers or {}, + "body": body, + "origin": origin, + }, + ) + resp.raise_for_status() + data: dict = resp.json() + # Defensive: контракт сервера — {"status": int, "body": str} (зеркалит как + # _post_fetch читает data["html"]). Если ключи пропали (несовместимый сервер / + # прокинутый error-payload) — падаем с понятной ошибкой, а не KeyError ниже по + # стеку у адаптера, который ждёт r["status"]/r["body"]. + if "status" not in data or "body" not in data: + raise RuntimeError( + f"BrowserFetcher: /fetch-json вернул некорректный ответ " + f"(нет 'status'/'body'): keys={sorted(data)} url={url!r}" + ) + logger.debug( + "BrowserFetcher: fetch-json OK url=%r status=%s body_len=%d", + url, + data.get("status"), + len(data.get("body") or ""), + ) + return data + + async def _post_login(self, body: dict[str, object]) -> dict[str, str]: + """Один HTTP POST к /login эндпоинту сервиса.""" + assert self._client is not None + assert self._endpoint is not None + + resp = await self._client.post( + f"{self._endpoint}/login", + json=body, + timeout=httpx.Timeout(60.0), + ) + if resp.status_code == 502: + data = resp.json() + has_screenshot = bool(data.get("screenshot_b64")) + logger.debug( + "BrowserFetcher: login 502 — has_screenshot=%s page_url=%r", + has_screenshot, + data.get("page_url"), + ) + raise RuntimeError( + f"browser login failed: {data.get('error')} url={data.get('page_url')}" + ) + resp.raise_for_status() + data = resp.json() + raw: list[dict[str, object]] = data["cookies"] + result = {str(c["name"]): str(c["value"]) for c in raw} + logger.info("BrowserFetcher: login OK cookie_count=%d", len(result)) + return result diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/cian_state_parser.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/cian_state_parser.py new file mode 100644 index 00000000..f6a46a7a --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/cian_state_parser.py @@ -0,0 +1,160 @@ +"""Shared utility для извлечения Cian Redux state из _cianConfig. + +Pattern: window._cianConfig[''].push({key:..., value:..., priority:..., filter:...}) +Used by SERP / detail / newbuilding / valuation scrapers. +""" + +import json +import logging +import re +from typing import Any + +logger = logging.getLogger(__name__) + +# Match both patterns Cian uses: +# 1. window._cianConfig['mfe'].push({key:..., value:..., priority:..., filter:...}) +# 2. window._cianConfig['mfe'] = window._cianConfig['mfe'] || []; +# window._cianConfig['mfe'].push({...}) +_RE_CIAN_PUSH = re.compile( + r"window\._cianConfig\['(?P[\w-]+)'\]" + r"(?:\.push\(\s*|" + r"\s*=\s*window\._cianConfig\['[\w-]+'\]\s*\|\|\s*\[\];\s*" + r"window\._cianConfig\['[\w-]+'\]\.push\(\s*)" + r"\{\s*key:\s*['\"](?P\w+)['\"]\s*,\s*" + r"value:\s*(?P.+?)\s*,\s*" + r"priority:", + re.DOTALL, +) + +# 2026-05 drift (#639): Cian сменил формат на +# window._cianConfig['mfe'] = (window._cianConfig['mfe'] || []).concat([{"key":..,"value":..},..]) +# вместо .push({key:..}). Ключи теперь в JSON-кавычках, аргумент concat([...]) — валидный +# JSON-массив → парсим его целиком (надёжнее regexp по одному value). Старый .push формат +# (_RE_CIAN_PUSH) оставлен как fallback для legacy-фикстур. +_RE_CIAN_CONCAT = re.compile( + r"window\._cianConfig\['(?P[\w-]+)'\]\s*=\s*" + r"\(\s*window\._cianConfig\['[\w-]+'\]\s*\|\|\s*\[\]\s*\)\.concat\(\s*" +) + + +def _balanced_array(s: str, start: int) -> str | None: + """Срез s[start:end] — сбалансированный ``[...]`` от индекса '[', с учётом JSON-строк.""" + depth = 0 + in_str = False + esc = False + for i in range(start, len(s)): + c = s[i] + if in_str: + if esc: + esc = False + elif c == "\\": + esc = True + elif c == '"': + in_str = False + continue + if c == '"': + in_str = True + elif c == "[": + depth += 1 + elif c == "]": + depth -= 1 + if depth == 0: + return s[start : i + 1] + return None + + +def _extract_from_concat(html: str, mfe: str, key: str) -> Any | None: + """Новый формат: .concat([{"key":..,"value":..},..]) — массив-аргумент = валидный JSON.""" + for m in _RE_CIAN_CONCAT.finditer(html): + if m.group("mfe") != mfe: + continue + start = m.end() + if start >= len(html) or html[start] != "[": + start = html.find("[", m.end()) + if start < 0: + continue + arr = _balanced_array(html, start) + if arr is None: + continue + try: + items = json.loads(arr) + except json.JSONDecodeError as exc: + logger.debug("concat array parse failed for mfe=%s key=%s: %s", mfe, key, exc) + continue + for it in items: + if isinstance(it, dict) and it.get("key") == key: + return it.get("value") + return None + + +def extract_state(html: str, mfe: str, key: str = "initialState") -> dict[str, Any] | None: + """Extract state.value where _cianConfig[] entry.key matches. + + Поддерживает оба формата Cian: новый ``.concat([{"key":..,"value":..}])`` (2026-05, + #639) и legacy ``.push({key:..,value:..})``. Returns parsed JSON dict, or None. + """ + # Новый формат (concat) — приоритетный. + concat_val = _extract_from_concat(html, mfe, key) + if concat_val is not None: + return concat_val + + for match in _RE_CIAN_PUSH.finditer(html): + if match.group("mfe") != mfe or match.group("key") != key: + continue + + value_str = match.group("value").strip() + + # Strategy 1: direct JSON parse (value уже JSON-like) + try: + return json.loads(value_str) + except json.JSONDecodeError: + pass + + # Strategy 2: JSON.parse("escaped string") + # Use json.loads on the outer quoted string to unescape atomically — avoids + # order-dependent manual .replace() chains that break when descriptions contain + # backslashes (e.g. '\\' in listing text corrupts the rest of the parse). + m_jp = re.match(r'JSON\.parse\(\s*("(?:[^"\\]|\\.)*")\s*\)\s*$', value_str, re.DOTALL) + if m_jp: + outer_json_str = m_jp.group(1) # the full "..." including surrounding quotes + try: + inner = json.loads(outer_json_str) # inner is now a Python str + except json.JSONDecodeError as exc: + logger.debug( + "JSON.parse() outer unescape failed for mfe=%s key=%s: %s", mfe, key, exc + ) + continue + try: + return json.loads(inner) + except json.JSONDecodeError as exc: + logger.debug("JSON.parse() inner parse failed for mfe=%s key=%s: %s", mfe, key, exc) + continue + + # Strategy 3: raw JS object literal — try demjson3 (optional dep) + try: + import demjson3 # type: ignore[import-untyped] + + return demjson3.decode(value_str) + except ImportError: + logger.debug( + "demjson3 not installed; cannot parse raw JS object for mfe=%s key=%s", mfe, key + ) + except Exception as exc: + logger.debug("demjson3 parse failed for mfe=%s key=%s: %s", mfe, key, exc) + + return None + + +def extract_all_states(html: str) -> dict[str, dict[str, Any]]: + """Returns {mfe_name: {key: parsed_value}} for all push() entries. + + Useful for discovery / debugging during scraper development. + """ + result: dict[str, dict[str, Any]] = {} + for match in _RE_CIAN_PUSH.finditer(html): + mfe = match.group("mfe") + key = match.group("key") + parsed = extract_state(html, mfe, key) + if parsed is not None: + result.setdefault(mfe, {})[key] = parsed + return result diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/domclick_exceptions.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/domclick_exceptions.py new file mode 100644 index 00000000..247de7fe --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/domclick_exceptions.py @@ -0,0 +1,27 @@ +"""DomClick-specific exceptions для anti-bot detection.""" + + +class DomClickBlockedError(Exception): + """DomClick BFF вернул QRATOR block-страницу (HTTP 200 + block HTML). + + QRATOR (qrator.net) — WAF/DDoS-защита domclick.ru. Блокирует datacenter-IP + и возвращает HTML с маркерами: "qrator", "bot_mitigation", "система защиты", + "403 | домклик", "captcha", "access denied". + + Обходится через shared mobile proxy (BrowserFetcher(source="domclick") → + generic provider → мобильный egress). + """ + + +class DomClickParseError(Exception): + """DomClick detail-карточка получена (HTTP 200, не block), но SSR-стейт + не извлекается / не парсится. + + Возникает в Layer B (domclick_detail.py) когда: + - `window.__SSR_STATE__` не найден в HTML (но это и не challenge-страница); + - object-литерал не сбалансирован по скобкам; + - json.loads падает после санитайза bare `undefined` → `null`. + + В отличие от DomClickBlockedError (anti-bot), это сигнал дрейфа схемы карточки + или повреждённого ответа — оркестратор считает его parse-failure, а не блоком. + """ diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/price_brackets.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/price_brackets.py new file mode 100644 index 00000000..f3c075fc --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/price_brackets.py @@ -0,0 +1,32 @@ +"""Общие seed price-brackets для exhaustive-скрапперов (avito/cian/yandex). + +EKB-настроенные (анализ прод-распределения вторички ЕКБ). Структура city-dict — +готова под per-city (multi-city: Москва пик 15-40М, Тюмень — иные; multi-city — +отдельная фича, см. vault decision). Сейчас единственный город — ekaterinburg. +""" + +# (lo, hi) рубли; hi=None — открытый верхний брекет (без потолка). Закрытые +# брекеты соприкасаются — вызывающий применяет hi-1 (эксклюзивный верх), дедуп +# по source_id страхует на стыках. +_EKB_SECONDARY: list[tuple[int, int | None]] = [ + (0, 4_000_000), + (4_000_000, 5_000_000), + (5_000_000, 6_000_000), + (6_000_000, 7_000_000), + (7_000_000, 8_000_000), + (8_000_000, 10_000_000), + (10_000_000, 13_000_000), + (13_000_000, 17_000_000), + (17_000_000, 25_000_000), + (25_000_000, 250_000_000), + (250_000_000, None), # 250М+ открытый catch-all (без потолка) +] + +CITY_PRICE_SEED_BRACKETS: dict[str, list[tuple[int, int | None]]] = { + "ekaterinburg": _EKB_SECONDARY, +} + + +def get_price_seed_brackets(city: str = "ekaterinburg") -> list[tuple[int, int | None]]: + """Seed price-brackets для города (fallback на ekaterinburg).""" + return CITY_PRICE_SEED_BRACKETS.get(city, _EKB_SECONDARY) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/repair_state_normalizer.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/repair_state_normalizer.py new file mode 100644 index 00000000..d31d70b2 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/repair_state_normalizer.py @@ -0,0 +1,162 @@ +"""Нормализация repair_state → каноничный enum для listings.repair_state. + +Целевой enum: needs_repair / standard / good / excellent. +Используется во всех инgest-путях: avito_detail, cian SERP, cian_detail. + +Источники raw-значений: +- Avito HTML: «Ремонт: евро» → avito_detail._REPAIR_MAP → «euro» → good +- Cian SERP JSON: offer.decoration → «euro» / «cosmetic» / «without» / ... +- Cian Detail JSON: offer.repairType → «cosmetic» / «design» / «no» / «euro» / ... + +Маппинг основан на _IMV_REPAIR_MAP из estimator.py (обратная форма): + needs_repair→«required», standard→«cosmetic», good→«euro», excellent→«designer». + +Связь с estimator._REPAIR_COEF: + needs_repair=0.94 / standard=1.00 / good=1.05 / excellent=1.10 +""" + +from __future__ import annotations + +import logging +import re + +logger = logging.getLogger(__name__) + +# Все известные raw-значения с источником в комментарии +_RAW_TO_ENUM: dict[str, str] = { + # ── needs_repair ──────────────────────────────────────────────────────────── + "required": "needs_repair", # Avito: «требуется» / «без ремонта» (через _REPAIR_MAP) + "without": "needs_repair", # Cian: без ремонта + "no": "needs_repair", # Cian: нет ремонта (синоним without) + "rough": "needs_repair", # Cian/общий: черновая отделка + "prefine": "needs_repair", # Cian detail: предчистовая/whitebox (offer.decoration, #1791) + # ── standard ──────────────────────────────────────────────────────────────── + "cosmetic": "standard", # Avito/Cian: косметический + # ── good ──────────────────────────────────────────────────────────────────── + "euro": "good", # Avito/Cian: евро + "fine": "good", # Cian: хорошая отделка (предчистовая / whitebox) + # ── excellent ─────────────────────────────────────────────────────────────── + "designer": "excellent", # Avito: дизайнерский + "design": "excellent", # Cian: дизайнерский (camelCase вариант) +} + +# Допустимые значения целевого enum — для pass-through защиты +_VALID_ENUM: frozenset[str] = frozenset({"needs_repair", "standard", "good", "excellent"}) + + +def normalize_repair_state(raw: str | None) -> str | None: + """Преобразовать raw repair-значение в каноничный enum. + + Если raw уже является корректным enum-значением — вернуть as-is (идемпотентно). + Если raw не распознан — вернуть None + залогировать WARNING один раз. + + Args: + raw: сырое значение из парсера (e.g. «euro», «without», «design») + или None. + + Returns: + Одно из needs_repair / standard / good / excellent, либо None. + """ + if raw is None: + return None + # Pass-through: уже нормализованный enum (idempotency — нужно при ре-обработке) + if raw in _VALID_ENUM: + return raw + # Cian отдаёт camelCase ('preFine', 'whiteBox') — ключи карты в нижнем регистре, + # сравниваем case-insensitive, чтобы не плодить дубль-ключи на каждый регистр. + result = _RAW_TO_ENUM.get(raw) or _RAW_TO_ENUM.get(raw.lower()) + if result is None: + logger.warning("repair_state_normalizer: unknown raw value %r — stored as NULL", raw) + return result + + +# ── Inference из текста описания (#622) ────────────────────────────────────── +# Когда структурное поле repair отсутствует (avito ~1% / cian ~5% / yandex 0%), +# извлекаем состояние из текста описания регэкспами по русским фразам. +# +# Порядок ВАЖЕН: проверяем от самого сильного сигнала к слабому +# excellent → good → needs_repair → standard, +# чтобы «после черновой сделали дизайнерский ремонт» дал excellent, а не +# needs_repair. Первое совпадение по порядку выигрывает. +# +# Целевой enum выровнен по normalize_repair_state() — те же 4 значения. +_TEXT_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = ( + # ── excellent: дизайнерский / премиум / эксклюзивный ремонт ────────────── + ( + re.compile( + r"дизайнерск(?:ий|ого|ом|ая)\s+ремонт" + r"|дизайнерск(?:ая|ой)\s+отделк" + r"|премиальн\w*\s+ремонт" + r"|эксклюзивн\w*\s+ремонт" + r"|авторск\w*\s+ремонт", + re.IGNORECASE, + ), + "excellent", + ), + # ── good: евроремонт / отличный / свежий / современный ремонт ──────────── + ( + re.compile( + r"евроремонт" + r"|евро[\s-]?ремонт" + r"|отличн\w*\s+ремонт" + r"|свеж\w*\s+ремонт" + r"|современн\w*\s+ремонт" + r"|качественн\w*\s+ремонт" + r"|ремонт\s+в\s+отличн", + re.IGNORECASE, + ), + "good", + ), + # ── needs_repair: без отделки / черновая / требует ремонта / под ремонт ── + ( + re.compile( + r"без\s+отделк" + r"|без\s+ремонт" + r"|черновая\s+отделк" + r"|черновой\s+ремонт" + r"|требует(?:ся)?\s+ремонт" + r"|требует(?:ся)?\s+космет" + r"|нужен\s+ремонт" + r"|под\s+ремонт" + r"|под\s+чистов\w+\s+отделк" + r"|предчистов", + re.IGNORECASE, + ), + "needs_repair", + ), + # ── standard: косметический / обычный / жилое состояние ───────────────── + ( + re.compile( + r"косметическ\w*\s+ремонт" + r"|косметическ\w*\s+отделк" + r"|обычн\w*\s+ремонт" + r"|жил(?:ое|ом)\s+состоян" + r"|хорош\w*\s+состоян", + re.IGNORECASE, + ), + "standard", + ), +) + + +def infer_repair_state_from_text(description: str | None) -> str | None: + """Извлечь repair_state из текста описания объявления (#622). + + Fallback-путь для повышения покрытия listings.repair_state, когда + структурное поле источника отсутствует. Проверяет паттерны в порядке + убывания силы сигнала (excellent → good → needs_repair → standard); + первое совпадение выигрывает. + + Args: + description: текст описания объявления (avito/cian/yandex) или None. + + Returns: + Одно из needs_repair / standard / good / excellent, либо None если + ни один паттерн не сработал (НЕ фабрикуем — unknown остаётся NULL). + """ + if not description: + return None + for pattern, enum_value in _TEXT_PATTERNS: + if pattern.search(description): + return enum_value + return None diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/snapshot_writer.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/snapshot_writer.py new file mode 100644 index 00000000..624637bd --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/snapshot_writer.py @@ -0,0 +1,92 @@ +"""listings_snapshots writer — point-in-time observation per listing per day. + +Таблица listings_snapshots (016_listings_snapshots.sql): + PRIMARY KEY (listing_id, snapshot_date) — максимум 1 snapshot в сутки. + ON CONFLICT DO UPDATE — берём последний за день (перезаписываем при повторном run'е). + +Используется двумя путями: + 1. SERP scrape (save_listings в base.py) — записывает цену + позицию в выдаче + каждый раз когда listing появляется в поиске. + 2. Detail backfill (save_detail_enrichment в cian_detail.py) — записывает цену + из detail-страницы для listings которые раньше не имели snapshot'а. +""" +from __future__ import annotations + +import logging +from datetime import date + +from sqlalchemy import text +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + + +def upsert_listing_snapshot( + db: Session, + *, + listing_id: int, + price_rub: int, + price_per_m2: int | None = None, + run_id: int | None = None, + snapshot_date: date | None = None, + position_in_serp: int | None = None, + status: str | None = "active", +) -> None: + """Записать / обновить snapshot для listing за текущий (или указанный) день. + + Идемпотентно: повторный вызов с теми же listing_id + snapshot_date перезаписывает + строку (берём последние данные за день — ON CONFLICT DO UPDATE). + + Args: + db: открытая SQLAlchemy session. Commit делает CALLER. + listing_id: PK из таблицы listings. + price_rub: текущая цена в рублях. + price_per_m2: цена за кв.м (вычисляется в caller'е, None если нет). + run_id: FK scrape_runs.id — с каким run'ом связан snapshot (None если backfill + запускается вне run'а). + snapshot_date: дата наблюдения; если None — используется CURRENT_DATE (по БД). + position_in_serp: позиция в SERP (1..60+), только при SERP-scrape. + status: 'active' / 'closed' / None. Default 'active' при обычном scrape. + """ + db.execute( + text(""" + INSERT INTO listings_snapshots + (listing_id, snapshot_date, run_id, + price_rub, price_per_m2, position_in_serp, status, observed_at) + VALUES ( + CAST(:lid AS bigint), + COALESCE(CAST(:snap_date AS date), CURRENT_DATE), + CAST(:run_id AS bigint), + CAST(:price AS bigint), + CAST(:ppm2 AS int), + CAST(:pos AS int), + :status, + NOW() + ) + ON CONFLICT (listing_id, snapshot_date) DO UPDATE SET + run_id = COALESCE(EXCLUDED.run_id, listings_snapshots.run_id), + price_rub = EXCLUDED.price_rub, + price_per_m2 = COALESCE(EXCLUDED.price_per_m2, listings_snapshots.price_per_m2), + position_in_serp = COALESCE( + EXCLUDED.position_in_serp, listings_snapshots.position_in_serp + ), + status = COALESCE(EXCLUDED.status, listings_snapshots.status), + observed_at = EXCLUDED.observed_at + """), + { + "lid": listing_id, + "snap_date": snapshot_date, + "run_id": run_id, + "price": price_rub, + "ppm2": price_per_m2, + "pos": position_in_serp, + "status": status, + }, + ) + logger.debug( + "upsert_listing_snapshot: listing_id=%s date=%s price=%s run_id=%s", + listing_id, + snapshot_date or "CURRENT_DATE", + price_rub, + run_id, + ) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/yandex_helpers.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/yandex_helpers.py new file mode 100644 index 00000000..a33d4501 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/yandex_helpers.py @@ -0,0 +1,355 @@ +"""Yandex Realty parsing helpers — JSON-LD + regex + DOM utilities for SERP, detail, newbuilding, +and valuation parsers.""" + +from __future__ import annotations + +import json +import re +from datetime import date, timedelta +from typing import Any + +from selectolax.parser import HTMLParser + +__all__ = [ + "RE_AGENCY_FOUNDED", + "RE_AGENCY_OBJECTS", + "RE_FLOOR", + "RE_JK_ID", + "RE_METRO_WALK", + "RE_OFFER_ID", + "RE_PPM2", + "RE_PRICE", + "RE_STREET_ID", + "RE_TITLE_AREA", + "RE_TITLE_ROOMS", + "RE_VIEWS", + "RE_YEAR", + "RU_MONTHS", + "RU_MONTH_NOMINATIVE", + "extract_json_ld", + "find_ld_by_type", + "parse_dmy", + "parse_house_class", + "parse_house_type", + "parse_listing_date", + "parse_ru_date", + "parse_rub", +] + +# --------------------------------------------------------------------------- +# Section 1 — JSON-LD extraction +# --------------------------------------------------------------------------- + + +def extract_json_ld(html: str) -> list[dict[str, Any]]: + """Extract all