"""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 random
import re
from collections.abc import Callable
from datetime import date, timedelta
from typing import TYPE_CHECKING, Any
from urllib.parse import urlencode, urljoin, urlparse, urlunparse

import httpx
from curl_cffi.requests import AsyncSession
from selectolax.parser import HTMLParser

from scraper_kit.avito_exceptions import (
    AvitoBlockedError,
    AvitoContentBlockedError,
    AvitoRateLimitedError,
)
from scraper_kit.base import BaseScraper, ScrapedLot
from scraper_kit.browser_fetcher import BrowserFetcher
from scraper_kit.price_brackets import get_price_seed_brackets
from scraper_kit.pricing import BisectionConfig, ProbeFailPolicy, ProbeResult, walk_price_range
from scraper_kit.providers._base import build_browser_fetcher, build_document_session
from scraper_kit.providers.avito.shared import _unix_to_date

if TYPE_CHECKING:
    from scraper_kit.contracts import ScraperConfig

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 через общий `_unix_to_date` (avito_shared.py, #2129), чтобы
# совпадало с тем, что видно на сайте (иначе объявления, поднятые в 00:00–03:00 MSK,
# давали бы дату на день раньше). #726.
_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-дублей).
    `_unix_to_date` ожидает СЕКУНДЫ — sortTimeStamp в МИЛЛИСЕКУНДАХ, делим на 1000.
    """
    out: dict[str, date] = {}
    for m in _SORT_TS_RE.finditer(html):
        item_id = m.group("id")
        if item_id in out:
            continue
        date_val = _unix_to_date(int(m.group("ts")) / 1000)
        if date_val is not None:
            out[item_id] = date_val
    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


def _avito_bisection_config(cap: int) -> BisectionConfig:
    """Конфиг единого движка бисекции (scraper_kit.pricing) для avito.

    probe_fail=SPLIT_OR_SKIP: probe вернул None (captcha/browser-timeout) → открытый
    брекет пагинируем best-effort, закрытый ещё делимый (bracket >= MIN) принудительно
    делим (тяжёлая выдача станет загружаемой в узком диапазоне), иначе пропуск.
    open_split_ceiling=None: открытый брекет (hi=None) пагинируется напрямую, не делится.
    """
    return BisectionConfig(
        cap=cap,
        min_bracket=_AVITO_MIN_BRACKET,
        probe_fail_policy=ProbeFailPolicy.SPLIT_OR_SKIP,
    )

# 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

# tradein-browser sidecar (POST /fetch) отдаёт HTTP 500 на ЛЮБУЮ внутреннюю
# ошибку страницы (тело {"error": "ErrorType: message"}) и HTTP 503 когда браузер
# не поднимается (прокси упал). Транзиентные сбои (Page.goto Timeout, Target
# closed, прочие NS_ERROR_/5xx) лечатся коротким backoff-retry того же url БЕЗ
# ротации IP; soft-ban (NS_ERROR_PROXY_TOO_MANY_REQUESTS / 503 / browser
# unavailable) — ротацией exit-IP. Когда ОБА бюджета исчерпаны → AvitoRateLimitedError,
# чтобы run_avito_full_load/run_avito_city_sweep ушли в graceful mark_banned путь
# (partial preserved), а не в generic except → mark_failed (прод-инцидент run #356:
# один поздний 500 ронял весь обход после 3186 сохранённых объявлений).
_AVITO_SIDECAR_TRANSIENT_RETRIES = 2
_AVITO_SIDECAR_TRANSIENT_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&nbsp;178</span>.
# Текст может содержать NBSP (\xa0), &nbsp;-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,
        config: ScraperConfig,
        *,
        delay_provider: Callable[[str], float] | None = None,
        target_city_slug: str | 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
        # #2487 oblast rollout: slug города-цели sweep'а (напр. "nizhniy_tagil").
        # None → ЕКБ (дефолт). Используется _parse_html-фильтром avito_serp_ekb_only,
        # чтобы оставлять карточки TARGET-города, а не хардкодить /ekaterinburg/.
        self._target_city_slug = target_city_slug

    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 = build_browser_fetcher(self._config, "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.
        """
        if self._config.scraper_proxy_url:
            logger.info("avito: routing through mobile proxy egress (#623)")
        return build_document_session(proxy_url=self._config.scraper_proxy_url, timeout=25)

    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-ошибка (#356, #2472): BrowserFetcher пробрасывает
        httpx.HTTPStatusError (500/503), если /fetch упал внутри (Page.goto Timeout,
        NS_ERROR_PROXY_TOO_MANY_REQUESTS, браузер не поднялся), либо
        httpx.TimeoutException/httpx.TransportError (ReadTimeout/ConnectError/
        RemoteProtocolError), если соединение до sidecar оборвалось ДО ответа (после
        internal-retry в BrowserFetcher.fetch). Раньше обе ветки (а до #2472 — только
        не-HTTPStatusError) всплывали неперехваченными → generic except в
        run_avito_full_load → mark_failed + re-raise, что роняло весь обход даже после
        тысяч инкрементально сохранённых карточек (и теряло done_buckets-чекпоинт —
        mark_failed его не пишет). Теперь классифицируем тело ошибки (HTTP-ошибки) или
        просто помечаем как transient (не-HTTP) и роутим в ту же ротация+retry-
        машинерию: soft-ban → ротация IP, transient → bounded backoff, оба бюджета
        исчерпаны → AvitoRateLimitedError (graceful mark_banned, partial preserved).
        """
        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, httpx.TimeoutException, httpx.TransportError) as exc:
                # HTTPStatusError: тело sidecar-ошибки {"error": "ErrorType: message"}.
                # Парсим устойчиво — невалидный JSON не должен ронять обработку.
                #
                # TimeoutException/TransportError (ReadTimeout/ConnectError/
                # RemoteProtocolError): сеть до sidecar оборвалась ДО ответа (после
                # internal-retry BrowserFetcher.fetch, browser_fetcher.py) — HTTP-статуса
                # нет. Раньше это вываливалось необработанным через
                # _paginate_incremental_bracket в generic except run_avito_full_load →
                # mark_failed(str(exc)) — а str(ReadTimeout())=="" (пустой error) теряет
                # done_buckets чекпоинт (mark_failed его не пишет, в отличие от mark_banned).
                # Роутим как transient (status=0, не soft-ban) — симметрично _probe_total.
                if isinstance(exc, httpx.HTTPStatusError):
                    status = exc.response.status_code
                    try:
                        error_text = str(exc.response.json().get("error", ""))
                    except Exception:
                        error_text = exc.response.text
                else:
                    status = 0
                    error_text = f"{type(exc).__name__}: {exc}"
                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 _city_seg(self) -> str:
        """City-slug для пути avito URL (#12: раньше хардкод /ekaterinburg/ во всех
        билдерах скоупил oblast city-sweep на ЕКБ, невзирая на geoCoords/фильтр)."""
        return self._target_city_slug or "ekaterinburg"

    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}/{self._city_seg()}/kvartiry/prodam-ASgBAgICAUSSA8YQ"
            f"?{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}/{self._city_seg()}/kvartiry/prodam-ASgBAgICAUSSA8YQ"
            f"?{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"/{self._city_seg()}/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}/{self._city_seg()}/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&nbsp;178</span>.
        Это аналог cian totalOffers — gate для бисекции (total > cap → split).
        Парсинг: текст span → убрать NBSP (\xa0)/пробелы/`&nbsp;` → 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), &nbsp;-entity и обычные пробелы-разделители разрядов.
            cleaned = raw.replace("\xa0", "").replace("&nbsp;", "").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 пробрасываются наверх.

        Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Avito-spec
        probe (page=1 + IP-rotate retry) и leaf-пагинация переданы callback'ами;
        пороги (cap, _AVITO_MIN_BRACKET) и политика probe-fail (SPLIT_OR_SKIP —
        принудительный split тяжёлой выдачи / best-effort для открытого) — в
        BisectionConfig. AvitoBlockedError/AvitoRateLimitedError из probe/leaf
        всплывают наверх (mark_banned).
        """

        async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult:
            lo_param = plo if plo and plo > 0 else None
            hi_repr = "open" if phi is None else str(phi)
            # _probe_total ловит browser-ошибки (HTTP 500 / timeout от tradein-browser
            # на тяжёлой выдаче) → total=None вместо краха. Block-исключения всплывают.
            html, total = await self._probe_total(room_slug, lo_param, phi)
            if total is None:
                logger.warning(
                    "avito: total=None for %s [%s, %s] depth=%d — rotating IP + retry",
                    room_label,
                    plo,
                    hi_repr,
                    depth,
                )
                rotated = await self._rotate_ip()
                if rotated:
                    html, total = await self._probe_total(room_slug, lo_param, phi)
            if total is not None:
                logger.info(
                    "avito: %s [%s, %s] total=%d depth=%d",
                    room_label,
                    plo,
                    hi_repr,
                    total,
                    depth,
                )
            return ProbeResult(count=total, payload=html)

        async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None:
            _lo = plo if plo is not None else 0
            total = result.count
            if total is None:
                # SPLIT_OR_SKIP на открытом брекете: probe провалился, делить нельзя —
                # пагинируем best-effort до max_pages_per_bucket (tail-loss accepted).
                html: str | None = None
                max_pages = max_pages_per_bucket
                logger.warning(
                    "avito: probe failed for OPEN bucket %s [%d, open] — "
                    "paginating up to max_pages (tail-loss accepted)",
                    room_label,
                    _lo,
                )
            else:
                if phi is None and 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)
                if phi is not None and 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,
                        phi,
                        total,
                        max_pages_per_bucket,
                        _AVITO_OFFERS_PER_PAGE,
                        max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE,
                        tail_loss,
                    )
                html = result.payload
            await self._paginate_leaf_bucket(
                room_slug=room_slug,
                room_label=room_label,
                lo=_lo,
                hi=phi,
                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,
            )

        await walk_price_range(
            lo=lo,
            hi=hi,
            config=_avito_bisection_config(price_cap_per_bucket),
            probe=_probe,
            on_leaf=_leaf,
            depth=_depth,
        )

    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 не содержит слаг ГОРОДА-ЦЕЛИ sweep'а — Avito добивает
        выдачу «по всей России» при малом числе результатов (4+ комн., дорогие сегменты).
        Город-цель = self._target_city_slug (напр. "nizhniy_tagil"), по умолчанию
        "ekaterinburg". Без параметризации (#2487) oblast-sweep по городу вне ЕКБ терял
        100% карточек: их per-card slug (/nizhniy_tagil/) не содержит /ekaterinburg/.
        Карточки с нераспознанным 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
        # #2487: слаг города, карточки которого оставляем. Дефолт "ekaterinburg" —
        # ЕКБ-поведение неизменно, когда target не задан (обычный ЕКБ-sweep/full-load).
        kept_slug = self._target_city_slug or "ekaterinburg"
        kept_seg = f"/{kept_slug}/"
        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/...        → нет слага города-цели.
                # Для href без явного city (/kvartiry/... — теоретически не должно быть,
                # но консервативно пропускаем).
                if kept_seg 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, но не города-цели → padding по России → дропаем
                    dropped += 1
                    continue
            lots.append(lot)
        if ekb_only and (dropped or no_city):
            logger.info(
                "avito SERP (kept=%s): dropped %d off-target cards (padding), "
                "%d cards without city-slug kept",
                kept_slug,
                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)

            # Адрес: первый <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,
)

# #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