feat(rewire): scheduler_main → kit-scheduler behind USE_KIT_SCHEDULER flag, ship-dark (#2192)

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

@@ -18,6 +18,16 @@ class Settings(BaseSettings):
     # Default true preserves existing behaviour.
     scheduler_enable: bool = Field(default=True, validation_alias="SCHEDULER_ENABLE")
 
+    # ── strangler-миграция scheduler → scraper_kit (#2192, ship-dark) ─────────
+    # Дефолт False → scheduler_main идёт СТАРЫМ путём (app.services.scheduler.scheduler_loop),
+    # прод не меняет поведение на деплое. True → kit-путь (scraper_kit.orchestration.scheduler
+    # + product_handlers). Переключается вручную после prod-smoke; боевой scheduler остаётся
+    # мгновенным fallback.
+    use_kit_scheduler: bool = Field(default=False, validation_alias="USE_KIT_SCHEDULER")
+    # Proxy-pool флаги — объявлены заранее (потребители P3/P4 отдельно), дефолт False.
+    use_proxy_pool_curl: bool = Field(default=False, validation_alias="USE_PROXY_POOL_CURL")
+    use_proxy_pool_browser: bool = Field(default=False, validation_alias="USE_PROXY_POOL_BROWSER")
+
     cors_origins: list[str] = ["http://localhost", "http://localhost:3000", "http://localhost:8080"]
     environment: str = "dev"
 

tradein-mvp/backend/app/scheduler_main.py

@@ -112,6 +112,47 @@ async def _await_scheduler(task: asyncio.Task[None]) -> None:
             await task
 
 
+async def _run_kit_scheduler() -> None:
+    """Kit-путь (#2192): собрать SchedulerContext + registry и крутить kit scheduler_loop.
+
+    Инжектируем продуктовые адаптеры (RealScraperConfig/Matcher/Enrichment/SessionFactory) +
+    боевой runs-модуль (app.services.scrape_runs) + текущий shutdown_requested callable, чтобы
+    kit-loop делил ту же concurrency/lifecycle-семантику с боевым scheduler'ом. Продуктовые
+    НЕ-sweep source'ы приходят из build_product_handlers; kit-native sweeps — из build_registry.
+
+    SIGTERM-drain: kit scheduler_loop сам следит за ctx.shutdown_requested() и дренит
+    in-flight задачи (ctx.drain_inflight) — та же кооперативная семантика, что у боевого.
+    """
+    from scraper_kit.orchestration.scheduler import (
+        SchedulerContext,
+        build_registry,
+    )
+    from scraper_kit.orchestration.scheduler import (
+        scheduler_loop as kit_scheduler_loop,
+    )
+
+    from app.services import scrape_runs as runs_mod
+    from app.services.product_handlers import build_product_handlers
+    from app.services.scraper_adapters import (
+        RealEnrichmentJobs,
+        RealMatcherAdapter,
+        RealScraperConfig,
+        RealSessionFactory,
+    )
+
+    ctx = SchedulerContext(
+        config=RealScraperConfig(),
+        matcher=RealMatcherAdapter(),
+        enrichment=RealEnrichmentJobs(),
+        session_factory=RealSessionFactory(),
+        shutdown_requested=shutdown_requested,
+        runs=runs_mod,
+    )
+    registry = build_registry(product_handlers=build_product_handlers(ctx))
+    logger.info("scheduler_main: kit-scheduler registry built (%d source handlers)", len(registry))
+    await kit_scheduler_loop(ctx, registry)
+
+
 async def _run() -> None:
     """Async entrypoint: scheduler_loop() с кооперативным SIGTERM/SIGINT-drain'ом.
 
@@ -119,10 +160,17 @@ async def _run() -> None:
     докоммитит текущую карточку и выйдет сам на ближайшем between-unit checkpoint'е
     (scheduler_loop / avito_detail_backfill / import_rosreestr_dkp). Safety-net в
     _await_scheduler гарантирует выход в пределах docker grace.
-    """
-    from app.services.scheduler import scheduler_loop
 
-    task = asyncio.create_task(scheduler_loop())
+    Развилка #2192 (ship-dark): USE_KIT_SCHEDULER=false (дефолт) → боевой
+    app.services.scheduler.scheduler_loop без изменений; =true → kit-путь.
+    """
+    if settings.use_kit_scheduler:
+        logger.info("scheduler_main: USE_KIT_SCHEDULER=true — running kit-scheduler path")
+        task = asyncio.create_task(_run_kit_scheduler())
+    else:
+        from app.services.scheduler import scheduler_loop
+
+        task = asyncio.create_task(scheduler_loop())
 
     loop = asyncio.get_running_loop()
 

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

@@ -0,0 +1,372 @@
+"""Продуктовые Handler'ы для kit-scheduler (#2192, strangler-миграция).
+
+`build_product_handlers(ctx)` собирает реестр НЕ-sweep source'ов, тело которых
+осталось в `app` (rosreestr_dkp / sber_index / deactivate_stale_* / *_backfill /
+cadastral_geo_match / house_* / proxy_healthcheck / …). Kit-native sweep-оркестраторы
+(avito/yandex/cian/domclick city/full-load) регистрируются самим `build_registry`
+через `_default_kit_handlers` — здесь их НЕТ.
+
+Дизайн-инвариант: НЕ дублируем логику. Каждый `job` переиспользует существующую боевую
+джоб-функцию (`app.services.scheduler.import_rosreestr_dkp`, `app.tasks.*`,
+`app.services.*`) — то же тело, что крутит боевой `trigger_*_run`. Kit `_dispatch` уже
+делает claim + fresh-session + spawn + close, поэтому `job` содержит ТОЛЬКО работу
+(без claim/spawn-boilerplate). Run-lifecycle (mark_done/mark_failed/heartbeat) для
+джоб, что не владеют им сами, делается через инжектированный `ctx.runs`.
+
+Ship-dark: модуль импортируется лишь когда `settings.use_kit_scheduler` True
+(scheduler_main._run_kit_scheduler). Боевой `app.services.scheduler` не тронут.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import TYPE_CHECKING, Any
+
+from scraper_kit.orchestration.scheduler import (
+    Handler,
+    reschedule_after_minutes,
+)
+from scraper_kit.orchestration.scheduler import (
+    _defer_next_run_at as kit_defer_next_run_at,
+)
+
+if TYPE_CHECKING:
+    from scraper_kit.orchestration.scheduler import SchedulerContext
+    from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+
+# ── cian_history_backfill — cookie-gated backfill ────────────────────────────
+async def _cian_pre_claim(db: Session, schedule_row: dict[str, Any], ctx: SchedulerContext) -> bool:
+    """Pre-claim gate: проверить наличие/валидность cian-cookies ДО claim (#1522).
+
+    Cookies отсутствуют/протухли → defer next_run_at на следующее окно и skip
+    (иначе get_due_schedules переотбирает schedule каждые 60с и verify_session
+    долбит Cian круглосуточно). Дословно из боевого trigger_cian_backfill_run.
+    """
+    import sentry_sdk
+
+    from app.services.cian_session import load_session, verify_session
+
+    cookies = load_session(db)
+    if cookies is None:
+        logger.warning("scheduler: cian_history_backfill skipped — no valid session cookies in DB")
+        kit_defer_next_run_at(db, schedule_row)
+        return False
+
+    state = await verify_session(cookies)
+    if state is None:
+        logger.warning(
+            "scheduler: cian_history_backfill — cookies expired or invalid, skipping run"
+        )
+        try:
+            sentry_sdk.capture_message(
+                "cian_history_backfill skipped: Cian session cookies expired — "
+                "please re-upload via admin UI",
+                level="warning",
+            )
+        except Exception:
+            pass  # sentry_sdk not initialised in dev
+        kit_defer_next_run_at(db, schedule_row)
+        return False
+
+    return True
+
+
+async def _job_cian_history_backfill(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.services.scheduler import _execute_cian_backfill
+
+    await _execute_cian_backfill(db, run_id=run_id, params=params)
+
+
+# ── rosreestr_dkp_import — sync FDW-импорт в executor ─────────────────────────
+async def _job_rosreestr_dkp(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.services.scheduler import import_rosreestr_dkp
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, import_rosreestr_dkp, db, run_id, params)
+
+
+# ── listing_source_snapshot — sync DB-snapshot в executor ────────────────────
+async def _job_listing_source_snapshot(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.listing_source_snapshot import snapshot_listing_sources
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, snapshot_listing_sources, db, run_id)
+
+
+# ── asking_to_sold_ratio_refresh — sync re-derive в executor ─────────────────
+async def _job_asking_to_sold_ratio(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.asking_to_sold_ratio import recompute_asking_to_sold_ratios
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, recompute_asking_to_sold_ratios, db, run_id)
+
+
+# ── refresh_search_matview — REFRESH MATVIEW CONCURRENTLY (own connection) ────
+async def _job_refresh_search_matview(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.refresh_search_matview import refresh_search_matview
+
+    loop = asyncio.get_event_loop()
+    try:
+        # refresh_search_matview() держит собственное autocommit-соединение
+        # (REFRESH CONCURRENTLY нельзя внутри транзакции).
+        await loop.run_in_executor(None, refresh_search_matview)
+        ctx.runs.mark_done(db, run_id, {})
+    except Exception:
+        logger.exception("scheduler: refresh_search_matview crashed run_id=%d", run_id)
+        ctx.runs.mark_failed(db, run_id, "refresh_search_matview failed", {})
+
+
+# ── yandex_address_backfill — async, owns lifecycle ──────────────────────────
+async def _job_yandex_address_backfill(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.yandex_address_backfill import run_yandex_address_backfill
+
+    await run_yandex_address_backfill(db, run_id=run_id, params=params)
+
+
+# ── deactivate_stale_* — sync UPDATE в executor (wildcard-семейство) ──────────
+async def _job_deactivate_stale(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.core.config import settings as _settings
+    from app.tasks.deactivate_stale_avito import deactivate_stale_listings
+
+    listing_source: str = params.get("listing_source", "avito")
+    ttl_days: int = params.get("ttl_days", _settings.avito_stale_ttl_days)
+    segments: list[str] | None = params.get("segments")
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(
+        None,
+        lambda: deactivate_stale_listings(
+            db,
+            run_id,
+            listing_source=listing_source,
+            ttl_days=ttl_days,
+            segments=segments,
+        ),
+    )
+
+
+# ── sber_index_pull — async, owns lifecycle ──────────────────────────────────
+async def _job_sber_index_pull(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.sber_index_pull import run_sber_index_pull
+
+    await run_sber_index_pull(db, run_id=run_id, params=params)
+
+
+# ── rosreestr_quarter_poll — async, owns lifecycle ───────────────────────────
+async def _job_rosreestr_quarter_poll(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.rosreestr_quarter_poll import run_rosreestr_quarter_poll
+
+    await run_rosreestr_quarter_poll(db, run_id=run_id, params=params)
+
+
+# ── newbuilding_enrich — async, owns lifecycle ───────────────────────────────
+async def _job_newbuilding_enrich(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.newbuilding_enrich_backfill import run_newbuilding_enrich
+
+    await run_newbuilding_enrich(db, run_id=run_id, params=params)
+
+
+# ── yandex_newbuilding_sweep — async, lifecycle в job ────────────────────────
+async def _job_yandex_newbuilding_sweep(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.yandex_newbuilding_sweep import enrich_yandex_newbuilding_sweep
+
+    try:
+        result = await enrich_yandex_newbuilding_sweep(
+            db,
+            limit=int(params.get("limit", 5)),
+            request_delay_sec=float(params.get("request_delay_sec", 8.0)),
+            city=str(params.get("city", "ekaterinburg")),
+        )
+        ctx.runs.mark_done(db, run_id, result.to_dict())
+    except Exception:
+        logger.exception("scheduler: enrich_yandex_newbuilding_sweep crashed run_id=%d", run_id)
+        try:
+            ctx.runs.mark_failed(db, run_id, "crashed", {})
+        except Exception:
+            pass
+
+
+# ── geocode_missing_listings — async, owns lifecycle ─────────────────────────
+async def _job_geocode_missing_listings(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.geocode_missing import run_geocode_missing_listings
+
+    await run_geocode_missing_listings(db, run_id=run_id, params=params)
+
+
+# ── avito_detail_backfill — async, owns lifecycle ────────────────────────────
+async def _job_avito_detail_backfill(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.avito_detail_backfill import run_avito_detail_backfill
+
+    await run_avito_detail_backfill(db, run_id=run_id, params=params)
+
+
+# ── yandex_detail_backfill — async, owns lifecycle ───────────────────────────
+async def _job_yandex_detail_backfill(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.yandex_detail_backfill import run_yandex_detail_backfill
+
+    await run_yandex_detail_backfill(db, run_id=run_id, params=params)
+
+
+# ── cadastral_geo_match — sync FDW-refresh+match в executor ───────────────────
+async def _job_cadastral_geo_match(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.cadastral_geo_match import run_cadastral_geo_match
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(
+        None,
+        lambda: run_cadastral_geo_match(db, run_id=run_id, params=params),
+    )
+
+
+# ── house_imv_backfill — async Avito-IMV с heartbeat, lifecycle в job ─────────
+async def _job_house_imv_backfill(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    batch_size = int(params.get("batch_size", 50))
+    request_delay_sec = float(params.get("request_delay_sec", 5.0))
+    only_status = str(params.get("only_status", "pending"))
+
+    def _heartbeat() -> None:
+        # heartbeat каждые N домов — иначе reap_zombies снимет живой долгий IMV-run (#1363).
+        ctx.runs.update_heartbeat(db, run_id, {})
+
+    try:
+        result = await ctx.enrichment.house_imv_backfill(
+            db,
+            batch_size=batch_size,
+            request_delay_sec=request_delay_sec,
+            only_status=only_status,
+            heartbeat=_heartbeat,
+        )
+        ctx.runs.mark_done(
+            db,
+            run_id,
+            {
+                "checked": result.checked,
+                "saved": result.saved,
+                "skipped": result.skipped,
+                "errors": result.errors,
+                "duration_sec": int(result.duration_sec),
+            },
+        )
+    except Exception as exc:
+        logger.exception("scheduler: house_imv_backfill crashed run_id=%d", run_id)
+        try:
+            ctx.runs.mark_failed(db, run_id, str(exc)[:1000], {})
+        except Exception:
+            logger.exception("scheduler: mark_failed crashed run_id=%d", run_id)
+
+
+# ── house_dedup_merge — sync destructive merge в executor, owns lifecycle ─────
+async def _job_house_dedup_merge(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.services.house_dedup_merge import run_house_dedup_merge
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(
+        None,
+        lambda: run_house_dedup_merge(db, run_id=run_id, params=params),
+    )
+
+
+# ── proxy_healthcheck — async, sub-hourly reschedule + lifecycle в job ────────
+async def _job_proxy_healthcheck(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.services.proxy_pool import run_proxy_healthcheck
+
+    try:
+        counters = await run_proxy_healthcheck(db)
+        ctx.runs.mark_done(db, run_id, counters)
+    except Exception as exc:
+        logger.exception("scheduler: run_proxy_healthcheck crashed run_id=%d", run_id)
+        try:
+            ctx.runs.mark_failed(db, run_id, str(exc)[:1000], {})
+        except Exception:
+            logger.exception("scheduler: mark_failed crashed run_id=%d", run_id)
+
+
+def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
+    """Реестр НЕ-sweep продуктовых source→Handler для kit build_registry.
+
+    Kit-native sweeps (avito/yandex/cian/domclick city/full-load/newbuilding) НЕ здесь —
+    их даёт build_registry(_default_kit_handlers). Здесь — 17 именованных + 1 wildcard
+    (deactivate_stale_*), покрывающие каждый НЕ-sweep source боевого scheduler-dispatch.
+
+    `ctx` — принят для симметрии контракта; сами Handler-job'ы получают ctx во время
+    dispatch (см. kit `_dispatch`), поэтому здесь он не замыкается.
+    """
+    return {
+        "cian_history_backfill": Handler(
+            _job_cian_history_backfill,
+            "cian_history_backfill",
+            pre_claim=_cian_pre_claim,
+        ),
+        "rosreestr_dkp_import": Handler(_job_rosreestr_dkp, "rosreestr_dkp_import"),
+        "listing_source_snapshot": Handler(_job_listing_source_snapshot, "listing_source_snapshot"),
+        "asking_to_sold_ratio_refresh": Handler(
+            _job_asking_to_sold_ratio, "asking_to_sold_ratio_refresh"
+        ),
+        "refresh_search_matview": Handler(_job_refresh_search_matview, "refresh_search_matview"),
+        "yandex_address_backfill": Handler(_job_yandex_address_backfill, "yandex_address_backfill"),
+        "sber_index_pull": Handler(_job_sber_index_pull, "sber_index_pull"),
+        "rosreestr_quarter_poll": Handler(_job_rosreestr_quarter_poll, "rosreestr_quarter_poll"),
+        "newbuilding_enrich": Handler(_job_newbuilding_enrich, "newbuilding_enrich"),
+        "yandex_newbuilding_sweep": Handler(
+            _job_yandex_newbuilding_sweep, "yandex_newbuilding_sweep"
+        ),
+        "geocode_missing_listings": Handler(
+            _job_geocode_missing_listings, "geocode_missing_listings"
+        ),
+        "avito_detail_backfill": Handler(_job_avito_detail_backfill, "avito_detail_backfill"),
+        "yandex_detail_backfill": Handler(_job_yandex_detail_backfill, "yandex_detail_backfill"),
+        "cadastral_geo_match": Handler(_job_cadastral_geo_match, "cadastral_geo_match"),
+        "house_imv_backfill": Handler(_job_house_imv_backfill, "house_imv_backfill"),
+        "house_dedup_merge": Handler(_job_house_dedup_merge, "house_dedup_merge"),
+        "proxy_healthcheck": Handler(
+            _job_proxy_healthcheck,
+            "proxy_healthcheck",
+            post_claim=reschedule_after_minutes(param="interval_minutes", default=30),
+        ),
+        # Wildcard: deactivate_stale_avito / _yandex / _cian — один Handler на семейство
+        # (resolve_handler матчит по префиксу key[:-1]="deactivate_stale_").
+        "deactivate_stale_*": Handler(_job_deactivate_stale, "deactivate_stale"),
+    }
+
+
+__all__ = ["build_product_handlers"]

tradein-mvp/backend/tests/test_kit_registry_completeness.py

@@ -0,0 +1,87 @@
+"""Полнота kit-registry vs боевой scheduler-dispatch (#2192).
+
+Гарантирует ship-dark инвариант: КАЖДЫЙ source, который обрабатывает боевой
+`app.services.scheduler.scheduler_loop` (if/elif + startswith), резолвится в
+kit-registry (`build_registry(product_handlers=build_product_handlers(ctx))`).
+Тест ПАДАЕТ, если source добавили в боевой dispatch, но забыли в registry —
+иначе после переключения USE_KIT_SCHEDULER=true такой source молча не запустится.
+
+Источник истины — сам файл боевого scheduler'а: парсим `source == "X"` (exact)
+и `source.startswith("X")` (wildcard), чтобы при дрейфе dispatch тест ловил
+рассинхрон автоматически, а не по захардкоженному списку.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from pathlib import Path
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from scraper_kit.orchestration.scheduler import (
+    SchedulerContext,
+    build_registry,
+    resolve_handler,
+)
+
+import app.services.scheduler as _sched_mod
+from app.services.product_handlers import build_product_handlers
+from app.services.scraper_adapters import (
+    RealEnrichmentJobs,
+    RealMatcherAdapter,
+    RealScraperConfig,
+    RealSessionFactory,
+)
+
+
+def _production_sources() -> tuple[set[str], set[str]]:
+    """(exact, prefixes) source'ов из боевого scheduler_loop dispatch."""
+    src = Path(_sched_mod.__file__).read_text(encoding="utf-8")
+    exact = set(re.findall(r'source == "([^"]+)"', src))
+    prefixes = set(re.findall(r'source\.startswith\("([^"]+)"\)', src))
+    return exact, prefixes
+
+
+def _build_registry() -> dict:
+    ctx = SchedulerContext(
+        config=RealScraperConfig(),
+        matcher=RealMatcherAdapter(),
+        enrichment=RealEnrichmentJobs(),
+        session_factory=RealSessionFactory(),
+    )
+    return build_registry(product_handlers=build_product_handlers(ctx))
+
+
+def test_production_dispatch_parse_sane() -> None:
+    """Sanity: парсер видит непустой набор source'ов (защита от regex-дрейфа)."""
+    exact, prefixes = _production_sources()
+    # На момент #2192 боевой dispatch держит 25 exact + 1 wildcard-семейство.
+    assert len(exact) >= 25, f"parse regressed — got {len(exact)} exact sources"
+    assert "deactivate_stale_" in prefixes
+
+
+def test_every_production_source_resolves_in_kit_registry() -> None:
+    """Каждый боевой exact-source резолвится в kit-registry."""
+    registry = _build_registry()
+    exact, _ = _production_sources()
+    missing = [s for s in sorted(exact) if resolve_handler(s, registry) is None]
+    assert not missing, f"sources missing from kit registry: {missing}"
+
+
+def test_wildcard_family_resolves() -> None:
+    """Каждый deactivate_stale_* член семейства резолвится через wildcard-Handler."""
+    registry = _build_registry()
+    _, prefixes = _production_sources()
+    for prefix in sorted(prefixes):
+        for member in ("avito", "yandex", "cian"):
+            source = f"{prefix}{member}"
+            assert (
+                resolve_handler(source, registry) is not None
+            ), f"wildcard source {source} did not resolve"
+
+
+def test_unknown_source_does_not_resolve() -> None:
+    """Негативный контроль: несуществующий source → None (тест реально дискриминирует)."""
+    registry = _build_registry()
+    assert resolve_handler("totally_unknown_source_xyz", registry) is None