lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

refactor(tradein): avito search — remove jitter (C-5) + mobile API dead code + parse houseLinks #441

Merged
lekss361 merged 1 commit from feat/tradein-avito-search-refactor into main 2026-05-23 11:55:19 +00:00
Conversation 1 Commits 1 Files changed 2 +57 -232
2 changed files with 57 additions and 232 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

277
tradein-mvp/backend/app/services/scrapers/avito.py
View file

@ -1,7 +1,6 @@
"""Avito.ru scraper — парсинг вторички вокруг точки.
Стратегия: HTML scrape карточек объявлений + извлечение JSON `__initialState__`
который Avito вставляет в `<script>` тег.
Стратегия: HTML scrape карточек объявлений через DOM.
URL pattern (EKB):
https://www.avito.ru/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ
@ -16,11 +15,8 @@ URL pattern (EKB):
from __future__ import annotations
import json
import logging
import random
import re
from datetime import date, datetime
from typing import Any
from urllib.parse import urlencode, urljoin
@ -47,7 +43,7 @@ class AvitoScraper(BaseScraper):
super().__init__()
self._cffi: AsyncSession | None = None
async def __aenter__(self) -> "AvitoScraper":
async def __aenter__(self) -> AvitoScraper:
await super().__aenter__()
self._cffi = AsyncSession(
impersonate="chrome120",
@ -78,23 +74,13 @@ class AvitoScraper(BaseScraper):
Avito работает в км — конвертируем. Минимум 1 км.
Возвращаем макс 50 объявлений (страница 1).
Coords НЕ заполняем из anchor (избегаем silent corruption через jitter).
Точные lat/lon приходят позже из avito_detail.py — для search-карточек
lat=lon=None, далее geocode-missing cron подтянет из address.
"""
radius_km = max(1, round(radius_m / 1000))
# Сохраняем центр для fallback заполнения coords (Avito HTML state не даёт coords)
self._anchor_lat = lat
self._anchor_lon = lon
self._anchor_radius_m = radius_m
# Strategy A: попробовать через mobile API (проще, меньше защиты)
try:
lots = await self._fetch_via_mobile_api(lat, lon, radius_km)
if lots:
logger.info("avito mobile_api: %d lots around (%.4f, %.4f)", len(lots), lat, lon)
return lots
except Exception as e: # noqa: BLE001
logger.warning("avito mobile_api failed: %s — falling back to HTML scrape", e)
# Strategy B: HTML scrape (web) через curl_cffi
url = self._build_web_url(lat, lon, radius_km, page=1)
try:
assert self._cffi is not None
@ -104,7 +90,7 @@ class AvitoScraper(BaseScraper):
"avito HTML returned %d for %s", response.status_code, url
)
return []
except Exception: # noqa: BLE001
except Exception:
logger.exception("avito HTML fetch failed for %s", url)
return []
@ -113,101 +99,6 @@ class AvitoScraper(BaseScraper):
await self.sleep_between_requests()
return lots
# ── Strategy A: mobile API ──────────────────────────────────────────────
async def _fetch_via_mobile_api(
self, lat: float, lon: float, radius_km: int
) -> list[ScrapedLot]:
"""Avito mobile API endpoint: https://m.avito.ru/api/15/items
Возвращает JSON с полным structured data — проще, чем HTML scrape.
Может быть закрыт за auth — обрабатываем в except выше.
"""
params = {
"categoryId": 24, # квартиры
"locationId": 653240, # ЕКБ (Avito internal)
"params[201]": 1, # тип: продам (вторичка)
"geoCoords": f"{lat},{lon}",
"radius": radius_km,
"page": 1,
"limit": 50,
"sort": "date",
}
url = f"https://m.avito.ru/api/15/items?{urlencode(params)}"
assert self._cffi is not None
response = await self._cffi.get(url, headers={"Accept": "application/json"})
if response.status_code != 200:
return []
data = response.json()
items = data.get("result", {}).get("items", []) if isinstance(data, dict) else []
return [self._mobile_item_to_lot(item) for item in items if isinstance(item, dict)]
def _mobile_item_to_lot(self, item: dict[str, Any]) -> ScrapedLot | None:
"""Маппинг Avito mobile API item → ScrapedLot.
Avito mobile API meanwhile отдаёт title с метаданными квартиры, но
address/coords могут быть в разных полях — пробуем все.
"""
try:
title = item.get("title", "")
# price может быть int или dict {value: N}
price_field = item.get("price")
if isinstance(price_field, dict):
price = price_field.get("value")
else:
price = price_field
if not price:
return None
rooms = _extract_rooms_from_title(title)
area = _extract_area_from_title(title)
floor, total = _extract_floor_from_title(title)
# Address — пробуем все возможные пути
address = (
item.get("address")
or item.get("location", {}).get("name") if isinstance(item.get("location"), dict) else ""
)
if not address and isinstance(item.get("geo"), dict):
address = item["geo"].get("formatted_address") or item["geo"].get("address", "")
if not address:
# Avito mobile API часто отдаёт address в "location.references" или "geo.geoReferences"
geo = item.get("geo") or {}
if isinstance(geo.get("geoReferences"), list) and geo["geoReferences"]:
address = geo["geoReferences"][0].get("content", "")
# Coords — тоже пробуем все варианты
coords = item.get("coords") or item.get("geo", {}).get("coords") or {}
lat = coords.get("lat") if isinstance(coords, dict) else None
lon = coords.get("lng") or coords.get("lon") if isinstance(coords, dict) else None
# Photos
photos = []
for img in (item.get("images") or [])[:5]:
if isinstance(img, dict):
photos.append(img.get("1280x960") or img.get("636x476") or img.get("864x648", ""))
elif isinstance(img, str):
photos.append(img)
return ScrapedLot(
source="avito",
source_url=urljoin("https://www.avito.ru", item.get("uri", "")),
source_id=str(item.get("id")),
address=address or None,
lat=lat,
lon=lon,
rooms=rooms,
area_m2=area,
floor=floor,
total_floors=total,
price_rub=int(price),
photo_urls=[p for p in photos if p],
raw_payload=item, # сохраняем ВСЁ для retrofit и дебага
)
except Exception: # noqa: BLE001
logger.exception("avito mobile item parse failed: %s", item.get("id"))
return None
# ── Strategy B: HTML scrape ─────────────────────────────────────────────
def _build_web_url(self, lat: float, lon: float, radius_km: int, page: int = 1) -> str:
params = {
@ -221,23 +112,10 @@ class AvitoScraper(BaseScraper):
)
def _parse_html(self, html: str, source_url_base: str) -> list[ScrapedLot]:
"""Парсим карточки объявлений из HTML.
"""Парсим карточки объявлений из HTML через DOM scrape.
Avito embed-ит JSON в `<script>` теге с initial state — пробуем найти.
Если структура поменялась — fallback на DOM scrape карточек.
Avito state давно пустой — используем только DOM путь.
"""
# Попробуем сначала __initialState__ JSON
state = _extract_avito_state(html)
if state:
items = _items_from_state(state)
if items:
return [
lot
for lot in (self._state_item_to_lot(it) for it in items)
if lot is not None
]
# Fallback: DOM scrape
tree = HTMLParser(html)
cards = tree.css('[data-marker="item"]')
lots: list[ScrapedLot] = []
@ -247,56 +125,12 @@ class AvitoScraper(BaseScraper):
lots.append(lot)
return lots
def _state_item_to_lot(self, item: dict[str, Any]) -> ScrapedLot | None:
"""Парсинг из __initialState__ JSON. Avito отдаёт минимум полей в нём
(часто только title) — fallback на anchor координаты с jitter."""
try:
price = item.get("priceDetailed", {}).get("value") or item.get("price")
if not price:
return None
title = item.get("title", "")
rooms = _extract_rooms_from_title(title)
area = _extract_area_from_title(title)
floor, total = _extract_floor_from_title(title)
coords = item.get("coords") or {}
location = item.get("location", {}) if isinstance(item.get("location"), dict) else {}
lat = coords.get("lat") if isinstance(coords, dict) else None
lon = coords.get("lng") if isinstance(coords, dict) else None
# Fallback: если Avito state не даёт coords — используем anchor с jitter.
# Это приближение, но позволяет PostGIS ST_DWithin работать в правильном радиусе.
if (lat is None or lon is None) and hasattr(self, "_anchor_lat"):
# Jitter в радиусе scrape — chunk радиуса в координатах (1° lat ≈ 111км)
jitter_deg = (self._anchor_radius_m / 1000) / 111 # км → градусы
lat = self._anchor_lat + (random.random() - 0.5) * jitter_deg
lon = self._anchor_lon + (random.random() - 0.5) * jitter_deg / 2 # коррекция по широте
return ScrapedLot(
source="avito",
source_url=urljoin("https://www.avito.ru", item.get("urlPath", "")),
source_id=str(item.get("id")),
address=location.get("name") or item.get("address") or "Екатеринбург (Avito)",
lat=lat,
lon=lon,
rooms=rooms,
area_m2=area,
floor=floor,
total_floors=total,
price_rub=int(price),
photo_urls=[
img.get("1280x960") or img.get("636x476", "")
for img in item.get("images", [])[:5]
if isinstance(img, dict)
],
raw_payload=item,
)
except Exception: # noqa: BLE001
return None
def _dom_card_to_lot(self, card: Any, source_url_base: str) -> ScrapedLot | None:
"""Парсинг через DOM. Avito state давно пустой (см. _parse_html) — почти все
объявления приходят сюда, поэтому путь обязан тащить и адрес, и фото."""
"""Парсинг карточки объявления через DOM.
Avito state давно пустой — единственный путь. Coords из карточки не отдаются,
оставляем lat=lon=None (geocode-missing cron подтянет из address).
"""
try:
link_el = card.css_first('a[data-marker="item-title"]')
if link_el is None:
@ -314,9 +148,7 @@ class AvitoScraper(BaseScraper):
area = _extract_area_from_title(title)
floor, total = _extract_floor_from_title(title)
# Адрес: первый <p> внутри data-marker="item-location" — это улица + дом.
# Второй <p> — метро/район, его отбрасываем. Старый селектор
# [class*="geo-address"] мёртв: Avito перешёл на CSS-модули с хешами.
# Адрес: первый <p> внутри data-marker="item-location"
address: str | None = None
loc_el = card.css_first('[data-marker="item-location"]')
if loc_el is not None:
@ -324,7 +156,7 @@ class AvitoScraper(BaseScraper):
if street_el is not None:
address = street_el.text(strip=True) or None
# Фото: <img itemprop="image"> карточки-слайдера, реальный URL прямо в src.
# Фото
photo_urls: list[str] = []
for img in card.css('img[itemprop="image"]'):
src = img.attributes.get("src") or ""
@ -333,27 +165,46 @@ class AvitoScraper(BaseScraper):
if len(photo_urls) >= 5:
break
# Anchor jitter — Avito DOM не отдаёт точные coords, используем центр scrape ±jitter
lat = lon = None
if hasattr(self, "_anchor_lat"):
jitter_deg = (self._anchor_radius_m / 1000) / 111
lat = self._anchor_lat + (random.random() - 0.5) * jitter_deg
lon = self._anchor_lon + (random.random() - 0.5) * jitter_deg / 2
# House link: <a href="/catalog/houses/<city>/<slug>/<id>">
# or /catalog/houses/<slug>/<id>
house_source: str | None = None
house_ext_id: str | None = None
house_url: str | None = None
house_link = card.css_first('a[href^="/catalog/houses/"]')
if house_link is not None:
h_href = house_link.attributes.get("href", "")
parts = [p for p in h_href.strip("/").split("/") if p]
# ['catalog', 'houses', <maybe city>, <slug>, <id>]
if len(parts) >= 4 and parts[-1].isdigit():
house_source = "avito"
house_ext_id = parts[-1]
house_url = urljoin("https://www.avito.ru", h_href)
# listing_segment по item URL pattern
listing_segment: str | None = None
if "/kvartiry/" in href:
listing_segment = "vtorichka"
elif "/novostroyki/" in href:
listing_segment = "novostroyki"
return ScrapedLot(
source="avito",
source_url=url,
source_id=card.attributes.get("data-item-id"),
address=address or "Екатеринбург (Avito)",
lat=lat,
lon=lon,
address=address, # БЕЗ fallback "Екатеринбург (Avito)" — пусть None
lat=None, # C-5 fix: НЕ jitter
lon=None, # C-5 fix: НЕ jitter
rooms=rooms,
area_m2=area,
floor=floor,
total_floors=total,
price_rub=price,
photo_urls=photo_urls,
raw_payload={"title": title, "address": address},
raw_payload={"title": title, "address": address, "house_ext_id": house_ext_id},
house_source=house_source,
house_ext_id=house_ext_id,
house_url=house_url,
listing_segment=listing_segment,
)
except Exception:
return None
@ -387,41 +238,3 @@ def _extract_floor_from_title(title: str) -> tuple[int | None, int | None]:
if m:
return int(m.group(1)), int(m.group(2))
return None, None
# ── Avito __initialState__ extraction ──────────────────────────────────────
_RE_AVITO_STATE = re.compile(
r"window\.__initialState__\s*=\s*\"([^\"]+)\"", re.DOTALL
)
def _extract_avito_state(html: str) -> dict[str, Any] | None:
"""Avito embed-ит initial state как escaped JSON в большом <script>."""
m = _RE_AVITO_STATE.search(html)
if not m:
return None
try:
raw = m.group(1).encode("utf-8").decode("unicode_escape")
return json.loads(raw)
except (UnicodeDecodeError, json.JSONDecodeError):
return None
def _items_from_state(state: dict[str, Any]) -> list[dict[str, Any]]:
"""Вытаскиваем items из state — структура у Avito гуляет, ищем рекурсивно."""
# типичный путь: state.items.catalog или state.search.results
for path in [
("items", "catalog"),
("search", "results"),
("catalog", "items"),
("singleCard", "items"),
]:
cursor: Any = state
for key in path:
if not isinstance(cursor, dict):
cursor = None
break
cursor = cursor.get(key)
if isinstance(cursor, list) and cursor:
return cursor
return []

12
tradein-mvp/backend/app/services/scrapers/base.py
View file

@ -77,6 +77,12 @@ class ScrapedLot(BaseModel):
has_balcony: bool | None = None
kadastr_num: str | 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'
# Цена (обязательно)
price_rub: int = Field(gt=0)
price_per_m2: int | None = None
@ -199,6 +205,7 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
address, lat, lon, region_code,
rooms, area_m2, floor, total_floors, year_built,
house_type, repair_state, has_balcony, kadastr_num,
house_source, house_ext_id, house_url, listing_segment,
price_rub, price_per_m2,
listing_date, days_on_market, photo_urls, raw_payload,
scraped_at, last_seen_at
@ -207,6 +214,7 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
:address, :lat, :lon, 66,
:rooms, :area_m2, :floor, :total_floors, :year_built,
:house_type, :repair_state, :has_balcony, :kadastr,
:house_source, :house_ext_id, :house_url, :listing_segment,
:price_rub, :ppm2,
:listing_date, :days_on_market,
CAST(:photos AS jsonb),
@ -239,6 +247,10 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
"repair_state": lot.repair_state,
"has_balcony": lot.has_balcony,
"kadastr": lot.kadastr_num,
"house_source": lot.house_source,
"house_ext_id": lot.house_ext_id,
"house_url": lot.house_url,
"listing_segment": lot.listing_segment,
"price_rub": lot.price_rub,
"ppm2": ppm2,
"listing_date": lot.listing_date,