fix(tradein/avito): anti-bot hardening — sleep + abort + shared session (#487)

tradein-mvp/backend/app/api/v1/admin.py

@@ -556,7 +556,7 @@ async def scrape_avito_imv(
 
 class CitySweepStartRequest(BaseModel):
     pages_per_anchor: int = Field(default=3, ge=1, le=10)
-    detail_top_n: int = Field(default=20, ge=0, le=30)
+    detail_top_n: int = Field(default=10, ge=0, le=30)
     request_delay_sec: float = Field(default=7.0, ge=3.0, le=15.0)
     enrich_houses: bool = True
     radius_m: int = Field(default=1500, ge=500, le=5000)

tradein-mvp/backend/app/services/scrape_pipeline.py

@@ -10,23 +10,34 @@
 City sweep: run_avito_city_sweep — итерирует EKB_ANCHORS × pages_per_anchor,
 с cooperative cancel через scrape_runs.status и heartbeat update каждый anchor.
 
+Anti-bot hardening (2023-05-23):
+  - Shared AsyncSession на весь sweep (один TLS handshake)
+  - asyncio.sleep с random +-20% jitter между detail requests
+  - AvitoBlockedError/AvitoRateLimitedError propagation → early abort
+  - После 3 consecutive blocks — abort sweep + mark_banned (status='banned')
+
 Graceful degradation на каждом step: exception в одной house/listing
-не валит весь pipeline.
+не валит весь pipeline (кроме blocked exceptions — они abort весь sweep).
 
 IMV ОТСУТСТВУЕТ — он on-demand (Stage 3 estimator integration), не cron.
 """
 
 from __future__ import annotations
 
+import asyncio
 import logging
+import random
 from dataclasses import dataclass, field, fields
 from urllib.parse import urlparse
 
+from curl_cffi.requests import AsyncSession
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.services import scrape_runs
 from app.services.scrapers.avito import AvitoScraper
 from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
+from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
 from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment
 from app.services.scrapers.base import ScrapedLot, save_listings
 
@@ -41,8 +52,19 @@ EKB_ANCHORS: list[tuple[float, float, str]] = [
     (56.8650, 60.6200, "Пионерский"),
 ]
 
+_CHROME_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",
+}
 
-# ── Result dataclasses ──────────────────────────────────────────────────────
+
+# ── Result dataclasses ──────────────────────────────────────────
 
 
 @dataclass
@@ -73,7 +95,7 @@ class PipelineResult:
     enrich_detail_top_n: int
 
 
-# ── Main orchestrator ───────────────────────────────────────────────────────
+# ── Main orchestrator ───────────────────────────────────────────
 
 
 async def run_avito_pipeline(
@@ -86,6 +108,7 @@ async def run_avito_pipeline(
     enrich_detail_top_n: int = 10,
     pages: int = 1,
     request_delay_sec: float | None = None,
+    shared_session: AsyncSession | None = None,
 ) -> PipelineResult:
     """Full Avito search → houses → detail enrichment pipeline.
 
@@ -93,42 +116,51 @@ async def run_avito_pipeline(
       1. SEARCH: AvitoScraper().fetch_around(lat, lon, radius_m, pages=pages)
       2. SAVE: save_listings(db, lots) → (inserted, updated)
       3. GROUP: dedupe house_url из новых lots → unique house path set
-      4. ENRICH_HOUSES (если enrich_houses=True): для каждого unique house path
-         → fetch_house_catalog + save_house_catalog_enrichment (try/except per house)
-      5. ENRICH_DETAIL (если enrich_detail_top_n > 0): select top-N priority listings
-         (приоритет — detail_enriched_at IS NULL + price > 0)
-         → fetch_detail + save_detail_enrichment (try/except per listing)
-      6. Return PipelineResult с counters
+      4. ENRICH_HOUSES: fetch_house_catalog + save per house (with sleep between)
+      5. ENRICH_DETAIL: top-N listings → fetch_detail + save (abort on 3 blocks)
 
-    pages=1 (default) — backward compat, одна страница.
-    pages>1 — pagination (city sweep). request_delay_sec — override задержки между страницами.
-
-    Graceful degradation на каждом step — exception на одной house/listing
-    не должна провалить весь pipeline.
+    shared_session — если передан, используется без закрытия (lifecycle у вызывающего).
+    AvitoBlockedError / AvitoRateLimitedError propagate наружу.
     """
     counters = PipelineCounters()
     lots: list[ScrapedLot] = []
+    detail_delay = request_delay_sec if request_delay_sec is not None else 7.0
 
-    # ── Step 1: search ──────────────────────────────────────────────────────
-    async with AvitoScraper() as scraper:
+    own_session = shared_session is None
+    session = shared_session or AsyncSession(
+        impersonate="chrome120",
+        timeout=25,
+        headers=_CHROME_HEADERS,
+    )
+
+    try:
+        # ── Step 1: search ──────────────────────────────────────
+        scraper = AvitoScraper()
+        scraper._cffi = session
         try:
             lots = await scraper.fetch_around(
-                lat, lon, radius_m, pages=pages, delay_override_sec=request_delay_sec
+                lat, lon, radius_m,
+                pages=pages,
+                delay_override_sec=request_delay_sec,
             )
             counters.lots_fetched = len(lots)
             logger.info(
                 "pipeline:search anchor=(%.4f,%.4f) r=%dm fetched=%d",
-                lat,
-                lon,
-                radius_m,
-                len(lots),
+                lat, lon, radius_m, len(lots),
             )
+        except (AvitoBlockedError, AvitoRateLimitedError):
+            logger.error(
+                "pipeline:search BLOCKED by Avito at anchor=(%.4f,%.4f)", lat, lon
+            )
+            raise
         except Exception as e:
             logger.exception("pipeline:search failed")
             counters.errors.append(f"search: {e}")
-            return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n)
+            return PipelineResult(
+                lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n
+            )
 
-    # ── Step 2: save listings ───────────────────────────────────────────────
+        # ── Step 2: save listings ───────────────────────────────
         if lots:
             try:
                 counters.lots_inserted, counters.lots_updated = save_listings(db, lots)
@@ -136,8 +168,7 @@ async def run_avito_pipeline(
                 logger.exception("pipeline:save_listings failed")
                 counters.errors.append(f"save_listings: {e}")
 
-    # ── Step 3: group by house ──────────────────────────────────────────────
-    # Dedupe house path (без host) для consistency; fetch_house_catalog принимает path
+        # ── Step 3: group by house ──────────────────────────────
         unique_house_paths: set[str] = set()
         for lot in lots:
             if lot.house_url:
@@ -151,13 +182,19 @@ async def run_avito_pipeline(
         counters.unique_houses = len(unique_house_paths)
         logger.info("pipeline:group_by_house unique=%d", counters.unique_houses)
 
-    # ── Step 4: enrich houses ───────────────────────────────────────────────
+        # ── Step 4: enrich houses с shared session + sleep ──────
         if enrich_houses and unique_house_paths:
-        for house_path in unique_house_paths:
+            house_paths_list = list(unique_house_paths)
+            for idx, house_path in enumerate(house_paths_list):
                 try:
-                enrichment = await fetch_house_catalog(house_path)
+                    enrichment = await fetch_house_catalog(house_path, cffi_session=session)
                     save_house_catalog_enrichment(db, enrichment)
                     counters.houses_enriched += 1
+                except (AvitoBlockedError, AvitoRateLimitedError):
+                    logger.error(
+                        "pipeline:house BLOCKED at %s — propagating", house_path
+                    )
+                    raise
                 except Exception as e:
                     logger.warning(
                         "pipeline:house_enrich failed for %s: %s", house_path, e
@@ -165,11 +202,11 @@ async def run_avito_pipeline(
                     counters.houses_failed += 1
                     counters.errors.append(f"house {house_path}: {e}")
 
-    # ── Step 5: enrich detail для top-N priority listings ──────────────────
+                if idx < len(house_paths_list) - 1:
+                    await asyncio.sleep(detail_delay)
+
+        # ── Step 5: enrich detail для top-N priority listings ───
         if enrich_detail_top_n > 0:
-        # Priority: листинги ещё не enriched (detail_enriched_at IS NULL)
-        # в радиусе 2× от anchor ИЛИ только что scraped (последние 2 часа).
-        # OR обёрнут в скобки — приоритет AND выше OR.
             priority_rows = db.execute(
                 text("""
                     SELECT source_url
@@ -202,43 +239,65 @@ async def run_avito_pipeline(
                 },
             ).mappings().all()
 
-        for row in priority_rows:
+            consecutive_blocks = 0
+            for idx, row in enumerate(priority_rows):
                 source_url: str = row["source_url"]
                 counters.detail_attempted += 1
                 try:
-                # fetch_detail принимает absolute URL или relative path
                     item_url = (
                         urlparse(source_url).path
                         if source_url.startswith("http")
                         else source_url
                     )
-                enrichment_detail = await fetch_detail(item_url)
+                    enrichment_detail = await fetch_detail(item_url, cffi_session=session)
                     if save_detail_enrichment(db, enrichment_detail):
                         counters.detail_enriched += 1
+                    consecutive_blocks = 0
+                except (AvitoBlockedError, AvitoRateLimitedError) as e:
+                    consecutive_blocks += 1
+                    counters.detail_failed += 1
+                    counters.errors.append(f"BLOCKED_DETAIL #{idx}: {e}")
+                    logger.warning(
+                        "pipeline:detail BLOCKED #%d/%d (consecutive=%d): %s",
+                        idx + 1, len(priority_rows), consecutive_blocks, e,
+                    )
+                    if consecutive_blocks >= 3:
+                        logger.error(
+                            "pipeline:detail ABORT — %d consecutive blocks, IP rate-limited",
+                            consecutive_blocks,
+                        )
+                        counters.errors.append(
+                            f"AVITO_RATE_LIMITED — sweep aborted after "
+                            f"{idx + 1}/{len(priority_rows)} details"
+                        )
+                        raise
                 except Exception as e:
+                    counters.detail_failed += 1
+                    counters.errors.append(f"detail {source_url}: {e}")
                     logger.warning(
                         "pipeline:detail_enrich failed for %s: %s", source_url, e
                     )
-                counters.detail_failed += 1
-                counters.errors.append(f"detail {source_url}: {e}")
+                    consecutive_blocks = 0
+
+                if idx < len(priority_rows) - 1:
+                    jitter = random.uniform(0.8, 1.2)
+                    await asyncio.sleep(detail_delay * jitter)
 
         logger.info(
             "pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) "
             "houses=%d/%d detail=%d/%d errors=%d",
-        lat,
-        lon,
-        counters.lots_fetched,
-        counters.lots_inserted,
-        counters.lots_updated,
-        counters.houses_enriched,
-        counters.unique_houses,
-        counters.detail_enriched,
-        counters.detail_attempted,
+            lat, lon,
+            counters.lots_fetched, counters.lots_inserted, counters.lots_updated,
+            counters.houses_enriched, counters.unique_houses,
+            counters.detail_enriched, counters.detail_attempted,
             len(counters.errors),
         )
-
         return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n)
 
+    finally:
+        if own_session:
+            await session.close()
+
 
 @dataclass
 class CitySweepCounters:
@@ -269,32 +328,31 @@ async def run_avito_city_sweep(
     radius_m: int = 1500,
     pages_per_anchor: int = 3,
     enrich_houses: bool = True,
-    detail_top_n: int = 20,
+    detail_top_n: int = 10,
     request_delay_sec: float = 7.0,
 ) -> CitySweepCounters:
     """Full city sweep: iterate anchors × pages → save → enrich houses + detail.
 
-    Алгоритм:
-      - Для каждого anchor: run_avito_pipeline(pages=pages_per_anchor)
-      - Перед каждым anchor проверяет scrape_runs.status — если 'cancelled' → stop
-      - Heartbeat update каждый anchor с текущими counters
-      - Errors per-anchor логируются, не валят весь sweep
-      - По завершению вызывает mark_done; при исключении — mark_failed
+    - Single shared AsyncSession на весь sweep (один TLS fingerprint)
+    - AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned')
+    - Прочие errors per-anchor логируются, не валят весь sweep
     """
-    from app.services import scrape_runs as runs
-
     _anchors = anchors if anchors is not None else EKB_ANCHORS
     counters = CitySweepCounters(anchors_total=len(_anchors))
 
+    async with AsyncSession(
+        impersonate="chrome120",
+        timeout=25,
+        headers=_CHROME_HEADERS,
+    ) as session:
         try:
             for idx, (lat, lon, name) in enumerate(_anchors, start=1):
-            # Cooperative cancel: проверяем статус перед каждым anchor
-            if runs.is_cancelled(db, run_id):
+                if scrape_runs.is_cancelled(db, run_id):
                     logger.info(
-                    "city-sweep run_id=%d: cancelled by signal at anchor #%d/%d (%s)",
+                        "city-sweep run_id=%d: cancelled at anchor #%d/%d (%s)",
                         run_id, idx, len(_anchors), name,
                     )
-                runs.update_heartbeat(db, run_id, counters.to_dict())
+                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                     return counters
 
                 logger.info(
@@ -311,6 +369,7 @@ async def run_avito_city_sweep(
                         enrich_detail_top_n=detail_top_n,
                         pages=pages_per_anchor,
                         request_delay_sec=request_delay_sec,
+                        shared_session=session,
                     )
                     counters.lots_fetched += result.counters.lots_fetched
                     counters.lots_inserted += result.counters.lots_inserted
@@ -322,14 +381,26 @@ async def run_avito_city_sweep(
                     counters.detail_enriched += result.counters.detail_enriched
                     counters.detail_failed += result.counters.detail_failed
                     counters.errors_count += len(result.counters.errors)
+                except (AvitoBlockedError, AvitoRateLimitedError) as e:
+                    logger.error(
+                        "city-sweep run_id=%d ABORT at anchor #%d/%d (%s) — blocked: %s",
+                        run_id, idx, len(_anchors), name, e,
+                    )
+                    counters.errors_count += 1
+                    counters.anchors_done = idx
+                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+                    scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict())
+                    return counters
                 except Exception:
-                logger.exception("city-sweep run_id=%d: anchor %s failed", run_id, name)
+                    logger.exception(
+                        "city-sweep run_id=%d: anchor %s failed", run_id, name
+                    )
                     counters.errors_count += 1
 
                 counters.anchors_done = idx
-            runs.update_heartbeat(db, run_id, counters.to_dict())
+                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
 
-        runs.mark_done(db, run_id, counters.to_dict())
+            scrape_runs.mark_done(db, run_id, counters.to_dict())
             logger.info(
                 "city-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) "
                 "houses=%d/%d detail=%d/%d errors=%d",
@@ -349,11 +420,11 @@ async def run_avito_city_sweep(
 
         except Exception as exc:
             logger.exception("city-sweep run_id=%d: fatal error", run_id)
-        runs.mark_failed(db, run_id, str(exc), counters.to_dict())
+            scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
             raise
 
 
-# ── Public re-exports ───────────────────────────────────────────────────────
+# ── Public re-exports ───────────────────────────────────────────
 __all__ = [
     "EKB_ANCHORS",
     "CitySweepCounters",

tradein-mvp/backend/app/services/scrape_runs.py

@@ -63,33 +63,77 @@ def is_cancelled(db: Session, run_id: int) -> bool:
 
 def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
     """Финализация run: status='done', finished_at=NOW(), counters."""
-    db.execute(
+    row = db.execute(
         text(
             """
             UPDATE scrape_runs
             SET status = 'done', finished_at = NOW(), heartbeat_at = NOW(),
                 counters = CAST(:counters AS jsonb)
             WHERE id = :run_id AND status = 'running'
+            RETURNING id
             """
         ),
         {"run_id": run_id, "counters": json.dumps(counters)},
-    )
+    ).first()
+    if row is None:
+        logger.warning("mark_done no-op: run_id=%d not in 'running' state", run_id)
     db.commit()
 
 
 def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None:
-    """Финализация run: status='failed', error (первые 1000 символов)."""
-    db.execute(
+    """Финализация run: status='failed', error (первые 1000 символов).
+
+    Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE,
+    он мог оставить сессию в error state — rollback сбрасывает состояние.
+    """
+    try:
+        db.rollback()  # defensive: cascade-safe if prior UPDATE left txn in error state
+    except Exception:
+        pass
+    row = db.execute(
         text(
             """
             UPDATE scrape_runs
             SET status = 'failed', finished_at = NOW(), heartbeat_at = NOW(),
                 error = :error, counters = CAST(:counters AS jsonb)
             WHERE id = :run_id AND status = 'running'
+            RETURNING id
             """
         ),
         {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)},
-    )
+    ).first()
+    if row is None:
+        logger.warning("mark_failed no-op: run_id=%d not in 'running' state", run_id)
+    db.commit()
+
+
+def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None:
+    """Финализация run: status='banned' (IP заблокирован Avito — 403/captcha).
+
+    Per migration 015 — 'banned' задокументирован как 'Avito вернул 403/captcha'.
+    Отличается от 'failed': это external constraint, не наш bug. Cooldown 2-4 часа.
+
+    Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE,
+    он мог оставить сессию в error state — rollback сбрасывает состояние.
+    """
+    try:
+        db.rollback()  # defensive: cascade-safe if prior UPDATE left txn in error state
+    except Exception:
+        pass
+    row = db.execute(
+        text(
+            """
+            UPDATE scrape_runs
+            SET status = 'banned', finished_at = NOW(), heartbeat_at = NOW(),
+                error = :error, counters = CAST(:counters AS jsonb)
+            WHERE id = :run_id AND status = 'running'
+            RETURNING id
+            """
+        ),
+        {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)},
+    ).first()
+    if row is None:
+        logger.warning("mark_banned no-op: run_id=%d not in 'running' state", run_id)
     db.commit()
 
 

tradein-mvp/backend/app/services/scrapers/avito.py

@@ -24,6 +24,7 @@ from curl_cffi.requests import AsyncSession
 from selectolax.parser import HTMLParser
 
 from app.services.scraper_settings import get_scraper_delay
+from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
 from app.services.scrapers.base import BaseScraper, ScrapedLot
 
 logger = logging.getLogger(__name__)
@@ -101,11 +102,27 @@ class AvitoScraper(BaseScraper):
             try:
                 assert self._cffi is not None
                 response = await self._cffi.get(url)
+                if response.status_code == 403:
+                    logger.error(
+                        "avito SERP HTTP 403 (IP blocked) page=%d url=%s", page, url
+                    )
+                    raise AvitoBlockedError(
+                        f"Avito SERP returned 403 — IP blocked at page={page}"
+                    )
+                if response.status_code == 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}"
+                    )
                 if response.status_code != 200:
                     logger.warning(
                         "avito HTML page=%d returned %d for %s", page, response.status_code, url
                     )
                     break
+            except (AvitoBlockedError, AvitoRateLimitedError):
+                raise
             except Exception:
                 logger.exception("avito HTML page=%d fetch failed for %s", page, url)
                 break

tradein-mvp/backend/app/services/scrapers/avito_detail.py

@@ -32,6 +32,8 @@ from selectolax.parser import HTMLParser, Node
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
+
 logger = logging.getLogger(__name__)
 
 AVITO_BASE = "https://www.avito.ru"
@@ -252,6 +254,10 @@ async def fetch_detail(
     full_url = item_url if item_url.startswith("http") else urljoin(AVITO_BASE, item_url)
     try:
         response = await session.get(full_url)
+        if response.status_code == 403:
+            raise AvitoBlockedError(f"Avito detail HTTP 403 for {full_url}")
+        if response.status_code == 429:
+            raise AvitoRateLimitedError(f"Avito detail HTTP 429 for {full_url}")
         if response.status_code != 200:
             raise ValueError(
                 f"avito detail HTTP {response.status_code} for {full_url}"

tradein-mvp/backend/app/services/scrapers/avito_exceptions.py

@@ -0,0 +1,17 @@
+"""Avito-specific exceptions для anti-bot detection."""
+
+
+class AvitoError(Exception):
+    """Base для всех Avito scraper exceptions."""
+
+
+class AvitoBlockedError(AvitoError):
+    """HTTP 403 от Avito — IP-level block detected."""
+
+
+class AvitoRateLimitedError(AvitoError):
+    """HTTP 429 от Avito — rate limit triggered."""
+
+
+class AvitoParseError(AvitoError):
+    """Cannot parse Avito HTML structure (selector changes)."""

tradein-mvp/backend/app/services/scrapers/avito_houses.py

@@ -32,6 +32,8 @@ from typing import Any
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
+
 logger = logging.getLogger(__name__)
 
 AVITO_BASE = "https://www.avito.ru"
@@ -621,6 +623,10 @@ async def fetch_house_catalog(
         logger.info("Houses Catalog fetch: %s", url)
 
         resp = await cffi_session.get(url)
+        if resp.status_code == 403:
+            raise AvitoBlockedError(f"Avito houses HTTP 403 for {url}")
+        if resp.status_code == 429:
+            raise AvitoRateLimitedError(f"Avito houses HTTP 429 for {url}")
         resp.raise_for_status()
         html: str = resp.text
 

tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py

@@ -0,0 +1,229 @@
+"""Test anti-bot hardening: blocked exceptions + early abort."""
+
+from unittest.mock import AsyncMock, MagicMock, call, patch
+
+import pytest
+
+from app.services.scrapers.avito_exceptions import (
+    AvitoBlockedError,
+    AvitoRateLimitedError,
+)
+
+# ── fetch_detail tests ──────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_raises_blocked_on_403() -> None:
+    """fetch_detail должен raise AvitoBlockedError на HTTP 403."""
+    from app.services.scrapers.avito_detail import fetch_detail
+
+    mock_session = AsyncMock()
+    mock_response = MagicMock()
+    mock_response.status_code = 403
+    mock_session.get = AsyncMock(return_value=mock_response)
+
+    with pytest.raises(AvitoBlockedError, match="403"):
+        await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session)
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_raises_rate_limited_on_429() -> None:
+    """fetch_detail должен raise AvitoRateLimitedError на HTTP 429."""
+    from app.services.scrapers.avito_detail import fetch_detail
+
+    mock_session = AsyncMock()
+    mock_response = MagicMock()
+    mock_response.status_code = 429
+    mock_session.get = AsyncMock(return_value=mock_response)
+
+    with pytest.raises(AvitoRateLimitedError, match="429"):
+        await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session)
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_raises_value_error_on_other_non_200() -> None:
+    """fetch_detail должен raise ValueError на прочие non-200 (не 403/429)."""
+    from app.services.scrapers.avito_detail import fetch_detail
+
+    mock_session = AsyncMock()
+    mock_response = MagicMock()
+    mock_response.status_code = 503
+    mock_session.get = AsyncMock(return_value=mock_response)
+
+    with pytest.raises(ValueError, match="503"):
+        await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session)
+
+
+# ── SERP (avito.py) tests ───────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_fetch_around_raises_blocked_on_403() -> None:
+    """AvitoScraper.fetch_around должен raise AvitoBlockedError на HTTP 403."""
+    from app.services.scrapers.avito import AvitoScraper
+
+    scraper = AvitoScraper()
+    mock_session = AsyncMock()
+    mock_response = MagicMock()
+    mock_response.status_code = 403
+    mock_session.get = AsyncMock(return_value=mock_response)
+    scraper._cffi = mock_session
+
+    with pytest.raises(AvitoBlockedError):
+        await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
+
+
+@pytest.mark.asyncio
+async def test_fetch_around_raises_rate_limited_on_429() -> None:
+    """AvitoScraper.fetch_around должен raise AvitoRateLimitedError на HTTP 429."""
+    from app.services.scrapers.avito import AvitoScraper
+
+    scraper = AvitoScraper()
+    mock_session = AsyncMock()
+    mock_response = MagicMock()
+    mock_response.status_code = 429
+    mock_session.get = AsyncMock(return_value=mock_response)
+    scraper._cffi = mock_session
+
+    with pytest.raises(AvitoRateLimitedError):
+        await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
+
+
+# ── Pipeline abort tests ────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_pipeline_aborts_after_3_consecutive_blocks() -> None:
+    """Pipeline должен прервать sweep после 3 consecutive 403 (re-raise)."""
+    from app.services.scrape_pipeline import run_avito_pipeline
+
+    mock_db = MagicMock()
+    mock_scraper = MagicMock()
+    mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper)
+    mock_scraper.__aexit__ = AsyncMock(return_value=None)
+    mock_scraper.fetch_around = AsyncMock(return_value=[])
+    mock_scraper._cffi = None
+
+    mock_row = MagicMock()
+    mock_row.__getitem__ = lambda self, key: (
+        "https://www.avito.ru/ekaterinburg/kvartiry/test-1" if key == "source_url" else None
+    )
+    mock_mappings = MagicMock()
+    mock_mappings.all.return_value = [mock_row, mock_row, mock_row, mock_row]
+    mock_execute = MagicMock()
+    mock_execute.mappings.return_value = mock_mappings
+    mock_db.execute.return_value = mock_execute
+
+    block_error = AvitoBlockedError("HTTP 403")
+
+    with (
+        patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper),
+        patch(
+            "app.services.scrape_pipeline.fetch_detail",
+            AsyncMock(side_effect=block_error),
+        ),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+        patch(
+            "curl_cffi.requests.AsyncSession",
+            return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()),
+        ),
+    ):
+        with pytest.raises(AvitoBlockedError):
+            await run_avito_pipeline(
+                mock_db, 56.8, 60.6, 1000,
+                enrich_houses=False,
+                enrich_detail_top_n=4,
+            )
+
+
+@pytest.mark.asyncio
+async def test_pipeline_sleep_between_detail_requests() -> None:
+    """Pipeline должен вызывать asyncio.sleep между каждым detail request."""
+    from app.services.scrape_pipeline import run_avito_pipeline
+
+    mock_db = MagicMock()
+    mock_scraper = MagicMock()
+    mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper)
+    mock_scraper.__aexit__ = AsyncMock(return_value=None)
+    mock_scraper.fetch_around = AsyncMock(return_value=[])
+    mock_scraper._cffi = None
+
+    mock_row = MagicMock()
+    mock_row.__getitem__ = lambda self, key: (
+        "https://www.avito.ru/ekaterinburg/kvartiry/test-1" if key == "source_url" else None
+    )
+    mock_mappings = MagicMock()
+    mock_mappings.all.return_value = [mock_row, mock_row, mock_row]
+    mock_execute = MagicMock()
+    mock_execute.mappings.return_value = mock_mappings
+    mock_db.execute.return_value = mock_execute
+
+    mock_enrichment = MagicMock()
+    sleep_mock = AsyncMock()
+
+    with (
+        patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper),
+        patch(
+            "app.services.scrape_pipeline.fetch_detail",
+            AsyncMock(return_value=mock_enrichment),
+        ),
+        patch("app.services.scrape_pipeline.save_detail_enrichment", return_value=True),
+        patch("app.services.scrape_pipeline.asyncio.sleep", sleep_mock),
+        patch(
+            "curl_cffi.requests.AsyncSession",
+            return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()),
+        ),
+        patch("app.services.scrape_pipeline.random.uniform", return_value=1.0),
+    ):
+        await run_avito_pipeline(
+            mock_db, 56.8, 60.6, 1000,
+            enrich_houses=False,
+            enrich_detail_top_n=3,
+            request_delay_sec=7.0,
+        )
+
+    assert sleep_mock.call_count == 2
+    for c in sleep_mock.call_args_list:
+        assert c == call(7.0)
+
+
+@pytest.mark.asyncio
+async def test_city_sweep_marks_banned_on_block() -> None:
+    """City sweep mark'ит run как banned при AvitoBlockedError.
+
+    Migration 015 задокументировал 'banned' как 'Avito вернул 403/captcha' — именно этот сценарий.
+    """
+    from app.services.scrape_pipeline import run_avito_city_sweep
+
+    mock_db = MagicMock()
+    mock_runs = MagicMock()
+    mock_runs.is_cancelled.return_value = False
+    mock_runs.update_heartbeat = MagicMock()
+    mock_runs.mark_banned = MagicMock()
+    mock_runs.mark_done = MagicMock()
+
+    block_error = AvitoBlockedError("IP blocked at page=1")
+
+    with (
+        patch("app.services.scrape_pipeline.scrape_runs", mock_runs),
+        patch(
+            "app.services.scrape_pipeline.run_avito_pipeline",
+            AsyncMock(side_effect=block_error),
+        ),
+        patch(
+            "curl_cffi.requests.AsyncSession",
+            return_value=AsyncMock(
+                __aenter__=AsyncMock(return_value=AsyncMock()),
+                __aexit__=AsyncMock(return_value=None),
+            ),
+        ),
+    ):
+        result = await run_avito_city_sweep(
+            mock_db,
+            run_id=42,
+            anchors=[(56.84, 60.60, "Центр"), (56.79, 60.53, "ЮЗ")],
+        )
+
+    assert mock_runs.mark_banned.called
+    assert not mock_runs.mark_done.called
+    assert result.anchors_done == 1