lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions
gendesign/tradein-mvp/backend/tests/test_cian_serp_scraper.py
bot-backend 1d5b081cb4
Some checks failed
Deploy Trade-In / changes (push) Successful in 13s
Details
Deploy Trade-In / build-frontend (push) Has been skipped
Details
Deploy Trade-In / build-browser (push) Has been skipped
Details
Deploy Trade-In / test (push) Successful in 1m28s
Details
Deploy Trade-In / deploy (push) Failing after 46s
Details
Deploy Trade-In / build-backend (push) Successful in 50s
Details
fix(cian): recover new-build house via cadastral anchor + listings (source,source_id) dedup (#1773) (#1912) 
cian new-build cadastral anchor + listings (source,source_id) dedup (#1773): migration 133 (deep-reviewed data-loss-safe). Refs #1773
2026-06-26 07:45:22 +00:00

729 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Smoke тесты для cian.py SERP scraper (Stage 3 рефактор).
Проверяют парсинг 137 полей из синтетического Redux state без сетевых запросов.
Fixture формирует минимальный JS push() conforming to _cianConfig pattern.
"""
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from app.services.scrapers.cian import CianScraper, _format_address
# ── Fixture helpers ──────────────────────────────────────────────────────────
def _make_cian_serp_html(offers: list[dict]) -> str:
"""Обернуть offers в валидный _cianConfig['frontend-serp'].push() HTML."""
state = {
"results": {
"totalOffers": len(offers),
"offers": offers,
}
}
state_json = json.dumps(state, ensure_ascii=False)
return f"""<html><body>
<script>
window._cianConfig['frontend-serp'] = window._cianConfig['frontend-serp'] || [];
window._cianConfig['frontend-serp'].push({{
key: 'initialState',
value: {state_json},
priority: 10,
filter: null
}});
</script>
</body></html>"""
def _make_full_offer(
cian_id: int = 328739308,
price_rur: int = 6500000,
rooms: int = 2,
total_area: str = "52.0",
living_area: str = "35.0",
kitchen_area: str = "11.5",
floor: int = 7,
floors_count: int = 16,
year_built: int = 2010,
material_type: str = "panel",
lat: float = 56.8385,
lng: float = 60.5940,
cadastral_number: str = "66:41:0204016:1234",
building_cadastral_number: str = "66:41:0204016:100",
description: str = "Продаётся отличная квартира",
description_minhash: str | list[int] = "abc123def456abc1",
newbuilding_id: int = 0,
has_phones: bool = True,
is_homeowner: bool = True,
is_pro: bool = False,
bargain_allowed: bool = True,
sale_type: str = "free",
) -> dict:
"""Синтетический Cian offer с полным набором полей (137 полей subset)."""
offer: dict = {
"cianId": cian_id,
"id": cian_id,
"fullUrl": f"https://ekb.cian.ru/sale/flat/{cian_id}/",
"category": "flatSale",
"dealType": "sale",
"offerType": "flat",
"flatType": "rooms",
"isApartments": False,
"roomsCount": rooms,
"status": "published",
# Area
"totalArea": total_area,
"livingArea": living_area,
"kitchenArea": kitchen_area,
"balconiesCount": 1,
"loggiasCount": 0,
"bedroomsCount": None,
"hasFurniture": False,
"isSoldFurnished": False,
"decoration": None,
# Floor / building
"floorNumber": floor,
"building": {
"floorsCount": floors_count,
"materialType": material_type,
"buildYear": year_built,
"deadline": None,
"parking": None,
},
# Cadastral
"cadastralNumber": cadastral_number,
"buildingCadastralNumber": building_cadastral_number,
# Price
"bargainTerms": {
"priceRur": price_rur,
"price": price_rur,
"currency": "rur",
"saleType": sale_type,
"mortgageAllowed": True,
"bargainAllowed": bargain_allowed,
"priceType": "all",
},
"formattedShortPrice": f"{price_rur // 1000000} млн ₽",
# Geo
"geo": {
"coordinates": {"lat": lat, "lng": lng},
"address": [
{"type": "location", "name": "Свердловская область"},
{"type": "city", "name": "Екатеринбург"},
{"type": "street", "name": "улица Постовского"},
{"type": "house", "name": "17А"},
],
"undergrounds": [
{
"name": "Геологическая",
"time": 10,
"transportType": "walk",
"lineId": 10,
"lineColor": "FF0000",
"isDefault": True,
"id": 100,
}
],
"districts": [],
},
# Newbuilding
"newbuilding": {"id": newbuilding_id, "name": ""},
"buildersIds": [],
# Seller
"phones": [{"number": "9012345678", "countryCode": "7"}] if has_phones else [],
"userId": 13332612,
"cianUserId": 13332612,
"publishedUserId": 13332612,
"isByHomeowner": is_homeowner,
"isByCommercialOwner": False,
"isCianPartner": False,
"isPro": is_pro,
"isCalltrackingEnabled": False,
# Description
"description": description,
"descriptionMinhash": description_minhash,
"isDuplicatedDescription": False,
# Time
"addedTimestamp": 1716451200, # 2024-05-23 00:00:00 UTC
"added": "вчера",
# Photos
"photos": [
{
"id": 1001,
"fullUrl": f"https://images.cdn-cian.ru/images/{cian_id}-1.jpg",
"thumbnailUrl": f"https://images.cdn-cian.ru/images/{cian_id}-2.jpg",
}
],
# Verification
"isRosreestrChecked": True,
"isLayoutApproved": None,
# Misc
"isNew": False,
"isTop3": False,
"isPremium": False,
}
return offer
# ── Tests ────────────────────────────────────────────────────────────────────
def test_cian_serp_parses_full_offer():
"""Базовый тест: полный offer → ScrapedLot с ключевыми полями."""
offer = _make_full_offer()
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
assert len(lots) == 1
lot = lots[0]
assert lot.source == "cian"
assert lot.source_id == "328739308"
assert lot.source_url == "https://ekb.cian.ru/sale/flat/328739308/"
def test_cian_serp_coords_no_jitter():
"""Координаты берутся точно из state — NO anchor jitter."""
offer = _make_full_offer(lat=56.8385, lng=60.5940)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
assert len(lots) == 1
lot = lots[0]
assert lot.lat == pytest.approx(56.8385, abs=1e-9)
assert lot.lon == pytest.approx(60.5940, abs=1e-9)
def test_cian_serp_kitchen_area_in_raw_payload():
"""kitchen_area_m2 не в ScrapedLot, но сохраняется в raw_payload."""
offer = _make_full_offer(kitchen_area="11.5")
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.raw_payload is not None
assert lot.raw_payload["kitchen_area_m2"] == "11.5"
def test_cian_serp_description_minhash():
"""description_minhash берётся из Cian state, если есть."""
offer = _make_full_offer(
description_minhash="cian_minhash_abc123",
description="Тест описания",
)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.description_minhash == "cian_minhash_abc123"
def test_cian_serp_description_minhash_fallback():
"""Если Cian не дал minhash — вычисляем SHA1 из description."""
offer = _make_full_offer(description_minhash="", description="Уникальное описание квартиры")
offer["descriptionMinhash"] = "" # явно пусто
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.description_minhash is not None
assert len(lot.description_minhash) == 32 # SHA1 hex [:32]
def test_cian_serp_description_minhash_list_int():
"""descriptionMinhash как list[int] (новый формат Cian ~2026-05-23) → строка через запятую.
Порядок элементов сохраняется как-есть — sorted() НЕ применяется,
т.к. порядок LSH-band'ов значим для dedup/cross-source matching.
Offer должен быть принят (не отброшен) — возвращает ScrapedLot, не None.
"""
minhash_list = [8439545, 5781654, 555882, 407717]
offer = _make_full_offer(
description_minhash=minhash_list, # type: ignore[arg-type]
description="Продаётся квартира с видом на парк",
)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
assert len(lots) == 1, "offer с list[int] minhash должен быть принят, не отброшен"
lot = lots[0]
assert lot.description_minhash == "8439545,5781654,555882,407717"
def test_cian_serp_phones():
"""phones — список dicts из state."""
offer = _make_full_offer(has_phones=True)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert isinstance(lot.phones, list)
assert len(lot.phones) == 1
assert lot.phones[0]["countryCode"] == "7"
assert lot.phones[0]["number"] == "9012345678"
def test_cian_serp_house_ext_id_newbuilding():
"""Новостройка с кадастром дома → anchor = building cad (#1773), не nb_id.
_make_full_offer несёт building_cadastral_number, поэтому same-building anchor
предпочитает кадастр дома (надёжнее, чем nb_id, который группирует весь ЖК)."""
offer = _make_full_offer(newbuilding_id=54016)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.house_source == "cian_building_cad"
assert lot.house_ext_id == "66:41:0204016:100"
assert lot.listing_segment == "novostroyki"
def test_cian_serp_house_ext_id_newbuilding_nb_id_fallback():
"""Без кадастра дома anchor падает на nb_id (ЖК) — back-compat fallback."""
offer = _make_full_offer(newbuilding_id=54016, building_cadastral_number="")
offer["buildingCadastralNumber"] = None
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.house_source == "cian_newbuilding"
assert lot.house_ext_id == "54016"
assert lot.listing_segment == "novostroyki"
def test_cian_serp_vtorichka_no_house_ext_id():
"""Вторичка (newbuilding.id == 0) — house_ext_id None."""
offer = _make_full_offer(newbuilding_id=0)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.house_source == "cian"
assert lot.house_ext_id is None
assert lot.listing_segment == "vtorichka"
def test_cian_serp_metro_stations():
"""metro_stations — список dicts с name/time/mode/line_color."""
offer = _make_full_offer()
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert isinstance(lot.metro_stations, list)
assert len(lot.metro_stations) == 1
station = lot.metro_stations[0]
assert station["name"] == "Геологическая"
assert station["time"] == 10
assert station["mode"] == "walk"
assert station["line_color"] == "FF0000"
def test_cian_serp_cadastral_numbers():
"""Кадастровые номера квартиры и здания."""
offer = _make_full_offer(
cadastral_number="66:41:0204016:1234",
building_cadastral_number="66:41:0204016:100",
)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.cadastral_number == "66:41:0204016:1234"
assert lot.building_cadastral_number == "66:41:0204016:100"
# kadastr_num (base field) тоже заполнен
assert lot.kadastr_num == "66:41:0204016:1234"
def test_cian_serp_is_homeowner_seller():
"""is_homeowner и is_pro_seller из offer."""
offer = _make_full_offer(is_homeowner=True, is_pro=False)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.is_homeowner is True
assert lot.is_pro_seller is False
def test_cian_serp_bargain_and_sale_type():
"""bargain_allowed и sale_type из bargainTerms."""
offer = _make_full_offer(bargain_allowed=True, sale_type="alternative")
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.bargain_allowed is True
assert lot.sale_type == "alternative"
def test_cian_serp_building_fields():
"""Поля здания: year_built, house_type, total_floors."""
offer = _make_full_offer(year_built=1988, material_type="brick", floors_count=9)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.year_built == 1988
assert lot.house_type == "brick"
assert lot.total_floors == 9
def test_cian_serp_living_area():
"""living_area_m2 из offer.livingArea."""
offer = _make_full_offer(living_area="35.0")
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.living_area_m2 == pytest.approx(35.0)
def test_cian_serp_state_missing_returns_empty():
"""Если state нет в HTML — возвращаем пустой список."""
html = "<html><body><p>no cian state here</p></body></html>"
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
assert lots == []
def test_cian_serp_skips_offer_without_price():
"""Offer без цены — пропускаем."""
offer = _make_full_offer()
del offer["bargainTerms"]["priceRur"]
del offer["bargainTerms"]["price"]
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
assert lots == []
def test_cian_serp_multiple_offers():
"""Несколько offers → список ScrapedLot."""
offers = [_make_full_offer(cian_id=100 + i) for i in range(5)]
html = _make_cian_serp_html(offers)
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
assert len(lots) == 5
ids = [lot.source_id for lot in lots]
assert sorted(ids) == ["100", "101", "102", "103", "104"]
def test_cian_serp_photo_urls():
"""photo_urls — полные URL из offer.photos[].fullUrl."""
offer = _make_full_offer(cian_id=999)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert len(lot.photo_urls) == 1
assert "cdn-cian.ru" in lot.photo_urls[0]
assert "999-1.jpg" in lot.photo_urls[0]
def test_cian_serp_no_coords_does_not_crash():
"""Если координаты отсутствуют — lot.lat/lon = None, не краш."""
offer = _make_full_offer()
del offer["geo"]["coordinates"]
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
assert len(lots) == 1
assert lots[0].lat is None
assert lots[0].lon is None
# ── Transport: BrowserFetcher (camoufox) миграция (#1806) ────────────────────
@pytest.mark.asyncio
async def test_cian_aenter_opens_browser_fetcher() -> None:
"""__aenter__ открывает ОДНУ BrowserFetcher(source="cian") сессию (camoufox)."""
mock_browser = MagicMock()
mock_browser.__aenter__ = AsyncMock(return_value=mock_browser)
mock_browser.__aexit__ = AsyncMock(return_value=None)
captured: dict = {}
def _fake_browser(*args, **kwargs):
captured["source"] = kwargs.get("source") if kwargs else (args[0] if args else None)
return mock_browser
scraper = CianScraper()
with patch("app.services.scrapers.cian.BrowserFetcher", _fake_browser):
await scraper.__aenter__()
mock_browser.__aenter__.assert_awaited_once()
assert scraper._browser is mock_browser
assert captured["source"] == "cian"
await scraper.__aexit__(None, None, None)
mock_browser.__aexit__.assert_awaited_once()
assert scraper._browser is None
@pytest.mark.asyncio
async def test_cian_fetch_page_html_uses_browser_fetcher() -> None:
"""_fetch_page_html тянет HTML через BrowserFetcher.fetch, не через curl_cffi."""
scraper = CianScraper.__new__(CianScraper)
scraper.name = "cian"
scraper.base_url = "https://ekb.cian.ru"
scraper.request_delay_sec = 0.0
fetched_urls: list[str] = []
async def fake_fetch(url: str) -> str:
fetched_urls.append(url)
return "<html>serp-page</html>"
scraper._browser = MagicMock()
scraper._browser.fetch = AsyncMock(side_effect=fake_fetch)
html = await scraper._fetch_page_html((2,), 1, None, None)
assert html == "<html>serp-page</html>"
assert len(fetched_urls) == 1
assert "cat.php" in fetched_urls[0]
@pytest.mark.asyncio
async def test_cian_fetch_page_html_returns_none_on_browser_error() -> None:
"""Если BrowserFetcher.fetch падает — _fetch_page_html возвращает None (не краш)."""
scraper = CianScraper.__new__(CianScraper)
scraper.name = "cian"
scraper.base_url = "https://ekb.cian.ru"
scraper.request_delay_sec = 0.0
scraper._browser = MagicMock()
scraper._browser.fetch = AsyncMock(side_effect=RuntimeError("browser down"))
html = await scraper._fetch_page_html((1,), 2, 1_000_000, 5_000_000)
assert html is None
@pytest.mark.asyncio
async def test_cian_fetch_around_parses_browser_html() -> None:
"""fetch_around парсит лоты из SSR-HTML, отданного BrowserFetcher."""
offer = _make_full_offer(cian_id=777)
serp_html = _make_cian_serp_html([offer])
scraper = CianScraper.__new__(CianScraper)
scraper.name = "cian"
scraper.base_url = "https://ekb.cian.ru"
scraper.request_delay_sec = 0.0
scraper._browser = MagicMock()
scraper._browser.fetch = AsyncMock(return_value=serp_html)
with patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock):
lots = await scraper.fetch_around(56.84, 60.60, rooms=(2,), page=1)
assert len(lots) == 1
assert lots[0].source_id == "777"
scraper._browser.fetch.assert_awaited_once()
@pytest.mark.asyncio
async def test_cian_rotate_ip_is_noop() -> None:
"""_rotate_ip после миграции — no-op, всегда False (прокси ротирует IP сам)."""
scraper = CianScraper.__new__(CianScraper)
result = await scraper._rotate_ip()
assert result is False
# ── Address formatter tests ──────────────────────────────────────────────────
def test_format_address_standard():
"""Стандартный geo.address[]."""
parts = [
{"type": "location", "name": "Свердловская область"},
{"type": "city", "name": "Екатеринбург"},
{"type": "district", "name": "Чкаловский"},
{"type": "street", "name": "улица Постовского"},
{"type": "house", "name": "17А"},
{"type": "metro", "name": "Геологическая"},
]
result = _format_address(parts)
# location и metro пропускаем
assert "Свердловская" not in result
assert "Геологическая" not in result
assert "Екатеринбург" in result
assert "17А" in result
def test_format_address_empty():
"""Пустой address[] → fallback string."""
assert _format_address([]) == "Екатеринбург (Cian)"
# ── #1773: новостройка без part{type:"house"} → восстановление номера дома ─────
def _make_newbuilding_houseless_offer(
cian_id: int = 330047129,
newbuilding_id: int = 54016,
newbuilding_name: str = "Суходольский квартал",
building_cadastral_number: str | None = "66:41:0313010:11396",
building_house_number: str | None = None,
) -> dict:
"""Cian-оффер новостройки: geo.address[] БЕЗ {type:"house"} (ЖК Суходольский,
ул. Евгения Савкова 29, cian 330047129 — реальный кейс из #1773)."""
offer = _make_full_offer(
cian_id=cian_id,
newbuilding_id=newbuilding_id,
building_cadastral_number=building_cadastral_number or "",
)
# geo.address[] без части house — корень бага: «…улица Евгения Савкова» без «29»
offer["geo"]["address"] = [
{"type": "location", "name": "Свердловская область"},
{"type": "city", "name": "Екатеринбург"},
{"type": "district", "name": "Академический"},
{"type": "street", "name": "улица Евгения Савкова"},
]
offer["newbuilding"] = {"id": newbuilding_id, "name": newbuilding_name}
building = offer.setdefault("building", {})
if building_house_number is not None:
building["houseNumber"] = building_house_number
if building_cadastral_number is None:
offer["buildingCadastralNumber"] = None
return offer
def test_cian_newbuilding_cad_number_anchor():
"""Новостройка с кадастром дома → house_source/house_ext_id = building cad
(same-building anchor #1773), адрес дополнен кадастром."""
offer = _make_newbuilding_houseless_offer(
building_cadastral_number="66:41:0313010:11396",
)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
assert len(lots) == 1
lot = lots[0]
# Кадастр дома предпочтён как anchor-ключ вместо nb_id (ЖК целиком)
assert lot.house_source == "cian_building_cad"
assert lot.house_ext_id == "66:41:0313010:11396"
assert lot.building_cadastral_number == "66:41:0313010:11396"
assert lot.listing_segment == "novostroyki"
# geo.address[] был без house → кадастр дописан в адрес как якорь
assert "Евгения Савкова" in (lot.address or "")
assert "66:41:0313010:11396" in (lot.address or "")
def test_cian_newbuilding_recovers_house_from_building_number():
"""building.houseNumber имеет приоритет (a): «29» восстановлен в адрес,
кадастр остаётся anchor-ключом."""
offer = _make_newbuilding_houseless_offer(
building_cadastral_number="66:41:0313010:11396",
building_house_number="29",
)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
addr = lot.address or ""
assert "Евгения Савкова" in addr
assert "29" in addr # номер дома восстановлен из building.houseNumber
# анкор по-прежнему кадастр дома (надёжнее nb_id)
assert lot.house_ext_id == "66:41:0313010:11396"
def test_cian_newbuilding_recovers_house_from_name_when_no_cad():
"""Без кадастра дома: дом-токен «29» вытягивается из названия ЖК (приоритет c),
anchor падает на nb_id."""
offer = _make_newbuilding_houseless_offer(
newbuilding_name="ЖК на Савкова 29",
building_cadastral_number=None,
)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
assert lot.building_cadastral_number is None
# нет кадастра → anchor = nb_id (ЖК)
assert lot.house_source == "cian_newbuilding"
assert lot.house_ext_id == "54016"
addr = lot.address or ""
assert "29" in addr # дом восстановлен из названия ЖК
def test_cian_newbuilding_no_house_no_fabrication():
"""Ни building, ни кадастра, ни номера в названии → номер НЕ выдумываем."""
offer = _make_newbuilding_houseless_offer(
newbuilding_name="Суходольский квартал", # без числа-хвоста
building_cadastral_number=None,
)
html = _make_cian_serp_html([offer])
scraper = CianScraper()
lots = scraper._parse_serp_html(html)
lot = lots[0]
addr = lot.address or ""
assert "Евгения Савкова" in addr
# никаких фейковых catch-all номеров (например 7019) в адресе
assert not addr.rstrip().endswith("7019")
def test_format_address_vtorichka_houseless_untouched():
"""Вторичка с house-less адресом НЕ трогается — номер не выдумываем."""
parts = [
{"type": "city", "name": "Екатеринбург"},
{"type": "street", "name": "улица Евгения Савкова"},
]
result = _format_address(
parts,
building={"houseNumber": "29"},
building_cadastral_number="66:41:0313010:11396",
newbuilding={"id": 0, "name": ""},
is_newbuilding=False,
)
# вторичка: is_newbuilding=False → ни «29», ни кадастр не дописываются
assert "29" not in result
assert "66:41" not in result
Reference in a new issue View git blame Copy permalink