diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index c2c1228d..1982a79e 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -22,7 +22,11 @@ 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, run_n1_city_sweep +from app.services.scrape_pipeline import ( + run_avito_city_sweep, + run_cian_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 @@ -759,6 +763,95 @@ def list_n1_city_sweep_runs( ] +# ── Cian city sweep (#860) — on-demand trigger + runs list ─────────────────── + + +class CianCitySweepStartRequest(BaseModel): + pages_per_anchor: int = Field(default=3, ge=1, le=10) + detail_top_n: int = Field(default=10, ge=0, le=30) + request_delay_sec: float = Field(default=5.0, ge=3.0, le=15.0) + enrich_houses: bool = True + radius_m: int = Field(default=1500, ge=500, le=5000) + + +@router.post("/scrape/cian-city-sweep", response_model=CitySweepStartResponse) +async def start_cian_city_sweep( + payload: CianCitySweepStartRequest, + background_tasks: BackgroundTasks, + db: Annotated[Session, Depends(get_db)], +) -> CitySweepStartResponse: + """Запустить Cian ЕКБ city sweep в background (#860). Returns run_id. + + Один прогон: SERP (fetch_around_multi_room) → detail-обогащение (incl. + price-history) → newbuilding/houses (если enrich_houses=True). + Прокси уже внутри CianScraper. + + Coop cancel: POST /scrape/cian-city-sweep/{run_id}/cancel. + НЕ добавляется в scrape_schedules — ручной/контролируемый запуск + (bulk cron → 429-бан на Cian). + """ + run_id = runs_mod.create_run(db, source="cian_city_sweep", params=payload.model_dump()) + + async def _sweep_task() -> None: + sweep_db = SessionLocal() + try: + await run_cian_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, + detail_top_n=payload.detail_top_n, + enrich_houses=payload.enrich_houses, + ) + except Exception: + logger.exception("cian-sweep background task run_id=%d crashed", run_id) + finally: + sweep_db.close() + + background_tasks.add_task(_sweep_task) + logger.info("cian-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/cian-city-sweep/runs", response_model=list[ScrapeRunRow]) +def list_cian_city_sweep_runs( + db: Annotated[Session, Depends(get_db)], + limit: int = 10, +) -> list[ScrapeRunRow]: + """Список последних N Cian city sweep runs (для UI polling). Default limit=10.""" + rows = runs_mod.list_recent(db, source="cian_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/cian-city-sweep/{run_id}/cancel") +def cancel_cian_city_sweep( + run_id: int, + db: Annotated[Session, Depends(get_db)], +) -> dict[str, object]: + """Отменить running Cian sweep. Cooperative: проверяется каждый anchor.""" + cancelled = runs_mod.mark_cancelled(db, run_id) + return {"ok": True, "run_id": run_id, "cancelled": cancelled} + + # ── Scraper settings: global + per-source delay management ─────────────────── diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 39d61f86..51ad7e15 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -107,6 +107,11 @@ class Settings(BaseSettings): # дают индекс > 2.0 при малой выборке → no-op чтобы избежать регрессию. estimate_quarter_index_max_for_small_n: float = 2.0 estimate_quarter_index_small_n_threshold: int = 50 + # Sanity-clamp на factor = target_index / avg_analog_index (#859). + # Belt-and-suspenders против патологичных FDW-данных. Нормальные квартальные + # индексы РФ лежат в [0.6, 1.8]; за этими порогами — артефакт, а не сигнал. + estimate_quarter_index_factor_min: float = 0.6 + estimate_quarter_index_factor_max: float = 1.8 # ── Estimate enrichment time-budgets (#654) ────────────────────────────── # POST /estimate делает несколько ПОСЛЕДОВАТЕЛЬНЫХ блокирующих сетевых @@ -182,6 +187,13 @@ class Settings(BaseSettings): # Сколько раз сменить IP при блоке прежде чем сдаться (на одну страницу). avito_proxy_max_rotations: int = 2 + # ── #759: deactivate stale avito listings ─────────────────────────────── + # Avito-объявления, не виденные scraper'ом более avito_stale_ttl_days дней, + # помечаются is_active=false ночной задачей deactivate_stale_avito_listings. + # TTL=10 — баланс между "снятое объявление" (обычно 3-7 дней молчания) и + # допуском на перерыв в работе scraper'а. ENV: AVITO_STALE_TTL_DAYS. + avito_stale_ttl_days: int = 10 + # ── Yandex SERP cookies (#801/T4) ─────────────────────────────────────── # Путь к JSON-файлу с cookies браузера (формат: [{name, value, ...}, ...]). # Если задан и файл существует — cookies передаются в curl_cffi-сессию при diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 51bff0d6..96e8c456 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -893,6 +893,8 @@ def _apply_quarter_index( base_range_high: int, target_index: float, avg_analog_index: float, + min_factor: float = 0.6, + max_factor: float = 1.8, ) -> tuple[float, int, int, int, float]: """Чистая (testable без БД) gap-correction квартального индекса (#764). @@ -904,10 +906,27 @@ def _apply_quarter_index( Все ценовые выходы масштабируются одним и тем же factor → median/range остаются геометрически консистентными. + min_factor / max_factor — sanity-clamp (#859): belt-and-suspenders против + патологичных FDW-данных. Калибруются через settings и передаются из + вызывающего кода, чтобы хелпер оставался чистым (без импорта settings). + Когда clamp меняет raw factor — логируется (см. caller). + Returns (adjusted_ppm2, adjusted_median_price, adjusted_range_low, adjusted_range_high, factor). """ - factor = target_index / avg_analog_index + raw_factor = target_index / avg_analog_index + factor = max(min_factor, min(max_factor, raw_factor)) + if raw_factor != factor: + logger.info( + "quarter_index: factor clamped raw=%.4f → %.4f (bounds [%.2f, %.2f])" + " target_index=%.3f avg_analog_index=%.3f", + raw_factor, + factor, + min_factor, + max_factor, + target_index, + avg_analog_index, + ) adjusted_ppm2 = base_median_ppm2 * factor adjusted_median_price = round(base_median_price * factor) adjusted_range_low = round(base_range_low * factor) @@ -2194,6 +2213,8 @@ async def estimate_quality( base_range_high=range_high, target_index=target_qi, avg_analog_index=avg_analog_index, + min_factor=settings.estimate_quarter_index_factor_min, + max_factor=settings.estimate_quarter_index_factor_max, ) analogs_with_qi = sum( 1 for lq, _lp in analog_quarters if lq in analog_index_map diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index cf83e05c..413232a7 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -24,6 +24,10 @@ Sources: yandex detail pages, extracts full address from , updates listings.address for street-only addresses; nightly 02:00-03:00 UTC, before refresh_search_matview) + - deactivate_stale_avito → deactivate_stale_avito_listings + (tasks/deactivate_stale_avito.py, #759; marks avito listings + with last_seen_at > TTL days as is_active=false — no DELETE, + no external calls, enabled by default; window after avito sweep) """ from __future__ import annotations @@ -556,6 +560,41 @@ async def trigger_refresh_search_matview_run( return run_id +async def trigger_deactivate_stale_avito_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch deactivate_stale_avito_listings в executor (sync DB-only task). + + Помечает avito-объявления, не виденные >TTL дней, как is_active=false (#759). + Задача синхронная (чистый internal UPDATE, БЕЗ внешних HTTP-вызовов) — гоняем в + run_in_executor по образцу trigger_listing_source_snapshot_run. SAFE to enable — + schedule seed enabled=true (090), pure internal DB op. + Окно 06:00-07:00 UTC — после avito_city_sweep (02:00-05:00 UTC). + + Returns run_id (или None если skip — есть running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.deactivate_stale_avito import deactivate_stale_avito_listings + + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, deactivate_stale_avito_listings, run_db, run_id) + except Exception: + logger.exception("scheduler: deactivate_stale_avito 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 deactivate_stale_avito run_id=%d", run_id) + return run_id + + 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). @@ -874,6 +913,8 @@ async def scheduler_loop() -> None: await trigger_refresh_search_matview_run(db, sch) elif source == "yandex_address_backfill": await trigger_yandex_address_backfill_run(db, sch) + elif source == "deactivate_stale_avito": + await trigger_deactivate_stale_avito_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 7c50d0b0..bd33cfee 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -490,8 +490,7 @@ async def run_avito_city_sweep( counters.errors_count += imv_result.errors scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) logger.info( - "city-sweep run_id=%d: IMV phase done — " - "attempted=%d enriched=%d failed=%d", + "city-sweep run_id=%d: IMV phase done — attempted=%d enriched=%d failed=%d", run_id, imv_result.checked, imv_result.saved, @@ -673,8 +672,7 @@ async def run_yandex_city_sweep( scrape_runs.mark_done(db, run_id, counters.to_dict()) logger.info( - "yandex-sweep run_id=%d done: anchors=%d/%d pages=%d lots=%d " - "(ins=%d/upd=%d) errors=%d", + "yandex-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, @@ -841,14 +839,333 @@ async def run_n1_city_sweep( raise +# ── Cian city sweep (#860) ────────────────────────────────────────────────── + + +@dataclass +class CianCitySweepCounters: + """Aggregate counters для Cian city sweep run. + + Поля симметричны CitySweepCounters, но без imv_* (у Cian нет IMV-фазы). + detail_* — обогащение через cian_detail (вкл. price-history). + houses_* — обогащение newbuilding/ЖК через cian_newbuilding. + """ + + anchors_total: int = 0 + anchors_done: int = 0 + lots_fetched: int = 0 + lots_inserted: int = 0 + lots_updated: int = 0 + detail_attempted: int = 0 + detail_enriched: int = 0 + detail_failed: int = 0 + houses_attempted: int = 0 + houses_enriched: int = 0 + houses_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)} + + +# Defensive abort threshold: если N подряд anchor'ов упали с исключением +# (не просто вернули пустой результат) — прерываем sweep. +CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES = 3 + + +async def run_cian_city_sweep( + db: Session, + *, + run_id: int, + anchors: list[tuple[float, float, str]] | None = None, + radius_m: int = 1500, + pages_per_anchor: int = 3, + request_delay_sec: float = 5.0, + detail_top_n: int = 10, + enrich_houses: bool = True, +) -> CianCitySweepCounters: + """Cian full city sweep: SERP → detail(+price-history) → newbuilding/houses. + + Симметрично run_avito_city_sweep. Использует EKB_ANCHORS (общий список). + + Фазы per anchor: + 1. SERP: CianScraper.fetch_around_multi_room(lat, lon, radius_m, pages=pages_per_anchor) + 2. SAVE: save_listings(db, lots, run_id=run_id) + 3. DETAIL: top-N свежих cian-листингов без detail_enriched_at → + fetch_detail(source_url) + save_detail_enrichment(db, listing_id, enrichment) + (save_detail_enrichment уже пишет price_changes → offer_price_history) + 4. HOUSES (если enrich_houses=True): для домов из этого anchor'а с cian_zhk_url → + fetch_newbuilding(zhk_url) + save_newbuilding_enrichment(db, house_id, enr) + + Anti-bot: + - CianScraper управляет своей curl_cffi-сессией (прокси уже внутри). + - asyncio.sleep(request_delay_sec) между detail-запросами (jitter ±20%). + - Defensive abort при CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES SERP-ошибок подряд. + + Cooperative cancel: scrape_runs.is_cancelled(db, run_id) перед каждым anchor. + mark_done вызывается ВСЕГДА (finally outer). + """ + from app.services.scrapers.cian import CianScraper + from app.services.scrapers.cian_detail import fetch_detail, save_detail_enrichment + from app.services.scrapers.cian_newbuilding import ( + fetch_newbuilding, + save_newbuilding_enrichment, + ) + + _anchors = anchors if anchors is not None else EKB_ANCHORS + counters = CianCitySweepCounters(anchors_total=len(_anchors)) + consecutive_failures = 0 + + try: + for idx, (lat, lon, name) in enumerate(_anchors, start=1): + # Cooperative cancel перед каждым anchor + if scrape_runs.is_cancelled(db, run_id): + logger.info( + "cian-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( + "cian-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", + run_id, + idx, + len(_anchors), + name, + lat, + lon, + ) + + # ── Phase 1+2: SERP + save ────────────────────────────────────── + anchor_lots: list[ScrapedLot] = [] + try: + async with CianScraper() as scraper: + anchor_lots = await scraper.fetch_around_multi_room( + lat, + lon, + radius_m, + pages=pages_per_anchor, + ) + 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 + logger.info( + "cian-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d", + run_id, + name, + len(anchor_lots), + counters.lots_inserted, + counters.lots_updated, + ) + except Exception as e: + logger.exception("cian-sweep run_id=%d: anchor %s SERP failed", run_id, name) + counters.errors_count += 1 + consecutive_failures += 1 + if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES: + logger.error( + "cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " + "%d consecutive SERP 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"cian sweep aborted: {consecutive_failures} consecutive " + f"anchor SERP failures (last: {e})", + counters.to_dict(), + ) + return counters + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + continue + + # ── Phase 3: detail enrichment (top-N) ───────────────────────── + if detail_top_n > 0: + from sqlalchemy import text as _text + + priority_rows = ( + db.execute( + _text(""" + SELECT id, source_url + FROM listings + WHERE source = 'cian' + AND source_url IS NOT NULL + AND detail_enriched_at IS NULL + AND scraped_at > NOW() - INTERVAL '2 hours' + ORDER BY scraped_at DESC NULLS LAST + LIMIT :lim + """), + {"lim": detail_top_n}, + ) + .mappings() + .all() + ) + for didx, row in enumerate(priority_rows): + listing_id: int = row["id"] + source_url: str = row["source_url"] + counters.detail_attempted += 1 + try: + enrichment = await fetch_detail(source_url) + if enrichment is not None: + save_detail_enrichment(db, listing_id, enrichment) + counters.detail_enriched += 1 + else: + counters.detail_failed += 1 + logger.warning( + "cian-sweep run_id=%d: detail returned None for " + "listing_id=%d url=%s", + run_id, + listing_id, + source_url, + ) + except Exception as exc: + counters.detail_failed += 1 + counters.errors_count += 1 + logger.warning( + "cian-sweep run_id=%d: detail failed listing_id=%d: %s", + run_id, + listing_id, + exc, + ) + if didx < len(priority_rows) - 1: + jitter = random.uniform(0.8, 1.2) + await asyncio.sleep(request_delay_sec * jitter) + logger.info( + "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d", + run_id, + name, + counters.detail_enriched, + counters.detail_attempted, + counters.detail_failed, + ) + + # ── Phase 4: newbuilding/houses enrichment ───────────────────── + if enrich_houses and anchor_lots: + from sqlalchemy import text as _text2 + + # Собираем cian_internal_house_ids из SERP-объявлений этого anchor'а. + # lot.raw_payload["newbuilding_id"] → SERP предоставляет house_ext_id. + # Используем houses.cian_zhk_url для тех домов, у которых она заполнена. + # Fallback: дома у которых cian_internal_house_id совпадает с house_ext_id. + cian_nb_ext_ids = { + lot.house_ext_id + for lot in anchor_lots + if lot.house_source == "cian_newbuilding" and lot.house_ext_id + } + if cian_nb_ext_ids: + nb_id_list = list(cian_nb_ext_ids) + house_rows = ( + db.execute( + _text2(""" + SELECT h.id, h.cian_zhk_url + FROM houses h + WHERE h.ext_source = 'cian_newbuilding' + AND h.ext_id = ANY(CAST(:ids AS text[])) + AND h.cian_zhk_url IS NOT NULL + """), + {"ids": nb_id_list}, + ) + .mappings() + .all() + ) + for hidx, hrow in enumerate(house_rows): + house_id_val: int = hrow["id"] + zhk_url: str = hrow["cian_zhk_url"] + counters.houses_attempted += 1 + try: + nb_enrichment = await fetch_newbuilding(zhk_url) + if nb_enrichment is not None: + save_newbuilding_enrichment(db, house_id_val, nb_enrichment) + counters.houses_enriched += 1 + else: + counters.houses_failed += 1 + logger.warning( + "cian-sweep run_id=%d: newbuilding returned None " + "house_id=%d url=%s", + run_id, + house_id_val, + zhk_url, + ) + except Exception as exc: + counters.houses_failed += 1 + counters.errors_count += 1 + logger.warning( + "cian-sweep run_id=%d: houses failed house_id=%d: %s", + run_id, + house_id_val, + exc, + ) + try: + db.rollback() + except Exception: + pass + if hidx < len(house_rows) - 1: + await asyncio.sleep(request_delay_sec) + logger.info( + "cian-sweep run_id=%d anchor %s: houses=%d/%d failed=%d", + run_id, + name, + counters.houses_enriched, + counters.houses_attempted, + counters.houses_failed, + ) + + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + + # Пауза между anchor'ами (поверх per-request sleep внутри scraper'а) + if idx < len(_anchors): + await asyncio.sleep(request_delay_sec) + + scrape_runs.mark_done(db, run_id, counters.to_dict()) + logger.info( + "cian-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) " + "detail=%d/%d houses=%d/%d errors=%d", + run_id, + counters.anchors_done, + counters.anchors_total, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.detail_enriched, + counters.detail_attempted, + counters.houses_enriched, + counters.houses_attempted, + counters.errors_count, + ) + return counters + + except Exception as exc: + logger.exception("cian-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__ = [ + "CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES", "EKB_ANCHORS", + "CianCitySweepCounters", "CitySweepCounters", "PipelineCounters", "PipelineResult", "YandexCitySweepCounters", "run_avito_city_sweep", "run_avito_pipeline", + "run_cian_city_sweep", "run_yandex_city_sweep", ] diff --git a/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py new file mode 100644 index 00000000..88386463 --- /dev/null +++ b/tradein-mvp/backend/app/tasks/deactivate_stale_avito.py @@ -0,0 +1,70 @@ +"""Deactivate stale avito listings that have not been seen in >TTL days (#759). + +Нечасто обновляемые объявления на Avito (last_seen_at < NOW() - TTL) помечаются +is_active=false, чтобы coverage/estimator denominators не раздувались мёртвыми +объявлениями. Строки НЕ удаляются — история нужна для бэктеста (#667). + +Только source='avito'. Cian/Yandex не затрагиваются. + +Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app +scheduler'ом через trigger_deactivate_stale_avito_run() (scheduler.py), +по образцу snapshot_listing_sources / recompute_asking_to_sold_ratios +(sync task в run_in_executor). + +TTL берётся из settings.avito_stale_ttl_days (env AVITO_STALE_TTL_DAYS, default 10). +""" + +from __future__ import annotations + +import logging + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.config import settings +from app.services import scrape_runs as runs_mod + +logger = logging.getLogger(__name__) + +# UPDATE: помечаем is_active=false для avito-объявлений, не виденных >TTL дней. +# CAST(:ttl_days || ' days' AS interval) — psycopg v3 safe (никаких :param::type). +# Только source='avito' и is_active=true — чтобы не трогать уже неактивные строки +# и не затрагивать cian/yandex. +_DEACTIVATE_SQL = text( + """ + UPDATE listings + SET is_active = false + WHERE source = 'avito' + AND is_active = true + AND last_seen_at < NOW() - CAST(:ttl_days || ' days' AS interval) + """ +) + + +def deactivate_stale_avito_listings(db: Session, run_id: int) -> dict[str, int]: + """Пометить is_active=false все avito-объявления с last_seen_at > TTL дней назад. + + Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources). + Один UPDATE в транзакции. Финализирует scrape_runs (mark_done / mark_failed). + + Returns {"deactivated": N} — количество обновлённых строк. + """ + counters: dict[str, int] = {"deactivated": 0} + try: + result = db.execute(_DEACTIVATE_SQL, {"ttl_days": settings.avito_stale_ttl_days}) + counters["deactivated"] = result.rowcount or 0 + + db.commit() + runs_mod.mark_done(db, run_id, counters) + logger.info( + "deactivate_stale_avito run_id=%d done: deactivated=%d (ttl_days=%d)", + run_id, + counters["deactivated"], + settings.avito_stale_ttl_days, + ) + return counters + except Exception as exc: + logger.exception("deactivate_stale_avito run_id=%d failed", run_id) + db.rollback() + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/data/sql/090_scrape_schedules_seed_deactivate_stale_avito.sql b/tradein-mvp/backend/data/sql/090_scrape_schedules_seed_deactivate_stale_avito.sql new file mode 100644 index 00000000..c741e58a --- /dev/null +++ b/tradein-mvp/backend/data/sql/090_scrape_schedules_seed_deactivate_stale_avito.sql @@ -0,0 +1,65 @@ +-- 090_scrape_schedules_seed_deactivate_stale_avito.sql +-- Issue #759 — seed scrape_schedules row for nightly deactivation of stale avito listings. +-- +-- Context: +-- Avito listings that haven't been seen by the scraper for >AVITO_STALE_TTL_DAYS days +-- (default 10) remain is_active=true indefinitely. This inflates the active-listing +-- denominator used by the coverage and estimator services, degrading accuracy. +-- +-- Fix: +-- 1. app/tasks/deactivate_stale_avito.py — deactivate_stale_avito_listings() +-- issues a single UPDATE listings SET is_active=false WHERE source='avito' +-- AND is_active=true AND last_seen_at < NOW()-INTERVAL(:ttl_days days). +-- Rows are NOT deleted — history is required for backtest (#667). +-- 2. scheduler.py — added trigger_deactivate_stale_avito_run() + dispatch branch +-- `elif source == "deactivate_stale_avito"` in scheduler_loop(). +-- 3. app/core/config.py — AVITO_STALE_TTL_DAYS setting (default 10). +-- 4. This migration — seeds the scrape_schedules row. +-- +-- Schedule window 06:00-07:00 UTC: +-- After the avito_city_sweep window (02:00-05:00 UTC per #814). Running after the +-- sweep ensures listings genuinely seen in the nightly pass are NOT deactivated. +-- +-- next_run_at bootstrapped to tomorrow 06:00 UTC so the scheduler does not fire +-- immediately on deploy (same pattern as 079/082/088). +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)). +-- listings table with last_seen_at column (002_core_tables.sql). +-- scheduler.py change (trigger_deactivate_stale_avito_run must be deployed). +-- +-- Deploy order: +-- Apply this migration after deploying the scheduler.py + task changes so the +-- scheduler can dispatch the new source name correctly on first fire. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'deactivate_stale_avito', + true, -- SAFE: pure internal DB UPDATE, no ext calls + 6, + 7, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 6)) AT TIME ZONE 'UTC', + '{}'::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), ' + 'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), ' + 'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), ' + 'deactivate_stale_avito (#759).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_cian_city_sweep.py b/tradein-mvp/backend/tests/test_cian_city_sweep.py new file mode 100644 index 00000000..e3a5a3a7 --- /dev/null +++ b/tradein-mvp/backend/tests/test_cian_city_sweep.py @@ -0,0 +1,574 @@ +"""Offline unit tests for run_cian_city_sweep (#860). + +Pure & fast: NO live network, NO DB. Mirrors test_n1_city_sweep / test_city_sweep. +Asserts orchestration semantics only. + +Coverage: + - Все фазы (SERP → save → detail → houses) вызываются в нужном порядке + - CianCitySweepCounters корректно заполняются + - Cooperative cancel прерывает sweep перед anchor'ом + - enrich_houses=False пропускает houses-фазу + - Consecutive SERP failures → mark_banned + abort + - Per-item detail/house error не валит весь sweep + - mark_done вызывается при нормальном завершении +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.services import scrape_pipeline +from app.services.scrapers.base import ScrapedLot +from app.services.scrapers.cian import CianScraper + +# Три тестовых anchor'а — не все 5, чтобы тесты были быстрее +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, *, nb: bool = False) -> ScrapedLot: + """Создаём ScrapedLot с опциональным newbuilding-ссылкой.""" + return ScrapedLot( + source="cian", + source_url=f"https://ekb.cian.ru/sale/flat/{offer_id}/", + source_id=offer_id, + address="Екатеринбург, улица Тестовая, 1", + price_rub=5_000_000, + area_m2=50.0, + rooms=2, + house_source="cian_newbuilding" if nb else "cian", + house_ext_id="99999" if nb else None, + ) + + +class _FakeRuns: + """Stub для scrape_runs: отслеживает вызовы, поддерживает cooperative cancel.""" + + 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: + """asyncio.sleep → instant (no test delays).""" + + async def _instant(_secs: float) -> None: + return None + + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", _instant) + + +@pytest.fixture(autouse=True) +def _stub_cian_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None: + """Нейтрализуем CianScraper.__init__/__aenter__/__aexit__ — нет DB вызовов.""" + + def _init(self: CianScraper) -> None: + self.request_delay_sec = 0.0 + self._cffi = None + + async def _aenter(self: CianScraper) -> CianScraper: + return self + + async def _aexit(self: CianScraper, *_args: Any) -> None: + return None + + monkeypatch.setattr(CianScraper, "__init__", _init) + monkeypatch.setattr(CianScraper, "__aenter__", _aenter) + monkeypatch.setattr(CianScraper, "__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) + + +# ── CianCitySweepCounters ─────────────────────────────────────────────────── + + +def test_cian_sweep_counters_defaults() -> None: + from app.services.scrape_pipeline import CianCitySweepCounters + + c = CianCitySweepCounters() + assert c.anchors_total == 0 + assert c.lots_fetched == 0 + assert c.detail_attempted == 0 + assert c.houses_attempted == 0 + assert c.errors_count == 0 + + +def test_cian_sweep_counters_to_dict_all_keys() -> None: + from dataclasses import fields + + from app.services.scrape_pipeline import CianCitySweepCounters + + c = CianCitySweepCounters() + d = c.to_dict() + expected = {f.name for f in fields(c)} + assert set(d.keys()) == expected + + +# ── Full sweep — фазы и агрегация ────────────────────────────────────────── + + +async def test_sweep_iterates_all_anchors_serp_and_save( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """SERP вызывается для каждого anchor, save_listings получает накопленные lots.""" + serp_calls: list[str] = [] + save_calls: list[int] = [] + + async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + serp_calls.append(f"{lat:.4f}") + return [_fake_lot(f"{lat}-{i}") for i in range(3)] + + def fake_save(_db: Any, lots: list, *, run_id: int | None = None): + save_calls.append(len(lots)) + return len(lots), 0 + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save) + + # Stub detail + houses fetchers (возвращают None → graceful skip) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", + AsyncMock(return_value=None), + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=1, + anchors=TEST_ANCHORS, + pages_per_anchor=2, + detail_top_n=0, # skip detail phase + enrich_houses=False, + request_delay_sec=0.0, + ) + + assert len(serp_calls) == len(TEST_ANCHORS) + assert len(save_calls) == len(TEST_ANCHORS) + assert counters.anchors_done == len(TEST_ANCHORS) + assert counters.lots_fetched == len(TEST_ANCHORS) * 3 + assert counters.lots_inserted == len(TEST_ANCHORS) * 3 + assert counters.errors_count == 0 + assert fake.done is not None + assert fake.done["lots_fetched"] == len(TEST_ANCHORS) * 3 + assert fake.failed is None + assert fake.banned is None + + +async def test_detail_phase_runs_and_fills_counters( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """detail-фаза вызывается и счётчики заполняются.""" + from app.services.scrapers.cian_detail import DetailEnrichment + + async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + return [_fake_lot(f"{lat}-{i}") for i in range(2)] + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + + # DB mock returning 2 rows for detail query + mock_db = MagicMock() + mock_db.execute.return_value.mappings.return_value.all.return_value = [ + {"id": 101, "source_url": "https://ekb.cian.ru/sale/flat/101/"}, + {"id": 102, "source_url": "https://ekb.cian.ru/sale/flat/102/"}, + ] + + fake_enrichment = DetailEnrichment(cian_id=101) + mock_fetch_detail = AsyncMock(return_value=fake_enrichment) + mock_save_detail = MagicMock() + + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", + mock_fetch_detail, + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.save_detail_enrichment", + mock_save_detail, + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=mock_db, + run_id=2, + anchors=TEST_ANCHORS[:1], # один anchor + detail_top_n=2, + enrich_houses=False, + request_delay_sec=0.0, + ) + + assert counters.detail_attempted == 2 + assert counters.detail_enriched == 2 + assert counters.detail_failed == 0 + assert mock_fetch_detail.call_count == 2 + assert mock_save_detail.call_count == 2 + assert fake.done is not None + + +async def test_enrich_houses_false_skips_houses_phase( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """enrich_houses=False → houses-фаза не выполняется.""" + nb_fetch_calls: list[str] = [] + + async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + return [_fake_lot("1001", nb=True)] + + async def fake_fetch_nb(url: str, **_: Any) -> None: + nb_fetch_calls.append(url) + return None + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + monkeypatch.setattr( + "app.services.scrapers.cian_newbuilding.fetch_newbuilding", + fake_fetch_nb, + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", + AsyncMock(return_value=None), + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=3, + anchors=TEST_ANCHORS[:1], + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + ) + + assert len(nb_fetch_calls) == 0 + assert counters.houses_attempted == 0 + assert counters.houses_enriched == 0 + assert fake.done is not None + + +async def test_cancel_midway_stops_remaining( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Cooperative cancel перед anchor #2 → anchor #1 выполняется, #2+ нет.""" + serp_calls: list[str] = [] + + async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + serp_calls.append(f"{lat:.4f}") + return [_fake_lot(f"{lat}-1")] + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", + AsyncMock(return_value=None), + ) + + fake = _FakeRuns(cancel_at_anchor=2) # cancel при is_cancelled_calls >= 2 → перед anchor #2 + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=4, + anchors=TEST_ANCHORS, + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + ) + + assert len(serp_calls) == 1, "только anchor #1 должен выполниться" + assert counters.anchors_done == 1 + assert fake.done is None + assert fake.banned is None + + +async def test_consecutive_serp_failures_abort_with_banned( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """N подряд SERP-ошибок → mark_banned + abort sweep.""" + from app.services.scrape_pipeline import CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES + + attempt_count = 0 + + async def fake_fetch_fail(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + nonlocal attempt_count + attempt_count += 1 + raise RuntimeError("cian blocked") + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_fail) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + # Передаём больше anchor'ов чем CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES + anchors = [*TEST_ANCHORS, (56.77, 60.55, "A4"), (56.865, 60.62, "A5")] + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=5, + anchors=anchors, + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + ) + + assert attempt_count == CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES + assert counters.errors_count == CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES + assert fake.banned is not None, "mark_banned должен быть вызван" + assert "consecutive" in fake.banned[0] + assert fake.done is None + + +async def test_per_item_detail_error_does_not_abort_sweep( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Ошибка при detail-обогащении одного листинга не валит sweep.""" + call_count = 0 + + async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + return [_fake_lot(f"lot-{idx}") for idx in range(2)] + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + + mock_db = MagicMock() + # Возвращаем 2 listing-а для detail + mock_db.execute.return_value.mappings.return_value.all.return_value = [ + {"id": 201, "source_url": "https://ekb.cian.ru/sale/flat/201/"}, + {"id": 202, "source_url": "https://ekb.cian.ru/sale/flat/202/"}, + ] + + async def fetch_detail_sometimes_fails(url: str, **_: Any): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise RuntimeError("transient network error") + from app.services.scrapers.cian_detail import DetailEnrichment + + return DetailEnrichment(cian_id=202) + + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", + fetch_detail_sometimes_fails, + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.save_detail_enrichment", + MagicMock(), + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=mock_db, + run_id=6, + anchors=TEST_ANCHORS[:1], + detail_top_n=2, + enrich_houses=False, + request_delay_sec=0.0, + ) + + assert counters.detail_attempted == 2 + assert counters.detail_failed == 1 + assert counters.detail_enriched == 1 + assert counters.errors_count >= 1 + assert fake.done is not None, "sweep должен завершиться mark_done несмотря на ошибку" + assert fake.banned is None + + +async def test_mark_done_always_called( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """mark_done вызывается даже при пустом SERP.""" + + async def fake_fetch_empty(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + return [] + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_empty) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", + AsyncMock(return_value=None), + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=7, + anchors=TEST_ANCHORS, + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + ) + + assert fake.done is not None + assert counters.lots_fetched == 0 + assert counters.anchors_done == len(TEST_ANCHORS) + + +# ── Admin endpoint tests (offline) ────────────────────────────────────────── + + +@pytest.fixture +def app_with_admin(): + from unittest.mock import MagicMock + + from fastapi import FastAPI + from fastapi.testclient import TestClient + + 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_cian_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/cian-city-sweep", + json={"pages_per_anchor": 99}, + ) + assert r.status_code == 422 + + +def test_cian_start_endpoint_validates_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/cian-city-sweep", + json={"request_delay_sec": 1.0}, + ) + assert r.status_code == 422 + + +def test_cian_start_endpoint_ok(app_with_admin) -> None: + """Valid request → 200, run_id в ответе.""" + from unittest.mock import AsyncMock, patch + + with ( + patch("app.services.scrape_runs.create_run", return_value=42), + patch("app.services.scrape_pipeline.run_cian_city_sweep", new_callable=AsyncMock), + ): + r = app_with_admin.post( + "/api/v1/admin/scrape/cian-city-sweep", + json={"pages_per_anchor": 2, "detail_top_n": 5, "enrich_houses": False}, + ) + assert r.status_code == 200 + body = r.json() + assert body["run_id"] == 42 + assert body["status"] == "running" + assert body["pages_per_anchor"] == 2 + assert body["detail_top_n"] == 5 + + +def test_cian_cancel_endpoint(app_with_admin) -> None: + """Cancel endpoint возвращает run_id + cancelled=True.""" + from unittest.mock import patch + + with patch("app.services.scrape_runs.mark_cancelled", return_value=True): + r = app_with_admin.post("/api/v1/admin/scrape/cian-city-sweep/77/cancel") + assert r.status_code == 200 + body = r.json() + assert body["run_id"] == 77 + assert body["cancelled"] is True + assert body["ok"] is True + + +def test_cian_cancel_endpoint_not_running(app_with_admin) -> None: + """Cancel на не-running sweep → cancelled=False.""" + from unittest.mock import patch + + with patch("app.services.scrape_runs.mark_cancelled", return_value=False): + r = app_with_admin.post("/api/v1/admin/scrape/cian-city-sweep/88/cancel") + assert r.status_code == 200 + assert r.json()["cancelled"] is False + + +def test_cian_list_runs_endpoint(app_with_admin) -> None: + """List runs endpoint возвращает список.""" + from datetime import UTC, datetime + from unittest.mock import patch + + fake_rows = [ + { + "run_id": 10, + "source": "cian_city_sweep", + "status": "done", + "params": {"pages_per_anchor": 3}, + "counters": {"lots_fetched": 150}, + "error": None, + "started_at": datetime(2025, 5, 1, tzinfo=UTC), + "finished_at": datetime(2025, 5, 1, 1, tzinfo=UTC), + "heartbeat_at": datetime(2025, 5, 1, 1, tzinfo=UTC), + } + ] + with patch("app.services.scrape_runs.list_recent", return_value=fake_rows): + r = app_with_admin.get("/api/v1/admin/scrape/cian-city-sweep/runs") + assert r.status_code == 200 + body = r.json() + assert len(body) == 1 + assert body[0]["run_id"] == 10 + assert body[0]["source"] == "cian_city_sweep" + assert "2025-05-01" in body[0]["started_at"] diff --git a/tradein-mvp/backend/tests/test_deactivate_stale_avito.py b/tradein-mvp/backend/tests/test_deactivate_stale_avito.py new file mode 100644 index 00000000..aed99581 --- /dev/null +++ b/tradein-mvp/backend/tests/test_deactivate_stale_avito.py @@ -0,0 +1,254 @@ +"""Tests for the stale avito listing deactivation task (#759). + +Static assertions (SQL shape, psycopg-v3 cast discipline, scheduler wiring) mirror +the pattern established in test_listing_source_snapshot.py. One behavioural test drives +the counter logic via a fake db without a live database. +""" + +import inspect +import os +import re +from pathlib import Path +from typing import Any + +import pytest + +# Stub DATABASE_URL before any app import (same guard as test_scheduler.py / +# test_listing_source_snapshot.py — these tests are static/fake-db, no live DB). +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services import scheduler +from app.tasks import deactivate_stale_avito as task_mod + +_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql" +_MIGRATION_090 = _SQL_DIR / "090_scrape_schedules_seed_deactivate_stale_avito.sql" + +_DEACTIVATE_SQL = str(task_mod._DEACTIVATE_SQL.text) +_TASK_SRC = inspect.getsource(task_mod.deactivate_stale_avito_listings) + + +# ── SQL shape ───────────────────────────────────────────────────────────────── + + +def test_sql_targets_only_avito_source() -> None: + """UPDATE must filter source='avito' — cian/yandex rows must not be touched.""" + assert "source = 'avito'" in _DEACTIVATE_SQL + + +def test_sql_only_deactivates_not_deletes() -> None: + """Rows must be marked is_active=false, never DELETEd (history needed for #667).""" + assert "SET is_active = false" in _DEACTIVATE_SQL + assert "DELETE" not in _DEACTIVATE_SQL.upper() + + +def test_sql_guards_already_inactive_rows() -> None: + """Filter AND is_active=true avoids redundant updates on rows already deactivated.""" + assert "is_active = true" in _DEACTIVATE_SQL + + +def test_sql_uses_last_seen_at_threshold() -> None: + """Threshold is last_seen_at < NOW() - interval, not any other column.""" + assert "last_seen_at" in _DEACTIVATE_SQL + assert "NOW()" in _DEACTIVATE_SQL + + +def test_sql_uses_cast_interval_not_double_colon() -> None: + """psycopg v3: bind param via CAST(:ttl_days || ' days' AS interval), never ::interval.""" + assert "CAST(:ttl_days || ' days' AS interval)" in _DEACTIVATE_SQL + # No bind-param (:name) followed by ::type anywhere. + assert not re.search(r":\w+::", _DEACTIVATE_SQL) + + +def test_sql_ttl_param_named_ttl_days() -> None: + """Bind param must be :ttl_days (wired from settings.avito_stale_ttl_days).""" + assert ":ttl_days" in _DEACTIVATE_SQL + + +# ── Settings wiring ─────────────────────────────────────────────────────────── + + +def test_settings_has_avito_stale_ttl_days() -> None: + from app.core.config import settings + + assert hasattr(settings, "avito_stale_ttl_days") + assert settings.avito_stale_ttl_days == 10 # default + + +def test_task_reads_ttl_from_settings() -> None: + """Task must use settings.avito_stale_ttl_days, not a hardcoded literal.""" + assert "settings.avito_stale_ttl_days" in _TASK_SRC + + +# ── Task function contract ──────────────────────────────────────────────────── + + +def test_task_returns_deactivated_counter() -> None: + assert '"deactivated"' in _TASK_SRC or "'deactivated'" in _TASK_SRC + + +def test_task_finalises_run() -> None: + assert "mark_done" in _TASK_SRC + assert "mark_failed" in _TASK_SRC + + +def test_task_commits_and_rollbacks_on_error() -> None: + assert "db.commit()" in _TASK_SRC + assert "db.rollback()" in _TASK_SRC + + +# ── Scheduler wiring ────────────────────────────────────────────────────────── + + +def test_trigger_fn_exists_and_is_async() -> None: + fn = getattr(scheduler, "trigger_deactivate_stale_avito_run", None) + assert fn is not None, "trigger_deactivate_stale_avito_run missing from scheduler" + assert inspect.iscoroutinefunction(fn) + + +def test_dispatch_branch_wired() -> None: + """scheduler_loop dispatches source='deactivate_stale_avito' to the trigger.""" + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "deactivate_stale_avito"' in loop_src + assert "trigger_deactivate_stale_avito_run(db, sch)" in loop_src + + +def test_trigger_runs_sync_task_in_executor() -> None: + """Sync DB-only task → run_in_executor, not a bare await.""" + trig_src = inspect.getsource(scheduler.trigger_deactivate_stale_avito_run) + # run_in_executor call may span two lines; check both parts separately. + assert "run_in_executor(" in trig_src + assert "deactivate_stale_avito_listings" in trig_src + assert "_claim_run(db, schedule_row)" in trig_src + + +# ── Migration 090 ───────────────────────────────────────────────────────────── + + +def test_migration_090_exists() -> None: + assert _MIGRATION_090.is_file(), f"missing migration: {_MIGRATION_090}" + + +def test_migration_090_seeds_correct_source_and_window() -> None: + sql = _MIGRATION_090.read_text("utf-8") + assert "'deactivate_stale_avito'" in sql + # Window 06:00-07:00 UTC — after avito sweep (02:00-05:00). + assert re.search(r"\b6\b", sql), "window_start_hour 6 missing" + assert re.search(r"\b7\b", sql), "window_end_hour 7 missing" + + +def test_migration_090_is_idempotent() -> None: + sql = _MIGRATION_090.read_text("utf-8") + assert "ON CONFLICT (source) DO NOTHING" in sql + + +def test_migration_090_enabled_true() -> None: + sql = _MIGRATION_090.read_text("utf-8") + insert_block = sql.split("INSERT INTO scrape_schedules")[1] + # enabled=true (SAFE — pure internal DB UPDATE). + assert "true," in insert_block + + +def test_migration_090_is_transactional() -> None: + sql = _MIGRATION_090.read_text("utf-8") + assert "BEGIN;" in sql + assert "COMMIT;" in sql + + +def test_migration_090_no_psycopg_trap() -> None: + """Migration is plain DDL/DML seed (no bind params), guard :x::type trap anyway.""" + sql = _MIGRATION_090.read_text("utf-8") + assert not re.search(r":\w+::", sql) + + +# ── Behavioural test: counter logic via fake db ─────────────────────────────── + + +class _FakeResult: + def __init__(self, rowcount: int) -> None: + self.rowcount = rowcount + + +class _FakeDB: + """Minimal stand-in for SQLAlchemy Session.""" + + def __init__(self, rowcount: int) -> None: + self._rowcount = rowcount + self.executed: list[Any] = [] + self.committed = False + self.rolled_back = False + + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + self.executed.append((stmt, params)) + return _FakeResult(self._rowcount) + + def commit(self) -> None: + self.committed = True + + def rollback(self) -> None: + self.rolled_back = True + + +def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None: + """deactivate_stale_avito_listings maps execute() rowcount → counters + marks done.""" + marked: dict[str, Any] = {} + monkeypatch.setattr( + task_mod.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + + db = _FakeDB(rowcount=137) + out = task_mod.deactivate_stale_avito_listings(db, run_id=42) # type: ignore[arg-type] + + assert out == {"deactivated": 137} + assert db.committed is True + assert len(db.executed) == 1 + # Bind params must contain ttl_days wired from settings. + _stmt, params = db.executed[0] + assert params is not None + assert "ttl_days" in params + assert params["ttl_days"] == 10 # default from settings + # run_id finalised via mark_done. + assert marked["run_id"] == 42 + assert marked["counters"] == {"deactivated": 137} + + +def test_counter_logic_zero_rows(monkeypatch: pytest.MonkeyPatch) -> None: + """rowcount=0 → deactivated=0, still commits and marks done.""" + marked: dict[str, Any] = {} + monkeypatch.setattr( + task_mod.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None) + + db = _FakeDB(rowcount=0) + out = task_mod.deactivate_stale_avito_listings(db, run_id=1) # type: ignore[arg-type] + + assert out == {"deactivated": 0} + assert db.committed is True + assert marked["counters"] == {"deactivated": 0} + + +def test_failure_path_rollback_and_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None: + """On execute error: rollback + mark_failed + re-raise (no silent swallow).""" + failed: dict[str, Any] = {} + monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr( + task_mod.runs_mod, + "mark_failed", + lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err), + ) + + class _BoomDB(_FakeDB): + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + raise RuntimeError("db exploded") + + db = _BoomDB(rowcount=0) + with pytest.raises(RuntimeError, match="db exploded"): + task_mod.deactivate_stale_avito_listings(db, run_id=7) # type: ignore[arg-type] + + assert db.rolled_back is True + assert failed["run_id"] == 7 diff --git a/tradein-mvp/backend/tests/test_estimator_quarter_index.py b/tradein-mvp/backend/tests/test_estimator_quarter_index.py index e701acdd..956d3f3e 100644 --- a/tradein-mvp/backend/tests/test_estimator_quarter_index.py +++ b/tradein-mvp/backend/tests/test_estimator_quarter_index.py @@ -581,7 +581,11 @@ def test_bimodal_guard_skips_high_index_small_n() -> None: def test_bimodal_guard_allows_high_index_large_n() -> None: - """Bimodal guard НЕ срабатывает при price_index>2.0 если n_deals>=50.""" + """Bimodal guard НЕ срабатывает при price_index>2.0 если n_deals>=50. + + Коррекция применяется, но raw factor=2.5 зажат #859-clamp до max_factor=1.8. + Медиана меняется (guard не блокирует), но масштабируется на 1.8, не 2.5. + """ analogs_no_cadnum = [ _make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3) ] @@ -589,12 +593,13 @@ def test_bimodal_guard_allows_high_index_large_n() -> None: est = _run_estimate_qi( analogs=analogs_no_cadnum, dadata_cadnum=f"{_TARGET_QUARTER}:350", - qi_lookup_result=(2.5, 60), # index>2.0 но n=60>=50 → применяется + qi_lookup_result=(2.5, 60), # index>2.0 но n=60>=50 → bimodal guard не срабатывает flag_enabled=True, ) - # factor = 2.5 → медиана должна измениться + # Коррекция применена: медиана != base_median (bimodal guard не заблокировал). + # factor=2.5 > max_factor=1.8 → зажат до 1.8 (#859). assert est.median_price_rub != base_median - assert est.median_price_rub == round(base_median * 2.5) + assert est.median_price_rub == round(base_median * 1.8) # ───────────────────────────────────────────────────────────────────────────── @@ -864,5 +869,119 @@ def test_guard1b_imv_anchor_below_blend_threshold_prevents_correction() -> None: assert "квартал" not in (est.confidence_explanation or "").lower() +# ───────────────────────────────────────────────────────────────────────────── +# #859 — sanity-clamp на factor = target_index / avg_analog_index +# ───────────────────────────────────────────────────────────────────────────── + + +def test_apply_quarter_index_clamp_extreme_high_raw_factor() -> None: + """Экстремально низкий avg_analog_index → raw factor >> max → зажат до max_factor. + + target_index=1.0, avg_analog_index=0.3 → raw=3.33, clamped→1.8. + Выходы масштабируются на 1.8, не на 3.33. + """ + max_f = 1.8 + ppm2, median, low, high, factor = _apply_quarter_index( + base_median_ppm2=100_000.0, + base_median_price=5_000_000, + base_range_low=4_000_000, + base_range_high=6_000_000, + target_index=1.0, + avg_analog_index=0.3, + min_factor=0.6, + max_factor=max_f, + ) + assert abs(factor - max_f) < 1e-9, f"Expected factor={max_f}, got {factor}" + assert abs(ppm2 - 100_000.0 * max_f) < 1.0 + assert median == round(5_000_000 * max_f) + assert low == round(4_000_000 * max_f) + assert high == round(6_000_000 * max_f) + # Убеждаемся, что raw factor действительно был бы за пределами clamp + raw = 1.0 / 0.3 + assert raw > max_f + + +def test_apply_quarter_index_clamp_extreme_low_raw_factor() -> None: + """Экстремально высокий avg_analog_index → raw factor << min → зажат до min_factor. + + target_index=0.5, avg_analog_index=2.0 → raw=0.25, clamped→0.6. + Выходы масштабируются на 0.6, не на 0.25. + """ + min_f = 0.6 + ppm2, median, low, high, factor = _apply_quarter_index( + base_median_ppm2=100_000.0, + base_median_price=5_000_000, + base_range_low=4_000_000, + base_range_high=6_000_000, + target_index=0.5, + avg_analog_index=2.0, + min_factor=min_f, + max_factor=1.8, + ) + assert abs(factor - min_f) < 1e-9, f"Expected factor={min_f}, got {factor}" + assert abs(ppm2 - 100_000.0 * min_f) < 1.0 + assert median == round(5_000_000 * min_f) + assert low == round(4_000_000 * min_f) + assert high == round(6_000_000 * min_f) + # Убеждаемся, что raw factor действительно был бы за пределами clamp + raw = 0.5 / 2.0 + assert raw < min_f + + +def test_apply_quarter_index_clamp_normal_factor_unchanged() -> None: + """Нормальный factor внутри [0.6, 1.8] → clamp не меняет значение (регрессия). + + target_index=1.3, avg_analog_index=1.1 → raw=1.182, внутри [0.6,1.8] → без изменений. + """ + raw_expected = 1.3 / 1.1 + ppm2, median, _low, _high, factor = _apply_quarter_index( + base_median_ppm2=200_000.0, + base_median_price=10_000_000, + base_range_low=8_000_000, + base_range_high=12_000_000, + target_index=1.3, + avg_analog_index=1.1, + min_factor=0.6, + max_factor=1.8, + ) + assert abs(factor - raw_expected) < 1e-9, f"Expected {raw_expected}, got {factor}" + assert abs(ppm2 - 200_000.0 * raw_expected) < 1.0 + assert median == round(10_000_000 * raw_expected) + + +def test_apply_quarter_index_clamp_boundary_at_max_exact() -> None: + """factor точно на верхней границе (=1.8) → clamp не применяется.""" + # target=1.8, avg=1.0 → raw=1.8 ровно = max_factor + _ppm2, median, _low, _high, factor = _apply_quarter_index( + base_median_ppm2=100_000.0, + base_median_price=5_000_000, + base_range_low=4_000_000, + base_range_high=6_000_000, + target_index=1.8, + avg_analog_index=1.0, + min_factor=0.6, + max_factor=1.8, + ) + assert abs(factor - 1.8) < 1e-9 + assert median == round(5_000_000 * 1.8) + + +def test_apply_quarter_index_clamp_boundary_at_min_exact() -> None: + """factor точно на нижней границе (=0.6) → clamp не применяется.""" + # target=0.6, avg=1.0 → raw=0.6 ровно = min_factor + _ppm2, median, _low, _high, factor = _apply_quarter_index( + base_median_ppm2=100_000.0, + base_median_price=5_000_000, + base_range_low=4_000_000, + base_range_high=6_000_000, + target_index=0.6, + avg_analog_index=1.0, + min_factor=0.6, + max_factor=1.8, + ) + assert abs(factor - 0.6) < 1e-9 + assert median == round(5_000_000 * 0.6) + + if __name__ == "__main__": # pragma: no cover raise SystemExit(pytest.main([__file__, "-q"]))