bot-backend
5d34bb7833
Прод-баг #2126: water_supply_reserves содержал мусор — резерв = № п/п формы (1.1, 1.3, 2.18…), а рубрики анкеты («Количество поданных…»), «Итого:» и шапки попадали в поле system_name. Итог §3 показывал «водоснабжение 1,1 / водоотведение 3,0 тыс. м³/сут» (номера строк, не резервы). Корень: extract_water_rows брал «первую текстовую ячейку» как имя и «первое число строки» как резерв. РЕАЛЬНАЯ сетка (снята с прода 2026-07-03 через docker exec тем же путём, что лоадер) устроена иначе, причём две страницы имеют РАЗНЫЙ layout: - водоснабжение: 6 колонок, каждый ЦСВ = 4 строки показателей; резерв — строка, где col2 = «Резерв мощности централизованной системы…», значение в col4; - водоотведение: 5 колонок, имя системы — отдельная строка-заголовок («Централизованная система…»/«Транспортировка сточных вод…»), резерв в col3. Переписал парсинг column-aligned по подписям шапки и метке показателя: детект layout по строке-шапке, резерв ТОЛЬКО из строки «Резерв мощности…» (+ защита ед.изм. «тыс…м3/сут»), заявки/отказы — из своих рубрик. Шапки, «Итого:», легенда отфильтрованы. Нули («Отсутствие резерва мощности», 0,0) сохраняются честно (0.0 + note), как в eias_heat_loader после того же класса бага. Дедуп байт-в-байт одинаковых имён суффиксом « (N)» (защита UNIQUE), warning при 0 записей. Прод-verify нового парсера на живом docx: water → 33 ЦСВ (ЗФС,ГСВ −168,5; Изоплит 0,0; Шабровский 1,03; ПТВ 73,34), sewerage → 6 систем (Южный бассейн −296,85; Северный 2,02). Ни один резерв больше не равен № п/п. Тесты: фикстуры из РЕАЛЬНОЙ сетки обоих layout (шапка + ЦСВ + рубрики + «Итого:» + легенда + ноль-резерв); ассерты — резервы из правильной колонки, мусор отфильтрован, нули сохранены; end-to-end через load_water_reserves_from_docx. 41 passed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
521 lines
25 KiB
Python
521 lines
25 KiB
Python
"""Загрузчик резервов ЦСВ/ЦСК Водоканала ЕКБ из docx-раскрытия (#2119, фикс #2126).
|
||
|
||
Две страницы водоканалекб.рф (водоснабжение / водоотведение) публикуют docx с
|
||
таблицами свободной мощности ЦСВ (центр. система водоснабжения) / ЦСК (центр.
|
||
система водоотведения). UPSERT-ит в ``water_supply_reserves``.
|
||
|
||
Источник ГЕО-БЛОКИРУЕТ non-RU IP → РАБОТАЕТ НА ПРОДЕ (Celery / docker exec).
|
||
Punycode-host обязателен (водоканалекб.рф → xn--80aadbki6adhshb.xn--p1ai).
|
||
|
||
Парсим таблицы через stdlib (zipfile + ElementTree по word/document.xml) — это
|
||
надёжнее python-docx для vMerge-детекта и тривиально тестируется на синтетическом
|
||
xml-фрагменте. Резерв бывает ОТРИЦАТЕЛЬНЫМ (дефицит, напр. −168,5) — знак сохраняем,
|
||
запятая-десятичный → float. period инферится из имени файла (1kv.-2026 → '2026-Q1').
|
||
|
||
РЕАЛЬНАЯ СЕТКА (снята с прода 2026-07-03, обе страницы — 1 таблица, но РАЗНЫЙ layout).
|
||
Парсинг COLUMN-ALIGNED по ПОДПИСЯМ ШАПКИ и метке показателя (НЕ по индексам вслепую):
|
||
|
||
* Страница «водоснабжение» (system_kind='water') — 6 колонок, «длинный» layout:
|
||
каждый ЦСВ занимает 4 СТРОКИ (по строке на показатель). Шапка: col2=='Наименование
|
||
показателя'. col0='№ ЦСВ' (номер п/п: 1.1, 2.1…2.26, ПТВ), col1='Название ЦСВ'
|
||
(имя системы, повторяется в каждой строке блока), col2=показатель, col3=ед.изм,
|
||
col4=значение периода, col5=примечание. РЕЗЕРВ = строка, где col2 начинается с
|
||
«Резерв мощности централизованной системы…» И col3 — «тыс…м3/сут» → число из col4.
|
||
Заявки/отказы — строки «Количество поданных…» / «…отказе в подключении».
|
||
Живые резервы: ЗФС,ГСВ = −168,5; Изоплит = 0,0; Шабровский = 1,03; ПТВ = 73,34.
|
||
* Страница «водоотведение» (system_kind='sewerage') — 5 колонок, «блочный» layout:
|
||
имя системы — ОТДЕЛЬНАЯ строка-заголовок (только col0, напр. «Централизованная
|
||
система водоотведения "Южный бассейн канализования"»), под ней строка-шапка
|
||
(col0='№ п/п') и 4 строки показателей (col0 = 1..4 ИЛИ пусто). РЕЗЕРВ = строка,
|
||
где col1 начинается с «Резерв мощности централизованной системы…» И col2 —
|
||
«тыс…м3/сут» → число из col3. Имя системы forward-fill-ится из последней
|
||
строки-заголовка. Живые резервы: Южный бассейн = −296,85; Северный = 2,02.
|
||
|
||
ЛОВУШКА (прод-баг #2126, был мусор 28+6 строк): старый парсер брал «первое число
|
||
строки» как резерв и «первую текстовую ячейку» как имя. На проде это давало № п/п
|
||
(1.1, 2.18…) в поле резерва и рубрики анкеты («Количество поданных…», «Итого:»,
|
||
служебные шапки «№ ЦСВ»/«№ п/п») как «системы». Column-aligned парс по метке
|
||
показателя их не видит: строки-рубрики, «Итого:», шапки и легенда отфильтрованы;
|
||
резерв читается ТОЛЬКО из строки «Резерв мощности…». НУЛИ («Отсутствие резерва
|
||
мощности», 0,0) сохраняем честно (0.0 + note), как в eias_heat_loader.
|
||
"""
|
||
|
||
import io
|
||
import logging
|
||
import re
|
||
import zipfile
|
||
from xml.etree import ElementTree as ET
|
||
|
||
import httpx
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# WordprocessingML namespace.
|
||
_W = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
|
||
_NS = {"w": _W}
|
||
|
||
# Punycode-host обязателен (водоканалекб.рф). xn--80aadbki6adhshb.xn--p1ai.
|
||
_BASE = "https://xn--80aadbki6adhshb.xn--p1ai"
|
||
PAGES: dict[str, str] = {
|
||
"water": f"{_BASE}/info/vodosnabzhenie/",
|
||
"sewerage": f"{_BASE}/info/vodotvedenie/",
|
||
}
|
||
|
||
_HTTP_TIMEOUT = 60
|
||
|
||
# docx-ссылка «для сайта»: dlya-sajta-*.docx.
|
||
_DOCX_HREF_RE = re.compile(r'href=["\']([^"\']*dlya-sajta-[^"\']*\.docx)["\']', re.IGNORECASE)
|
||
# Дата в имени файла для выбора свежайшего.
|
||
_FILE_DATE_RE = re.compile(r"(\d{4})|(\d{1,2})[.\-](\d{4})")
|
||
# Период вида «1kv.-2026» / «1кв-2026» / «1 квартал 2026».
|
||
_PERIOD_RE = re.compile(r"(\d)\s*(?:kv|кв)\w*[.\s-]*(\d{4})", re.IGNORECASE)
|
||
|
||
_NODATA_TOKENS = frozenset({"", "н/д", "нд", "n/a", "-", "—", "–", "нет данных", "х", "x"})
|
||
|
||
# Метки показателей формы раскрытия (нормализуем casefold + схлопнутые пробелы перед
|
||
# сравнением: в docx встречаются \xa0 и двойные пробелы). Резерв — по префиксу, т.к.
|
||
# хвост различается («…холодного водоснабжения» / «…водоотведения»).
|
||
_RESERVE_LABEL_PREFIX = "резерв мощности централизованной системы"
|
||
_REQUESTED_LABEL = "количество поданных и зарегистрированных заявок на подключение"
|
||
_REFUSED_LABEL_PREFIX = "количество заявок на подключение, по которым принято решение об отказе"
|
||
# Ед.изм. резерва: «тыс. м3/сут» / «тыс.м3/сут» (пробелы схлопнуты) — защита от того,
|
||
# чтобы «Резерв…» из чужой формы с иной размерностью не попал в тыс.м³/сут.
|
||
_RESERVE_UNIT_MARKER = "тыс"
|
||
# Шапки таблицы и служебные строки — не данные.
|
||
_HEADER_MARKERS = frozenset({"№ п/п", "№ цсв", "наименование показателя", "название цсв"})
|
||
_TOTAL_MARKER = "итого"
|
||
# Строка-заголовок системы в «блочном» layout водоотведения: только col0 заполнен и
|
||
# начинается с одного из этих префиксов (сама система, а не показатель/шапка).
|
||
_SYSTEM_TITLE_PREFIXES = (
|
||
"централизованная система",
|
||
"транспортировка сточных вод",
|
||
)
|
||
# Легенда в конце водоснабжения: «ЦСВ – централизованная система…» / «ПТВ – …».
|
||
_LEGEND_MARKER = " – "
|
||
|
||
|
||
def parse_water_number(value: str | None) -> float | None:
|
||
"""Числовая ячейка резерва тыс. м³/сут → float. Сохраняет знак (дефицит <0).
|
||
|
||
Запятая-десятичный → точка, минусы всех видов (−/–/-) нормализуются.
|
||
«н/д»/пусто → None.
|
||
"""
|
||
if value is None:
|
||
return None
|
||
s = str(value).strip().lower()
|
||
if s in _NODATA_TOKENS:
|
||
return None
|
||
s = s.replace("\xa0", "").replace(" ", "").replace(",", ".")
|
||
# Нормализуем юникод-минусы к ASCII '-'.
|
||
s = s.replace("−", "-").replace("–", "-").replace("—", "-")
|
||
m = re.search(r"-?\d+(?:\.\d+)?", s)
|
||
if not m:
|
||
return None
|
||
try:
|
||
return float(m.group())
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def infer_period(filename: str) -> str | None:
|
||
"""Инферит период отчёта из имени файла: «1kv.-2026» → '2026-Q1'. None — нет."""
|
||
m = _PERIOD_RE.search(filename)
|
||
if not m:
|
||
return None
|
||
quarter, year = m.group(1), m.group(2)
|
||
return f"{year}-Q{quarter}"
|
||
|
||
|
||
def _cell_text(tc: ET.Element) -> str:
|
||
"""Весь текст ячейки таблицы (<w:tc>) — конкатенация всех <w:t>."""
|
||
parts = [t.text or "" for t in tc.iter(f"{{{_W}}}t")]
|
||
return "".join(parts).strip()
|
||
|
||
|
||
def _vmerge_state(tc: ET.Element) -> str:
|
||
"""Состояние вертикального merge ячейки: 'restart' | 'continue' | 'none'.
|
||
|
||
<w:vMerge w:val="restart"/> — начало блока (значение здесь).
|
||
<w:vMerge/> (без val или val='continue') — продолжение (forward-fill сверху).
|
||
Нет <w:vMerge> — обычная ячейка.
|
||
"""
|
||
tc_pr = tc.find(f"{{{_W}}}tcPr")
|
||
if tc_pr is None:
|
||
return "none"
|
||
vmerge = tc_pr.find(f"{{{_W}}}vMerge")
|
||
if vmerge is None:
|
||
return "none"
|
||
val = vmerge.get(f"{{{_W}}}val")
|
||
return "restart" if val == "restart" else "continue"
|
||
|
||
|
||
def parse_docx_table_rows(document_xml: bytes) -> list[list[str]]:
|
||
"""Парсит ПЕРВУЮ таблицу document.xml → list строк (list ячеек-строк).
|
||
|
||
Forward-fill по вертикальному merge: ячейка в состоянии 'continue' наследует
|
||
текст ячейки той же колонки из предыдущей строки (system-name + note мержатся).
|
||
Возвращает матрицу текстов [row][col]. Пусто → [].
|
||
"""
|
||
root = ET.fromstring(document_xml)
|
||
tables = root.iter(f"{{{_W}}}tbl")
|
||
tbl = next(tables, None)
|
||
if tbl is None:
|
||
return []
|
||
|
||
matrix: list[list[str]] = []
|
||
prev_row: list[str] = []
|
||
for tr in tbl.findall(f"{{{_W}}}tr"):
|
||
cells = tr.findall(f"{{{_W}}}tc")
|
||
row: list[str] = []
|
||
for col_idx, tc in enumerate(cells):
|
||
state = _vmerge_state(tc)
|
||
if state == "continue":
|
||
# forward-fill из той же колонки предыдущей строки.
|
||
filled = prev_row[col_idx] if col_idx < len(prev_row) else ""
|
||
row.append(filled)
|
||
else:
|
||
row.append(_cell_text(tc))
|
||
matrix.append(row)
|
||
prev_row = row
|
||
return matrix
|
||
|
||
|
||
def _norm(s: str | None) -> str:
|
||
"""Ячейку → casefold + схлопнутые пробелы (\xa0/двойные) для сравнения с метками."""
|
||
if not s:
|
||
return ""
|
||
return re.sub(r"\s+", " ", s.replace("\xa0", " ")).strip().casefold()
|
||
|
||
|
||
def _cell(row: list[str], idx: int) -> str:
|
||
"""Ячейка row[idx] со .strip() или '' если колонки нет."""
|
||
return row[idx].strip() if idx < len(row) and row[idx] else ""
|
||
|
||
|
||
def _is_header_row(row: list[str]) -> bool:
|
||
"""Строка-шапка таблицы (№ п/п / Наименование показателя / …)?"""
|
||
return any(_norm(c) in _HEADER_MARKERS for c in row)
|
||
|
||
|
||
def _is_total_row(row: list[str]) -> bool:
|
||
"""Строка «Итого:» (агрегат по всем системам — не резерв конкретной ЦСВ)."""
|
||
return _norm(_cell(row, 0)).startswith(_TOTAL_MARKER)
|
||
|
||
|
||
def _row_metric(row: list[str], label_col: int) -> str:
|
||
"""Метка показателя строки (нормализованная) из колонки label_col."""
|
||
return _norm(_cell(row, label_col))
|
||
|
||
|
||
def _is_reserve_metric(metric: str) -> bool:
|
||
return metric.startswith(_RESERVE_LABEL_PREFIX)
|
||
|
||
|
||
def _is_requested_metric(metric: str) -> bool:
|
||
return metric == _REQUESTED_LABEL
|
||
|
||
|
||
def _is_refused_metric(metric: str) -> bool:
|
||
return metric.startswith(_REFUSED_LABEL_PREFIX)
|
||
|
||
|
||
def _system_title(row: list[str]) -> str | None:
|
||
"""Строка-заголовок системы «блочного» layout (водоотведение): имя или None.
|
||
|
||
Заполнена ТОЛЬКО col0, текст начинается с «Централизованная система…» /
|
||
«Транспортировка сточных вод…». Возвращает имя системы (col0.strip()).
|
||
"""
|
||
if any(_cell(row, i) for i in range(1, len(row))):
|
||
return None # не одиночный заголовок — заполнены и другие колонки
|
||
text0 = _cell(row, 0)
|
||
norm0 = _norm(text0)
|
||
if any(norm0.startswith(p) for p in _SYSTEM_TITLE_PREFIXES):
|
||
return text0
|
||
return None
|
||
|
||
|
||
def _detect_layout(matrix: list[list[str]]) -> tuple[str, int, int, int]:
|
||
"""По шапке определяет layout → (kind, name_col, label_col, value_col).
|
||
|
||
kind='water' : шапка с col2=='Наименование показателя', имя ЦСВ в col1,
|
||
показатель col2, значение col4.
|
||
kind='block' : шапка с col1=='Наименование показателя' (водоотведение),
|
||
имя системы — отдельная строка-заголовок, значение col3.
|
||
Не нашли шапку → ('', -1, -1, -1) (graceful).
|
||
"""
|
||
for row in matrix:
|
||
if _norm(_cell(row, 2)) == "наименование показателя":
|
||
return ("water", 1, 2, 4)
|
||
if _norm(_cell(row, 1)) == "наименование показателя":
|
||
return ("block", -1, 1, 3)
|
||
return ("", -1, -1, -1)
|
||
|
||
|
||
def _flush_group(
|
||
name: str,
|
||
reserve_cell: str | None,
|
||
note_cell: str,
|
||
requested: int | None,
|
||
refused: int | None,
|
||
records: list[dict],
|
||
) -> None:
|
||
"""Собирает запись ЦСВ/ЦСК из накопленной группы строк, если есть имя + резерв."""
|
||
if not name or reserve_cell is None:
|
||
return
|
||
reserve = parse_water_number(reserve_cell)
|
||
# note: явное примечание строки резерва (напр. «Отсутствие резерва мощности»)
|
||
# честно фиксирует ноль-резерв; иначе None.
|
||
note = note_cell.strip() or None
|
||
records.append(
|
||
{
|
||
"system_name": name,
|
||
"reserve_thousand_m3_day": reserve,
|
||
"connections_requested": requested,
|
||
"connections_refused": refused,
|
||
"note": note,
|
||
}
|
||
)
|
||
|
||
|
||
def extract_water_rows(matrix: list[list[str]]) -> list[dict]:
|
||
"""Из матрицы docx → записи ЦСВ/ЦСК, COLUMN-ALIGNED по метке показателя.
|
||
|
||
Определяет layout по шапке (см. ``_detect_layout``): «water» — имя ЦСВ в col1,
|
||
«block» (водоотведение) — имя в отдельной строке-заголовке. Затем группирует
|
||
строки показателей по системе и берёт резерв ТОЛЬКО из строки «Резерв мощности
|
||
централизованной системы…» (col — value_col), заявки/отказы — из строк
|
||
«Количество поданных…» / «…отказе…». Ноль-резерв (0,0 + «Отсутствие резерва
|
||
мощности») сохраняется честно. Шапки, «Итого:», легенда — отфильтрованы.
|
||
|
||
Не распознан layout → [] (graceful, лог наверху).
|
||
"""
|
||
kind, name_col, label_col, value_col = _detect_layout(matrix)
|
||
if not kind:
|
||
return []
|
||
|
||
records: list[dict] = []
|
||
# Накопитель текущей группы (одна система = один блок строк показателей).
|
||
cur_name = ""
|
||
cur_reserve: str | None = None
|
||
cur_note = ""
|
||
cur_requested: int | None = None
|
||
cur_refused: int | None = None
|
||
|
||
def flush() -> None:
|
||
nonlocal cur_name, cur_reserve, cur_note, cur_requested, cur_refused
|
||
_flush_group(cur_name, cur_reserve, cur_note, cur_requested, cur_refused, records)
|
||
cur_name, cur_reserve, cur_note = "", None, ""
|
||
cur_requested, cur_refused = None, None
|
||
|
||
for row in matrix:
|
||
if not any(c.strip() for c in row):
|
||
continue
|
||
if _is_header_row(row):
|
||
continue
|
||
if _is_total_row(row):
|
||
flush() # «Итого:» закрывает предыдущую систему, само не данные
|
||
continue
|
||
# Легенда в конце водоснабжения («ЦСВ – централизованная система…»).
|
||
if _norm(_cell(row, 0)).startswith(("цсв ", "птв ")) and _LEGEND_MARKER in _cell(row, 0):
|
||
flush()
|
||
continue
|
||
|
||
if kind == "block":
|
||
title = _system_title(row)
|
||
if title is not None:
|
||
flush() # новая система-заголовок
|
||
cur_name = title
|
||
continue
|
||
|
||
metric = _row_metric(row, label_col)
|
||
if kind == "water":
|
||
# Имя ЦСВ повторяется в каждой строке блока (col1). Смена имени → новая
|
||
# система: сбрасываем накопитель.
|
||
row_name = _cell(row, name_col)
|
||
if row_name and row_name != cur_name:
|
||
flush()
|
||
cur_name = row_name
|
||
|
||
value = _cell(row, value_col)
|
||
if _is_reserve_metric(metric):
|
||
# Защита размерности: у резерва ед.изм. — «тыс…м3/сут» (col label+1).
|
||
unit = _norm(_cell(row, label_col + 1))
|
||
if _RESERVE_UNIT_MARKER in unit:
|
||
cur_reserve = value
|
||
cur_note = _cell(row, value_col + 1) # Примечание строки резерва
|
||
elif _is_requested_metric(metric):
|
||
n = parse_water_number(value)
|
||
if n is not None and float(n).is_integer():
|
||
cur_requested = int(n)
|
||
elif _is_refused_metric(metric):
|
||
n = parse_water_number(value)
|
||
if n is not None and float(n).is_integer():
|
||
cur_refused = int(n)
|
||
|
||
flush() # последняя система
|
||
return records
|
||
|
||
|
||
def fetch_docx_link(page_url: str) -> str:
|
||
"""Скрейпит страницу → ссылка на СВЕЖАЙШИЙ dlya-sajta-*.docx. '' если нет.
|
||
|
||
RUN-ON-PROD (гео-блок). GET follow_redirects, regex по href.
|
||
"""
|
||
resp = httpx.get(page_url, timeout=_HTTP_TIMEOUT, follow_redirects=True)
|
||
resp.raise_for_status()
|
||
hrefs = _DOCX_HREF_RE.findall(resp.text)
|
||
if not hrefs:
|
||
return ""
|
||
|
||
def _key(h: str) -> tuple:
|
||
years = [int(y) for grp in _FILE_DATE_RE.findall(h) for y in grp if y]
|
||
return (max(years) if years else 0, h)
|
||
|
||
best = max(hrefs, key=_key)
|
||
return best if best.startswith("http") else _BASE + best
|
||
|
||
|
||
def _dedupe_names(records: list[dict]) -> list[dict]:
|
||
"""Разводит БАЙТ-В-БАЙТ одинаковые system_name суффиксом « (N)».
|
||
|
||
UNIQUE (system_kind, system_name, period) схлопнул бы две разные системы с
|
||
одинаковым именем в одну (с потерей резерва). Имена ЦСВ на проде обычно уникальны
|
||
(well-суффиксы), но защищаемся — как в eias_heat_loader.
|
||
"""
|
||
seen: dict[str, int] = {}
|
||
for rec in records:
|
||
name = rec["system_name"]
|
||
seen[name] = seen.get(name, 0) + 1
|
||
if seen[name] > 1:
|
||
rec["system_name"] = f"{name} ({seen[name]})"
|
||
return records
|
||
|
||
|
||
def _upsert_water_rows(
|
||
db: Session,
|
||
system_kind: str,
|
||
records: list[dict],
|
||
period: str | None,
|
||
source_url: str,
|
||
) -> dict[str, int]:
|
||
"""UPSERT записей ЦСВ/ЦСК в water_supply_reserves. Per-row SAVEPOINT."""
|
||
inserted = 0
|
||
updated = 0
|
||
skipped = 0
|
||
for rec in records:
|
||
params = {
|
||
"system_kind": system_kind,
|
||
"system_name": rec["system_name"],
|
||
"reserve": rec["reserve_thousand_m3_day"],
|
||
"requested": rec["connections_requested"],
|
||
"refused": rec["connections_refused"],
|
||
"note": rec["note"],
|
||
"period": period,
|
||
"source_url": source_url,
|
||
}
|
||
try:
|
||
with db.begin_nested(): # SAVEPOINT — битая строка не валит батч
|
||
result = db.execute(
|
||
text("""
|
||
INSERT INTO water_supply_reserves
|
||
(system_kind, system_name, reserve_thousand_m3_day,
|
||
connections_requested, connections_refused, note,
|
||
period, source_url, fetched_at)
|
||
VALUES (
|
||
:system_kind, :system_name, :reserve,
|
||
:requested, :refused, :note,
|
||
:period, :source_url, NOW()
|
||
)
|
||
ON CONFLICT (system_kind, system_name, period) DO UPDATE
|
||
SET reserve_thousand_m3_day = EXCLUDED.reserve_thousand_m3_day,
|
||
connections_requested = EXCLUDED.connections_requested,
|
||
connections_refused = EXCLUDED.connections_refused,
|
||
note = EXCLUDED.note,
|
||
source_url = EXCLUDED.source_url,
|
||
fetched_at = NOW()
|
||
RETURNING (xmax = 0) AS is_insert
|
||
"""),
|
||
params,
|
||
).scalar()
|
||
if result:
|
||
inserted += 1
|
||
else:
|
||
updated += 1
|
||
except Exception as e:
|
||
logger.warning("water_reserve upsert failed for %r: %s", rec["system_name"], e)
|
||
skipped += 1
|
||
return {"inserted": inserted, "updated": updated, "skipped": skipped}
|
||
|
||
|
||
def load_water_reserves_from_docx(
|
||
db: Session,
|
||
system_kind: str,
|
||
docx_bytes: bytes,
|
||
source_url: str = "",
|
||
) -> dict[str, int]:
|
||
"""Парсит docx-байты → UPSERT ЦСВ/ЦСК в water_supply_reserves.
|
||
|
||
Выделено из load_water_reserves для юнит-теста на синтетическом docx.
|
||
Читает word/document.xml из zip, forward-fill vMerge, извлечение записей.
|
||
"""
|
||
with zipfile.ZipFile(io.BytesIO(docx_bytes)) as zf:
|
||
document_xml = zf.read("word/document.xml")
|
||
matrix = parse_docx_table_rows(document_xml)
|
||
records = _dedupe_names(extract_water_rows(matrix))
|
||
period = infer_period(source_url)
|
||
if not records:
|
||
logger.warning(
|
||
"water_reserves[%s]: 0 записей распознано (layout не найден или пустая "
|
||
"таблица); первые 6 строк матрицы: %s",
|
||
system_kind,
|
||
matrix[:6],
|
||
)
|
||
|
||
try:
|
||
counts = _upsert_water_rows(db, system_kind, records, period, source_url)
|
||
db.commit()
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.exception("load_water_reserves_from_docx: outer tx rolled back: %s", e)
|
||
raise
|
||
|
||
result = {"records": len(records), **counts, "period": period} # type: ignore[dict-item]
|
||
logger.info("water_reserves[%s] done: %s", system_kind, result)
|
||
return {k: v for k, v in result.items() if isinstance(v, int)}
|
||
|
||
|
||
def load_water_reserves(db: Session | None = None) -> dict[str, dict]:
|
||
"""Полный прогон: обе страницы (водоснабжение/водоотведение) → UPSERT.
|
||
|
||
RUN-ON-PROD. Каждая страница graceful: сбой одной не валит другую.
|
||
Returns: {"water": {...}, "sewerage": {...}}.
|
||
"""
|
||
owns_session = db is None
|
||
if db is None:
|
||
db = SessionLocal()
|
||
|
||
out: dict[str, dict] = {}
|
||
try:
|
||
with httpx.Client(timeout=_HTTP_TIMEOUT, follow_redirects=True) as client:
|
||
for kind, page_url in PAGES.items():
|
||
try:
|
||
link = fetch_docx_link(page_url)
|
||
if not link:
|
||
out[kind] = {"error": "no docx link found"}
|
||
continue
|
||
r = client.get(link)
|
||
r.raise_for_status()
|
||
out[kind] = load_water_reserves_from_docx(db, kind, r.content, link)
|
||
except Exception as e:
|
||
logger.exception("load_water_reserves: %s failed: %s", kind, e)
|
||
out[kind] = {"error": str(e)}
|
||
finally:
|
||
if owns_session:
|
||
db.close()
|
||
|
||
logger.info("load_water_reserves done: %s", out)
|
||
return out
|