694 lines
31 KiB
Python
694 lines
31 KiB
Python
"""In-app scheduler — strangler-копия `app.services.scheduler` в scraper_kit (#2136).
|
||
|
||
Разница с боевым `app.services.scheduler`:
|
||
|
||
1. Полная развязка от `app.*` — все продуктовые зависимости инжектируются через
|
||
`SchedulerContext` (config/matcher/enrichment/session_factory/runs/shutdown),
|
||
описанные Protocol'ами в `scraper_kit.contracts`. `grep "from app"` — пусто.
|
||
|
||
2. **Registry-dispatch вместо copy-paste.** Боевой модуль держит 25+ почти
|
||
одинаковых `trigger_*_run` (claim → new-session → create_task → close) плюс
|
||
27-веточный `if/elif` в `scheduler_loop`. Здесь — единственный параметризованный
|
||
путь `_dispatch()` + `HANDLERS: dict[str, Handler]`. Каждый source — одна запись
|
||
реестра; общий claim/session/drain-boilerplate вынесен в `_dispatch`.
|
||
|
||
Kit-native (sweep-оркестраторы, уже перенесённые в `scraper_kit.orchestration.pipeline`)
|
||
регистрируются встроенно (`_default_kit_handlers`). Продуктовые джобы, тело которых
|
||
осталось в `app` (rosreestr_dkp / sber_index / deactivate_stale / *_backfill / …),
|
||
инжектируются извне как `Handler` через `build_registry(product_handlers=...)`.
|
||
|
||
Боевой рантайм (scraper-контейнер) по-прежнему крутит старый `app.services.scheduler` —
|
||
это COPY, не MOVE. Переключение — отдельный поздний strangler-шаг.
|
||
|
||
Критичная concurrency-логика (`_claim_run` advisory-lock + double-check, `reap_zombies`
|
||
порог, heartbeat, SIGTERM-drain) перенесена ДОСЛОВНО — тот же SQL, то же ветвление.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import random
|
||
from collections.abc import Callable, Coroutine, Mapping
|
||
from dataclasses import dataclass, field
|
||
from datetime import UTC, datetime, time, timedelta
|
||
from typing import TYPE_CHECKING, Any
|
||
|
||
from sqlalchemy import text
|
||
|
||
from scraper_kit.orchestration import runs as _kit_runs
|
||
from scraper_kit.orchestration.pipeline import (
|
||
run_avito_city_sweep,
|
||
run_avito_full_load,
|
||
run_avito_newbuilding_sweep,
|
||
run_cian_city_sweep,
|
||
run_cian_full_load,
|
||
run_domclick_city_sweep,
|
||
run_yandex_city_sweep,
|
||
)
|
||
|
||
if TYPE_CHECKING:
|
||
from sqlalchemy.orm import Session
|
||
|
||
from scraper_kit.contracts import (
|
||
EnrichmentJobs,
|
||
HouseMatcher,
|
||
ScraperConfig,
|
||
SessionFactory,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Loop interval — check каждую минуту
|
||
SCHEDULER_TICK_SEC = 60
|
||
ZOMBIE_THRESHOLD_HOURS = 6
|
||
|
||
# Бюджет дренажа детей < scheduler_main._DRAIN_TIMEOUT_S (100s) < docker grace (120s)
|
||
# (см. #1182 P2 — идентично боевому scheduler'у).
|
||
_CHILD_DRAIN_TIMEOUT_S = 80.0
|
||
|
||
# ── типы job/handler ─────────────────────────────────────────────────────────
|
||
# Job получает свежую сессию (открыта `_dispatch`), run_id, params и весь контекст
|
||
# (config/matcher/enrichment/runs) — чтобы иметь доступ к инжектированным зависимостям.
|
||
Job = Callable[["Session", int, dict[str, Any], "SchedulerContext"], Coroutine[Any, Any, None]]
|
||
# Pre-claim gate: вернуть False → пропустить claim (напр. cian: нет/протухли cookies).
|
||
PreClaim = Callable[["Session", dict[str, Any], "SchedulerContext"], Coroutine[Any, Any, bool]]
|
||
# Post-claim hook: сразу после успешного claim (напр. proxy_healthcheck sub-hourly reschedule).
|
||
PostClaim = Callable[["Session", int, dict[str, Any], "SchedulerContext"], None]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Handler:
|
||
"""Одна запись реестра source→обработчик.
|
||
|
||
Инкапсулирует всё, что раньше размазывалось по индивидуальному `trigger_*_run`:
|
||
- `job` — асинхронное тело (SERP/backfill/enrichment). Общий boilerplate
|
||
(claim, свежая сессия, create_task, close, log) делает `_dispatch`.
|
||
- `pre_claim` — опциональный гейт ДО claim (cookie-проверка cian → defer).
|
||
- `post_claim` — опциональный хук сразу ПОСЛЕ claim (sub-hourly reschedule).
|
||
- `log_name` — имя источника в логах ("scheduler: triggered <log_name> run_id=…").
|
||
"""
|
||
|
||
job: Job
|
||
log_name: str
|
||
pre_claim: PreClaim | None = None
|
||
post_claim: PostClaim | None = None
|
||
|
||
|
||
@dataclass
|
||
class SchedulerContext:
|
||
"""Инжектируемые зависимости + рантайм-состояние планировщика.
|
||
|
||
Заменяет прямые импорты `app.core.config.settings` / `app.services.matching` /
|
||
`app.services.house_imv_backfill` / `app.core.db.SessionLocal` /
|
||
`app.services.scrape_runs` / `app.core.shutdown.shutdown_requested`.
|
||
"""
|
||
|
||
config: ScraperConfig
|
||
matcher: HouseMatcher
|
||
enrichment: EnrichmentJobs
|
||
session_factory: SessionFactory
|
||
shutdown_requested: Callable[[], bool] = lambda: False
|
||
# runs-модуль (create_run/update_heartbeat/mark_done/… ). По умолчанию — kit-копия
|
||
# scraper_kit.orchestration.runs; в тестах подменяется recorder'ом.
|
||
runs: Any = _kit_runs
|
||
# #1182 P2: strong-ref'ы detached run-задач для graceful SIGTERM-drain'а.
|
||
_inflight_tasks: set[asyncio.Task[None]] = field(default_factory=set)
|
||
|
||
def spawn_tracked(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]:
|
||
"""create_task + регистрация в _inflight_tasks для graceful-drain'а (#1182 P2).
|
||
|
||
strong-ref в set'е держит задачу до завершения (RUF006) и даёт scheduler_loop'у
|
||
дождаться её на SIGTERM-drain'е; done-callback ретривит exception и убирает задачу.
|
||
"""
|
||
task = asyncio.create_task(coro)
|
||
self._inflight_tasks.add(task)
|
||
|
||
def _on_done(t: asyncio.Task[None]) -> None:
|
||
self._inflight_tasks.discard(t)
|
||
if not t.cancelled():
|
||
t.exception()
|
||
|
||
task.add_done_callback(_on_done)
|
||
return task
|
||
|
||
async def drain_inflight(self) -> None:
|
||
"""Дождаться завершения detached run-задач перед teardown'ом (#1182 P2).
|
||
|
||
Кооперативные дети дочекивают текущий unit + mark_done и резолвятся;
|
||
некооперативные упираются в _CHILD_DRAIN_TIMEOUT_S и остаются на внешний
|
||
hard-cancel fallback. Не busy-spin — один asyncio.wait.
|
||
"""
|
||
pending = [t for t in self._inflight_tasks if not t.done()]
|
||
if not pending:
|
||
return
|
||
logger.info("scheduler: draining %d in-flight run task(s) on shutdown", len(pending))
|
||
_done, still = await asyncio.wait(pending, timeout=_CHILD_DRAIN_TIMEOUT_S)
|
||
if still:
|
||
logger.warning(
|
||
"scheduler: %d task(s) did not drain in %.0fs — leaving for hard-cancel",
|
||
len(still),
|
||
_CHILD_DRAIN_TIMEOUT_S,
|
||
)
|
||
else:
|
||
logger.info("scheduler: all in-flight run task(s) drained cleanly")
|
||
|
||
|
||
def compute_next_run_at(
|
||
window_start_hour: int,
|
||
window_end_hour: int,
|
||
*,
|
||
now: datetime | None = None,
|
||
interval_days: int = 1,
|
||
) -> datetime:
|
||
"""Pick random datetime в window [start, end) UTC, через interval_days суток после now.
|
||
|
||
interval_days задаёт каденс источника: 1 (default) = daily (back-compat), 7 = weekly.
|
||
Берётся из schedule.default_params["interval_days"] вызывающим кодом (_claim_run /
|
||
_defer_next_run_at); отсутствие ключа → 1 → прежнее ежедневное поведение.
|
||
|
||
Если window_end_hour <= window_start_hour → cross-midnight window
|
||
(например 22→3 → окно 22:00-23:59 ИЛИ 00:00-02:59).
|
||
"""
|
||
now = now or datetime.now(tz=UTC)
|
||
interval_days = max(1, int(interval_days))
|
||
# Целевая дата = now + interval_days суток (interval_days=1 → завтра, как раньше).
|
||
target = (now + timedelta(days=interval_days)).date()
|
||
|
||
if window_end_hour > window_start_hour:
|
||
# Обычное окно (например 2..5 → 02:00-04:59)
|
||
start_seconds = window_start_hour * 3600
|
||
end_seconds = window_end_hour * 3600
|
||
rand_seconds = random.randint(start_seconds, end_seconds - 1)
|
||
return datetime.combine(target, time(0, 0), tzinfo=UTC) + timedelta(seconds=rand_seconds)
|
||
else:
|
||
# Cross-midnight (22..3 → 22:00-23:59 + 00:00-02:59)
|
||
# Длина окна = (24-start) + end часов
|
||
total_seconds = ((24 - window_start_hour) + window_end_hour) * 3600
|
||
rand_seconds = random.randint(0, total_seconds - 1)
|
||
# Если rand попадает в первую часть (start..24)
|
||
first_half = (24 - window_start_hour) * 3600
|
||
if rand_seconds < first_half:
|
||
# interval_days=1: текущая дата (если окно ещё не наступило сегодня) или next day.
|
||
# interval_days>1: всегда целевая дата (стаггер на N суток вперёд).
|
||
today_ok = interval_days == 1 and now.hour < window_start_hour
|
||
base_date = now.date() if today_ok else target
|
||
return datetime.combine(base_date, time(0, 0), tzinfo=UTC) + timedelta(
|
||
seconds=window_start_hour * 3600 + rand_seconds
|
||
)
|
||
else:
|
||
# Во второй части (0..end), целевого дня
|
||
offset = rand_seconds - first_half
|
||
return datetime.combine(target, time(0, 0), tzinfo=UTC) + timedelta(seconds=offset)
|
||
|
||
|
||
def has_running_run(db: Session, source: str) -> bool:
|
||
"""Есть ли активный run для source (status='running')."""
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
SELECT 1 FROM scrape_runs
|
||
WHERE source = :source AND status = 'running'
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"source": source},
|
||
).fetchone()
|
||
return row is not None
|
||
|
||
|
||
def reap_zombies(db: Session) -> int:
|
||
"""Mark scrape_runs as 'zombie' если heartbeat не обновлялся > ZOMBIE_THRESHOLD_HOURS hours."""
|
||
zombie_interval = f"{ZOMBIE_THRESHOLD_HOURS} hours"
|
||
result = db.execute(
|
||
text(
|
||
"""
|
||
UPDATE scrape_runs
|
||
SET status = 'zombie', finished_at = NOW()
|
||
WHERE status = 'running'
|
||
AND (heartbeat_at IS NULL
|
||
OR heartbeat_at < NOW() - CAST(:interval AS interval))
|
||
RETURNING id
|
||
"""
|
||
),
|
||
{"interval": zombie_interval},
|
||
)
|
||
rows = result.fetchall()
|
||
db.commit()
|
||
if rows:
|
||
logger.warning("scheduler: reaped %d zombie runs: %s", len(rows), [r.id for r in rows])
|
||
return len(rows)
|
||
|
||
|
||
def _claim_run(db: Session, schedule_row: dict[str, Any], ctx: SchedulerContext) -> int | None:
|
||
"""Claim run: INSERT scrape_runs + UPDATE scrape_schedules next_run_at.
|
||
|
||
Returns run_id или None если уже есть running run для этого source.
|
||
Общий helper для всех source-handler'ов. Advisory-lock concurrency-логика перенесена
|
||
ДОСЛОВНО из боевого scheduler'а (создание run — через инжектированный ctx.runs).
|
||
"""
|
||
source = schedule_row["source"]
|
||
if has_running_run(db, source):
|
||
logger.info("scheduler: skip — already running for source=%s", source)
|
||
return None
|
||
|
||
# #750: атомарный claim через transaction-scoped advisory lock. Сериализует
|
||
# конкурентные _claim_run одного source (redeploy-overlap старого+нового контейнера
|
||
# / 2 реплики), иначе оба пройдут has_running_run-pre-check и запустят ДВОЙНОЙ sweep
|
||
# → 2× rate к Avito/Cian → бан. pg_try_advisory_xact_lock НЕ блокирует: lock занят
|
||
# другим тиком → false → return None. Лок авто-снимается на commit/rollback этой
|
||
# транзакции (has_running_run выше — быстрый pre-check, обычно без взятия лока).
|
||
got_lock = db.execute(
|
||
text("SELECT pg_try_advisory_xact_lock(hashtext(:source))"),
|
||
{"source": source},
|
||
).scalar()
|
||
if not got_lock:
|
||
logger.info("scheduler: skip — concurrent claim in progress for source=%s", source)
|
||
return None
|
||
|
||
# Double-checked под локом: конкурентный тик мог закоммитить running-run МЕЖДУ
|
||
# pre-check и взятием лока (затем освободить lock на commit). Перепроверяем под
|
||
# локом — иначе словим двойной run на узком окне.
|
||
if has_running_run(db, source):
|
||
logger.info("scheduler: skip — running appeared under lock for source=%s", source)
|
||
db.rollback() # освобождаем advisory lock (claim не состоялся)
|
||
return None
|
||
|
||
params = schedule_row.get("default_params") or {}
|
||
run_id = ctx.runs.create_run(db, source=source, params=params)
|
||
|
||
next_at = compute_next_run_at(
|
||
schedule_row["window_start_hour"],
|
||
schedule_row["window_end_hour"],
|
||
interval_days=int(params.get("interval_days", 1)),
|
||
)
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE scrape_schedules
|
||
SET last_run_id = :run_id, last_run_at = NOW(),
|
||
next_run_at = :next_at, updated_at = NOW()
|
||
WHERE source = :source
|
||
"""
|
||
),
|
||
{"run_id": run_id, "next_at": next_at, "source": source},
|
||
)
|
||
db.commit()
|
||
logger.info(
|
||
"scheduler: claimed run_id=%d source=%s next_run_at=%s",
|
||
run_id,
|
||
source,
|
||
next_at.isoformat(),
|
||
)
|
||
return run_id
|
||
|
||
|
||
def _defer_next_run_at(db: Session, schedule_row: dict[str, Any]) -> None:
|
||
"""Сдвинуть next_run_at на следующее окно БЕЗ создания run (#1522).
|
||
|
||
Используется когда pre_claim делает early-return (например cian: отсутствуют/протухли
|
||
cookies) ДО _claim_run. Без этого get_due_schedules переотбирает schedule на каждом
|
||
тике (SCHEDULER_TICK_SEC), и pre-check гоняется раз в минуту круглосуточно.
|
||
"""
|
||
source = schedule_row["source"]
|
||
params = schedule_row.get("default_params") or {}
|
||
next_at = compute_next_run_at(
|
||
schedule_row["window_start_hour"],
|
||
schedule_row["window_end_hour"],
|
||
interval_days=int(params.get("interval_days", 1)),
|
||
)
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE scrape_schedules
|
||
SET next_run_at = :next_at, updated_at = NOW()
|
||
WHERE source = :source
|
||
"""
|
||
),
|
||
{"next_at": next_at, "source": source},
|
||
)
|
||
db.commit()
|
||
logger.info(
|
||
"scheduler: deferred next_run_at source=%s next_run_at=%s (no run claimed)",
|
||
source,
|
||
next_at.isoformat(),
|
||
)
|
||
|
||
|
||
def reschedule_after_minutes(*, param: str = "interval_minutes", default: int = 30) -> PostClaim:
|
||
"""Фабрика post_claim-хука: sub-hourly re-schedule сразу после claim (#2162).
|
||
|
||
Боевой scheduler держит суточную гранулярность в compute_next_run_at, а
|
||
proxy_healthcheck хочет каденс раз в N минут. Хук СРАЗУ после claim переопределяет
|
||
next_run_at на now() + interval_minutes (source-специфично, не трогая shared
|
||
compute_next_run_at). Если run ещё идёт на следующем тике — has_running_run в
|
||
_claim_run вернёт None (skip) и next_run_at не сбросится → авто-throttle по факту.
|
||
"""
|
||
|
||
def _hook(db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext) -> None:
|
||
# source читаем из ЗАКЛЕЙМЛЕННОГО schedule — берём из params-owner через closure нельзя,
|
||
# поэтому апдейтим по run_id → source (schedule owner). Проще: UPDATE по source из
|
||
# scrape_runs текущего run_id, что эквивалентно WHERE source = <this source>.
|
||
minutes = int(params.get(param, default))
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE scrape_schedules
|
||
SET next_run_at = now() + make_interval(mins => CAST(:mins AS integer)),
|
||
updated_at = now()
|
||
WHERE source = (SELECT source FROM scrape_runs WHERE id = CAST(:rid AS bigint))
|
||
"""
|
||
),
|
||
{"mins": minutes, "rid": run_id},
|
||
)
|
||
db.commit()
|
||
|
||
return _hook
|
||
|
||
|
||
# ── единственный параметризованный dispatch (заменяет 25 trigger_* + 27 if/elif) ──
|
||
|
||
|
||
async def _dispatch(
|
||
handler: Handler,
|
||
db: Session,
|
||
schedule_row: dict[str, Any],
|
||
ctx: SchedulerContext,
|
||
) -> int | None:
|
||
"""Общий путь claim→spawn для ЛЮБОГО source (весь boilerplate прежних trigger_*_run).
|
||
|
||
1. Опциональный pre_claim-гейт (False → skip без claim).
|
||
2. _claim_run (advisory-lock; None → уже running / concurrent claim).
|
||
3. Опциональный post_claim-хук (sub-hourly reschedule).
|
||
4. spawn detached run-задачи: свежая сессия → handler.job → close (+ exception-log).
|
||
"""
|
||
if handler.pre_claim is not None:
|
||
proceed = await handler.pre_claim(db, schedule_row, ctx)
|
||
if not proceed:
|
||
return None
|
||
|
||
run_id = _claim_run(db, schedule_row, ctx)
|
||
if run_id is None:
|
||
return None
|
||
|
||
params = schedule_row.get("default_params") or {}
|
||
|
||
if handler.post_claim is not None:
|
||
handler.post_claim(db, run_id, params, ctx)
|
||
|
||
async def _run() -> None:
|
||
run_db = ctx.session_factory()
|
||
try:
|
||
await handler.job(run_db, run_id, params, ctx)
|
||
except Exception:
|
||
logger.exception("scheduler: %s crashed run_id=%d", handler.log_name, run_id)
|
||
finally:
|
||
run_db.close()
|
||
|
||
ctx.spawn_tracked(_run())
|
||
logger.info("scheduler: triggered %s run_id=%d", handler.log_name, run_id)
|
||
return run_id
|
||
|
||
|
||
def resolve_handler(source: str, registry: Mapping[str, Handler]) -> Handler | None:
|
||
"""Найти handler по source: exact-match, затем wildcard-префикс ("deactivate_stale_*").
|
||
|
||
Единственная точка ветвления вместо 27 if/elif. Wildcard-ключ (кончается на "*")
|
||
матчит по префиксу — покрывает семейство deactivate_stale_avito/yandex/cian одной
|
||
записью реестра (как боевой `source.startswith("deactivate_stale_")`).
|
||
"""
|
||
handler = registry.get(source)
|
||
if handler is not None:
|
||
return handler
|
||
for key, h in registry.items():
|
||
if key.endswith("*") and source.startswith(key[:-1]):
|
||
return h
|
||
return None
|
||
|
||
|
||
# ── kit-native sweep-обработчики (тело в scraper_kit.orchestration.pipeline) ──────
|
||
# Param-чтение идентично боевым trigger_*_run — та же семантика окон/дефолтов.
|
||
|
||
|
||
async def _job_avito_city_sweep(
|
||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
||
) -> None:
|
||
await run_avito_city_sweep(
|
||
db,
|
||
run_id=run_id,
|
||
config=ctx.config,
|
||
matcher=ctx.matcher,
|
||
enrichment=ctx.enrichment,
|
||
shutdown_requested=ctx.shutdown_requested,
|
||
pages_per_anchor=int(params.get("pages_per_anchor", 3)),
|
||
detail_top_n=int(params.get("detail_top_n", 20)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 7.0)),
|
||
enrich_houses=bool(params.get("enrich_houses", True)),
|
||
radius_m=int(params.get("radius_m", 1500)),
|
||
)
|
||
|
||
|
||
async def _job_avito_newbuilding_sweep(
|
||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
||
) -> None:
|
||
await run_avito_newbuilding_sweep(
|
||
db,
|
||
run_id=run_id,
|
||
config=ctx.config,
|
||
matcher=ctx.matcher,
|
||
shutdown_requested=ctx.shutdown_requested,
|
||
pages=int(params.get("pages", 20)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 7.0)),
|
||
)
|
||
|
||
|
||
async def _job_avito_full_load(
|
||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
||
) -> None:
|
||
_incremental_days = params.get("incremental_days")
|
||
incremental_days = int(_incremental_days) if _incremental_days is not None else None
|
||
await run_avito_full_load(
|
||
db,
|
||
run_id=run_id,
|
||
config=ctx.config,
|
||
matcher=ctx.matcher,
|
||
shutdown_requested=ctx.shutdown_requested,
|
||
price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
|
||
concurrency=int(params.get("concurrency", 5)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 7.0)),
|
||
secondary_only=bool(params.get("secondary_only", True)),
|
||
resume_run_id=None,
|
||
incremental_days=incremental_days,
|
||
)
|
||
|
||
|
||
async def _job_avito_full_load_exhaustive(
|
||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
||
) -> None:
|
||
# incremental_days форсится в None → полный exhaustive обход (weekly cadence).
|
||
await run_avito_full_load(
|
||
db,
|
||
run_id=run_id,
|
||
config=ctx.config,
|
||
matcher=ctx.matcher,
|
||
shutdown_requested=ctx.shutdown_requested,
|
||
price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
|
||
concurrency=int(params.get("concurrency", 5)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 7.0)),
|
||
secondary_only=bool(params.get("secondary_only", True)),
|
||
resume_run_id=None,
|
||
incremental_days=None,
|
||
)
|
||
|
||
|
||
async def _job_yandex_city_sweep(
|
||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
||
) -> None:
|
||
await run_yandex_city_sweep(
|
||
db,
|
||
run_id=run_id,
|
||
config=ctx.config,
|
||
matcher=ctx.matcher,
|
||
enrichment=ctx.enrichment,
|
||
shutdown_requested=ctx.shutdown_requested,
|
||
pages_per_anchor=int(params.get("pages_per_anchor", 2)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 9.0)),
|
||
radius_m=int(params.get("radius_m", 1500)),
|
||
enrich_address=bool(params.get("enrich_address", True)),
|
||
segments=params.get("segments"),
|
||
)
|
||
|
||
|
||
async def _job_cian_city_sweep(
|
||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
||
) -> None:
|
||
await run_cian_city_sweep(
|
||
db,
|
||
run_id=run_id,
|
||
config=ctx.config,
|
||
matcher=ctx.matcher,
|
||
shutdown_requested=ctx.shutdown_requested,
|
||
pages_per_anchor=int(params.get("pages_per_anchor", 3)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 5.0)),
|
||
radius_m=int(params.get("radius_m", 1500)),
|
||
detail_top_n=int(params.get("detail_top_n", 10)),
|
||
enrich_houses=bool(params.get("enrich_houses", True)),
|
||
newbuilding_only=bool(params.get("newbuilding_only", True)),
|
||
)
|
||
|
||
|
||
async def _job_cian_full_load(
|
||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
||
) -> None:
|
||
await run_cian_full_load(
|
||
db,
|
||
run_id=run_id,
|
||
config=ctx.config,
|
||
matcher=ctx.matcher,
|
||
shutdown_requested=ctx.shutdown_requested,
|
||
price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
|
||
concurrency=int(params.get("concurrency", 5)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 4.0)),
|
||
enrich_detail=bool(params.get("enrich_detail", False)),
|
||
detail_top_n=int(params.get("detail_top_n", 0)),
|
||
resume_run_id=None,
|
||
)
|
||
|
||
|
||
async def _job_domclick_city_sweep(
|
||
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
|
||
) -> None:
|
||
await run_domclick_city_sweep(
|
||
db,
|
||
run_id=run_id,
|
||
config=ctx.config,
|
||
matcher=ctx.matcher,
|
||
shutdown_requested=ctx.shutdown_requested,
|
||
city_id=int(params.get("city_id", 4)),
|
||
rooms=params.get("rooms"),
|
||
pages=int(params.get("pages_per_anchor", 5)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 6.0)),
|
||
)
|
||
|
||
|
||
def _default_kit_handlers() -> dict[str, Handler]:
|
||
"""Встроенные kit-native sweep-обработчики (тело в scraper_kit.orchestration.pipeline)."""
|
||
return {
|
||
"avito_city_sweep": Handler(_job_avito_city_sweep, "avito_city_sweep"),
|
||
"avito_newbuilding_sweep": Handler(_job_avito_newbuilding_sweep, "avito_newbuilding_sweep"),
|
||
"avito_full_load": Handler(_job_avito_full_load, "avito_full_load"),
|
||
"avito_full_load_exhaustive": Handler(
|
||
_job_avito_full_load_exhaustive, "avito_full_load_exhaustive"
|
||
),
|
||
"yandex_city_sweep": Handler(_job_yandex_city_sweep, "yandex_city_sweep"),
|
||
"cian_city_sweep": Handler(_job_cian_city_sweep, "cian_city_sweep"),
|
||
"cian_full_load": Handler(_job_cian_full_load, "cian_full_load"),
|
||
"domclick_city_sweep": Handler(_job_domclick_city_sweep, "domclick_city_sweep"),
|
||
}
|
||
|
||
|
||
def build_registry(product_handlers: Mapping[str, Handler] | None = None) -> dict[str, Handler]:
|
||
"""Собрать реестр source→Handler: kit-native sweeps + инжектированные продуктовые джобы.
|
||
|
||
`product_handlers` — джобы, тело которых осталось в `app` (rosreestr_dkp / sber_index /
|
||
deactivate_stale_* / *_backfill / cadastral_geo_match / house_* / proxy_healthcheck / …).
|
||
Их строит app-side wiring и передаёт сюда. Продуктовые ключи МОГУТ переопределять
|
||
kit-native (последнее слово за продуктом).
|
||
|
||
Wildcard-ключи ("deactivate_stale_*") резолвятся `resolve_handler` по префиксу.
|
||
"""
|
||
registry = dict(_default_kit_handlers())
|
||
if product_handlers:
|
||
registry.update(product_handlers)
|
||
return registry
|
||
|
||
|
||
def get_due_schedules(db: Session) -> list[dict[str, Any]]:
|
||
"""SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT id, source, enabled, window_start_hour, window_end_hour,
|
||
default_params, last_run_id, last_run_at, next_run_at
|
||
FROM scrape_schedules
|
||
WHERE enabled = true
|
||
AND (next_run_at IS NULL OR next_run_at <= NOW())
|
||
"""
|
||
),
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
async def scheduler_loop(ctx: SchedulerContext, registry: Mapping[str, Handler]) -> None:
|
||
"""Бесконечный async loop — tick каждые SCHEDULER_TICK_SEC секунд.
|
||
|
||
Структура (reap → get_due → dispatch → SIGTERM-drain) идентична боевому
|
||
scheduler'у; 27-веточный if/elif заменён единственным `resolve_handler` + `_dispatch`.
|
||
"""
|
||
logger.info("scheduler: started (tick=%ds)", SCHEDULER_TICK_SEC)
|
||
# Initial sleep 30s чтобы дать FastAPI startup завершиться
|
||
await asyncio.sleep(30)
|
||
while True:
|
||
# #1182 Phase 2: кооперативный SIGTERM-drain. Не reap'аем и не claim'аем
|
||
# новые run'ы во время shutdown — даём текущему dispatch'у докатиться и выходим.
|
||
if ctx.shutdown_requested():
|
||
logger.info("scheduler: SIGTERM-drain — stop claiming new runs, exiting tick loop")
|
||
break
|
||
try:
|
||
db = ctx.session_factory()
|
||
try:
|
||
# Reap zombies first
|
||
reap_zombies(db)
|
||
# Process due schedules
|
||
due = get_due_schedules(db)
|
||
for sch in due:
|
||
source = sch["source"]
|
||
handler = resolve_handler(source, registry)
|
||
if handler is None:
|
||
logger.warning("scheduler: unknown source=%s, skip", source)
|
||
else:
|
||
await _dispatch(handler, db, sch, ctx)
|
||
|
||
# #1182 Phase 2: после каждого dispatch'а проверяем drain — текущий
|
||
# run уже отпущен в свою asyncio-задачу (сам докоммитит/выйдет по
|
||
# своему checkpoint'у), а новые в этом тике не запускаем.
|
||
if ctx.shutdown_requested():
|
||
logger.info(
|
||
"scheduler: SIGTERM-drain — stop dispatch after source=%s", source
|
||
)
|
||
break
|
||
finally:
|
||
db.close()
|
||
except Exception:
|
||
logger.exception("scheduler: tick failed")
|
||
# Промежуточная проверка перед 60s-сном: выходим сразу, не ждём целый тик.
|
||
if ctx.shutdown_requested():
|
||
logger.info("scheduler: SIGTERM-drain — exiting tick loop after dispatch")
|
||
break
|
||
await asyncio.sleep(SCHEDULER_TICK_SEC)
|
||
|
||
# Tick-loop вышел только по SIGTERM-drain'у (иначе while True бесконечен): дожидаемся
|
||
# detached run-задач (spawn_tracked) — пусть докоммитят текущий unit и сделают
|
||
# mark_done, а не дадим asyncio.run() teardown'у хард-кансельнуть их mid-await (#1182).
|
||
await ctx.drain_inflight()
|
||
logger.info("scheduler: tick loop exited (in-flight drain complete)")
|
||
|
||
|
||
__all__ = [
|
||
"SCHEDULER_TICK_SEC",
|
||
"ZOMBIE_THRESHOLD_HOURS",
|
||
"Handler",
|
||
"SchedulerContext",
|
||
"build_registry",
|
||
"compute_next_run_at",
|
||
"get_due_schedules",
|
||
"has_running_run",
|
||
"reap_zombies",
|
||
"reschedule_after_minutes",
|
||
"resolve_handler",
|
||
"scheduler_loop",
|
||
]
|