"""Avito.ru scraper — парсинг вторички вокруг точки + citywide/byrooms sweep.

Стратегия: HTML scrape карточек объявлений через DOM.

URL patterns (EKB):
  geo-mode (default):
    https://www.avito.ru/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ
      ?geoCoords=56.838,60.605
      &radius=1                  # в км
      &s=104                     # sort by date
      &p=1                       # страница

  citywide-mode (без geo, весь город, T6):
    https://www.avito.ru/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ
      ?s=104&p=1

  byrooms-mode (citywide + фильтр комнатности, T6):
    https://www.avito.ru/ekaterinburg/kvartiry/prodam/<room-slug>?s=104&p=1

ВАЖНО: Avito банит httpx (403/429) по TLS fingerprint от server IP.
Используем curl_cffi с impersonate='chrome120' — настоящий Chrome TLS ClientHello.
"""

from __future__ import annotations

import asyncio
import inspect
import logging
import math
import re
from collections.abc import Callable
from datetime import date, datetime, timedelta, timezone
from typing import Any
from urllib.parse import urlencode, urljoin, urlparse, urlunparse

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

from app.core.config import settings
from app.services.scraper_settings import get_scraper_delay
from app.services.scrapers.avito_exceptions import (
    AvitoBlockedError,
    AvitoContentBlockedError,
    AvitoRateLimitedError,
)
from app.services.scrapers.base import BaseScraper, ScrapedLot
from app.services.scrapers.browser_fetcher import BrowserFetcher
from app.services.scrapers.price_brackets import get_price_seed_brackets

logger = logging.getLogger(__name__)

# ── Relative date parsing ────────────────────────────────────────────────────
_REL_DATE_RE = re.compile(
    r"(?P<n>\d+)\s+(?P<unit>"
    r"секунд[ауы]?|"
    r"минут[ауы]?|"
    r"час(?:а|ов)?|"
    r"день|дн(?:я|ей?)|"
    r"недел[июяь]+|"
    r"месяц(?:а|ев)?|"
    r"год(?:а|ов)?"
    r")\s+назад",
    flags=re.I,
)
_UNIT_DAYS: dict[str, int] = {
    "секунд": 0,
    "минут": 0,
    "час": 0,
    "день": 1,
    "дн": 1,
    "недел": 7,
    "месяц": 30,
    "год": 365,
}
# #1387: Avito SERP часто показывает давность в единственном числе БЕЗ ведущего
# числа — «час назад», «минуту назад», «день назад», «неделю назад», «месяц
# назад», «год назад». Числовой `_REL_DATE_RE` их не матчит (требует `\d+`),
# поэтому отдельная ветка с неявным n=1. Юниты в винительном падеже («минуту»,
# «неделю») плюс именительный («час», «день»).
_REL_DATE_SINGULAR_RE = re.compile(
    r"(?P<unit>" r"секунду|" r"минуту|" r"час|" r"день|" r"неделю|" r"месяц|" r"год" r")\s+назад",
    flags=re.I,
)

# Russian month names → number. Avito показывает абсолютные даты как «18 мая».
_RU_MONTHS: dict[str, int] = {
    "январ": 1,
    "феврал": 2,
    "март": 3,
    "апрел": 4,
    "ма": 5,
    "июн": 6,
    "июл": 7,
    "август": 8,
    "сентябр": 9,
    "октябр": 10,
    "ноябр": 11,
    "декабр": 12,
}
# «18 мая», «5 июня в 12:30»  — day + month (year inferred current).
_ABS_DATE_RE = re.compile(
    r"(?P<day>\d{1,2})\s+(?P<month>январ\w*|феврал\w*|март\w*|апрел\w*|"
    r"ма[яй]|июн\w*|июл\w*|август\w*|сентябр\w*|октябр\w*|ноябр\w*|декабр\w*)",
    flags=re.I,
)


def _parse_relative_date(s: str | None) -> date | None:
    """Avito показывает дату публикации в трёх форматах:
    * '5 дней назад' / 'час назад' / 'неделю назад'  — relative numeric
    * 'сегодня в 12:30' / 'вчера в 18:45' / 'позавчера'  — relative keyword
    * '18 мая' / '5 июня в 12:30'  — absolute day + month (year inferred)

    PR M-followup 2026-05-27: расширено покрытие — раньше только relative numeric
    matched (regex `\\d+ unit назад`), 100% Avito listings заходило с listing_date=None
    т.к. Avito SERP для свежих cards использует keyword/absolute форматы. См.
    тесты для каждого case.
    """
    if not s:
        return None
    text_lower = s.lower()

    # Keyword shortcuts (no \d+).
    if "позавчера" in text_lower:
        return date.today() - timedelta(days=2)
    if "вчера" in text_lower:
        return date.today() - timedelta(days=1)
    if "сегодня" in text_lower or "только что" in text_lower:
        return date.today()

    # Relative numeric («5 дней назад»).
    m = _REL_DATE_RE.search(s)
    if m:
        n = int(m["n"])
        unit_raw = m["unit"].lower()
        for prefix, days in _UNIT_DAYS.items():
            if unit_raw.startswith(prefix):
                return date.today() - timedelta(days=n * days)

    # #1387: Relative singular без числа («час назад», «неделю назад»). Неявное
    # n=1; те же _UNIT_DAYS. Винительный падеж юнита нормализуем к ключу-префиксу.
    m = _REL_DATE_SINGULAR_RE.search(s)
    if m:
        unit_raw = m["unit"].lower()
        # «минуту»→«минут», «неделю»→«недел», «секунду»→«секунд»; остальные уже
        # совпадают с префиксами _UNIT_DAYS.
        unit_key = {"минуту": "минут", "неделю": "недел", "секунду": "секунд"}.get(
            unit_raw, unit_raw
        )
        for prefix, days in _UNIT_DAYS.items():
            if unit_key.startswith(prefix):
                return date.today() - timedelta(days=days)

    # Absolute («18 мая», «5 июня в 12:30»). Year — current; если месяц
    # в будущем относительно сегодня (например сейчас декабрь, увидели
    # «5 января») — откатываем к прошлому году.
    m = _ABS_DATE_RE.search(s)
    if m:
        day = int(m["day"])
        month_raw = m["month"].lower()
        month: int | None = None
        for prefix, num in _RU_MONTHS.items():
            if month_raw.startswith(prefix):
                month = num
                break
        if month is not None and 1 <= day <= 31:
            today = date.today()
            year = today.year
            try:
                parsed = date(year, month, day)
            except ValueError:
                return None
            # Если parsed в будущем больше чем на 30 дней — Avito показывает
            # прошлогоднюю дату («декабрь» в феврале значит прошлый декабрь).
            if (parsed - today).days > 30:
                try:
                    parsed = date(year - 1, month, day)
                except ValueError:
                    return None
            return parsed

    return None


# #726: Avito SERP лениво рендерит DOM — `data-marker="item-date"` присутствует только
# у верхних (above-the-fold) карточек (~10 из ~46), у остальных его НЕТ в raw-HTML, что
# curl_cffi получает (JS-гидрация не выполняется). Но per-item JSON в той же HTML несёт
# `sortTimeStamp` (epoch-ms) почти у всех карточек — это drift-resistant источник даты.
# Маппим item_id (= DOM `data-item-id`) → дата. `{0,1500}` ограничивает поиск телом
# одного item-объекта (id и sortTimeStamp лежат в нём; внутри ещё есть короткий вложенный
# `addressDetailed{…}`, поэтому не `[^{}]`). re.S — sortTimeStamp может быть на новой строке.
# sortTimeStamp — UTC epoch-ms. Avito показывает дату в MSK (UTC+3, без DST с 2014).
# Берём .date() в MSK, чтобы совпадало с тем, что видно на сайте (иначе объявления,
# поднятые в 00:00–03:00 MSK, давали бы дату на день раньше). Fixed offset, без
# зависимости от системной tzdata в контейнере. #726.
_MSK = timezone(timedelta(hours=3))
_SORT_TS_RE = re.compile(r'"id":(?P<id>\d{8,12})\b.{0,1500}?"sortTimeStamp":(?P<ts>\d+)', re.S)


def _build_sort_timestamp_map(html: str) -> dict[str, date]:
    """item_id → дата публикации из embedded JSON (`sortTimeStamp`, epoch-ms).

    Best-effort: при любой ошибке конкретной записи — пропускаем её (не роняем парс).
    Первое вхождение id выигрывает (item-объект идёт раньше recommendations-дублей).
    """
    out: dict[str, date] = {}
    for m in _SORT_TS_RE.finditer(html):
        item_id = m.group("id")
        if item_id in out:
            continue
        try:
            out[item_id] = datetime.fromtimestamp(int(m.group("ts")) / 1000, tz=_MSK).date()
        except (ValueError, OSError, OverflowError):
            continue
    return out


# #623: Avito банит datacenter-IP — на HTTP 200 отдаёт firewall-интерстициал
# «Доступ ограничен: проблема с IP» (title в <head>), а НЕ 403. Детектим по
# маркерам в начале документа (firewall-страница объёмна, не сканируем целиком).
_FIREWALL_MARKERS = ("доступ ограничен", "проблема с ip", "firewall-container")

# EKB-фильтр «Новостройка» (slug категории kvartiry/prodam). Кодирует
# ASgB-параметры выборки только новостроек — SERP возвращает 100% new-build
# карточки (с data-marker="item-development-name"). Извлечён из реального
# search-URL: /ekaterinburg/kvartiry/prodam/novostroyka-ASgBAgICAkSSA8YQ5geOUg
NOVOSTROYKA_SLUG = "novostroyka-ASgBAgICAkSSA8YQ5geOUg"

# ── Exhaustive full-load (room×price bisection) ───────────────────────────────
# Верхняя граница цены при первом рекурсивном делении (нет явного hi).
# 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 результатов.
# Больше НЕ корень бисекции (см. _AVITO_PRICE_SEED_BRACKETS) — оставлен для
# совместимости (используется в тестах probe-устойчивости как sentinel).
_AVITO_MAX_PRICE = 200_000_000
# Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET прекращаем
# деление и пагинируем как есть (принимаем возможный «хвост»).
_AVITO_MIN_BRACKET = 50_000

# Seed price-brackets (руб) — data-driven по прод-распределению вторички ЕКБ
# (пик 4-8М, bulk 3-16М). Теперь общий констант для всех exhaustive-скрапперов
# (avito/cian/yandex) — app.services.scrapers.price_brackets.get_price_seed_brackets().
# Локальный алиас сохранён для читаемости и тестов; значения идентичны shared EKB.
# Каждый (комнатность × брекет) дальше дробится _walk_price_range если total > cap.
# Последний брекет ОТКРЫТ (hi=None → pmax не ставится) — ловит весь хвост люкса
# без потолка; он крошечный (avito >50М ≈ 73), пагинируется напрямую без бисекции.
_AVITO_PRICE_SEED_BRACKETS: list[tuple[int, int | None]] = get_price_seed_brackets()
# Avito SERP показывает ~50 карточек на страницу.
_AVITO_OFFERS_PER_PAGE = 50

# HTTP 429 в curl_cffi-режиме через backconnect-прокси (mproxy.site) — НЕ IP-ban, а
# transient «слишком много одновременных соединений» (лимит 5). Проходит на коротком
# retry без ротации IP. Делаем до _AVITO_429_MAX_RETRIES коротких пауз перед тем как
# свалиться в старое поведение (rotation / AvitoRateLimitedError).
_AVITO_429_MAX_RETRIES = 4
_AVITO_429_BACKOFF_SEC = 1.5

# Счётчик результатов в шапке 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,
)


def _is_firewall_page(html: str) -> bool:
    """True если Avito вернул firewall-страницу IP-блока (на HTTP 200)."""
    head = html[:4096].lower()
    return any(marker in head for marker in _FIREWALL_MARKERS)


class AvitoScraper(BaseScraper):
    """Avito vtorichka parser. Источник = 'avito'.

    Использует curl_cffi с impersonate=chrome120 для обхода TLS fingerprint бана.
    """

    name = "avito"
    base_url = "https://www.avito.ru"
    # Avito жёстко мониторит — спим долго между запросами.
    # Класс-дефолт; реальное значение загружается из scraper_settings при создании экземпляра.
    request_delay_sec = 7.0

    def __init__(self) -> None:
        super().__init__()
        self.request_delay_sec = get_scraper_delay(self.name)
        self._cffi: AsyncSession | None = None
        self._browser: BrowserFetcher | None = None
        # #823: счётчик карточек, которые не удалось распарсить из-за неожиданной структуры DOM.
        self.parse_failures: int = 0

    async def __aenter__(self) -> AvitoScraper:
        # Проактивная ротация IP в начале sweep (#1731): apw-IP «протухает» по
        # Datadome между суточными прогонами, поэтому первые якоря блокируются и
        # реактивного бюджета max_rotations не хватает (3 подряд блока →
        # mark_banned). Один changeip-вызов до первого SERP-fetch даёт свежий IP.
        # _rotate_ip() сам ждёт ~9с (settle) и логирует/глотает ошибку, поэтому
        # тут только дополнительный guard: сбой ротации НЕ должен ронять sweep.
        if settings.avito_proxy_rotate_url:
            try:
                if await self._rotate_ip():
                    logger.info("avito: proactive IP rotate at sweep start (#1731)")
                else:
                    logger.warning("avito: proactive IP rotate at sweep start failed — proceeding")
            except Exception:
                logger.warning(
                    "avito: proactive IP rotate at sweep start raised — proceeding", exc_info=True
                )
        if settings.scraper_fetch_mode == "browser":
            self._browser = BrowserFetcher(source="avito")
            await self._browser.__aenter__()
            logger.info("avito: SERP fetch via BrowserFetcher (camoufox) — #901")
            return self
        await super().__aenter__()
        proxies = None
        if settings.scraper_proxy_url:
            proxies = {"http": settings.scraper_proxy_url, "https": settings.scraper_proxy_url}
            logger.info("avito: routing through mobile proxy egress (#623)")
        self._cffi = AsyncSession(
            impersonate="chrome120",
            timeout=25,
            proxies=proxies,
            headers={
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
                "Cache-Control": "max-age=0",
                "Sec-Fetch-Dest": "document",
                "Sec-Fetch-Mode": "navigate",
                "Sec-Fetch-Site": "none",
                "Sec-Fetch-User": "?1",
                "Upgrade-Insecure-Requests": "1",
            },
        )
        return self

    async def __aexit__(self, *args: Any) -> None:
        if self._browser is not None:
            await self._browser.__aexit__(*args)
        if self._cffi is not None:
            await self._cffi.close()
        await super().__aexit__(*args)

    # ── Anti-block (#623) ─────────────────────────────────────────────────────
    async def _rotate_ip(self) -> bool:
        """Сменить мобильный IP через changeip-ссылку mobileproxy (#623).

        Дёргается напрямую (без прокси) — это API провайдера, не Avito. Ждём ~9с:
        мобильному модему нужно время поднять новый IP. Returns True при успехе.
        """
        rotate_url = settings.avito_proxy_rotate_url
        if not rotate_url:
            return False
        sep = "&" if "?" in rotate_url else "?"
        try:
            async with AsyncSession(timeout=30) as rot:
                await rot.get(f"{rotate_url}{sep}format=json")
            await asyncio.sleep(9)
            logger.info("avito proxy: IP rotated via changeip")
            return True
        except Exception:
            logger.warning("avito proxy: IP rotation failed", exc_info=True)
            return False

    async def _fetch_serp_html(self, url: str, page: int) -> str | None:
        """GET SERP HTML с детектом IP-бана и ротацией мобильного IP (#623).

        Avito на забаненном datacenter-IP отдаёт HTTP 200 + firewall-страницу
        (не 403). Детектим по тексту; если задана changeip-ссылка — меняем IP и
        повторяем до ``avito_proxy_max_rotations`` раз. Транзиентные сетевые
        обрывы (мобильный канал) ретраим без ротации.

        Returns:
            HTML при HTTP 200, либо ``None`` при non-200 (конец пагинации).
            В browser-mode HTTP-статус недоступен: метод возвращает HTML или
            поднимает ``AvitoBlockedError`` (pagination ends via empty-parse in callers).
        Raises:
            AvitoBlockedError / AvitoRateLimitedError — если бан не снят ротацией.
        """
        if self._browser is not None:
            # Browser-mode (#901): no HTTP status code — soft firewall-detect + rotate.
            max_rot = settings.avito_proxy_max_rotations if settings.avito_proxy_rotate_url else 0
            rot_done = 0
            while True:
                html = await self._browser.fetch(url)
                if _is_firewall_page(html):
                    if rot_done < max_rot and await self._rotate_ip():
                        rot_done += 1
                        logger.info(
                            "avito page=%d browser firewall — retry after rotation #%d",
                            page,
                            rot_done,
                        )
                        continue
                    logger.error("avito SERP firewall in browser-mode page=%d url=%s", page, url)
                    raise AvitoBlockedError(
                        f"Avito SERP firewall (browser-mode) at page={page} — IP banned"
                    )
                return html
        assert self._cffi is not None
        max_rot = settings.avito_proxy_max_rotations if settings.avito_proxy_rotate_url else 0
        rot_done = 0
        transient_left = 2
        r429 = 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
                if sc == 429:
                    logger.error("avito SERP HTTP 429 (rate limited) page=%d url=%s", page, url)
                    raise AvitoRateLimitedError(
                        f"Avito SERP returned 429 — rate limited at page={page}"
                    )
                logger.error(
                    "avito SERP blocked page=%d (HTTP %d, firewall=%s) url=%s",
                    page,
                    sc,
                    is_firewall,
                    url,
                )
                raise AvitoBlockedError(
                    f"Avito SERP blocked (HTTP {sc}, firewall={is_firewall}) "
                    f"at page={page} — IP banned"
                )
            if sc != 200:
                logger.warning("avito HTML page=%d returned %d for %s", page, sc, url)
                return None
            return response.text

    async def _fetch_serp_html_with_retry(
        self, url: str, page: int, *, max_retries: int = 2
    ) -> str | None:
        """Page-level retry поверх _fetch_serp_html для sweep-обходов (P2 resilience).

        _fetch_serp_html уже ретраит транзиентные fetch-обрывы и ротирует IP, но
        если после внутренних попыток он всё равно поднимает generic Exception —
        внешний sweep-цикл (citywide/byrooms) делал break и терял весь хвост.
        Здесь даём странице ещё ``max_retries`` попыток перед тем как пробросить.

        Поведение:
          - AvitoBlockedError / AvitoRateLimitedError — hard-block, пробрасываются
            немедленно (НЕ retry).
          - generic Exception — если попытки остались: warning + короткий backoff +
            следующая попытка; иначе re-raise последнего исключения (внешний цикл
            решит break).
          - None — это «non-200 / конец пагинации», НЕ ошибка → возвращается как есть
            без ретраев.
        """
        attempts = max_retries + 1
        backoff = max(2.0, min(self.request_delay_sec, 4.0))
        for attempt in range(1, attempts + 1):
            try:
                return await self._fetch_serp_html(url, page)
            except (AvitoBlockedError, AvitoRateLimitedError):
                raise
            except Exception:
                if attempt >= attempts:
                    raise
                logger.warning(
                    "avito page=%d transient error — retry (attempt %d/%d)",
                    page,
                    attempt,
                    max_retries,
                    exc_info=True,
                )
                await asyncio.sleep(backoff)
        return None  # pragma: no cover — loop either returns or raises

    # ── Public ──────────────────────────────────────────────────────────────
    async def fetch_around(
        self,
        lat: float,
        lon: float,
        radius_m: int = 1000,
        *,
        pages: int = 1,
        delay_override_sec: float | None = None,
    ) -> list[ScrapedLot]:
        """Найти объявления Авито вокруг (lat, lon) в radius_m метрах.

        Avito работает в км — конвертируем. Минимум 1 км.

        pages=1 (default) — backward compat, одна страница (~50 lots).
        pages>1 — итерируем p=1..pages, собираем до 50*pages lots.
        Между запросами sleep request_delay_sec (или delay_override_sec если передан).

        Coords НЕ заполняем из anchor (избегаем silent corruption через jitter).
        Точные lat/lon приходят позже из avito_detail.py — для search-карточек
        lat=lon=None, далее geocode-missing cron подтянет из address.
        """
        radius_km = max(1, round(radius_m / 1000))
        if delay_override_sec is not None:
            self.request_delay_sec = delay_override_sec

        all_lots: list[ScrapedLot] = []
        for page in range(1, pages + 1):
            url = self._build_web_url(lat, lon, radius_km, page=page)
            try:
                html = await self._fetch_serp_html(url, page)
            except (AvitoBlockedError, AvitoRateLimitedError):
                raise
            except Exception:
                logger.exception("avito HTML page=%d fetch failed for %s", page, url)
                break
            if html is None:
                break

            lots = self._parse_html(html, source_url_base=url)
            if not lots:
                if page == 1:
                    logger.error(
                        "avito SERP page=1 returned HTTP 200 but 0 cards — "
                        "likely content-block/captcha url=%s",
                        url,
                    )
                    raise AvitoContentBlockedError(
                        "Avito SERP HTTP 200 with 0 cards on page=1 — content-block suspected"
                    )
                logger.info("avito page=%d: 0 lots — end of pagination", page)
                break
            all_lots.extend(lots)
            logger.info(
                "avito page=%d: %d lots (total %d) around (%.4f, %.4f)",
                page,
                len(lots),
                len(all_lots),
                lat,
                lon,
            )
            if page < pages:
                await self.sleep_between_requests()

        if pages == 1 and all_lots:
            # Backward compat: single-page path does trailing sleep (как раньше)
            await self.sleep_between_requests()
        elif pages > 1:
            logger.info(
                "avito fetch_around pages=%d total_lots=%d around (%.4f, %.4f)",
                pages,
                len(all_lots),
                lat,
                lon,
            )

        return all_lots

    # ── Strategy B: HTML scrape ─────────────────────────────────────────────
    def _build_web_url(self, lat: float, lon: float, radius_km: int, page: int = 1) -> str:
        params = {
            "geoCoords": f"{lat},{lon}",
            "radius": radius_km,
            "s": 104,  # sort by date
            "p": page,
        }
        return f"{self.base_url}/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ?{urlencode(params)}"

    def _build_citywide_url(self, page: int = 1) -> str:
        """T6: URL всего города ЕКБ без geo-фильтра, сортировка по дате.

        Avito отдаёт все объявления города (cap ~5000 = 100 страниц × 50 карточек).
        """
        params = {"s": 104, "p": page}
        return f"{self.base_url}/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ?{urlencode(params)}"

    def _build_newbuilding_url(self, page: int = 1) -> str:
        """URL ЕКБ-выборки только новостроек (novostroyka-filter), сортировка по дате.

        Тот же shape параметров, что _build_citywide_url (s=104, p=page), но slug
        категории NOVOSTROYKA_SLUG отдаёт исключительно new-build карточки.
        """
        params = {"s": 104, "p": page}
        path = f"/ekaterinburg/kvartiry/prodam/{NOVOSTROYKA_SLUG}"
        return f"{self.base_url}{path}?{urlencode(params)}"

    def _build_rooms_url(
        self,
        room_slug: str,
        page: int = 1,
        min_price: int | None = None,
        max_price: int | None = None,
    ) -> str:
        """T6: URL с фильтром по комнатности для всего ЕКБ (no geo).

        Дробление по комнатности позволяет обойти cap ~5000 на category.
        room_slug — берётся из ROOM_SLUGS (кодирует ASgB-параметры выборки).

        min_price/max_price — опциональные ценовые границы (Avito `pmin`/`pmax`),
        добавляются в query ТОЛЬКО если заданы (backward-compat: вызовы без цены
        не меняют URL). Используются exhaustive full-load для room×price бисекции.
        """
        params: dict[str, Any] = {"s": 104, "p": page}
        if min_price is not None:
            params["pmin"] = min_price
        if max_price is not None:
            params["pmax"] = max_price
        return f"{self.base_url}/ekaterinburg/kvartiry/prodam/{room_slug}?{urlencode(params)}"

    def _extract_total_count(self, html: str) -> int | None:
        """Извлечь общее число результатов из Avito SERP (`page-title/count`).

        Avito рендерит счётчик в <span data-marker="page-title/count">1&nbsp;178</span>.
        Это аналог cian totalOffers — gate для бисекции (total > cap → split).
        Парсинг: текст span → убрать NBSP (\xa0)/пробелы/`&nbsp;` → int.

        Returns:
            int при успехе (включая 0), либо None при captcha/firewall/missing span.
        """
        m = _TOTAL_COUNT_RE.search(html)
        if m is None:
            return None
        raw = m.group("count")
        # Убрать NBSP (\xa0), &nbsp;-entity и обычные пробелы-разделители разрядов.
        cleaned = raw.replace("\xa0", "").replace("&nbsp;", "").replace(" ", "")
        if not cleaned:
            return None
        try:
            return int(cleaned)
        except (TypeError, ValueError):
            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,
    ) -> 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-запросы выполняются.

        Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url).
        """
        _buckets = rooms_buckets if rooms_buckets is not None else _AVITO_DEFAULT_ROOMS
        seen: dict[str, ScrapedLot] = {}

        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
                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,
                )
            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", len(seen))
        return list(seen.values())

    async def _walk_price_range(
        self,
        *,
        room_slug: str,
        room_label: str,
        lo: int,
        hi: int | None,
        seen: dict[str, ScrapedLot],
        price_cap_per_bucket: int,
        max_pages_per_bucket: int,
        concurrency: int = 5,
        secondary_only: bool = True,
        on_bucket: Callable[..., Any] | None = None,
        skip_buckets: set[str] | None = None,
        _depth: int = 0,
    ) -> None:
        """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].

        Алгоритм для ЗАКРЫТОГО брекета (hi is not None):
          1. Запросить page=1 с pmin=lo, pmax=hi → total (page-title/count).
          2. Если total <= cap → пагинировать leaf-бакет параллельно (concurrency).
          3. Если total > cap → разбить бакет пополам (рекурсия).
          Guard: hi - lo < _AVITO_MIN_BRACKET → пагинировать как есть (WARNING).

        ОТКРЫТЫЙ брекет (hi is None) — верхний seed-брекет без потолка (pmax не
        ставится, avito отдаёт всё ≥lo). Делить нельзя (mid=(lo+hi)//2 невозможен) —
        probe + пагинируем leaf напрямую. Хвост 250М+ крошечный (≈73), за cap
        обычно не выходит; если вышел — пагинируем до max_pages + WARNING (acceptable).

        После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots).
        bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый).
        on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон.
        skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются.
        secondary_only: новостройки (listing_segment=="novostroyki") отбрасываются после
          сбора bucket_lots, до дедупа в seen и вызова on_bucket.
        AvitoBlockedError/AvitoRateLimitedError из _fetch_serp_html пробрасываются наверх.
        """
        _lo_param = lo if lo > 0 else None
        _hi_repr = "open" if hi is None else str(hi)

        # ── Шаг 1: probe page 1 ────────────────────────────────────────────────
        # _probe_total ловит browser-ошибки (HTTP 500 / timeout от tradein-browser
        # на тяжёлой выдаче, напр. студии pmax=200M) → total=None вместо краха.
        # AvitoBlockedError/AvitoRateLimitedError всплывают наверх (mark_banned).
        html, total = await self._probe_total(room_slug, _lo_param, hi)

        # Ретрай на captcha/missing-count/browser-ошибке: rotate IP + 1 retry.
        if total is None:
            logger.warning(
                "avito: total=None for %s [%d, %s] depth=%d — rotating IP + retry",
                room_label,
                lo,
                _hi_repr,
                _depth,
            )
            rotated = await self._rotate_ip()
            if rotated:
                html, total = await self._probe_total(room_slug, _lo_param, hi)

        if total is None:
            # ОТКРЫТЫЙ брекет: делить нельзя (нет hi). Probe failed → нет total для
            # ceil(); пагинируем до max_pages_per_bucket (best-effort, хвост 250М+
            # крошечный) и WARNING. НЕ роняем прогон.
            if hi is None:
                logger.warning(
                    "avito: probe failed for OPEN bucket %s [%d, open] depth=%d — "
                    "paginating up to max_pages (tail-loss accepted)",
                    room_label,
                    lo,
                    _depth,
                )
                await self._paginate_leaf_bucket(
                    room_slug=room_slug,
                    room_label=room_label,
                    lo=lo,
                    hi=None,
                    html=None,
                    max_pages=max_pages_per_bucket,
                    seen=seen,
                    price_cap_per_bucket=price_cap_per_bucket,
                    max_pages_per_bucket=max_pages_per_bucket,
                    concurrency=concurrency,
                    secondary_only=secondary_only,
                    on_bucket=on_bucket,
                    skip_buckets=skip_buckets,
                )
                return
            bracket_size = hi - lo
            # Probe не удался даже после ретрая. Если бакет ещё делим — НЕ теряем
            # его целиком: тяжёлая страница (огромная выдача) скорее всего станет
            # загружаемой в более узком ценовом диапазоне → принудительный split.
            if bracket_size >= _AVITO_MIN_BRACKET:
                logger.warning(
                    "avito: probe failed for %s [%d, %d] depth=%d — splitting (assume heavy)",
                    room_label,
                    lo,
                    hi,
                    _depth,
                )
                mid = (lo + hi) // 2
                await self._walk_price_range(
                    room_slug=room_slug,
                    room_label=room_label,
                    lo=lo,
                    hi=mid,
                    seen=seen,
                    price_cap_per_bucket=price_cap_per_bucket,
                    max_pages_per_bucket=max_pages_per_bucket,
                    concurrency=concurrency,
                    secondary_only=secondary_only,
                    on_bucket=on_bucket,
                    skip_buckets=skip_buckets,
                    _depth=_depth + 1,
                )
                await self._walk_price_range(
                    room_slug=room_slug,
                    room_label=room_label,
                    lo=mid + 1,
                    hi=hi,
                    seen=seen,
                    price_cap_per_bucket=price_cap_per_bucket,
                    max_pages_per_bucket=max_pages_per_bucket,
                    concurrency=concurrency,
                    secondary_only=secondary_only,
                    on_bucket=on_bucket,
                    skip_buckets=skip_buckets,
                    _depth=_depth + 1,
                )
                return
            # Бакет уже узкий (< MIN_BRACKET) — split не поможет, хвост теряем
            # осознанно (как cian) и логируем.
            logger.error(
                "avito: skipping bucket %s [%d, %d] — total unavailable after retry "
                "(bracket=%d < MIN_BRACKET=%d, tail loss accepted)",
                room_label,
                lo,
                hi,
                bracket_size,
                _AVITO_MIN_BRACKET,
            )
            return

        logger.info(
            "avito: %s [%d, %s] total=%d depth=%d",
            room_label,
            lo,
            _hi_repr,
            total,
            _depth,
        )

        if total == 0:
            return

        # ── ОТКРЫТЫЙ брекет (hi is None): делить нельзя — пагинируем leaf напрямую ─
        # total известен из probe → max_pages = min(ceil(total/per_page), cap).
        # Если total > cap (не должно для 250М+, но guard) — пагинируем как есть до
        # max_pages_per_bucket с WARNING (хвост люкса крошечный, tail-loss accepted).
        if hi is None:
            if total > price_cap_per_bucket:
                logger.warning(
                    "avito: OPEN bucket %s [%d, open] total=%d > cap=%d — paginating "
                    "without split (tail-loss accepted, lux tail tiny)",
                    room_label,
                    lo,
                    total,
                    price_cap_per_bucket,
                )
            pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE)
            max_pages = min(pages_needed, max_pages_per_bucket)
            await self._paginate_leaf_bucket(
                room_slug=room_slug,
                room_label=room_label,
                lo=lo,
                hi=None,
                html=html,
                max_pages=max_pages,
                seen=seen,
                price_cap_per_bucket=price_cap_per_bucket,
                max_pages_per_bucket=max_pages_per_bucket,
                concurrency=concurrency,
                secondary_only=secondary_only,
                on_bucket=on_bucket,
                skip_buckets=skip_buckets,
            )
            return

        # ── Шаг 2: деление или пагинация ──────────────────────────────────────
        bracket_size = hi - lo
        need_split = total > price_cap_per_bucket
        too_narrow = bracket_size < _AVITO_MIN_BRACKET

        if need_split and too_narrow:
            logger.warning(
                "avito: %s [%d, %d] total=%d > cap=%d but bracket=%d < MIN_BRACKET=%d "
                "— paginating as-is (tail loss ~%d)",
                room_label,
                lo,
                hi,
                total,
                price_cap_per_bucket,
                bracket_size,
                _AVITO_MIN_BRACKET,
                max(0, total - price_cap_per_bucket),
            )
            need_split = False  # принудительно пагинируем

        if need_split:
            mid = (lo + hi) // 2
            # [lo, mid]
            await self._walk_price_range(
                room_slug=room_slug,
                room_label=room_label,
                lo=lo,
                hi=mid,
                seen=seen,
                price_cap_per_bucket=price_cap_per_bucket,
                max_pages_per_bucket=max_pages_per_bucket,
                concurrency=concurrency,
                secondary_only=secondary_only,
                on_bucket=on_bucket,
                skip_buckets=skip_buckets,
                _depth=_depth + 1,
            )
            # [mid+1, hi]
            await self._walk_price_range(
                room_slug=room_slug,
                room_label=room_label,
                lo=mid + 1,
                hi=hi,
                seen=seen,
                price_cap_per_bucket=price_cap_per_bucket,
                max_pages_per_bucket=max_pages_per_bucket,
                concurrency=concurrency,
                secondary_only=secondary_only,
                on_bucket=on_bucket,
                skip_buckets=skip_buckets,
                _depth=_depth + 1,
            )
            return

        # ── Закрытый leaf-бакет: считаем max_pages + warning, делегируем helper'у ─
        pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE)
        max_pages = min(pages_needed, max_pages_per_bucket)

        if pages_needed > max_pages_per_bucket:
            tail_loss = total - max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE
            logger.warning(
                "avito: %s [%d, %d] total=%d exceeds page cap "
                "(max_pages=%d × %d=%d offers) — tail loss ~%d offers",
                room_label,
                lo,
                hi,
                total,
                max_pages_per_bucket,
                _AVITO_OFFERS_PER_PAGE,
                max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE,
                tail_loss,
            )

        await self._paginate_leaf_bucket(
            room_slug=room_slug,
            room_label=room_label,
            lo=lo,
            hi=hi,
            html=html,
            max_pages=max_pages,
            seen=seen,
            price_cap_per_bucket=price_cap_per_bucket,
            max_pages_per_bucket=max_pages_per_bucket,
            concurrency=concurrency,
            secondary_only=secondary_only,
            on_bucket=on_bucket,
            skip_buckets=skip_buckets,
        )

    async def _paginate_leaf_bucket(
        self,
        *,
        room_slug: str,
        room_label: str,
        lo: int,
        hi: int | None,
        html: str | None,
        max_pages: int,
        seen: dict[str, ScrapedLot],
        price_cap_per_bucket: int,
        max_pages_per_bucket: int,
        concurrency: int,
        secondary_only: bool,
        on_bucket: Callable[..., Any] | None,
        skip_buckets: set[str] | None,
    ) -> None:
        """Параллельная пагинация одного leaf-бакета + фильтр + дедуп + on_bucket.

        Общая для ЗАКРЫТЫХ ([lo, hi]) и ОТКРЫТОГО (hi=None → pmax не ставится)
        брекетов. ``html`` — HTML страницы 1 из probe (переиспользуем, не
        перезапрашиваем); для открытого-без-probe передаётся None → стр.1 фетчится
        наравне с остальными. ``max_pages`` уже посчитан вызывающим.

        bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый).
        skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются.
        AvitoBlockedError/AvitoRateLimitedError из page-фетчей пробрасываются наверх.
        """
        _lo_param = lo if lo > 0 else None
        _hi_param = hi  # None → _build_rooms_url не ставит pmax
        _hi_repr = "open" if hi is None else str(hi)
        bucket_key = f"{room_label}:{lo}:{_hi_repr}"

        # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket.
        if skip_buckets and bucket_key in skip_buckets:
            logger.info("avito: skip bucket %s — already done (resume)", bucket_key)
            return

        # Страница 1 — из probe-html если он есть (закрытый/открытый с probe);
        # иначе фетчим её как обычную страницу (открытый брекет без probe-html).
        sem = asyncio.Semaphore(concurrency)
        first_url = self._build_rooms_url(room_slug, 1, _lo_param, _hi_param)

        async def _one_page(p: int) -> list[ScrapedLot]:
            if p == 1 and html is not None:
                return self._parse_html(html, source_url_base=first_url)
            async with sem:
                page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, _hi_param)
                await asyncio.sleep(self.request_delay_sec)
                if page_html is None:
                    logger.warning(
                        "avito: page_html=None %s [%d, %s] page=%d — skipping page",
                        room_label,
                        lo,
                        _hi_repr,
                        p,
                    )
                    return []
                page_url = self._build_rooms_url(room_slug, p, _lo_param, _hi_param)
                return self._parse_html(page_html, source_url_base=page_url)

        page_results = await asyncio.gather(
            *[_one_page(p) for p in range(1, max_pages + 1)],
            return_exceptions=True,
        )

        bucket_lots: list[ScrapedLot] = []
        for p_idx, res in enumerate(page_results, start=1):
            if isinstance(res, BaseException):
                # Блокировки пробрасываем наверх (mark_banned в pipeline-обёртке).
                if isinstance(res, AvitoBlockedError | AvitoRateLimitedError):
                    raise res
                logger.warning(
                    "avito: page exception %s [%d, %s] page=%d — %r",
                    room_label,
                    lo,
                    _hi_repr,
                    p_idx,
                    res,
                )
            else:
                bucket_lots.extend(res)

        collected_this_bucket = len(bucket_lots)

        # ── Фильтр новостроек (secondary_only) ────────────────────────────────
        dropped_nb = 0
        if secondary_only:
            filtered = [lot for lot in bucket_lots if lot.listing_segment != "novostroyki"]
            dropped_nb = collected_this_bucket - len(filtered)
            bucket_lots = filtered

        # Дедуп в общий seen
        for lot in bucket_lots:
            key = lot.source_id or lot.source_url
            if key:
                seen[key] = lot

        logger.info(
            "avito: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d",
            room_label,
            lo,
            _hi_repr,
            max_pages,
            collected_this_bucket,
            dropped_nb,
            len(seen),
        )

        # ── on_bucket callback: инкрементальный save ──────────────────────────
        if on_bucket is not None and bucket_lots:
            res_cb = on_bucket(bucket_key, bucket_lots)
            if inspect.isawaitable(res_cb):
                await res_cb

    # ── T6: citywide / byrooms fetch methods ──────────────────────────────────

    async def fetch_city_wide(
        self,
        pages: int = 100,
        *,
        delay_override_sec: float | None = None,
    ) -> list[ScrapedLot]:
        """T6: Обход всего ЕКБ без geo-фильтра (citywide-mode), paginated.

        Возвращает дедуплицированный список лотов (по source_id/source_url).
        Break-on-empty: останавливается когда страница отдаёт 0 карточек.
        Сохраняет весь anti-block pipeline (_fetch_serp_html: firewall-detect,
        IP rotation, транзиентные ретраи). Не трогает fetch_around.

        Args:
            pages: максимальное число страниц (default 100 = Avito cap ~5000 лотов).
            delay_override_sec: если задан — переопределяет request_delay_sec для
                этого вызова.

        Returns:
            Список ScrapedLot (все страницы, дедуп по source_id).
        """
        all_lots = await self._paginate_sweep(
            pages,
            self._build_citywide_url,
            label="citywide",
            delay_override_sec=delay_override_sec,
        )
        logger.info("avito fetch_city_wide pages=%d total_lots=%d", pages, len(all_lots))
        return all_lots

    async def _paginate_sweep(
        self,
        pages: int,
        url_builder: Callable[[int], str],
        *,
        label: str,
        delay_override_sec: float | None = None,
    ) -> list[ScrapedLot]:
        """Общий paginated-обход ЕКБ (citywide / novostroyka) с break-on-empty.

        url_builder(page) — функция построения URL страницы (citywide или
        novostroyka). Сохраняет anti-block pipeline (_fetch_serp_html: firewall,
        IP rotation, ретраи), дедуп по source_id, break-on-empty. label — только
        для логов. Не меняет наблюдаемое поведение fetch_city_wide.
        """
        if delay_override_sec is not None:
            self.request_delay_sec = delay_override_sec

        all_lots: list[ScrapedLot] = []
        seen_ids: set[str] = set()

        for page in range(1, pages + 1):
            url = url_builder(page)
            try:
                html = await self._fetch_serp_html_with_retry(url, page)
            except (AvitoBlockedError, AvitoRateLimitedError):
                raise
            except Exception:
                logger.exception(
                    "avito %s page=%d fetch failed after retries url=%s", label, page, url
                )
                break
            if html is None:
                logger.info("avito %s page=%d: non-200 — end of pagination", label, page)
                break

            lots = self._parse_html(html, source_url_base=url)
            if not lots:
                logger.info("avito %s page=%d: 0 lots — end of pagination", label, page)
                break

            new_lots = [lot for lot in lots if lot.source_id not in seen_ids]
            for lot in new_lots:
                if lot.source_id:
                    seen_ids.add(lot.source_id)
            all_lots.extend(new_lots)
            logger.info(
                "avito %s page=%d: %d lots (%d new, %d dedup'd, total=%d)",
                label,
                page,
                len(lots),
                len(new_lots),
                len(lots) - len(new_lots),
                len(all_lots),
            )
            if page < pages:
                await self.sleep_between_requests()
        return all_lots

    async def fetch_newbuildings(
        self,
        pages: int = 30,
        *,
        delay_override_sec: float | None = None,
    ) -> list[ScrapedLot]:
        """Обход ЕКБ-выборки только новостроек (novostroyka-filter), paginated.

        Avito отдаёт dedicated SERP с фильтром «Новостройка» — 100% new-build
        карточек (с маркером застройщика → listing_segment="novostroyki",
        newbuilding_id/newbuilding_url заполнены). Citywide (без geo/anchor).
        Возвращает дедуплицированный список (по source_id), break-on-empty.
        Сохраняет весь anti-block pipeline. Не трогает fetch_city_wide/fetch_around.

        Args:
            pages: максимальное число страниц (default 30).
            delay_override_sec: если задан — переопределяет request_delay_sec для
                этого вызова.

        Returns:
            Список ScrapedLot новостроек (все страницы, дедуп по source_id).
        """
        all_lots = await self._paginate_sweep(
            pages,
            self._build_newbuilding_url,
            label="newbuilding",
            delay_override_sec=delay_override_sec,
        )
        logger.info("avito fetch_newbuildings pages=%d total_lots=%d", pages, len(all_lots))
        return all_lots

    async def fetch_by_rooms(
        self,
        pages: int = 100,
        *,
        delay_override_sec: float | None = None,
        room_slugs: list[tuple[str, str]] | None = None,
    ) -> list[ScrapedLot]:
        """T6: Обход ЕКБ по категориям комнатности (byrooms-mode).

        Дробление по 7 категориям комнатности позволяет обойти Avito cap ~5000
        результатов на поиск: каждый сегмент < 5000. Итого до 7 × 5000 = 35k лотов.
        Break-on-empty на каждой категории. Глобальный dedup по source_id.

        Args:
            pages: страниц на категорию (default 100).
            delay_override_sec: переопределяет request_delay_sec.
            room_slugs: список (name, slug) для обхода; по умолчанию ROOM_SLUGS
                (все 7 категорий). Передай подмножество для частичного обхода.

        Returns:
            Список ScrapedLot (все категории, глобальный дедуп по source_id).
        """
        if delay_override_sec is not None:
            self.request_delay_sec = delay_override_sec

        slugs = room_slugs if room_slugs is not None else ROOM_SLUGS
        all_lots: list[ScrapedLot] = []
        seen_ids: set[str] = set()

        for ci, (name, slug) in enumerate(slugs, 1):
            logger.info(
                "avito byrooms [%d/%d] category=%s slug_prefix=%.30s",
                ci,
                len(slugs),
                name,
                slug,
            )
            cat_count = 0
            for page in range(1, pages + 1):
                url = self._build_rooms_url(slug, page)
                try:
                    html = await self._fetch_serp_html_with_retry(url, page)
                except (AvitoBlockedError, AvitoRateLimitedError):
                    raise
                except Exception:
                    logger.exception(
                        "avito byrooms category=%s page=%d fetch failed after retries url=%s",
                        name,
                        page,
                        url,
                    )
                    break
                if html is None:
                    logger.info("avito byrooms category=%s page=%d: non-200 — end", name, page)
                    break

                lots = self._parse_html(html, source_url_base=url)
                if not lots:
                    logger.info(
                        "avito byrooms category=%s page=%d: 0 lots — end of category",
                        name,
                        page,
                    )
                    break

                new_lots = [lot for lot in lots if lot.source_id not in seen_ids]
                for lot in new_lots:
                    if lot.source_id:
                        seen_ids.add(lot.source_id)
                all_lots.extend(new_lots)
                cat_count += len(new_lots)
                logger.info(
                    "avito byrooms category=%s page=%d: %d lots (%d new, total=%d)",
                    name,
                    page,
                    len(lots),
                    len(new_lots),
                    len(all_lots),
                )
                if page < pages:
                    await self.sleep_between_requests()

            logger.info(
                "avito byrooms category=%s done: cat_new=%d grand_total=%d",
                name,
                cat_count,
                len(all_lots),
            )

        logger.info("avito fetch_by_rooms pages_per_cat=%d total_lots=%d", pages, len(all_lots))
        return all_lots

    def _parse_html(self, html: str, source_url_base: str) -> list[ScrapedLot]:
        """Парсим карточки объявлений из HTML через DOM scrape.

        Поля карточки берём из DOM (`data-marker="item-*"`). Дату публикации — из
        embedded per-item JSON (`sortTimeStamp`), т.к. DOM `item-date` рендерится
        лениво только у верхних карточек (#726); DOM-marker остаётся fallback.
        """
        tree = HTMLParser(html)
        ts_map = _build_sort_timestamp_map(html)
        cards = tree.css('[data-marker="item"]')
        lots: list[ScrapedLot] = []
        for card in cards:
            lot = self._dom_card_to_lot(card, source_url_base, ts_map)
            if lot is not None:
                lots.append(lot)
        return lots

    def _dom_card_to_lot(
        self, card: Any, source_url_base: str, ts_map: dict[str, date] | None = None
    ) -> ScrapedLot | None:
        """Парсинг карточки объявления через DOM.

        Avito state давно пустой — единственный путь. Coords из карточки не отдаются,
        оставляем lat=lon=None (geocode-missing cron подтянет из address).
        """
        try:
            link_el = card.css_first('a[data-marker="item-title"]')
            if link_el is None:
                return None
            href = link_el.attributes.get("href", "")
            url = urljoin("https://www.avito.ru", href)
            title = link_el.text(strip=True)

            price_el = card.css_first('meta[itemprop="price"]')
            price = int(price_el.attributes.get("content", 0)) if price_el is not None else 0
            if price <= 0:
                return None

            rooms = _extract_rooms_from_title(title)
            area = _extract_area_from_title(title)
            floor, total = _extract_floor_from_title(title)

            # Адрес: первый <p> внутри data-marker="item-location"
            address: str | None = None
            loc_el = card.css_first('[data-marker="item-location"]')
            if loc_el is not None:
                street_el = loc_el.css_first("p")
                if street_el is not None:
                    address = _clean_address(street_el.text(strip=True))

            # Фото
            photo_urls: list[str] = []
            for img in card.css('img[itemprop="image"]'):
                src = img.attributes.get("src") or ""
                if src.startswith("http") and src not in photo_urls:
                    photo_urls.append(src)
                if len(photo_urls) >= 5:
                    break

            # Дата публикации (#726): primary — sortTimeStamp из per-item JSON по
            # data-item-id (DOM item-date рендерится лениво только у верхних карточек),
            # fallback — DOM marker item-date/item-date-info + relative-date parse.
            listing_date: date | None = None
            item_id = card.attributes.get("data-item-id")
            if ts_map and item_id:
                listing_date = ts_map.get(item_id)
            if listing_date is None:
                date_el = card.css_first('[data-marker="item-date"]')
                if date_el is None:
                    date_el = card.css_first('[data-marker="item-date-info"]')
                if date_el is not None:
                    listing_date = _parse_relative_date(date_el.text(strip=True))
            if listing_date is None:
                # Fallback: ищем любой текст «N дней/недель/месяцев назад» в карточке
                card_text = card.text(strip=True)
                listing_date = _parse_relative_date(card_text)

            # House link: <a href="/catalog/houses/<city>/<slug>/<id>">
            # or /catalog/houses/<slug>/<id>
            house_source: str | None = None
            house_ext_id: str | None = None
            house_url: str | None = None
            house_link = card.css_first('a[href^="/catalog/houses/"]')
            if house_link is not None:
                h_href = house_link.attributes.get("href", "")
                parts = [p for p in h_href.strip("/").split("/") if p]
                # ['catalog', 'houses', <maybe city>, <slug>, <id>]
                if len(parts) >= 4 and parts[-1].isdigit():
                    house_source = "avito"
                    house_ext_id = parts[-1]
                    house_url = urljoin("https://www.avito.ru", h_href)

            # listing_segment по DOM-маркеру карточки: новостройки несут
            # data-marker="item-development-name" (название ЖК/застройщика),
            # вторичка — нет. URL-паттерн ненадёжен: каждый avito-URL квартиры
            # содержит /kvartiry/ (вкл. новостройки) → ветка всегда vtorichka.
            dev_name = card.css_first('[data-marker="item-development-name"]')
            listing_segment = "novostroyki" if dev_name is not None else "vtorichka"

            # Newbuilding (ЖК) link: новостройки несут якорь на ЖК-сабдомен
            # <a href="https://zhk-<slug>-ekaterinburg.avito.ru?context=...">.
            # newbuilding_url — host+path без ?context (mirror urlparse-cleaning),
            # newbuilding_id — <slug> = host без 'zhk-' префикса и '.avito.ru' суффикса.
            newbuilding_id: str | None = None
            newbuilding_url: str | None = None
            if dev_name is not None:
                nb_link = card.css_first('a[href*="zhk-"][href*=".avito.ru"]')
                if nb_link is not None:
                    nb_href = nb_link.attributes.get("href", "") or ""
                    parsed = urlparse(nb_href)
                    host = parsed.netloc
                    if host.startswith("zhk-") and host.endswith(".avito.ru"):
                        # strip query/fragment, keep scheme+host+path
                        newbuilding_url = urlunparse((parsed.scheme, host, parsed.path, "", "", ""))
                        newbuilding_id = host[len("zhk-") : -len(".avito.ru")]

            return ScrapedLot(
                source="avito",
                source_url=url,
                source_id=card.attributes.get("data-item-id"),
                address=address,  # БЕЗ fallback "Екатеринбург (Avito)" — пусть None
                lat=None,  # C-5 fix: НЕ jitter
                lon=None,  # C-5 fix: НЕ jitter
                rooms=rooms,
                area_m2=area,
                floor=floor,
                total_floors=total,
                price_rub=price,
                photo_urls=photo_urls,
                listing_date=listing_date,
                raw_payload={"title": title, "address": address, "house_ext_id": house_ext_id},
                house_source=house_source,
                house_ext_id=house_ext_id,
                house_url=house_url,
                listing_segment=listing_segment,
                newbuilding_id=newbuilding_id,
                newbuilding_url=newbuilding_url,
            )
        except Exception:
            # #823: логируем каждую сломанную карточку и считаем для observability.
            # Карточки с неожиданной DOM-структурой скипаем, но не роняем весь парс.
            self.parse_failures += 1
            item_id_attr = getattr(card, "attributes", {}).get("data-item-id", "?")
            logger.warning(
                "avito _dom_card_to_lot: failed to parse card item_id=%s (parse_failures=%d)",
                item_id_attr,
                self.parse_failures,
                exc_info=True,
            )
            return None


# ── Helpers: title parser ───────────────────────────────────────────────────
_RE_ROOMS = re.compile(r"(\d)-к\.?\s*(квартира|кв\.)", re.IGNORECASE)
_RE_STUDIO = re.compile(r"студи[яиюей]", re.IGNORECASE)
_RE_AREA = re.compile(r"(\d+[.,]?\d*)\s*м[²2]", re.IGNORECASE)
_RE_FLOOR = re.compile(r"(\d+)\s*/\s*(\d+)\s*эт\.?", re.IGNORECASE)

# ── T6: citywide / byrooms sweep ──────────────────────────────────────────────
# Avito URL slugs для room-filter (извлечены из реальных search-страниц 2026-05-23).
# Каждый slug кодирует ASgB-параметры выборки комнатности.
ROOM_SLUGS: list[tuple[str, str]] = [
    ("студии", "studii-ASgBAgICAkSSA8YQygj~WA"),
    ("1-комн.", "1-komnatnye-ASgBAgICAkSSA8YQygiAWQ"),
    ("2-комн.", "2-komnatnye-ASgBAgICAkSSA8YQygiCWQ"),
    ("3-комн.", "3-komnatnye-ASgBAgICAkSSA8YQygiEWQ"),
    ("4-комн.", "4-komnatnye-ASgBAgICAkSSA8YQygiGWQ"),
    ("5+комн.", "5-komnatnye-ASgBAgICAkSSA8YQygiIWQ"),
    ("своб.планир.", "svobodnaya_planirovka-ASgBAgICAkSSA8YQygj8zzI"),
]

# Дефолтные комнатности для exhaustive full-load — все ROOM_SLUGS (вкл. студии и
# своб.планировку), т.к. каждая категория < SERP-cap после room×price бисекции.
_AVITO_DEFAULT_ROOMS: list[tuple[str, str]] = list(ROOM_SLUGS)

_NON_SLUG_CHARS_RE = re.compile(r"[^a-z0-9]+")


def _avito_room_label(name: str) -> str:
    """Стабильный bucket_key-префикс из человекочитаемого имени комнатности.

    «1-комн.» → «room_1_komn», «студии» → «room_studii», «5+комн.» → «room_5_komn».
    Транслитерация не нужна — берём латиницу/цифры из slug-friendly формы; для
    кириллицы используем порядковый индекс категории как fallback стабильности.
    """
    # Цифры из имени (комнатность) + транслит ключевых маркеров для читабельности логов.
    digits = "".join(ch for ch in name if ch.isdigit())
    base = name.lower()
    if "студ" in base:
        return "room_studii"
    if "планир" in base:
        return "room_svobodnaya"
    if digits:
        return f"room_{digits}_komn"
    # Fallback: нормализуем что есть (на случай новых категорий).
    norm = _NON_SLUG_CHARS_RE.sub("_", base).strip("_")
    return f"room_{norm or 'unknown'}"


_CSS_NOISE_RE = re.compile(r"\.?css-[a-z0-9_-]+\s*\{[^}]*\}", flags=re.I)
_NOT_ADDRESS_TAIL_RE = re.compile(
    r"\s*(Площадь \d|от \d+\s?мин\.|css-[a-z0-9_-]+)",
    flags=re.I,
)


def _clean_address(raw: str | None) -> str | None:
    """Strip Emotion CSS and post-address noise that Avito DOM leaks via .text().

    Examples:
        "ул. Токарей, 56к1Площадь 1905 года.css-39hgr0{fill:..."
            -> "ул. Токарей, 56к1"
        "ул. Малышева, 1.css-xxx{...}"
            -> "ул. Малышева, 1"
    """
    if not raw:
        return None
    cleaned = _CSS_NOISE_RE.sub("", raw)
    cleaned = _NOT_ADDRESS_TAIL_RE.split(cleaned, maxsplit=1)[0]
    cleaned = cleaned.strip(" ,.\n\t")
    return cleaned or None


def _extract_rooms_from_title(title: str) -> int | None:
    if _RE_STUDIO.search(title):
        return 0
    m = _RE_ROOMS.search(title)
    if m:
        return int(m.group(1))
    return None


def _extract_area_from_title(title: str) -> float | None:
    m = _RE_AREA.search(title)
    if m:
        return float(m.group(1).replace(",", "."))
    return None


def _extract_floor_from_title(title: str) -> tuple[int | None, int | None]:
    m = _RE_FLOOR.search(title)
    if m:
        return int(m.group(1)), int(m.group(2))
    return None, None