349 lines
14 KiB
Python
349 lines
14 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 хранятся как строки — 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,
|
||
)
|