lekss361
ef15190690
Co-authored-by: lekss361 <lekss361@gendsgn.local> Co-committed-by: lekss361 <lekss361@gendsgn.local>
444 lines
19 KiB
Python
444 lines
19 KiB
Python
"""ЕКБ РНС/РВЭ xlsx parser (Issue #105).
|
||
|
||
Source: https://xn--80acgfbsl1azdqr.xn--p1ai/дляработы/гиз/градостроительство/разрешение
|
||
Format: Form 3 (РНС) + Form 4 (РВЭ) — стандартные Росстат шаблоны.
|
||
|
||
Структура xlsx (проверена на всех файлах 2022-2026):
|
||
Лист «реестр разрешений на строительс» → РНС (Таблица 3)
|
||
Лист «реестр разрешений на ввод» → РВЭ (Таблица 4)
|
||
|
||
Расположение заголовков:
|
||
2022: строки 5-7 — заголовок/подзаголовок/номера; данные с row 8
|
||
2023+: строки 4-6 — заголовок/подзаголовок/номера; данные с row 7
|
||
|
||
Колонки (0-based) одинаковы во всех годах:
|
||
0 developer_name — Наименование застройщика
|
||
1 developer_inn — ИНН (int или str)
|
||
2 developer_address — Адрес застройщика (не сохраняем)
|
||
3 object_type — Тип строительного объекта
|
||
4 object_name — Наименование объекта КС
|
||
5 cadastral_number — Кадастровый номер ЗУ
|
||
6 raw_coord_x — X (СКФ-66, местная СК Свердловской обл.)
|
||
7 raw_coord_y — Y (СКФ-66)
|
||
8 construction_address — Адрес объекта
|
||
9 permit_number — Номер разрешения на строительство
|
||
10 issue_date — Дата разрешения на строительство
|
||
11 expiry_date — Дата окончания разрешения
|
||
12 total_area_sqm — Общая площадь, м²
|
||
13 living_area_sqm — Площадь жилых помещений по проекту, м²
|
||
-- дополнительно только для РВЭ:
|
||
14 living_area_fact_sqm — Площадь жилых помещений фактически, м²
|
||
15 rve_number — Номер разрешения на ввод
|
||
16 rve_date — Дата разрешения на ввод
|
||
|
||
Координаты raw_coord_x / raw_coord_y — СКФ-66 (МСК-66, Свердловская обл., зона 1).
|
||
Параметры проекции: tmerc, lon_0=60°E, k=1, false_easting=1496944, false_northing=-5911077,
|
||
эллипсоид Красовского, сдвиг datum towgs84=23.92,-141.27,-80.9,0,0.35,0.82,-0.12.
|
||
Параметры откалиброваны по cad_parcels ground-truth (738 точек): avg_err≈14м, median≈7м.
|
||
Ячейки могут содержать несколько точек через пробел — берём первую.
|
||
Геокодирование выполняется при парсинге (geom, geocoded_lat/lon).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from collections.abc import Iterator
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
from io import BytesIO
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from openpyxl import load_workbook
|
||
from pyproj import CRS, Transformer
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── МСК-66 → WGS84 трансформер ───────────────────────────────────────────────
|
||
# СКФ-66 (местная СК Свердловской обл., зона 1):
|
||
# tmerc, осевой меридиан 60°E, масштаб 1.0
|
||
# x_0=1496944 (false_easting), y_0=-5911077 (false_northing)
|
||
# Откалиброваны по cad_parcels ground-truth (738 точек permit'ов с кад.номером):
|
||
# avg_err≈14м, median≈7м, p95≈58м (без 8 записей с ошибочными Y в исходных данных).
|
||
# Предыдущее x_0=1500000 давало систематический сдвиг ~3050м западнее по долготе.
|
||
# эллипсоид Красовского + сдвиг Пулково→WGS84
|
||
_MSK66_PROJ4 = (
|
||
"+proj=tmerc +lat_0=0 +lon_0=60 +k=1"
|
||
" +x_0=1496944 +y_0=-5911077"
|
||
" +ellps=krass +towgs84=23.92,-141.27,-80.9,0,0.35,0.82,-0.12"
|
||
" +units=m +no_defs"
|
||
)
|
||
_MSK66_TO_WGS84: Transformer = Transformer.from_crs(
|
||
CRS.from_proj4(_MSK66_PROJ4),
|
||
CRS.from_epsg(4326),
|
||
always_xy=True,
|
||
)
|
||
|
||
# Валидный bbox ЕКБ и ближайших районов (WGS84)
|
||
_EKB_LON_MIN = 59.5
|
||
_EKB_LON_MAX = 61.5
|
||
_EKB_LAT_MIN = 56.3
|
||
_EKB_LAT_MAX = 57.5
|
||
|
||
# Паттерн для разбора первого числа из строки координат
|
||
# (ячейки могут содержать несколько точек через пробелы или запятую-десятичный разделитель)
|
||
_COORD_FIRST_RE = re.compile(r"[\d]+[.,][\d]+")
|
||
|
||
|
||
def msk66_to_wgs84(raw_x: str | None, raw_y: str | None) -> tuple[float, float] | None:
|
||
"""Преобразовать первую точку МСК-66 (raw_coord_x, raw_coord_y) в (lon, lat) WGS84.
|
||
|
||
Берёт первое числовое значение из каждой ячейки (в xlsx могут быть несколько
|
||
точек через пробел). Возвращает None если координаты отсутствуют, не парсятся
|
||
или результат выходит за пределы ЕКБ-региона.
|
||
|
||
В МСК-66 Свердловской обл.:
|
||
raw_coord_x ≈ 1 500 000 + смещение (откалиброванный false_easting = 1 496 944)
|
||
raw_coord_y ≈ northing + смещение (откалиброванный false_northing = -5 911 077)
|
||
"""
|
||
if not raw_x or not raw_y:
|
||
return None
|
||
|
||
# Нормализуем запятую как десятичный разделитель
|
||
x_str = raw_x.replace(",", ".")
|
||
y_str = raw_y.replace(",", ".")
|
||
|
||
# Берём первое число из (возможно) многоточечной строки
|
||
mx = _COORD_FIRST_RE.search(x_str)
|
||
my = _COORD_FIRST_RE.search(y_str)
|
||
if not mx or not my:
|
||
return None
|
||
|
||
try:
|
||
msk_x = float(mx.group())
|
||
msk_y = float(my.group())
|
||
except ValueError:
|
||
return None
|
||
|
||
# Грубая проверка диапазона МСК-66 X (false_easting=1500000, типичные отклонения ±50 км)
|
||
if not (1_450_000 <= msk_x <= 1_600_000):
|
||
logger.debug("msk66_to_wgs84: X=%s вне ожидаемого диапазона МСК-66", msk_x)
|
||
return None
|
||
|
||
try:
|
||
lon, lat = _MSK66_TO_WGS84.transform(msk_x, msk_y)
|
||
except Exception as exc: # pyproj может поднять ProjError
|
||
logger.debug("msk66_to_wgs84: transform failed X=%s Y=%s: %s", msk_x, msk_y, exc)
|
||
return None
|
||
|
||
if not (_EKB_LON_MIN <= lon <= _EKB_LON_MAX and _EKB_LAT_MIN <= lat <= _EKB_LAT_MAX):
|
||
logger.debug("msk66_to_wgs84: результат (%.4f, %.4f) вне ЕКБ-региона", lon, lat)
|
||
return None
|
||
|
||
return lon, lat
|
||
|
||
|
||
EKBURG_PERMITS_URLS: dict[int, str] = {
|
||
2026: "https://xn--80acgfbsl1azdqr.xn--p1ai/file/70bf01bf31538ee9dd82dadfc47192a0",
|
||
2025: "https://xn--80acgfbsl1azdqr.xn--p1ai/file/6a0a18c9ee327e6e4f76c32a5385a6bd",
|
||
2024: "https://xn--80acgfbsl1azdqr.xn--p1ai/file/907bfa0cf78d5a93c6ccafed1af51fc5",
|
||
2023: "https://xn--80acgfbsl1azdqr.xn--p1ai/file/3b2ef86bf5673adaa65263672f0c623f",
|
||
2022: "https://xn--80acgfbsl1azdqr.xn--p1ai/file/51a7e5654d2fa018bc6db402ab4d5775",
|
||
}
|
||
|
||
# Листы которые нужно пропустить (справочник, вспомогательные)
|
||
_SKIP_SHEETS = {"справочник", "лист1", "sheet1"}
|
||
|
||
# Паттерн для проверки что строка — строка данных (первая ячейка непустая и не число-нумерация)
|
||
_INN_RE = re.compile(r"\d{10,12}")
|
||
|
||
|
||
@dataclass
|
||
class PermitRow:
|
||
"""Parsed row from РНС/РВЭ xlsx."""
|
||
|
||
permit_type: str # "RNS" | "RVE"
|
||
permit_number: str
|
||
issue_date: date | None
|
||
expiry_date: date | None
|
||
developer_inn: str | None
|
||
developer_name: str | None
|
||
object_name: str | None
|
||
object_type: str | None
|
||
construction_address: str | None
|
||
cadastral_number: str | None
|
||
total_area_sqm: float | None
|
||
living_area_sqm: float | None
|
||
living_area_fact_sqm: float | None # только РВЭ
|
||
rve_number: str | None # только РВЭ
|
||
rve_date: date | None # только РВЭ
|
||
raw_coord_x: str | None
|
||
raw_coord_y: str | None
|
||
geocoded_lat: float | None # WGS84 широта (из МСК-66, None если не сконвертировано)
|
||
geocoded_lon: float | None # WGS84 долгота (из МСК-66, None если не сконвертировано)
|
||
source_year: int
|
||
source_url: str
|
||
raw_row: dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _to_str(v: Any) -> str | None:
|
||
"""Привести значение ячейки к строке, убрать лишние пробелы. None если пусто."""
|
||
if v is None:
|
||
return None
|
||
s = str(v).strip().replace("\xa0", " ")
|
||
return s if s and s != "-" else None
|
||
|
||
|
||
def _to_date(v: Any) -> date | None:
|
||
"""Привести значение ячейки к date.
|
||
|
||
Принимает: datetime (openpyxl), date, строки 'DD.MM.YYYY', 'YYYY-MM-DD'.
|
||
"""
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, datetime):
|
||
return v.date()
|
||
if isinstance(v, date):
|
||
return v
|
||
if isinstance(v, str):
|
||
s = v.strip()
|
||
for fmt in ("%d.%m.%Y", "%Y-%m-%d", "%d-%m-%Y"):
|
||
try:
|
||
return datetime.strptime(s, fmt).date()
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _to_float(v: Any) -> float | None:
|
||
"""Привести значение ячейки к float. Обрабатывает запятую как разделитель."""
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, int | float):
|
||
return float(v) if not isinstance(v, bool) else None
|
||
s = str(v).strip().replace(",", ".").replace("\xa0", "").replace(" ", "")
|
||
if not s or s == "-":
|
||
return None
|
||
try:
|
||
return float(s)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _clean_inn(v: Any) -> str | None:
|
||
"""Извлечь ИНН из значения ячейки. ИНН — 10 или 12 цифр."""
|
||
if v is None:
|
||
return None
|
||
s = str(v).strip().replace("\xa0", "")
|
||
# Если ячейка — число (openpyxl выдаёт int/float)
|
||
if isinstance(v, int | float) and not isinstance(v, bool):
|
||
digits = str(int(v))
|
||
if 10 <= len(digits) <= 12:
|
||
return digits
|
||
return None
|
||
m = _INN_RE.search(s)
|
||
return m.group(0) if m else None
|
||
|
||
|
||
def _detect_permit_type(sheet_name: str) -> str | None:
|
||
"""Определить тип разрешения по имени листа."""
|
||
name = sheet_name.lower()
|
||
if "строит" in name:
|
||
return "RNS"
|
||
if "ввод" in name:
|
||
return "RVE"
|
||
return None
|
||
|
||
|
||
def _detect_header_row(sheet: Any) -> int:
|
||
"""Найти строку с заголовком 'Наименование застройщика' (1-based).
|
||
|
||
Возвращает номер строки-данных (header_row + 3, так как за заголовком идут
|
||
две строки подзаголовков и нумерация).
|
||
"""
|
||
for row_idx, row in enumerate(
|
||
sheet.iter_rows(min_row=1, max_row=10, values_only=True), start=1
|
||
):
|
||
if row and row[0] and "застройщик" in str(row[0]).lower():
|
||
# +3: подзаголовок (X/Y), нумерация (1,2,3...), первая данных
|
||
return row_idx + 3
|
||
# Fallback: стандартные позиции
|
||
# 2022: header=5, данные с 8; 2023+: header=4, данные с 7
|
||
return 7
|
||
|
||
|
||
def _is_data_row(row: tuple[Any, ...]) -> bool:
|
||
"""Вернуть True если строка содержит реальные данные (не пустая, не заголовок)."""
|
||
if not row or all(v is None for v in row):
|
||
return False
|
||
first = row[0]
|
||
if first is None:
|
||
# Продолжение предыдущей записи (merged cells) — пропускаем
|
||
return False
|
||
# Проверяем, что первая ячейка — не номер (нумерация столбцов в header)
|
||
if isinstance(first, int | float) and not isinstance(first, bool):
|
||
val = int(first)
|
||
if 1 <= val <= 30:
|
||
return False
|
||
return True
|
||
|
||
|
||
def _parse_row(
|
||
row: tuple[Any, ...],
|
||
permit_type: str,
|
||
year: int,
|
||
source_url: str,
|
||
) -> PermitRow | None:
|
||
"""Распарсить одну строку данных в PermitRow.
|
||
|
||
Возвращает None если строка не содержит номера разрешения (обязательное поле).
|
||
"""
|
||
# permit_number — колонка 9 (0-based)
|
||
permit_number = _to_str(row[9]) if len(row) > 9 else None
|
||
if not permit_number:
|
||
return None
|
||
|
||
raw: dict[str, Any] = {str(i): str(v) for i, v in enumerate(row) if v is not None}
|
||
|
||
raw_x = _to_str(row[6]) if len(row) > 6 else None
|
||
raw_y = _to_str(row[7]) if len(row) > 7 else None
|
||
coords = msk66_to_wgs84(raw_x, raw_y)
|
||
geocoded_lon: float | None = coords[0] if coords else None
|
||
geocoded_lat: float | None = coords[1] if coords else None
|
||
|
||
return PermitRow(
|
||
permit_type=permit_type,
|
||
permit_number=permit_number,
|
||
issue_date=_to_date(row[10]) if len(row) > 10 else None,
|
||
expiry_date=_to_date(row[11]) if len(row) > 11 else None,
|
||
developer_inn=_clean_inn(row[1]) if len(row) > 1 else None,
|
||
developer_name=_to_str(row[0]) if len(row) > 0 else None,
|
||
object_name=_to_str(row[4]) if len(row) > 4 else None,
|
||
object_type=_to_str(row[3]) if len(row) > 3 else None,
|
||
construction_address=_to_str(row[8]) if len(row) > 8 else None,
|
||
cadastral_number=_to_str(row[5]) if len(row) > 5 else None,
|
||
total_area_sqm=_to_float(row[12]) if len(row) > 12 else None,
|
||
living_area_sqm=_to_float(row[13]) if len(row) > 13 else None,
|
||
living_area_fact_sqm=(
|
||
_to_float(row[14]) if permit_type == "RVE" and len(row) > 14 else None
|
||
),
|
||
rve_number=_to_str(row[15]) if permit_type == "RVE" and len(row) > 15 else None,
|
||
rve_date=_to_date(row[16]) if permit_type == "RVE" and len(row) > 16 else None,
|
||
raw_coord_x=raw_x,
|
||
raw_coord_y=raw_y,
|
||
geocoded_lat=geocoded_lat,
|
||
geocoded_lon=geocoded_lon,
|
||
source_year=year,
|
||
source_url=source_url,
|
||
raw_row=raw,
|
||
)
|
||
|
||
|
||
# ── client ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class EkburgPermitsClient:
|
||
"""Client для загрузки + парсинга РНС/РВЭ xlsx с екатеринбург.рф."""
|
||
|
||
DEFAULT_TIMEOUT = 60.0
|
||
USER_AGENT = "GenDesign/1.0 (+https://gendsgn.ru) Site Finder permits scraper"
|
||
|
||
def __init__(self, *, timeout: float = DEFAULT_TIMEOUT) -> None:
|
||
# verify=False: екатеринбург.рф подписан CA Минцифры РФ (нет в certifi).
|
||
# Данные публичные open-data — SSL pinning здесь не требуется. Issue #242.
|
||
self._client = httpx.Client(
|
||
timeout=timeout,
|
||
follow_redirects=True,
|
||
headers={"User-Agent": self.USER_AGENT},
|
||
verify=False,
|
||
)
|
||
|
||
def __enter__(self) -> EkburgPermitsClient:
|
||
return self
|
||
|
||
def __exit__(self, *_: Any) -> None:
|
||
self._client.close()
|
||
|
||
def download_xlsx(self, year: int) -> bytes:
|
||
"""GET xlsx для заданного года. Поднимает ValueError для неизвестного года."""
|
||
url = EKBURG_PERMITS_URLS.get(year)
|
||
if not url:
|
||
raise ValueError(f"No URL configured for year={year}")
|
||
response = self._client.get(url)
|
||
response.raise_for_status()
|
||
logger.info("Downloaded ekburg permits xlsx %d: %d bytes", year, len(response.content))
|
||
return response.content
|
||
|
||
def parse_xlsx(self, content: bytes, year: int, source_url: str) -> Iterator[PermitRow]:
|
||
"""Parse xlsx bytes → yield PermitRow.
|
||
|
||
Автоматически определяет тип листа (РНС/РВЭ) по названию.
|
||
Пропускает листы «Справочник», «Лист1» и неизвестные.
|
||
"""
|
||
wb = load_workbook(BytesIO(content), read_only=True, data_only=True)
|
||
|
||
for sheet_name in wb.sheetnames:
|
||
if sheet_name.lower() in _SKIP_SHEETS:
|
||
continue
|
||
|
||
permit_type = _detect_permit_type(sheet_name)
|
||
if permit_type is None:
|
||
logger.debug("Skipping unknown sheet %r in year %d", sheet_name, year)
|
||
continue
|
||
|
||
sheet = wb[sheet_name]
|
||
data_start = _detect_header_row(sheet)
|
||
logger.info(
|
||
"Parsing sheet %r (%s) year=%d, data starts at row %d",
|
||
sheet_name,
|
||
permit_type,
|
||
year,
|
||
data_start,
|
||
)
|
||
yield from self._parse_sheet(sheet, permit_type, year, source_url, data_start)
|
||
|
||
def _parse_sheet(
|
||
self,
|
||
sheet: Any,
|
||
permit_type: str,
|
||
year: int,
|
||
source_url: str,
|
||
data_start: int,
|
||
) -> Iterator[PermitRow]:
|
||
"""Parse один лист → yield PermitRow per data row."""
|
||
row_count = 0
|
||
skip_count = 0
|
||
|
||
for row_idx, row in enumerate(
|
||
sheet.iter_rows(min_row=data_start, values_only=True), start=data_start
|
||
):
|
||
if not _is_data_row(row):
|
||
skip_count += 1
|
||
continue
|
||
|
||
try:
|
||
permit = _parse_row(row, permit_type, year, source_url)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"Failed to parse row %d sheet %s year %d: %s",
|
||
row_idx,
|
||
permit_type,
|
||
year,
|
||
exc,
|
||
)
|
||
skip_count += 1
|
||
continue
|
||
|
||
if permit is None:
|
||
skip_count += 1
|
||
continue
|
||
|
||
row_count += 1
|
||
yield permit
|
||
|
||
logger.info(
|
||
"Sheet %s year=%d: parsed=%d skipped=%d",
|
||
permit_type,
|
||
year,
|
||
row_count,
|
||
skip_count,
|
||
)
|