add log scrapper

backend/app/services/scrapers/domrf_kn.py

@@ -920,6 +920,29 @@ def log_progress(
             pass
 
 
+def _checkpoint(db: Session, run_id: int, progress_index: int) -> None:
+    """Записать прогресс + heartbeat. Вызывается раз в N объектов и в конце."""
+    try:
+        db.execute(
+            text(
+                """
+                UPDATE kn_scrape_runs
+                   SET progress_obj_index = :idx,
+                       heartbeat_at = NOW()
+                 WHERE run_id = :rid
+                """
+            ),
+            {"idx": progress_index, "rid": run_id},
+        )
+        db.commit()
+    except Exception as e:
+        logger.warning("checkpoint failed run=%s: %s", run_id, e)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+
+
 async def run_region_sweep(
     region_code: int,
     developers: list[str] | None = None,
@@ -933,38 +956,137 @@ async def run_region_sweep(
     download_photos_binary: bool = False,
     photos_dir: Path | str | None = None,
     place_override: str | None = None,
+    resume_from_run_id: int | None = None,
 ) -> dict[str, Any]:
-    """Full sweep for one region: bootstrap → fetch objects (each objStatus) → fetch flats.
+    """Full sweep for one region: bootstrap → fetch objects → per-object processing.
 
-    `developers` filters the OBJECT set locally by `developer.companyGroup` matching the
+    Resume-aware: если `resume_from_run_id` задан, читаем objects_snapshot и
-    "<group>_0" id pattern (since DOM.РФ doesn't honour developer filter on the API).
+    progress_obj_index из исходного run и продолжаем с этого индекса. Новый
+    run_id создаётся (с resumed_from_run_id ссылкой), исходный остаётся
+    'zombie' для аудита. Heartbeat обновляется каждые 10 объектов — старт-хук
+    worker'а ловит застывшие row >5 мин.
     """
     snapshot_date = snapshot_date or date.today()
     db = SessionLocal()
-    run_id = db.execute(
-        text(
-            """
-            INSERT INTO kn_scrape_runs (region_codes, developer_ids, snapshot_date, status)
-            VALUES (ARRAY[:rc]::int[], CAST(:devs AS text[]), :snap, 'running')
-            RETURNING run_id
-            """
-        ),
-        {
-            "rc": region_code,
-            "devs": developers if developers else None,
-            "snap": snapshot_date,
-        },
-    ).scalar_one()
-    db.commit()
 
+    # ── Init: либо новый run, либо resume ──────────────────────────────────
+    all_objects: list[dict[str, Any]] = []
+    start_index = 0
     place = place_override or _place_str(region_code)
-    log_progress(
+    params_dict: dict[str, Any] = {
-        db,
+        "extras": extras,
-        run_id,
+        "fetch_flats": fetch_flats,
-        f"Старт sweep региона {region_code} (place={place}, devs={developers or '*'},"
+        "download_photos": download_photos_binary,
-        f" extras={extras}, photos={download_photos_binary})",
+        "place": place,
-        stage="start",
+        "statuses": list(statuses),
-    )
+    }
+
+    if resume_from_run_id:
+        prev = (
+            db.execute(
+                text(
+                    """
+                    SELECT region_codes, developer_ids, snapshot_date, params,
+                           objects_snapshot,
+                           COALESCE(progress_obj_index, 0) AS idx
+                      FROM kn_scrape_runs
+                     WHERE run_id = :rid
+                    """
+                ),
+                {"rid": resume_from_run_id},
+            )
+            .mappings()
+            .first()
+        )
+        if not prev or not prev["objects_snapshot"]:
+            db.close()
+            raise RuntimeError(f"resume_from_run_id={resume_from_run_id}: нет objects_snapshot")
+        all_objects = list(prev["objects_snapshot"])
+        start_index = int(prev["idx"] or 0)
+        # Восстанавливаем параметры старого run'а — прелюдия может отличаться
+        # от вызывающего кода (например beat schedule поменял extras).
+        if prev["params"]:
+            saved = prev["params"]
+            extras = saved.get("extras", extras)
+            fetch_flats = saved.get("fetch_flats", fetch_flats)
+            download_photos_binary = saved.get("download_photos", download_photos_binary)
+        snapshot_date = prev["snapshot_date"] or snapshot_date
+
+        run_id = db.execute(
+            text(
+                """
+                INSERT INTO kn_scrape_runs (
+                    region_codes, developer_ids, snapshot_date, status,
+                    params, objects_snapshot, progress_obj_index,
+                    total_obj_count, heartbeat_at, resumed_from_run_id
+                )
+                VALUES (
+                    :rc::int[], :devs::text[], :snap, 'running',
+                    :params::jsonb, :objs::jsonb, :idx,
+                    :total, NOW(), :prev
+                )
+                RETURNING run_id
+                """
+            ),
+            {
+                "rc": list(prev["region_codes"]),
+                "devs": list(prev["developer_ids"]) if prev["developer_ids"] else None,
+                "snap": snapshot_date,
+                "params": json.dumps(prev["params"] or params_dict),
+                "objs": json.dumps(all_objects),
+                "idx": start_index,
+                "total": len(all_objects),
+                "prev": resume_from_run_id,
+            },
+        ).scalar_one()
+        db.commit()
+        log_progress(
+            db,
+            run_id,
+            f"♻ Resume from index {start_index}/{len(all_objects)}"
+            f" (предыдущий run #{resume_from_run_id})",
+            stage="resume",
+        )
+    else:
+        run_id = db.execute(
+            text(
+                """
+                INSERT INTO kn_scrape_runs (
+                    region_codes, developer_ids, snapshot_date, status,
+                    params, heartbeat_at
+                )
+                VALUES (
+                    ARRAY[:rc]::int[], CAST(:devs AS text[]), :snap, 'running',
+                    :params::jsonb, NOW()
+                )
+                RETURNING run_id
+                """
+            ),
+            {
+                "rc": region_code,
+                "devs": developers if developers else None,
+                "snap": snapshot_date,
+                "params": json.dumps(params_dict),
+            },
+        ).scalar_one()
+        db.commit()
+        log_progress(
+            db,
+            run_id,
+            f"Старт sweep региона {region_code} (place={place}, devs={developers or '*'},"
+            f" extras={extras}, photos={download_photos_binary})",
+            stage="start",
+        )
+
+    extras_counts = {
+        "sale_graph_rows": 0,
+        "sales_agg_rows": 0,
+        "infra_rows": 0,
+        "photos_rows": 0,
+        "photos_downloaded": 0,
+    }
+    total_flats = 0
+    request_count = 0
 
     try:
         async with BrowserSession(
@@ -973,88 +1095,94 @@ async def run_region_sweep(
             load_state=load_state,
             save_state=save_state,
         ) as sess:
-            # --- objects across statuses -------------------------------------
+            # ── Phase A — fetch objects (skip on resume) ────────────────────
-            all_objects: list[dict[str, Any]] = []
+            if not resume_from_run_id:
-            for status in statuses:
+                for status in statuses:
-                rows = await fetch_objects_for_status(sess, place, status)
+                    rows = await fetch_objects_for_status(sess, place, status)
-                all_objects.extend(rows)
+                    all_objects.extend(rows)
+                    log_progress(
+                        db,
+                        run_id,
+                        f"objStatus={status}: получено {len(rows)} объектов",
+                        stage="fetch_objects",
+                    )
+                logger.info(
+                    "place=%s total objects across %s = %d",
+                    place,
+                    statuses,
+                    len(all_objects),
+                )
+                if developers:
+                    wanted_groups = {int(d.split("_")[0]) for d in developers if "_" in d}
+                    all_objects = [
+                        o
+                        for o in all_objects
+                        if isinstance(o.get("developer"), dict)
+                        and o["developer"].get("companyGroup") in wanted_groups
+                    ]
+                    log_progress(
+                        db,
+                        run_id,
+                        f"После фильтра devs={developers}: {len(all_objects)} объектов",
+                        stage="filter_devs",
+                    )
+
+                # Phase A commit: upsert_objects + raw + objects_snapshot.
+                # Это критично для resume — при падении ниже worker сможет
+                # подхватить с правильным списком объектов и кадастровым snapshot'ом.
+                obj_count = upsert_objects(db, all_objects, snapshot_date, region_code)
+                _insert_raw(db, snapshot_date, f"kn_object_place_{place}", all_objects)
+                db.execute(
+                    text(
+                        """
+                        UPDATE kn_scrape_runs
+                           SET objects_snapshot = :objs::jsonb,
+                               total_obj_count  = :total,
+                               objects_count    = :total,
+                               heartbeat_at     = NOW()
+                         WHERE run_id = :rid
+                        """
+                    ),
+                    {
+                        "objs": json.dumps(all_objects),
+                        "total": len(all_objects),
+                        "rid": run_id,
+                    },
+                )
+                db.commit()
                 log_progress(
                     db,
                     run_id,
-                    f"objStatus={status}: получено {len(rows)} объектов",
+                    f"Phase A done: {obj_count} объектов в БД, snapshot закреплён",
-                    stage="fetch_objects",
+                    stage="phase_a",
-                )
-            logger.info("place=%s total objects across %s = %d", place, statuses, len(all_objects))
-            log_progress(
-                db,
-                run_id,
-                f"Всего объектов: {len(all_objects)} (по статусам {statuses})",
-                stage="fetch_objects",
-            )
-
-            # local developer filter
-            if developers:
-                wanted_groups = {int(d.split("_")[0]) for d in developers if "_" in d}
-                all_objects = [
-                    o
-                    for o in all_objects
-                    if isinstance(o.get("developer"), dict)
-                    and o["developer"].get("companyGroup") in wanted_groups
-                ]
-                logger.info("after developers filter: %d objects", len(all_objects))
-                log_progress(
-                    db,
-                    run_id,
-                    f"После фильтра devs={developers}: {len(all_objects)} объектов",
-                    stage="filter_devs",
                 )
 
-            # --- flats per object --------------------------------------------
+            # ── Phase B/C — per-object processing (resumable) ───────────────
-            all_flats: list[dict[str, Any]] = []
+            pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT
-            if fetch_flats:
+            total = len(all_objects)
-                for i, obj in enumerate(all_objects):
+            for i in range(start_index, total):
-                    obj_id = obj.get("objId") or obj.get("obj_id")
+                obj = all_objects[i]
-                    if not obj_id:
+                obj_id = obj.get("objId") or obj.get("obj_id")
-                        continue
+                if not obj_id:
-                    flats = await fetch_flats_for_object(sess, obj_id)
+                    continue
-                    all_flats.extend(flats)
+
-                    if (i + 1) % 25 == 0:
+                # Flats per obj — committed immediately
-                        logger.info(
+                if fetch_flats:
-                            "flats progress: %d/%d objs, %d flats so far",
+                    try:
-                            i + 1,
+                        flats = await fetch_flats_for_object(sess, obj_id)
-                            len(all_objects),
+                        if flats:
-                            len(all_flats),
+                            total_flats += upsert_flats(db, flats, snapshot_date, region_code)
-                        )
+                    except Exception as e:
                         log_progress(
                             db,
                             run_id,
-                            f"Квартиры: {i + 1}/{len(all_objects)} объектов,"
+                            f"flats failed obj={obj_id}: {type(e).__name__}: {str(e)[:120]}",
-                            f" {len(all_flats)} квартир",
+                            level="warn",
                             stage="fetch_flats",
+                            obj_id=obj_id,
                         )
-                logger.info("total flats: %d (across %d objs)", len(all_flats), len(all_objects))
-                log_progress(
-                    db,
-                    run_id,
-                    f"Квартиры собраны: {len(all_flats)} шт по {len(all_objects)} объектам",
-                    stage="fetch_flats",
-                )
-
-            # --- extras (sale_graph, sales_agg, infra, photos) per object ----
-            extras_counts = {
-                "sale_graph_rows": 0,
-                "sales_agg_rows": 0,
-                "infra_rows": 0,
-                "photos_rows": 0,
-                "photos_downloaded": 0,
-            }
-            if extras:
-                pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT
-                for i, obj in enumerate(all_objects):
-                    obj_id = obj.get("objId") or obj.get("obj_id")
-                    if not obj_id:
-                        continue
 
+                if extras:
                     # sale_graph (apartments + parking)
                     for type_ in SALE_GRAPH_TYPES:
                         try:
@@ -1064,7 +1192,8 @@ async def run_region_sweep(
                             )
                         except Exception as e:
                             full_url = (
-                                f"{BASE_URL}{PATH_SALE_GRAPH.format(obj_id=obj_id)}?type={type_}"
+                                f"{BASE_URL}"
+                                f"{PATH_SALE_GRAPH.format(obj_id=obj_id)}?type={type_}"
                             )
                             _classify_and_log(
                                 db, run_id, obj_id, f"sale_graph_{type_}", full_url, e
@@ -1090,7 +1219,7 @@ async def run_region_sweep(
                         full_url = f"{BASE_URL}{PATH_INFRA.format(obj_id=obj_id)}"
                         _classify_and_log(db, run_id, obj_id, "infrastructure", full_url, e)
 
-                    # photos (metadata + optional binary)
+                    # photos
                     try:
                         photos, full_url = await fetch_photos(sess, obj_id)
                         local_paths: dict[str, str] = {}
@@ -1107,68 +1236,55 @@ async def run_region_sweep(
                         full_url = f"{BASE_URL}{PATH_PHOTOS.format(obj_id=obj_id)}"
                         _classify_and_log(db, run_id, obj_id, "photos", full_url, e)
 
-                    if (i + 1) % 10 == 0:
+                # Checkpoint раз в 10 объектов: запись прогресса + heartbeat.
-                        logger.info(
+                if (i + 1) % 10 == 0:
-                            "extras progress: %d/%d objs, %s",
+                    _checkpoint(db, run_id, i + 1)
-                            i + 1,
+                    log_progress(
-                            len(all_objects),
+                        db,
-                            extras_counts,
+                        run_id,
-                        )
+                        f"Progress: {i + 1}/{total} | flats={total_flats}"
-                        # Periodic commit so failures are visible in admin UI even mid-run
+                        f" sale_graph={extras_counts['sale_graph_rows']}"
-                        db.commit()
+                        f" agg={extras_counts['sales_agg_rows']}"
-                        log_progress(
+                        f" infra={extras_counts['infra_rows']}"
-                            db,
+                        f" photos={extras_counts['photos_rows']}"
-                            run_id,
+                        f" downloaded={extras_counts['photos_downloaded']}",
-                            f"Extras: {i + 1}/{len(all_objects)} объектов |"
+                        stage="extras" if extras else "fetch_flats",
-                            f" sale_graph={extras_counts['sale_graph_rows']}"
+                    )
-                            f" agg={extras_counts['sales_agg_rows']}"
-                            f" infra={extras_counts['infra_rows']}"
-                            f" photos={extras_counts['photos_rows']}"
-                            f" downloaded={extras_counts['photos_downloaded']}",
-                            stage="extras",
-                        )
-                logger.info("extras done: %s", extras_counts)
-                log_progress(
-                    db,
-                    run_id,
-                    f"Extras готовы: {extras_counts}",
-                    stage="extras",
-                )
 
+            _checkpoint(db, run_id, total)
             request_count = sess.request_count
 
-        # --- write all in one transaction ------------------------------------
+        # ── Phase D — finalize ─────────────────────────────────────────────
-        obj_count = upsert_objects(db, all_objects, snapshot_date, region_code)
-        flat_count = upsert_flats(db, all_flats, snapshot_date, region_code) if fetch_flats else 0
-        # Save raw payloads (one row per status; flats are huge so we store the count, not bodies).
-        _insert_raw(db, snapshot_date, f"kn_object_place_{place}", all_objects)
-
         db.execute(
             text(
                 """
                 UPDATE kn_scrape_runs
-                   SET finished_at=NOW(), objects_count=:oc, flats_count=:fc,
+                   SET finished_at  = NOW(),
-                       requests_count=:rq, status='done'
+                       flats_count  = :fc,
-                 WHERE run_id=:rid
+                       requests_count = :rq,
+                       status       = 'done',
+                       heartbeat_at = NOW()
+                 WHERE run_id = :rid
                 """
             ),
-            {"oc": obj_count, "fc": flat_count, "rq": request_count, "rid": run_id},
+            {"fc": total_flats, "rq": request_count, "rid": run_id},
         )
         db.commit()
         log_progress(
             db,
             run_id,
-            f"Готово ✅ objects={obj_count} flats={flat_count} requests={request_count}",
+            f"Готово ✅ objects={total} flats={total_flats} requests={request_count}",
             stage="done",
         )
         return {
             "run_id": run_id,
             "region_code": region_code,
             "place": place,
-            "objects": obj_count,
+            "objects": total,
-            "flats": flat_count,
+            "flats": total_flats,
             "requests": request_count,
             "snapshot_date": snapshot_date.isoformat(),
+            "resumed_from": resume_from_run_id,
             **extras_counts,
         }
     except Exception as e:

backend/app/workers/celery_app.py

@@ -1,10 +1,15 @@
 """Celery app + beat schedule."""
 
+import logging
+
 from celery import Celery
 from celery.schedules import crontab
+from celery.signals import worker_ready
 
 from app.core.config import settings
 
+logger = logging.getLogger(__name__)
+
 
 def _parse_cron(spec: str) -> crontab:
     """Parse 'M H DoM Mon DoW' crontab string into Celery crontab. Empty fields default to '*'."""
@@ -44,3 +49,76 @@ celery_app.conf.beat_schedule = {
     }
     for rc in _default_regions()
 }
+
+
+@worker_ready.connect
+def _resume_zombie_runs(sender=None, **_kwargs) -> None:
+    """When a worker finishes booting (after redeploy/restart), find any sweep
+    that was 'running' with a stale heartbeat (>5 min) and re-enqueue a resume
+    task. Each resume creates a fresh run_id linked via resumed_from_run_id;
+    the original row is marked 'zombie' so the audit trail is preserved.
+    """
+    from sqlalchemy import text
+
+    from app.core.db import SessionLocal
+
+    db = SessionLocal()
+    try:
+        rows = (
+            db.execute(
+                text(
+                    """
+                    SELECT run_id
+                      FROM kn_scrape_runs
+                     WHERE status = 'running'
+                       AND objects_snapshot IS NOT NULL
+                       AND COALESCE(heartbeat_at, started_at)
+                           < NOW() - INTERVAL '5 minutes'
+                     ORDER BY started_at ASC
+                     LIMIT 20
+                    """
+                )
+            )
+            .mappings()
+            .all()
+        )
+        if not rows:
+            logger.info("worker_ready: нет stale runs для resume")
+            return
+
+        ids = [int(r["run_id"]) for r in rows]
+        # Помечаем найденные как 'zombie' одним апдейтом — resume создаст новые
+        # run_id со ссылкой resumed_from_run_id.
+        db.execute(
+            text(
+                """
+                UPDATE kn_scrape_runs
+                   SET status = 'zombie',
+                       finished_at = NOW(),
+                       error = COALESCE(error,
+                           'auto-zombie at worker_ready, resume scheduled')
+                 WHERE run_id = ANY(:ids)
+                """
+            ),
+            {"ids": ids},
+        )
+        db.commit()
+    except Exception as e:
+        logger.exception("worker_ready resume scan failed: %s", e)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        return
+    finally:
+        db.close()
+
+    # Enqueue resume tasks. Late import to avoid circular at module load.
+    from app.workers.tasks.scrape_kn import resume_kn_run
+
+    for rid in ids:
+        try:
+            resume_kn_run.apply_async(args=[rid])
+            logger.info("worker_ready: resume_kn_run enqueued for run=%s", rid)
+        except Exception as e:
+            logger.warning("worker_ready: failed to enqueue resume for run=%s: %s", rid, e)

backend/app/workers/tasks/scrape_kn.py

@@ -102,3 +102,64 @@ def scrape_kn_region(
                 download_photos_binary=download_photos,
             )
         )
+
+
+@celery_app.task(bind=True, name="tasks.scrape_kn.resume_kn_run", max_retries=2)
+def resume_kn_run(self: Any, run_id: int) -> dict[str, Any]:
+    """Resume a previously interrupted sweep using objects_snapshot +
+    progress_obj_index from kn_scrape_runs. Triggered by worker_ready hook
+    when a stale 'running' row is found (heartbeat >5 min)."""
+    from sqlalchemy import text
+
+    from app.core.db import SessionLocal
+
+    state_path: str | None = settings.scrape_kn_state_path or None
+    if state_path and not Path(state_path).exists():
+        state_path = None
+
+    # Reconstruct region/devs from the stale row to scope the lock correctly.
+    db = SessionLocal()
+    try:
+        row = (
+            db.execute(
+                text(
+                    """
+                    SELECT region_codes, developer_ids
+                      FROM kn_scrape_runs
+                     WHERE run_id = :rid
+                    """
+                ),
+                {"rid": run_id},
+            )
+            .mappings()
+            .first()
+        )
+    finally:
+        db.close()
+    if not row:
+        return {"skipped": True, "reason": "run_not_found", "run_id": run_id}
+
+    region_codes = list(row["region_codes"] or [])
+    developers = list(row["developer_ids"]) if row["developer_ids"] else None
+    if not region_codes:
+        return {"skipped": True, "reason": "no_region", "run_id": run_id}
+    region_code = region_codes[0]
+
+    with _region_lock(region_code, developers) as got_lock:
+        if not got_lock:
+            logger.warning(
+                "resume_kn_run SKIPPED run=%s region=%s — lock held by another sweep",
+                run_id,
+                region_code,
+            )
+            return {"skipped": True, "reason": "lock_held", "run_id": run_id}
+
+        logger.info("resume_kn_run start run=%s region=%s devs=%s", run_id, region_code, developers)
+        return asyncio.run(
+            run_region_sweep(
+                region_code=region_code,
+                developers=developers,
+                load_state=state_path,
+                resume_from_run_id=run_id,
+            )
+        )

data/sql/55_schema_scrape_resume.sql

@@ -0,0 +1,17 @@
+-- Resume support for kn-API sweeps.
+-- When a worker dies (redeploy/SIGKILL/OOM), the run row stays 'running'.
+-- Worker startup hook finds rows with stale heartbeat (>5 min) and re-enqueues
+-- a resume task that picks up from progress_obj_index using objects_snapshot.
+-- Idempotent.
+
+ALTER TABLE kn_scrape_runs
+    ADD COLUMN IF NOT EXISTS heartbeat_at         TIMESTAMPTZ,
+    ADD COLUMN IF NOT EXISTS progress_obj_index   INT,
+    ADD COLUMN IF NOT EXISTS total_obj_count      INT,
+    ADD COLUMN IF NOT EXISTS objects_snapshot     JSONB,
+    ADD COLUMN IF NOT EXISTS params               JSONB,
+    ADD COLUMN IF NOT EXISTS resumed_from_run_id  BIGINT REFERENCES kn_scrape_runs(run_id);
+
+CREATE INDEX IF NOT EXISTS idx_kn_scrape_runs_running_heartbeat
+    ON kn_scrape_runs (heartbeat_at)
+    WHERE status IN ('running', 'resuming');