lekss361
1e53416277
Co-authored-by: lekss361 <lekss361@gendsgn.local> Co-committed-by: lekss361 <lekss361@gendsgn.local>
416 lines
18 KiB
Python
416 lines
18 KiB
Python
"""Парсер ТЭП из пояснительных записок ППТ/ПМТ (#1085, GG-форсайт #1067).
|
||
|
||
Источник: PDF-документы планировочных проектов (ППТ/ПМТ) Екатеринбурга.
|
||
Геометрия ППТ/ПМТ уже в БД (planning_projects, 1170 объектов, WFS).
|
||
|
||
Три целевые таблицы:
|
||
Табл.11 — Баланс территории по зонам (га, %);
|
||
Табл.12 — Основные технико-экономические показатели;
|
||
Табл.13 — Очередность строительства (зона / состав / финансирование).
|
||
|
||
Особенности:
|
||
- Таблицы обнаруживаются по заголовку-caption через extract_text() на каждой странице.
|
||
- Каждая таблица может занимать несколько страниц (особенно Табл.12 и 13).
|
||
- Структура ППТ разнится — парсер работает best-effort:
|
||
не найдена таблица → соответствующее поле = [] (не исключение).
|
||
- Кириллица: используем ТОЛЬКО pdfplumber (pdftotext/poppler даёт mojibake).
|
||
- Нормализация числовых значений: «15,41 га», «(42,61 %)» → float.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import logging
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
import pdfplumber
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Нормализация значений ────────────────────────────────────────────────────
|
||
|
||
_RE_NUM = re.compile(r"[\d\s]+(?:[.,]\d+)?")
|
||
|
||
|
||
def _norm_value(raw: str | None) -> str:
|
||
"""Очищаем ячейку: убираем лишние пробелы, переносы, управляющие символы."""
|
||
if raw is None:
|
||
return ""
|
||
# Убираем управляющие глифы (например — «»)
|
||
text = re.sub(r"[-]", "", raw)
|
||
# Нормализуем переносы строк внутри ячейки → пробел
|
||
text = re.sub(r"\s*\n\s*", " ", text)
|
||
return text.strip()
|
||
|
||
|
||
def _try_float(value: str) -> float | None:
|
||
"""Пытаемся превратить строку вида '15,41 га (42,61 %)' в float первого числа."""
|
||
m = re.search(r"(\d[\d\s]*(?:[.,]\d+)?)", value)
|
||
if not m:
|
||
return None
|
||
num_str = m.group(1).replace(" ", "").replace(",", ".")
|
||
try:
|
||
return float(num_str)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
# ── Обнаружение таблиц ───────────────────────────────────────────────────────
|
||
|
||
# Паттерны для определения номера таблицы в тексте страницы.
|
||
# Ищем «Таблица 11», «Таблица 12», «Таблица 13» (латинская «а» вместо кириллической
|
||
# тоже встречается в некоторых PDF — поэтому без локали).
|
||
_TABLE_CAPTION_RE = {
|
||
11: re.compile(r"Таблица\s+11\b", re.IGNORECASE),
|
||
12: re.compile(r"Таблица\s+12\b", re.IGNORECASE),
|
||
13: re.compile(r"Таблица\s+13\b", re.IGNORECASE),
|
||
}
|
||
|
||
|
||
def _page_contains_table(text: str, table_no: int) -> bool:
|
||
"""Возвращает True если текст страницы содержит заголовок таблицы N.
|
||
|
||
Требуем явный caption «Таблица N» — hint-только режим (оглавление, перекрёстные ссылки)
|
||
даёт false positive и подавляется. Если caption присутствует — достаточно.
|
||
"""
|
||
cap = _TABLE_CAPTION_RE[table_no]
|
||
return bool(cap.search(text))
|
||
|
||
|
||
# ── Парсинг конкретных таблиц ────────────────────────────────────────────────
|
||
|
||
|
||
def _parse_table11(raw_rows: list[list[str | None]]) -> list[dict[str, Any]]:
|
||
"""Баланс территории по зонам (Табл.11).
|
||
|
||
Структура: № | Наименование зоны | Сущ., га (%) | Проект, га (%)
|
||
Заголовок (первые 2 строки) пропускается по признаку: №/None в col[0].
|
||
"""
|
||
result: list[dict[str, Any]] = []
|
||
header_skipped = False
|
||
|
||
for row in raw_rows:
|
||
if not row:
|
||
continue
|
||
no = _norm_value(row[0] if len(row) > 0 else None)
|
||
zone = _norm_value(row[1] if len(row) > 1 else None)
|
||
existing = _norm_value(row[2] if len(row) > 2 else None)
|
||
project = _norm_value(row[3] if len(row) > 3 else None)
|
||
|
||
# Строка-заголовок: col[0] в {№, None, ''} и col[1] содержит
|
||
# «Наименование» или «Площадь» — пропускаем.
|
||
if not header_skipped:
|
||
zone_lo = zone.lower()
|
||
if no in ("", "№", None) and ("наименование" in zone_lo or "площадь" in zone_lo):
|
||
header_skipped = True
|
||
continue
|
||
if no in ("", "№", None) and not zone:
|
||
continue
|
||
|
||
# Пропускаем строки без номера — вложенные подзаголовки.
|
||
# Итоговую строку включаем (распознаём по «Итого»).
|
||
is_total = "итого" in (no + zone).lower()
|
||
|
||
if not no and not is_total:
|
||
continue
|
||
|
||
record: dict[str, Any] = {
|
||
"no": no,
|
||
"zone_name": zone,
|
||
"existing_ha": _norm_value(existing),
|
||
"existing_ha_num": _try_float(existing),
|
||
"project_ha": _norm_value(project),
|
||
"project_ha_num": _try_float(project),
|
||
}
|
||
if is_total:
|
||
record["is_total"] = True
|
||
result.append(record)
|
||
|
||
return result
|
||
|
||
|
||
def _parse_table12(raw_rows: list[list[str | None]]) -> list[dict[str, Any]]:
|
||
"""Основные ТЭП (Табл.12).
|
||
|
||
Структура: № | Наименование показателей | Единица измерения |
|
||
Современное состояние | Расчётный срок.
|
||
Таблица может содержать дополнительный столбец «I этап»
|
||
(на страницах 112-113 этапы разнесены).
|
||
"""
|
||
result: list[dict[str, Any]] = []
|
||
header_skipped = False
|
||
|
||
for row in raw_rows:
|
||
if not row:
|
||
continue
|
||
# Чистим все ячейки
|
||
cells = [_norm_value(c) for c in row]
|
||
if not any(cells):
|
||
continue
|
||
|
||
no = cells[0] if cells else ""
|
||
name = cells[1] if len(cells) > 1 else ""
|
||
unit = cells[2] if len(cells) > 2 else ""
|
||
current = cells[3] if len(cells) > 3 else ""
|
||
target = cells[4] if len(cells) > 4 else ""
|
||
|
||
# Пропуск строки-заголовка таблицы
|
||
if not header_skipped:
|
||
if "наименование" in name.lower() or "показател" in name.lower():
|
||
header_skipped = True
|
||
continue
|
||
if no in ("", "№") and ("единица" in unit.lower() or "измерени" in unit.lower()):
|
||
header_skipped = True
|
||
continue
|
||
|
||
# Пропускаем полностью пустые строки
|
||
if not (no or name or current or target):
|
||
continue
|
||
|
||
record: dict[str, Any] = {
|
||
"no": no,
|
||
"indicator": name,
|
||
"unit": unit,
|
||
"current": current,
|
||
"target": target,
|
||
}
|
||
# Если 6 колонок — есть этапная разбивка (I этап / II этап)
|
||
if len(cells) > 5:
|
||
record["stage1"] = cells[4]
|
||
record["target"] = cells[5]
|
||
|
||
result.append(record)
|
||
|
||
return result
|
||
|
||
|
||
def _parse_table13(raw_rows: list[list[str | None]]) -> list[dict[str, Any]]:
|
||
"""Очередность строительства (Табл.13).
|
||
|
||
Структура сложная (20 колонок из-за merged cells):
|
||
Нас интересуют: фаза (col[0]), состав объекта (col[1..2]), зона (col[13]),
|
||
площадь (col[15]), финансирование (col[17]), программа (col[19]).
|
||
Извлекаем только строки с непустым полем «зона» или «состав объекта».
|
||
"""
|
||
result: list[dict[str, Any]] = []
|
||
current_phase: str = ""
|
||
header_skipped = False
|
||
|
||
for row in raw_rows:
|
||
if not row:
|
||
continue
|
||
cells = [_norm_value(c) for c in row]
|
||
|
||
# Пропуск заголовка (первые 2 строки)
|
||
if not header_skipped:
|
||
first_meaningful = next((c for c in cells if c), "")
|
||
if "номер" in first_meaningful.lower() or "состав" in first_meaningful.lower():
|
||
header_skipped = True
|
||
continue
|
||
# Строка с «1 2 4 5 6 7» (номера колонок)
|
||
if first_meaningful in ("1", ""):
|
||
continue
|
||
|
||
# Определяем фазу/этап строительства (ячейка 0, начинается с «I», «II», «III»).
|
||
col0 = cells[0] if cells else ""
|
||
if re.match(r"^(I{1,3}V?|IV|V)\s*(этап|очередь)?", col0, re.IGNORECASE) and col0:
|
||
current_phase = col0
|
||
|
||
# Состав объекта — из колонок 1 или 3 (после слияния).
|
||
# При 20-колоночной таблице ищем непустой текст в диапазоне 1..5.
|
||
composition = ""
|
||
for ci in range(1, min(6, len(cells))):
|
||
if cells[ci]:
|
||
composition = cells[ci]
|
||
break
|
||
|
||
# Зона планируемого размещения.
|
||
zone = ""
|
||
area_ha_str = ""
|
||
funding = ""
|
||
program = ""
|
||
|
||
if len(cells) >= 14:
|
||
zone = cells[13]
|
||
if len(cells) >= 16:
|
||
area_ha_str = cells[15]
|
||
if len(cells) >= 18:
|
||
funding = cells[17]
|
||
if len(cells) >= 20:
|
||
program = cells[19]
|
||
|
||
# Пропускаем строки без осмысленного содержания
|
||
if not (composition or zone):
|
||
continue
|
||
# Пропускаем строки-секционные заголовки (нет зоны, состав = подзаголовок раздела)
|
||
if not zone and "объект" in composition.lower() and len(composition) < 50:
|
||
continue
|
||
|
||
result.append(
|
||
{
|
||
"phase": current_phase,
|
||
"composition": composition,
|
||
"zone": zone,
|
||
"area_ha": area_ha_str,
|
||
"area_ha_num": _try_float(area_ha_str),
|
||
"funding": funding,
|
||
"program": program,
|
||
}
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
# ── Сборка сырых строк со всех страниц блока таблицы ────────────────────────
|
||
|
||
|
||
def _collect_table_rows(
|
||
pages: list[pdfplumber.page.Page],
|
||
start_page_idx: int,
|
||
table_no: int,
|
||
next_table_no: int | None,
|
||
) -> tuple[list[list[str | None]], list[dict[str, Any]]]:
|
||
"""Собирает строки таблицы N начиная со страницы start_page_idx.
|
||
|
||
Останавливается когда встречает заголовок следующей таблицы (next_table_no)
|
||
или «Рис.» / другой раздел.
|
||
|
||
Returns:
|
||
(all_rows, raw_table_records) — строки для парсинга и сырые данные для
|
||
поля raw_tables.
|
||
"""
|
||
all_rows: list[list[str | None]] = []
|
||
raw_records: list[dict[str, Any]] = []
|
||
found_start = False
|
||
|
||
for pi in range(start_page_idx, len(pages)):
|
||
page = pages[pi]
|
||
page_text = page.extract_text() or ""
|
||
|
||
# Обнаружение начала нашей таблицы
|
||
if not found_start:
|
||
if _page_contains_table(page_text, table_no):
|
||
found_start = True
|
||
else:
|
||
continue
|
||
|
||
# Стоп-условие: начался следующий раздел/таблица
|
||
if found_start and pi > start_page_idx:
|
||
if next_table_no and _page_contains_table(page_text, next_table_no):
|
||
break
|
||
|
||
# Извлекаем таблицы со страницы
|
||
page_tables = page.extract_tables()
|
||
for ti, tbl in enumerate(page_tables):
|
||
if not tbl:
|
||
continue
|
||
# Берём только таблицы с ≥2 строками и ≥2 колонками
|
||
if len(tbl) < 2 or (tbl[0] and len(tbl[0]) < 2):
|
||
continue
|
||
|
||
# На первой странице таблицы пропускаем маленькие таблицы (заголовки/рамки)
|
||
# — они идут ДО основной таблицы данных.
|
||
if pi == start_page_idx and len(tbl) < 4:
|
||
continue
|
||
|
||
all_rows.extend(tbl)
|
||
raw_records.append(
|
||
{
|
||
"page": pi + 1, # 1-based для отладки
|
||
"table_index": ti,
|
||
"rows": [[_norm_value(c) for c in row] for row in tbl[:10]], # cap 10 rows
|
||
}
|
||
)
|
||
|
||
return all_rows, raw_records
|
||
|
||
|
||
# ── Публичный API ────────────────────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class PptTep:
|
||
"""Результат парсинга пояснительной записки ППТ/ПМТ.
|
||
|
||
Каждое поле — list[dict] с нормализованными строками таблицы.
|
||
Пустой список означает «таблица не найдена» (best-effort, не ошибка).
|
||
"""
|
||
|
||
zone_balance: list[dict[str, Any]] = field(default_factory=list)
|
||
tep: list[dict[str, Any]] = field(default_factory=list)
|
||
phasing: list[dict[str, Any]] = field(default_factory=list)
|
||
raw_tables: list[dict[str, Any]] = field(default_factory=list)
|
||
|
||
|
||
def parse_ppt_tep(pdf_bytes: bytes) -> PptTep:
|
||
"""Парсит PDF пояснительной записки ППТ/ПМТ → ТЭП (Табл.11/12/13).
|
||
|
||
Чистая функция от bytes; сеть/fetch — вне этой функции.
|
||
Graceful при отсутствии таблиц: поле = [] вместо исключения.
|
||
|
||
Args:
|
||
pdf_bytes: сырые байты PDF-документа.
|
||
|
||
Returns:
|
||
PptTep с заполненными полями zone_balance / tep / phasing / raw_tables.
|
||
"""
|
||
result = PptTep()
|
||
|
||
try:
|
||
with pdfplumber.open(io.BytesIO(pdf_bytes)) as pdf:
|
||
pages = pdf.pages
|
||
n_pages = len(pages)
|
||
logger.info("parse_ppt_tep: opened PDF, %d pages", n_pages)
|
||
|
||
# Находим страницы-начала Табл.11/12/13
|
||
page_11 = page_12 = page_13 = None
|
||
for pi, page in enumerate(pages):
|
||
text = page.extract_text() or ""
|
||
if page_11 is None and _page_contains_table(text, 11):
|
||
page_11 = pi
|
||
if page_12 is None and _page_contains_table(text, 12):
|
||
page_12 = pi
|
||
if page_13 is None and _page_contains_table(text, 13):
|
||
page_13 = pi
|
||
|
||
logger.info(
|
||
"parse_ppt_tep: Табл.11 на стр.%s, Табл.12 на стр.%s, Табл.13 на стр.%s",
|
||
page_11 and page_11 + 1,
|
||
page_12 and page_12 + 1,
|
||
page_13 and page_13 + 1,
|
||
)
|
||
|
||
# Табл.11 — Баланс территории
|
||
if page_11 is not None:
|
||
rows_11, raw_11 = _collect_table_rows(pages, page_11, 11, 12)
|
||
result.zone_balance = _parse_table11(rows_11)
|
||
result.raw_tables.extend(raw_11)
|
||
logger.info("parse_ppt_tep: Табл.11 → %d строк", len(result.zone_balance))
|
||
else:
|
||
logger.warning("parse_ppt_tep: Таблица 11 не найдена в PDF")
|
||
|
||
# Табл.12 — ТЭП
|
||
if page_12 is not None:
|
||
rows_12, raw_12 = _collect_table_rows(pages, page_12, 12, 13)
|
||
result.tep = _parse_table12(rows_12)
|
||
result.raw_tables.extend(raw_12)
|
||
logger.info("parse_ppt_tep: Табл.12 → %d строк", len(result.tep))
|
||
else:
|
||
logger.warning("parse_ppt_tep: Таблица 12 не найдена в PDF")
|
||
|
||
# Табл.13 — Очередность строительства
|
||
if page_13 is not None:
|
||
rows_13, raw_13 = _collect_table_rows(pages, page_13, 13, None)
|
||
result.phasing = _parse_table13(rows_13)
|
||
result.raw_tables.extend(raw_13)
|
||
logger.info("parse_ppt_tep: Табл.13 → %d строк", len(result.phasing))
|
||
else:
|
||
logger.warning("parse_ppt_tep: Таблица 13 не найдена в PDF")
|
||
|
||
except Exception as exc:
|
||
logger.error("parse_ppt_tep: ошибка парсинга PDF: %s", exc)
|
||
raise
|
||
|
||
return result
|