bot-backend
bee44ffd3d
Run 216 recovered 403s (merged fix) but still banned on a 429 at page 36 (deep pagination, backconnect conn-limit). Widen the 429 retry budget + one reconnect on exhaustion in _fetch_serp_html, and make fetch_all_secondary skip a blocked bucket and continue rather than aborting the run; only N consecutive blocked buckets (real hard ban) re-raises. Preserves hard-ban detection.
1927 lines
96 KiB
Python
1927 lines
96 KiB
Python
"""Avito.ru scraper — парсинг вторички вокруг точки + citywide/byrooms sweep.
|
||
|
||
Стратегия: HTML scrape карточек объявлений через DOM.
|
||
|
||
URL patterns (EKB):
|
||
geo-mode (default):
|
||
https://www.avito.ru/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ
|
||
?geoCoords=56.838,60.605
|
||
&radius=1 # в км
|
||
&s=104 # sort by date
|
||
&p=1 # страница
|
||
|
||
citywide-mode (без geo, весь город, T6):
|
||
https://www.avito.ru/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ
|
||
?s=104&p=1
|
||
|
||
byrooms-mode (citywide + фильтр комнатности, T6):
|
||
https://www.avito.ru/ekaterinburg/kvartiry/prodam/<room-slug>?s=104&p=1
|
||
|
||
ВАЖНО: Avito банит httpx (403/429) по TLS fingerprint от server IP.
|
||
Используем curl_cffi с impersonate='chrome120' — настоящий Chrome TLS ClientHello.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import inspect
|
||
import logging
|
||
import math
|
||
import re
|
||
from collections.abc import Callable
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from typing import Any
|
||
from urllib.parse import urlencode, urljoin, urlparse, urlunparse
|
||
|
||
import httpx
|
||
from curl_cffi.requests import AsyncSession
|
||
from selectolax.parser import HTMLParser
|
||
|
||
from app.core.config import settings
|
||
from app.services.scraper_settings import get_scraper_delay
|
||
from app.services.scrapers.avito_exceptions import (
|
||
AvitoBlockedError,
|
||
AvitoContentBlockedError,
|
||
AvitoRateLimitedError,
|
||
)
|
||
from app.services.scrapers.base import BaseScraper, ScrapedLot
|
||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||
from app.services.scrapers.price_brackets import get_price_seed_brackets
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Relative date parsing ────────────────────────────────────────────────────
|
||
_REL_DATE_RE = re.compile(
|
||
r"(?P<n>\d+)\s+(?P<unit>"
|
||
r"секунд[ауы]?|"
|
||
r"минут[ауы]?|"
|
||
r"час(?:а|ов)?|"
|
||
r"день|дн(?:я|ей?)|"
|
||
r"недел[июяь]+|"
|
||
r"месяц(?:а|ев)?|"
|
||
r"год(?:а|ов)?"
|
||
r")\s+назад",
|
||
flags=re.I,
|
||
)
|
||
_UNIT_DAYS: dict[str, int] = {
|
||
"секунд": 0,
|
||
"минут": 0,
|
||
"час": 0,
|
||
"день": 1,
|
||
"дн": 1,
|
||
"недел": 7,
|
||
"месяц": 30,
|
||
"год": 365,
|
||
}
|
||
# #1387: Avito SERP часто показывает давность в единственном числе БЕЗ ведущего
|
||
# числа — «час назад», «минуту назад», «день назад», «неделю назад», «месяц
|
||
# назад», «год назад». Числовой `_REL_DATE_RE` их не матчит (требует `\d+`),
|
||
# поэтому отдельная ветка с неявным n=1. Юниты в винительном падеже («минуту»,
|
||
# «неделю») плюс именительный («час», «день»).
|
||
_REL_DATE_SINGULAR_RE = re.compile(
|
||
r"(?P<unit>" r"секунду|" r"минуту|" r"час|" r"день|" r"неделю|" r"месяц|" r"год" r")\s+назад",
|
||
flags=re.I,
|
||
)
|
||
|
||
# Russian month names → number. Avito показывает абсолютные даты как «18 мая».
|
||
_RU_MONTHS: dict[str, int] = {
|
||
"январ": 1,
|
||
"феврал": 2,
|
||
"март": 3,
|
||
"апрел": 4,
|
||
"ма": 5,
|
||
"июн": 6,
|
||
"июл": 7,
|
||
"август": 8,
|
||
"сентябр": 9,
|
||
"октябр": 10,
|
||
"ноябр": 11,
|
||
"декабр": 12,
|
||
}
|
||
# «18 мая», «5 июня в 12:30» — day + month (year inferred current).
|
||
_ABS_DATE_RE = re.compile(
|
||
r"(?P<day>\d{1,2})\s+(?P<month>январ\w*|феврал\w*|март\w*|апрел\w*|"
|
||
r"ма[яй]|июн\w*|июл\w*|август\w*|сентябр\w*|октябр\w*|ноябр\w*|декабр\w*)",
|
||
flags=re.I,
|
||
)
|
||
|
||
|
||
def _parse_relative_date(s: str | None) -> date | None:
|
||
"""Avito показывает дату публикации в трёх форматах:
|
||
* '5 дней назад' / 'час назад' / 'неделю назад' — relative numeric
|
||
* 'сегодня в 12:30' / 'вчера в 18:45' / 'позавчера' — relative keyword
|
||
* '18 мая' / '5 июня в 12:30' — absolute day + month (year inferred)
|
||
|
||
PR M-followup 2026-05-27: расширено покрытие — раньше только relative numeric
|
||
matched (regex `\\d+ unit назад`), 100% Avito listings заходило с listing_date=None
|
||
т.к. Avito SERP для свежих cards использует keyword/absolute форматы. См.
|
||
тесты для каждого case.
|
||
"""
|
||
if not s:
|
||
return None
|
||
text_lower = s.lower()
|
||
|
||
# Keyword shortcuts (no \d+).
|
||
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()
|
||
|
||
# Relative numeric («5 дней назад»).
|
||
m = _REL_DATE_RE.search(s)
|
||
if m:
|
||
n = int(m["n"])
|
||
unit_raw = m["unit"].lower()
|
||
for prefix, days in _UNIT_DAYS.items():
|
||
if unit_raw.startswith(prefix):
|
||
return date.today() - timedelta(days=n * days)
|
||
|
||
# #1387: Relative singular без числа («час назад», «неделю назад»). Неявное
|
||
# n=1; те же _UNIT_DAYS. Винительный падеж юнита нормализуем к ключу-префиксу.
|
||
m = _REL_DATE_SINGULAR_RE.search(s)
|
||
if m:
|
||
unit_raw = m["unit"].lower()
|
||
# «минуту»→«минут», «неделю»→«недел», «секунду»→«секунд»; остальные уже
|
||
# совпадают с префиксами _UNIT_DAYS.
|
||
unit_key = {"минуту": "минут", "неделю": "недел", "секунду": "секунд"}.get(
|
||
unit_raw, unit_raw
|
||
)
|
||
for prefix, days in _UNIT_DAYS.items():
|
||
if unit_key.startswith(prefix):
|
||
return date.today() - timedelta(days=days)
|
||
|
||
# Absolute («18 мая», «5 июня в 12:30»). Year — current; если месяц
|
||
# в будущем относительно сегодня (например сейчас декабрь, увидели
|
||
# «5 января») — откатываем к прошлому году.
|
||
m = _ABS_DATE_RE.search(s)
|
||
if m:
|
||
day = int(m["day"])
|
||
month_raw = m["month"].lower()
|
||
month: int | None = None
|
||
for prefix, num in _RU_MONTHS.items():
|
||
if month_raw.startswith(prefix):
|
||
month = num
|
||
break
|
||
if month is not None and 1 <= day <= 31:
|
||
today = date.today()
|
||
year = today.year
|
||
try:
|
||
parsed = date(year, month, day)
|
||
except ValueError:
|
||
return None
|
||
# Если parsed в будущем больше чем на 30 дней — Avito показывает
|
||
# прошлогоднюю дату («декабрь» в феврале значит прошлый декабрь).
|
||
if (parsed - today).days > 30:
|
||
try:
|
||
parsed = date(year - 1, month, day)
|
||
except ValueError:
|
||
return None
|
||
return parsed
|
||
|
||
return None
|
||
|
||
|
||
# #726: Avito SERP лениво рендерит DOM — `data-marker="item-date"` присутствует только
|
||
# у верхних (above-the-fold) карточек (~10 из ~46), у остальных его НЕТ в raw-HTML, что
|
||
# curl_cffi получает (JS-гидрация не выполняется). Но per-item JSON в той же HTML несёт
|
||
# `sortTimeStamp` (epoch-ms) почти у всех карточек — это drift-resistant источник даты.
|
||
# Маппим item_id (= DOM `data-item-id`) → дата. `{0,1500}` ограничивает поиск телом
|
||
# одного item-объекта (id и sortTimeStamp лежат в нём; внутри ещё есть короткий вложенный
|
||
# `addressDetailed{…}`, поэтому не `[^{}]`). re.S — sortTimeStamp может быть на новой строке.
|
||
# sortTimeStamp — UTC epoch-ms. Avito показывает дату в MSK (UTC+3, без DST с 2014).
|
||
# Берём .date() в MSK, чтобы совпадало с тем, что видно на сайте (иначе объявления,
|
||
# поднятые в 00:00–03:00 MSK, давали бы дату на день раньше). Fixed offset, без
|
||
# зависимости от системной tzdata в контейнере. #726.
|
||
_MSK = timezone(timedelta(hours=3))
|
||
_SORT_TS_RE = re.compile(r'"id":(?P<id>\d{8,12})\b.{0,1500}?"sortTimeStamp":(?P<ts>\d+)', re.S)
|
||
|
||
|
||
def _build_sort_timestamp_map(html: str) -> dict[str, date]:
|
||
"""item_id → дата публикации из embedded JSON (`sortTimeStamp`, epoch-ms).
|
||
|
||
Best-effort: при любой ошибке конкретной записи — пропускаем её (не роняем парс).
|
||
Первое вхождение id выигрывает (item-объект идёт раньше recommendations-дублей).
|
||
"""
|
||
out: dict[str, date] = {}
|
||
for m in _SORT_TS_RE.finditer(html):
|
||
item_id = m.group("id")
|
||
if item_id in out:
|
||
continue
|
||
try:
|
||
out[item_id] = datetime.fromtimestamp(int(m.group("ts")) / 1000, tz=_MSK).date()
|
||
except (ValueError, OSError, OverflowError):
|
||
continue
|
||
return out
|
||
|
||
|
||
# #623: Avito банит datacenter-IP — на HTTP 200 отдаёт firewall-интерстициал
|
||
# «Доступ ограничен: проблема с IP» (title в <head>), а НЕ 403. Детектим по
|
||
# маркерам в начале документа (firewall-страница объёмна, не сканируем целиком).
|
||
_FIREWALL_MARKERS = ("доступ ограничен", "проблема с ip", "firewall-container")
|
||
|
||
# EKB-фильтр «Новостройка» (slug категории kvartiry/prodam). Кодирует
|
||
# ASgB-параметры выборки только новостроек — SERP возвращает 100% new-build
|
||
# карточки (с data-marker="item-development-name"). Извлечён из реального
|
||
# search-URL: /ekaterinburg/kvartiry/prodam/novostroyka-ASgBAgICAkSSA8YQ5geOUg
|
||
NOVOSTROYKA_SLUG = "novostroyka-ASgBAgICAkSSA8YQ5geOUg"
|
||
|
||
# ── Exhaustive full-load (room×price bisection) ───────────────────────────────
|
||
# Верхняя граница цены при первом рекурсивном делении (нет явного hi).
|
||
# 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 результатов.
|
||
# Больше НЕ корень бисекции (см. _AVITO_PRICE_SEED_BRACKETS) — оставлен для
|
||
# совместимости (используется в тестах probe-устойчивости как sentinel).
|
||
_AVITO_MAX_PRICE = 200_000_000
|
||
# Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET прекращаем
|
||
# деление и пагинируем как есть (принимаем возможный «хвост»).
|
||
_AVITO_MIN_BRACKET = 50_000
|
||
|
||
# Seed price-brackets (руб) — data-driven по прод-распределению вторички ЕКБ
|
||
# (пик 4-8М, bulk 3-16М). Теперь общий констант для всех exhaustive-скрапперов
|
||
# (avito/cian/yandex) — app.services.scrapers.price_brackets.get_price_seed_brackets().
|
||
# Локальный алиас сохранён для читаемости и тестов; значения идентичны shared EKB.
|
||
# Каждый (комнатность × брекет) дальше дробится _walk_price_range если total > cap.
|
||
# Последний брекет ОТКРЫТ (hi=None → pmax не ставится) — ловит весь хвост люкса
|
||
# без потолка; он крошечный (avito >50М ≈ 73), пагинируется напрямую без бисекции.
|
||
_AVITO_PRICE_SEED_BRACKETS: list[tuple[int, int | None]] = get_price_seed_brackets()
|
||
# Avito SERP показывает ~50 карточек на страницу.
|
||
_AVITO_OFFERS_PER_PAGE = 50
|
||
|
||
# HTTP 429 в curl_cffi-режиме через backconnect-прокси (mproxy.site) — НЕ IP-ban, а
|
||
# transient «слишком много одновременных соединений» (лимит 5). Проходит на коротком
|
||
# retry без ротации IP. Делаем до _AVITO_429_MAX_RETRIES коротких пауз; если они
|
||
# исчерпаны под backconnect — ещё _AVITO_429_RECONNECT_RETRIES попыток через
|
||
# пересоздание сессии (_reset_cffi): свежее соединение освобождает удерживаемый
|
||
# conn-slot и может попасть на менее загруженный путь. Только потом —
|
||
# AvitoRateLimitedError (run 216: deep-pagination 429 на page 36 ронял весь sweep).
|
||
_AVITO_429_MAX_RETRIES = 8
|
||
_AVITO_429_BACKOFF_SEC = 1.5
|
||
# Reconnect-fallback на исчерпанном 429 short-retry (только backconnect): дольше спим
|
||
# (_AVITO_403_BACKOFF_SEC), чтобы дать conn-slot'ам разгрузиться.
|
||
_AVITO_429_RECONNECT_RETRIES = 3
|
||
|
||
# HTTP 403 / firewall-интерстициал в curl_cffi-режиме через backconnect-прокси
|
||
# (mproxy.site, нет changeip-URL): провайдер выдаёт СВЕЖИЙ exit-IP на каждое НОВОЕ
|
||
# соединение (5 последовательных запросов → 5 разных IP). 403 значит залочен лишь
|
||
# ТЕКУЩИЙ exit-IP — открытие нового CONNECT-туннеля escape'ит бан. Keep-alive curl_cffi
|
||
# сессия переиспользует тот же туннель = тот же IP, поэтому retry на том же self._cffi
|
||
# не помогает: нужно пересоздать сессию (_reset_cffi). Делаем до _AVITO_403_MAX_RETRIES
|
||
# reconnect-попыток прежде чем свалиться в AvitoBlockedError (mark_banned). Старый
|
||
# single-IP changeip-путь и browser-режим не затрагиваются.
|
||
_AVITO_403_MAX_RETRIES = 5
|
||
_AVITO_403_BACKOFF_SEC = 2.0
|
||
|
||
# Per-bucket block tolerance в fetch_all_secondary: одиночный 429/блок на ОДНОМ
|
||
# (комнатность × ценовой брекет) бакете (напр. deep-pagination dense-бакета) НЕ должен
|
||
# ронять весь sweep — бакет скипается, обход продолжается. Только N ПОДРЯД заблокированных
|
||
# бакетов = реальный hard IP-ban → re-raise (run_avito_full_load → mark_banned).
|
||
_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED = 4
|
||
|
||
# Счётчик результатов в шапке SERP: <span data-marker="page-title/count">1 178</span>.
|
||
# Текст может содержать NBSP (\xa0), -entity или обычные пробелы как разделители
|
||
# разрядов — чистятся в _extract_total_count. [^<]* нежадно ловит весь текст до </span>.
|
||
_TOTAL_COUNT_RE = re.compile(
|
||
r'data-marker="page-title/count"[^>]*>(?P<count>[^<]*)<',
|
||
flags=re.I,
|
||
)
|
||
|
||
# Маркер валидной ПУСТОЙ выборки. На 0-результатной странице (напр. студии с
|
||
# pmin=17M, которых нет) Avito отдаёт HTTP 200 БЕЗ `page-title/count`, но с
|
||
# текстом «ничего не найдено» И ~50 fallback-«похожими» объявлениями (НЕ
|
||
# подходящими под ценовой фильтр). Без этого маркера _extract_total_count
|
||
# возвращал None → resilience-логика дробила пустой брекет + ротировала IP
|
||
# (прод-инцидент run 209: 14+ мин в пустых верхних брекетах, errors=1058).
|
||
# Lowercase substring «ничего не найден» ловит «найдено»/«найдены»/любой регистр.
|
||
_AVITO_NO_RESULTS_MARKERS = ("ничего не найден",)
|
||
|
||
|
||
def _is_firewall_page(html: str) -> bool:
|
||
"""True если Avito вернул firewall-страницу IP-блока (на HTTP 200)."""
|
||
head = html[:4096].lower()
|
||
return any(marker in head for marker in _FIREWALL_MARKERS)
|
||
|
||
|
||
class AvitoScraper(BaseScraper):
|
||
"""Avito vtorichka parser. Источник = 'avito'.
|
||
|
||
Использует curl_cffi с impersonate=chrome120 для обхода TLS fingerprint бана.
|
||
"""
|
||
|
||
name = "avito"
|
||
base_url = "https://www.avito.ru"
|
||
# Avito жёстко мониторит — спим долго между запросами.
|
||
# Класс-дефолт; реальное значение загружается из scraper_settings при создании экземпляра.
|
||
request_delay_sec = 7.0
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.request_delay_sec = get_scraper_delay(self.name)
|
||
self._cffi: AsyncSession | None = None
|
||
self._browser: BrowserFetcher | None = None
|
||
# #823: счётчик карточек, которые не удалось распарсить из-за неожиданной структуры DOM.
|
||
self.parse_failures: int = 0
|
||
|
||
async def __aenter__(self) -> AvitoScraper:
|
||
# Проактивная ротация IP в начале sweep (#1731): apw-IP «протухает» по
|
||
# Datadome между суточными прогонами, поэтому первые якоря блокируются и
|
||
# реактивного бюджета max_rotations не хватает (3 подряд блока →
|
||
# mark_banned). Один changeip-вызов до первого SERP-fetch даёт свежий IP.
|
||
# _rotate_ip() сам ждёт ~9с (settle) и логирует/глотает ошибку, поэтому
|
||
# тут только дополнительный guard: сбой ротации НЕ должен ронять sweep.
|
||
if settings.avito_proxy_rotate_url:
|
||
try:
|
||
if await self._rotate_ip():
|
||
logger.info("avito: proactive IP rotate at sweep start (#1731)")
|
||
else:
|
||
logger.warning("avito: proactive IP rotate at sweep start failed — proceeding")
|
||
except Exception:
|
||
logger.warning(
|
||
"avito: proactive IP rotate at sweep start raised — proceeding", exc_info=True
|
||
)
|
||
if settings.scraper_fetch_mode == "browser":
|
||
self._browser = BrowserFetcher(source="avito")
|
||
await self._browser.__aenter__()
|
||
logger.info("avito: SERP fetch via BrowserFetcher (camoufox) — #901")
|
||
return self
|
||
await super().__aenter__()
|
||
self._cffi = self._build_cffi_session()
|
||
return self
|
||
|
||
def _build_cffi_session(self) -> AsyncSession:
|
||
"""Создать curl_cffi-сессию (impersonate chrome120, через прокси если задан).
|
||
|
||
Вынесено из __aenter__ чтобы _reset_cffi мог пересоздать сессию (новый
|
||
CONNECT-туннель) для escape 403 на залоченном backconnect exit-IP.
|
||
"""
|
||
proxies = None
|
||
if settings.scraper_proxy_url:
|
||
proxies = {"http": settings.scraper_proxy_url, "https": settings.scraper_proxy_url}
|
||
logger.info("avito: routing through mobile proxy egress (#623)")
|
||
return AsyncSession(
|
||
impersonate="chrome120",
|
||
timeout=25,
|
||
proxies=proxies,
|
||
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",
|
||
},
|
||
)
|
||
|
||
async def _reset_cffi(self) -> None:
|
||
"""Закрыть и пересоздать curl_cffi-сессию — новый CONNECT-туннель через
|
||
backconnect-прокси = свежий exit-IP (escape 403 на залоченном IP)."""
|
||
if self._cffi is not None:
|
||
try:
|
||
await self._cffi.close()
|
||
except Exception:
|
||
pass
|
||
self._cffi = self._build_cffi_session()
|
||
|
||
async def __aexit__(self, *args: Any) -> None:
|
||
if self._browser is not None:
|
||
await self._browser.__aexit__(*args)
|
||
if self._cffi is not None:
|
||
await self._cffi.close()
|
||
await super().__aexit__(*args)
|
||
|
||
# ── Anti-block (#623) ─────────────────────────────────────────────────────
|
||
async def _rotate_ip(self) -> bool:
|
||
"""Сменить мобильный IP через changeip-ссылку mobileproxy (#623).
|
||
|
||
Дёргается напрямую (без прокси) — это API провайдера, не Avito. Ждём ~9с:
|
||
мобильному модему нужно время поднять новый IP. Returns True при успехе.
|
||
"""
|
||
rotate_url = settings.avito_proxy_rotate_url
|
||
if not rotate_url:
|
||
return False
|
||
sep = "&" if "?" in rotate_url else "?"
|
||
try:
|
||
async with AsyncSession(timeout=30) as rot:
|
||
await rot.get(f"{rotate_url}{sep}format=json")
|
||
await asyncio.sleep(9)
|
||
logger.info("avito proxy: IP rotated via changeip")
|
||
return True
|
||
except Exception:
|
||
logger.warning("avito proxy: IP rotation failed", exc_info=True)
|
||
return False
|
||
|
||
async def _fetch_serp_html(self, url: str, page: int) -> str | None:
|
||
"""GET SERP HTML с детектом IP-бана и ротацией мобильного IP (#623).
|
||
|
||
Avito на забаненном datacenter-IP отдаёт HTTP 200 + firewall-страницу
|
||
(не 403). Детектим по тексту; если задана changeip-ссылка — меняем IP и
|
||
повторяем до ``avito_proxy_max_rotations`` раз. Транзиентные сетевые
|
||
обрывы (мобильный канал) ретраим без ротации.
|
||
|
||
Returns:
|
||
HTML при HTTP 200, либо ``None`` при non-200 (конец пагинации).
|
||
В browser-mode HTTP-статус недоступен: метод возвращает HTML или
|
||
поднимает ``AvitoBlockedError`` (pagination ends via empty-parse in callers).
|
||
Raises:
|
||
AvitoBlockedError / AvitoRateLimitedError — если бан не снят ротацией.
|
||
"""
|
||
if self._browser is not None:
|
||
# Browser-mode (#901): no HTTP status code — soft firewall-detect + rotate.
|
||
max_rot = settings.avito_proxy_max_rotations if settings.avito_proxy_rotate_url else 0
|
||
rot_done = 0
|
||
while True:
|
||
html = await self._browser.fetch(url)
|
||
if _is_firewall_page(html):
|
||
if rot_done < max_rot and await self._rotate_ip():
|
||
rot_done += 1
|
||
logger.info(
|
||
"avito page=%d browser firewall — retry after rotation #%d",
|
||
page,
|
||
rot_done,
|
||
)
|
||
continue
|
||
logger.error("avito SERP firewall in browser-mode page=%d url=%s", page, url)
|
||
raise AvitoBlockedError(
|
||
f"Avito SERP firewall (browser-mode) at page={page} — IP banned"
|
||
)
|
||
return html
|
||
assert self._cffi is not None
|
||
max_rot = settings.avito_proxy_max_rotations if settings.avito_proxy_rotate_url else 0
|
||
# curl_cffi-сессия всегда egress через settings.scraper_proxy_url (backconnect
|
||
# mproxy). Пересоздание сессии (_reset_cffi) даёт свежий exit-IP вне зависимости
|
||
# от наличия changeip-URL (тот относится к ДРУГОМУ, browser/auv-прокси). Поэтому
|
||
# reconnect-retry гейтим ТОЛЬКО на наличие proxy, не на отсутствие rotate_url.
|
||
backconnect = bool(settings.scraper_proxy_url)
|
||
rot_done = 0
|
||
transient_left = 2
|
||
r429 = 0
|
||
r429_reconnect = 0
|
||
r403 = 0
|
||
while True:
|
||
try:
|
||
response = await self._cffi.get(url)
|
||
except Exception:
|
||
if transient_left > 0:
|
||
transient_left -= 1
|
||
logger.warning(
|
||
"avito page=%d transient fetch error — retry (left=%d)",
|
||
page,
|
||
transient_left,
|
||
exc_info=True,
|
||
)
|
||
await asyncio.sleep(3)
|
||
continue
|
||
raise
|
||
|
||
sc = response.status_code
|
||
is_firewall = sc == 200 and _is_firewall_page(response.text)
|
||
# 429 в curl_cffi через backconnect-прокси — transient conn-limit, НЕ
|
||
# IP-ban: короткий retry того же url без ротации. Только если короткие
|
||
# retry исчерпаны — проваливаемся в общую блок-логику ниже (rotation/raise).
|
||
if sc == 429 and r429 < _AVITO_429_MAX_RETRIES:
|
||
r429 += 1
|
||
logger.info(
|
||
"avito page=%d HTTP 429 (backconnect conn-limit) — short retry %d/%d",
|
||
page,
|
||
r429,
|
||
_AVITO_429_MAX_RETRIES,
|
||
)
|
||
jitter = (r429 % 3) * 0.1
|
||
await asyncio.sleep(_AVITO_429_BACKOFF_SEC + jitter)
|
||
continue
|
||
if sc in (403, 429) or is_firewall:
|
||
if rot_done < max_rot and await self._rotate_ip():
|
||
rot_done += 1
|
||
logger.info(
|
||
"avito page=%d blocked (HTTP %d, firewall=%s) — retry after rotation #%d",
|
||
page,
|
||
sc,
|
||
is_firewall,
|
||
rot_done,
|
||
)
|
||
continue
|
||
# backconnect: 403/firewall = только текущий exit-IP залочен. Новый
|
||
# CONNECT-туннель (пересоздание сессии) = свежий IP → reconnect-retry.
|
||
# Только для 403/firewall (НЕ 429 — у него свой short-retry выше).
|
||
if backconnect and (sc == 403 or is_firewall) and r403 < _AVITO_403_MAX_RETRIES:
|
||
r403 += 1
|
||
logger.info(
|
||
"avito page=%d blocked (HTTP %d, firewall=%s) — "
|
||
"backconnect reconnect retry %d/%d",
|
||
page,
|
||
sc,
|
||
is_firewall,
|
||
r403,
|
||
_AVITO_403_MAX_RETRIES,
|
||
)
|
||
await self._reset_cffi()
|
||
await asyncio.sleep(_AVITO_403_BACKOFF_SEC)
|
||
continue
|
||
if sc == 429:
|
||
# short-retry (тот же session) исчерпан. Под backconnect одна
|
||
# фаза reconnect-fallback: пересоздание сессии освобождает
|
||
# удерживаемый conn-slot и может лечь на менее загруженный путь
|
||
# (deep-pagination 429 = conn-limit, не IP-ban). Только потом raise.
|
||
if backconnect and r429_reconnect < _AVITO_429_RECONNECT_RETRIES:
|
||
r429_reconnect += 1
|
||
logger.info(
|
||
"avito page=%d HTTP 429 short-retry exhausted — "
|
||
"backconnect reconnect retry %d/%d",
|
||
page,
|
||
r429_reconnect,
|
||
_AVITO_429_RECONNECT_RETRIES,
|
||
)
|
||
await self._reset_cffi()
|
||
await asyncio.sleep(_AVITO_403_BACKOFF_SEC)
|
||
continue
|
||
logger.error("avito SERP HTTP 429 (rate limited) page=%d url=%s", page, url)
|
||
raise AvitoRateLimitedError(
|
||
f"Avito SERP returned 429 — rate limited at page={page}"
|
||
)
|
||
logger.error(
|
||
"avito SERP blocked page=%d (HTTP %d, firewall=%s) url=%s",
|
||
page,
|
||
sc,
|
||
is_firewall,
|
||
url,
|
||
)
|
||
raise AvitoBlockedError(
|
||
f"Avito SERP blocked (HTTP {sc}, firewall={is_firewall}) "
|
||
f"at page={page} — IP banned"
|
||
)
|
||
if sc != 200:
|
||
logger.warning("avito HTML page=%d returned %d for %s", page, sc, url)
|
||
return None
|
||
return response.text
|
||
|
||
async def _fetch_serp_html_with_retry(
|
||
self, url: str, page: int, *, max_retries: int = 2
|
||
) -> str | None:
|
||
"""Page-level retry поверх _fetch_serp_html для sweep-обходов (P2 resilience).
|
||
|
||
_fetch_serp_html уже ретраит транзиентные fetch-обрывы и ротирует IP, но
|
||
если после внутренних попыток он всё равно поднимает generic Exception —
|
||
внешний sweep-цикл (citywide/byrooms) делал break и терял весь хвост.
|
||
Здесь даём странице ещё ``max_retries`` попыток перед тем как пробросить.
|
||
|
||
Поведение:
|
||
- AvitoBlockedError / AvitoRateLimitedError — hard-block, пробрасываются
|
||
немедленно (НЕ retry).
|
||
- generic Exception — если попытки остались: warning + короткий backoff +
|
||
следующая попытка; иначе re-raise последнего исключения (внешний цикл
|
||
решит break).
|
||
- None — это «non-200 / конец пагинации», НЕ ошибка → возвращается как есть
|
||
без ретраев.
|
||
"""
|
||
attempts = max_retries + 1
|
||
backoff = max(2.0, min(self.request_delay_sec, 4.0))
|
||
for attempt in range(1, attempts + 1):
|
||
try:
|
||
return await self._fetch_serp_html(url, page)
|
||
except (AvitoBlockedError, AvitoRateLimitedError):
|
||
raise
|
||
except Exception:
|
||
if attempt >= attempts:
|
||
raise
|
||
logger.warning(
|
||
"avito page=%d transient error — retry (attempt %d/%d)",
|
||
page,
|
||
attempt,
|
||
max_retries,
|
||
exc_info=True,
|
||
)
|
||
await asyncio.sleep(backoff)
|
||
return None # pragma: no cover — loop either returns or raises
|
||
|
||
# ── 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:
|
||
html = await self._fetch_serp_html(url, page)
|
||
except (AvitoBlockedError, AvitoRateLimitedError):
|
||
raise
|
||
except Exception:
|
||
logger.exception("avito HTML page=%d fetch failed for %s", page, url)
|
||
break
|
||
if html is None:
|
||
break
|
||
|
||
lots = self._parse_html(html, source_url_base=url)
|
||
if not lots:
|
||
if page == 1:
|
||
logger.error(
|
||
"avito SERP page=1 returned HTTP 200 but 0 cards — "
|
||
"likely content-block/captcha url=%s",
|
||
url,
|
||
)
|
||
raise AvitoContentBlockedError(
|
||
"Avito SERP HTTP 200 with 0 cards on page=1 — content-block suspected"
|
||
)
|
||
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 _build_citywide_url(self, page: int = 1) -> str:
|
||
"""T6: URL всего города ЕКБ без geo-фильтра, сортировка по дате.
|
||
|
||
Avito отдаёт все объявления города (cap ~5000 = 100 страниц × 50 карточек).
|
||
"""
|
||
params = {"s": 104, "p": page}
|
||
return f"{self.base_url}/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ?{urlencode(params)}"
|
||
|
||
def _build_newbuilding_url(self, page: int = 1) -> str:
|
||
"""URL ЕКБ-выборки только новостроек (novostroyka-filter), сортировка по дате.
|
||
|
||
Тот же shape параметров, что _build_citywide_url (s=104, p=page), но slug
|
||
категории NOVOSTROYKA_SLUG отдаёт исключительно new-build карточки.
|
||
"""
|
||
params = {"s": 104, "p": page}
|
||
path = f"/ekaterinburg/kvartiry/prodam/{NOVOSTROYKA_SLUG}"
|
||
return f"{self.base_url}{path}?{urlencode(params)}"
|
||
|
||
def _build_rooms_url(
|
||
self,
|
||
room_slug: str,
|
||
page: int = 1,
|
||
min_price: int | None = None,
|
||
max_price: int | None = None,
|
||
) -> str:
|
||
"""T6: URL с фильтром по комнатности для всего ЕКБ (no geo).
|
||
|
||
Дробление по комнатности позволяет обойти cap ~5000 на category.
|
||
room_slug — берётся из ROOM_SLUGS (кодирует ASgB-параметры выборки).
|
||
|
||
min_price/max_price — опциональные ценовые границы (Avito `pmin`/`pmax`),
|
||
добавляются в query ТОЛЬКО если заданы (backward-compat: вызовы без цены
|
||
не меняют URL). Используются exhaustive full-load для room×price бисекции.
|
||
"""
|
||
params: dict[str, Any] = {"s": 104, "p": page}
|
||
if min_price is not None:
|
||
params["pmin"] = min_price
|
||
if max_price is not None:
|
||
params["pmax"] = max_price
|
||
return f"{self.base_url}/ekaterinburg/kvartiry/prodam/{room_slug}?{urlencode(params)}"
|
||
|
||
def _extract_total_count(self, html: str) -> int | None:
|
||
"""Извлечь общее число результатов из Avito SERP (`page-title/count`).
|
||
|
||
Avito рендерит счётчик в <span data-marker="page-title/count">1 178</span>.
|
||
Это аналог cian totalOffers — gate для бисекции (total > cap → split).
|
||
Парсинг: текст span → убрать NBSP (\xa0)/пробелы/` ` → int.
|
||
|
||
Порядок: сначала `page-title/count` (его несёт ТОЛЬКО непустая выборка),
|
||
потом no-results-маркер → 0 (валидная пустая выборка: брекет скипается без
|
||
split/rotation и без скрапа ~50 fallback-«похожих»), и только при отсутствии
|
||
обоих — None (реально битая/captcha/firewall/незагруженная страница).
|
||
|
||
Returns:
|
||
int при успехе (включая 0 на no-results-странице), либо None при
|
||
captcha/firewall/незагруженной странице (ни count, ни no-results).
|
||
"""
|
||
m = _TOTAL_COUNT_RE.search(html)
|
||
if m is not None:
|
||
raw = m.group("count")
|
||
# Убрать NBSP (\xa0), -entity и обычные пробелы-разделители разрядов.
|
||
cleaned = raw.replace("\xa0", "").replace(" ", "").replace(" ", "")
|
||
if cleaned:
|
||
try:
|
||
return int(cleaned)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
# Счётчик не найден/не распарсился. Проверяем no-results-маркер: его наличие
|
||
# = валидная пустая выборка (total=0), несмотря на ~50 fallback-объявлений.
|
||
lowered = html.lower()
|
||
if any(marker in lowered for marker in _AVITO_NO_RESULTS_MARKERS):
|
||
return 0
|
||
return None
|
||
|
||
async def _fetch_rooms_page_html(
|
||
self,
|
||
room_slug: str,
|
||
page: int,
|
||
min_price: int | None,
|
||
max_price: int | None,
|
||
) -> str | None:
|
||
"""GET одной room×price SERP-страницы через anti-block pipeline.
|
||
|
||
Делегирует _fetch_serp_html (firewall-detect, IP-rotation, транзиентные
|
||
ретраи). AvitoBlockedError/AvitoRateLimitedError пробрасываются наверх.
|
||
"""
|
||
url = self._build_rooms_url(room_slug, page, min_price, max_price)
|
||
return await self._fetch_serp_html(url, page)
|
||
|
||
async def _probe_total(
|
||
self, room_slug: str, min_price: int | None, max_price: int | None
|
||
) -> tuple[str | None, int | None]:
|
||
"""Probe page=1 бакета и извлекает total, устойчиво к browser-ошибкам.
|
||
|
||
tradein-browser на тяжёлой выдаче (огромный диапазон, напр. студии
|
||
pmax=200M) может таймаутить на goto → HTTP 500 → ``httpx.HTTPStatusError``,
|
||
либо отдавать ``httpx.TimeoutException``/``ReadTimeout``. Раньше эти ошибки
|
||
пробрасывались из probe до самого верха и роняли весь full-load с 0 uniq.
|
||
Теперь они трактуются как «total недоступен» (``None``) — вызывающий
|
||
``_walk_price_range`` решит: split (бакет делим) или skip (узкий бакет).
|
||
|
||
AvitoBlockedError/AvitoRateLimitedError НЕ глушатся — это hard-block,
|
||
должен всплыть в pipeline (mark_banned). Возвращает (html, total); при
|
||
ошибке fetch'а html=None.
|
||
"""
|
||
try:
|
||
html = await self._fetch_rooms_page_html(room_slug, 1, min_price, max_price)
|
||
except (AvitoBlockedError, AvitoRateLimitedError):
|
||
# Hard-block — пробрасываем наверх (pipeline пометит run banned).
|
||
raise
|
||
except (httpx.HTTPStatusError, httpx.TimeoutException, httpx.HTTPError) as exc:
|
||
logger.warning(
|
||
"avito: probe browser error %s for %s [%s, %s] — treating as total=None",
|
||
type(exc).__name__,
|
||
room_slug,
|
||
min_price,
|
||
max_price,
|
||
)
|
||
return None, None
|
||
except Exception:
|
||
logger.warning(
|
||
"avito: probe unexpected error for %s [%s, %s] — treating as total=None",
|
||
room_slug,
|
||
min_price,
|
||
max_price,
|
||
exc_info=True,
|
||
)
|
||
return None, None
|
||
|
||
total = self._extract_total_count(html) if html is not None else None
|
||
return html, total
|
||
|
||
async def fetch_all_secondary(
|
||
self,
|
||
*,
|
||
rooms_buckets: list[tuple[str, str]] | None = None,
|
||
price_cap_per_bucket: int = 1400,
|
||
max_pages_per_bucket: int = 100,
|
||
concurrency: int = 5,
|
||
secondary_only: bool = True,
|
||
on_bucket: Callable[..., Any] | None = None,
|
||
on_progress: Callable[[int], None] | None = None,
|
||
skip_buckets: set[str] | None = None,
|
||
since: date | None = None,
|
||
) -> list[ScrapedLot]:
|
||
"""Exhaustive- ИЛИ инкрементальная загрузка Avito ЕКБ вторички (КОМНАТНОСТЬ × ЦЕНА).
|
||
|
||
Обходит Avito SERP-cap (~5000 результатов на запрос). Вместо корень-бисекции
|
||
[0, _AVITO_MAX_PRICE] на каждую комнатность (≈5 холостых probe-спусков через
|
||
пустой верх, т.к. вторичка ЕКБ почти вся <16М) — стартуем с data-driven
|
||
seed-сетки ценовых брекетов (_AVITO_PRICE_SEED_BRACKETS) по прод-распределению.
|
||
Каждый (комнатность × seed-брекет) дальше дробится _walk_price_range если
|
||
total > price_cap_per_bucket. Последний seed-брекет ОТКРЫТ (hi=None → pmax
|
||
не ставится) — ловит весь хвост люкса без потолка.
|
||
|
||
Страницы leaf-бакета — параллельно (Semaphore). Дедуп по source_id (dict seen),
|
||
ОБЩИЙ по всем брекетам и комнатностям.
|
||
|
||
Параметры:
|
||
rooms_buckets: список (name, slug) комнатностей (default: все ROOM_SLUGS,
|
||
вкл. студии и своб.планировку).
|
||
price_cap_per_bucket: максимум результатов в бакете перед делением (< 5000).
|
||
max_pages_per_bucket: верхний предел страниц на leaf-бакет (default=100).
|
||
concurrency: максимум параллельных page-фетчей в leaf-бакете (default=5).
|
||
secondary_only: если True (default) — отбрасывает новостройки
|
||
(listing_segment=="novostroyki") после парсинга, до save/on_bucket.
|
||
Avito тегирует сегмент по DOM-маркеру item-development-name в _parse_html.
|
||
on_bucket: опциональный callback(bucket_key, list[ScrapedLot]) после каждого
|
||
leaf-бакета. Может быть async или sync. Исключение прерывает прогон.
|
||
on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket).
|
||
skip_buckets: множество ключей «room_label:lo:hi» уже завершённых бакетов —
|
||
пагинация и on_bucket для них пропускаются. Probe-запросы выполняются.
|
||
since: если None (default) — EXHAUSTIVE bisection-обход (поведение без
|
||
изменений). Если задана date — INCREMENTAL: на каждый (комнатность ×
|
||
seed-брекет) последовательная пагинация newest-first с ранней остановкой,
|
||
как только страница уходит ниже watermark `since`. Бисекция по цене НЕ
|
||
выполняется (читаем только верхние свежие страницы, до SERP-cap далеко).
|
||
|
||
Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url).
|
||
"""
|
||
_buckets = rooms_buckets if rooms_buckets is not None else _AVITO_DEFAULT_ROOMS
|
||
seen: dict[str, ScrapedLot] = {}
|
||
# Tolerance к одиночным per-bucket блокам (deep-pagination 429 dense-бакета):
|
||
# скипаем заблокированный бакет и продолжаем; N ПОДРЯД блоков = hard ban → re-raise.
|
||
consecutive_blocked = 0
|
||
skipped_blocked = 0
|
||
|
||
for name, slug in _buckets:
|
||
room_label = _avito_room_label(name)
|
||
logger.info(
|
||
"avito exhaustive: starting %s (price_cap=%d concurrency=%d secondary_only=%s)",
|
||
room_label,
|
||
price_cap_per_bucket,
|
||
concurrency,
|
||
secondary_only,
|
||
)
|
||
before = len(seen)
|
||
# Перебираем seed-брекеты: для ЗАКРЫТЫХ передаём hi = br_hi - 1
|
||
# (pmax эксклюзивный сверху → соседние брекеты не пересекаются), для
|
||
# ОТКРЫТОГО (br_hi is None) — hi=None (без потолка). Дедуп по source_id
|
||
# в общем `seen` страхует на стыках.
|
||
for br_lo, br_hi in _AVITO_PRICE_SEED_BRACKETS:
|
||
walk_hi = br_hi - 1 if br_hi is not None else None
|
||
try:
|
||
if since is not None:
|
||
# INCREMENTAL: последовательная newest-first пагинация с ранней
|
||
# остановкой по watermark `since`; без бисекции по цене.
|
||
await self._paginate_incremental_bracket(
|
||
room_slug=slug,
|
||
room_label=room_label,
|
||
lo=br_lo,
|
||
hi=walk_hi,
|
||
since=since,
|
||
seen=seen,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
)
|
||
else:
|
||
await self._walk_price_range(
|
||
room_slug=slug,
|
||
room_label=room_label,
|
||
lo=br_lo,
|
||
hi=walk_hi,
|
||
seen=seen,
|
||
price_cap_per_bucket=price_cap_per_bucket,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
)
|
||
except (AvitoRateLimitedError, AvitoBlockedError) as exc:
|
||
# Одиночный блок на бакете (deep-pagination 429 dense-бакета) НЕ
|
||
# роняет sweep: скипаем бакет, продолжаем. N ПОДРЯД блоков = реальный
|
||
# hard IP-ban → re-raise (run_avito_full_load → mark_banned).
|
||
consecutive_blocked += 1
|
||
skipped_blocked += 1
|
||
logger.warning(
|
||
"avito exhaustive: %s bracket [%s..%s] skipped — blocked (%s); "
|
||
"consecutive_blocked=%d/%d",
|
||
room_label,
|
||
br_lo,
|
||
walk_hi,
|
||
type(exc).__name__,
|
||
consecutive_blocked,
|
||
_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED,
|
||
)
|
||
if consecutive_blocked >= _AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED:
|
||
logger.error(
|
||
"avito exhaustive: %d consecutive blocked buckets — "
|
||
"treating as hard IP-ban, aborting sweep",
|
||
consecutive_blocked,
|
||
)
|
||
raise
|
||
continue
|
||
# Успешный бакет — сбрасываем счётчик подряд-блоков.
|
||
consecutive_blocked = 0
|
||
room_collected = len(seen) - before
|
||
logger.info(
|
||
"avito exhaustive: %s done — collected %d (total unique=%d)",
|
||
room_label,
|
||
room_collected,
|
||
len(seen),
|
||
)
|
||
if on_progress is not None:
|
||
on_progress(len(seen))
|
||
|
||
logger.info(
|
||
"avito exhaustive: DONE — total unique=%d lots skipped_blocked=%d",
|
||
len(seen),
|
||
skipped_blocked,
|
||
)
|
||
return list(seen.values())
|
||
|
||
async def _walk_price_range(
|
||
self,
|
||
*,
|
||
room_slug: str,
|
||
room_label: str,
|
||
lo: int,
|
||
hi: int | None,
|
||
seen: dict[str, ScrapedLot],
|
||
price_cap_per_bucket: int,
|
||
max_pages_per_bucket: int,
|
||
concurrency: int = 5,
|
||
secondary_only: bool = True,
|
||
on_bucket: Callable[..., Any] | None = None,
|
||
skip_buckets: set[str] | None = None,
|
||
_depth: int = 0,
|
||
) -> None:
|
||
"""Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].
|
||
|
||
Алгоритм для ЗАКРЫТОГО брекета (hi is not None):
|
||
1. Запросить page=1 с pmin=lo, pmax=hi → total (page-title/count).
|
||
2. Если total <= cap → пагинировать leaf-бакет параллельно (concurrency).
|
||
3. Если total > cap → разбить бакет пополам (рекурсия).
|
||
Guard: hi - lo < _AVITO_MIN_BRACKET → пагинировать как есть (WARNING).
|
||
|
||
ОТКРЫТЫЙ брекет (hi is None) — верхний seed-брекет без потолка (pmax не
|
||
ставится, avito отдаёт всё ≥lo). Делить нельзя (mid=(lo+hi)//2 невозможен) —
|
||
probe + пагинируем leaf напрямую. Хвост 250М+ крошечный (≈73), за cap
|
||
обычно не выходит; если вышел — пагинируем до max_pages + WARNING (acceptable).
|
||
|
||
После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots).
|
||
bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый).
|
||
on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон.
|
||
skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются.
|
||
secondary_only: новостройки (listing_segment=="novostroyki") отбрасываются после
|
||
сбора bucket_lots, до дедупа в seen и вызова on_bucket.
|
||
AvitoBlockedError/AvitoRateLimitedError из _fetch_serp_html пробрасываются наверх.
|
||
"""
|
||
_lo_param = lo if lo > 0 else None
|
||
_hi_repr = "open" if hi is None else str(hi)
|
||
|
||
# ── Шаг 1: probe page 1 ────────────────────────────────────────────────
|
||
# _probe_total ловит browser-ошибки (HTTP 500 / timeout от tradein-browser
|
||
# на тяжёлой выдаче, напр. студии pmax=200M) → total=None вместо краха.
|
||
# AvitoBlockedError/AvitoRateLimitedError всплывают наверх (mark_banned).
|
||
html, total = await self._probe_total(room_slug, _lo_param, hi)
|
||
|
||
# Ретрай на captcha/missing-count/browser-ошибке: rotate IP + 1 retry.
|
||
if total is None:
|
||
logger.warning(
|
||
"avito: total=None for %s [%d, %s] depth=%d — rotating IP + retry",
|
||
room_label,
|
||
lo,
|
||
_hi_repr,
|
||
_depth,
|
||
)
|
||
rotated = await self._rotate_ip()
|
||
if rotated:
|
||
html, total = await self._probe_total(room_slug, _lo_param, hi)
|
||
|
||
if total is None:
|
||
# ОТКРЫТЫЙ брекет: делить нельзя (нет hi). Probe failed → нет total для
|
||
# ceil(); пагинируем до max_pages_per_bucket (best-effort, хвост 250М+
|
||
# крошечный) и WARNING. НЕ роняем прогон.
|
||
if hi is None:
|
||
logger.warning(
|
||
"avito: probe failed for OPEN bucket %s [%d, open] depth=%d — "
|
||
"paginating up to max_pages (tail-loss accepted)",
|
||
room_label,
|
||
lo,
|
||
_depth,
|
||
)
|
||
await self._paginate_leaf_bucket(
|
||
room_slug=room_slug,
|
||
room_label=room_label,
|
||
lo=lo,
|
||
hi=None,
|
||
html=None,
|
||
max_pages=max_pages_per_bucket,
|
||
seen=seen,
|
||
price_cap_per_bucket=price_cap_per_bucket,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
)
|
||
return
|
||
bracket_size = hi - lo
|
||
# Probe не удался даже после ретрая. Если бакет ещё делим — НЕ теряем
|
||
# его целиком: тяжёлая страница (огромная выдача) скорее всего станет
|
||
# загружаемой в более узком ценовом диапазоне → принудительный split.
|
||
if bracket_size >= _AVITO_MIN_BRACKET:
|
||
logger.warning(
|
||
"avito: probe failed for %s [%d, %d] depth=%d — splitting (assume heavy)",
|
||
room_label,
|
||
lo,
|
||
hi,
|
||
_depth,
|
||
)
|
||
mid = (lo + hi) // 2
|
||
await self._walk_price_range(
|
||
room_slug=room_slug,
|
||
room_label=room_label,
|
||
lo=lo,
|
||
hi=mid,
|
||
seen=seen,
|
||
price_cap_per_bucket=price_cap_per_bucket,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
_depth=_depth + 1,
|
||
)
|
||
await self._walk_price_range(
|
||
room_slug=room_slug,
|
||
room_label=room_label,
|
||
lo=mid + 1,
|
||
hi=hi,
|
||
seen=seen,
|
||
price_cap_per_bucket=price_cap_per_bucket,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
_depth=_depth + 1,
|
||
)
|
||
return
|
||
# Бакет уже узкий (< MIN_BRACKET) — split не поможет, хвост теряем
|
||
# осознанно (как cian) и логируем.
|
||
logger.error(
|
||
"avito: skipping bucket %s [%d, %d] — total unavailable after retry "
|
||
"(bracket=%d < MIN_BRACKET=%d, tail loss accepted)",
|
||
room_label,
|
||
lo,
|
||
hi,
|
||
bracket_size,
|
||
_AVITO_MIN_BRACKET,
|
||
)
|
||
return
|
||
|
||
logger.info(
|
||
"avito: %s [%d, %s] total=%d depth=%d",
|
||
room_label,
|
||
lo,
|
||
_hi_repr,
|
||
total,
|
||
_depth,
|
||
)
|
||
|
||
if total == 0:
|
||
return
|
||
|
||
# ── ОТКРЫТЫЙ брекет (hi is None): делить нельзя — пагинируем leaf напрямую ─
|
||
# total известен из probe → max_pages = min(ceil(total/per_page), cap).
|
||
# Если total > cap (не должно для 250М+, но guard) — пагинируем как есть до
|
||
# max_pages_per_bucket с WARNING (хвост люкса крошечный, tail-loss accepted).
|
||
if hi is None:
|
||
if total > price_cap_per_bucket:
|
||
logger.warning(
|
||
"avito: OPEN bucket %s [%d, open] total=%d > cap=%d — paginating "
|
||
"without split (tail-loss accepted, lux tail tiny)",
|
||
room_label,
|
||
lo,
|
||
total,
|
||
price_cap_per_bucket,
|
||
)
|
||
pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE)
|
||
max_pages = min(pages_needed, max_pages_per_bucket)
|
||
await self._paginate_leaf_bucket(
|
||
room_slug=room_slug,
|
||
room_label=room_label,
|
||
lo=lo,
|
||
hi=None,
|
||
html=html,
|
||
max_pages=max_pages,
|
||
seen=seen,
|
||
price_cap_per_bucket=price_cap_per_bucket,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
)
|
||
return
|
||
|
||
# ── Шаг 2: деление или пагинация ──────────────────────────────────────
|
||
bracket_size = hi - lo
|
||
need_split = total > price_cap_per_bucket
|
||
too_narrow = bracket_size < _AVITO_MIN_BRACKET
|
||
|
||
if need_split and too_narrow:
|
||
logger.warning(
|
||
"avito: %s [%d, %d] total=%d > cap=%d but bracket=%d < MIN_BRACKET=%d "
|
||
"— paginating as-is (tail loss ~%d)",
|
||
room_label,
|
||
lo,
|
||
hi,
|
||
total,
|
||
price_cap_per_bucket,
|
||
bracket_size,
|
||
_AVITO_MIN_BRACKET,
|
||
max(0, total - price_cap_per_bucket),
|
||
)
|
||
need_split = False # принудительно пагинируем
|
||
|
||
if need_split:
|
||
mid = (lo + hi) // 2
|
||
# [lo, mid]
|
||
await self._walk_price_range(
|
||
room_slug=room_slug,
|
||
room_label=room_label,
|
||
lo=lo,
|
||
hi=mid,
|
||
seen=seen,
|
||
price_cap_per_bucket=price_cap_per_bucket,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
_depth=_depth + 1,
|
||
)
|
||
# [mid+1, hi]
|
||
await self._walk_price_range(
|
||
room_slug=room_slug,
|
||
room_label=room_label,
|
||
lo=mid + 1,
|
||
hi=hi,
|
||
seen=seen,
|
||
price_cap_per_bucket=price_cap_per_bucket,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
_depth=_depth + 1,
|
||
)
|
||
return
|
||
|
||
# ── Закрытый leaf-бакет: считаем max_pages + warning, делегируем helper'у ─
|
||
pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE)
|
||
max_pages = min(pages_needed, max_pages_per_bucket)
|
||
|
||
if pages_needed > max_pages_per_bucket:
|
||
tail_loss = total - max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE
|
||
logger.warning(
|
||
"avito: %s [%d, %d] total=%d exceeds page cap "
|
||
"(max_pages=%d × %d=%d offers) — tail loss ~%d offers",
|
||
room_label,
|
||
lo,
|
||
hi,
|
||
total,
|
||
max_pages_per_bucket,
|
||
_AVITO_OFFERS_PER_PAGE,
|
||
max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE,
|
||
tail_loss,
|
||
)
|
||
|
||
await self._paginate_leaf_bucket(
|
||
room_slug=room_slug,
|
||
room_label=room_label,
|
||
lo=lo,
|
||
hi=hi,
|
||
html=html,
|
||
max_pages=max_pages,
|
||
seen=seen,
|
||
price_cap_per_bucket=price_cap_per_bucket,
|
||
max_pages_per_bucket=max_pages_per_bucket,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
)
|
||
|
||
async def _paginate_leaf_bucket(
|
||
self,
|
||
*,
|
||
room_slug: str,
|
||
room_label: str,
|
||
lo: int,
|
||
hi: int | None,
|
||
html: str | None,
|
||
max_pages: int,
|
||
seen: dict[str, ScrapedLot],
|
||
price_cap_per_bucket: int,
|
||
max_pages_per_bucket: int,
|
||
concurrency: int,
|
||
secondary_only: bool,
|
||
on_bucket: Callable[..., Any] | None,
|
||
skip_buckets: set[str] | None,
|
||
) -> None:
|
||
"""Параллельная пагинация одного leaf-бакета + фильтр + дедуп + on_bucket.
|
||
|
||
Общая для ЗАКРЫТЫХ ([lo, hi]) и ОТКРЫТОГО (hi=None → pmax не ставится)
|
||
брекетов. ``html`` — HTML страницы 1 из probe (переиспользуем, не
|
||
перезапрашиваем); для открытого-без-probe передаётся None → стр.1 фетчится
|
||
наравне с остальными. ``max_pages`` уже посчитан вызывающим.
|
||
|
||
bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый).
|
||
skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются.
|
||
AvitoBlockedError/AvitoRateLimitedError из page-фетчей пробрасываются наверх.
|
||
"""
|
||
_lo_param = lo if lo > 0 else None
|
||
_hi_param = hi # None → _build_rooms_url не ставит pmax
|
||
_hi_repr = "open" if hi is None else str(hi)
|
||
bucket_key = f"{room_label}:{lo}:{_hi_repr}"
|
||
|
||
# Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket.
|
||
if skip_buckets and bucket_key in skip_buckets:
|
||
logger.info("avito: skip bucket %s — already done (resume)", bucket_key)
|
||
return
|
||
|
||
# Страница 1 — из probe-html если он есть (закрытый/открытый с probe);
|
||
# иначе фетчим её как обычную страницу (открытый брекет без probe-html).
|
||
sem = asyncio.Semaphore(concurrency)
|
||
first_url = self._build_rooms_url(room_slug, 1, _lo_param, _hi_param)
|
||
|
||
async def _one_page(p: int) -> list[ScrapedLot]:
|
||
if p == 1 and html is not None:
|
||
return self._parse_html(html, source_url_base=first_url)
|
||
async with sem:
|
||
page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, _hi_param)
|
||
await asyncio.sleep(self.request_delay_sec)
|
||
if page_html is None:
|
||
logger.warning(
|
||
"avito: page_html=None %s [%d, %s] page=%d — skipping page",
|
||
room_label,
|
||
lo,
|
||
_hi_repr,
|
||
p,
|
||
)
|
||
return []
|
||
page_url = self._build_rooms_url(room_slug, p, _lo_param, _hi_param)
|
||
return self._parse_html(page_html, source_url_base=page_url)
|
||
|
||
page_results = await asyncio.gather(
|
||
*[_one_page(p) for p in range(1, max_pages + 1)],
|
||
return_exceptions=True,
|
||
)
|
||
|
||
bucket_lots: list[ScrapedLot] = []
|
||
for p_idx, res in enumerate(page_results, start=1):
|
||
if isinstance(res, BaseException):
|
||
# Блокировки пробрасываем наверх (mark_banned в pipeline-обёртке).
|
||
if isinstance(res, AvitoBlockedError | AvitoRateLimitedError):
|
||
raise res
|
||
logger.warning(
|
||
"avito: page exception %s [%d, %s] page=%d — %r",
|
||
room_label,
|
||
lo,
|
||
_hi_repr,
|
||
p_idx,
|
||
res,
|
||
)
|
||
else:
|
||
bucket_lots.extend(res)
|
||
|
||
collected_this_bucket = len(bucket_lots)
|
||
|
||
# ── Фильтр новостроек (secondary_only) ────────────────────────────────
|
||
dropped_nb = 0
|
||
if secondary_only:
|
||
filtered = [lot for lot in bucket_lots if lot.listing_segment != "novostroyki"]
|
||
dropped_nb = collected_this_bucket - len(filtered)
|
||
bucket_lots = filtered
|
||
|
||
# Дедуп в общий seen
|
||
for lot in bucket_lots:
|
||
key = lot.source_id or lot.source_url
|
||
if key:
|
||
seen[key] = lot
|
||
|
||
logger.info(
|
||
"avito: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d",
|
||
room_label,
|
||
lo,
|
||
_hi_repr,
|
||
max_pages,
|
||
collected_this_bucket,
|
||
dropped_nb,
|
||
len(seen),
|
||
)
|
||
|
||
# ── on_bucket callback: инкрементальный save ──────────────────────────
|
||
if on_bucket is not None and bucket_lots:
|
||
res_cb = on_bucket(bucket_key, bucket_lots)
|
||
if inspect.isawaitable(res_cb):
|
||
await res_cb
|
||
|
||
async def _paginate_incremental_bracket(
|
||
self,
|
||
*,
|
||
room_slug: str,
|
||
room_label: str,
|
||
lo: int,
|
||
hi: int | None,
|
||
since: date,
|
||
seen: dict[str, ScrapedLot],
|
||
max_pages_per_bucket: int,
|
||
secondary_only: bool,
|
||
on_bucket: Callable[..., Any] | None,
|
||
) -> None:
|
||
"""INCREMENTAL пагинация одного (комнатность × seed-брекет) с ранней остановкой.
|
||
|
||
SERP отсортирован newest-first (s=104). Идём ПОСЛЕДОВАТЕЛЬНО со страницы 1;
|
||
как только страница не содержит ни одной карточки с listing_date >= since —
|
||
весь остаток бакета старше watermark → останавливаемся. Бисекция по цене НЕ
|
||
выполняется: читаем только верхние свежие страницы, до SERP-cap (~5000) далеко.
|
||
|
||
Early-stop semantics:
|
||
- На каждой странице берём max(listing_date) среди датированных карточек.
|
||
Если ни одна карточка не имеет listing_date >= since → страница «старая».
|
||
- Карточки с listing_date is None НЕ триггерят стоп сами по себе (трактуем
|
||
как «продолжить», чтобы не потерять свежие-но-недатированные карточки).
|
||
- Но страница, целиком состоящая из None-или-старых карточек, считается
|
||
«не-свежей»; 2 ПОДРЯД не-свежие страницы → стоп (grace против runaway,
|
||
когда весь хвост недатирован).
|
||
- page_html is None или parse == [] → конец пагинации, стоп.
|
||
|
||
bucket_key, secondary_only-фильтр и дедуп в seen — идентичны _paginate_leaf_bucket.
|
||
on_bucket вызывается один раз для собранного бакета (async/sync-aware).
|
||
AvitoBlockedError/AvitoRateLimitedError из page-фетчей пробрасываются наверх.
|
||
"""
|
||
_lo_param = lo if lo > 0 else None
|
||
_hi_param = hi # None → _build_rooms_url не ставит pmax
|
||
_hi_repr = "open" if hi is None else str(hi)
|
||
bucket_key = f"{room_label}:{lo}:{_hi_repr}"
|
||
|
||
bucket_lots: list[ScrapedLot] = []
|
||
pages_fetched = 0
|
||
not_fresh_streak = 0 # подряд идущие не-свежие (all None-или-старые) страницы
|
||
stop_reason = "max-pages" # перетирается ниже на реальную причину
|
||
|
||
for p in range(1, max_pages_per_bucket + 1):
|
||
# Последовательный фетч (НЕ asyncio.gather): early-stop требует читать
|
||
# страницы по порядку. Тот же helper, что и в _paginate_leaf_bucket.
|
||
page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, _hi_param)
|
||
await asyncio.sleep(self.request_delay_sec)
|
||
pages_fetched += 1
|
||
|
||
if page_html is None:
|
||
stop_reason = "end-of-pages (page_html=None)"
|
||
break
|
||
|
||
page_url = self._build_rooms_url(room_slug, p, _lo_param, _hi_param)
|
||
page_lots = self._parse_html(page_html, source_url_base=page_url)
|
||
if not page_lots:
|
||
stop_reason = "end-of-pages (empty parse)"
|
||
break
|
||
|
||
bucket_lots.extend(page_lots)
|
||
|
||
# Свежесть страницы: есть ли хоть одна карточка с датой >= since.
|
||
dated = [lot.listing_date for lot in page_lots if lot.listing_date is not None]
|
||
has_fresh = any(d >= since for d in dated)
|
||
if has_fresh:
|
||
not_fresh_streak = 0
|
||
continue
|
||
|
||
# Страница не содержит свежих датированных карточек.
|
||
if dated:
|
||
# Есть даты, но все < since → newest-first гарантирует, что дальше
|
||
# только старее → стоп немедленно.
|
||
stop_reason = "early-stop (page all older than since)"
|
||
break
|
||
# Все карточки undated (None) → не стопим сразу (могут быть свежие без
|
||
# даты), но копим streak; 2 подряд недатированные страницы → grace-стоп.
|
||
not_fresh_streak += 1
|
||
if not_fresh_streak >= 2:
|
||
stop_reason = "early-stop (2 consecutive all-undated pages, grace)"
|
||
break
|
||
|
||
collected_this_bucket = len(bucket_lots)
|
||
|
||
# ── Фильтр новостроек (secondary_only) — идентично _paginate_leaf_bucket ──
|
||
dropped_nb = 0
|
||
if secondary_only:
|
||
filtered = [lot for lot in bucket_lots if lot.listing_segment != "novostroyki"]
|
||
dropped_nb = collected_this_bucket - len(filtered)
|
||
bucket_lots = filtered
|
||
|
||
# Дедуп в общий seen
|
||
for lot in bucket_lots:
|
||
key = lot.source_id or lot.source_url
|
||
if key:
|
||
seen[key] = lot
|
||
|
||
logger.info(
|
||
"avito: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d "
|
||
"unique_total=%d incremental since=%s stop=%s",
|
||
room_label,
|
||
lo,
|
||
_hi_repr,
|
||
pages_fetched,
|
||
collected_this_bucket,
|
||
dropped_nb,
|
||
len(seen),
|
||
since.isoformat(),
|
||
stop_reason,
|
||
)
|
||
|
||
# ── on_bucket callback: инкрементальный save ──────────────────────────
|
||
if on_bucket is not None and bucket_lots:
|
||
res_cb = on_bucket(bucket_key, bucket_lots)
|
||
if inspect.isawaitable(res_cb):
|
||
await res_cb
|
||
|
||
# ── T6: citywide / byrooms fetch methods ──────────────────────────────────
|
||
|
||
async def fetch_city_wide(
|
||
self,
|
||
pages: int = 100,
|
||
*,
|
||
delay_override_sec: float | None = None,
|
||
) -> list[ScrapedLot]:
|
||
"""T6: Обход всего ЕКБ без geo-фильтра (citywide-mode), paginated.
|
||
|
||
Возвращает дедуплицированный список лотов (по source_id/source_url).
|
||
Break-on-empty: останавливается когда страница отдаёт 0 карточек.
|
||
Сохраняет весь anti-block pipeline (_fetch_serp_html: firewall-detect,
|
||
IP rotation, транзиентные ретраи). Не трогает fetch_around.
|
||
|
||
Args:
|
||
pages: максимальное число страниц (default 100 = Avito cap ~5000 лотов).
|
||
delay_override_sec: если задан — переопределяет request_delay_sec для
|
||
этого вызова.
|
||
|
||
Returns:
|
||
Список ScrapedLot (все страницы, дедуп по source_id).
|
||
"""
|
||
all_lots = await self._paginate_sweep(
|
||
pages,
|
||
self._build_citywide_url,
|
||
label="citywide",
|
||
delay_override_sec=delay_override_sec,
|
||
)
|
||
logger.info("avito fetch_city_wide pages=%d total_lots=%d", pages, len(all_lots))
|
||
return all_lots
|
||
|
||
async def _paginate_sweep(
|
||
self,
|
||
pages: int,
|
||
url_builder: Callable[[int], str],
|
||
*,
|
||
label: str,
|
||
delay_override_sec: float | None = None,
|
||
) -> list[ScrapedLot]:
|
||
"""Общий paginated-обход ЕКБ (citywide / novostroyka) с break-on-empty.
|
||
|
||
url_builder(page) — функция построения URL страницы (citywide или
|
||
novostroyka). Сохраняет anti-block pipeline (_fetch_serp_html: firewall,
|
||
IP rotation, ретраи), дедуп по source_id, break-on-empty. label — только
|
||
для логов. Не меняет наблюдаемое поведение fetch_city_wide.
|
||
"""
|
||
if delay_override_sec is not None:
|
||
self.request_delay_sec = delay_override_sec
|
||
|
||
all_lots: list[ScrapedLot] = []
|
||
seen_ids: set[str] = set()
|
||
|
||
for page in range(1, pages + 1):
|
||
url = url_builder(page)
|
||
try:
|
||
html = await self._fetch_serp_html_with_retry(url, page)
|
||
except (AvitoBlockedError, AvitoRateLimitedError):
|
||
raise
|
||
except Exception:
|
||
logger.exception(
|
||
"avito %s page=%d fetch failed after retries url=%s", label, page, url
|
||
)
|
||
break
|
||
if html is None:
|
||
logger.info("avito %s page=%d: non-200 — end of pagination", label, page)
|
||
break
|
||
|
||
lots = self._parse_html(html, source_url_base=url)
|
||
if not lots:
|
||
logger.info("avito %s page=%d: 0 lots — end of pagination", label, page)
|
||
break
|
||
|
||
new_lots = [lot for lot in lots if lot.source_id not in seen_ids]
|
||
for lot in new_lots:
|
||
if lot.source_id:
|
||
seen_ids.add(lot.source_id)
|
||
all_lots.extend(new_lots)
|
||
logger.info(
|
||
"avito %s page=%d: %d lots (%d new, %d dedup'd, total=%d)",
|
||
label,
|
||
page,
|
||
len(lots),
|
||
len(new_lots),
|
||
len(lots) - len(new_lots),
|
||
len(all_lots),
|
||
)
|
||
if page < pages:
|
||
await self.sleep_between_requests()
|
||
return all_lots
|
||
|
||
async def fetch_newbuildings(
|
||
self,
|
||
pages: int = 30,
|
||
*,
|
||
delay_override_sec: float | None = None,
|
||
) -> list[ScrapedLot]:
|
||
"""Обход ЕКБ-выборки только новостроек (novostroyka-filter), paginated.
|
||
|
||
Avito отдаёт dedicated SERP с фильтром «Новостройка» — 100% new-build
|
||
карточек (с маркером застройщика → listing_segment="novostroyki",
|
||
newbuilding_id/newbuilding_url заполнены). Citywide (без geo/anchor).
|
||
Возвращает дедуплицированный список (по source_id), break-on-empty.
|
||
Сохраняет весь anti-block pipeline. Не трогает fetch_city_wide/fetch_around.
|
||
|
||
Args:
|
||
pages: максимальное число страниц (default 30).
|
||
delay_override_sec: если задан — переопределяет request_delay_sec для
|
||
этого вызова.
|
||
|
||
Returns:
|
||
Список ScrapedLot новостроек (все страницы, дедуп по source_id).
|
||
"""
|
||
all_lots = await self._paginate_sweep(
|
||
pages,
|
||
self._build_newbuilding_url,
|
||
label="newbuilding",
|
||
delay_override_sec=delay_override_sec,
|
||
)
|
||
logger.info("avito fetch_newbuildings pages=%d total_lots=%d", pages, len(all_lots))
|
||
return all_lots
|
||
|
||
async def fetch_by_rooms(
|
||
self,
|
||
pages: int = 100,
|
||
*,
|
||
delay_override_sec: float | None = None,
|
||
room_slugs: list[tuple[str, str]] | None = None,
|
||
) -> list[ScrapedLot]:
|
||
"""T6: Обход ЕКБ по категориям комнатности (byrooms-mode).
|
||
|
||
Дробление по 7 категориям комнатности позволяет обойти Avito cap ~5000
|
||
результатов на поиск: каждый сегмент < 5000. Итого до 7 × 5000 = 35k лотов.
|
||
Break-on-empty на каждой категории. Глобальный dedup по source_id.
|
||
|
||
Args:
|
||
pages: страниц на категорию (default 100).
|
||
delay_override_sec: переопределяет request_delay_sec.
|
||
room_slugs: список (name, slug) для обхода; по умолчанию ROOM_SLUGS
|
||
(все 7 категорий). Передай подмножество для частичного обхода.
|
||
|
||
Returns:
|
||
Список ScrapedLot (все категории, глобальный дедуп по source_id).
|
||
"""
|
||
if delay_override_sec is not None:
|
||
self.request_delay_sec = delay_override_sec
|
||
|
||
slugs = room_slugs if room_slugs is not None else ROOM_SLUGS
|
||
all_lots: list[ScrapedLot] = []
|
||
seen_ids: set[str] = set()
|
||
|
||
for ci, (name, slug) in enumerate(slugs, 1):
|
||
logger.info(
|
||
"avito byrooms [%d/%d] category=%s slug_prefix=%.30s",
|
||
ci,
|
||
len(slugs),
|
||
name,
|
||
slug,
|
||
)
|
||
cat_count = 0
|
||
for page in range(1, pages + 1):
|
||
url = self._build_rooms_url(slug, page)
|
||
try:
|
||
html = await self._fetch_serp_html_with_retry(url, page)
|
||
except (AvitoBlockedError, AvitoRateLimitedError):
|
||
raise
|
||
except Exception:
|
||
logger.exception(
|
||
"avito byrooms category=%s page=%d fetch failed after retries url=%s",
|
||
name,
|
||
page,
|
||
url,
|
||
)
|
||
break
|
||
if html is None:
|
||
logger.info("avito byrooms category=%s page=%d: non-200 — end", name, page)
|
||
break
|
||
|
||
lots = self._parse_html(html, source_url_base=url)
|
||
if not lots:
|
||
logger.info(
|
||
"avito byrooms category=%s page=%d: 0 lots — end of category",
|
||
name,
|
||
page,
|
||
)
|
||
break
|
||
|
||
new_lots = [lot for lot in lots if lot.source_id not in seen_ids]
|
||
for lot in new_lots:
|
||
if lot.source_id:
|
||
seen_ids.add(lot.source_id)
|
||
all_lots.extend(new_lots)
|
||
cat_count += len(new_lots)
|
||
logger.info(
|
||
"avito byrooms category=%s page=%d: %d lots (%d new, total=%d)",
|
||
name,
|
||
page,
|
||
len(lots),
|
||
len(new_lots),
|
||
len(all_lots),
|
||
)
|
||
if page < pages:
|
||
await self.sleep_between_requests()
|
||
|
||
logger.info(
|
||
"avito byrooms category=%s done: cat_new=%d grand_total=%d",
|
||
name,
|
||
cat_count,
|
||
len(all_lots),
|
||
)
|
||
|
||
logger.info("avito fetch_by_rooms pages_per_cat=%d total_lots=%d", pages, len(all_lots))
|
||
return all_lots
|
||
|
||
def _parse_html(self, html: str, source_url_base: str) -> list[ScrapedLot]:
|
||
"""Парсим карточки объявлений из HTML через DOM scrape.
|
||
|
||
Поля карточки берём из DOM (`data-marker="item-*"`). Дату публикации — из
|
||
embedded per-item JSON (`sortTimeStamp`), т.к. DOM `item-date` рендерится
|
||
лениво только у верхних карточек (#726); DOM-marker остаётся fallback.
|
||
"""
|
||
tree = HTMLParser(html)
|
||
ts_map = _build_sort_timestamp_map(html)
|
||
cards = tree.css('[data-marker="item"]')
|
||
lots: list[ScrapedLot] = []
|
||
for card in cards:
|
||
lot = self._dom_card_to_lot(card, source_url_base, ts_map)
|
||
if lot is not None:
|
||
lots.append(lot)
|
||
return lots
|
||
|
||
def _dom_card_to_lot(
|
||
self, card: Any, source_url_base: str, ts_map: dict[str, date] | None = None
|
||
) -> 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)
|
||
|
||
# Адрес: первый <p> внутри 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 = _clean_address(street_el.text(strip=True))
|
||
|
||
# Фото
|
||
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
|
||
|
||
# Дата публикации (#726): primary — sortTimeStamp из per-item JSON по
|
||
# data-item-id (DOM item-date рендерится лениво только у верхних карточек),
|
||
# fallback — DOM marker item-date/item-date-info + relative-date parse.
|
||
listing_date: date | None = None
|
||
item_id = card.attributes.get("data-item-id")
|
||
if ts_map and item_id:
|
||
listing_date = ts_map.get(item_id)
|
||
if listing_date is None:
|
||
date_el = card.css_first('[data-marker="item-date"]')
|
||
if date_el is None:
|
||
date_el = card.css_first('[data-marker="item-date-info"]')
|
||
if date_el is not None:
|
||
listing_date = _parse_relative_date(date_el.text(strip=True))
|
||
if listing_date is None:
|
||
# Fallback: ищем любой текст «N дней/недель/месяцев назад» в карточке
|
||
card_text = card.text(strip=True)
|
||
listing_date = _parse_relative_date(card_text)
|
||
|
||
# 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 по DOM-маркеру карточки: новостройки несут
|
||
# data-marker="item-development-name" (название ЖК/застройщика),
|
||
# вторичка — нет. URL-паттерн ненадёжен: каждый avito-URL квартиры
|
||
# содержит /kvartiry/ (вкл. новостройки) → ветка всегда vtorichka.
|
||
dev_name = card.css_first('[data-marker="item-development-name"]')
|
||
listing_segment = "novostroyki" if dev_name is not None else "vtorichka"
|
||
|
||
# Newbuilding (ЖК) link: новостройки несут якорь на ЖК-сабдомен
|
||
# <a href="https://zhk-<slug>-ekaterinburg.avito.ru?context=...">.
|
||
# newbuilding_url — host+path без ?context (mirror urlparse-cleaning),
|
||
# newbuilding_id — <slug> = host без 'zhk-' префикса и '.avito.ru' суффикса.
|
||
newbuilding_id: str | None = None
|
||
newbuilding_url: str | None = None
|
||
if dev_name is not None:
|
||
nb_link = card.css_first('a[href*="zhk-"][href*=".avito.ru"]')
|
||
if nb_link is not None:
|
||
nb_href = nb_link.attributes.get("href", "") or ""
|
||
parsed = urlparse(nb_href)
|
||
host = parsed.netloc
|
||
if host.startswith("zhk-") and host.endswith(".avito.ru"):
|
||
# strip query/fragment, keep scheme+host+path
|
||
newbuilding_url = urlunparse((parsed.scheme, host, parsed.path, "", "", ""))
|
||
newbuilding_id = host[len("zhk-") : -len(".avito.ru")]
|
||
|
||
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,
|
||
listing_date=listing_date,
|
||
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,
|
||
newbuilding_id=newbuilding_id,
|
||
newbuilding_url=newbuilding_url,
|
||
)
|
||
except Exception:
|
||
# #823: логируем каждую сломанную карточку и считаем для observability.
|
||
# Карточки с неожиданной DOM-структурой скипаем, но не роняем весь парс.
|
||
self.parse_failures += 1
|
||
item_id_attr = getattr(card, "attributes", {}).get("data-item-id", "?")
|
||
logger.warning(
|
||
"avito _dom_card_to_lot: failed to parse card item_id=%s (parse_failures=%d)",
|
||
item_id_attr,
|
||
self.parse_failures,
|
||
exc_info=True,
|
||
)
|
||
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)
|
||
|
||
# ── T6: citywide / byrooms sweep ──────────────────────────────────────────────
|
||
# Avito URL slugs для room-filter (извлечены из реальных search-страниц 2026-05-23).
|
||
# Каждый slug кодирует ASgB-параметры выборки комнатности.
|
||
ROOM_SLUGS: list[tuple[str, str]] = [
|
||
("студии", "studii-ASgBAgICAkSSA8YQygj~WA"),
|
||
("1-комн.", "1-komnatnye-ASgBAgICAkSSA8YQygiAWQ"),
|
||
("2-комн.", "2-komnatnye-ASgBAgICAkSSA8YQygiCWQ"),
|
||
("3-комн.", "3-komnatnye-ASgBAgICAkSSA8YQygiEWQ"),
|
||
("4-комн.", "4-komnatnye-ASgBAgICAkSSA8YQygiGWQ"),
|
||
("5+комн.", "5-komnatnye-ASgBAgICAkSSA8YQygiIWQ"),
|
||
("своб.планир.", "svobodnaya_planirovka-ASgBAgICAkSSA8YQygj8zzI"),
|
||
]
|
||
|
||
# Дефолтные комнатности для exhaustive full-load — все ROOM_SLUGS (вкл. студии и
|
||
# своб.планировку), т.к. каждая категория < SERP-cap после room×price бисекции.
|
||
_AVITO_DEFAULT_ROOMS: list[tuple[str, str]] = list(ROOM_SLUGS)
|
||
|
||
_NON_SLUG_CHARS_RE = re.compile(r"[^a-z0-9]+")
|
||
|
||
|
||
def _avito_room_label(name: str) -> str:
|
||
"""Стабильный bucket_key-префикс из человекочитаемого имени комнатности.
|
||
|
||
«1-комн.» → «room_1_komn», «студии» → «room_studii», «5+комн.» → «room_5_komn».
|
||
Транслитерация не нужна — берём латиницу/цифры из slug-friendly формы; для
|
||
кириллицы используем порядковый индекс категории как fallback стабильности.
|
||
"""
|
||
# Цифры из имени (комнатность) + транслит ключевых маркеров для читабельности логов.
|
||
digits = "".join(ch for ch in name if ch.isdigit())
|
||
base = name.lower()
|
||
if "студ" in base:
|
||
return "room_studii"
|
||
if "планир" in base:
|
||
return "room_svobodnaya"
|
||
if digits:
|
||
return f"room_{digits}_komn"
|
||
# Fallback: нормализуем что есть (на случай новых категорий).
|
||
norm = _NON_SLUG_CHARS_RE.sub("_", base).strip("_")
|
||
return f"room_{norm or 'unknown'}"
|
||
|
||
|
||
_CSS_NOISE_RE = re.compile(r"\.?css-[a-z0-9_-]+\s*\{[^}]*\}", flags=re.I)
|
||
_NOT_ADDRESS_TAIL_RE = re.compile(
|
||
r"\s*(Площадь \d|от \d+\s?мин\.|css-[a-z0-9_-]+)",
|
||
flags=re.I,
|
||
)
|
||
|
||
|
||
def _clean_address(raw: str | None) -> str | None:
|
||
"""Strip Emotion CSS and post-address noise that Avito DOM leaks via .text().
|
||
|
||
Examples:
|
||
"ул. Токарей, 56к1Площадь 1905 года.css-39hgr0{fill:..."
|
||
-> "ул. Токарей, 56к1"
|
||
"ул. Малышева, 1.css-xxx{...}"
|
||
-> "ул. Малышева, 1"
|
||
"""
|
||
if not raw:
|
||
return None
|
||
cleaned = _CSS_NOISE_RE.sub("", raw)
|
||
cleaned = _NOT_ADDRESS_TAIL_RE.split(cleaned, maxsplit=1)[0]
|
||
cleaned = cleaned.strip(" ,.\n\t")
|
||
return cleaned or None
|
||
|
||
|
||
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
|