"""ЕКБ РНС/РВЭ 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 хранятся как строки — CRS неизвестна (предположительно СНСК-66 / СКФ-66 Свердловская обл., local offset system). Геокодирование через construction_address планируется в Phase 3 (отдельный PR). """ 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 logger = logging.getLogger(__name__) 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 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} 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=_to_str(row[6]) if len(row) > 6 else None, raw_coord_y=_to_str(row[7]) if len(row) > 7 else None, 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, )