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

"""Продуктовые 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")
    staleness_column: str = params.get("staleness_column", "last_seen_at")

    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,
            staleness_column=staleness_column,
        ),
    )


# ── 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)


# ── deals_freshness_monitor — sync DB-only freshness check в executor ─────────
async def _job_deals_freshness_monitor(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    from app.tasks.deals_freshness_monitor import check_deals_freshness

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, check_deals_freshness, db, run_id, 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),
    )


# ── osm_poi_ekb_refresh — sync FDW-refresh в executor ─────────────────────────
async def _job_osm_poi_ekb_refresh(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    from app.tasks.osm_poi_ekb_refresh import run_osm_poi_ekb_refresh

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(
        None,
        lambda: run_osm_poi_ekb_refresh(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"),
        "deals_freshness_monitor": Handler(_job_deals_freshness_monitor, "deals_freshness_monitor"),
        "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"),
        "osm_poi_ekb_refresh": Handler(_job_osm_poi_ekb_refresh, "osm_poi_ekb_refresh"),
        "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"]