diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 1edd2bd9..363ac731 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -358,6 +358,214 @@ def list_logs( ] +# ── NSPD (cad_quarters / cad_buildings) ───────────────────────────────────── + + +class TriggerNspdRequest(BaseModel): + region_code: int = Field(default=66, ge=1, le=99) + limit: int | None = Field(default=None, ge=1, le=2000) + rate_ms: int = Field(default=600, ge=200, le=5000) + force: bool = False + + +@router.post("/nspd") +def trigger_nspd_sweep( + payload: TriggerNspdRequest, + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + _check_token(x_admin_token) + from app.workers.tasks.scrape_nspd import ( + force_release_lock as nspd_force_release_lock, + ) + from app.workers.tasks.scrape_nspd import ( + scrape_nspd_region, + ) + + lock_released = False + if payload.force: + lock_released = nspd_force_release_lock(payload.region_code) + + result = scrape_nspd_region.apply_async( + kwargs={ + "region_code": payload.region_code, + "triggered_by": "manual", + "limit": payload.limit, + "rate_ms": payload.rate_ms, + }, + ) + return { + "task_id": result.id, + "region_code": payload.region_code, + "limit": payload.limit, + "rate_ms": payload.rate_ms, + "force": payload.force, + "lock_was_released": lock_released, + } + + +class NspdReleaseLockRequest(BaseModel): + region_code: int = Field(..., ge=1, le=99) + + +@router.post("/nspd/release-lock") +def nspd_release_lock( + payload: NspdReleaseLockRequest, + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + _check_token(x_admin_token) + from app.workers.tasks.scrape_nspd import force_release_lock as nspd_force_release_lock + + released = nspd_force_release_lock(payload.region_code) + return {"region_code": payload.region_code, "released": released} + + +@router.get("/nspd/runs") +def list_nspd_runs( + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + limit: int = 20, +) -> list[dict[str, Any]]: + _check_token(x_admin_token) + rows = ( + db.execute( + text( + """ + SELECT run_id, started_at, finished_at, heartbeat_at, + region_code, pending_count, + quarters_ok, quarters_failed, buildings_ok, + requests_count, waf_429_count, + status, error, triggered_by + FROM nspd_scrape_runs + ORDER BY started_at DESC + LIMIT :lim + """ + ), + {"lim": min(limit, 100)}, + ) + .mappings() + .all() + ) + return [ + { + "run_id": r["run_id"], + "started_at": r["started_at"].isoformat() if r["started_at"] else None, + "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None, + "heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None, + "region_code": r["region_code"], + "pending_count": r["pending_count"], + "quarters_ok": r["quarters_ok"], + "quarters_failed": r["quarters_failed"], + "buildings_ok": r["buildings_ok"], + "requests_count": r["requests_count"], + "waf_429_count": r["waf_429_count"], + "status": r["status"], + "error": r["error"], + "triggered_by": r["triggered_by"], + } + for r in rows + ] + + +@router.get("/nspd/logs") +def list_nspd_logs( + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + run_id: int | None = None, + since_id: int | None = None, + limit: int = 200, +) -> list[dict[str, Any]]: + _check_token(x_admin_token) + where: list[str] = [] + params: dict[str, Any] = {"lim": min(limit, 1000)} + if run_id is not None: + where.append("run_id = :rid") + params["rid"] = run_id + if since_id is not None: + where.append("log_id > :sid") + params["sid"] = since_id + where_sql = ("WHERE " + " AND ".join(where)) if where else "" + rows = ( + db.execute( + text( + f""" + SELECT log_id, run_id, ts, level, stage, cad_number, message + FROM nspd_scrape_log + {where_sql} + ORDER BY log_id DESC + LIMIT :lim + """ + ), + params, + ) + .mappings() + .all() + ) + return [ + { + "log_id": r["log_id"], + "run_id": r["run_id"], + "ts": r["ts"].isoformat() if r["ts"] else None, + "level": r["level"], + "stage": r["stage"], + "cad_number": r["cad_number"], + "message": r["message"], + } + for r in rows + ] + + +@router.get("/nspd/coverage") +def nspd_coverage( + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + region_code: int = 66, +) -> dict[str, Any]: + """Покрытие cad_quarters_geom относительно ДДУ-сделок rosreestr. + Полезно перед запуском sweep — видишь сколько ещё осталось скрейпить.""" + _check_token(x_admin_token) + row = ( + db.execute( + text( + """ + WITH pending AS ( + SELECT DISTINCT quarter_cad_number AS cad + FROM rosreestr_deals + WHERE region_code = :rc + AND doc_type = 'ДДУ' + AND realestate_type_code = '002001003000' + AND quarter_cad_number IS NOT NULL + AND quarter_cad_number !~ '^00:00:' + AND quarter_cad_number !~ ':0000000$' + ), + done AS (SELECT cad_number AS cad FROM cad_quarters_geom) + SELECT + (SELECT COUNT(*) FROM pending) AS total_with_ddu, + (SELECT COUNT(*) FROM pending p + WHERE EXISTS (SELECT 1 FROM done d WHERE d.cad = p.cad)) + AS already_scraped, + (SELECT COUNT(*) FROM cad_buildings) AS buildings_total + """ + ), + {"rc": region_code}, + ) + .mappings() + .first() + ) + total = int(row["total_with_ddu"] or 0) + done = int(row["already_scraped"] or 0) + return { + "region_code": region_code, + "total_quarters_with_ddu": total, + "already_scraped": done, + "pending": max(total - done, 0), + "buildings_total": int(row["buildings_total"] or 0), + "coverage_pct": round(done / max(total, 1) * 100, 1) if total else None, + } + + +# ── DomRF (kn) — runs list ─────────────────────────────────────────────────── + + @router.get("/runs") def list_runs( db: Annotated[Session, Depends(get_db)], diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 805f2b26..e3785cd4 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -149,25 +149,46 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: logger.warning("worker_ready: failed to enqueue resume for run=%s: %s", rid, e) # NSPD-runs: помечаем зомби (>15 мин без heartbeat — NSPD сам по себе - # медленнее kn, дольше дельта). Resume не делаем — следующий beat - # подберёт pending (idempotent через cad_quarters_geom). + # медленнее kn, дельта больше). Resume = просто заново ставим + # scrape_nspd_region с тем же region_code и triggered_by='resume': + # сервис идемпотентен — pending = (rosreestr ДДУ cads) − (cad_quarters_geom done), + # уже скрейпнутые квартала пропустятся автоматом. db = SessionLocal() + nspd_resume_targets: list[int] = [] try: - db.execute( - text( - """ - UPDATE nspd_scrape_runs - SET status = 'zombie', - finished_at = NOW(), - error = COALESCE(error, - 'auto-zombie at worker_ready') - WHERE status = 'running' - AND COALESCE(heartbeat_at, started_at) - < NOW() - INTERVAL '15 minutes' - """ + zombie_rows = ( + db.execute( + text( + """ + UPDATE nspd_scrape_runs + SET status = 'zombie', + finished_at = NOW(), + error = COALESCE(error, + 'auto-zombie at worker_ready, resume scheduled') + WHERE status = 'running' + AND COALESCE(heartbeat_at, started_at) + < NOW() - INTERVAL '15 minutes' + RETURNING run_id, region_code + """ + ) ) + .mappings() + .all() ) db.commit() + # Уникальные регионы для resume (разные ран-id для одного региона — + # повторно не enqueueum, lock и idempotency пропустят дубль). + seen: set[int] = set() + for r in zombie_rows: + rc = int(r["region_code"]) + if rc not in seen: + seen.add(rc) + nspd_resume_targets.append(rc) + logger.info( + "worker_ready: NSPD zombie run=%s region=%s — resume scheduled", + r["run_id"], + rc, + ) except Exception as e: logger.warning("worker_ready nspd zombie scan failed: %s", e) try: @@ -176,3 +197,19 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: pass finally: db.close() + + if nspd_resume_targets: + from app.workers.tasks.scrape_nspd import scrape_nspd_region + + for rc in nspd_resume_targets: + try: + scrape_nspd_region.apply_async( + kwargs={"region_code": rc, "triggered_by": "resume"}, + ) + logger.info("worker_ready: scrape_nspd_region enqueued region=%s", rc) + except Exception as e: + logger.warning( + "worker_ready: failed to enqueue NSPD resume for region=%s: %s", + rc, + e, + ) diff --git a/frontend/src/app/admin/layout.tsx b/frontend/src/app/admin/layout.tsx index 6bfa4730..fb87372c 100644 --- a/frontend/src/app/admin/layout.tsx +++ b/frontend/src/app/admin/layout.tsx @@ -4,7 +4,8 @@ import Link from "next/link"; import { usePathname } from "next/navigation"; const TABS = [ - { href: "/admin/scrape", label: "Скрапер" }, + { href: "/admin/scrape", label: "Скрапер DOM.РФ" }, + { href: "/admin/scrape/nspd", label: "Скрапер NSPD" }, { href: "/admin/leads", label: "Лиды PRINZIP" }, ]; @@ -35,8 +36,19 @@ export default function AdminLayout({ }} > {TABS.map((t) => { - const active = + // Активна вкладка с самым длинным совпадающим префиксом — + // иначе /admin/scrape тоже подсветится при заходе на + // /admin/scrape/nspd. + const matchesPrefix = pathname === t.href || pathname?.startsWith(t.href + "/"); + const longerMatch = TABS.some( + (other) => + other.href !== t.href && + other.href.length > t.href.length && + (pathname === other.href || + pathname?.startsWith(other.href + "/")), + ); + const active = matchesPrefix && !longerMatch; return ( (() => + typeof window === "undefined" + ? "" + : (localStorage.getItem("admin_token") ?? ""), + ); + const [region, setRegion] = useState(66); + const [limit, setLimit] = useState(""); + const [rateMs, setRateMs] = useState(600); + const [forceRun, setForceRun] = useState(false); + const [logsRunId, setLogsRunId] = useState(""); + const queryClient = useQueryClient(); + + const saveToken = (v: string) => { + setToken(v); + localStorage.setItem("admin_token", v); + }; + + const coverage = useQuery({ + queryKey: ["nspd-coverage", region, token], + queryFn: () => + apiFetch( + `/api/v1/admin/scrape/nspd/coverage?region_code=${region}`, + { + headers: { "X-Admin-Token": token }, + }, + ), + enabled: !!token, + refetchInterval: 10000, + }); + + const runs = useQuery({ + queryKey: ["nspd-runs", token], + queryFn: () => + apiFetch("/api/v1/admin/scrape/nspd/runs?limit=20", { + headers: { "X-Admin-Token": token }, + }), + enabled: !!token, + refetchInterval: 5000, + }); + + const logs = useQuery({ + queryKey: ["nspd-logs", logsRunId, token], + queryFn: () => { + const q = + logsRunId !== "" ? `?run_id=${logsRunId}&limit=200` : "?limit=200"; + return apiFetch(`/api/v1/admin/scrape/nspd/logs${q}`, { + headers: { "X-Admin-Token": token }, + }); + }, + enabled: !!token, + refetchInterval: 3000, + }); + + const trigger = useMutation({ + mutationFn: () => + apiFetch("/api/v1/admin/scrape/nspd", { + method: "POST", + headers: { "X-Admin-Token": token }, + body: JSON.stringify({ + region_code: region, + limit: limit === "" ? null : Number(limit), + rate_ms: rateMs, + force: forceRun, + }), + }), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ["nspd-runs"] }); + queryClient.invalidateQueries({ queryKey: ["nspd-coverage"] }); + }, + }); + + const releaseLock = useMutation({ + mutationFn: () => + apiFetch<{ region_code: number; released: boolean }>( + "/api/v1/admin/scrape/nspd/release-lock", + { + method: "POST", + headers: { "X-Admin-Token": token }, + body: JSON.stringify({ region_code: region }), + }, + ), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ["nspd-runs"] }); + }, + }); + + return ( + <> +

+ Скрапер NSPD — кадастровые кварталы и здания +

+

+ Берёт quarter_cad_number из rosreestr_deals (ДДУ-сделки на квартиры) → + тянет полигон cad-квартала + здания внутри из NSPD geoportal. Расписание + (beat): 20-е число фев/мая/авг/нояб (после квартальных публикаций + rosreestr). См. SCRAPE_NSPD_CRON. +

+ +
+ +
+ + + + +
+ + {/* Pre-launch summary: сколько кварталов реально пойдёт в обработку + и сколько примерно займёт. Считаем тут, а не на бэке — дешевле. */} + {token && coverage.data + ? (() => { + const limitNum = limit === "" ? null : Number(limit); + const willProcess = limitNum + ? Math.min(limitNum, coverage.data.pending) + : coverage.data.pending; + // 2 запроса на квартал (полигон + здания) × rate_ms, +1с на запись + const etaSec = willProcess * ((rateMs * 2) / 1000 + 1); + const etaMin = Math.round(etaSec / 60); + const etaH = Math.floor(etaMin / 60); + const etaText = + etaMin >= 60 ? `~${etaH}ч ${etaMin % 60}мин` : `~${etaMin} мин`; + return ( +
+ {willProcess === 0 ? ( + <> + ✅ Всё уже скрейпнуто для региона {region}. Запускать + бесполезно — pending=0. + + ) : ( + <> + При запуске будет обработано{" "} + + {willProcess.toLocaleString("ru")} кварталов + {" "} + {limitNum && limitNum < coverage.data.pending ? ( + + (limit {limitNum} из {coverage.data.pending} pending) + + ) : ( + + (вся очередь pending) + + )} + . Расчётное время: {etaText} при rate{" "} + {rateMs}мс ({((rateMs * 2) / 1000).toFixed(1)}с / квартал + × 2 запроса). + + )} +
+ ); + })() + : null} + +
+ + + {!token ? ( + + ← введите Admin Token выше + + ) : null} +
+ + {trigger.isError ? ( +
+ Ошибка: {String(trigger.error)} +
+ ) : null} + + {trigger.isSuccess ? ( +
+ ✓ Поставлено в очередь. task_id={" "} + + {trigger.data.task_id} + {" "} + · регион {trigger.data.region_code} + {trigger.data.limit ? ` · limit=${trigger.data.limit}` : ""} · rate= + {trigger.data.rate_ms}мс +
+ ) : null} +
+ +
+

+ Покрытие cad_quarters_geom (регион {region}) +

+ {!token ? ( +

+ Введите Admin Token выше. +

+ ) : coverage.isLoading ? ( +

Загрузка…

+ ) : coverage.isError ? ( +

Ошибка: {String(coverage.error)}

+ ) : coverage.data ? ( +
+ + + 0 ? "#9a6700" : "#0a7a3a"} + /> + + = 90 + ? "#0a7a3a" + : "#9a6700" + } + /> +
+ ) : null} +
+ +
+
+

+ Лог скрейпа{" "} + + · обновляется каждые 3 сек + +

+ + setLogsRunId(e.target.value === "" ? "" : Number(e.target.value)) + } + placeholder="run_id (опц.)" + style={{ ...inputStyle, width: 130 }} + /> +
+ {!token ? ( +

+ Введите Admin Token чтобы видеть лог. +

+ ) : logs.isLoading ? ( +

Загрузка…

+ ) : logs.isError ? ( +

Ошибка: {String(logs.error)}

+ ) : (logs.data?.length ?? 0) === 0 ? ( +

+ Нет событий{logsRunId !== "" ? ` для run_id=${logsRunId}` : ""}. +

+ ) : ( +
+ {(logs.data ?? []).map((l) => { + const color = + l.level === "error" + ? "#fca5a5" + : l.level === "warn" + ? "#fcd34d" + : "#a3e635"; + return ( +
+ + {l.ts?.slice(11, 19)} + {" "} + {l.level.toUpperCase()}{" "} + + [run-{l.run_id} + {l.stage ? ` · ${l.stage}` : ""} + {l.cad_number ? ` · ${l.cad_number}` : ""}] + {" "} + {l.message} +
+ ); + })} +
+ )} +
+ +
+

Последние прогоны

+ {!token ? ( +

+ Введите Admin Token чтобы увидеть журнал. +

+ ) : runs.isLoading ? ( +

Загрузка…

+ ) : runs.isError ? ( +

Ошибка: {String(runs.error)}

+ ) : ( + + + + {[ + "#", + "Старт", + "HB", + "Статус", + "Прогресс", + "Pending", + "OK", + "Fail", + "Зданий", + "Req", + "WAF", + "Trigger", + "Ошибка", + ].map((h) => ( + + ))} + + + + {(runs.data ?? []).map((r) => ( + + + + + + + + + + + + + + + + ))} + +
+ {h} +
{r.run_id} + {r.started_at?.slice(0, 19).replace("T", " ")} + + {r.heartbeat_at ? r.heartbeat_at.slice(11, 19) : "—"} + + {r.status} + + {r.pending_count && r.pending_count > 0 + ? (() => { + const pct = Math.min( + 100, + Math.round( + ((r.quarters_ok + r.quarters_failed) / + r.pending_count) * + 100, + ), + ); + return ( +
+
+
0 + ? "#1d4ed8" + : "#9a6700", + }} + /> +
+ {pct}% +
+ ); + })() + : "—"} +
{r.pending_count ?? "—"}{r.quarters_ok} 0 ? "#b3261e" : undefined, + }} + > + {r.quarters_failed} + {r.buildings_ok}{r.requests_count} 0 ? "#b3261e" : undefined, + }} + > + {r.waf_429_count} + {r.triggered_by} + {r.error ?? ""} +
+ )} +
+ + ); +} + +function CoverageCard({ + label, + value, + color, +}: { + label: string; + value: string; + color?: string; +}) { + return ( +
+
+ {label} +
+
+ {value} +
+
+ ); +} + +const cardStyle = { + background: "#fff", + border: "1px solid #e6e8ec", + borderRadius: 12, + padding: 20, +}; +const labelStyle = { + display: "block", + fontSize: 12, + color: "#5b6066", + marginBottom: 4, + textTransform: "uppercase" as const, + letterSpacing: 0.4, +}; +const inputStyle = { + padding: "8px 10px", + border: "1px solid #d1d5db", + borderRadius: 6, + fontSize: 14, + width: "100%", + boxSizing: "border-box" as const, +}; +const primaryBtn = { + padding: "10px 18px", + background: "#1d4ed8", + color: "#fff", + border: "none", + borderRadius: 6, + fontSize: 14, + fontWeight: 600, + cursor: "pointer", +}; +const secondaryBtn = { + padding: "8px 14px", + background: "#f3f4f6", + color: "#1f2937", + border: "1px solid #d1d5db", + borderRadius: 6, + fontSize: 13, + cursor: "pointer", +}; +const th = { + padding: "8px 10px", + textAlign: "left" as const, + fontWeight: 600, + borderBottom: "1px solid #e6e8ec", +}; +const td = { padding: "8px 10px" };