diff --git a/data/sql/122_grant_rosreestr_to_fdw_reader.sql b/data/sql/122_grant_rosreestr_to_fdw_reader.sql new file mode 100644 index 00000000..eae57e5d --- /dev/null +++ b/data/sql/122_grant_rosreestr_to_fdw_reader.sql @@ -0,0 +1,7 @@ +-- 122_grant_rosreestr_to_fdw_reader.sql +-- #563: tradein scheduler reads gendesign.rosreestr_deals via FDW; tradein_fdw_reader +-- was REVOKE ALL (100_tradein_fdw_role.sql) → grant SELECT on the partitioned parent. +-- Single parent grant suffices (PG declarative partitioning checks parent ACL). +BEGIN; +GRANT SELECT ON public.rosreestr_deals TO tradein_fdw_reader; +COMMIT; diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 9469d51f..2094d01b 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -6,6 +6,11 @@ 2. Для каждого: проверить нет ли уже running run этого source — если нет, trigger. 3. После trigger — recompute next_run_at в следующих суток. 4. На old runs (orphaned status='running' > 6h без heartbeat) — пометить как 'zombie'. + +Sources: + - avito_city_sweep → run_avito_city_sweep (scrape_pipeline.py) + - cian_history_backfill → run_cian_backfill (this module, #560) + - rosreestr_dkp_import → import_rosreestr_dkp (this module, #563) """ from __future__ import annotations @@ -15,6 +20,7 @@ import random from datetime import UTC, datetime, time, timedelta from typing import Any +import sentry_sdk from sqlalchemy import text from sqlalchemy.orm import Session @@ -105,19 +111,20 @@ def reap_zombies(db: Session) -> int: return len(rows) -async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None: - """Создать новый scrape_runs + launch run_avito_city_sweep в asyncio.create_task. +def _claim_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Claim run: INSERT scrape_runs + UPDATE scrape_schedules next_run_at. - Returns run_id (или None если skip — есть running run). + Returns run_id или None если уже есть running run для этого source. + Общий helper для всех source-handler'ов. """ - if has_running_run(db, schedule_row["source"]): - logger.info("scheduler: skip — already running for source=%s", schedule_row["source"]) + source = schedule_row["source"] + if has_running_run(db, source): + logger.info("scheduler: skip — already running for source=%s", source) return None params = schedule_row.get("default_params") or {} - run_id = runs_mod.create_run(db, source=schedule_row["source"], params=params) + run_id = runs_mod.create_run(db, source=source, params=params) - # Update schedule: last_run_id, last_run_at, recompute next_run_at next_at = compute_next_run_at( schedule_row["window_start_hour"], schedule_row["window_end_hour"], @@ -131,9 +138,28 @@ async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any] WHERE source = :source """ ), - {"run_id": run_id, "next_at": next_at, "source": schedule_row["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 + + +async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать новый scrape_runs + launch run_avito_city_sweep в asyncio.create_task. + + Returns run_id (или None если skip — есть running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick) async def _run() -> None: @@ -156,12 +182,378 @@ async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any] task = asyncio.create_task(_run()) # Keep reference to avoid GC before task completes (RUF006) task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) - logger.info( - "scheduler: triggered city_sweep run_id=%d next_run_at=%s", run_id, next_at.isoformat() - ) + logger.info("scheduler: triggered city_sweep run_id=%d", run_id) return run_id +async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать scrape_runs + launch run_cian_backfill в asyncio.create_task. + + Pre-checks Cian session validity before claiming the run. + Marks nothing as failed if cookies are absent/expired — just logs + GlitchTip warning. + Returns run_id или None (skip / expired cookies). + """ + from app.services.cian_session import load_session, verify_session + + # Pre-check: load and verify cookies before claiming the run slot + cookies = load_session(db) + if cookies is None: + logger.warning("scheduler: cian_history_backfill skipped — no valid session cookies in DB") + return None + + state = await verify_session(cookies) + if state is None: + logger.warning( + "scheduler: cian_history_backfill — cookies expired or invalid, skipping run" + ) + try: + sentry_sdk.capture_message( + "cian_history_backfill skipped: Cian session cookies expired — " + "please re-upload via admin UI", + level="warning", + ) + except Exception: + pass # sentry_sdk not initialised in dev + return None + + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + await _execute_cian_backfill(run_db, run_id=run_id, params=params) + except Exception: + logger.exception("scheduler: _execute_cian_backfill crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered cian_history_backfill run_id=%d", run_id) + return run_id + + +async def _execute_cian_backfill( + db: Session, + *, + run_id: int, + params: dict[str, Any], +) -> None: + """Orchestrate Cian history backfill with heartbeat + checkpoint. + + Wraps backfill_cian_history(), updating scrape_runs counters (via update_heartbeat) + before and after the batch call for zombie-detection visibility. + + Checkpoint/resume semantics: backfill_cian_history() queries rows WHERE history IS + NULL via LEFT JOIN — so re-running after a partial completion naturally skips + already-processed rows (idempotent by design). + + Params (from default_params jsonb): + batch_size: int — rows per run (listings + houses counted separately). + """ + from app.tasks.cian_history_backfill import backfill_cian_history + + batch_size = int(params.get("batch_size", 100)) + counters: dict[str, int] = { + "listings_processed": 0, + "listings_succeeded": 0, + "listings_failed": 0, + "houses_processed": 0, + "houses_succeeded": 0, + "houses_failed": 0, + } + + try: + runs_mod.update_heartbeat(db, run_id, counters) + + result = await backfill_cian_history( + db, + batch_size=batch_size, + do_listings=True, + do_houses=True, + do_valuations=False, + ) + + counters = { + "listings_processed": result.listings_processed, + "listings_succeeded": result.listings_succeeded, + "listings_failed": result.listings_failed_fetch + result.listings_failed_save, + "houses_processed": result.houses_processed, + "houses_succeeded": result.houses_succeeded, + "houses_failed": result.houses_failed_fetch + result.houses_failed_save, + "duration_sec": int(result.duration_sec), + } + runs_mod.mark_done(db, run_id, counters) + logger.info( + "scheduler: cian_history_backfill run_id=%d done — " + "listings=%d/%d houses=%d/%d %.1fs", + run_id, + result.listings_succeeded, + result.listings_total, + result.houses_succeeded, + result.houses_total, + result.duration_sec, + ) + except Exception as exc: + logger.exception("scheduler: cian_history_backfill run_id=%d failed", run_id) + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise + + +async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task).""" + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + # import_rosreestr_dkp is synchronous (DB-only, no async HTTP) + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, import_rosreestr_dkp, run_db, run_id, params) + except Exception: + logger.exception("scheduler: import_rosreestr_dkp crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered rosreestr_dkp_import run_id=%d", run_id) + return run_id + + +def import_rosreestr_dkp( + db: Session, + run_id: int, + params: dict[str, Any], +) -> None: + """Import ДКП-сделок из gendesign rosreestr_deals через postgres_fdw. + + Python-порт import-rosreestr.sh (Variant C из #563). + + Источник: foreign table gendesign_rosreestr_deals (создана в migration 072). + SERVER gendesign_remote настроен в 060_postgres_fdw_extension.sql. + USER MAPPING создаётся при startup в core/fdw.py (tradein_fdw_reader). + + Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24): + - region_code = 66 (Свердловская область) + - city ILIKE '%катеринбург%' + - realestate_type_code = '002001003000' (квартира) + - area BETWEEN 18 AND 200 + - deal_price BETWEEN 1000000 AND 100000000 + - street IS NOT NULL AND trim(street) != '' + - doc_type = 'ДКП' (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24) + - period_start_date >= since (default '2024-01-01') + + dedup_hash: md5('ros:dkp:' || id) — соответствует #549 / Fix_Rosreestr_Dkp_Filter_May24. + Rooms: выводятся из площади (Росреестр не отдаёт кол-во комнат). + + Batch-процессинг: читаем из FDW батчами по batch_size через cursor-based пагинацию + (WHERE id > last_id ORDER BY id). Heartbeat обновляется каждый батч (= checkpoint). + SAVEPOINT per row — один сбойный row не откатывает батч. + + Координаты: NULL после импорта — геокодинг остаётся follow-up (geocode-deals). + TODO (follow-up): запустить geocode backfill после import. + + Cleanup: удаляет legacy строки address='Екатеринбург, реальная сделка' (pre-#549). + """ + since: str = str(params.get("since", "2024-01-01")) + batch_size: int = int(params.get("batch_size", 2000)) + + counters: dict[str, int] = { + "rows_fetched": 0, + "rows_inserted": 0, + "rows_skipped": 0, + "batches_done": 0, + } + + # Cleanup legacy synthetic rows (pre-#549, idempotent) + try: + deleted = db.execute( + text( + "DELETE FROM deals WHERE address = CAST(:addr AS text) RETURNING id" + ), + {"addr": "Екатеринбург, реальная сделка"}, + ).fetchall() + db.commit() + if deleted: + logger.info( + "rosreestr_dkp_import run_id=%d: removed %d legacy synthetic rows", + run_id, + len(deleted), + ) + except Exception as exc: + logger.warning( + "rosreestr_dkp_import run_id=%d: cleanup failed (non-fatal): %s", run_id, exc + ) + db.rollback() + + last_id: int = 0 + total_batches = 0 + + try: + while True: + if runs_mod.is_cancelled(db, run_id): + logger.info("rosreestr_dkp_import run_id=%d: cancelled by user", run_id) + runs_mod.mark_cancelled(db, run_id) + return + + # Cursor-based pagination via foreign table gendesign_rosreestr_deals. + # FDW pushes WHERE + ORDER BY + LIMIT to gendesign-postgres automatically + # (postgres_fdw 'use_remote_estimate' is off by default but clause push-down + # still happens for simple predicates on the remote side). + batch_rows = db.execute( + text(""" + SELECT + id, + md5('ros:dkp:' || CAST(id AS text)) AS dedup_hash, + 'Екатеринбург, ' || trim(street) AS address, + CASE + WHEN area < 30 THEN 0 + WHEN area < 44 THEN 1 + WHEN area < 62 THEN 2 + WHEN area < 85 THEN 3 + ELSE 4 + END AS rooms, + round(area, 2) AS area_m2, + LEAST( + NULLIF( + substring(floor FROM '[0-9]+'), '' + )::int, + 100 + ) AS floor_num, + year_build AS year_built, + round(deal_price)::bigint AS price_rub, + round(price_per_sqm)::int AS price_per_m2, + period_start_date AS deal_date + FROM gendesign_rosreestr_deals + WHERE region_code = 66 + AND city ILIKE '%катеринбург%' + AND realestate_type_code = '002001003000' + AND area BETWEEN 18 AND 200 + AND deal_price BETWEEN 1000000 AND 100000000 + AND street IS NOT NULL AND trim(street) <> '' + AND doc_type = 'ДКП' + AND period_start_date >= CAST(:since AS date) + AND id > CAST(:last_id AS bigint) + ORDER BY id + LIMIT CAST(:batch_size AS int) + """), + {"since": since, "last_id": last_id, "batch_size": batch_size}, + ).mappings().all() + + if not batch_rows: + break + + total_batches += 1 + batch_inserted = 0 + batch_skipped = 0 + batch_max_id = last_id + + for row in batch_rows: + row_id: int = int(row["id"]) + if row_id > batch_max_id: + batch_max_id = row_id + + try: + with db.begin_nested(): # SAVEPOINT per row + inserted = db.execute( + text(""" + INSERT INTO deals ( + source, dedup_hash, address, rooms, area_m2, + floor, year_built, price_rub, price_per_m2, deal_date + ) + VALUES ( + 'rosreestr', + CAST(:dedup_hash AS text), + CAST(:address AS text), + CAST(:rooms AS int), + CAST(:area_m2 AS numeric), + CAST(:floor_num AS int), + CAST(:year_built AS int), + CAST(:price_rub AS bigint), + CAST(:price_per_m2 AS int), + CAST(:deal_date AS date) + ) + ON CONFLICT (dedup_hash) DO NOTHING + RETURNING id + """), + { + "dedup_hash": row["dedup_hash"], + "address": row["address"], + "rooms": row["rooms"], + "area_m2": row["area_m2"], + "floor_num": row["floor_num"], + "year_built": row["year_built"], + "price_rub": row["price_rub"], + "price_per_m2": row["price_per_m2"], + "deal_date": row["deal_date"], + }, + ).fetchone() + if inserted is not None: + batch_inserted += 1 + else: + batch_skipped += 1 + except Exception as exc: + logger.warning( + "rosreestr_dkp_import run_id=%d: row id=%d INSERT failed: %s", + run_id, + row_id, + exc, + ) + batch_skipped += 1 + + last_id = batch_max_id + db.commit() + + counters["rows_fetched"] += len(batch_rows) + counters["rows_inserted"] += batch_inserted + counters["rows_skipped"] += batch_skipped + counters["batches_done"] = total_batches + counters["last_id"] = last_id # type: ignore[assignment] + + # Heartbeat = checkpoint: allows zombie detection + resume visibility + runs_mod.update_heartbeat(db, run_id, counters) + logger.info( + "rosreestr_dkp_import run_id=%d: batch=%d fetched=%d " + "inserted=%d skipped=%d last_id=%d", + run_id, + total_batches, + len(batch_rows), + batch_inserted, + batch_skipped, + last_id, + ) + + if len(batch_rows) < batch_size: + break # Last partial batch — no more rows + + runs_mod.mark_done(db, run_id, counters) + logger.info( + "rosreestr_dkp_import run_id=%d done: " + "total_fetched=%d inserted=%d skipped=%d batches=%d", + run_id, + counters["rows_fetched"], + counters["rows_inserted"], + counters["rows_skipped"], + total_batches, + ) + + except Exception as exc: + logger.exception( + "rosreestr_dkp_import run_id=%d failed at last_id=%d", run_id, last_id + ) + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise + + 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( @@ -192,10 +584,15 @@ async def scheduler_loop() -> None: # Process due schedules due = get_due_schedules(db) for sch in due: - if sch["source"] == "avito_city_sweep": + source = sch["source"] + if source == "avito_city_sweep": await trigger_avito_city_sweep_run(db, sch) + elif source == "cian_history_backfill": + await trigger_cian_backfill_run(db, sch) + elif source == "rosreestr_dkp_import": + await trigger_rosreestr_dkp_run(db, sch) else: - logger.warning("scheduler: unknown source=%s, skip", sch["source"]) + logger.warning("scheduler: unknown source=%s, skip", source) finally: db.close() except Exception: diff --git a/tradein-mvp/backend/data/sql/072_scrape_schedules_seed_cian_rosreestr.sql b/tradein-mvp/backend/data/sql/072_scrape_schedules_seed_cian_rosreestr.sql new file mode 100644 index 00000000..99465fb8 --- /dev/null +++ b/tradein-mvp/backend/data/sql/072_scrape_schedules_seed_cian_rosreestr.sql @@ -0,0 +1,92 @@ +-- 072_scrape_schedules_seed_cian_rosreestr.sql +-- Seed rows для двух новых автоматизированных задач (#560 + #563). +-- Добавляет foreign table gendesign_rosreestr_deals для импорта через FDW. +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql, 060_postgres_fdw_extension.sql. +-- Idempotent: безопасно запускать повторно. +-- +-- cian_history_backfill — 02:00-05:00 UTC (05:00-08:00 МСК) +-- Аналог ручного POST /api/v1/admin/scrape/cian-backfill-history. +-- default_params соответствуют API-параметрам backfill_cian_history(): +-- batch_size — кол-во listing/house-строк за один run. +-- Если Cian-сессия истекла — run помечается failed + отправляется capture_message. +-- +-- rosreestr_dkp_import — 04:00-06:00 UTC (07:00-09:00 МСК) +-- Аналог ручного ./deploy/import-rosreestr.sh через FDW + pure-Python psycopg v3. +-- SINCE по умолчанию '2024-01-01' — резонное начало истории ДКП. +-- Idempotent: ON CONFLICT (dedup_hash) DO NOTHING. +-- +-- gendesign_rosreestr_deals — foreign table через gendesign_remote FDW server. +-- Колонки: только те, что используются импортёром (минимальный набор). +-- Партиционированная таблица на gendesign-стороне; FDW пушит WHERE вниз автоматически. + +BEGIN; + +-- ── Foreign table для rosreestr_deals ───────────────────────────────────────── +-- server gendesign_remote создан в 060_postgres_fdw_extension.sql. +-- USER MAPPING управляется backend startup (core/fdw.py, role tradein_fdw_reader). +-- tradein_fdw_reader должен иметь SELECT на rosreestr_deals в gendesign-БД. +-- Колонки подобраны под SELECT в import_rosreestr_dkp() в scheduler.py. + +DROP FOREIGN TABLE IF EXISTS gendesign_rosreestr_deals; + +CREATE FOREIGN TABLE gendesign_rosreestr_deals ( + id bigint, + region_code int, + city text, + street text, + realestate_type_code text, + doc_type text, + area numeric, + floor text, + year_build int, + deal_price numeric, + price_per_sqm numeric, + period_start_date date +) +SERVER gendesign_remote +OPTIONS (schema_name 'public', table_name 'rosreestr_deals'); + +COMMENT ON FOREIGN TABLE gendesign_rosreestr_deals IS + 'Live read из gendesign.rosreestr_deals для rosreestr_dkp_import (FDW, #563). ' + 'USER MAPPING управляется backend startup (core/fdw.py).'; + +-- ── Seed rows для scrape_schedules ──────────────────────────────────────────── + +-- next_run_at = следующее наступление night-window (tomorrow + window_start_hour), +-- иначе при NULL get_due_schedules() выстрелит сразу после деплоя (вне окна, +-- cian — на момент деплоя / до того как ляжет FDW-грант миграции 122). +-- window_*_hour документированы в UTC (см. 052) → AT TIME ZONE 'UTC' фиксирует +-- wall-clock час в UTC независимо от session TimeZone GUC. +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'cian_history_backfill', + true, + 2, + 5, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC', + '{"batch_size": 100}'::jsonb +), +( + 'rosreestr_dkp_import', + true, + 4, + 6, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 4)) AT TIME ZONE 'UTC', + '{"since": "2024-01-01", "batch_size": 2000}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (заменяет cron-script setup). ' + 'Sources: avito_city_sweep, cian_history_backfill, rosreestr_dkp_import.'; + +COMMIT;