"""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 logging
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

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

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"


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

    # ── 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) -> str:
        """T6: URL с фильтром по комнатности для всего ЕКБ (no geo).

        Дробление по комнатности позволяет обойти cap ~5000 на category.
        room_slug — берётся из ROOM_SLUGS (кодирует ASgB-параметры выборки).
        """
        params = {"s": 104, "p": page}
        return f"{self.base_url}/ekaterinburg/kvartiry/prodam/{room_slug}?{urlencode(params)}"

    # ── 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(url, page)
            except (AvitoBlockedError, AvitoRateLimitedError):
                raise
            except Exception:
                logger.exception("avito %s page=%d fetch failed 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(url, page)
                except (AvitoBlockedError, AvitoRateLimitedError):
                    raise
                except Exception:
                    logger.exception(
                        "avito byrooms category=%s page=%d fetch failed 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"),
]

_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