Планировщик кита передаёт регион из строки расписания в обход #3515

Merged
lekss361 merged 2 commits from feat/kit-scheduler-region-param into main 2026-09-13 11:45:58 +00:00
3 changed files with 148 additions and 1 deletions

View file

@ -39,6 +39,9 @@ from scraper_kit.orchestration.scheduler import (
from scraper_kit.orchestration.scheduler import (
_pick_resume as kit_pick_resume,
)
from scraper_kit.orchestration.scheduler import (
_resolve_region_code as kit_resolve_region_code,
)
if TYPE_CHECKING:
from scraper_kit.orchestration.scheduler import SchedulerContext
@ -515,6 +518,7 @@ async def _job_domclick_city_sweep(
rooms=params.get("rooms"),
pages=int(params.get("pages_per_anchor", 5)),
request_delay_sec=float(params.get("request_delay_sec", 6.0)),
region_code=kit_resolve_region_code(params),
resume_run_id=kit_pick_resume(db, run_id),
cookies=cookies,
)

View file

@ -25,10 +25,12 @@ import os
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
from scraper_kit.orchestration import scheduler as kit_sched
from scraper_kit.orchestration.pipeline import CITY_ANCHORS
from scraper_kit.orchestration.pipeline import CITY_ANCHORS, DEFAULT_REGION_CODE
from scraper_kit.orchestration.scheduler import (
Handler,
SchedulerContext,
@ -38,7 +40,10 @@ from scraper_kit.orchestration.scheduler import (
_job_avito_full_load,
_job_avito_full_load_exhaustive,
_job_cian_city_sweep,
_job_cian_full_load,
_job_domclick_city_sweep,
_job_yandex_city_sweep,
_resolve_region_code,
build_registry,
reap_zombies,
reschedule_after_minutes,
@ -623,3 +628,101 @@ def test_wildcard_resolves_oblast_city_sweep_sources() -> None:
# EKB source (no city) — по-прежнему резолвится на exact non-wildcard entry.
assert resolve_handler("avito_city_sweep", registry) is registry["avito_city_sweep"]
assert registry["avito_city_sweep"] is not registry["avito_city_sweep_*"]
# ── region_code: default_params → run_*-функции pipeline.py ─────────────────────────
#
# Планировщик кита раньше НЕ читал default_params["region_code"] ни разу — все sweep'ы
# писали листings регионом 66 (DEFAULT_REGION_CODE) независимо от того, что лежит в
# scrape_schedules.default_params. Это единственная причина, по которой московский обход
# нельзя было завести конфигом — pipeline.py уже принимает region_code параметром,
# scheduler.py его просто не прокидывал. Ниже: (1) отсутствие ключа — БЕЗ изменений
# поведения для всех существующих строк расписания (default 66); (2) валидный код (77)
# доезжает до run_*; (3) мусорное значение — явный отказ ДО сетевого похода, а не тихая
# запись в listings.region_code.
def test_resolve_region_code_missing_key_defaults_to_66() -> None:
"""Нет ключа в default_params → DEFAULT_REGION_CODE (66) — прежнее поведение всех
123 существующих строк расписания не меняется."""
assert _resolve_region_code({}) == DEFAULT_REGION_CODE == 66
def test_resolve_region_code_known_value_passes_through() -> None:
assert _resolve_region_code({"region_code": 77}) == 77
def test_resolve_region_code_string_digit_coerces_to_int() -> None:
"""jsonb default_params может прислать региона строкой — тот же int(...), что и
у остальных числовых параметров этого файла (params.get("pages", ...) и т.п.)."""
assert _resolve_region_code({"region_code": "77"}) == 77
def test_resolve_region_code_unknown_value_raises() -> None:
"""Регион вне реестра — явный ValueError, а не тихая запись мусора в listings."""
with pytest.raises(ValueError, match="реестре регионов"):
_resolve_region_code({"region_code": 999})
def test_resolve_region_code_non_numeric_raises() -> None:
with pytest.raises(ValueError, match="не приводится к int"):
_resolve_region_code({"region_code": "москва"})
async def test_job_avito_city_sweep_region_code_absent_defaults() -> None:
with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
await _job_avito_city_sweep(MagicMock(), 1, {}, _oblast_ctx())
_args, kwargs = mock_run.call_args
assert kwargs["region_code"] == DEFAULT_REGION_CODE
async def test_job_avito_city_sweep_region_code_from_params() -> None:
with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
await _job_avito_city_sweep(MagicMock(), 1, {"region_code": 77}, _oblast_ctx())
_args, kwargs = mock_run.call_args
assert kwargs["region_code"] == 77
async def test_job_avito_city_sweep_region_code_garbage_raises_before_run() -> None:
"""Мусорный region_code роняет job ДО вызова run_* — сеть/скрапинг не трогаются."""
with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
with pytest.raises(ValueError):
await _job_avito_city_sweep(MagicMock(), 1, {"region_code": 12345}, _oblast_ctx())
mock_run.assert_not_awaited()
async def test_job_avito_full_load_region_code_from_params() -> None:
with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run:
await _job_avito_full_load(MagicMock(), 1, {"region_code": 77}, _oblast_ctx())
_args, kwargs = mock_run.call_args
assert kwargs["region_code"] == 77
async def test_job_domclick_city_sweep_region_code_absent_defaults() -> None:
"""Регионный проброс не должен ломать уже читаемый city_id (тот же params dict)."""
with patch.object(kit_sched, "run_domclick_city_sweep", AsyncMock()) as mock_run:
await _job_domclick_city_sweep(MagicMock(), 1, {"city_id": 9}, _oblast_ctx())
_args, kwargs = mock_run.call_args
assert kwargs["region_code"] == DEFAULT_REGION_CODE
assert kwargs["city_id"] == 9
async def test_job_domclick_city_sweep_region_code_from_params() -> None:
with patch.object(kit_sched, "run_domclick_city_sweep", AsyncMock()) as mock_run:
await _job_domclick_city_sweep(MagicMock(), 1, {"region_code": 77}, _oblast_ctx())
_args, kwargs = mock_run.call_args
assert kwargs["region_code"] == 77
async def test_job_cian_full_load_region_code_absent_defaults() -> None:
with patch.object(kit_sched, "run_cian_full_load", AsyncMock()) as mock_run:
await _job_cian_full_load(MagicMock(), 1, {}, _oblast_ctx())
_args, kwargs = mock_run.call_args
assert kwargs["region_code"] == DEFAULT_REGION_CODE

View file

@ -47,6 +47,7 @@ except ImportError: # pragma: no cover - в проде backend-env sentry_sdk
from scraper_kit.orchestration import runs as _kit_runs
from scraper_kit.orchestration.pipeline import (
DEFAULT_REGION_CODE,
get_city_anchors,
run_avito_city_sweep,
run_avito_full_load,
@ -880,6 +881,37 @@ def _interval_minutes(params: dict[str, Any], *, param: str = "interval_minutes"
return None if raw is None else int(raw)
# Kit не импортирует `app.*` (contracts.py — граница strangler'а, см. её docstring),
# поэтому реестр `app.services.regions.REGIONS` сюда не дотянуть напрямую. Список
# ниже — суженная копия ЕГО КЛЮЧЕЙ только для отказа мусорного region_code из
# default_params; она не заменяет REGIONS и не должна расти самостоятельно — при
# добавлении региона в REGIONS обновить и здесь.
_KNOWN_REGION_CODES: frozenset[int] = frozenset({50, 66, 77})
def _resolve_region_code(params: dict[str, Any]) -> int:
"""default_params["region_code"] → валидированный int для run_*.
Ключа нет DEFAULT_REGION_CODE (66, тот же дефолт что и у run_*-функций
pipeline.py) прежнее поведение всех 123 существующих строк расписания не
меняется. Значение не парсится в int ИЛИ отсутствует в _KNOWN_REGION_CODES
ValueError: явный отказ прогона (падает сквозь job в `_dispatch`'s
`except Exception` `logger.exception`, тот же путь, что и любой другой
crash хендлера), а не тихая запись мусора в listings.region_code.
"""
raw = params.get("region_code", DEFAULT_REGION_CODE)
try:
region_code = int(raw)
except (TypeError, ValueError) as exc:
raise ValueError(f"region_code={raw!r} из default_params не приводится к int") from exc
if region_code not in _KNOWN_REGION_CODES:
raise ValueError(
f"region_code={region_code} отсутствует в реестре регионов "
f"(известны: {sorted(_KNOWN_REGION_CODES)})"
)
return region_code
# Нижний порог defer'а (#1522): пропуск обязан пережить несколько get_due_schedules,
# иначе pre-check снова гоняется каждый тик. 3 тика = минимум один промах даже при
# рассинхроне часов планировщика и БД на такт в любую сторону.
@ -1047,6 +1079,7 @@ async def _job_avito_city_sweep(
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)),
region_code=_resolve_region_code(params),
# #3074: подхват якорей у оборванного предшественника. Ключ чекпоинта —
# имя якоря, оно не зависит от количества якорей, поэтому в отличие от
# combo-чекпоинта yandex-свипа гарда по числу якорей здесь не требуется.
@ -1066,6 +1099,7 @@ async def _job_avito_newbuilding_sweep(
proxy_provider=ctx.proxy_provider,
pages=int(params.get("pages", 20)),
request_delay_sec=float(params.get("request_delay_sec", 7.0)),
region_code=_resolve_region_code(params),
# #3074: подхват страниц у оборванного предшественника — см.
# _job_avito_city_sweep выше.
resume_run_id=_pick_resume(db, run_id),
@ -1110,6 +1144,7 @@ async def _job_avito_full_load(
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)),
region_code=_resolve_region_code(params),
resume_run_id=_pick_resume(db, run_id),
incremental_days=incremental_days,
)
@ -1130,6 +1165,7 @@ async def _job_avito_full_load_exhaustive(
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)),
region_code=_resolve_region_code(params),
resume_run_id=_pick_resume(db, run_id),
incremental_days=None,
)
@ -1159,6 +1195,7 @@ async def _job_yandex_city_sweep(
radius_m=int(params.get("radius_m", 1500)),
enrich_address=bool(params.get("enrich_address", True)),
segments=params.get("segments"),
region_code=_resolve_region_code(params),
resume_run_id=_pick_resume(db, run_id),
)
@ -1184,6 +1221,7 @@ async def _job_cian_city_sweep(
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)),
region_code=_resolve_region_code(params),
# #3074: подхват якорей у оборванного предшественника. Ключ чекпоинта —
# имя якоря, от их количества не зависит, поэтому гарда как у combo-
# чекпоинта яндекса здесь не требуется.
@ -1206,6 +1244,7 @@ async def _job_cian_full_load(
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)),
region_code=_resolve_region_code(params),
resume_run_id=_pick_resume(db, run_id),
)
@ -1224,6 +1263,7 @@ async def _job_domclick_city_sweep(
rooms=params.get("rooms"),
pages=int(params.get("pages_per_anchor", 5)),
request_delay_sec=float(params.get("request_delay_sec", 6.0)),
region_code=_resolve_region_code(params),
resume_run_id=_pick_resume(db, run_id),
)