[^<]*)<',
+ 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,
+ config: ScraperConfig,
+ *,
+ delay_provider: Callable[[str], float] | None = None,
+ ) -> None:
+ super().__init__()
+ # Strangler-инжекция (#2133): конфиг и провайдер задержки приходят снаружи
+ # вместо прямого импорта app.core.config.settings /
+ # app.services.scraper_settings.get_scraper_delay. Продуктовая сторона
+ # передаёт RealScraperConfig + get_scraper_delay; kit не знает про app / БД.
+ self._config = config
+ if delay_provider is not None:
+ self.request_delay_sec = delay_provider(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 self._config.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 self._config.scraper_fetch_mode == "browser":
+ # Default (#1805): SERP-фетч через camoufox (tradein-browser per-provider
+ # + ротирующий backconnect-прокси). curl_cffi поднимаем тут же как
+ # FALLBACK — если браузер отдаёт firewall/пусто, _fetch_serp_html делает
+ # один retry через curl_cffi. _build_cffi_session самодостаточен
+ # (отдельный AsyncSession, не зависит от BaseScraper.__aenter__).
+ self._browser = BrowserFetcher(source="avito")
+ await self._browser.__aenter__()
+ self._cffi = self._build_cffi_session()
+ logger.info(
+ "avito: SERP fetch via BrowserFetcher (camoufox) default, "
+ "curl_cffi fallback armed — #1805"
+ )
+ return self
+ # Explicit opt-out: legacy curl_cffi-only путь (SCRAPER_FETCH_MODE=curl_cffi).
+ 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 self._config.scraper_proxy_url:
+ proxies = {"http": self._config.scraper_proxy_url, "https": self._config.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 = self._config.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 — dispatcher между browser-default и curl_cffi (#1805).
+
+ Если поднят BrowserFetcher (default с #1805) — фетчим через camoufox с
+ curl_cffi-fallback на firewall (_fetch_serp_html_browser). Иначе
+ (legacy/explicit curl_cffi opt-out, либо тесты с прямым _cffi) — прямой
+ curl_cffi-путь (_fetch_serp_html_cffi).
+
+ Returns:
+ HTML при успехе, либо ``None`` при non-200 (конец пагинации).
+ Raises:
+ AvitoBlockedError / AvitoRateLimitedError — если бан не снят.
+ """
+ if self._browser is not None:
+ return await self._fetch_serp_html_browser(url, page)
+ return await self._fetch_serp_html_cffi(url, page)
+
+ async def _fetch_serp_html_browser(self, url: str, page: int) -> str | None:
+ """SERP-фетч через camoufox-браузер (#1805, default) с curl_cffi-fallback.
+
+ Браузер не отдаёт HTTP-статус — бан виден только как rendered firewall-HTML
+ (camoufox исполняет JS, поэтому firewall-интерстициал приходит готовым
+ документом; ловим теми же текст-маркерами _is_firewall_page).
+
+ Pipeline:
+ 1. browser.fetch → firewall? → ротация IP (до max_rotations) + retry.
+ 2. Ротация исчерпана и всё ещё firewall/пусто → ОДИН retry через curl_cffi
+ (другой egress/TLS-fingerprint может проскочить). curl-fallback
+ доступен только если _cffi поднят (standalone __aenter__); в
+ pipeline-режиме с shared-browser _cffi=None → fallback пропускается.
+ 3. curl-fallback тоже firewall/пусто (или недоступен) → AvitoBlockedError.
+
+ Отдельная ветка — sidecar HTTP-ошибка (#356): BrowserFetcher пробрасывает
+ httpx.HTTPStatusError (500/503), если /fetch упал внутри (Page.goto Timeout,
+ NS_ERROR_PROXY_TOO_MANY_REQUESTS, браузер не поднялся). Раньше она всплывала
+ неперехваченной → generic except в run_avito_full_load → mark_failed +
+ re-raise, что роняло весь обход даже после тысяч инкрементально сохранённых
+ карточек. Теперь классифицируем тело ошибки и роутим в ту же
+ ротация+retry-машинерию: soft-ban → ротация IP, transient → bounded backoff,
+ оба бюджета исчерпаны → AvitoRateLimitedError (graceful mark_banned).
+ """
+ assert self._browser is not None
+ max_rot = self._config.avito_proxy_max_rotations if self._config.avito_proxy_rotate_url else 0
+ rot_done = 0
+ # Независимый бюджет коротких retry на transient sidecar-ошибки (timeout/
+ # crash) — НЕ тратит бюджет ротации. Каждая ветка ниже (ротация / transient)
+ # уменьшает свой счётчик → цикл всегда движется к условию исчерпания.
+ transient_left = _AVITO_SIDECAR_TRANSIENT_RETRIES
+ while True:
+ try:
+ html = await self._browser.fetch(url)
+ except httpx.HTTPStatusError as exc:
+ # Тело sidecar-ошибки: {"error": "ErrorType: message"}. Парсим
+ # устойчиво — невалидный JSON не должен ронять обработку.
+ status = exc.response.status_code
+ try:
+ error_text = str(exc.response.json().get("error", ""))
+ except Exception:
+ error_text = exc.response.text
+ lowered = error_text.lower()
+
+ is_soft_ban = (
+ "too_many_requests" in lowered
+ or status == 503
+ or "browser unavailable" in lowered
+ or "proxy may be down" in lowered
+ )
+ if is_soft_ban:
+ # Soft proxy rate-limit / прокси недоступен → свежий exit-IP.
+ if rot_done < max_rot and await self._rotate_ip():
+ rot_done += 1
+ logger.info(
+ "avito page=%d sidecar soft-ban (status=%d) — "
+ "retry after rotation #%d: %s",
+ page,
+ status,
+ rot_done,
+ error_text[:200],
+ )
+ continue
+ else:
+ # Transient (Page.goto Timeout / Target closed / прочие 5xx) →
+ # короткий backoff-retry того же url БЕЗ ротации.
+ if transient_left > 0:
+ transient_left -= 1
+ backoff = _AVITO_SIDECAR_TRANSIENT_BACKOFF_SEC + random.uniform(0, 0.5)
+ logger.warning(
+ "avito page=%d sidecar transient (status=%d) — "
+ "retry in %.1fs (left=%d): %s",
+ page,
+ status,
+ backoff,
+ transient_left,
+ error_text[:200],
+ )
+ await asyncio.sleep(backoff)
+ continue
+ # Оба бюджета (ротация + transient) исчерпаны → graceful soft-ban,
+ # чтобы run_avito_full_load/city_sweep ушли в mark_banned (partial
+ # preserved), а НЕ в generic except → mark_failed. Намеренно
+ # AvitoRateLimitedError, не AvitoBlockedError и не bare exception.
+ logger.error(
+ "avito page=%d sidecar error budget exhausted (status=%d) url=%s: %s",
+ page,
+ status,
+ url,
+ error_text[:200],
+ )
+ raise AvitoRateLimitedError(
+ f"avito SERP browser-sidecar error (page={page}): {error_text[:200]}"
+ ) from exc
+ # Пустой рендер ("" / None) = таймаут рендера sidecar'а, НЕ firewall:
+ # _is_firewall_page("") вернул бы False → пустой HTML утёк бы как
+ # «валидный». Возвращаем None — caller (fetch_around) трактует это как
+ # конец пагинации (break), а не как бан.
+ if not html:
+ return None
+ if not _is_firewall_page(html):
+ return 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
+ # Ротация исчерпана/недоступна. Fallback: один retry через curl_cffi.
+ if self._cffi is not None:
+ logger.warning(
+ "avito page=%d browser firewall — falling back to curl_cffi (#1805)",
+ page,
+ )
+ try:
+ fb_html = await self._fetch_serp_html_cffi(url, page)
+ except (AvitoBlockedError, AvitoRateLimitedError):
+ # curl-fallback тоже забанен → пробрасываем как hard-block.
+ logger.error("avito page=%d curl_cffi fallback also blocked url=%s", page, url)
+ raise
+ if fb_html is None or not _is_firewall_page(fb_html):
+ # None = non-200/конец пагинации; иначе валидный HTML.
+ return fb_html
+ logger.error("avito page=%d curl_cffi fallback still firewall url=%s", page, url)
+ 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"
+ )
+
+ async def _fetch_serp_html_cffi(self, url: str, page: int) -> str | None:
+ """GET SERP HTML через curl_cffi (legacy/fallback путь) с anti-block-логикой.
+
+ Avito на забаненном datacenter-IP отдаёт HTTP 200 + firewall-страницу (не
+ 403). 429 через backconnect-прокси — transient conn-limit (short-retry без
+ ротации). 403/firewall на backconnect — залочен текущий exit-IP, escape
+ через reconnect (_reset_cffi) либо changeip-ротацию.
+
+ Returns:
+ HTML при HTTP 200, либо ``None`` при non-200 (конец пагинации).
+ Raises:
+ AvitoBlockedError / AvitoRateLimitedError — если бан не снят.
+ """
+ assert self._cffi is not None
+ max_rot = self._config.avito_proxy_max_rotations if self._config.avito_proxy_rotate_url else 0
+ # curl_cffi-сессия всегда egress через self._config.scraper_proxy_url (backconnect
+ # mproxy). Пересоздание сессии (_reset_cffi) даёт свежий exit-IP вне зависимости
+ # от наличия changeip-URL (тот относится к ДРУГОМУ, browser/auv-прокси). Поэтому
+ # reconnect-retry гейтим ТОЛЬКО на наличие proxy, не на отсутствие rotate_url.
+ backconnect = bool(self._config.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 рендерит счётчик в 1 178.
+ Это аналог 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.
+
+ Если self._config.avito_serp_ekb_only=True (дефолт), отбрасываем карточки,
+ чей per-card source_url не содержит /ekaterinburg/ — Avito добивает выдачу
+ «по всей России» при малом числе ЕКБ-результатов (4+ комн., дорогие сегменты).
+ Карточки с нераспознанным city-slug (href без ведущего /city/) пропускаются
+ консервативно (не дропаются по ошибке).
+ """
+ tree = HTMLParser(html)
+ ts_map = _build_sort_timestamp_map(html)
+ cards = tree.css('[data-marker="item"]')
+ lots: list[ScrapedLot] = []
+ ekb_only = self._config.avito_serp_ekb_only
+ dropped = 0
+ no_city = 0
+ for card in cards:
+ lot = self._dom_card_to_lot(card, source_url_base, ts_map)
+ if lot is None:
+ continue
+ if ekb_only:
+ card_url = lot.source_url or ""
+ # Определяем city-slug по первому сегменту пути в абсолютном URL.
+ # urljoin всегда возвращает абсолютный URL (https://www.avito.ru/...).
+ # Avito city-slug — первый path-сегмент после netloc.
+ # Пример: https://www.avito.ru/ekaterinburg/... → /ekaterinburg/ присутствует.
+ # https://www.avito.ru/moskva/... → нет /ekaterinburg/.
+ # Для href без явного city (/kvartiry/... — теоретически не должно быть,
+ # но консервативно пропускаем).
+ if "/ekaterinburg/" in card_url:
+ # ЕКБ-карточка — берём
+ pass
+ else:
+ # Проверяем, есть ли вообще city-slug (хотя бы один path-сегмент
+ # после авито-хоста). Если нет — консервативно берём.
+ parsed_path = urlparse(card_url).path
+ # Первый непустой сегмент пути — city-slug в Avito URL.
+ first_seg = next((s for s in parsed_path.split("/") if s), None)
+ # Известные не-city сегменты Avito (/kvartiry/, /item/, ...) —
+ # если первый сегмент — не похоже на city-slug, пропускаем консервативно.
+ # Простая эвристика: city-slug содержит только строчные буквы/дефис/цифры
+ # и НЕ является известным non-city сегментом.
+ _non_city_segs = {"kvartiry", "item", "catalog", "user", "api", ""}
+ if first_seg is None or first_seg in _non_city_segs:
+ no_city += 1
+ # Нет city-slug → консервативно берём карточку
+ lots.append(lot)
+ continue
+ # Есть city-slug, но не /ekaterinburg/ → padding по России → дропаем
+ dropped += 1
+ continue
+ lots.append(lot)
+ if ekb_only and (dropped or no_city):
+ logger.info(
+ "avito SERP: dropped %d non-EKB cards (padding), %d cards without city-slug kept",
+ dropped,
+ no_city,
+ )
+ 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)
+
+ # Адрес: первый внутри 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:
+ # 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 по 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: новостройки несут якорь на ЖК-сабдомен
+ # .
+ # newbuilding_url — host+path без ?context (mirror urlparse-cleaning),
+ # newbuilding_id — = 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,
+)
+
+# #1773: Avito DOM иногда выдаёт номер дома СЛИПШИМСЯ с маркером района/нас.пункта
+# без разделителя — «…Савкова, 29р-н Академический», «17Ар-н», «35/2р-н», «33с1р-н».
+# Это ломает извлечение номера дома и матчинг «того же дома» (см. issue #1773).
+# Де-слип вставляет «, » между токеном номера дома и следующим маркером.
+#
+# Группа 1 — токен номера дома: цифры + опц. кириллическая литера (А/Б/...) +
+# опц. корпус/строение (/N, сN, кN, корпус N, стр N) — напр. 29, 29А, 35/2, 33с1, 44Б.
+# Группа 2 — маркер района/нас.пункта, начинающийся СРАЗУ за номером без пробела.
+#
+# Тонкость со слипшейся литерой: в «17Ар-н» литера «А» и маркер «р-н» — соседние
+# кириллические буквы без границы. Поэтому литеру дома матчим ТОЛЬКО когда сразу
+# за ней идёт известный маркер (lookahead на _GLUE_MARKER) — иначе риск съесть
+# первую букву обычного слова. Корпус «сN/кN» требует цифры, поэтому не путается
+# с маркерами «с.» (село) / «к» — у тех нет цифры вплотную.
+_GLUE_MARKER = (
+ r"(?:р-?н|мкр\.?|микрорайон|район|г\.|пос\.?|посёлок|поселок" r"|снт|кв-л|квартал|д\.|с\.|пгт)"
+)
+_GLUE_HOUSE_TOKEN = (
+ r"\d+" # дом: цифры
+ # опц. одиночная литера (29А) — только если за ней сразу маркер
+ rf"(?:\s?[А-Яа-я](?={_GLUE_MARKER}))?"
+ r"(?:\s?(?:/\d+|с\d+|к\d+|корп\.?\s?\d+|стр\.?\s?\d+))?" # опц. корпус/строение
+)
+# Хвост `(?![а-яёa-z])` гарантирует, что маркер — отдельный токен, а не начало слова.
+_DEGLUE_RE = re.compile(
+ rf"({_GLUE_HOUSE_TOKEN})({_GLUE_MARKER}(?![а-яёa-z]))",
+ flags=re.IGNORECASE,
+)
+
+
+def _deglue_house_marker(addr: str) -> str:
+ """Вставить «, » между слипшимся номером дома и маркером района/нас.пункта.
+
+ «…Савкова, 29р-н Академический» -> «…Савкова, 29, р-н Академический»
+ «…, 17Ар-н Пионерский» -> «…, 17А, р-н Пионерский»
+ «…, 35/2р-н …» -> «…, 35/2, р-н …»
+ «…, 33с1р-н …» -> «…, 33с1, р-н …»
+
+ Идемпотентно: на уже-расклеенной строке («29, р-н») паттерн не срабатывает,
+ т.к. между номером и маркером уже есть запятая/пробел.
+ """
+ return _DEGLUE_RE.sub(r"\1, \2", addr)
+
+
+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"
+ "ул. Евгения Савкова, 29р-н Академический"
+ -> "ул. Евгения Савкова, 29, р-н Академический"
+ """
+ if not raw:
+ return None
+ cleaned = _CSS_NOISE_RE.sub("", raw)
+ cleaned = _NOT_ADDRESS_TAIL_RE.split(cleaned, maxsplit=1)[0]
+ cleaned = _deglue_house_marker(cleaned)
+ 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
diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/shared.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/shared.py
new file mode 100644
index 00000000..c499178d
--- /dev/null
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/shared.py
@@ -0,0 +1,79 @@
+"""avito_shared.py — общие константы/хелперы для avito*.py модулей.
+
+Единая точка правды для:
+ - `RUS_MONTHS` — рус. месяц (родительный падеж) → номер, для парсинга дат вида
+ "9 октября 2025" (было продублировано в avito.py/avito_detail.py/avito_houses.py).
+ - `_unix_to_date` — unix-epoch (в СЕКУНДАХ) → `date`, в таймзоне Europe/Moscow.
+
+TZ-обоснование (issue #2129): было 3 независимые копии конвертации unix→date с
+РАЗНЫМИ timezone — MSK (avito.py, документировано issue #726), UTC (avito_houses.py),
+naive/system-local (avito_imv.py). Один и тот же epoch давал разные `date` в
+SERP / houses / IMV путях для одного и того же объявления.
+
+Почему MSK — канонический выбор (не UTC, не local):
+ - Avito — российская площадка, отображает даты объявлений (`item-date` на SERP,
+ `publishDate`, история цен `startPriceDate`/`lastPriceDate`/`removedDate`) в
+ Europe/Moscow. Объявление, поднятое в 00:00–03:00 MSK, при наивной UTC-конвертации
+ получило бы дату на день раньше, чем видно на самом Avito (обоснование #726).
+ - avito_houses.py и avito_imv.py используются на одних и тех же данных
+ (`housePlacementHistory` widget / IMV `placementHistory` — идентичные поля
+ startPriceDate/lastPriceDate/removedDate, см. fixtures/avito_imv_getdata.json) —
+ единственный смысл держать их согласованными с уже принятым MSK-решением из #726.
+ - Проверено на живой fixture: startPriceDate/lastPriceDate — всегда полночь UTC
+ (00:00:00), поэтому MSK vs UTC не меняет date для них. `removedDate`/`publishDate`
+ несут реальное время суток — там UTC/naive-local и MSK ДЕЙСТВИТЕЛЬНО расходятся
+ (пример из fixture: removedDate=1756077885 → UTC 2025-08-24, MSK 2025-08-25).
+ Это единственный реальный сдвиг данных от этого фикса: `avito_imv_evaluations` /
+ `house_placement_history`, поля `removed_date`/`publish_date`, у ~12.5% строк
+ (события в 21:00–23:59 UTC/local) дата сдвинется на +1 день при следующем прогоне
+ скрейпера. Схему/уже сохранённые строки не трогаем — это единственно правильное
+ представление на будущее.
+ - Fixed-offset `timezone(timedelta(hours=3))`, а не `zoneinfo("Europe/Moscow")` —
+ не зависит от системной tzdata в контейнере; Москва не переходит на DST с 2014,
+ так что fixed offset корректен всегда.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime, timedelta, timezone
+
+logger = logging.getLogger(__name__)
+
+# Europe/Moscow, UTC+3, без перехода на летнее время (с 2014).
+MSK = timezone(timedelta(hours=3))
+
+# Русские месяцы (родительный падеж) для парсинга дат вида "9 октября 2025".
+RUS_MONTHS: dict[str, int] = {
+ "января": 1,
+ "февраля": 2,
+ "марта": 3,
+ "апреля": 4,
+ "мая": 5,
+ "июня": 6,
+ "июля": 7,
+ "августа": 8,
+ "сентября": 9,
+ "октября": 10,
+ "ноября": 11,
+ "декабря": 12,
+}
+
+
+def _unix_to_date(ts: int | float | None) -> date | None:
+ """Конвертирует unix-epoch (в СЕКУНДАХ) в `date`, в таймзоне Europe/Moscow (MSK).
+
+ Возвращает `None` для `None`/`0` (Avito использует 0 как sentinel «нет даты») и
+ для значений, которые не конвертируются в валидную дату.
+
+ ВАЖНО: ожидает epoch в СЕКУНДАХ. Avito SERP-поле `sortTimeStamp` отдаёт epoch в
+ МИЛЛИСЕКУНДАХ — вызывающий код обязан поделить на 1000 до вызова (см.
+ `avito.py::_build_sort_timestamp_map`).
+ """
+ if not ts:
+ return None
+ try:
+ return datetime.fromtimestamp(float(ts), tz=MSK).date()
+ except (OSError, OverflowError, ValueError):
+ logger.warning("Не удалось конвертировать unix timestamp: %r", ts)
+ return None
diff --git a/tradein-mvp/uv.lock b/tradein-mvp/uv.lock
index 28de8dbb..e03ef752 100644
--- a/tradein-mvp/uv.lock
+++ b/tradein-mvp/uv.lock
@@ -1114,6 +1114,7 @@ name = "scraper-kit"
version = "0.1.0"
source = { editable = "packages/scraper-kit" }
dependencies = [
+ { name = "curl-cffi" },
{ name = "httpx", extra = ["socks"] },
{ name = "psycopg", extra = ["binary"] },
{ name = "pydantic" },
@@ -1124,6 +1125,7 @@ dependencies = [
[package.metadata]
requires-dist = [
+ { name = "curl-cffi", specifier = ">=0.7.0" },
{ name = "httpx", extras = ["socks"], specifier = ">=0.27.0" },
{ name = "psycopg", extras = ["binary"], specifier = ">=3.2.0" },
{ name = "pydantic", specifier = ">=2.7.0" },