bot-backend
c76785308b
All checks were successful
Deploy Trade-In / changes (push) Successful in 5s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / test (push) Successful in 25s
Deploy Trade-In / build-backend (push) Successful in 41s
Deploy Trade-In / deploy (push) Successful in 34s
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
271 lines
10 KiB
Python
271 lines
10 KiB
Python
"""N1.ru scraper — вторичка ЕКБ (ekaterinburg.n1.ru).
|
||
|
||
N1.ru — региональный портал недвижимости (хорошо представлен на Урале).
|
||
SSR HTML с DOM-карточками `data-test="offers-list-item"`.
|
||
|
||
Использует curl_cffi (impersonate=chrome120) — как Cian/Avito.
|
||
|
||
Карточка:
|
||
- a[href^="/view/NNN/"] — ссылка на объявление + offer_id
|
||
текст ссылки = "1/11 2-к, Репина, 75/2 стр." (этаж/этажность + комнаты + адрес)
|
||
Этаж/этажность — leading-токен «N/M» перед сегментом комнат.
|
||
- [data-test="offers-list-item-param-total-area"] — "79 м2"
|
||
- [data-test="offers-list-item-param-price"] — "15 700 000"
|
||
- img.src — фото
|
||
|
||
Координаты карточки не отдаёт → оставляем lat/lon = None,
|
||
geocode-missing проставит реальные по адресу.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from typing import Any
|
||
|
||
from curl_cffi.requests import AsyncSession
|
||
from selectolax.parser import HTMLParser
|
||
|
||
from app.services.scraper_settings import get_scraper_delay
|
||
from app.services.scrapers.base import BaseScraper, ScrapedLot
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class N1Scraper(BaseScraper):
|
||
"""N1.ru vtorichka parser. Источник = 'n1'."""
|
||
|
||
name = "n1"
|
||
base_url = "https://ekaterinburg.n1.ru"
|
||
# Класс-дефолт; реальное значение загружается из scraper_settings при создании экземпляра.
|
||
request_delay_sec = 5.0
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.request_delay_sec = get_scraper_delay(self.name)
|
||
self._cffi: AsyncSession | None = None
|
||
|
||
async def __aenter__(self) -> N1Scraper:
|
||
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",
|
||
"Sec-Fetch-Dest": "document",
|
||
"Sec-Fetch-Mode": "navigate",
|
||
"Sec-Fetch-Site": "none",
|
||
"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)
|
||
|
||
async def fetch_around(
|
||
self,
|
||
lat: float,
|
||
lon: float,
|
||
radius_m: int = 1000,
|
||
rooms: int | None = None,
|
||
page: int = 1,
|
||
) -> list[ScrapedLot]:
|
||
self._anchor_lat = lat
|
||
self._anchor_lon = lon
|
||
self._anchor_radius_m = radius_m
|
||
|
||
url = self._build_url(rooms, page)
|
||
try:
|
||
assert self._cffi is not None
|
||
response = await self._cffi.get(url)
|
||
except Exception:
|
||
logger.exception("n1 fetch failed for %s", url)
|
||
return []
|
||
if response.status_code != 200:
|
||
logger.warning("n1 returned %d for %s", response.status_code, url)
|
||
return []
|
||
|
||
lots = self._parse_html(response.text)
|
||
logger.info(
|
||
"n1: %d lots around (%.4f, %.4f) rooms=%s page=%d",
|
||
len(lots),
|
||
lat,
|
||
lon,
|
||
rooms,
|
||
page,
|
||
)
|
||
await self.sleep_between_requests()
|
||
return lots
|
||
|
||
async def fetch_around_multi_room(
|
||
self,
|
||
lat: float,
|
||
lon: float,
|
||
radius_m: int = 1000,
|
||
pages: int = 8,
|
||
) -> list[ScrapedLot]:
|
||
"""Скрейп N1 по сегментам комнат 1/2/3/4 × N страниц — расширяет выборку."""
|
||
seen: dict[str, ScrapedLot] = {}
|
||
for rooms in (1, 2, 3, 4):
|
||
for page in range(1, pages + 1):
|
||
try:
|
||
lots = await self.fetch_around(lat, lon, radius_m, rooms=rooms, page=page)
|
||
except Exception:
|
||
logger.exception("n1 multi-room fetch failed rooms=%s page=%d", rooms, page)
|
||
continue
|
||
if not lots:
|
||
break # пустая страница → дальше нет смысла
|
||
for lot in lots:
|
||
key = lot.source_id or lot.source_url
|
||
if key and key not in seen:
|
||
seen[key] = lot
|
||
logger.info("n1 multi-room: %d unique lots around (%.4f, %.4f)", len(seen), lat, lon)
|
||
return list(seen.values())
|
||
|
||
def _build_url(self, rooms: int | None = None, page: int = 1) -> str:
|
||
"""URL каталога вторички ЕКБ.
|
||
|
||
N1 фильтр комнат: /kupit/kvartiry/vtorichka/1-komnatnye/ итд.
|
||
Пагинация: ?page=N.
|
||
"""
|
||
rooms_segment = {
|
||
1: "1-komnatnye/",
|
||
2: "2-komnatnye/",
|
||
3: "3-komnatnye/",
|
||
4: "4-komnatnye/",
|
||
}.get(rooms or 0, "")
|
||
url = f"{self.base_url}/kupit/kvartiry/vtorichka/{rooms_segment}"
|
||
if page > 1:
|
||
url += f"?page={page}"
|
||
return url
|
||
|
||
def _parse_html(self, html: str) -> list[ScrapedLot]:
|
||
tree = HTMLParser(html)
|
||
cards = tree.css('[data-test="offers-list-item"]')
|
||
return [lot for lot in (self._card_to_lot(c) for c in cards) if lot is not None]
|
||
|
||
def _card_to_lot(self, card: Any) -> ScrapedLot | None:
|
||
try:
|
||
# Ссылка на объявление /view/NNN/
|
||
link = card.css_first('a[href*="/view/"]')
|
||
if link is None:
|
||
return None
|
||
href = link.attributes.get("href", "")
|
||
m_id = re.search(r"/view/(\d+)/", href)
|
||
offer_id = m_id.group(1) if m_id else None
|
||
url = href if href.startswith("http") else f"{self.base_url}{href}"
|
||
|
||
# Заголовок ссылки: "1/11 2-к, Репина, 75/2 стр." (этаж/этажность + комнаты + адрес)
|
||
title = link.text(strip=True) or ""
|
||
floor, total_floors = _extract_floor_from_title(title)
|
||
rooms = _extract_rooms(title)
|
||
# Адрес = title без ведущего «N/M» и без префикса комнат
|
||
address = re.sub(
|
||
r"^\s*(\d+-к|студи[яи])\s*,?\s*", "", title, flags=re.IGNORECASE
|
||
).strip()
|
||
if not address:
|
||
address = "Екатеринбург (N1)"
|
||
|
||
# Площадь
|
||
area_el = card.css_first('[data-test="offers-list-item-param-total-area"]')
|
||
area = _extract_area(area_el.text(strip=True) if area_el else "")
|
||
|
||
# Цена
|
||
price_el = card.css_first('[data-test="offers-list-item-param-price"]')
|
||
price = _extract_price(price_el.text(strip=True) if price_el else "")
|
||
if not price:
|
||
return None
|
||
|
||
# Координаты НЕ ставим — N1 card не отдаёт. Оставляем None,
|
||
# geocode-missing проставит реальные по address (N1 отдаёт всё ЕКБ,
|
||
# jitter вокруг якоря был бы фейком).
|
||
lat = lon = None
|
||
|
||
# Фото
|
||
photos: list[str] = []
|
||
for img in card.css("img")[:5]:
|
||
src = img.attributes.get("src", "")
|
||
if src and src.startswith("http") and src not in photos:
|
||
photos.append(src)
|
||
|
||
return ScrapedLot(
|
||
source="n1",
|
||
source_url=url,
|
||
source_id=offer_id,
|
||
address=address,
|
||
lat=lat,
|
||
lon=lon,
|
||
rooms=rooms,
|
||
area_m2=area,
|
||
floor=floor,
|
||
total_floors=total_floors,
|
||
price_rub=price,
|
||
photo_urls=photos,
|
||
raw_payload={"title": title},
|
||
)
|
||
except Exception:
|
||
logger.exception("n1 card parse failed")
|
||
return None
|
||
|
||
|
||
# ── Helpers ─────────────────────────────────────────────────────────────────
|
||
|
||
_RE_ROOMS = re.compile(r"(\d+)\s*-?\s*к", re.IGNORECASE)
|
||
_RE_STUDIO = re.compile(r"студи", re.IGNORECASE)
|
||
_RE_AREA = re.compile(r"([\d]+(?:[.,]\d+)?)")
|
||
_RE_DIGITS = re.compile(r"\d")
|
||
|
||
# Ведущий токен «N/M» в тексте карточки N1: «1/11 2-к, Репина, 75/2 стр.»
|
||
# Якорь ^ гарантирует: только если N/M стоит В НАЧАЛЕ строки (до сегмента комнат).
|
||
# Дробь в адресе вида «75/2» в середине строки НЕ совпадёт.
|
||
_RE_FLOOR_LEADING = re.compile(r"^\s*(\d{1,3})\s*/\s*(\d{1,3})\b")
|
||
|
||
|
||
def _extract_rooms(text: str) -> int | None:
|
||
if _RE_STUDIO.search(text):
|
||
return 0
|
||
m = _RE_ROOMS.search(text)
|
||
return int(m.group(1)) if m else None
|
||
|
||
|
||
def _extract_area(text: str) -> float | None:
|
||
m = _RE_AREA.search(text)
|
||
if m:
|
||
try:
|
||
return float(m.group(1).replace(",", "."))
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _extract_price(text: str) -> int | None:
|
||
digits = "".join(_RE_DIGITS.findall(text))
|
||
return int(digits) if digits else None
|
||
|
||
|
||
def _extract_floor_from_title(text: str) -> tuple[int | None, int | None]:
|
||
"""Извлекает этаж и этажность из ведущего токена «N/M» в тексте карточки N1.
|
||
|
||
N1 SERP отдаёт текст вида «1/11 2-к, Репина, 75/2 стр.» — этаж/этажность
|
||
идут первыми, до сегмента комнат. Регулярка привязана к началу строки (^),
|
||
поэтому дробь в номере дома («75/2») в середине строки не захватывается.
|
||
|
||
Sanity-проверки:
|
||
- floor >= 1
|
||
- total_floors >= floor
|
||
- total_floors <= 100
|
||
|
||
Возвращает (None, None) при отсутствии совпадения или нарушении диапазона.
|
||
"""
|
||
m = _RE_FLOOR_LEADING.match(text)
|
||
if m is None:
|
||
return None, None
|
||
floor = int(m.group(1))
|
||
total_floors = int(m.group(2))
|
||
if floor < 1 or total_floors < floor or total_floors > 100:
|
||
return None, None
|
||
return floor, total_floors
|