diff --git a/backend/app/services/scrapers/domrf_kn.py b/backend/app/services/scrapers/domrf_kn.py index ed2829be..58577898 100644 --- a/backend/app/services/scrapers/domrf_kn.py +++ b/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 - "_0" id pattern (since DOM.РФ doesn't honour developer filter on the API). + Resume-aware: если `resume_from_run_id` задан, читаем objects_snapshot и + 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( - db, - run_id, - f"Старт sweep региона {region_code} (place={place}, devs={developers or '*'}," - f" extras={extras}, photos={download_photos_binary})", - stage="start", - ) + params_dict: dict[str, Any] = { + "extras": extras, + "fetch_flats": fetch_flats, + "download_photos": download_photos_binary, + "place": place, + "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 ------------------------------------- - all_objects: list[dict[str, Any]] = [] - for status in statuses: - rows = await fetch_objects_for_status(sess, place, status) - all_objects.extend(rows) + # ── Phase A — fetch objects (skip on resume) ──────────────────── + if not resume_from_run_id: + for status in statuses: + rows = await fetch_objects_for_status(sess, place, status) + 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)} объектов", - stage="fetch_objects", - ) - 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", + f"Phase A done: {obj_count} объектов в БД, snapshot закреплён", + stage="phase_a", ) - # --- flats per object -------------------------------------------- - all_flats: list[dict[str, Any]] = [] - if fetch_flats: - for i, obj in enumerate(all_objects): - obj_id = obj.get("objId") or obj.get("obj_id") - if not obj_id: - continue - flats = await fetch_flats_for_object(sess, obj_id) - all_flats.extend(flats) - if (i + 1) % 25 == 0: - logger.info( - "flats progress: %d/%d objs, %d flats so far", - i + 1, - len(all_objects), - len(all_flats), - ) + # ── Phase B/C — per-object processing (resumable) ─────────────── + pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT + total = len(all_objects) + for i in range(start_index, total): + obj = all_objects[i] + obj_id = obj.get("objId") or obj.get("obj_id") + if not obj_id: + continue + + # Flats per obj — committed immediately + if fetch_flats: + try: + flats = await fetch_flats_for_object(sess, obj_id) + if 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" {len(all_flats)} квартир", + f"flats failed obj={obj_id}: {type(e).__name__}: {str(e)[:120]}", + 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: - logger.info( - "extras progress: %d/%d objs, %s", - i + 1, - len(all_objects), - extras_counts, - ) - # Periodic commit so failures are visible in admin UI even mid-run - db.commit() - log_progress( - db, - run_id, - f"Extras: {i + 1}/{len(all_objects)} объектов |" - 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 раз в 10 объектов: запись прогресса + heartbeat. + if (i + 1) % 10 == 0: + _checkpoint(db, run_id, i + 1) + log_progress( + db, + run_id, + f"Progress: {i + 1}/{total} | flats={total_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" if extras else "fetch_flats", + ) + _checkpoint(db, run_id, total) request_count = sess.request_count - # --- write all in one transaction ------------------------------------ - 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) - + # ── Phase D — finalize ───────────────────────────────────────────── db.execute( text( """ UPDATE kn_scrape_runs - SET finished_at=NOW(), objects_count=:oc, flats_count=:fc, - requests_count=:rq, status='done' - WHERE run_id=:rid + SET finished_at = NOW(), + flats_count = :fc, + 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, - "flats": flat_count, + "objects": total, + "flats": total_flats, "requests": request_count, "snapshot_date": snapshot_date.isoformat(), + "resumed_from": resume_from_run_id, **extras_counts, } except Exception as e: diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 0d5c6646..b3d6b411 100644 --- a/backend/app/workers/celery_app.py +++ b/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) diff --git a/backend/app/workers/tasks/scrape_kn.py b/backend/app/workers/tasks/scrape_kn.py index 1a55e783..c3e75f37 100644 --- a/backend/app/workers/tasks/scrape_kn.py +++ b/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, + ) + ) diff --git a/data/sql/55_schema_scrape_resume.sql b/data/sql/55_schema_scrape_resume.sql new file mode 100644 index 00000000..08524406 --- /dev/null +++ b/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');