"""Импорт ПЗЗ территориальных зон ЕКБ из Росреестр PKK6.""" import asyncio import json import logging import httpx from sqlalchemy import text from app.core.db import SessionLocal logger = logging.getLogger(__name__) PKK6_URL = "https://pkk.rosreestr.ru/arcgis/rest/services/PKK6/ZONES/" "MapServer/5/query" # bbox ЕКБ: (xmin, ymin, xmax, ymax) в WGS84 EKB_BBOX = (60.5, 56.7, 60.75, 56.95) async def fetch_pkk6_zones() -> list[dict]: """Запросить ВСЕ территориальные зоны для bbox ЕКБ с пагинацией. PKK6 maxRecordCount = 1000 per request. Пагинируем через resultOffset. Rate-limit: asyncio.sleep(0.5) между страницами. """ xmin, ymin, xmax, ymax = EKB_BBOX geometry = json.dumps( { "xmin": xmin, "ymin": ymin, "xmax": xmax, "ymax": ymax, "spatialReference": {"wkid": 4326}, } ) headers = { "User-Agent": "GenDesign-SiteFinder/1.0 (+https://gendsgn.ru)", "Accept": "application/json", } all_features: list[dict] = [] offset = 0 page_size = 1000 # verify=False: Росреестр PKK6 self-signed cert. Публичный read-only API, # MITM risk минимален. follow_redirects=True: PKK6 возвращает 301 на # некоторые запросы (вероятно legacy URL redirect). async with httpx.AsyncClient( timeout=60, headers=headers, verify=False, follow_redirects=True ) as client: while True: params = { "where": "1=1", "geometry": geometry, "geometryType": "esriGeometryEnvelope", "inSR": "4326", "outSR": "4326", "outFields": "*", "f": "geojson", "resultOffset": offset, "resultRecordCount": page_size, } r = await client.get(PKK6_URL, params=params) r.raise_for_status() data = r.json() features = data.get("features", []) if not features: break all_features.extend(features) logger.info("PKK6 page offset=%d → %d features", offset, len(features)) if len(features) < page_size: break offset += page_size await asyncio.sleep(0.5) # вежливый rate-limit logger.info("PKK6: total %d zones for EKB bbox", len(all_features)) return all_features def sync_pzz_zones_to_db() -> dict[str, int]: """Загрузить зоны из PKK6 и UPSERT в pzz_zones_ekb. Возвращает словарь {fetched, inserted, updated, skipped}. Каждый UPSERT обёрнут в SAVEPOINT (db.begin_nested), чтобы ошибка одной записи не откатывала всю транзакцию и счётчики оставались согласованными с тем, что реально записано в DB. """ features = asyncio.run(fetch_pkk6_zones()) inserted = 0 updated = 0 skipped = 0 db = SessionLocal() try: for f in features: props = f.get("properties") or {} geom = f.get("geometry") if not geom: skipped += 1 continue rosreestr_id = props.get("OBJECTID") or props.get("objectid") zone_code = ( props.get("zone_number") or props.get("zone_code") or props.get("identification_number") or None ) zone_name = ( props.get("description") or props.get("name") or props.get("zone_name") or None ) description = props.get("description") try: with db.begin_nested(): # SAVEPOINT — откат только этой записи result = db.execute( text(""" INSERT INTO pzz_zones_ekb (rosreestr_id, zone_code, zone_name, description, raw_props, geom) VALUES ( :rid, :code, :nm, :desc, CAST(:props AS jsonb), ST_Multi(ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326)) ) ON CONFLICT (rosreestr_id) DO UPDATE SET zone_code = EXCLUDED.zone_code, zone_name = EXCLUDED.zone_name, description = EXCLUDED.description, raw_props = EXCLUDED.raw_props, geom = EXCLUDED.geom, fetched_at = NOW() RETURNING (xmax = 0) AS is_insert """), { "rid": rosreestr_id, "code": zone_code, "nm": zone_name, "desc": description, "props": json.dumps(props, ensure_ascii=False), "g": json.dumps(geom), }, ).scalar() if result: inserted += 1 else: updated += 1 except Exception as e: logger.warning("PZZ insert failed for rosreestr_id=%s: %s", rosreestr_id, e) skipped += 1 db.commit() except Exception as e: db.rollback() logger.exception("pzz_loader: unexpected error, outer tx rolled back: %s", e) raise finally: db.close() return { "fetched": len(features), "inserted": inserted, "updated": updated, "skipped": skipped, }