feat(tradein): city sweep — auto ЕКБ pipeline (#477)

Stage 4d: full city sweep ЕКБ (5 anchors × N pages + enrichment + cooperative cancel).

- avito.py fetch_around(pages=N) — pagination backward compat default=1
- scrape_pipeline.run_avito_city_sweep — anchors loop + heartbeat + cancel check
- scrape_runs.py utility (create/heartbeat/done/failed/cancelled/list_recent)
- admin.py 3 endpoints: start (BackgroundTasks) / runs / cancel
- migration 051 ADD COLUMN IF NOT EXISTS + DROP/ADD CHECK constraint (text+CHECK, не enum — safe в transaction)
- deploy/avito-city-sweep.sh с random delay 0-3h (anti-pattern)
- 16 offline tests

Deep review approved: migration idempotent, cancel priority correct (mark_done guarded by status='running'), no SQL injection, psycopg v3 compliant.

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

@@ -9,15 +9,17 @@ import logging
 import time
 from typing import Annotated, Literal
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
 from pydantic import BaseModel, Field
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
-from app.core.db import get_db
+from app.core.db import SessionLocal, get_db
 from app.services import cian_session as cian_session_svc
+from app.services import scrape_runs as runs_mod
 from app.services.geocoder import geocode
+from app.services.scrape_pipeline import run_avito_city_sweep
 from app.services.scrapers.avito import AvitoScraper
 from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
 from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment
@@ -37,15 +39,24 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
+_ALL_SOURCES = ["avito", "cian", "yandex", "n1"]
+
+
 class ScrapeRequest(BaseModel):
     lat: float
     lon: float
     radius_m: int = Field(default=1000, ge=100, le=20000)
-    sources: list[Literal["avito", "cian", "yandex", "n1"]] = Field(default_factory=lambda: ["avito", "cian", "yandex", "n1"])
+    sources: list[Literal["avito", "cian", "yandex", "n1"]] = Field(
-    multi_room_yandex: bool = False  # Если True — скрейп Yandex по 5 сегментам комнат, не одним общим запросом.
+        default_factory=lambda: list(_ALL_SOURCES)
-    deep_yandex: bool = False  # Если True — Yandex полный обход 5 rooms × 3 sorts × 2 pages = 30 запросов / ~150s.
+    )
-    multi_room_cian: bool = False  # Если True — скрейп Cian по 4 сегментам комнат отдельно (~4× лотов).
+    # Если True — скрейп Yandex по 5 сегментам комнат, не одним общим запросом.
-    multi_room_n1: bool = False  # Если True — скрейп N1 по 4 сегментам комнат отдельно (~4× лотов).
+    multi_room_yandex: bool = False
+    # Если True — Yandex полный обход 5 rooms × 3 sorts × 2 pages = 30 запросов / ~150s.
+    deep_yandex: bool = False
+    # Если True — скрейп Cian по 4 сегментам комнат отдельно (~4× лотов).
+    multi_room_cian: bool = False
+    # Если True — скрейп N1 по 4 сегментам комнат отдельно (~4× лотов).
+    multi_room_n1: bool = False
 
 
 class ScrapeResult(BaseModel):
@@ -424,3 +435,109 @@ async def scrape_avito_imv(
         "evaluation_id": eval_id,
         "history_saved": history_saved,
     }
+
+
+# ── City sweep: background full-ЕКБ pipeline ────────────────────────────────
+
+
+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)
+    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)
+
+
+class CitySweepStartResponse(BaseModel):
+    run_id: int
+    status: str
+    pages_per_anchor: int
+    detail_top_n: int
+
+
+class ScrapeRunRow(BaseModel):
+    run_id: int
+    source: str
+    status: str
+    params: dict | None = None
+    counters: dict | None = None
+    error: str | None = None
+    started_at: str | None = None
+    finished_at: str | None = None
+    heartbeat_at: str | None = None
+
+
+@router.post("/scrape/avito-city-sweep", response_model=CitySweepStartResponse)
+async def start_avito_city_sweep(
+    payload: CitySweepStartRequest,
+    background_tasks: BackgroundTasks,
+    db: Annotated[Session, Depends(get_db)],
+) -> CitySweepStartResponse:
+    """Запустить full ЕКБ city sweep в background. Returns run_id для polling/cancel.
+
+    Sweep итерирует 5 anchor-точек ЕКБ × pages_per_anchor страниц Avito,
+    сохраняет listings, обогащает houses и detail. Runs in FastAPI BackgroundTasks.
+
+    Coop cancel: POST /scrape/avito-city-sweep/{run_id}/cancel помечает run,
+    pipeline проверяет статус каждый anchor.
+    """
+    run_id = runs_mod.create_run(db, source="avito_city_sweep", params=payload.model_dump())
+
+    async def _sweep_task() -> None:
+        sweep_db = SessionLocal()
+        try:
+            await run_avito_city_sweep(
+                sweep_db,
+                run_id=run_id,
+                radius_m=payload.radius_m,
+                pages_per_anchor=payload.pages_per_anchor,
+                enrich_houses=payload.enrich_houses,
+                detail_top_n=payload.detail_top_n,
+                request_delay_sec=payload.request_delay_sec,
+            )
+        except Exception:
+            logger.exception("city-sweep background task run_id=%d crashed", run_id)
+        finally:
+            sweep_db.close()
+
+    background_tasks.add_task(_sweep_task)
+    logger.info("city-sweep queued run_id=%d params=%s", run_id, payload.model_dump())
+    return CitySweepStartResponse(
+        run_id=run_id,
+        status="running",
+        pages_per_anchor=payload.pages_per_anchor,
+        detail_top_n=payload.detail_top_n,
+    )
+
+
+@router.get("/scrape/avito-city-sweep/runs", response_model=list[ScrapeRunRow])
+def list_avito_city_sweep_runs(
+    db: Annotated[Session, Depends(get_db)],
+    limit: int = 10,
+) -> list[ScrapeRunRow]:
+    """Список последних N runs (для UI polling). Default limit=10."""
+    rows = runs_mod.list_recent(db, source="avito_city_sweep", limit=limit)
+    return [
+        ScrapeRunRow(
+            run_id=r["run_id"],
+            source=r["source"],
+            status=r["status"],
+            params=r.get("params"),
+            counters=r.get("counters"),
+            error=r.get("error"),
+            started_at=r["started_at"].isoformat() if r.get("started_at") else None,
+            finished_at=r["finished_at"].isoformat() if r.get("finished_at") else None,
+            heartbeat_at=r["heartbeat_at"].isoformat() if r.get("heartbeat_at") else None,
+        )
+        for r in rows
+    ]
+
+
+@router.post("/scrape/avito-city-sweep/{run_id}/cancel")
+def cancel_avito_city_sweep(
+    run_id: int,
+    db: Annotated[Session, Depends(get_db)],
+) -> dict[str, object]:
+    """Отменить running sweep. Cooperative: pipeline проверяет scrape_runs.status каждый anchor."""
+    cancelled = runs_mod.mark_cancelled(db, run_id)
+    return {"ok": True, "run_id": run_id, "cancelled": cancelled}

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

@@ -1,12 +1,15 @@
-"""scrape_pipeline.py — Avito full orchestrator (Stage 2e).
+"""scrape_pipeline.py — Avito full orchestrator (Stage 2e + city sweep).
 
 Координирует 5 шагов одного pipeline run:
-  1. SEARCH   — AvitoScraper().fetch_around(lat, lon, radius_m)
+  1. SEARCH   — AvitoScraper().fetch_around(lat, lon, radius_m, pages=N)
   2. SAVE     — save_listings(db, lots) → (inserted, updated)
   3. GROUP    — dedupe house_url из lots → unique set
   4. ENRICH_HOUSES — fetch_house_catalog + save_house_catalog_enrichment per unique house
   5. ENRICH_DETAIL — top-N listings (detail_enriched_at IS NULL) → fetch_detail + save
 
+City sweep: run_avito_city_sweep — итерирует EKB_ANCHORS × pages_per_anchor,
+с cooperative cancel через scrape_runs.status и heartbeat update каждый anchor.
+
 Graceful degradation на каждом step: exception в одной house/listing
 не валит весь pipeline.
 
@@ -16,7 +19,7 @@ IMV ОТСУТСТВУЕТ — он on-demand (Stage 3 estimator integration),
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, fields
 from urllib.parse import urlparse
 
 from sqlalchemy import text
@@ -29,6 +32,15 @@ from app.services.scrapers.base import ScrapedLot, save_listings
 
 logger = logging.getLogger(__name__)
 
+# Default anchors ЕКБ — 5 точек покрытия города
+EKB_ANCHORS: list[tuple[float, float, str]] = [
+    (56.8400, 60.6050, "Центр"),
+    (56.7950, 60.5300, "ЮЗ"),
+    (56.8970, 60.6100, "Уралмаш"),
+    (56.7700, 60.5500, "Академический"),
+    (56.8650, 60.6200, "Пионерский"),
+]
+
 
 # ── Result dataclasses ──────────────────────────────────────────────────────
 
@@ -72,11 +84,13 @@ async def run_avito_pipeline(
     *,
     enrich_houses: bool = True,
     enrich_detail_top_n: int = 10,
+    pages: int = 1,
+    request_delay_sec: float | None = None,
 ) -> PipelineResult:
     """Full Avito search → houses → detail enrichment pipeline.
 
     Steps:
-      1. SEARCH: AvitoScraper().fetch_around(lat, lon, radius_m) → list[ScrapedLot]
+      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
@@ -86,6 +100,9 @@ async def run_avito_pipeline(
          → fetch_detail + save_detail_enrichment (try/except per listing)
       6. Return PipelineResult с counters
 
+    pages=1 (default) — backward compat, одна страница.
+    pages>1 — pagination (city sweep). request_delay_sec — override задержки между страницами.
+
     Graceful degradation на каждом step — exception на одной house/listing
     не должна провалить весь pipeline.
     """
@@ -95,7 +112,9 @@ async def run_avito_pipeline(
     # ── Step 1: search ──────────────────────────────────────────────────────
     async with AvitoScraper() as scraper:
         try:
-            lots = await scraper.fetch_around(lat, lon, radius_m)
+            lots = await scraper.fetch_around(
+                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",
@@ -221,9 +240,125 @@ async def run_avito_pipeline(
     return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n)
 
 
+@dataclass
+class CitySweepCounters:
+    """Aggregate counters для full city sweep run."""
+
+    anchors_total: int = 0
+    anchors_done: int = 0
+    lots_fetched: int = 0
+    lots_inserted: int = 0
+    lots_updated: int = 0
+    unique_houses: int = 0
+    houses_enriched: int = 0
+    houses_failed: int = 0
+    detail_attempted: int = 0
+    detail_enriched: int = 0
+    detail_failed: int = 0
+    errors_count: int = 0
+
+    def to_dict(self) -> dict[str, int]:
+        return {f.name: getattr(self, f.name) for f in fields(self)}
+
+
+async def run_avito_city_sweep(
+    db: Session,
+    *,
+    run_id: int,
+    anchors: list[tuple[float, float, str]] | None = None,
+    radius_m: int = 1500,
+    pages_per_anchor: int = 3,
+    enrich_houses: bool = True,
+    detail_top_n: int = 20,
+    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
+    """
+    from app.services import scrape_runs as runs
+
+    _anchors = anchors if anchors is not None else EKB_ANCHORS
+    counters = CitySweepCounters(anchors_total=len(_anchors))
+
+    try:
+        for idx, (lat, lon, name) in enumerate(_anchors, start=1):
+            # Cooperative cancel: проверяем статус перед каждым anchor
+            if runs.is_cancelled(db, run_id):
+                logger.info(
+                    "city-sweep run_id=%d: cancelled by signal at anchor #%d/%d (%s)",
+                    run_id, idx, len(_anchors), name,
+                )
+                runs.update_heartbeat(db, run_id, counters.to_dict())
+                return counters
+
+            logger.info(
+                "city-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
+                run_id, idx, len(_anchors), name, lat, lon,
+            )
+            try:
+                result = await run_avito_pipeline(
+                    db,
+                    lat=lat,
+                    lon=lon,
+                    radius_m=radius_m,
+                    enrich_houses=enrich_houses,
+                    enrich_detail_top_n=detail_top_n,
+                    pages=pages_per_anchor,
+                    request_delay_sec=request_delay_sec,
+                )
+                counters.lots_fetched += result.counters.lots_fetched
+                counters.lots_inserted += result.counters.lots_inserted
+                counters.lots_updated += result.counters.lots_updated
+                counters.unique_houses += result.counters.unique_houses
+                counters.houses_enriched += result.counters.houses_enriched
+                counters.houses_failed += result.counters.houses_failed
+                counters.detail_attempted += result.counters.detail_attempted
+                counters.detail_enriched += result.counters.detail_enriched
+                counters.detail_failed += result.counters.detail_failed
+                counters.errors_count += len(result.counters.errors)
+            except Exception:
+                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())
+
+        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",
+            run_id,
+            counters.anchors_done,
+            counters.anchors_total,
+            counters.lots_fetched,
+            counters.lots_inserted,
+            counters.lots_updated,
+            counters.houses_enriched,
+            counters.unique_houses,
+            counters.detail_enriched,
+            counters.detail_attempted,
+            counters.errors_count,
+        )
+        return counters
+
+    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())
+        raise
+
+
 # ── Public re-exports ───────────────────────────────────────────────────────
 __all__ = [
+    "EKB_ANCHORS",
+    "CitySweepCounters",
     "PipelineCounters",
     "PipelineResult",
+    "run_avito_city_sweep",
     "run_avito_pipeline",
 ]

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

@@ -0,0 +1,128 @@
+"""Utility functions для scrape_runs table — tracking long-running pipeline runs.
+
+Таблица scrape_runs создана в 015_scrape_runs.sql.
+Расширена в 051_scrape_runs_extend.sql: params/counters/error/finished_at/cancelled.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+
+def create_run(db: Session, *, source: str, params: dict[str, Any]) -> int:
+    """INSERT scrape_runs(source, status='running', params, started_at=NOW()).
+
+    run_type DEFAULT 'city_sweep' (из 051 миграции).
+    Returns run_id (bigint).
+    """
+    row = db.execute(
+        text(
+            """
+            INSERT INTO scrape_runs (source, status, params, started_at, heartbeat_at)
+            VALUES (:source, 'running', CAST(:params AS jsonb), NOW(), NOW())
+            RETURNING id
+            """
+        ),
+        {"source": source, "params": json.dumps(params, ensure_ascii=False)},
+    ).fetchone()
+    db.commit()
+    assert row is not None, "scrape_runs INSERT returned no id"
+    return int(row.id)
+
+
+def update_heartbeat(db: Session, run_id: int, counters: dict[str, int]) -> None:
+    """UPDATE heartbeat_at=NOW(), counters=:counters."""
+    db.execute(
+        text(
+            """
+            UPDATE scrape_runs
+            SET heartbeat_at = NOW(), counters = CAST(:counters AS jsonb)
+            WHERE id = :run_id
+            """
+        ),
+        {"run_id": run_id, "counters": json.dumps(counters)},
+    )
+    db.commit()
+
+
+def is_cancelled(db: Session, run_id: int) -> bool:
+    """Проверить status='cancelled' (cooperative cancel в long-running pipeline)."""
+    row = db.execute(
+        text("SELECT status FROM scrape_runs WHERE id = :id"),
+        {"id": run_id},
+    ).fetchone()
+    return row is not None and row.status == "cancelled"
+
+
+def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
+    """Финализация run: status='done', finished_at=NOW(), counters."""
+    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'
+            """
+        ),
+        {"run_id": run_id, "counters": json.dumps(counters)},
+    )
+    db.commit()
+
+
+def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None:
+    """Финализация run: status='failed', error (первые 1000 символов)."""
+    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'
+            """
+        ),
+        {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)},
+    )
+    db.commit()
+
+
+def mark_cancelled(db: Session, run_id: int) -> bool:
+    """Set status='cancelled' если currently 'running'. Returns True если cancelled."""
+    result = db.execute(
+        text(
+            """
+            UPDATE scrape_runs
+            SET status = 'cancelled', finished_at = NOW()
+            WHERE id = :run_id AND status = 'running'
+            RETURNING id
+            """
+        ),
+        {"run_id": run_id},
+    ).fetchone()
+    db.commit()
+    return result is not None
+
+
+def list_recent(db: Session, *, source: str, limit: int = 10) -> list[dict[str, Any]]:
+    """Список последних N runs по source, в порядке убывания started_at."""
+    rows = db.execute(
+        text(
+            """
+            SELECT id AS run_id, source, status, params, counters, error,
+                   started_at, finished_at, heartbeat_at
+            FROM scrape_runs
+            WHERE source = :source
+            ORDER BY started_at DESC NULLS LAST
+            LIMIT :limit
+            """
+        ),
+        {"source": source, "limit": limit},
+    ).mappings().all()
+    return [dict(r) for r in rows]

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

@@ -68,36 +68,67 @@ class AvitoScraper(BaseScraper):
 
     # ── Public ──────────────────────────────────────────────────────────────
     async def fetch_around(
-        self, lat: float, lon: float, radius_m: int = 1000
+        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 км.
-        Возвращаем макс 50 объявлений (страница 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
 
-        url = self._build_web_url(lat, lon, radius_km, page=1)
+        all_lots: list[ScrapedLot] = []
-        try:
+        for page in range(1, pages + 1):
-            assert self._cffi is not None
+            url = self._build_web_url(lat, lon, radius_km, page=page)
-            response = await self._cffi.get(url)
+            try:
-            if response.status_code != 200:
+                assert self._cffi is not None
-                logger.warning(
+                response = await self._cffi.get(url)
-                    "avito HTML returned %d for %s", response.status_code, url
+                if response.status_code != 200:
-                )
+                    logger.warning(
-                return []
+                        "avito HTML page=%d returned %d for %s", page, response.status_code, url
-        except Exception:
+                    )
-            logger.exception("avito HTML fetch failed for %s", url)
+                    break
-            return []
+            except Exception:
+                logger.exception("avito HTML page=%d fetch failed for %s", page, url)
+                break
 
-        lots = self._parse_html(response.text, source_url_base=url)
+            lots = self._parse_html(response.text, source_url_base=url)
-        logger.info("avito html_scrape: %d lots around (%.4f, %.4f)", len(lots), lat, lon)
+            if not lots:
-        await self.sleep_between_requests()
+                logger.info("avito page=%d: 0 lots — end of pagination", page)
-        return lots
+                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:

tradein-mvp/backend/data/sql/051_scrape_runs_extend.sql

@@ -0,0 +1,57 @@
+-- 051_scrape_runs_extend.sql
+-- Расширение таблицы scrape_runs для поддержки city sweep pipeline:
+--   - params jsonb      — параметры запуска (radius_m, pages_per_anchor, …)
+--   - counters jsonb    — агрегированные счётчики (aggregate anchors_done, lots_fetched, …)
+--   - error text        — краткое сообщение об ошибке (alias для error_text)
+--   - finished_at       — время завершения run (done/failed/cancelled)
+--   - 'cancelled' в status CHECK
+--
+-- Dependencies: 015_scrape_runs.sql
+-- Идемпотентно: безопасно запускать повторно (ALTER ADD COLUMN IF NOT EXISTS).
+
+BEGIN;
+
+-- 1. Добавить params jsonb — конфигурация запуска
+ALTER TABLE scrape_runs
+    ADD COLUMN IF NOT EXISTS params jsonb;
+
+-- 2. Добавить counters jsonb — агрегированные счётчики (city sweep)
+ALTER TABLE scrape_runs
+    ADD COLUMN IF NOT EXISTS counters jsonb;
+
+-- 3. Добавить finished_at (было только в error_text / heartbeat_at)
+ALTER TABLE scrape_runs
+    ADD COLUMN IF NOT EXISTS finished_at timestamptz;
+
+-- 4. Добавить error text (alias для краткого сообщения, совместимо с scrape_runs utility module)
+--    error_text уже существует (из 015) — добавляем error как синоним для новых runs.
+--    Utility module пишет в 'error' — перенаправляем на error_text через generated column
+--    НЕ работает в PostGIS/PG16 для text → используем просто второй столбец.
+ALTER TABLE scrape_runs
+    ADD COLUMN IF NOT EXISTS error text;
+
+-- 5. Расширить CHECK constraint status чтобы допускал 'cancelled'
+-- DROP + ADD (нельзя ALTER CHECK в Postgres без пересоздания)
+ALTER TABLE scrape_runs
+    DROP CONSTRAINT IF EXISTS scrape_runs_status_check;
+
+ALTER TABLE scrape_runs
+    ADD CONSTRAINT scrape_runs_status_check
+    CHECK (status IN ('running', 'done', 'failed', 'zombie', 'skipped', 'banned', 'cancelled'));
+
+-- 6. run_type имеет NOT NULL в 015 — для city sweep run_type = 'city_sweep'
+--    Убедимся что INSERT в scrape_runs.py покрывает это через DEFAULT
+ALTER TABLE scrape_runs
+    ALTER COLUMN run_type SET DEFAULT 'city_sweep';
+
+COMMENT ON COLUMN scrape_runs.params IS
+    'JSON-параметры запуска pipeline (pages_per_anchor, radius_m, detail_top_n, …)';
+COMMENT ON COLUMN scrape_runs.counters IS
+    'JSON-счётчики прогресса (anchors_done, lots_fetched, houses_enriched, …). '
+    'Обновляется heartbeat''ом каждый anchor.';
+COMMENT ON COLUMN scrape_runs.finished_at IS
+    'Время завершения run (status=done/failed/cancelled).';
+COMMENT ON COLUMN scrape_runs.error IS
+    'Краткое сообщение об ошибке (для status=failed). Первые 1000 символов.';
+
+COMMIT;

tradein-mvp/backend/tests/test_city_sweep.py

@@ -0,0 +1,234 @@
+"""Offline smoke tests for city sweep utilities + endpoints.
+
+Тесты НЕ требуют реального DB — мокируем SessionLocal и scrape_runs модуль.
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+# ── EKB_ANCHORS ────────────────────────────────────────────────────────────
+
+
+def test_ekb_anchors_count() -> None:
+    from app.services.scrape_pipeline import EKB_ANCHORS
+
+    assert len(EKB_ANCHORS) == 5
+    for lat, lon, name in EKB_ANCHORS:
+        assert 56.5 < lat < 57.0, f"lat={lat} out of EKB range for anchor {name}"
+        assert 60.3 < lon < 60.8, f"lon={lon} out of EKB range for anchor {name}"
+        assert isinstance(name, str) and name
+
+
+# ── CitySweepCounters ───────────────────────────────────────────────────────
+
+
+def test_city_sweep_counters_defaults() -> None:
+    from app.services.scrape_pipeline import CitySweepCounters
+
+    c = CitySweepCounters()
+    assert c.anchors_total == 0
+    assert c.lots_fetched == 0
+    assert c.errors_count == 0
+
+
+def test_city_sweep_counters_to_dict() -> None:
+    from app.services.scrape_pipeline import CitySweepCounters
+
+    c = CitySweepCounters(anchors_total=5, anchors_done=2, lots_fetched=150)
+    d = c.to_dict()
+    assert d["anchors_total"] == 5
+    assert d["anchors_done"] == 2
+    assert d["lots_fetched"] == 150
+    assert "errors_count" in d
+    assert "houses_enriched" in d
+    assert "detail_enriched" in d
+
+
+def test_city_sweep_counters_to_dict_all_keys() -> None:
+    from app.services.scrape_pipeline import CitySweepCounters
+    from dataclasses import fields
+
+    c = CitySweepCounters()
+    d = c.to_dict()
+    expected_keys = {f.name for f in fields(c)}
+    assert set(d.keys()) == expected_keys
+
+
+# ── scrape_runs utilities (unit, no DB) ────────────────────────────────────
+
+
+def test_scrape_runs_create_run_returns_id() -> None:
+    from app.services.scrape_runs import create_run
+
+    mock_db = MagicMock()
+    mock_row = MagicMock()
+    mock_row.id = 42
+    mock_db.execute.return_value.fetchone.return_value = mock_row
+
+    run_id = create_run(mock_db, source="avito_city_sweep", params={"pages_per_anchor": 3})
+
+    assert run_id == 42
+    mock_db.commit.assert_called()
+
+
+def test_scrape_runs_is_cancelled_true() -> None:
+    from app.services.scrape_runs import is_cancelled
+
+    mock_db = MagicMock()
+    mock_row = MagicMock()
+    mock_row.status = "cancelled"
+    mock_db.execute.return_value.fetchone.return_value = mock_row
+
+    assert is_cancelled(mock_db, 99) is True
+
+
+def test_scrape_runs_is_cancelled_false_when_running() -> None:
+    from app.services.scrape_runs import is_cancelled
+
+    mock_db = MagicMock()
+    mock_row = MagicMock()
+    mock_row.status = "running"
+    mock_db.execute.return_value.fetchone.return_value = mock_row
+
+    assert is_cancelled(mock_db, 99) is False
+
+
+def test_scrape_runs_is_cancelled_false_when_not_found() -> None:
+    from app.services.scrape_runs import is_cancelled
+
+    mock_db = MagicMock()
+    mock_db.execute.return_value.fetchone.return_value = None
+
+    assert is_cancelled(mock_db, 999) is False
+
+
+def test_scrape_runs_mark_cancelled_returns_bool() -> None:
+    from app.services.scrape_runs import mark_cancelled
+
+    mock_db = MagicMock()
+    mock_db.execute.return_value.fetchone.return_value = MagicMock()  # row found
+
+    result = mark_cancelled(mock_db, 10)
+    assert result is True
+    mock_db.commit.assert_called()
+
+
+def test_scrape_runs_mark_cancelled_returns_false_when_not_running() -> None:
+    from app.services.scrape_runs import mark_cancelled
+
+    mock_db = MagicMock()
+    mock_db.execute.return_value.fetchone.return_value = None  # no row (not running)
+
+    result = mark_cancelled(mock_db, 10)
+    assert result is False
+
+
+# ── FastAPI endpoints (offline) ─────────────────────────────────────────────
+
+
+@pytest.fixture
+def app_with_admin():
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+
+    # Lazy import to avoid side effects before env is set
+    from app.api.v1 import admin as admin_module
+    from app.core.db import get_db
+
+    app = FastAPI()
+    app.include_router(admin_module.router, prefix="/api/v1/admin")
+
+    def fake_db():
+        yield MagicMock()
+
+    app.dependency_overrides[get_db] = fake_db
+    return TestClient(app)
+
+
+def test_start_endpoint_validates_pages_per_anchor(app_with_admin) -> None:
+    """pages_per_anchor=99 превышает max=10 → 422."""
+    r = app_with_admin.post(
+        "/api/v1/admin/scrape/avito-city-sweep",
+        json={"pages_per_anchor": 99},
+    )
+    assert r.status_code == 422
+
+
+def test_start_endpoint_validates_request_delay_too_low(app_with_admin) -> None:
+    """request_delay_sec=1.0 меньше min=3.0 → 422."""
+    r = app_with_admin.post(
+        "/api/v1/admin/scrape/avito-city-sweep",
+        json={"request_delay_sec": 1.0},
+    )
+    assert r.status_code == 422
+
+
+def test_start_endpoint_ok(app_with_admin) -> None:
+    """Valid request → 200, run_id в ответе."""
+    with (
+        patch("app.services.scrape_runs.create_run", return_value=7) as mock_create,
+        patch("app.services.scrape_pipeline.run_avito_city_sweep", new_callable=AsyncMock),
+    ):
+        r = app_with_admin.post(
+            "/api/v1/admin/scrape/avito-city-sweep",
+            json={"pages_per_anchor": 2, "detail_top_n": 10},
+        )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["run_id"] == 7
+    assert body["status"] == "running"
+    assert body["pages_per_anchor"] == 2
+
+
+def test_cancel_endpoint_exists(app_with_admin) -> None:
+    """Cancel endpoint returns run_id + cancelled flag."""
+    with patch("app.services.scrape_runs.mark_cancelled", return_value=True):
+        r = app_with_admin.post("/api/v1/admin/scrape/avito-city-sweep/123/cancel")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["run_id"] == 123
+    assert body["cancelled"] is True
+    assert body["ok"] is True
+
+
+def test_cancel_endpoint_not_running(app_with_admin) -> None:
+    """Cancel на уже завершённый run → cancelled=False."""
+    with patch("app.services.scrape_runs.mark_cancelled", return_value=False):
+        r = app_with_admin.post("/api/v1/admin/scrape/avito-city-sweep/456/cancel")
+    assert r.status_code == 200
+    assert r.json()["cancelled"] is False
+
+
+def test_list_runs_endpoint(app_with_admin) -> None:
+    """List runs endpoint возвращает список."""
+    from datetime import datetime, timezone
+
+    fake_rows = [
+        {
+            "run_id": 1,
+            "source": "avito_city_sweep",
+            "status": "done",
+            "params": {"pages_per_anchor": 3},
+            "counters": {"lots_fetched": 300},
+            "error": None,
+            "started_at": datetime(2025, 1, 1, tzinfo=timezone.utc),
+            "finished_at": datetime(2025, 1, 1, 1, tzinfo=timezone.utc),
+            "heartbeat_at": datetime(2025, 1, 1, 1, tzinfo=timezone.utc),
+        }
+    ]
+    with patch("app.services.scrape_runs.list_recent", return_value=fake_rows):
+        r = app_with_admin.get("/api/v1/admin/scrape/avito-city-sweep/runs")
+    assert r.status_code == 200
+    body = r.json()
+    assert len(body) == 1
+    assert body[0]["run_id"] == 1
+    assert body[0]["status"] == "done"
+    assert "2025-01-01" in body[0]["started_at"]

tradein-mvp/deploy/avito-city-sweep.sh

@@ -0,0 +1,64 @@
+#!/usr/bin/env bash
+# Avito city sweep — full ЕКБ pagination + enrichment.
+# Random delay 0-3h перед запуском чтобы cron не бил Avito в одно и то же время каждый день.
+#
+# Crontab: 0 2 * * * /opt/gendesign/tradein-mvp/deploy/avito-city-sweep.sh
+# Реальный запуск: 02:00-05:00 UTC случайно каждый день.
+#
+# Env overrides:
+#   SWEEP_PAGES=3           — страниц на anchor (default 3)
+#   SWEEP_DETAIL_TOP_N=20   — detail enrichment top-N (default 20)
+#   SWEEP_REQUEST_DELAY=7.0 — задержка между запросами в секундах (default 7.0)
+#   LOG=/var/log/tradein-avito-city-sweep.log
+#   TRADEIN_BACKEND_CONTAINER=tradein-backend
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+LOG="${LOG:-/var/log/tradein-avito-city-sweep.log}"
+CONTAINER="${TRADEIN_BACKEND_CONTAINER:-tradein-backend}"
+
+log() {
+    echo "[$(date -u +'%Y-%m-%dT%H:%M:%SZ')] $*" | tee -a "$LOG"
+}
+
+# Random delay 0-10800 sec (0-3h). RANDOM gives 0..32767, mod 10800 → uniform 0..3h.
+DELAY=$((RANDOM % 10800))
+log "=== avito-city-sweep: sleeping ${DELAY}s before start ==="
+sleep "$DELAY"
+
+# Параметры sweep (overrideable via env)
+PAGES_PER_ANCHOR="${SWEEP_PAGES:-3}"
+DETAIL_TOP_N="${SWEEP_DETAIL_TOP_N:-20}"
+REQUEST_DELAY="${SWEEP_REQUEST_DELAY:-7.0}"
+
+PAYLOAD=$(cat <<EOF
+{
+  "pages_per_anchor": ${PAGES_PER_ANCHOR},
+  "detail_top_n": ${DETAIL_TOP_N},
+  "request_delay_sec": ${REQUEST_DELAY},
+  "enrich_houses": true,
+  "radius_m": 1500
+}
+EOF
+)
+
+log "=== avito-city-sweep: triggering with pages=${PAGES_PER_ANCHOR} detail_top_n=${DETAIL_TOP_N} ==="
+
+resp=$(docker exec "$CONTAINER" curl -s -X POST \
+    http://localhost:8000/api/v1/admin/scrape/avito-city-sweep \
+    -H "Content-Type: application/json" \
+    -d "$PAYLOAD" \
+    -m 30 2>/dev/null || echo '{"error":"trigger_failed"}')
+
+log "<- response: $resp"
+
+# Проверка что run_id вернулся
+if echo "$resp" | grep -q '"run_id"'; then
+    RUN_ID=$(echo "$resp" | grep -o '"run_id":[0-9]*' | grep -o '[0-9]*')
+    log "=== avito-city-sweep: started run_id=${RUN_ID} (background task на backend)"
+    log "=== Polling: GET /api/v1/admin/scrape/avito-city-sweep/runs"
+else
+    log "=== avito-city-sweep: WARNING — unexpected response, check backend logs ==="
+    exit 1
+fi