feat(scraper-kit): copy core layer (base/save_listings/transport) with protocol injection, strangler (#2132) #2153

Merged
lekss361 merged 2 commits from feat/scraper-kit-core-copy into main 2026-07-02 15:43:16 +00:00
16 changed files with 2848 additions and 3 deletions

View file

@ -0,0 +1,211 @@
"""Продуктовые адаптеры над `app.services.*`, удовлетворяющие scraper_kit.contracts.
Тонкие врапперы, инжектируемые в скрапперы (после переноса в scraper_kit, шаги C-G
#2131) вместо прямых импортов `app.*`. Каждый класс структурно удовлетворяет
соответствующему Protocol'у из `scraper_kit.contracts`:
RealMatcherAdapter HouseMatcher
RealScraperConfig ScraperConfig
RealSessionFactory SessionFactory
RealEnrichmentJobs EnrichmentJobs
Пока НИКТО их не использует скрапперы всё ещё на прямых импортах. Это подготовка:
переключение вызовов на инжекцию отдельный шаг (C). Держим сигнатуры в точном
соответствии с исходными функциями, чтобы шаг C состыковался без правок контрактов.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from app.core.config import settings as _settings
from app.core.db import SessionLocal as _SessionLocal
from app.services.house_dedup_merge import run_house_dedup_merge as _run_house_dedup_merge
from app.services.house_imv_backfill import backfill_house_imv as _backfill_house_imv
from app.services.matching import match_or_create_house as _match_or_create_house
from app.services.matching import upsert_listing_source as _upsert_listing_source
from app.services.yandex_address_backfill import (
backfill_yandex_addresses as _backfill_yandex_addresses,
)
if TYPE_CHECKING:
from sqlalchemy.orm import Session
from app.services.house_imv_backfill import HouseIMVBackfillResult
from app.services.yandex_address_backfill import YandexAddressBackfillResult
class RealMatcherAdapter:
"""HouseMatcher-адаптер над `app.services.matching`."""
def match_or_create_house(
self,
db: Session,
ext_source: str,
ext_id: str,
address: str | None = None,
lat: float | None = None,
lon: float | None = None,
*,
year_built: int | None = None,
building_cadastral_number: str | None = None,
cadastral_number: str | None = None,
source_url: str | None = None,
) -> tuple[int, float, str]:
return _match_or_create_house(
db,
ext_source,
ext_id,
address,
lat,
lon,
year_built=year_built,
building_cadastral_number=building_cadastral_number,
cadastral_number=cadastral_number,
source_url=source_url,
)
def upsert_listing_source(
self,
db: Session,
*,
listing_id: int,
ext_source: str,
ext_id: str,
method: str,
confidence: float,
price_rub: float | None = None,
area_m2: float | None = None,
floor: int | None = None,
rooms_count: int | None = None,
source_url: str | None = None,
source_data: dict[str, Any] | None = None,
) -> None:
_upsert_listing_source(
db,
listing_id=listing_id,
ext_source=ext_source,
ext_id=ext_id,
method=method,
confidence=confidence,
price_rub=price_rub,
area_m2=area_m2,
floor=floor,
rooms_count=rooms_count,
source_url=source_url,
source_data=source_data,
)
class RealScraperConfig:
"""ScraperConfig-адаптер: read-only проекция singleton'а `settings`.
Свойства проксируют в `app.core.config.settings`, включая computed-properties
(`scraper_proxy_url`, `cian_proxy_url`), чтобы скрапперы читали ровно те же
значения, что и сейчас через прямой `settings.<name>`.
"""
@property
def scraper_fetch_mode(self) -> str:
return _settings.scraper_fetch_mode
@property
def browser_http_endpoint(self) -> str:
return _settings.browser_http_endpoint
@property
def scraper_proxy_url(self) -> str | None:
return _settings.scraper_proxy_url
@property
def avito_proxy_rotate_url(self) -> str | None:
return _settings.avito_proxy_rotate_url
@property
def avito_proxy_max_rotations(self) -> int:
return _settings.avito_proxy_max_rotations
@property
def avito_serp_ekb_only(self) -> bool:
return _settings.avito_serp_ekb_only
@property
def yandex_proxy_rotate_url(self) -> str | None:
return _settings.yandex_proxy_rotate_url
@property
def cian_proxy_url(self) -> str | None:
return _settings.cian_proxy_url
@property
def cian_valuation_min_rub(self) -> float:
return _settings.cian_valuation_min_rub
@property
def cian_valuation_max_rub(self) -> float:
return _settings.cian_valuation_max_rub
@property
def glitchtip_dsn(self) -> str | None:
return _settings.glitchtip_dsn
class RealSessionFactory:
"""SessionFactory-адаптер над `app.core.db.SessionLocal`."""
def __call__(self) -> Session:
return _SessionLocal()
class RealEnrichmentJobs:
"""EnrichmentJobs-адаптер над enrichment-сервисами `app.services.*`."""
async def house_imv_backfill(
self,
db: Session,
*,
batch_size: int = 50,
request_delay_sec: float = 5.0,
only_status: str = "pending",
house_id: int | None = None,
heartbeat: Callable[[], None] | None = None,
) -> HouseIMVBackfillResult:
return await _backfill_house_imv(
db,
batch_size=batch_size,
request_delay_sec=request_delay_sec,
only_status=only_status,
house_id=house_id,
heartbeat=heartbeat,
)
def house_dedup_merge(
self,
db: Session,
*,
run_id: int,
params: dict[str, Any],
) -> dict[str, int]:
return _run_house_dedup_merge(db, run_id=run_id, params=params)
async def yandex_address_backfill(
self,
db: Session,
*,
limit: int = 200,
request_delay_sec: float = 3.0,
) -> YandexAddressBackfillResult:
return await _backfill_yandex_addresses(
db,
limit=limit,
request_delay_sec=request_delay_sec,
)
__all__ = [
"RealEnrichmentJobs",
"RealMatcherAdapter",
"RealScraperConfig",
"RealSessionFactory",
]

View file

@ -0,0 +1,88 @@
"""Проверка, что backend-адаптеры структурно удовлетворяют scraper_kit.contracts.
Двойная проверка:
1. Статическая `_check_*` функции с аннотацией Protocol-типа. mypy проверит,
что каждый адаптер присваивается соответствующему Protocol'у (структурно).
2. Рантайм `isinstance` против `@runtime_checkable` Protocol'ов (проверяет
наличие методов/атрибутов). Ловит рассинхрон имён без запуска mypy.
Границы контрактов зафиксированы для #2131 (шаг B); переключение скрапперов на
инжекцию шаг C.
"""
from __future__ import annotations
from scraper_kit.contracts import (
EnrichmentJobs,
HouseMatcher,
ScraperConfig,
SessionFactory,
)
from app.services.scraper_adapters import (
RealEnrichmentJobs,
RealMatcherAdapter,
RealScraperConfig,
RealSessionFactory,
)
def _check_matcher(m: HouseMatcher) -> None:
"""Статическая проверка: RealMatcherAdapter ⊑ HouseMatcher."""
def _check_config(c: ScraperConfig) -> None:
"""Статическая проверка: RealScraperConfig ⊑ ScraperConfig."""
def _check_session_factory(f: SessionFactory) -> None:
"""Статическая проверка: RealSessionFactory ⊑ SessionFactory."""
def _check_enrichment(j: EnrichmentJobs) -> None:
"""Статическая проверка: RealEnrichmentJobs ⊑ EnrichmentJobs."""
def test_matcher_adapter_satisfies_protocol() -> None:
adapter = RealMatcherAdapter()
_check_matcher(adapter)
assert isinstance(adapter, HouseMatcher)
assert callable(adapter.match_or_create_house)
assert callable(adapter.upsert_listing_source)
def test_scraper_config_satisfies_protocol() -> None:
config = RealScraperConfig()
_check_config(config)
assert isinstance(config, ScraperConfig)
# Ключевые поля читаются без ошибок (проксируют в singleton settings).
assert isinstance(config.scraper_fetch_mode, str)
assert isinstance(config.browser_http_endpoint, str)
assert isinstance(config.avito_proxy_max_rotations, int)
assert isinstance(config.avito_serp_ekb_only, bool)
def test_session_factory_satisfies_protocol() -> None:
factory = RealSessionFactory()
_check_session_factory(factory)
assert isinstance(factory, SessionFactory)
def test_enrichment_jobs_satisfies_protocol() -> None:
jobs = RealEnrichmentJobs()
_check_enrichment(jobs)
assert isinstance(jobs, EnrichmentJobs)
assert callable(jobs.house_imv_backfill)
assert callable(jobs.house_dedup_merge)
assert callable(jobs.yandex_address_backfill)
def test_contracts_module_has_no_app_imports() -> None:
"""contracts.py не должен импортировать app.* (иначе circular / coupling)."""
import inspect
import scraper_kit.contracts as contracts_mod
source = inspect.getsource(contracts_mod)
assert "from app" not in source
assert "import app" not in source

View file

@ -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}"

View file

@ -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"]

View file

@ -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"

View file

@ -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)."""

View file

@ -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,
)

View file

@ -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

View file

@ -0,0 +1,160 @@
"""Shared utility для извлечения Cian Redux state из _cianConfig.
Pattern: window._cianConfig['<mfe>'].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<mfe>[\w-]+)'\]"
r"(?:\.push\(\s*|"
r"\s*=\s*window\._cianConfig\['[\w-]+'\]\s*\|\|\s*\[\];\s*"
r"window\._cianConfig\['[\w-]+'\]\.push\(\s*)"
r"\{\s*key:\s*['\"](?P<key>\w+)['\"]\s*,\s*"
r"value:\s*(?P<value>.+?)\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<mfe>[\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[<mfe>] 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

View file

@ -0,0 +1,178 @@
"""Protocol-контракты (границы) между scraper_kit и продуктовым backend'ом.
Скрапперы (после переноса в scraper_kit, шаги C-G #2131) не должны напрямую
импортировать `app.*`. Вместо этого они принимают инжектируемые зависимости,
описанные здесь через `typing.Protocol` (structural typing): продуктовая сторона
предоставляет конкретные реализации (см. `app.services.scraper_adapters`),
которые структурно удовлетворяют этим Protocol'ам.
ВАЖНО: этот модуль НЕ импортирует `app.*` и не тянет тяжёлых зависимостей в
рантайме. Тип `Session` (SQLAlchemy) используется только в аннотациях под
`TYPE_CHECKING` благодаря `from __future__ import annotations` аннотации не
вычисляются при импорте, поэтому пакет остаётся standalone-импортируемым.
Сигнатуры Protocol'ов зафиксированы по факту из backend-кода:
- HouseMatcher app.services.matching.houses.match_or_create_house
+ app.services.matching.listings.upsert_listing_source
- ScraperConfig поля app.core.config.Settings, читаемые скрапперами
- SessionFactory app.core.db.SessionLocal (sessionmaker Session)
- EnrichmentJobs app.services.house_imv_backfill / house_dedup_merge /
yandex_address_backfill
"""
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
if TYPE_CHECKING:
from sqlalchemy.orm import Session
@runtime_checkable
class HouseMatcher(Protocol):
"""Кросс-источниковый матчинг домов/листингов (продуктовая логика matching-сервиса).
Реализуется `app.services.scraper_adapters.RealMatcherAdapter`, который проксирует
в `app.services.matching`. Скрапперы вызывают методы вместо прямого импорта.
"""
def match_or_create_house(
self,
db: Session,
ext_source: str,
ext_id: str,
address: str | None = ...,
lat: float | None = ...,
lon: float | None = ...,
*,
year_built: int | None = ...,
building_cadastral_number: str | None = ...,
cadastral_number: str | None = ...,
source_url: str | None = ...,
) -> tuple[int, float, str]:
"""Найти или создать канонический дом.
Returns:
(house_id, confidence [0.0, 1.0], method), где method {
'cadastr_exact', 'source_exact', 'fingerprint', 'geo_proximity', 'new'
}.
"""
...
def upsert_listing_source(
self,
db: Session,
*,
listing_id: int,
ext_source: str,
ext_id: str,
method: str,
confidence: float,
price_rub: float | None = ...,
area_m2: float | None = ...,
floor: int | None = ...,
rooms_count: int | None = ...,
source_url: str | None = ...,
source_data: dict[str, Any] | None = ...,
) -> None:
"""Зарегистрировать/обновить строку listing_sources для (ext_source, ext_id)."""
...
@runtime_checkable
class ScraperConfig(Protocol):
"""Конфигурация, читаемая скрапперами из `app.core.config.settings`.
Поля собраны grep'ом `settings.<name>` по `app/services/scrapers/`. Типы —
как в `Settings` (proxy-свойства возвращают `str | None`). Реализуется
`app.services.scraper_adapters.RealScraperConfig` (проксирует в singleton
`settings`); pydantic-`Settings` также структурно удовлетворяет этому Protocol.
"""
# Режим загрузки: "curl_cffi" | "browser".
scraper_fetch_mode: str
# HTTP-эндпоинт headless-браузера (tradein-browser).
browser_http_endpoint: str
# Backconnect-прокси общий (Avito/Cian fallback).
scraper_proxy_url: str | None
# Avito rotate-прокси + лимит ротаций на прогон.
avito_proxy_rotate_url: str | None
avito_proxy_max_rotations: int
# Ограничивать Avito SERP только ЕКБ.
avito_serp_ekb_only: bool
# Yandex rotate-прокси.
yandex_proxy_rotate_url: str | None
# Cian-прокси (CIAN_PROXY_URL > scraper_proxy_url fallback).
cian_proxy_url: str | None
# Границы валидности cian-оценки, руб.
cian_valuation_min_rub: float
cian_valuation_max_rub: float
# Sentry/GlitchTip DSN (опционально).
glitchtip_dsn: str | None
@runtime_checkable
class SessionFactory(Protocol):
"""Фабрика SQLAlchemy-сессии (аналог `app.core.db.SessionLocal`).
Вызов без аргументов возвращает новую `Session`. Реализуется
`app.services.scraper_adapters.RealSessionFactory`; сам `sessionmaker`
(SessionLocal) также структурно удовлетворяет.
"""
def __call__(self) -> Session:
"""Открыть новую сессию БД."""
...
@runtime_checkable
class EnrichmentJobs(Protocol):
"""Реестр enrichment-джоб, дёргаемых после сохранения скрапом.
Callback'и проксируют в соответствующие `app.services.*`. Возвращаемые
доменные результат-типы (`HouseIMVBackfillResult`, `YandexAddressBackfillResult`)
остаются `Any`, чтобы не тянуть `app.*` в контракты. Реализуется
`app.services.scraper_adapters.RealEnrichmentJobs`.
"""
async def house_imv_backfill(
self,
db: Session,
*,
batch_size: int = ...,
request_delay_sec: float = ...,
only_status: str = ...,
house_id: int | None = ...,
heartbeat: Callable[[], None] | None = ...,
) -> Any:
"""Avito IMV-оценка для домов в scope. Проксирует backfill_house_imv."""
...
def house_dedup_merge(
self,
db: Session,
*,
run_id: int,
params: dict[str, Any],
) -> dict[str, int]:
"""Слить дубли домов. Проксирует run_house_dedup_merge. Returns счётчики."""
...
async def yandex_address_backfill(
self,
db: Session,
*,
limit: int = ...,
request_delay_sec: float = ...,
) -> Any:
"""Догрузить точные адреса Yandex. Проксирует backfill_yandex_addresses."""
...
__all__ = [
"EnrichmentJobs",
"HouseMatcher",
"ScraperConfig",
"SessionFactory",
]

View file

@ -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, а не блоком.
"""

View file

@ -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)

View file

@ -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

View file

@ -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,
)

View file

@ -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 <script type='application/ld+json'> blocks from HTML.
Returns a list of parsed JSON-LD dicts. Skips blocks that fail to parse.
"""
tree = HTMLParser(html)
results: list[dict[str, Any]] = []
for script in tree.css('script[type="application/ld+json"]'):
text = script.text() or ""
if not text.strip():
continue
try:
parsed = json.loads(text)
except json.JSONDecodeError:
continue
if isinstance(parsed, dict):
results.append(parsed)
elif isinstance(parsed, list):
# Some pages wrap multiple LDs in a JSON array
results.extend(item for item in parsed if isinstance(item, dict))
return results
def find_ld_by_type(html: str, type_name: str) -> dict[str, Any] | None:
"""Return the first JSON-LD dict whose @type equals type_name.
If @type is a list, returns the dict if type_name is contained in it.
"""
for ld in extract_json_ld(html):
t = ld.get("@type")
if t == type_name:
return ld
if isinstance(t, list) and type_name in t:
return ld
return None
# ---------------------------------------------------------------------------
# Section 2 — Regex extractors (module-level compiled)
# ---------------------------------------------------------------------------
# URL pattern extractors
RE_OFFER_ID = re.compile(r"/offer/(\d+)/?")
RE_JK_ID = re.compile(r"/novostrojka/([\w-]+?)-(\d+)/?") # group(1)=slug, group(2)=id
RE_STREET_ID = re.compile(r"/kupit/kvartira/st-([\w-]+?)-(\d+)/?")
# Card / title text extractors
RE_TITLE_AREA = re.compile(r"(\d+[.,]?\d*)\s*м²")
RE_TITLE_ROOMS = re.compile(r"(\d+)\s*-?\s*комнатн|(студи[яюй])", re.IGNORECASE)
RE_FLOOR = re.compile(r"(\d+)\s+этаж\s+из\s+(\d+)", re.IGNORECASE)
RE_PRICE = re.compile(r"(\d[\d\s]+\d)\s*₽")
RE_PPM2 = re.compile(r"(\d[\d\s]+)\s*₽\s*за\s*м²", re.IGNORECASE)
RE_VIEWS = re.compile(r"(\d+)\s+просмотр", re.IGNORECASE)
# Detail / NLP extractors
RE_METRO_WALK = re.compile(
r"(\d+)\s+минут\s+(?:неспешной\s+прогулки|ходьбы|пешком)", re.IGNORECASE
)
RE_YEAR = re.compile(r"\b(19\d{2}|20[0-2]\d)\b")
# Agency block ("Год основания 1995", "150 объектов")
RE_AGENCY_FOUNDED = re.compile(r"Год\s+основания\s+(\d{4})", re.IGNORECASE)
RE_AGENCY_OBJECTS = re.compile(r"(\d+)\s+объект", re.IGNORECASE)
# ---------------------------------------------------------------------------
# Section 3 — Russian date parsing
# ---------------------------------------------------------------------------
RU_MONTHS: dict[str, int] = {
"января": 1, "февраля": 2, "марта": 3, "апреля": 4, "мая": 5, "июня": 6,
"июля": 7, "августа": 8, "сентября": 9, "октября": 10, "ноября": 11, "декабря": 12,
}
RU_MONTH_NOMINATIVE: dict[str, int] = {
"январь": 1, "февраль": 2, "март": 3, "апрель": 4, "май": 5, "июнь": 6,
"июль": 7, "август": 8, "сентябрь": 9, "октябрь": 10, "ноябрь": 11, "декабрь": 12,
}
RE_RU_DATE = re.compile(r"(\d{1,2})\s+(\w+)\s+(\d{4})")
RE_DMY = re.compile(r"(\d{2})\.(\d{2})\.(\d{4})")
def parse_ru_date(text: str | None) -> date | None:
"""Parse Russian date string like '9 мая 2026' → date(2026, 5, 9). Genitive month form only."""
if not text:
return None
m = RE_RU_DATE.search(text)
if not m:
return None
day_s, month_ru, year_s = m.groups()
month = RU_MONTHS.get(month_ru.lower())
if not month:
return None
try:
return date(int(year_s), month, int(day_s))
except (ValueError, TypeError):
return None
def parse_dmy(text: str | None) -> date | None:
"""Parse DD.MM.YYYY format (used by Yandex Valuation tool history) → date."""
if not text:
return None
m = RE_DMY.search(text)
if not m:
return None
try:
return date(int(m.group(3)), int(m.group(2)), int(m.group(1)))
except (ValueError, TypeError):
return None
# Relative date parsing (Yandex SERP shows "N дней назад" / "N часов назад")
_RE_REL_DATE = re.compile(
r"(?P<n>\d+)\s+(?P<unit>"
r"секунд[ауы]?|"
r"минут[ауы]?|"
r"час(?:а|ов)?|"
r"день|дн(?:я|ей?)|"
r"недел[июяь]+|"
r"месяц(?:а|ев)?|"
r"год(?:а|ов)?"
r")\s+назад",
flags=re.I,
)
_REL_UNIT_DAYS: dict[str, int] = {
"секунд": 0, "минут": 0, "час": 0,
"день": 1, "дн": 1,
"недел": 7,
"месяц": 30,
"год": 365,
}
def _parse_relative_date_yandex(s: str | None) -> date | None:
"""Parse 'N дней/недель/месяцев назад' → date. Returns None if not matched."""
if not s:
return None
m = _RE_REL_DATE.search(s)
if not m:
return None
n = int(m["n"])
unit_raw = m["unit"].lower()
for prefix, days in _REL_UNIT_DAYS.items():
if unit_raw.startswith(prefix):
return date.today() - timedelta(days=n * days)
return None
RE_RU_DATE_NO_YEAR = re.compile(
r"(\d{1,2})\s+(январ\w*|феврал\w*|март\w*|апрел\w*|ма[яй]|"
r"июн\w*|июл\w*|август\w*|сентябр\w*|октябр\w*|ноябр\w*|декабр\w*)"
r"(?!\s+\d{4})",
flags=re.I,
)
_RU_MONTH_PREFIXES: dict[str, int] = {
"январ": 1, "феврал": 2, "март": 3, "апрел": 4,
"ма": 5, "июн": 6, "июл": 7, "август": 8,
"сентябр": 9, "октябр": 10, "ноябр": 11, "декабр": 12,
}
def parse_listing_date(text: str | None) -> date | None:
"""Parse listing date from Yandex card text.
Tries in order:
1. Keyword shortcut: 'сегодня' / 'вчера' / 'позавчера' relative day
2. Absolute Russian date with year: '9 мая 2026' date(2026, 5, 9)
3. Absolute without year: '18 мая' current year (rollback to prev
year if month >30 дней в будущем)
4. Relative date: 'N дней назад' date.today() - N days
Returns None if none match.
"""
if not text:
return None
text_lower = text.lower()
# 1. Keyword shortcuts (common on Yandex SERP for recent cards).
if "позавчера" in text_lower:
return date.today() - timedelta(days=2)
if "вчера" in text_lower:
return date.today() - timedelta(days=1)
if "сегодня" in text_lower or "только что" in text_lower:
return date.today()
# 2. Absolute with year (existing).
result = parse_ru_date(text)
if result is not None:
return result
# 3. Absolute without year — infer current, rollback to prev if future.
m = RE_RU_DATE_NO_YEAR.search(text)
if m:
day = int(m.group(1))
month_raw = m.group(2).lower()
month: int | None = None
for prefix, num in _RU_MONTH_PREFIXES.items():
if month_raw.startswith(prefix):
month = num
break
if month is not None and 1 <= day <= 31:
today = date.today()
try:
parsed = date(today.year, month, day)
except ValueError:
parsed = None
if parsed and (parsed - today).days > 30:
try:
parsed = date(today.year - 1, month, day)
except ValueError:
parsed = None
if parsed:
return parsed
# 4. Relative numeric.
return _parse_relative_date_yandex(text)
# ---------------------------------------------------------------------------
# Section 4 — House type / class NLP from description text
# ---------------------------------------------------------------------------
def parse_house_type(text: str | None) -> str | None:
"""Extract house material type from free-form description.
Returns: 'panel' | 'monolith_brick' | 'monolith' | 'brick' | 'block' | 'wood' | None.
Order matters: 'monolith_brick' is checked before 'monolith' and 'brick' alone.
"""
if not text:
return None
t = text.lower()
if "панельн" in t:
return "panel"
if "монолит" in t and "кирпич" in t:
return "monolith_brick"
if "монолит" in t:
return "monolith"
if "кирпичн" in t:
return "brick"
if "блочн" in t:
return "block"
if "деревянн" in t:
return "wood"
return None
def parse_house_class(text: str | None) -> str | None:
"""Extract ЖК class from description.
Returns: 'elite' | 'premium' | 'business' | 'comfort_plus' | 'comfort' | 'economy' | None.
"""
if not text:
return None
t = text.lower()
if re.search(r"класса?\s+элит|элитный\s+класс", t):
return "elite"
if re.search(r"класса?\s+премиум|премиум.{0,5}класс", t):
return "premium"
if re.search(r"класса?\s+бизнес", t):
return "business"
if re.search(r"класса?\s+комфорт\s*\+", t):
return "comfort_plus"
if re.search(r"класса?\s+комфорт(?!\s*\+)", t):
return "comfort"
if re.search(r"класса?\s+эконом", t):
return "economy"
return None
# ---------------------------------------------------------------------------
# Section 5 — Money parsing
# ---------------------------------------------------------------------------
RE_RUB_MLN = re.compile(r"([\d,.]+)\s*млн\s*₽", re.IGNORECASE)
RE_RUB_RAW = re.compile(r"(\d[\d\s]*\d|\d)")
def parse_rub(text: str | None) -> int | None:
"""Parse Russian price string → integer rubles.
Supports:
'4 399 000 ₽' 4_399_000
'4,4 млн ₽' 4_400_000
'4.4 млн ₽' 4_400_000
'117 500' 117_500 (raw digits without currency, e.g. ppm2)
None / 'без цены' / '' None
"""
if not text:
return None
text = text.strip()
if not text:
return None
m = RE_RUB_MLN.search(text)
if m:
try:
# `\s` strips also NBSP (\xa0), thin space, и т.д. — Yandex SERP
# форматирует «9\xa0800\xa0000» через NBSP, не regular space.
cleaned = re.sub(r"\s", "", m.group(1)).replace(",", ".")
return int(float(cleaned) * 1_000_000)
except (ValueError, TypeError):
pass
m = RE_RUB_RAW.search(text)
if m:
# Same NBSP issue: `.replace(" ", "")` оставляет \xa0 → int() raises.
cleaned = re.sub(r"\s", "", m.group(1))
try:
return int(cleaned)
except ValueError:
return None
return None

18
tradein-mvp/uv.lock generated
View file

@ -1113,6 +1113,24 @@ wheels = [
name = "scraper-kit"
version = "0.1.0"
source = { editable = "packages/scraper-kit" }
dependencies = [
{ name = "httpx", extra = ["socks"] },
{ name = "psycopg", extra = ["binary"] },
{ name = "pydantic" },
{ name = "selectolax" },
{ name = "sqlalchemy" },
{ name = "tenacity" },
]
[package.metadata]
requires-dist = [
{ name = "httpx", extras = ["socks"], specifier = ">=0.27.0" },
{ name = "psycopg", extras = ["binary"], specifier = ">=3.2.0" },
{ name = "pydantic", specifier = ">=2.7.0" },
{ name = "selectolax", specifier = ">=0.3.0" },
{ name = "sqlalchemy", specifier = ">=2.0.30" },
{ name = "tenacity", specifier = ">=9.0.0" },
]
[[package]]
name = "segno"