From fe2a881cec8615ec6c070f484057dee206977485 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Mon, 11 May 2026 15:56:17 +0300 Subject: [PATCH] feat(api,analytics): bulk geo backfill + complex_buildings query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - POST /api/v1/admin/scrape/geo/bulk — splits pending Sverdlovsk cad-nums into N chunks (parallelism 1..10, default 5), creates N jobs and enqueues each in queue=geo. source_kind='rosreestr_pending_chunk' for tracking. - analytics_queries.complex_buildings(db, obj_id) — returns list of buildings from cad_buildings (cad_num, floors, area, purpose, name, address, geom). - object_detail: LEFT JOIN v_complex_buildings, adds buildings_count. - top_developers: adds complexes_count via correlated subquery. - GET /api/v1/analytics/object/{obj_id}/buildings → list[ComplexBuilding]. --- backend/app/api/v1/admin_scrape.py | 84 +++++++++++++++++++++++ backend/app/api/v1/analytics.py | 10 +++ backend/app/schemas/complex_buildings.py | 15 ++++ backend/app/services/analytics_queries.py | 76 +++++++++++++++++--- 4 files changed, 176 insertions(+), 9 deletions(-) create mode 100644 backend/app/schemas/complex_buildings.py diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index f3777cc6..7c4d9774 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -693,6 +693,90 @@ def enqueue_geo_job( } +class BulkGeoEnqueueRequest(BaseModel): + """Параметры для параллельного backfill geo по Свердловской обл.""" + + parallelism: int = Field(default=5, ge=1, le=10) + thematic_id: int = Field(default=2, ge=1, le=15, description="1=parcel, 2=quarter, 5=building") + + +@router.post("/geo/bulk") +def bulk_enqueue_geo( + payload: BulkGeoEnqueueRequest, + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Разбить pending cad-номера (region 66) на N чанков и запустить N geo-jobs параллельно. + + Логика выборки pending — те же правила что у source_kind='rosreestr_pending': + quarter_cad_number из rosreestr_deals (region 66, ДДУ, тип 002001003000) + которых нет в cad_quarters_geom. + """ + _check_token(x_admin_token) + from app.services.job_settings import get_setting_value + from app.workers.tasks.nspd_geo import enqueue_geo_job as enqueue_helper + from app.workers.tasks.nspd_geo import process_nspd_geo_job + + # 1) Собрать все pending cad-номера для region 66 + rows = db.execute( + text( + """ + SELECT DISTINCT quarter_cad_number AS cad + FROM rosreestr_deals + WHERE region_code = ANY(:rc) + AND doc_type = 'ДДУ' + AND realestate_type_code = '002001003000' + AND quarter_cad_number IS NOT NULL + AND quarter_cad_number NOT LIKE :bp + AND quarter_cad_number NOT LIKE :bs + AND NOT EXISTS (SELECT 1 FROM cad_quarters_geom g + WHERE g.cad_number = rosreestr_deals.quarter_cad_number) + LIMIT 50000 + """ + ), + {"rc": [66], "bp": "00:00:%", "bs": "%:0000000"}, + ).all() + + all_cad: list[str] = [r[0] for r in rows] + pending_total = len(all_cad) + + if pending_total == 0: + raise HTTPException(status_code=400, detail="Нет cad-номеров для backfill") + + # 2) Разбить на чанки (numpy-style array_split — равные куски, хвост меньше) + n_jobs = min(payload.parallelism, pending_total) + chunk_size, remainder = divmod(pending_total, n_jobs) + chunks: list[list[str]] = [] + start = 0 + for i in range(n_jobs): + end = start + chunk_size + (1 if i < remainder else 0) + chunks.append(all_cad[start:end]) + start = end + + # 3) Создать job + enqueue для каждого чанка + geo_queue = get_setting_value("nspd_geo", "queue_name", "geo") + job_ids: list[int] = [] + for idx, chunk in enumerate(chunks): + cad_with_thematic = [(c, payload.thematic_id) for c in chunk] + job_id = enqueue_helper( + name=f"bulk_svrd_{idx + 1}/{n_jobs}", + job_kind="quarters", + source_kind="rosreestr_pending_chunk", + source_params={"region_codes": [66], "thematic_id": payload.thematic_id}, + cad_nums_with_thematic=cad_with_thematic, + triggered_by="bulk_admin", + ) + process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue) + job_ids.append(job_id) + + return { + "job_ids": job_ids, + "targets_total": pending_total, + "parallelism": n_jobs, + "targets_per_job": chunk_size + (1 if remainder else 0), + } + + @router.get("/geo/jobs") def list_geo_jobs( db: Annotated[Session, Depends(get_db)], diff --git a/backend/app/api/v1/analytics.py b/backend/app/api/v1/analytics.py index 64b23fe5..61028b56 100644 --- a/backend/app/api/v1/analytics.py +++ b/backend/app/api/v1/analytics.py @@ -9,6 +9,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from app.core.db import get_db +from app.schemas.complex_buildings import ComplexBuilding from app.schemas.recommend import RecommendMixInput, RecommendMixOutput from app.services import analytics_queries as q @@ -159,6 +160,15 @@ def object_detail( return result +@router.get("/object/{obj_id}/buildings", response_model=list[ComplexBuilding]) +def object_buildings( + db: Annotated[Session, Depends(get_db)], + obj_id: int, +) -> list[dict[str, Any]]: + """Список кадастровых зданий ЖК из cad_buildings.""" + return q.complex_buildings(db, obj_id=obj_id) + + # ---- PRINZIP-specific ------------------------------------------------------- diff --git a/backend/app/schemas/complex_buildings.py b/backend/app/schemas/complex_buildings.py new file mode 100644 index 00000000..ba364797 --- /dev/null +++ b/backend/app/schemas/complex_buildings.py @@ -0,0 +1,15 @@ +"""Pydantic schemas for complex cadastral buildings endpoint.""" + +from __future__ import annotations + +from pydantic import BaseModel + + +class ComplexBuilding(BaseModel): + cad_num: str + floors: int | None + area: float | None + purpose: str | None + building_name: str | None + readable_address: str | None + geom_geojson: dict | None # GeoJSON Geometry, e.g. {"type": "Polygon", "coordinates": [...]} diff --git a/backend/app/services/analytics_queries.py b/backend/app/services/analytics_queries.py index d3c01197..d04752e5 100644 --- a/backend/app/services/analytics_queries.py +++ b/backend/app/services/analytics_queries.py @@ -318,7 +318,9 @@ def top_developers(db: Session, region_code: int = 66, limit: int = 15) -> list[ m.avg_area_sqm, m.pct_one, m.pct_three_plus, fl.sold_first, fl.sold_last, (fl.sold_last - fl.sold_first) AS sold_delta_pp, - fl.first_dt, fl.last_dt + fl.first_dt, fl.last_dt, + (SELECT COUNT(*) FROM complexes c + WHERE c.developer_id = m.developer_id) AS complexes_count FROM v_developer_full_metrics m LEFT JOIN first_last fl ON fl.developer_id = m.developer_id WHERE m.sverdl_sqm IS NOT NULL @@ -347,6 +349,7 @@ def top_developers(db: Session, region_code: int = 66, limit: int = 15) -> list[ "avg_area_sqm": _f(r["avg_area_sqm"]), "pct_one": _f(r["pct_one"]), "pct_three_plus": _f(r["pct_three_plus"]), + "complexes_count": int(r["complexes_count"] or 0), } for r in rows ] @@ -603,18 +606,23 @@ def prinzip_insights() -> dict[str, Any]: def object_detail(db: Session, obj_id: int) -> dict[str, Any] | None: - """Базовая инфа объекта из domrf_kn_objects (последний snapshot).""" + """Базовая инфа объекта из domrf_kn_objects (последний snapshot). + + Также возвращает buildings_count из v_complex_buildings (0 если зданий нет). + """ row = ( db.execute( text( """ - SELECT obj_id, hobj_id, comm_name, addr, short_addr, region_cd, - dev_id, dev_name, floor_min, floor_max, flat_count, square_living, - ready_dt, site_status, escrow, obj_class, latitude, longitude, - obj_status, snapshot_date - FROM domrf_kn_objects - WHERE obj_id = :obj - ORDER BY snapshot_date DESC + SELECT o.obj_id, o.hobj_id, o.comm_name, o.addr, o.short_addr, o.region_cd, + o.dev_id, o.dev_name, o.floor_min, o.floor_max, o.flat_count, + o.square_living, o.ready_dt, o.site_status, o.escrow, o.obj_class, + o.latitude, o.longitude, o.obj_status, o.snapshot_date, + COALESCE(cb.buildings_count, 0) AS buildings_count + FROM domrf_kn_objects o + LEFT JOIN v_complex_buildings cb ON cb.complex_id = o.obj_id + WHERE o.obj_id = :obj + ORDER BY o.snapshot_date DESC LIMIT 1 """ ), @@ -646,6 +654,7 @@ def object_detail(db: Session, obj_id: int) -> dict[str, Any] | None: "longitude": _f(row["longitude"]), "obj_status": row["obj_status"], "snapshot_date": row["snapshot_date"].isoformat() if row["snapshot_date"] else None, + "buildings_count": int(row["buildings_count"]), } @@ -2198,3 +2207,52 @@ def recommend_mix( for r in cmp_rows ], } + + +# ── Cadastral buildings per complex ────────────────────────────────────────── + + +def complex_buildings(db: Session, obj_id: int) -> list[dict[str, Any]]: + """Список зданий из cad_buildings для данного ЖК. + + Возвращает [] если ни одного здания не найдено. + """ + rows = ( + db.execute( + text( + """ + SELECT cad_num, floors, area, purpose, building_name, + readable_address, ST_AsGeoJSON(geom) AS geom_geojson + FROM cad_buildings + WHERE complex_id = :obj_id + ORDER BY cad_num + """ + ), + {"obj_id": obj_id}, + ) + .mappings() + .all() + ) + import json as _json + + result: list[dict[str, Any]] = [] + for r in rows: + geom_raw = r["geom_geojson"] + geom: dict[str, Any] | None = None + if geom_raw: + try: + geom = _json.loads(geom_raw) + except (ValueError, TypeError): + geom = None + result.append( + { + "cad_num": r["cad_num"], + "floors": r["floors"], + "area": _f(r["area"]), + "purpose": r["purpose"], + "building_name": r["building_name"], + "readable_address": r["readable_address"], + "geom_geojson": geom, + } + ) + return result