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

@@ -22,11 +22,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 (
+from app.services.scrape_pipeline import run_avito_city_sweep, run_n1_city_sweep
-    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
@@ -763,95 +759,6 @@ 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 ───────────────────
 
 

tradein-mvp/backend/app/core/config.py

@@ -107,11 +107,6 @@ 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 делает несколько ПОСЛЕДОВАТЕЛЬНЫХ блокирующих сетевых
@@ -187,13 +182,6 @@ 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-сессию при

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

@@ -893,8 +893,6 @@ 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).
 
@@ -906,27 +904,10 @@ 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).
     """
-    raw_factor = target_index / avg_analog_index
+    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)
@@ -2213,8 +2194,6 @@ 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

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

@@ -24,10 +24,6 @@ Sources:
                               yandex detail pages, extracts full address from <title>,
                               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
@@ -560,41 +556,6 @@ 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).
 
@@ -913,8 +874,6 @@ 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:

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

@@ -490,7 +490,8 @@ 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,
@@ -672,7 +673,8 @@ 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,
@@ -839,333 +841,14 @@ 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",
 ]

tradein-mvp/backend/app/tasks/deactivate_stale_avito.py

@@ -1,70 +0,0 @@
-"""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

tradein-mvp/backend/data/sql/090_scrape_schedules_seed_deactivate_stale_avito.sql

@@ -1,65 +0,0 @@
--- 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;

tradein-mvp/backend/tests/test_cian_city_sweep.py

@@ -1,574 +0,0 @@
-"""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"]

tradein-mvp/backend/tests/test_deactivate_stale_avito.py

@@ -1,254 +0,0 @@
-"""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

tradein-mvp/backend/tests/test_estimator_quarter_index.py

@@ -581,11 +581,7 @@ 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)
     ]
@@ -593,13 +589,12 @@ 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 → bimodal guard не срабатывает
+        qi_lookup_result=(2.5, 60),  # index>2.0 но n=60>=50 → применяется
         flag_enabled=True,
     )
-    # Коррекция применена: медиана != base_median (bimodal guard не заблокировал).
+    # factor = 2.5 → медиана должна измениться
-    # 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 * 1.8)
+    assert est.median_price_rub == round(base_median * 2.5)
 
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -869,119 +864,5 @@ 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"]))