diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 6dbe6923..fa9599a4 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/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"]) - multi_room_yandex: bool = False # Если True — скрейп Yandex по 5 сегментам комнат, не одним общим запросом. - deep_yandex: bool = False # Если True — Yandex полный обход 5 rooms × 3 sorts × 2 pages = 30 запросов / ~150s. - multi_room_cian: bool = False # Если True — скрейп Cian по 4 сегментам комнат отдельно (~4× лотов). - multi_room_n1: bool = False # Если True — скрейп N1 по 4 сегментам комнат отдельно (~4× лотов). + sources: list[Literal["avito", "cian", "yandex", "n1"]] = Field( + default_factory=lambda: list(_ALL_SOURCES) + ) + # Если True — скрейп Yandex по 5 сегментам комнат, не одним общим запросом. + 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} diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 0353513d..b674ade0 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/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", ] diff --git a/tradein-mvp/backend/app/services/scrape_runs.py b/tradein-mvp/backend/app/services/scrape_runs.py new file mode 100644 index 00000000..ffbf9a51 --- /dev/null +++ b/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] diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py index 6ec92b8c..f6571b23 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito.py +++ b/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) - try: - assert self._cffi is not None - response = await self._cffi.get(url) - if response.status_code != 200: - logger.warning( - "avito HTML returned %d for %s", response.status_code, url - ) - return [] - except Exception: - logger.exception("avito HTML fetch failed for %s", url) - return [] + all_lots: list[ScrapedLot] = [] + for page in range(1, pages + 1): + url = self._build_web_url(lat, lon, radius_km, page=page) + try: + assert self._cffi is not None + response = await self._cffi.get(url) + if response.status_code != 200: + logger.warning( + "avito HTML page=%d returned %d for %s", page, response.status_code, url + ) + break + 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) - logger.info("avito html_scrape: %d lots around (%.4f, %.4f)", len(lots), lat, lon) - await self.sleep_between_requests() - return lots + lots = self._parse_html(response.text, source_url_base=url) + if not lots: + logger.info("avito page=%d: 0 lots — end of pagination", page) + break + all_lots.extend(lots) + logger.info( + "avito page=%d: %d lots (total %d) around (%.4f, %.4f)", + page, len(lots), len(all_lots), lat, lon, + ) + if page < pages: + await self.sleep_between_requests() + + if pages == 1 and all_lots: + # Backward compat: single-page path does trailing sleep (как раньше) + await self.sleep_between_requests() + elif pages > 1: + logger.info( + "avito fetch_around pages=%d total_lots=%d around (%.4f, %.4f)", + pages, len(all_lots), lat, lon, + ) + + return all_lots # ── Strategy B: HTML scrape ───────────────────────────────────────────── def _build_web_url(self, lat: float, lon: float, radius_km: int, page: int = 1) -> str: diff --git a/tradein-mvp/backend/data/sql/051_scrape_runs_extend.sql b/tradein-mvp/backend/data/sql/051_scrape_runs_extend.sql new file mode 100644 index 00000000..cd67f870 --- /dev/null +++ b/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; diff --git a/tradein-mvp/backend/tests/test_city_sweep.py b/tradein-mvp/backend/tests/test_city_sweep.py new file mode 100644 index 00000000..3bc76100 --- /dev/null +++ b/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"] diff --git a/tradein-mvp/deploy/avito-city-sweep.sh b/tradein-mvp/deploy/avito-city-sweep.sh new file mode 100644 index 00000000..082325d2 --- /dev/null +++ b/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 </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