3919a40c49
Объединяю несколько связанных изменений вокруг NSPD geo bulk-fetcher: Adapt to rosreestr2coord v5 API: - nspd_lite.fetch_via_rosreestr2coord: drop `delay` kwarg from Area() (removed upstream in v5); keep it in our function signature for backward-compat, comment why. - nspd_geo worker: add explicit time.sleep(rate_ms/1000) after lib-branch fetch — in v4 the library throttled internally via delay, in v5 rate limiting is the caller's job. Без этого получали Area.__init__() unexpected kwarg `delay` на каждом target. Drop use_rosreestr2coord switch: - Removed urllib-vs-lib choice everywhere. We always use community rosreestr2coord library — авторы регулярно обновляют WAF-tricks, наш urllib-fetcher (fetch_geoportal) уже неактуален. - admin_scrape.py: Pydantic schema, INSERT, SELECT, API response cleaned of `use_rosreestr2coord`. - nspd_geo.enqueue_geo_job: param dropped, INSERT shrunk. - worker process loop: dropped `if use_lib:` branch + import of fetch_geoportal. - frontend/geo/page.tsx: removed checkbox + GeoJob.use_rosreestr2coord field + POST body field. DB column drop: - data/sql/78_drop_use_rosreestr2coord.sql (NEW): DROP COLUMN nspd_geo_jobs.use_rosreestr2coord + CREATE OR REPLACE VIEW v_scrape_runs_unified (которая depended on the column). - data/sql/77_nspd_geo_jobs.sql: cleaned historical DDL for fresh setups. - Migration applied to prod (in-conversation via postgres MCP). Frontend polish: - Thematic ID changed from free-form number input to labeled select (1=parcel / 2=quarter / 4=admin / 5=building / 7=zone / 15=complex). - Auto-sync thematic_id from Job kind on change (override possible). - ScrapeLogsPanel: extended union type with "nspd_geo" + fixed /admin/scrape/geo to pass scraperType="nspd_geo" (was "nspd", filtering empty legacy nspd_scrape_log table; real logs live in nspd_geo_log via v_scrape_log_unified). Verified: ruff ✓, tsc --noEmit ✓, migration ran (BEGIN..COMMIT clean). Deploy order safe: prod column уже удалена → новый backend код, который не INSERT'ит use_rosreestr2coord, совпадёт со схемой после deploy.
490 lines
18 KiB
Python
490 lines
18 KiB
Python
"""Bulk NSPD geo backfill — resume-friendly Celery task.
|
||
|
||
Архитектура (см. data/sql/77_nspd_geo_jobs.sql):
|
||
nspd_geo_jobs — журнал заданий (один job = одна bulk-операция)
|
||
nspd_geo_targets — список cad-номеров со status (pending/done/failed)
|
||
|
||
Resume:
|
||
1. При worker_ready signal: ищем jobs со status='running' и stale heartbeat
|
||
(>10 мин), переключаем их статус на 'resume_needed' + re-enqueue task.
|
||
2. process_nspd_geo_job(job_id) сразу читает SELECT WHERE status='pending'
|
||
— идемпотентно продолжает с того места, где остановился (UPSERT в
|
||
cad_quarters_geom / cad_buildings гарантирует no dup'ов).
|
||
|
||
Запуск (3 пути):
|
||
1. UI: POST /api/v1/admin/scrape/geo (создаёт job, наполняет targets, enqueue task)
|
||
2. CLI: python -m app.workers.tasks.nspd_geo --enqueue-region 66
|
||
3. Beat: можно добавить cron-расписание в celery_app.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.scrapers.nspd_lite import (
|
||
NspdLiteError,
|
||
NspdLiteWafError,
|
||
fetch_via_rosreestr2coord,
|
||
)
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Heartbeat каждые N items (компромисс: чаще = выше DB write, реже = stale-window больше)
|
||
HEARTBEAT_EVERY = 5
|
||
# WAF backoff (seconds) после первого 403/429
|
||
WAF_BACKOFF_BASE_S = 30
|
||
|
||
|
||
# ── Helpers для job/target lifecycle ────────────────────────────────────────
|
||
|
||
|
||
def _log(
|
||
db: Session, job_id: int, level: str, stage: str, message: str, cad_num: str | None = None
|
||
) -> None:
|
||
"""Запись в nspd_geo_log (виден через v_scrape_log_unified в admin UI)."""
|
||
try:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO nspd_geo_log (job_id, level, stage, cad_num, message)
|
||
VALUES (:jid, :lvl, :stg, :cad, :msg)
|
||
"""
|
||
),
|
||
{"jid": job_id, "lvl": level, "stg": stage, "cad": cad_num, "msg": message[:1000]},
|
||
)
|
||
db.commit()
|
||
except Exception as e:
|
||
logger.warning("nspd_geo_log write failed for job=%s: %s", job_id, e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _start_job(db: Session, job_id: int) -> None:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE nspd_geo_jobs
|
||
SET status = 'running',
|
||
started_at = COALESCE(started_at, NOW()),
|
||
heartbeat_at = NOW(),
|
||
updated_at = NOW()
|
||
WHERE job_id = :id
|
||
"""
|
||
),
|
||
{"id": job_id},
|
||
)
|
||
db.commit()
|
||
_log(db, job_id, "info", "start", "job started/resumed")
|
||
|
||
|
||
def _heartbeat(db: Session, job_id: int, **counts: int) -> None:
|
||
sets = ["heartbeat_at = NOW()", "updated_at = NOW()"]
|
||
params: dict[str, Any] = {"id": job_id}
|
||
for k, v in counts.items():
|
||
sets.append(f"{k} = :{k}")
|
||
params[k] = v
|
||
try:
|
||
db.execute(
|
||
text(f"UPDATE nspd_geo_jobs SET {', '.join(sets)} WHERE job_id = :id"),
|
||
params,
|
||
)
|
||
db.commit()
|
||
except Exception as e:
|
||
logger.warning("heartbeat failed for job=%s: %s", job_id, e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _finish_job(db: Session, job_id: int, status: str, error: str | None = None) -> None:
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE nspd_geo_jobs
|
||
SET status = :st, finished_at = NOW(), error = :err, updated_at = NOW()
|
||
WHERE job_id = :id
|
||
"""
|
||
),
|
||
{"id": job_id, "st": status, "err": error},
|
||
)
|
||
db.commit()
|
||
|
||
|
||
# ── Persistence для разных видов targets ────────────────────────────────────
|
||
|
||
|
||
def _save_quarter(db: Session, payload: dict, cad_num: str) -> int:
|
||
"""Парсит NSPD-response для квартала и UPSERT'ит в cad_quarters_geom."""
|
||
feats = (payload.get("data") or {}).get("features") or []
|
||
if not feats:
|
||
return 0
|
||
f = feats[0]
|
||
geom_geojson = f.get("geometry")
|
||
if not geom_geojson:
|
||
return 0
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, fetched_at)
|
||
VALUES (:cad, ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326),
|
||
CAST(:props AS jsonb), NOW())
|
||
ON CONFLICT (cad_number) DO UPDATE
|
||
SET geom = EXCLUDED.geom,
|
||
raw_props = EXCLUDED.raw_props,
|
||
fetched_at = NOW()
|
||
"""
|
||
),
|
||
{
|
||
"cad": cad_num,
|
||
"g": str(geom_geojson).replace("'", '"'),
|
||
"props": str(f.get("properties", {})).replace("'", '"'),
|
||
},
|
||
)
|
||
return 1
|
||
|
||
|
||
def _save_building(db: Session, payload: dict, cad_num: str) -> int:
|
||
"""Парсит NSPD-response для здания и UPSERT'ит в cad_buildings."""
|
||
feats = (payload.get("data") or {}).get("features") or []
|
||
if not feats:
|
||
return 0
|
||
n = 0
|
||
for f in feats:
|
||
geom_geojson = f.get("geometry")
|
||
opts = (f.get("properties") or {}).get("options") or {}
|
||
if not geom_geojson:
|
||
continue
|
||
# Только этого здания (NSPD может вернуть здания всего квартала)
|
||
if opts.get("cad_num") != cad_num:
|
||
continue
|
||
import json as _json
|
||
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO cad_buildings (
|
||
cad_num, quarter_cad_num, geom, purpose, building_name,
|
||
readable_address, area, floors, raw_props
|
||
) VALUES (
|
||
:cad, :qcad,
|
||
ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326),
|
||
:purpose, :name, :addr, :area, :floors, CAST(:props AS jsonb)
|
||
)
|
||
ON CONFLICT (cad_num) DO UPDATE
|
||
SET quarter_cad_num = EXCLUDED.quarter_cad_num,
|
||
geom = EXCLUDED.geom,
|
||
purpose = EXCLUDED.purpose,
|
||
building_name = EXCLUDED.building_name,
|
||
readable_address = EXCLUDED.readable_address,
|
||
area = EXCLUDED.area,
|
||
floors = EXCLUDED.floors,
|
||
raw_props = EXCLUDED.raw_props
|
||
"""
|
||
),
|
||
{
|
||
"cad": cad_num,
|
||
"qcad": opts.get("quarter_cad_number"),
|
||
"g": _json.dumps(geom_geojson, ensure_ascii=False),
|
||
"purpose": opts.get("purpose"),
|
||
"name": opts.get("building_name") or opts.get("name"),
|
||
"addr": opts.get("readable_address"),
|
||
"area": opts.get("area"),
|
||
"floors": opts.get("floors"),
|
||
"props": _json.dumps(opts, ensure_ascii=False),
|
||
},
|
||
)
|
||
n += 1
|
||
return n
|
||
|
||
|
||
def _persist_target(db: Session, thematic_id: int, payload: dict, cad_num: str) -> tuple[int, str]:
|
||
"""Returns (n_features_saved, table_name). Полиморфно по thematic_id."""
|
||
if thematic_id == 2:
|
||
return _save_quarter(db, payload, cad_num), "cad_quarters_geom"
|
||
if thematic_id == 5:
|
||
return _save_building(db, payload, cad_num), "cad_buildings"
|
||
# Для parcel (1) пока нет нашей normalized-таблицы — только raw в targets.result_features
|
||
return 0, ""
|
||
|
||
|
||
# ── Helper для создания job + targets ──────────────────────────────────────
|
||
|
||
|
||
def enqueue_geo_job(
|
||
*,
|
||
name: str | None,
|
||
job_kind: str,
|
||
source_kind: str,
|
||
source_params: dict[str, Any],
|
||
cad_nums_with_thematic: list[tuple[str, int]],
|
||
rate_ms: int = 600,
|
||
triggered_by: str = "manual",
|
||
) -> int:
|
||
"""Создаёт job + targets и возвращает job_id. Не enqueue'ит task сам.
|
||
Caller должен сделать `process_nspd_geo_job.apply_async(args=[job_id])`.
|
||
"""
|
||
import json as _json
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
job_id = db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO nspd_geo_jobs (
|
||
name, job_kind, source_kind, source_params,
|
||
rate_ms, triggered_by, status, targets_total
|
||
) VALUES (:n, :k, :sk, CAST(:sp AS jsonb), :r, :tb, 'queued', :tt)
|
||
RETURNING job_id
|
||
"""
|
||
),
|
||
{
|
||
"n": name,
|
||
"k": job_kind,
|
||
"sk": source_kind,
|
||
"sp": _json.dumps(source_params, ensure_ascii=False),
|
||
"r": rate_ms,
|
||
"tb": triggered_by,
|
||
"tt": len(cad_nums_with_thematic),
|
||
},
|
||
).scalar_one()
|
||
|
||
# Bulk INSERT targets
|
||
if cad_nums_with_thematic:
|
||
values = ",".join(
|
||
f"({job_id}, '{c.replace(chr(39), chr(39)*2)}', {t}, 'pending')"
|
||
for c, t in cad_nums_with_thematic
|
||
)
|
||
db.execute(
|
||
text(
|
||
f"INSERT INTO nspd_geo_targets (job_id, cad_num, thematic_id, status) "
|
||
f"VALUES {values} ON CONFLICT (job_id, cad_num, thematic_id) DO NOTHING"
|
||
)
|
||
)
|
||
db.commit()
|
||
return int(job_id)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# ── Main task ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
@celery_app.task(
|
||
bind=True,
|
||
name="tasks.nspd_geo.process_job",
|
||
max_retries=0, # resume через worker_ready, не через retry
|
||
soft_time_limit=3600 * 6, # 6 часов max — крупные jobs
|
||
)
|
||
def process_nspd_geo_job(self: Any, job_id: int) -> dict[str, Any]:
|
||
"""Идёт по nspd_geo_targets WHERE job_id=:id AND status='pending'.
|
||
|
||
Heartbeat каждые HEARTBEAT_EVERY items, persistent counters в job-row.
|
||
Idempotent: уже-сохранённые UPSERT'ятся, status target → 'done'.
|
||
"""
|
||
import time
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
# Загружаем параметры job'а
|
||
job = (
|
||
db.execute(
|
||
text("SELECT * FROM nspd_geo_jobs WHERE job_id = :id"),
|
||
{"id": job_id},
|
||
)
|
||
.mappings()
|
||
.first()
|
||
)
|
||
if not job:
|
||
return {"error": "job_not_found", "job_id": job_id}
|
||
|
||
_start_job(db, job_id)
|
||
rate_ms = int(job["rate_ms"] or 600)
|
||
|
||
# Текущие счётчики (для resume — продолжаем с тех значений)
|
||
done = int(job["targets_done"] or 0)
|
||
failed = int(job["targets_failed"] or 0)
|
||
skipped = int(job["targets_skipped"] or 0)
|
||
n_requests = int(job["requests_count"] or 0)
|
||
n_waf = int(job["waf_blocked_count"] or 0)
|
||
consecutive_waf = 0
|
||
|
||
logger.info("process_nspd_geo_job start: job=%s rate=%dms", job_id, rate_ms)
|
||
|
||
while True:
|
||
target = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT target_id, cad_num, thematic_id, attempts
|
||
FROM nspd_geo_targets
|
||
WHERE job_id = :id AND status = 'pending'
|
||
ORDER BY target_id
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"id": job_id},
|
||
)
|
||
.mappings()
|
||
.first()
|
||
)
|
||
if not target:
|
||
break
|
||
|
||
tid = target["thematic_id"]
|
||
cad = target["cad_num"]
|
||
|
||
try:
|
||
# NSPD geo fetch теперь только через rosreestr2coord lib (v5+).
|
||
# Старая ветка с urllib-fetch (fetch_geoportal) удалена — community
|
||
# пакет нам надёжнее, авторы обновляют под WAF.
|
||
payload = {
|
||
"data": {
|
||
"features": [fetch_via_rosreestr2coord(cad, area_type=tid, timeout=15)]
|
||
}
|
||
}
|
||
payload["data"]["features"] = [f for f in payload["data"]["features"] if f]
|
||
# rosreestr2coord v5 убрал internal delay-параметр → rate limit
|
||
# делаем здесь, чтобы не словить WAF за высокую частоту.
|
||
if rate_ms > 0:
|
||
time.sleep(rate_ms / 1000.0)
|
||
n_requests += 1
|
||
consecutive_waf = 0
|
||
|
||
n_feats, saved_to = _persist_target(db, tid, payload, cad)
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE nspd_geo_targets
|
||
SET status = 'done', result_features = :f,
|
||
saved_to_table = :s, fetched_at = NOW(),
|
||
attempts = attempts + 1
|
||
WHERE target_id = :tid
|
||
"""
|
||
),
|
||
{"f": n_feats, "s": saved_to or None, "tid": target["target_id"]},
|
||
)
|
||
db.commit()
|
||
done += 1
|
||
|
||
except NspdLiteWafError as e:
|
||
consecutive_waf += 1
|
||
n_waf += 1
|
||
wait = WAF_BACKOFF_BASE_S * min(consecutive_waf, 8)
|
||
logger.warning(
|
||
"WAF on %s (consecutive=%d), backoff %ds: %s",
|
||
cad,
|
||
consecutive_waf,
|
||
wait,
|
||
e,
|
||
)
|
||
# НЕ помечаем failed — остаётся pending для следующей итерации
|
||
_heartbeat(db, job_id, requests_count=n_requests, waf_blocked_count=n_waf)
|
||
_log(
|
||
db,
|
||
job_id,
|
||
"warn",
|
||
"waf",
|
||
f"WAF/429 (consecutive={consecutive_waf}), backoff {wait}s",
|
||
cad_num=cad,
|
||
)
|
||
time.sleep(wait)
|
||
if consecutive_waf >= 8:
|
||
# 8 подряд WAF — пауза job'а до следующего worker_ready
|
||
_finish_job(
|
||
db,
|
||
job_id,
|
||
"paused",
|
||
error=f"WAF blocked {consecutive_waf} consecutive",
|
||
)
|
||
_log(
|
||
db,
|
||
job_id,
|
||
"error",
|
||
"paused",
|
||
f"job paused after {consecutive_waf} consecutive WAF blocks",
|
||
)
|
||
return {"job_id": job_id, "paused": True, "reason": "waf"}
|
||
continue
|
||
|
||
except (NspdLiteError, Exception) as e:
|
||
logger.warning("fetch failed for %s: %s", cad, e)
|
||
_log(db, job_id, "warn", "fetch_fail", str(e)[:300], cad_num=cad)
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE nspd_geo_targets
|
||
SET status = CASE WHEN attempts >= 2 THEN 'failed' ELSE 'pending' END,
|
||
attempts = attempts + 1,
|
||
error_msg = :em
|
||
WHERE target_id = :tid
|
||
"""
|
||
),
|
||
{"em": str(e)[:500], "tid": target["target_id"]},
|
||
)
|
||
db.commit()
|
||
if int(target["attempts"]) >= 2:
|
||
failed += 1
|
||
|
||
if (done + failed + skipped) % HEARTBEAT_EVERY == 0:
|
||
_heartbeat(
|
||
db,
|
||
job_id,
|
||
targets_done=done,
|
||
targets_failed=failed,
|
||
targets_skipped=skipped,
|
||
requests_count=n_requests,
|
||
waf_blocked_count=n_waf,
|
||
)
|
||
|
||
_heartbeat(
|
||
db,
|
||
job_id,
|
||
targets_done=done,
|
||
targets_failed=failed,
|
||
targets_skipped=skipped,
|
||
requests_count=n_requests,
|
||
waf_blocked_count=n_waf,
|
||
)
|
||
# done даже при failed targets — UI покажет failed_count отдельно
|
||
_finish_job(db, job_id, "done")
|
||
_log(
|
||
db,
|
||
job_id,
|
||
"info",
|
||
"done",
|
||
f"done: {done} ok / {failed} failed / {n_requests} req / {n_waf} WAF blocks",
|
||
)
|
||
logger.info(
|
||
"process_nspd_geo_job done: job=%s done=%d failed=%d req=%d waf=%d",
|
||
job_id,
|
||
done,
|
||
failed,
|
||
n_requests,
|
||
n_waf,
|
||
)
|
||
return {
|
||
"job_id": job_id,
|
||
"done": done,
|
||
"failed": failed,
|
||
"skipped": skipped,
|
||
"requests": n_requests,
|
||
"waf_blocked": n_waf,
|
||
}
|
||
except Exception as e:
|
||
logger.exception("process_nspd_geo_job crashed: job=%s: %s", job_id, e)
|
||
try:
|
||
_finish_job(db, job_id, "failed", error=f"{type(e).__name__}: {e}")
|
||
except Exception:
|
||
pass
|
||
raise
|
||
finally:
|
||
db.close()
|