From f6fb79522a168821e6efbdf5c81aed624e6f4dd3 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 5 Jul 2026 21:39:39 +0500 Subject: [PATCH] =?UTF-8?q?feat(scrapers):=20wire=20per-flat=20DOM.=D0=A0?= =?UTF-8?q?=D0=A4=20catalog=20scraper=20+=20propagate=20catalog=5Furl=5Fha?= =?UTF-8?q?sh=20(#2442)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 1: _norm_flat теперь читает catalog_url_hash из уже доступного kn-API elemId вместо жёсткого None (domrf_kn.py:479) — разблокирует источник хэшей для 983k квартир при следующем kn-sweep. elemId or None корректно нормализует и отсутствие, и пустую строку. Task 2: новый scrape_kn_catalog_flats Celery task оборачивает уже смерженный scrape_catalog_batch (#2441) — batch-выборка по последнему snapshot WHERE catalog_url_hash IS NOT NULL + 30d freshness/force. Зарегистрирован в celery_app, ручной триггер POST /kn-catalog-flats. Beat-запись оставлена закомментированной — по образцу уже отключённого object-level catalog scraper: DOM.РФ WAF hard-ban на VPS IP с 2026-05-24 (см. #2443), cooldown не подтверждён. Task ships полностью dormant: beat не назначен в рантайме, admin-endpoint требует авторизации, browser- stack импортируется лениво внутри функции (не на старте воркера), живой fetch недостижим без непустого batch. code-reviewer отдельно проверил WAF-safety независимо — подтверждено. Batch наполнится непустыми строками только после того, как Task 1 задеплоится И пройдёт свежий kn-sweep — ожидаемо 0 строк первое время. Refs #2442, #2441, #2443 --- backend/app/api/v1/admin_scrape.py | 47 ++++ backend/app/services/scrapers/domrf_kn.py | 13 +- backend/app/workers/beat_schedule.py | 18 ++ backend/app/workers/celery_app.py | 1 + .../workers/tasks/scrape_kn_catalog_flats.py | 214 ++++++++++++++++++ .../scrapers/test_domrf_kn_normalize.py | 31 +++ .../workers/test_scrape_kn_catalog_flats.py | 175 ++++++++++++++ 7 files changed, 498 insertions(+), 1 deletion(-) create mode 100644 backend/app/workers/tasks/scrape_kn_catalog_flats.py create mode 100644 backend/tests/workers/test_scrape_kn_catalog_flats.py diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index d84ce126..b1001074 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -1834,3 +1834,50 @@ def trigger_kn_catalog_objects( "force": payload.force, "queued_at": "now", } + + +class TriggerKnCatalogFlatsRequest(BaseModel): + region_code: int = Field(default=66, ge=1, le=99) + max_flats: int | None = Field(default=None, ge=1, le=5000) + force: bool = Field( + default=False, + description=( + "True — игнорировать фильтр свежести ('catalog_updated_at свежий') и " + "грузить ВСЕ квартиры последнего snapshot с непустым catalog_url_hash " + "('Загрузить все'). По умолчанию пропускает то, что скраплено < 30 дней назад." + ), + ) + + +@router.post("/kn-catalog-flats") +def trigger_kn_catalog_flats( + payload: TriggerKnCatalogFlatsRequest, +) -> dict[str, Any]: + """Manual trigger для catalog-FLAT scraper (#2442): цена/статус/отделка/потолки/ + дата обновления + plan-изображения квартир из SSR-страницы каталога. + + Селектит domrf_kn_flats WHERE catalog_url_hash IS NOT NULL. До тех пор пока + #2442 Task 1 (elemId → catalog_url_hash) не задеплоен и свежий kn-sweep не + наполнил hash — вернёт 0 обработанных строк (ожидаемо, не баг). + + - max_flats=None → дефолтный лимит таска (300). + - max_flats=3 → smoke-тест. + - force=True → 'Загрузить все': игнорирует фильтр свежести, грузит всё с hash. + """ + from app.workers.tasks.scrape_kn_catalog_flats import scrape_kn_catalog_flats + + kwargs: dict[str, Any] = { + "region_code": payload.region_code, + "force": payload.force, + } + if payload.max_flats is not None: + kwargs["max_flats"] = payload.max_flats + result = scrape_kn_catalog_flats.apply_async(kwargs=kwargs) + + return { + "task_id": result.id, + "region_code": payload.region_code, + "max_flats": payload.max_flats, + "force": payload.force, + "queued_at": "now", + } diff --git a/backend/app/services/scrapers/domrf_kn.py b/backend/app/services/scrapers/domrf_kn.py index 44fe97cd..3aa23056 100644 --- a/backend/app/services/scrapers/domrf_kn.py +++ b/backend/app/services/scrapers/domrf_kn.py @@ -432,6 +432,14 @@ def _norm_flat(row: dict[str, Any], region_cd: int | None) -> dict[str, Any]: digest = hashlib.sha256(str(elem).encode("utf-8")).digest()[:8] flat_id = int.from_bytes(digest, "big") % (2**63 - 1) + # catalog_url_hash — тот же elemId (hash-часть URL квартиры в каталоге DOM.РФ): + # https://наш.дом.рф/сервисы/каталог-квартир/квартира/{catalog_url_hash}. + # Читаем ЗДЕСЬ (а не только в fallback-ветке flat_id выше) — при наличии flatId + # ветка fallback не выполняется и elem остаётся непрочитанным. Пустую строку + # нормализуем в None: `or None` зеркалит трактовку `if elem:` в fallback-ветке + # и общий null-handling соседних optional-полей (_g отдаёт None при отсутствии). + catalog_url_hash = _g(row, "elemId") or None + price_rub = _g(row, "price") price_per_m2 = _g(row, "pricePerSquareMeter") total_area = _g(row, "totalArea") @@ -476,7 +484,10 @@ def _norm_flat(row: dict[str, Any], region_cd: int | None) -> dict[str, Any]: "ceiling_height_m": None, # заполнит catalog scraper 22d "key_handover_dt": None, # заполнит catalog scraper 22d "catalog_updated_at": None, # заполнит catalog scraper 22d - "catalog_url_hash": None, # заполнит catalog scraper 22d + # catalog_url_hash приходит из kn-API (elemId) уже здесь — нужен catalog + # scraper'у (22d) для построения URL страницы квартиры. Ранее хардкодился + # в None, из-за чего flat-catalog scraper не мог выбрать ни одной строки. + "catalog_url_hash": catalog_url_hash, } diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index 28dd70ec..ff2d812b 100644 --- a/backend/app/workers/beat_schedule.py +++ b/backend/app/workers/beat_schedule.py @@ -420,6 +420,24 @@ def build_beat_schedule() -> dict: # "options": {"queue": "celery"}, # } + # Catalog-FLAT scrape (#2442 Task 2) — наполняет per-flat catalog-поля + # (цена/статус/отделка/потолки/дата обновления) + plan-изображения квартир из + # SSR-страниц каталога. Селектит domrf_kn_flats WHERE catalog_url_hash IS NOT NULL. + # + # DISABLED (как и object-level выше): 1) тот же DOM.РФ WAF hard-ban на VPS IP — + # каталог-квартиры используют тот же BrowserSession + /сервисы/* paths. + # 2) Sequencing: пока #2442 Task 1 (elemId → catalog_url_hash) не задеплоен и + # свежий kn-sweep не наполнил hash, SELECT вернёт 0 строк — включать смысла нет. + # Возврат после WAF-cooldown + первого kn-sweep с hash (проверить targeted-тестом). + # Разнести по времени с object-scrape (вт 04:00), чтобы не двоить WAF-нагрузку — + # напр. четверг 04:00 UTC. + # schedule["scrape-kn-catalog-flats-weekly"] = { + # "task": "tasks.scrape_kn_catalog_flats.scrape_kn_catalog_flats", + # "schedule": _parse_cron("0 4 * * 4"), # Thursday 04:00 UTC + # "kwargs": {"region_code": 66, "max_flats": 300}, + # "options": {"queue": "celery"}, + # } + # Quarter price index refresh (#762) — monthly on the 5th at 05:00 MSK. # Celery conf.timezone=Europe/Moscow → crontab трактуется в МСК (#1233). # Chains: mv_quarter_price_per_m2 → mv_quarter_price_index (in order, both CONCURRENTLY). diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 3bc3a232..1268bfe6 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -50,6 +50,7 @@ celery_app = Celery( include=[ "app.workers.tasks.scrape_kn", "app.workers.tasks.scrape_kn_catalog_objects", + "app.workers.tasks.scrape_kn_catalog_flats", "app.workers.tasks.refresh_analytics", "app.workers.tasks.scrape_objective", "app.workers.tasks.objective_etl", diff --git a/backend/app/workers/tasks/scrape_kn_catalog_flats.py b/backend/app/workers/tasks/scrape_kn_catalog_flats.py new file mode 100644 index 00000000..0e30d1c6 --- /dev/null +++ b/backend/app/workers/tasks/scrape_kn_catalog_flats.py @@ -0,0 +1,214 @@ +"""Celery task: periodic catalog-FLAT scrape для DOM.РФ (issue #2442 Task 2). + +Дополняет per-flat catalog-поля (price_rub, price_per_m2, status, finishing_type, +ceiling_height_m, section_no, catalog_updated_at) + plan-изображения квартир в +domrf_kn_flats из SSR-страниц каталога: + https://наш.дом.рф/сервисы/каталог-квартир/квартира/{catalog_url_hash} + +kn-API отдаёт цену для меньшинства квартир (~8.5% NULL-free); остальные поля +(отделка, потолки, дата обновления) живут только на публичной странице квартиры. +Оркестрация здесь: выборка batch'а из БД + вызов `scrape_catalog_batch`, которая +сама держит один BrowserSession + встроенный jitter_sleep (rate-limit/stealth). + +Selector logic (аналог object-level scraper): + - catalog_url_hash IS NOT NULL — берём только строки, где kn-sweep уже записал + реальный hash. До #2442 Task 1 + свежего kn-sweep это 0 строк (hash хардкодился + в NULL) — пустой batch ожидаем/не баг. + - catalog_updated_at IS NULL — квартира ещё не скрапилась из каталога, грузим. + - catalog_updated_at < CURRENT_DATE - INTERVAL '30 days' — данные протухли, refresh. + - force=True — игнорирует «уже свежий» фильтр, грузит все строки последнего + snapshot с непустым hash. + +Sequencing dependency (важно): эта задача НЕ дублирует object-level scrape и +начнёт возвращать непустые batch'и ТОЛЬКО после того, как: + 1. #2442 Task 1 (elemId → catalog_url_hash) задеплоен, И + 2. прошёл свежий kn-sweep, наполнивший catalog_url_hash на новых snapshot'ах. +До этого SELECT вернёт 0 строк (см. выше). Beat-entry намеренно оставлен +ЗАКОММЕНТИРОВАННЫМ в beat_schedule.py (как и object-level) — WAF-cooldown на +VPS IP; включается вручную после проверки targeted-тестом. +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import date +from typing import Any + +from sqlalchemy import text + +from app.core.db import SessionLocal +from app.workers.celery_app import celery_app + +logger = logging.getLogger(__name__) + +# Выбор квартир на catalog-скрап одним запросом (obj-scraper-паттерн: obj_id+snapshot +# атомарно, здесь ods_id+catalog_url_hash). snapshot_date = MAX(...) для того же +# региона фиксирует последний snapshot и не тянет исторические строки — UPSERT в +# scrape_catalog_batch пишет `WHERE ods_id = :ods_id` (без snapshot-скоупа), поэтому +# отбираем только актуальный snapshot, чтобы не скрапить устаревшие ods_id. +# +# catalog_url_hash IS NOT NULL — жёсткое условие: без hash URL страницы не собрать. +# Пока Task 1 не задеплоен и kn-sweep не наполнил hash — 0 строк (ожидаемо). +# +# Фильтр свежести (:force = false): catalog_updated_at IS NULL (никогда не скрапили) +# ИЛИ старше 30 дней. При :force = true фильтр снимается — грузим все строки +# последнего snapshot с непустым hash. +# +# LIMIT :max_flats: в PostgreSQL `LIMIT NULL` == без лимита, поэтому при +# max_flats=None (force «Загрузить все» без явного потолка) грузим ВСЕ подходящие +# строки, а не молча режем до _DEFAULT_MAX_FLATS. +_SELECT_TARGETS_SQL = text( + """ + SELECT ods_id, catalog_url_hash, snapshot_date + FROM domrf_kn_flats + WHERE region_cd = :region_code + AND catalog_url_hash IS NOT NULL + AND ods_id IS NOT NULL + AND ( + CAST(:force AS boolean) + OR catalog_updated_at IS NULL + OR catalog_updated_at < CURRENT_DATE - CAST(:stale_days AS integer) + ) + AND snapshot_date = ( + SELECT MAX(snapshot_date) + FROM domrf_kn_flats + WHERE region_cd = :region_code + ) + ORDER BY catalog_updated_at NULLS FIRST + LIMIT :max_flats + """ +) + +# Лимит по умолчанию если max_flats не задан явно (обычный beat/ad-hoc pass). +_DEFAULT_MAX_FLATS = 300 + +# Считаем данные протухшими старше этого числа дней (refresh-порог). +_STALE_DAYS = 30 + + +@celery_app.task( + bind=True, + name="tasks.scrape_kn_catalog_flats.scrape_kn_catalog_flats", + time_limit=3600, +) +def scrape_kn_catalog_flats( + self: Any, + region_code: int = 66, + max_flats: int | None = None, + force: bool = False, +) -> dict[str, Any]: + """Periodic catalog-flat scrape (цены/отделка/планировки квартир из SSR каталога). + + Args: + region_code: Код региона (ОКАТО prefix). Default 66 = Свердловская обл. + max_flats: Максимум квартир за один run. Если не задан: при force=True + лимита нет (грузим все), при force=False — _DEFAULT_MAX_FLATS (300). + force: Если True — игнорирует фильтр свежести («catalog_updated_at свежий») + и (при не заданном max_flats) снимает лимит, грузя ВСЕ квартиры + последнего snapshot с непустым catalog_url_hash (admin «Загрузить все»). + По умолчанию False — пропускает то, что скраплено < 30 дней назад, и + режет batch до 300. + + Returns: + dict с ключами: region_code, force, snapshot_date, flats_count + поля из + scrape_catalog_batch (total, success, updated, failed, fields_total). + + Concurrency: + No Redis lock — consistent с sibling tasks (scrape_kn_catalog_objects и т.д.). + Beat отключён (WAF cooldown); при ручном/ad-hoc запуске пересечение крайне + маловероятно. Если случится: + - UPDATE идемпотентен (COALESCE в upsert_catalog_data, WHERE ods_id). + - Max risk: 2x WAF-нагрузка на DOM.РФ для того же batch. + Add Redis lock если WAF блокирует наблюдается или beat включается на overlap. + """ + from app.services.scrapers.domrf_catalog import scrape_catalog_batch + + # Явный max_flats всегда уважается. Без него: + # force=True («Загрузить все») → лимита нет (limit=None → SQL LIMIT NULL = все строки); + # force=False (beat / ad-hoc pass) → дефолтный batch _DEFAULT_MAX_FLATS. + if max_flats is not None: + limit: int | None = max_flats + elif force: + limit = None + else: + limit = _DEFAULT_MAX_FLATS + + db = SessionLocal() + try: + rows = ( + db.execute( + _SELECT_TARGETS_SQL, + { + "region_code": region_code, + "max_flats": limit, + "force": force, + "stale_days": _STALE_DAYS, + }, + ) + .mappings() + .all() + ) + flats: list[dict[str, Any]] = [ + {"ods_id": r["ods_id"], "catalog_url_hash": r["catalog_url_hash"]} for r in rows + ] + except Exception as exc: + logger.error("scrape_kn_catalog_flats: failed to fetch flats: %s", exc) + db.close() + raise + + if not flats: + logger.info( + "scrape_kn_catalog_flats: nothing to do for region=%d (force=%s) — " + "likely no catalog_url_hash populated yet (see #2442 sequencing)", + region_code, + force, + ) + db.close() + return { + "region_code": region_code, + "force": force, + "flats_count": 0, + "total": 0, + "success": 0, + "updated": 0, + "failed": 0, + "fields_total": 0, + } + + # snapshot_date из первой строки — все строки одинаковые (WHERE snapshot_date = + # MAX(snapshot_date)). Атомарно: один SELECT, без race с kn-scraper. + snapshot_date_val: date = rows[0]["snapshot_date"] + + logger.info( + "scrape_kn_catalog_flats: region=%d snapshot_date=%s flats=%d limit=%s force=%s", + region_code, + snapshot_date_val, + len(flats), + "ALL" if limit is None else limit, + force, + ) + + try: + stats = asyncio.run( + scrape_catalog_batch( + db=db, + flats=flats, + region_code=region_code, + ) + ) + except Exception as exc: + logger.error("scrape_kn_catalog_flats: scrape failed: %s", exc) + raise + finally: + db.close() + + result: dict[str, Any] = { + "region_code": region_code, + "force": force, + "snapshot_date": str(snapshot_date_val), + "flats_count": len(flats), + **stats, + } + logger.info("scrape_kn_catalog_flats done: %s", result) + return result diff --git a/backend/tests/services/scrapers/test_domrf_kn_normalize.py b/backend/tests/services/scrapers/test_domrf_kn_normalize.py index a8d98e62..35d2203c 100644 --- a/backend/tests/services/scrapers/test_domrf_kn_normalize.py +++ b/backend/tests/services/scrapers/test_domrf_kn_normalize.py @@ -55,3 +55,34 @@ class TestFlatIdFallbackDeterministic: out = _norm_flat({"elemId": "x" * 1000}, region_cd=66) assert out["id"] is not None assert 0 <= out["id"] < 2**63 + + +class TestCatalogUrlHash: + """catalog_url_hash ОБЯЗАН пробрасываться из elemId (#2442 Task 1). + + Ранее хардкодился в None → flat-catalog scraper выбирал 0 строк + (WHERE catalog_url_hash IS NOT NULL). Теперь берётся из того же elemId, + что и fallback flat_id, и живёт независимо от того, пришёл ли flatId. + """ + + def test_hash_populated_from_elem_id(self) -> None: + row = {"elemId": "abc-hash-123"} + out = _norm_flat(row, region_cd=66) + assert out["catalog_url_hash"] == "abc-hash-123" + + def test_hash_populated_even_when_flat_id_present(self) -> None: + # flatId присутствует → fallback-ветка flat_id НЕ выполняется, но hash + # всё равно должен прочитаться (регрессия, которую чинит Task 1). + row = {"flatId": 999, "elemId": "hash-with-flatid"} + out = _norm_flat(row, region_cd=66) + assert out["id"] == 999 + assert out["catalog_url_hash"] == "hash-with-flatid" + + def test_hash_none_when_elem_id_absent(self) -> None: + out = _norm_flat({"flatId": 1}, region_cd=66) + assert out["catalog_url_hash"] is None + + def test_hash_empty_string_normalized_to_none(self) -> None: + # Пустую строку трактуем как отсутствие (не пишем "" в БД). + out = _norm_flat({"flatId": 1, "elemId": ""}, region_cd=66) + assert out["catalog_url_hash"] is None diff --git a/backend/tests/workers/test_scrape_kn_catalog_flats.py b/backend/tests/workers/test_scrape_kn_catalog_flats.py new file mode 100644 index 00000000..a2963716 --- /dev/null +++ b/backend/tests/workers/test_scrape_kn_catalog_flats.py @@ -0,0 +1,175 @@ +"""Тесты для backend/app/workers/tasks/scrape_kn_catalog_flats.py (#2442 Task 2). + +Покрывают: batch-selection bind params (force / max_flats / stale_days), +маппинг rows → flats-dicts для scrape_catalog_batch, пустой batch (ожидаемо до +наполнения catalog_url_hash), и жёсткие предикаты SELECT (hash IS NOT NULL, CAST). +""" + +from __future__ import annotations + +from datetime import date +from typing import Any +from unittest.mock import MagicMock, patch + + +def _make_mock_db(rows: list[dict[str, Any]]) -> tuple[MagicMock, list[dict[str, Any]]]: + """Mock DB session. Возвращает (db, captured_params).""" + captured: list[dict[str, Any]] = [] + + def execute_router(stmt: Any, params: Any = None) -> MagicMock: + if params is not None: + captured.append(params) + result = MagicMock() + result.mappings.return_value.all.return_value = rows + return result + + db = MagicMock() + db.execute = execute_router + return db, captured + + +def test_skip_fresh_filter_active_by_default() -> None: + """force=False (default) → :force = False → SQL применит фильтр свежести.""" + from app.workers.tasks import scrape_kn_catalog_flats as task_mod + + db, captured = _make_mock_db(rows=[]) + + with patch.object(task_mod, "SessionLocal", return_value=db): + result = task_mod.scrape_kn_catalog_flats.run(region_code=66) + + assert len(captured) == 1 + assert captured[0]["force"] is False + assert captured[0]["region_code"] == 66 + assert captured[0]["stale_days"] == task_mod._STALE_DAYS + # Пустой результат → ранний выход (ожидаемо до наполнения catalog_url_hash). + assert result["flats_count"] == 0 + assert result["force"] is False + + +def test_force_true_passes_through_to_sql() -> None: + """force=True → :force = True → SQL грузит все строки с непустым hash.""" + from app.workers.tasks import scrape_kn_catalog_flats as task_mod + + db, captured = _make_mock_db(rows=[]) + + with patch.object(task_mod, "SessionLocal", return_value=db): + result = task_mod.scrape_kn_catalog_flats.run(region_code=66, force=True) + + assert len(captured) == 1 + assert captured[0]["force"] is True + assert result["force"] is True + + +def test_max_flats_passed_as_limit() -> None: + """max_flats явный — прилетает в SQL bind как :max_flats.""" + from app.workers.tasks import scrape_kn_catalog_flats as task_mod + + db, captured = _make_mock_db(rows=[]) + + with patch.object(task_mod, "SessionLocal", return_value=db): + task_mod.scrape_kn_catalog_flats.run(region_code=66, max_flats=10) + + assert captured[0]["max_flats"] == 10 + + +def test_default_limit_when_not_forced() -> None: + """Без max_flats и без force → дефолтный лимит _DEFAULT_MAX_FLATS.""" + from app.workers.tasks import scrape_kn_catalog_flats as task_mod + + db, captured = _make_mock_db(rows=[]) + + with patch.object(task_mod, "SessionLocal", return_value=db): + task_mod.scrape_kn_catalog_flats.run(region_code=66) + + assert captured[0]["max_flats"] == task_mod._DEFAULT_MAX_FLATS + + +def test_force_without_max_flats_is_unlimited() -> None: + """force=True без max_flats → limit=None (SQL LIMIT NULL = все строки).""" + from app.workers.tasks import scrape_kn_catalog_flats as task_mod + + db, captured = _make_mock_db(rows=[]) + + with patch.object(task_mod, "SessionLocal", return_value=db): + task_mod.scrape_kn_catalog_flats.run(region_code=66, force=True) + + assert captured[0]["max_flats"] is None + + +def test_processes_rows_when_returned() -> None: + """SELECT вернул строки → вызывается scrape_catalog_batch с flats-dicts + (ods_id + catalog_url_hash), snapshot_date из первой строки.""" + from app.workers.tasks import scrape_kn_catalog_flats as task_mod + + rows = [ + { + "ods_id": "65136/1/1.4.3", + "catalog_url_hash": "hash-a", + "snapshot_date": date(2026, 7, 5), + }, + { + "ods_id": "65136/1/1.4.4", + "catalog_url_hash": "hash-b", + "snapshot_date": date(2026, 7, 5), + }, + ] + db, _ = _make_mock_db(rows=rows) + + mock_stats = {"total": 2, "success": 2, "updated": 2, "failed": 0, "fields_total": 8} + with ( + patch.object(task_mod, "SessionLocal", return_value=db), + patch( + "app.services.scrapers.domrf_catalog.scrape_catalog_batch", + return_value=mock_stats, + ) as mock_scrape, + ): + result = task_mod.scrape_kn_catalog_flats.run(region_code=66) + + mock_scrape.assert_called_once() + call_kwargs = mock_scrape.call_args[1] + assert call_kwargs["flats"] == [ + {"ods_id": "65136/1/1.4.3", "catalog_url_hash": "hash-a"}, + {"ods_id": "65136/1/1.4.4", "catalog_url_hash": "hash-b"}, + ] + assert call_kwargs["region_code"] == 66 + assert result["flats_count"] == 2 + assert result["snapshot_date"] == "2026-07-05" + # stats из scrape_catalog_batch прокинуты в result. + assert result["updated"] == 2 + assert result["total"] == 2 + + +def test_empty_batch_returns_zeroed_stats() -> None: + """Пустой SELECT → все счётчики 0, scrape_catalog_batch НЕ вызывается.""" + from app.workers.tasks import scrape_kn_catalog_flats as task_mod + + db, _ = _make_mock_db(rows=[]) + + with ( + patch.object(task_mod, "SessionLocal", return_value=db), + patch( + "app.services.scrapers.domrf_catalog.scrape_catalog_batch", + ) as mock_scrape, + ): + result = task_mod.scrape_kn_catalog_flats.run(region_code=66) + + mock_scrape.assert_not_called() + assert result["flats_count"] == 0 + assert result["updated"] == 0 + assert result["failed"] == 0 + assert result["total"] == 0 + + +def test_sql_contains_hash_notnull_and_cast_predicates() -> None: + """Smoke: SELECT требует catalog_url_hash IS NOT NULL + использует CAST(...) + (psycopg v3 — никогда :x::type) для force/stale_days.""" + from app.workers.tasks.scrape_kn_catalog_flats import _SELECT_TARGETS_SQL + + sql = str(_SELECT_TARGETS_SQL) + upper = sql.upper() + assert "CATALOG_URL_HASH IS NOT NULL" in upper + assert "CAST(:FORCE AS BOOLEAN)" in upper + assert "CAST(:STALE_DAYS AS INTEGER)" in upper + # psycopg v3: никаких ::type после bind-имени. + assert ":force::" not in sql + assert ":stale_days::" not in sql