feat(scrapers): wire N1.ru city sweep — pipeline + scheduler + admin endpoint (#575)

n1.py scraper existed and was wired into the generic point-/scrape endpoint, but
had no city-sweep + schedule entry. Mirror the yandex sweep (N1 is a simple
regional SERP — search → save, no houses/detail enrich):

- scrape_pipeline.run_n1_city_sweep + N1CitySweepCounters (anchors × pages →
  save_listings; cooperative cancel; abort after N consecutive anchor failures).
- scheduler.trigger_n1_city_sweep_run + dispatch (source='n1_city_sweep').
- data/sql/084: scrape_schedules seed, enabled=true (vs dormant avito/yandex) —
  #575 activates N1 as a source; one-line UPDATE to disable if it blocks.
- POST /api/v1/admin/scrape/n1 (+ /scrape/n1/runs) for on-demand city-sweep.

estimator.sources_used is derived from {lot.source} — 'n1' flows in automatically
once n1 listings exist (no estimator change).

Tests: test_n1_city_sweep.py (5) — aggregate, run_id threading, empty-page stop,
mid-sweep cancel, consecutive-failure abort. ruff clean; 32 scheduler/sweep
regression tests pass.

Refs #575

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

@@ -21,7 +21,7 @@ from app.schemas.trade_in import ScheduleConfig, ScheduleConfigUpdate
 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.scrape_pipeline import run_avito_city_sweep, run_n1_city_sweep
 from app.services.scraper_settings import invalidate_cache
 from app.services.scrapers.avito import AvitoScraper
 from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
@@ -106,7 +106,9 @@ async def scrape_around(
         async with scraper_cls() as scraper:
             if source == "yandex" and payload.deep_yandex:
                 lots = await scraper.fetch_around_multi_room(
-                    payload.lat, payload.lon, payload.radius_m,
+                    payload.lat,
+                    payload.lon,
+                    payload.radius_m,
                     sorts=("DATE_DESC", "PRICE", "AREA_DESC"),
                     pages=(0, 1),
                 )
@@ -123,9 +125,7 @@ async def scrape_around(
                     payload.lat, payload.lon, payload.radius_m
                 )
             else:
-                lots = await scraper.fetch_around(
-                    payload.lat, payload.lon, payload.radius_m
-                )
+                lots = await scraper.fetch_around(payload.lat, payload.lon, payload.radius_m)
         inserted, updated = save_listings(db, lots)
         results.append(
             ScrapeResult(source=source, fetched=len(lots), inserted=inserted, updated=updated)
@@ -172,9 +172,10 @@ async def geocode_missing(
         if target == "listings"
         else ""
     )
-    rows = db.execute(
-        text(
-            f"""
+    rows = (
+        db.execute(
+            text(
+                f"""
             SELECT id, address
             FROM {target}
             WHERE lat IS NULL
@@ -185,9 +186,12 @@ async def geocode_missing(
             ORDER BY geocode_tried_at NULLS FIRST
             LIMIT :limit
             """
-        ),
-        {"limit": limit},
-    ).mappings().all()
+            ),
+            {"limit": limit},
+        )
+        .mappings()
+        .all()
+    )
 
     # Бюджет по времени: геокодинг упирается в Nominatim (1 req/sec + typo-тиры
     # на провалах), 100 адресов могут не уложиться в cron-таймаут `curl -m 320`.
@@ -201,7 +205,8 @@ async def geocode_missing(
         if time.monotonic() - started > budget_sec:
             logger.info(
                 "geocode-missing: бюджет %.0fс исчерпан, обработано %d — cron продолжит",
-                budget_sec, geocoded + skipped,
+                budget_sec,
+                geocoded + skipped,
             )
             break
         clean = _clean_address_for_geocode(row["address"])
@@ -328,9 +333,10 @@ async def get_geocode_backfill_status(
 
     Включает per-source breakdown и оценку времени при Nominatim 1 req/sec.
     """
-    stats = db.execute(
-        text(
-            """
+    stats = (
+        db.execute(
+            text(
+                """
             SELECT
               source,
               COUNT(*) AS total,
@@ -341,22 +347,29 @@ async def get_geocode_backfill_status(
             GROUP BY source
             ORDER BY pending_geocode DESC
             """
+            )
         )
-    ).mappings().all()
+        .mappings()
+        .all()
+    )
 
     total_pending = sum(s["pending_geocode"] for s in stats)
-    unique_addresses_pending = db.execute(
-        text(
-            """
+    unique_addresses_pending = (
+        db.execute(
+            text(
+                """
             SELECT COUNT(DISTINCT address) FROM listings
             WHERE lat IS NULL AND address IS NOT NULL AND length(trim(address)) >= 5
             """
-        )
-    ).scalar() or 0
+            )
+        ).scalar()
+        or 0
+    )
 
-    cache_size = db.execute(
-        text("SELECT COUNT(*) FROM geocode_cache WHERE expires_at > NOW()")
-    ).scalar() or 0
+    cache_size = (
+        db.execute(text("SELECT COUNT(*) FROM geocode_cache WHERE expires_at > NOW()")).scalar()
+        or 0
+    )
 
     return {
         "total_pending_listings": total_pending,
@@ -389,6 +402,7 @@ async def trigger_geocode_missing_listings(
     from app.core.db import SessionLocal
 
     if background:
+
         async def _bg_task() -> None:
             bg_db = SessionLocal()
             try:
@@ -657,6 +671,72 @@ def cancel_avito_city_sweep(
     return {"ok": True, "run_id": run_id, "cancelled": cancelled}
 
 
+# ── N1.ru city sweep (#575) — on-demand trigger + runs list ──────────────────
+
+
+@router.post("/scrape/n1", response_model=CitySweepStartResponse)
+async def start_n1_city_sweep(
+    payload: CitySweepStartRequest,
+    background_tasks: BackgroundTasks,
+    db: Annotated[Session, Depends(get_db)],
+) -> CitySweepStartResponse:
+    """Запустить N1.ru ЕКБ city sweep в background (#575). Returns run_id.
+
+    Зеркало avito-city-sweep, но N1 — простой SERP (search → save), без
+    houses/detail-enrich: поля enrich_houses/detail_top_n из payload игнорируются.
+    Помимо этого эндпоинта N1 крутится по расписанию (scrape_schedules
+    source='n1_city_sweep', seed 084).
+    """
+    run_id = runs_mod.create_run(db, source="n1_city_sweep", params=payload.model_dump())
+
+    async def _sweep_task() -> None:
+        sweep_db = SessionLocal()
+        try:
+            await run_n1_city_sweep(
+                sweep_db,
+                run_id=run_id,
+                radius_m=payload.radius_m,
+                pages_per_anchor=payload.pages_per_anchor,
+                request_delay_sec=payload.request_delay_sec,
+            )
+        except Exception:
+            logger.exception("n1-sweep background task run_id=%d crashed", run_id)
+        finally:
+            sweep_db.close()
+
+    background_tasks.add_task(_sweep_task)
+    logger.info("n1-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=0,  # N1 sweep не имеет detail-шага
+    )
+
+
+@router.get("/scrape/n1/runs", response_model=list[ScrapeRunRow])
+def list_n1_city_sweep_runs(
+    db: Annotated[Session, Depends(get_db)],
+    limit: int = 10,
+) -> list[ScrapeRunRow]:
+    """Список последних N N1-sweep runs (для UI polling). Default limit=10."""
+    rows = runs_mod.list_recent(db, source="n1_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
+    ]
+
+
 # ── Scraper settings: global + per-source delay management ───────────────────
 
 
@@ -671,13 +751,17 @@ def list_scraper_settings(
     db: Annotated[Session, Depends(get_db)],
 ) -> dict:
     """Список всех настроек задержки (per-source + global)."""
-    rows = db.execute(
-        text("""
+    rows = (
+        db.execute(
+            text("""
             SELECT source, request_delay_sec, description, updated_at
             FROM scraper_settings
             ORDER BY source ASC
         """)
-    ).mappings().all()
+        )
+        .mappings()
+        .all()
+    )
     return {"settings": [dict(r) for r in rows]}
 
 
@@ -700,8 +784,9 @@ def update_scraper_setting(
             detail=f"Path source={source!r} does not match body source={payload.source!r}",
         )
 
-    row = db.execute(
-        text("""
+    row = (
+        db.execute(
+            text("""
             INSERT INTO scraper_settings (source, request_delay_sec, description, updated_at)
             VALUES (:s, CAST(:d AS numeric), :desc, NOW())
             ON CONFLICT (source) DO UPDATE SET
@@ -710,18 +795,19 @@ def update_scraper_setting(
                 updated_at = NOW()
             RETURNING source, request_delay_sec, description, updated_at
         """),
-        {
-            "s": source,
-            "d": payload.request_delay_sec,
-            "desc": payload.description,
-        },
-    ).mappings().one()
+            {
+                "s": source,
+                "d": payload.request_delay_sec,
+                "desc": payload.description,
+            },
+        )
+        .mappings()
+        .one()
+    )
     db.commit()
 
     invalidate_cache(source)
-    logger.info(
-        "scraper-settings: updated source=%s delay=%.1f", source, payload.request_delay_sec
-    )
+    logger.info("scraper-settings: updated source=%s delay=%.1f", source, payload.request_delay_sec)
     return dict(row)
 
 
@@ -731,17 +817,21 @@ def update_scraper_setting(
 @router.get("/scrape/schedules", response_model=list[ScheduleConfig])
 def list_schedules(db: Annotated[Session, Depends(get_db)]) -> list[ScheduleConfig]:
     """Список всех schedules. Сейчас single row 'avito_city_sweep', extensible."""
-    rows = db.execute(
-        text(
-            """
+    rows = (
+        db.execute(
+            text(
+                """
             SELECT id, source, enabled, window_start_hour, window_end_hour,
                    default_params, last_run_id, last_run_at, next_run_at,
                    created_at, updated_at
             FROM scrape_schedules
             ORDER BY source
             """
-        ),
-    ).mappings().all()
+            ),
+        )
+        .mappings()
+        .all()
+    )
     return [
         ScheduleConfig(
             id=r["id"],
@@ -771,9 +861,10 @@ def update_schedule(
     # Compute new next_run_at если window изменился — recompute, иначе keep existing
     next_at = compute_next_run_at(payload.window_start_hour, payload.window_end_hour)
 
-    row = db.execute(
-        text(
-            """
+    row = (
+        db.execute(
+            text(
+                """
             INSERT INTO scrape_schedules
                 (source, enabled, window_start_hour, window_end_hour, default_params, next_run_at)
             VALUES (:source, :enabled, :ws, :we, CAST(:params AS jsonb), :next_at)
@@ -787,16 +878,19 @@ def update_schedule(
             RETURNING id, source, enabled, window_start_hour, window_end_hour,
                       default_params, last_run_id, last_run_at, next_run_at, updated_at
             """
-        ),
-        {
-            "source": source,
-            "enabled": payload.enabled,
-            "ws": payload.window_start_hour,
-            "we": payload.window_end_hour,
-            "params": json.dumps(payload.default_params, ensure_ascii=False),
-            "next_at": next_at,
-        },
-    ).mappings().fetchone()
+            ),
+            {
+                "source": source,
+                "enabled": payload.enabled,
+                "ws": payload.window_start_hour,
+                "we": payload.window_end_hour,
+                "params": json.dumps(payload.default_params, ensure_ascii=False),
+                "next_at": next_at,
+            },
+        )
+        .mappings()
+        .fetchone()
+    )
     db.commit()
 
     assert row is not None, f"scrape_schedules upsert returned no row for source={source!r}"

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

@@ -10,6 +10,7 @@
 Sources:
   - avito_city_sweep        → run_avito_city_sweep (scrape_pipeline.py)
   - yandex_city_sweep       → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT)
+  - n1_city_sweep           → run_n1_city_sweep (scrape_pipeline.py, #575)
   - cian_history_backfill   → run_cian_backfill (this module, #560)
   - rosreestr_dkp_import    → import_rosreestr_dkp (this module, #563)
   - listing_source_snapshot → snapshot_listing_sources (tasks/listing_source_snapshot.py, #570;
@@ -19,6 +20,7 @@ Sources:
                               re-derivation of the asking→sold ratios — no external calls,
                               enabled by default; window after rosreestr_dkp_import)
 """
+
 from __future__ import annotations
 
 import asyncio
@@ -33,7 +35,11 @@ from sqlalchemy.orm import Session
 
 from app.core.db import SessionLocal
 from app.services import scrape_runs as runs_mod
-from app.services.scrape_pipeline import run_avito_city_sweep, run_yandex_city_sweep
+from app.services.scrape_pipeline import (
+    run_avito_city_sweep,
+    run_n1_city_sweep,
+    run_yandex_city_sweep,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -58,9 +64,7 @@ def compute_next_run_at(
         start_seconds = window_start_hour * 3600
         end_seconds = window_end_hour * 3600
         rand_seconds = random.randint(start_seconds, end_seconds - 1)
-        return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(
-            seconds=rand_seconds
-        )
+        return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(seconds=rand_seconds)
     else:
         # Cross-midnight (22..3 → 22:00-23:59 + 00:00-02:59)
         # Длина окна = (24-start) + end часов
@@ -231,6 +235,39 @@ async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any
     return run_id
 
 
+async def trigger_n1_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать новый scrape_runs + launch run_n1_city_sweep в asyncio.create_task (#575).
+
+    Зеркало trigger_yandex_city_sweep_run — N1 sweep простой (search → save, без
+    houses/detail). Returns run_id (или None если skip — есть running run).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            await run_n1_city_sweep(
+                run_db,
+                run_id=run_id,
+                pages_per_anchor=int(params.get("pages_per_anchor", 2)),
+                request_delay_sec=float(params.get("request_delay_sec", 7.0)),
+                radius_m=int(params.get("radius_m", 1500)),
+            )
+        except Exception:
+            logger.exception("scheduler: run_n1_city_sweep crashed run_id=%d", run_id)
+        finally:
+            run_db.close()
+
+    task = asyncio.create_task(_run())
+    task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+    logger.info("scheduler: triggered n1_city_sweep run_id=%d", run_id)
+    return run_id
+
+
 async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch run_cian_backfill в asyncio.create_task.
 
@@ -409,9 +446,7 @@ async def trigger_listing_source_snapshot_run(
     return run_id
 
 
-async def trigger_asking_to_sold_ratio_run(
-    db: Session, schedule_row: dict[str, Any]
-) -> int | None:
+async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).
 
     Daily TRUE-MIRROR refresh асking→sold коэффициентов (#648 Stage 4): DELETE WHERE
@@ -436,9 +471,7 @@ async def trigger_asking_to_sold_ratio_run(
             loop = asyncio.get_event_loop()
             await loop.run_in_executor(None, recompute_asking_to_sold_ratios, run_db, run_id)
         except Exception:
-            logger.exception(
-                "scheduler: recompute_asking_to_sold_ratios crashed run_id=%d", run_id
-            )
+            logger.exception("scheduler: recompute_asking_to_sold_ratios crashed run_id=%d", run_id)
         finally:
             run_db.close()
 
@@ -498,9 +531,7 @@ def import_rosreestr_dkp(
     # Cleanup legacy synthetic rows (pre-#549, idempotent)
     try:
         deleted = db.execute(
-            text(
-                "DELETE FROM deals WHERE address = CAST(:addr AS text) RETURNING id"
-            ),
+            text("DELETE FROM deals WHERE address = CAST(:addr AS text) RETURNING id"),
             {"addr": "Екатеринбург, реальная сделка"},
         ).fetchall()
         db.commit()
@@ -530,8 +561,9 @@ def import_rosreestr_dkp(
             # FDW pushes WHERE + ORDER BY + LIMIT to gendesign-postgres automatically
             # (postgres_fdw 'use_remote_estimate' is off by default but clause push-down
             # still happens for simple predicates on the remote side).
-            batch_rows = db.execute(
-                text("""
+            batch_rows = (
+                db.execute(
+                    text("""
                     SELECT
                         id,
                         id                                       AS source_id_src,
@@ -568,8 +600,11 @@ def import_rosreestr_dkp(
                     ORDER BY id
                     LIMIT CAST(:batch_size AS int)
                 """),
-                {"since": since, "last_id": last_id, "batch_size": batch_size},
-            ).mappings().all()
+                    {"since": since, "last_id": last_id, "batch_size": batch_size},
+                )
+                .mappings()
+                .all()
+            )
 
             if not batch_rows:
                 break
@@ -671,26 +706,28 @@ def import_rosreestr_dkp(
         )
 
     except Exception as exc:
-        logger.exception(
-            "rosreestr_dkp_import run_id=%d failed at last_id=%d", run_id, last_id
-        )
+        logger.exception("rosreestr_dkp_import run_id=%d failed at last_id=%d", run_id, last_id)
         runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
         raise
 
 
 def get_due_schedules(db: Session) -> list[dict[str, Any]]:
     """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
-    rows = db.execute(
-        text(
-            """
+    rows = (
+        db.execute(
+            text(
+                """
             SELECT id, source, enabled, window_start_hour, window_end_hour,
                    default_params, last_run_id, last_run_at, next_run_at
             FROM scrape_schedules
             WHERE enabled = true
               AND (next_run_at IS NULL OR next_run_at <= NOW())
             """
-        ),
-    ).mappings().all()
+            ),
+        )
+        .mappings()
+        .all()
+    )
     return [dict(r) for r in rows]
 
 
@@ -713,6 +750,8 @@ async def scheduler_loop() -> None:
                         await trigger_avito_city_sweep_run(db, sch)
                     elif source == "yandex_city_sweep":
                         await trigger_yandex_city_sweep_run(db, sch)
+                    elif source == "n1_city_sweep":
+                        await trigger_n1_city_sweep_run(db, sch)
                     elif source == "cian_history_backfill":
                         await trigger_cian_backfill_run(db, sch)
                     elif source == "rosreestr_dkp_import":

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

@@ -40,6 +40,7 @@ from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichm
 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
+from app.services.scrapers.n1 import N1Scraper
 from app.services.scrapers.yandex_realty import YandexRealtyScraper
 
 logger = logging.getLogger(__name__)
@@ -140,26 +141,27 @@ async def run_avito_pipeline(
         scraper._cffi = session
         try:
             lots = await scraper.fetch_around(
-                lat, lon, radius_m,
+                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
-            )
+            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 ───────────────────────────────
         if lots:
@@ -192,14 +194,10 @@ async def run_avito_pipeline(
                     save_house_catalog_enrichment(db, enrichment)
                     counters.houses_enriched += 1
                 except (AvitoBlockedError, AvitoRateLimitedError):
-                    logger.error(
-                        "pipeline:house BLOCKED at %s — propagating", house_path
-                    )
+                    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
-                    )
+                    logger.warning("pipeline:house_enrich failed for %s: %s", house_path, e)
                     counters.houses_failed += 1
                     counters.errors.append(f"house {house_path}: {e}")
 
@@ -208,8 +206,9 @@ async def run_avito_pipeline(
 
         # ── Step 5: enrich detail для top-N priority listings ───
         if enrich_detail_top_n > 0:
-            priority_rows = db.execute(
-                text("""
+            priority_rows = (
+                db.execute(
+                    text("""
                     SELECT source_url
                     FROM listings
                     WHERE source = 'avito'
@@ -232,13 +231,16 @@ async def run_avito_pipeline(
                     ORDER BY scraped_at DESC NULLS LAST
                     LIMIT :limit
                 """),
-                {
-                    "lat": lat,
-                    "lon": lon,
-                    "radius": radius_m * 2,
-                    "limit": enrich_detail_top_n,
-                },
-            ).mappings().all()
+                    {
+                        "lat": lat,
+                        "lon": lon,
+                        "radius": radius_m * 2,
+                        "limit": enrich_detail_top_n,
+                    },
+                )
+                .mappings()
+                .all()
+            )
 
             consecutive_blocks = 0
             for idx, row in enumerate(priority_rows):
@@ -246,9 +248,7 @@ async def run_avito_pipeline(
                 counters.detail_attempted += 1
                 try:
                     item_url = (
-                        urlparse(source_url).path
-                        if source_url.startswith("http")
-                        else source_url
+                        urlparse(source_url).path if source_url.startswith("http") else source_url
                     )
                     enrichment_detail = await fetch_detail(item_url, cffi_session=session)
                     if save_detail_enrichment(db, enrichment_detail):
@@ -260,7 +260,10 @@ async def run_avito_pipeline(
                     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,
+                        idx + 1,
+                        len(priority_rows),
+                        consecutive_blocks,
+                        e,
                     )
                     if consecutive_blocks >= 3:
                         logger.error(
@@ -275,9 +278,7 @@ async def run_avito_pipeline(
                 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
-                    )
+                    logger.warning("pipeline:detail_enrich failed for %s: %s", source_url, e)
                     consecutive_blocks = 0
 
                 if idx < len(priority_rows) - 1:
@@ -287,10 +288,15 @@ async def run_avito_pipeline(
         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)
@@ -351,14 +357,22 @@ async def run_avito_city_sweep(
                 if scrape_runs.is_cancelled(db, run_id):
                     logger.info(
                         "city-sweep run_id=%d: cancelled at anchor #%d/%d (%s)",
-                        run_id, idx, len(_anchors), name,
+                        run_id,
+                        idx,
+                        len(_anchors),
+                        name,
                     )
                     scrape_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,
+                    run_id,
+                    idx,
+                    len(_anchors),
+                    name,
+                    lat,
+                    lon,
                 )
                 try:
                     result = await run_avito_pipeline(
@@ -385,7 +399,11 @@ async def run_avito_city_sweep(
                 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,
+                        run_id,
+                        idx,
+                        len(_anchors),
+                        name,
+                        e,
                     )
                     counters.errors_count += 1
                     counters.anchors_done = idx
@@ -393,9 +411,7 @@ async def run_avito_city_sweep(
                     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
@@ -493,14 +509,22 @@ async def run_yandex_city_sweep(
             if scrape_runs.is_cancelled(db, run_id):
                 logger.info(
                     "yandex-sweep run_id=%d: cancelled at anchor #%d/%d (%s)",
-                    run_id, idx, len(_anchors), name,
+                    run_id,
+                    idx,
+                    len(_anchors),
+                    name,
                 )
                 scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                 return counters
 
             logger.info(
                 "yandex-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
-                run_id, idx, len(_anchors), name, lat, lon,
+                run_id,
+                idx,
+                len(_anchors),
+                name,
+                lat,
+                lon,
             )
             try:
                 # Fresh scraper per anchor — uses its own httpx.AsyncClient via the
@@ -508,9 +532,7 @@ async def run_yandex_city_sweep(
                 anchor_lots: list[ScrapedLot] = []
                 async with YandexRealtyScraper() as scraper:
                     for page in range(pages_per_anchor):
-                        page_lots = await scraper.fetch_around(
-                            lat, lon, radius_m, page=page
-                        )
+                        page_lots = await scraper.fetch_around(lat, lon, radius_m, page=page)
                         counters.pages_fetched += 1
                         if not page_lots:
                             # Пустая страница → дальше результатов нет, прекращаем пагинацию.
@@ -526,16 +548,19 @@ async def run_yandex_city_sweep(
             except Exception as e:
                 # YandexRealtyScraper не propagate'ит block/rate exceptions, поэтому здесь
                 # ловим broad Exception per-anchor: логируем и continue (изоляция anchor'а).
-                logger.exception(
-                    "yandex-sweep run_id=%d: anchor %s failed", run_id, name
-                )
+                logger.exception("yandex-sweep run_id=%d: anchor %s failed", run_id, name)
                 counters.errors_count += 1
                 consecutive_failures += 1
                 if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES:
                     logger.error(
                         "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
                         "%d consecutive failures (IP likely blocked): %s",
-                        run_id, idx, len(_anchors), name, consecutive_failures, e,
+                        run_id,
+                        idx,
+                        len(_anchors),
+                        name,
+                        consecutive_failures,
+                        e,
                     )
                     counters.anchors_done = idx
                     scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
@@ -576,6 +601,155 @@ async def run_yandex_city_sweep(
         raise
 
 
+# ── N1.ru city sweep (#575) ─────────────────────────────────────────
+
+
+@dataclass
+class N1CitySweepCounters:
+    """Aggregate counters для N1.ru city sweep run.
+
+    Как Yandex: N1 SERP — простой search → save, без house-catalog / detail-enrich
+    шага, поэтому houses/detail-полей нет.
+    """
+
+    anchors_total: int = 0
+    anchors_done: int = 0
+    pages_fetched: int = 0
+    lots_fetched: int = 0
+    lots_inserted: int = 0
+    lots_updated: 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)}
+
+
+# Defensive abort: N1Scraper.fetch_around глотает ошибки и возвращает [] (нет
+# block/rate exception-классов). Прерываем sweep после N подряд anchor'ов,
+# упавших с исключением — сигнал системного блока IP. Пустой результат (0 lots)
+# — НЕ ошибка, счётчик не растёт (зеркало yandex).
+N1_SWEEP_MAX_CONSECUTIVE_FAILURES = 3
+
+
+async def run_n1_city_sweep(
+    db: Session,
+    *,
+    run_id: int,
+    anchors: list[tuple[float, float, str]] | None = None,
+    radius_m: int = 1500,
+    pages_per_anchor: int = 2,
+    request_delay_sec: float | None = None,
+) -> N1CitySweepCounters:
+    """N1.ru city sweep: iterate anchors × pages → save (#575).
+
+    Зеркало run_yandex_city_sweep — N1 это региональный SERP-портал ЕКБ без
+    house-catalog / detail-enrich шага:
+      per anchor: N1Scraper().fetch_around(lat, lon, radius_m, page=p)
+      → save_listings(db, lots) → accumulate counters.
+
+    Anti-bot / cancel семантика (зеркало yandex/avito sweep):
+      - Cooperative cancel: is_cancelled(db, run_id) перед каждым anchor.
+      - N1Scraper не определяет block/rate exception-классов (fetch_around ловит
+        ошибки внутри → []). Per-anchor ловим broad Exception, логируем, continue;
+        защитно прерываем (early abort + mark_banned) при N подряд исключениях.
+      - request_delay_sec применяется как доп. asyncio.sleep МЕЖДУ anchor'ами
+        (N1Scraper сам спит self.request_delay_sec между страницами).
+    """
+    _anchors = anchors if anchors is not None else EKB_ANCHORS
+    counters = N1CitySweepCounters(anchors_total=len(_anchors))
+    inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0
+    consecutive_failures = 0
+
+    try:
+        for idx, (lat, lon, name) in enumerate(_anchors, start=1):
+            if scrape_runs.is_cancelled(db, run_id):
+                logger.info(
+                    "n1-sweep run_id=%d: cancelled at anchor #%d/%d (%s)",
+                    run_id,
+                    idx,
+                    len(_anchors),
+                    name,
+                )
+                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+                return counters
+
+            logger.info(
+                "n1-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
+                run_id,
+                idx,
+                len(_anchors),
+                name,
+                lat,
+                lon,
+            )
+            try:
+                anchor_lots: list[ScrapedLot] = []
+                async with N1Scraper() as scraper:
+                    for page in range(1, pages_per_anchor + 1):
+                        page_lots = await scraper.fetch_around(lat, lon, radius_m, page=page)
+                        counters.pages_fetched += 1
+                        if not page_lots:
+                            break
+                        anchor_lots.extend(page_lots)
+
+                counters.lots_fetched += len(anchor_lots)
+                if anchor_lots:
+                    inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
+                    counters.lots_inserted += inserted
+                    counters.lots_updated += updated
+                consecutive_failures = 0
+            except Exception as e:
+                logger.exception("n1-sweep run_id=%d: anchor %s failed", run_id, name)
+                counters.errors_count += 1
+                consecutive_failures += 1
+                if consecutive_failures >= N1_SWEEP_MAX_CONSECUTIVE_FAILURES:
+                    logger.error(
+                        "n1-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
+                        "%d consecutive failures (IP likely blocked): %s",
+                        run_id,
+                        idx,
+                        len(_anchors),
+                        name,
+                        consecutive_failures,
+                        e,
+                    )
+                    counters.anchors_done = idx
+                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+                    scrape_runs.mark_banned(
+                        db,
+                        run_id,
+                        f"n1 sweep aborted: {consecutive_failures} consecutive "
+                        f"anchor failures (last: {e})",
+                        counters.to_dict(),
+                    )
+                    return counters
+
+            counters.anchors_done = idx
+            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+
+            if idx < len(_anchors):
+                await asyncio.sleep(inter_anchor_delay)
+
+        scrape_runs.mark_done(db, run_id, counters.to_dict())
+        logger.info(
+            "n1-sweep run_id=%d done: anchors=%d/%d pages=%d lots=%d (ins=%d/upd=%d) errors=%d",
+            run_id,
+            counters.anchors_done,
+            counters.anchors_total,
+            counters.pages_fetched,
+            counters.lots_fetched,
+            counters.lots_inserted,
+            counters.lots_updated,
+            counters.errors_count,
+        )
+        return counters
+
+    except Exception as exc:
+        logger.exception("n1-sweep run_id=%d: fatal error", run_id)
+        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
+        raise
+
+
 # ── Public re-exports ───────────────────────────────────────────
 __all__ = [
     "EKB_ANCHORS",

tradein-mvp/backend/data/sql/084_scrape_schedules_seed_n1_sweep.sql

@@ -0,0 +1,49 @@
+-- 084_scrape_schedules_seed_n1_sweep.sql
+-- Seed row для N1.ru периодического city-sweep (#575).
+--
+-- N1 — региональный портал недвижимости ЕКБ. В отличие от avito/yandex sweep
+-- (засеяны DORMANT из-за анти-бот риска с prod-IP), N1 засеян ВКЛЮЧЁННЫМ
+-- (enabled = true): задача #575 — активировать N1 как источник listings
+-- (acceptance: ≥100 строк source='n1' за сутки). Anti-bot у регионального портала
+-- мягче, чем у Avito/Yandex; параметры консервативные. Capability полностью wired
+-- (run_n1_city_sweep + trigger_n1_city_sweep_run + dispatch в scheduler).
+--
+-- ЕСЛИ N1 начнёт блокировать prod-IP — выключить одной строкой:
+--     UPDATE scrape_schedules SET enabled = false WHERE source = 'n1_city_sweep';
+-- Ручной on-demand триггер для проверки до ночного окна — POST /api/v1/admin/scrape/n1.
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)).
+-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно.
+--
+-- n1_city_sweep — окно 02:00-05:00 UTC (05:00-08:00 МСК), как avito/yandex sweep.
+--   default_params: pages_per_anchor (страниц SERP на anchor), request_delay_sec
+--   (пауза между anchor'ами), radius_m (прокидывается в fetch_around).
+--   У N1 sweep НЕТ detail/houses-enrich шага (как у yandex).
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'n1_city_sweep',
+    true,                                        -- #575: активируем N1 как источник
+    2,
+    5,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC',
+    '{"pages_per_anchor": 2, "request_delay_sec": 7, "radius_m": 1500}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (заменяет cron-script setup). '
+    'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), '
+    'n1_city_sweep (#575), cian_history_backfill, rosreestr_dkp_import.';
+
+COMMIT;

tradein-mvp/backend/tests/test_n1_city_sweep.py

@@ -0,0 +1,228 @@
+"""Offline unit tests for run_n1_city_sweep (#575).
+
+Pure & fast: NO live network, NO DB. Mirrors test_yandex_city_sweep — monkeypatch
+N1Scraper lifecycle + fetch_around, scrape_pipeline.save_listings, scrape_runs.*.
+Asserts orchestration semantics only (live N1 behavior verified by post-deploy QA).
+
+NB: run_n1_city_sweep пагинирует page=1..pages_per_anchor (1-based), в отличие от
+yandex (0-based) — тесты учитывают это.
+"""
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from typing import Any
+
+import pytest
+
+from app.services import scrape_pipeline
+from app.services.scrapers.base import ScrapedLot
+from app.services.scrapers.n1 import N1Scraper
+
+TEST_ANCHORS: list[tuple[float, float, str]] = [
+    (56.8400, 60.6050, "A1"),
+    (56.7950, 60.5300, "A2"),
+    (56.8970, 60.6100, "A3"),
+]
+
+
+def _fake_lot(offer_id: str) -> ScrapedLot:
+    return ScrapedLot(
+        source="n1",
+        source_url=f"https://ekb.n1.ru/view/{offer_id}/",
+        source_id=offer_id,
+        address="Екатеринбург, улица Тестовая, 1",
+        price_rub=5_000_000,
+        area_m2=50.0,
+        rooms=2,
+    )
+
+
+class _FakeRuns:
+    def __init__(self, *, cancel_at_anchor: int | None = None) -> None:
+        self.cancel_at_anchor = cancel_at_anchor
+        self.is_cancelled_calls = 0
+        self.heartbeats: list[dict[str, int]] = []
+        self.done: dict[str, int] | None = None
+        self.failed: tuple[str, dict[str, int]] | None = None
+        self.banned: tuple[str, dict[str, int]] | None = None
+
+    def is_cancelled(self, _db: Any, _run_id: int) -> bool:
+        self.is_cancelled_calls += 1
+        if self.cancel_at_anchor is None:
+            return False
+        return self.is_cancelled_calls >= self.cancel_at_anchor
+
+    def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
+        self.heartbeats.append(dict(counters))
+
+    def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
+        self.done = dict(counters)
+
+    def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
+        self.failed = (error, dict(counters))
+
+    def mark_banned(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
+        self.banned = (error, dict(counters))
+
+
+@pytest.fixture(autouse=True)
+def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def _instant(_secs: float) -> None:
+        return None
+
+    monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", _instant)
+
+
+@pytest.fixture(autouse=True)
+def _stub_scraper_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Neutralize __init__/__aenter__/__aexit__ — no get_scraper_delay DB, no curl_cffi."""
+
+    def _init(self: N1Scraper) -> None:
+        self.request_delay_sec = 0.0
+        self._cffi = None
+
+    async def _aenter(self: N1Scraper) -> N1Scraper:
+        return self
+
+    async def _aexit(self: N1Scraper, *_args: Any) -> None:
+        return None
+
+    monkeypatch.setattr(N1Scraper, "__init__", _init)
+    monkeypatch.setattr(N1Scraper, "__aenter__", _aenter)
+    monkeypatch.setattr(N1Scraper, "__aexit__", _aexit)
+
+
+def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None:
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "is_cancelled", fake.is_cancelled)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "update_heartbeat", fake.update_heartbeat)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_done", fake.mark_done)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_failed", fake.mark_failed)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_banned", fake.mark_banned)
+
+
+async def test_sweep_iterates_all_anchors_and_aggregates(monkeypatch: pytest.MonkeyPatch) -> None:
+    fetch_calls: list[tuple[float, float, int]] = []
+    save_calls: list[int] = []
+
+    async def fake_fetch(self, lat, lon, radius_m=1000, rooms=None, page=1):
+        fetch_calls.append((lat, lon, page))
+        return [_fake_lot(f"{lat}-{page}-{i}") for i in range(2)]
+
+    def fake_save(_db, lots, *, run_id=None):
+        save_calls.append(len(lots))
+        return len(lots), 0
+
+    monkeypatch.setattr(N1Scraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_n1_city_sweep(
+        db=object(), run_id=1, anchors=TEST_ANCHORS, pages_per_anchor=2, request_delay_sec=0.0
+    )
+
+    assert len(fetch_calls) == len(TEST_ANCHORS) * 2
+    assert [p for _, _, p in fetch_calls[:2]] == [1, 2]  # 1-based pagination
+    assert len(save_calls) == len(TEST_ANCHORS)
+    assert counters.anchors_done == len(TEST_ANCHORS)
+    assert counters.pages_fetched == len(TEST_ANCHORS) * 2
+    assert counters.lots_fetched == 12
+    assert counters.lots_inserted == 12
+    assert counters.errors_count == 0
+    assert fake.done is not None and fake.done["lots_fetched"] == 12
+    assert fake.failed is None and fake.banned is None
+
+
+async def test_save_listings_receives_run_id(monkeypatch: pytest.MonkeyPatch) -> None:
+    seen: list[int | None] = []
+
+    async def fake_fetch(self, lat, lon, radius_m=1000, rooms=None, page=1):
+        return [_fake_lot(f"{lat}-{page}")]
+
+    def fake_save(_db, lots, *, run_id=None):
+        seen.append(run_id)
+        return len(lots), 0
+
+    monkeypatch.setattr(N1Scraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
+    _install_runs(monkeypatch, _FakeRuns())
+
+    await scrape_pipeline.run_n1_city_sweep(
+        db=object(), run_id=777, anchors=TEST_ANCHORS, pages_per_anchor=1
+    )
+    assert seen == [777, 777, 777]
+
+
+async def test_empty_page_stops_pagination(monkeypatch: pytest.MonkeyPatch) -> None:
+    pages_seen: list[int] = []
+
+    async def fake_fetch(self, lat, lon, radius_m=1000, rooms=None, page=1):
+        pages_seen.append(page)
+        if page == 1:
+            return [_fake_lot(f"{lat}-1")]
+        return []  # page 2 empty → stop
+
+    monkeypatch.setattr(N1Scraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_n1_city_sweep(
+        db=object(), run_id=2, anchors=TEST_ANCHORS, pages_per_anchor=5
+    )
+    assert pages_seen == [1, 2, 1, 2, 1, 2]
+    assert counters.pages_fetched == 6
+    assert counters.lots_fetched == 3
+    assert counters.anchors_done == 3
+
+
+async def test_cancel_midway_stops_remaining(monkeypatch: pytest.MonkeyPatch) -> None:
+    fetch_calls: list[str] = []
+
+    async def fake_fetch(self, lat, lon, radius_m=1000, rooms=None, page=1):
+        fetch_calls.append(f"{lat}")
+        return [_fake_lot(f"{lat}-{page}")]
+
+    monkeypatch.setattr(N1Scraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    fake = _FakeRuns(cancel_at_anchor=2)
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_n1_city_sweep(
+        db=object(), run_id=3, anchors=TEST_ANCHORS, pages_per_anchor=1
+    )
+    assert len(fetch_calls) == 1
+    assert counters.anchors_done == 1
+    assert fake.done is None and fake.banned is None
+
+
+async def test_consecutive_failures_abort_sweep(monkeypatch: pytest.MonkeyPatch) -> None:
+    attempts: list[str] = []
+
+    async def fake_fetch(self, lat, lon, radius_m=1000, rooms=None, page=1):
+        attempts.append(f"{lat}")
+        raise RuntimeError("n1 blocked")
+
+    monkeypatch.setattr(N1Scraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    anchors = [*TEST_ANCHORS, (56.77, 60.55, "A4"), (56.865, 60.62, "A5")]
+    counters = await scrape_pipeline.run_n1_city_sweep(
+        db=object(), run_id=5, anchors=anchors, pages_per_anchor=1
+    )
+    n = scrape_pipeline.N1_SWEEP_MAX_CONSECUTIVE_FAILURES
+    assert len(attempts) == n
+    assert counters.errors_count == n
+    assert counters.anchors_done == n
+    assert fake.banned is not None and "consecutive" in fake.banned[0]
+    assert fake.done is None