diff --git a/backend/app/services/cadastre/bulk_harvest.py b/backend/app/services/cadastre/bulk_harvest.py index 173a328d..289d4469 100644 --- a/backend/app/services/cadastre/bulk_harvest.py +++ b/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"), - "registration_date": raw_opts.get("registration_date"), + "build_record_registration_date": _parse_nspd_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 diff --git a/backend/app/workers/tasks/scrape_cadastre.py b/backend/app/workers/tasks/scrape_cadastre.py index 27dd9ec9..4c1ec35d 100644 --- a/backend/app/workers/tasks/scrape_cadastre.py +++ b/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 (последний успешный обход 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() diff --git a/backend/tests/services/test_cadastre_bulk.py b/backend/tests/services/test_cadastre_bulk.py index 19909e3b..1252ccc2 100644 --- a/backend/tests/services/test_cadastre_bulk.py +++ b/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 diff --git a/backend/tests/workers/test_scrape_cadastre.py b/backend/tests/workers/test_scrape_cadastre.py new file mode 100644 index 00000000..fbcd6e1f --- /dev/null +++ b/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)