"""Avito.ru scraper — парсинг вторички вокруг точки. Стратегия: HTML scrape карточек объявлений через DOM. URL pattern (EKB): https://www.avito.ru/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ ?geoCoords=56.838,60.605 &radius=1 # в км &s=104 # sort by date &p=1 # страница ВАЖНО: Avito банит httpx (403/429) по TLS fingerprint от server IP. Используем curl_cffi с impersonate='chrome120' — настоящий Chrome TLS ClientHello. """ from __future__ import annotations import logging import re from typing import Any from urllib.parse import urlencode, urljoin from curl_cffi.requests import AsyncSession from selectolax.parser import HTMLParser from app.services.scrapers.base import BaseScraper, ScrapedLot logger = logging.getLogger(__name__) class AvitoScraper(BaseScraper): """Avito vtorichka parser. Источник = 'avito'. Использует curl_cffi с impersonate=chrome120 для обхода TLS fingerprint бана. """ name = "avito" base_url = "https://www.avito.ru" # Avito жёстко мониторит — спим долго между запросами request_delay_sec = 7.0 def __init__(self) -> None: super().__init__() self._cffi: AsyncSession | None = None async def __aenter__(self) -> AvitoScraper: await super().__aenter__() self._cffi = AsyncSession( impersonate="chrome120", timeout=25, headers={ "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", "Cache-Control": "max-age=0", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Sec-Fetch-User": "?1", "Upgrade-Insecure-Requests": "1", }, ) return self async def __aexit__(self, *args: Any) -> None: if self._cffi is not None: await self._cffi.close() await super().__aexit__(*args) # ── Public ────────────────────────────────────────────────────────────── async def fetch_around( self, lat: float, lon: float, radius_m: int = 1000, *, pages: int = 1, delay_override_sec: float | None = None, ) -> list[ScrapedLot]: """Найти объявления Авито вокруг (lat, lon) в radius_m метрах. Avito работает в км — конвертируем. Минимум 1 км. pages=1 (default) — backward compat, одна страница (~50 lots). pages>1 — итерируем p=1..pages, собираем до 50*pages lots. Между запросами sleep request_delay_sec (или delay_override_sec если передан). 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)) if delay_override_sec is not None: self.request_delay_sec = delay_override_sec all_lots: list[ScrapedLot] = [] for page in range(1, pages + 1): url = self._build_web_url(lat, lon, radius_km, page=page) try: assert self._cffi is not None response = await self._cffi.get(url) if response.status_code != 200: logger.warning( "avito HTML page=%d returned %d for %s", page, response.status_code, url ) break except Exception: logger.exception("avito HTML page=%d fetch failed for %s", page, url) break lots = self._parse_html(response.text, source_url_base=url) if not lots: logger.info("avito page=%d: 0 lots — end of pagination", page) break all_lots.extend(lots) logger.info( "avito page=%d: %d lots (total %d) around (%.4f, %.4f)", page, len(lots), len(all_lots), lat, lon, ) if page < pages: await self.sleep_between_requests() if pages == 1 and all_lots: # Backward compat: single-page path does trailing sleep (как раньше) await self.sleep_between_requests() elif pages > 1: logger.info( "avito fetch_around pages=%d total_lots=%d around (%.4f, %.4f)", pages, len(all_lots), lat, lon, ) return all_lots # ── Strategy B: HTML scrape ───────────────────────────────────────────── def _build_web_url(self, lat: float, lon: float, radius_km: int, page: int = 1) -> str: params = { "geoCoords": f"{lat},{lon}", "radius": radius_km, "s": 104, # sort by date "p": page, } return ( f"{self.base_url}/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ?{urlencode(params)}" ) def _parse_html(self, html: str, source_url_base: str) -> list[ScrapedLot]: """Парсим карточки объявлений из HTML через DOM scrape. Avito state давно пустой — используем только DOM путь. """ tree = HTMLParser(html) cards = tree.css('[data-marker="item"]') lots: list[ScrapedLot] = [] for card in cards: lot = self._dom_card_to_lot(card, source_url_base) if lot is not None: lots.append(lot) return lots def _dom_card_to_lot(self, card: Any, source_url_base: str) -> ScrapedLot | None: """Парсинг карточки объявления через 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: return None href = link_el.attributes.get("href", "") url = urljoin("https://www.avito.ru", href) title = link_el.text(strip=True) price_el = card.css_first('meta[itemprop="price"]') price = int(price_el.attributes.get("content", 0)) if price_el is not None else 0 if price <= 0: return None rooms = _extract_rooms_from_title(title) area = _extract_area_from_title(title) floor, total = _extract_floor_from_title(title) # Адрес: первый

внутри data-marker="item-location" address: str | None = None loc_el = card.css_first('[data-marker="item-location"]') if loc_el is not None: street_el = loc_el.css_first("p") if street_el is not None: address = street_el.text(strip=True) or None # Фото photo_urls: list[str] = [] for img in card.css('img[itemprop="image"]'): src = img.attributes.get("src") or "" if src.startswith("http") and src not in photo_urls: photo_urls.append(src) if len(photo_urls) >= 5: break # House link: # or /catalog/houses// 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', , , ] 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, # БЕЗ 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, "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 # ── Helpers: title parser ─────────────────────────────────────────────────── _RE_ROOMS = re.compile(r"(\d)-к\.?\s*(квартира|кв\.)", re.IGNORECASE) _RE_STUDIO = re.compile(r"студи[яиюей]", re.IGNORECASE) _RE_AREA = re.compile(r"(\d+[.,]?\d*)\s*м[²2]", re.IGNORECASE) _RE_FLOOR = re.compile(r"(\d+)\s*/\s*(\d+)\s*эт\.?", re.IGNORECASE) def _extract_rooms_from_title(title: str) -> int | None: if _RE_STUDIO.search(title): return 0 m = _RE_ROOMS.search(title) if m: return int(m.group(1)) return None def _extract_area_from_title(title: str) -> float | None: m = _RE_AREA.search(title) if m: return float(m.group(1).replace(",", ".")) return None def _extract_floor_from_title(title: str) -> tuple[int | None, int | None]: m = _RE_FLOOR.search(title) if m: return int(m.group(1)), int(m.group(2)) return None, None