From 6762a532eb07afe048abd3cff66b48c77e0e12f5 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 30 May 2026 15:43:38 +0000 Subject: [PATCH] =?UTF-8?q?feat(scrapers):=20wire=20N1.ru=20city=20sweep?= =?UTF-8?q?=20=E2=80=94=20pipeline=20+=20scheduler=20+=20admin=20endpoint?= =?UTF-8?q?=20(#575)=20(#745)=20Co-authored-by:=20bot-backend=20=20Co-committed-by:=20bot-backend=20?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/api/v1/admin.py | 206 +++++++++---- tradein-mvp/backend/app/services/scheduler.py | 89 ++++-- .../backend/app/services/scrape_pipeline.py | 272 ++++++++++++++---- .../084_scrape_schedules_seed_n1_sweep.sql | 49 ++++ .../backend/tests/test_n1_city_sweep.py | 228 +++++++++++++++ 5 files changed, 714 insertions(+), 130 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/084_scrape_schedules_seed_n1_sweep.sql create mode 100644 tradein-mvp/backend/tests/test_n1_city_sweep.py diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 2eba9cdf..825528dd 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/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}" diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 8a3cfd2e..32956e49 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/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": diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 91b19878..81cddaf6 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/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", diff --git a/tradein-mvp/backend/data/sql/084_scrape_schedules_seed_n1_sweep.sql b/tradein-mvp/backend/data/sql/084_scrape_schedules_seed_n1_sweep.sql new file mode 100644 index 00000000..4e05d73d --- /dev/null +++ b/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; diff --git a/tradein-mvp/backend/tests/test_n1_city_sweep.py b/tradein-mvp/backend/tests/test_n1_city_sweep.py new file mode 100644 index 00000000..e24b4f4d --- /dev/null +++ b/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