backend/app/services/cadastre/bulk_harvest.py

@@ -218,6 +218,7 @@ async def harvest_quarter(
             client=client,
             quarter=quarter,
             layer_id=cat_id,
+            update_progress=update_progress,
         )
         result.grid_walk_requests += n_requests
 
@@ -276,12 +277,19 @@ async def _grid_walk_category(
     layer_id: int,
     grid_size: int = 15,
     tile_size: int = 512,
+    update_progress: Callable[[dict[str, Any]], None] | None = None,
+    heartbeat_every: int = 50,
 ) -> tuple[int, int]:
     """Grid-walk layer_id в bbox квартала.
 
     Генерирует grid_size×grid_size ячеек, для каждой делает wms_feature_info.
     Дедуплицирует по cad_num, upsert'ит новые объекты.
 
+    Args:
+        update_progress: Fix I — callback для heartbeat внутри grid-walk loop.
+            225 cells × ~70ms = ~15-75s без heartbeat, zombie-cleanup ложно
+            помечал worker как мёртвого. Вызываем каждые heartbeat_every cells.
+
     Returns:
         (upserted_count, requests_count)
     """
@@ -300,7 +308,19 @@ async def _grid_walk_category(
     upserted = 0
     requests = 0
 
-    for cell_bbox, click_xy in grid_points:
+    for idx, (cell_bbox, click_xy) in enumerate(grid_points):
+        # Fix I: heartbeat каждые N cells — обновляет cadastre_jobs.heartbeat_at
+        # чтобы cleanup_zombies (5min stale check) не cancel'нул живой worker.
+        if update_progress is not None and idx > 0 and idx % heartbeat_every == 0:
+            update_progress(
+                {
+                    "phase": f"grid_walk_{layer_id}_cell_{idx}",
+                    "quarter": quarter,
+                    "grid_cells_done": idx,
+                    "grid_cells_total": len(grid_points),
+                }
+            )
+
         try:
             features = await client.wms_feature_info(
                 layer_id=layer_id,
@@ -311,7 +331,10 @@ async def _grid_walk_category(
             )
             requests += 1
         except Exception as e:
-            logger.warning(
+            # Fix J: NSPD WMS возвращает HTTP 500 на десятки cells per quarter —
+            # это server-side noise, не наш bug. logger.debug чтобы не засорять
+            # prod логи (раньше было warning → 200+ warnings per quarter).
+            logger.debug(
                 "_grid_walk_category: wms_feature_info error layer=%d quarter=%s: %s",
                 layer_id,
                 quarter,
@@ -500,7 +523,7 @@ def upsert_parcel(db: Session, feature: NSPDBulkFeature, source: str = "search")
             "readable_address": raw_opts.get("readable_address"),
             "status": raw_opts.get("status"),
             "previously_posted": raw_opts.get("previously_posted"),
-            "registration_date": raw_opts.get("registration_date"),
+            "registration_date": _parse_nspd_date(raw_opts.get("registration_date")),
             "subcategory": _safe_int(raw_opts.get("subcategory")),
             "cadastral_districts_code": _safe_int(raw_opts.get("cadastral_districts_code")),
             "geom": geom_json,
@@ -615,8 +638,10 @@ def upsert_building(db: Session, feature: NSPDBulkFeature, source: str = "search
             "cost_index": _safe_numeric(raw_opts.get("cost_index")),
             "year_built": _safe_int(raw_opts.get("year_built")),
             "year_commisioning": _safe_int(raw_opts.get("year_commisioning")),
-            "build_record_registration_date": raw_opts.get("build_record_registration_date"),
+            "build_record_registration_date": _parse_nspd_date(
-            "registration_date": raw_opts.get("registration_date"),
+                raw_opts.get("build_record_registration_date")
+            ),
+            "registration_date": _parse_nspd_date(raw_opts.get("registration_date")),
             "ownership_type": raw_opts.get("ownership_type"),
             "cultural_heritage_object": cultural_heritage_bool,
             "cultural_heritage_val": ch_str,
@@ -746,7 +771,7 @@ def upsert_construction(db: Session, feature: NSPDBulkFeature, source: str = "se
             "address_readable_address": raw_opts.get("address_readable_address"),
             "status": raw_opts.get("status"),
             "common_data_status": raw_opts.get("common_data_status"),
-            "registration_date": raw_opts.get("registration_date"),
+            "registration_date": _parse_nspd_date(raw_opts.get("registration_date")),
             "registers_id": _safe_int(raw_opts.get("registers_id")),
             "united_cad_number": raw_opts.get("united_cad_number"),
             "geom": geom_json,
@@ -870,7 +895,7 @@ def upsert_enk(db: Session, feature: NSPDBulkFeature, source: str = "search") ->
             "cost_value": _safe_numeric(raw_opts.get("cost_value")),
             "facility_cad_number": raw_opts.get("facility_cad_number"),
             "readable_address": raw_opts.get("readable_address"),
-            "registration_date": raw_opts.get("registration_date"),
+            "registration_date": _parse_nspd_date(raw_opts.get("registration_date")),
             "registers_id": _safe_int(raw_opts.get("registers_id")),
             "geom": geom_json,
             "raw_props": json.dumps(raw_opts, ensure_ascii=False),
@@ -973,10 +998,10 @@ def upsert_zouit(db: Session, feature: NSPDBulkFeature, source: str = "search")
             "content_restrict_encumbrances": raw_opts.get("content_restrict_encumbrances"),
             "name_by_doc": raw_opts.get("name_by_doc"),
             "legal_act_name": raw_opts.get("legal_act_name"),
-            "legal_act_date": raw_opts.get("legal_act_date"),
+            "legal_act_date": _parse_nspd_date(raw_opts.get("legal_act_date")),
             "legal_act_number": raw_opts.get("legal_act_number"),
             "legal_act_issuer": raw_opts.get("legal_act_issuer"),
-            "registration_date": raw_opts.get("registration_date"),
+            "registration_date": _parse_nspd_date(raw_opts.get("registration_date")),
             "old_account_number": raw_opts.get("old_account_number"),
             "geom": geom_json,
             "raw_props": json.dumps(raw_opts, ensure_ascii=False),
@@ -1180,3 +1205,42 @@ def _safe_int(value: Any) -> int | None:
         return int(value)
     except (ValueError, TypeError):
         return None
+
+
+def _parse_nspd_date(value: Any) -> str | None:
+    """Конвертировать NSPD date в ISO YYYY-MM-DD string.
+
+    NSPD возвращает даты в смешанных форматах:
+    - '13.03.2023' (DD.MM.YYYY — для Сооружения/ОНС/ЕНК)
+    - '2023-03-13' (ISO — для ЗУ/Здания)
+    - '2023-03-13 00:00:00' (ISO with time)
+    - None / empty string
+
+    PostgreSQL DATE cast принимает ISO. Без conversion DD.MM.YYYY → DatetimeFieldOverflow
+    crash в upsert_*. Также валидируем что результат — actually-parseable date чтобы
+    invalid input не уехал в SQL.
+    """
+    import datetime as _dt
+
+    if value is None or value == "":
+        return None
+    if not isinstance(value, str):
+        return None
+    s = value.strip()
+    candidate: str | None = None
+    # ISO format already — pass through (PG accept YYYY-MM-DD)
+    if len(s) >= 10 and s[4] == "-" and s[7] == "-":
+        candidate = s[:10]
+    # DD.MM.YYYY → YYYY-MM-DD
+    elif len(s) == 10 and s[2] == "." and s[5] == ".":
+        parts = s.split(".")
+        if len(parts) == 3 and all(p.isdigit() for p in parts):
+            candidate = f"{parts[2]}-{parts[1]}-{parts[0]}"
+    if candidate is None:
+        return None
+    # Validate: actually parseable as date (catches '2023-13-45', '32.13.2023', etc)
+    try:
+        _dt.date.fromisoformat(candidate)
+        return candidate
+    except ValueError:
+        return None

backend/app/workers/tasks/scrape_cadastre.py

@@ -357,22 +357,65 @@ def enqueue_cadastre_harvest(self: Any, job_id: int) -> dict[str, Any]:
             logger.warning("enqueue_cadastre_harvest: job=%d имеет пустой список quarters", job_id)
             return {"job_id": job_id, "enqueued": 0}
 
+        # Fix H: per-quarter freshness skip — не переобрабатывать quarters с свежим
+        # cad_quarter_stats.fetched_at (последний успешный обход <FRESHNESS_HOURS).
+        # raw_targets.skip_fresh_hours=N управляет порогом; 0 → отключает skip.
+        # Хранится в существующем `raw_targets` JSONB (no schema change).
+        # Это сэкономит ~30k NSPD calls при повторных запусках того же scope.
+        skip_fresh_hours = int(raw_targets.get("skip_fresh_hours", 1) or 0)
+        skipped_fresh: set[str] = set()
+        if skip_fresh_hours > 0:
+            threshold = _dt.datetime.now(_dt.UTC) - _dt.timedelta(hours=skip_fresh_hours)
+            rows = (
+                db.execute(
+                    text(
+                        "SELECT cad_number FROM cad_quarter_stats "
+                        "WHERE cad_number = ANY(CAST(:qs AS text[])) "
+                        "AND fetched_at >= :t"
+                    ),
+                    {"qs": quarters, "t": threshold},
+                )
+                .scalars()
+                .all()
+            )
+            skipped_fresh = set(rows)
+            if skipped_fresh:
+                # Корректируем targets_total — skipped quarters не входят в прогресс
+                db.execute(
+                    text(
+                        "UPDATE cadastre_jobs SET targets_total = :new_total, "
+                        "targets_skipped = COALESCE(targets_skipped, 0) + :sk "
+                        "WHERE job_id = :id"
+                    ),
+                    {
+                        "new_total": len(quarters) - len(skipped_fresh),
+                        "sk": len(skipped_fresh),
+                        "id": job_id,
+                    },
+                )
+
         # Обновить статус на running
         _update_job(db, job_id, status="running", started_at=_dt.datetime.now(_dt.UTC))
         db.commit()
 
+        enqueued = 0
         for quarter in quarters:
+            if quarter in skipped_fresh:
+                continue
             bulk_harvest_quarter_task.apply_async(
                 args=[quarter, job_id],
                 queue=CADASTRE_QUEUE,
             )
+            enqueued += 1
 
         logger.info(
-            "enqueue_cadastre_harvest: job=%d enqueued %d quarters",
+            "enqueue_cadastre_harvest: job=%d enqueued %d (skipped_fresh=%d, threshold=%dh)",
             job_id,
-            len(quarters),
+            enqueued,
+            len(skipped_fresh),
+            skip_fresh_hours,
         )
-        return {"job_id": job_id, "enqueued": len(quarters)}
+        return {"job_id": job_id, "enqueued": enqueued, "skipped_fresh": len(skipped_fresh)}
 
     finally:
         db.close()

backend/tests/services/test_cadastre_bulk.py

@@ -168,6 +168,44 @@ def test_safe_int_none_returns_none() -> None:
     assert _safe_int(None) is None
 
 
+def test_parse_nspd_date_handles_dd_mm_yyyy() -> None:
+    """NSPD Сооружения/ОНС/ЕНК отдают даты в DD.MM.YYYY → конвертируем в ISO."""
+    from app.services.cadastre.bulk_harvest import _parse_nspd_date
+
+    assert _parse_nspd_date("13.03.2023") == "2023-03-13"
+    assert _parse_nspd_date("24.02.2022") == "2022-02-24"
+
+
+def test_parse_nspd_date_passes_through_iso() -> None:
+    """ISO формат (YYYY-MM-DD) уже валидный — pass через."""
+    from app.services.cadastre.bulk_harvest import _parse_nspd_date
+
+    assert _parse_nspd_date("2023-03-13") == "2023-03-13"
+    assert _parse_nspd_date("2010-08-12 00:00:00") == "2010-08-12"
+
+
+def test_parse_nspd_date_none_and_invalid() -> None:
+    """None / empty string / garbage → None."""
+    from app.services.cadastre.bulk_harvest import _parse_nspd_date
+
+    assert _parse_nspd_date(None) is None
+    assert _parse_nspd_date("") is None
+    assert _parse_nspd_date("nonsense") is None
+    assert _parse_nspd_date(123) is None  # not str
+
+
+def test_parse_nspd_date_validates_month_day_ranges() -> None:
+    """Invalid month/day (13.03 / 32.13 / etc) → None — не пропускать в SQL."""
+    from app.services.cadastre.bulk_harvest import _parse_nspd_date
+
+    # Невалидный месяц (13) в ISO формате
+    assert _parse_nspd_date("2023-13-45") is None
+    # Невалидный день в DD.MM.YYYY
+    assert _parse_nspd_date("32.13.2023") is None
+    # Невалидный month
+    assert _parse_nspd_date("31.02.2023") is None  # Feb 31
+
+
 # ── Grid geometry ────────────────────────────────────────────────────────────
 
 
@@ -1096,3 +1134,73 @@ async def test_harvest_quarter_geom_heal_failure_does_not_propagate() -> None:
     mock_heal.assert_called_once()
     # Phase=done — harvest завершился штатно
     assert result.phase_state == {"phase": "done", "quarter": "66:41:0303161"}
+
+
+# ── Fix I: heartbeat в grid-walk loop ────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_grid_walk_emits_heartbeat_callbacks() -> None:
+    """_grid_walk_category должен вызывать update_progress каждые heartbeat_every cells.
+
+    225 cells (15×15) / 50 = 4 heartbeats minimum (idx=50, 100, 150, 200).
+    """
+    from app.services.cadastre.bulk_harvest import _grid_walk_category
+
+    # Mock db: quarter_bbox_3857 returns valid bbox
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = {
+        "xmin": 6735845.0,
+        "ymin": 8329000.0,
+        "xmax": 6736595.0,
+        "ymax": 8329750.0,
+    }
+
+    # Mock client: wms_feature_info returns empty list (no upserts to worry about)
+    client = AsyncMock()
+    client.wms_feature_info = AsyncMock(return_value=[])
+
+    progress_states: list[dict[str, Any]] = []
+
+    upserted, requests = await _grid_walk_category(
+        db=db,
+        client=client,
+        quarter="66:41:0303161",
+        layer_id=36368,
+        update_progress=lambda s: progress_states.append(s),
+        heartbeat_every=50,
+    )
+
+    # 225 cells / 50 = 4 heartbeats at idx=50/100/150/200
+    assert len(progress_states) == 4
+    assert all("grid_cells_done" in s for s in progress_states)
+    assert progress_states[0]["grid_cells_done"] == 50
+    assert progress_states[-1]["grid_cells_done"] == 200
+    assert progress_states[0]["quarter"] == "66:41:0303161"
+    # 225 WMS calls (request counter)
+    assert requests == 225
+
+
+@pytest.mark.asyncio
+async def test_grid_walk_no_heartbeat_when_callback_none() -> None:
+    """Если update_progress=None, не должны падать (callback optional)."""
+    from app.services.cadastre.bulk_harvest import _grid_walk_category
+
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = {
+        "xmin": 6735845.0,
+        "ymin": 8329000.0,
+        "xmax": 6736595.0,
+        "ymax": 8329750.0,
+    }
+    client = AsyncMock()
+    client.wms_feature_info = AsyncMock(return_value=[])
+
+    # Не передаём update_progress — должно не падать
+    upserted, requests = await _grid_walk_category(
+        db=db,
+        client=client,
+        quarter="66:41:0303161",
+        layer_id=36368,
+    )
+    assert requests == 225

backend/tests/workers/test_scrape_cadastre.py

@@ -0,0 +1,132 @@
+"""Тесты для backend/app/workers/tasks/scrape_cadastre.py.
+
+Покрывает Fix H — per-quarter freshness skip в enqueue_cadastre_harvest.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+
+def _make_mock_job(quarters: list[str], raw_targets_extra: dict[str, Any] | None = None) -> dict:
+    """Mock cadastre_jobs row."""
+    raw_targets = {"quarters": quarters}
+    if raw_targets_extra:
+        raw_targets.update(raw_targets_extra)
+    return {
+        "job_id": 99,
+        "status": "pending",
+        "raw_targets": raw_targets,
+    }
+
+
+def _make_mock_db(job: dict, fresh_quarters: list[str] | None = None) -> MagicMock:
+    """Mock DB session.
+
+    1. db.execute("SELECT * FROM cadastre_jobs ...").mappings().first() → job
+    2. db.execute("SELECT cad_number FROM cad_quarter_stats ...").scalars().all() → fresh
+    3. db.execute(UPDATE) → no-op
+    """
+    db = MagicMock()
+    fresh_quarters = fresh_quarters or []
+
+    def execute_router(stmt: Any, params: Any = None) -> MagicMock:
+        sql = str(stmt).upper()
+        result = MagicMock()
+        if "FROM CADASTRE_JOBS" in sql and "WHERE JOB_ID" in sql and "SELECT *" in str(stmt):
+            result.mappings.return_value.first.return_value = job
+        elif "FROM CAD_QUARTER_STATS" in sql:
+            result.scalars.return_value.all.return_value = fresh_quarters
+        else:
+            # UPDATE / other queries
+            result.scalars.return_value.all.return_value = []
+            result.mappings.return_value.first.return_value = None
+        return result
+
+    db.execute = execute_router
+    return db
+
+
+def test_enqueue_skips_fresh_quarters_with_default_threshold() -> None:
+    """Fix H: фреш кварталы (cad_quarter_stats.fetched_at < 1h) пропускаются."""
+    from app.workers.tasks import scrape_cadastre
+
+    quarters = ["66:41:0001001", "66:41:0001002", "66:41:0001003"]
+    fresh = ["66:41:0001001", "66:41:0001002"]  # 2 свежих, 1 stale
+
+    job = _make_mock_job(quarters)
+    db = _make_mock_db(job, fresh_quarters=fresh)
+
+    with (
+        patch.object(scrape_cadastre, "SessionLocal", return_value=db),
+        patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async") as mock_apply,
+    ):
+        result = scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
+
+    # Только 1 stale quarter должен попасть в apply_async
+    assert mock_apply.call_count == 1
+    assert mock_apply.call_args[1]["args"] == ["66:41:0001003", 99]
+    assert result["enqueued"] == 1
+    assert result["skipped_fresh"] == 2
+
+
+def test_enqueue_no_skip_when_skip_fresh_hours_zero() -> None:
+    """skip_fresh_hours=0 → отключает freshness check, все quarters enqueued."""
+    from app.workers.tasks import scrape_cadastre
+
+    quarters = ["66:41:0001001", "66:41:0001002"]
+
+    job = _make_mock_job(quarters, raw_targets_extra={"skip_fresh_hours": 0})
+    db = _make_mock_db(job, fresh_quarters=["66:41:0001001"])  # один свежий
+
+    with (
+        patch.object(scrape_cadastre, "SessionLocal", return_value=db),
+        patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async") as mock_apply,
+    ):
+        result = scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
+
+    # Все 2 quarters должны быть enqueued (skip отключен)
+    assert mock_apply.call_count == 2
+    assert result["enqueued"] == 2
+    assert result["skipped_fresh"] == 0
+
+
+def test_enqueue_freshness_threshold_passed_in_sql() -> None:
+    """skip_fresh_hours=2 → threshold = NOW() - 2h в SQL query."""
+    from app.workers.tasks import scrape_cadastre
+
+    quarters = ["66:41:0001001"]
+    captured_params: list[dict[str, Any]] = []
+
+    job = _make_mock_job(quarters, raw_targets_extra={"skip_fresh_hours": 2})
+
+    def execute_router(stmt: Any, params: Any = None) -> MagicMock:
+        sql = str(stmt).upper()
+        result = MagicMock()
+        if "FROM CADASTRE_JOBS" in sql and "SELECT *" in str(stmt):
+            result.mappings.return_value.first.return_value = job
+        elif "FROM CAD_QUARTER_STATS" in sql:
+            if params:
+                captured_params.append(params)
+            result.scalars.return_value.all.return_value = []
+        else:
+            result.scalars.return_value.all.return_value = []
+        return result
+
+    db = MagicMock()
+    db.execute = execute_router
+
+    with (
+        patch.object(scrape_cadastre, "SessionLocal", return_value=db),
+        patch.object(scrape_cadastre.bulk_harvest_quarter_task, "apply_async"),
+    ):
+        scrape_cadastre.enqueue_cadastre_harvest.run(job_id=99)
+
+    assert len(captured_params) == 1
+    threshold: _dt.datetime = captured_params[0]["t"]
+    now = _dt.datetime.now(_dt.UTC)
+    # Threshold должен быть приблизительно 2h назад (±5s tolerance)
+    delta = now - threshold
+    assert _dt.timedelta(hours=1, minutes=59) < delta < _dt.timedelta(hours=2, minutes=1)