From bcdec5ebd4db8212c10515614942f2e549e4fb11 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Thu, 2 Jul 2026 18:41:12 +0000 Subject: [PATCH] =?UTF-8?q?feat(rewire):=20scheduler=5Fmain=20=E2=86=92=20?= =?UTF-8?q?kit-scheduler=20behind=20USE=5FKIT=5FSCHEDULER=20flag,=20ship-d?= =?UTF-8?q?ark=20(#2192)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/core/config.py | 10 + tradein-mvp/backend/app/scheduler_main.py | 54 ++- .../backend/app/services/product_handlers.py | 372 ++++++++++++++++++ .../tests/test_kit_registry_completeness.py | 87 ++++ 4 files changed, 520 insertions(+), 3 deletions(-) create mode 100644 tradein-mvp/backend/app/services/product_handlers.py create mode 100644 tradein-mvp/backend/tests/test_kit_registry_completeness.py diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index a7d3e79d..1486ef59 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/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" diff --git a/tradein-mvp/backend/app/scheduler_main.py b/tradein-mvp/backend/app/scheduler_main.py index 51f356cc..a0cf5571 100644 --- a/tradein-mvp/backend/app/scheduler_main.py +++ b/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() diff --git a/tradein-mvp/backend/app/services/product_handlers.py b/tradein-mvp/backend/app/services/product_handlers.py new file mode 100644 index 00000000..fb38506d --- /dev/null +++ b/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"] diff --git a/tradein-mvp/backend/tests/test_kit_registry_completeness.py b/tradein-mvp/backend/tests/test_kit_registry_completeness.py new file mode 100644 index 00000000..fef35c0d --- /dev/null +++ b/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