"""ЕКБ РНС/РВЭ 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, )