Kinda fix

Compared our report against the customer's own new-format export for
10.08.2026 (121 columns) over the 124 source files that produced it. Records
and participant rows line up exactly: 11393 records and 22839 rows on both
sides, no record with a different participant count. 90 of the 104 comparable
columns already matched value for value.

Two differences were ours to fix:

- Column 102, the card number in the foreign bank card block, took its value
  from СведенияВнесениеПолучениеНалДС/НомерКарта whenever the foreign card
  block was missing, because short lookup matches the tag НомерКарта in both
  blocks. The reference leaves the column empty there and shows the number in
  column 81 only, which is what we now do. 63 records on this export.
- The operation-document aggregate numbered its entries "1. ", the reference
  uses "1, ". That was 11377 rows out of 11393.

The remaining differences are the reference's own presentation of aggregate
cells and need the customer's word before touching them: it joins ФИО with
", " and keeps a trailing comma for a missing patronymic, spells the document
that proves identity as labelled lines ("Вид документа: 21\nСерия…") where we
use a comma list, and inserts the name of the region decoded from the ОКАТО
code into address cells, which the schema does not carry at all.

Two columns the reference fills and our form has no place for: the reporting
bank's own registration number, БИК and ОКАТО, and the authorised officer's
name, position, phone and e-mail. Both come from ИнфБанк and СлужЧасть.

app/db/base.py

@@ -5,17 +5,35 @@ from sqlalchemy.orm import DeclarativeBase
 
 from app.core.config import settings
 
+# Без явных таймаутов зависший/протухший сокет к Postgres (например, тихо
+# оборванное NAT/LB соединение после долгого простоя — ровно то, что могло
+# случиться во время получасовой обработки батча из 230 файлов) блокирует
+# asyncpg НАВСЕГДА: ни ошибки, ни лога, просто вечное ожидание чтения
+# сокета. Это тот же класс проблемы с asyncio.run()-за-asyncio.run() вокруг
+# общего пула соединений, что уже ловили раньше (см. комментарий про
+# engine.dispose() в opencloud_token_store.py) — тут решаем его с другой
+# стороны: даём зависшему соединению упасть с понятной ошибкой вместо
+# бесконечного ожидания.
+_CONNECT_TIMEOUT_SECONDS = 10
+_COMMAND_TIMEOUT_SECONDS = 15
+
 connect_args: dict = {}
 if settings.DATABASE_SCHEMA:
     connect_args["server_settings"] = {"search_path": settings.DATABASE_SCHEMA}
 if "sqlite" in settings.DATABASE_URL:
     connect_args["check_same_thread"] = False
+else:
+    # timeout/command_timeout — параметры asyncpg.connect(), не SQLAlchemy;
+    # aiosqlite их не понимает, поэтому только для не-sqlite URL.
+    connect_args["timeout"] = _CONNECT_TIMEOUT_SECONDS
+    connect_args["command_timeout"] = _COMMAND_TIMEOUT_SECONDS
 
 engine = create_async_engine(
     settings.DATABASE_URL,
     pool_pre_ping=True,
     pool_size=5,
     max_overflow=10,
+    pool_timeout=_CONNECT_TIMEOUT_SECONDS,
     connect_args=connect_args,
 )
 

app/pipeline/mapping.py

@@ -88,6 +88,10 @@ class _CompiledCandidate:
     normalized: str
     short: str
     strict_scoped: bool
+    # Собственные суффиксы normalized по границе ".", из двух и более
+    # сегментов: "a.b.c.d" -> ("b.c.d", "c.d"). Нужны _pick_value для поиска
+    # через словарь вместо прохода по всем полям строки (см. там).
+    dotted_suffixes: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -117,6 +121,10 @@ class _PayloadLookup:
     normalized_exact_first: dict[str, str]
     normalized_suffix_first: dict[str, str]
     raw_short_first: dict[str, str]
+    # normalized_path (только составные, с ".") -> (позиция первого поля с
+    # непустым значением, значение). Позиция нужна, чтобы при нескольких
+    # совпадениях вернуть то же поле, что и прежний линейный проход.
+    dotted_path_first: dict[str, tuple[int, str]]
 
 
 @dataclass(frozen=True)
@@ -197,11 +205,17 @@ def _compile_candidates(candidates: tuple[str, ...]) -> tuple[_CompiledCandidate
                 other != normalized and other.endswith(f".{normalized}")
                 for other in normalized_candidates
             ),
+            dotted_suffixes=_dotted_suffixes(normalized),
         )
         for candidate, normalized in zip(candidates, normalized_candidates)
     )
 
 
+def _dotted_suffixes(normalized: str) -> tuple[str, ...]:
+    parts = normalized.split(".")
+    return tuple(".".join(parts[start:]) for start in range(1, len(parts) - 1))
+
+
 def _build_rules(mapping_data: tuple[ColumnMeta, ...]) -> tuple[MappingRule, ...]:
     rules: list[MappingRule] = []
     for meta in mapping_data:
@@ -620,10 +634,13 @@ def _build_payload_lookup_from_entries(
     normalized_exact_first: dict[str, str] = {}
     normalized_suffix_first: dict[str, str] = {}
     raw_short_first: dict[str, str] = {}
+    dotted_path_first: dict[str, tuple[int, str]] = {}
 
-    for entry in entries:
+    for position, entry in enumerate(entries):
         if not entry.value:
             continue
+        if "." in entry.normalized_path:
+            dotted_path_first.setdefault(entry.normalized_path, (position, entry.value))
         normalized_exact_first.setdefault(entry.normalized_path, entry.value)
         path_parts = entry.normalized_path.split(".")
         for start_index in range(len(path_parts)):
@@ -639,6 +656,7 @@ def _build_payload_lookup_from_entries(
         normalized_exact_first=normalized_exact_first,
         normalized_suffix_first=normalized_suffix_first,
         raw_short_first=raw_short_first,
+        dotted_path_first=dotted_path_first,
     )
 
 
@@ -791,13 +809,18 @@ def _pick_value(
         if value := lookup.normalized_suffix_first.get(candidate.normalized):
             return value
         if not allow_short_lookup:
-            for entry in lookup.entries:
+            # Ищем первое (в порядке полей) непустое составное поле, путь
-                if (
+            # которого является собственным суффиксом кандидата. Раньше это
-                    entry.value
+            # был проход по ВСЕМ полям строки с endswith на каждый кандидат
-                    and "." in entry.normalized_path
+            # каждой колонки — ~40% времени формирования отчёта (профиль на
-                    and candidate.normalized.endswith(f".{entry.normalized_path}")
+            # 123 реальных файлах). Теперь — несколько обращений к словарю.
-                ):
+            best: tuple[int, str] | None = None
-                    return entry.value
+            for suffix in candidate.dotted_suffixes:
+                found = lookup.dotted_path_first.get(suffix)
+                if found is not None and (best is None or found[0] < best[0]):
+                    best = found
+            if best is not None:
+                return best[1]
             continue
         if value := lookup.values.get(candidate.short):
             return value

app/pipeline/report.py

@@ -164,7 +164,16 @@ class StreamingReportWriter:
         self._text_format = self.workbook.add_format(
             {"text_wrap": True, "valign": "top"}
         )
+        # Явный «обычный» формат для ячеек, которые раньше писались вообще
+        # без формата (например, невалидная дата в колонке дат): без него
+        # xlsxwriter подставил бы формат колонки (DD.MM.YYYY) из set_column.
+        self._plain_format = self.workbook.add_format()
         self._column_formats = [self._resolve_format(column) for column in self.columns]
+        # Признак зависит только от имени колонки — считаем один раз, а не
+        # на каждую ячейку каждой строки.
+        self._text_identifier_flags = [
+            self._is_text_identifier_column(column) for column in self.columns
+        ]
         self._build_group_header_row()
         # В режиме constant_memory строка "уходит" на диск сразу после того,
         # как начата следующая — поэтому set_row для строки 0 нужно вызывать
@@ -174,10 +183,14 @@ class StreamingReportWriter:
         self.sheet.set_row(1, 46)
         for column_index, column_name in enumerate(self.columns):
             column_format = self._column_formats[column_index]
+            # Формат колонки задаём на уровне колонки: пустые ячейки данных
+            # больше не пишутся (см. append_row), и вводимое вручную в пустую
+            # ячейку значение получит тот же формат, что и раньше.
             self.sheet.set_column(
                 column_index,
                 column_index,
                 self._resolve_column_width(column_name, column_format),
+                column_format,
             )
         # Шапка из двух строк уже настроена выше (до перехода на следующую
         # строку — см. комментарий про constant_memory).
@@ -204,20 +217,29 @@ class StreamingReportWriter:
                 raw_value=raw_value,
                 default_format=default_format,
             )
+            # Пустые ячейки не пишем: в отчёте их ~70%, и каждая со стилем
+            # раздувала XML листа (15 МБ xlsx против 5 МБ на 123 файлах) и
+            # съедала ~40% времени записи. Внешне ничего не меняется — формат
+            # колонки задан в __init__ через set_column.
+            if value is None or value == "":
+                continue
             if isinstance(value, datetime) and cell_format is self._date_format:
                 self.sheet.write_datetime(
                     self._next_data_row, zero_based_index, value, cell_format
                 )
-            elif self._is_text_identifier_column(column_name):
+            elif self._text_identifier_flags[zero_based_index]:
-                self.sheet.write_string(
+                identifier = self._normalize_identifier_string(
-                    self._next_data_row,
-                    zero_based_index,
-                    self._normalize_identifier_string(
                     column_name,
                     value,
                     column_index=output_index,
                     row_by_index=data_by_index,
-                    ),
+                )
+                if identifier == "":
+                    continue
+                self.sheet.write_string(
+                    self._next_data_row,
+                    zero_based_index,
+                    identifier,
                     default_format,
                 )
             elif cell_format is not None:
@@ -225,7 +247,9 @@ class StreamingReportWriter:
                     self._next_data_row, zero_based_index, value, cell_format
                 )
             else:
-                self.sheet.write(self._next_data_row, zero_based_index, value)
+                self.sheet.write(
+                    self._next_data_row, zero_based_index, value, self._plain_format
+                )
         self._next_data_row += 1
 
     def _resolve_output_value(

app/pipeline/service.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import tempfile
 import time
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
@@ -28,6 +28,13 @@ from .report import ReportRow, StreamingReportWriter
 
 EMPTY_BATCH_STATUS = "info"
 
+# Вызывается после каждого обработанного XML-файла: (номер файла, всего
+# файлов, имя файла). Нужен, чтобы вызывающий код (main.py) мог показывать
+# прогресс длинной обработки вместо одного блокирующего спиннера — см.
+# комментарий в run_batch про причину (жалоба СФМ на зависание RAISA при
+# батче 230 файлов).
+ProgressCallback = Callable[[int, int, str], None]
+
 
 @dataclass
 class BatchResult:
@@ -40,6 +47,11 @@ class BatchResult:
     log_path: str
     messages: list[str] = field(default_factory=list)
     duration_seconds: float = 0.0
+    # Заполняются, только если файлы обработаны, но загрузка в OpenCloud не
+    # удалась: раньше в этом случае отчёт удалялся вместе с временной
+    # папкой, и 40 минут обработки большого батча пропадали впустую.
+    # main.py отдаёт их пользователю через st.download_button.
+    fallback_files: dict[str, bytes] = field(default_factory=dict)
 
 
 def run_batch(
@@ -47,6 +59,7 @@ def run_batch(
     launcher: str,
     opencloud_username: str | None,
     opencloud_access_token: str | None,
+    progress_callback: ProgressCallback | None = None,
 ) -> BatchResult:
     """Прикреплённые XML и архивы -> отчёт/протокол/лог -> OpenCloud.
 
@@ -62,6 +75,14 @@ def run_batch(
     XML просто попадают в протокол со статусом "error" — никуда не
     перемещаются (перемещать их и некуда, это временная распакованная копия
     вложения).
+
+    progress_callback, если передан, вызывается после разбора каждого
+    XML-файла с (обработано_файлов, всего_файлов, имя_файла). Весь батч
+    по-прежнему выполняется синхронно в одном вызове (отдельный воркер/очередь
+    — отдельная задача), но колбэк даёт вызывающему коду возможность
+    обновлять прогресс по ходу обработки, а не показывать один спиннер на
+    всё время — именно отсутствие видимого прогресса на большом батче (230
+    файлов, тестирование СФМ) навело на мысль, что приложение "зависло".
     """
     started_at = time.perf_counter()
     if not opencloud_username:
@@ -102,6 +123,26 @@ def run_batch(
             messages=["Не прикреплено ни одного файла."],
         )
 
+    # Доступ к OpenCloud проверяем ДО обработки, а не после: на батче из 230
+    # файлов СФМ получили HTTP 401 от OpenCloud только через ~40 минут
+    # обработки, когда дело дошло до загрузки. Неверный/отозванный App Token
+    # или несовпадение username должны всплывать сразу.
+    try:
+        opencloud_space_id = resolve_personal_drive_id(
+            opencloud_username, opencloud_access_token
+        )
+    except OpenCloudUploadError as exc:
+        return BatchResult(
+            status="error",
+            processed=0,
+            partial=0,
+            errors=1,
+            report_path="",
+            protocol_path="",
+            log_path="",
+            messages=[_describe_opencloud_access_error(exc, opencloud_username)],
+        )
+
     protocol_rows: list[ProtocolRow] = []
     attachment_failures: list[str] = []
     attachment_names = ", ".join(path.name for path in input_paths)
@@ -109,12 +150,15 @@ def run_batch(
 
     with tempfile.TemporaryDirectory() as extract_dir:
         extract_dir_path = Path(extract_dir)
+        stage_started = time.perf_counter()
         collected = collect_xml_attachments(input_paths, extract_dir_path)
+        extract_seconds = time.perf_counter() - stage_started
         log_lines.append(
             f"{_now()} | вложений: {len(input_paths)} "
             f"(архивов: {collected.archives}, "
             f"отдельных XML: {collected.loose_xml_files}), "
-            f"найдено XML-файлов: {len(collected.files)}"
+            f"найдено XML-файлов: {len(collected.files)}, "
+            f"распаковка: {extract_seconds:.1f} с"
         )
         for problem in collected.problems:
             protocol_rows.append(
@@ -152,7 +196,12 @@ def run_batch(
             )
 
         report_writer = StreamingReportWriter()
-        for collected_xml in collected.files:
+        total_files = len(collected.files)
+        total_rows_written = 0
+        parse_total_seconds = 0.0
+        write_total_seconds = 0.0
+        files_loop_started = time.perf_counter()
+        for file_number, collected_xml in enumerate(collected.files, start=1):
             file_name = collected_xml.file_name
             try:
                 content = collected_xml.path.read_bytes()
@@ -170,12 +219,17 @@ def run_batch(
                     )
                 )
                 log_lines.append(f"{_now()} | read failed: {file_name}. {exc}")
+                if progress_callback is not None:
+                    progress_callback(file_number, total_files, file_name)
                 continue
 
             # parse_xml_content сам проверяет имя файла и проставляет
             # fatal_error, если оно не соответствует шаблону — отдельная
             # валидация здесь не нужна.
+            parse_started = time.perf_counter()
             parse_result = parse_xml_content(file_name=file_name, xml_content=content)
+            parse_seconds = time.perf_counter() - parse_started
+            write_started = time.perf_counter()
             for row in parse_result.rows:
                 report_writer.append_row(
                     ReportRow(
@@ -187,6 +241,10 @@ def run_batch(
                         is_continuation=row.is_continuation,
                     )
                 )
+            write_seconds = time.perf_counter() - write_started
+            parse_total_seconds += parse_seconds
+            write_total_seconds += write_seconds
+            total_rows_written += len(parse_result.rows)
             status, message = _resolve_parse_status(parse_result)
             protocol_rows.append(
                 ProtocolRow(
@@ -200,7 +258,23 @@ def run_batch(
                     processed_at=datetime.now(),
                 )
             )
-            log_lines.append(f"{_now()} | parsed {file_name}: {status}")
+            # Замер на каждый файл — чтобы по логу было видно, растёт ли
+            # время на файл по ходу батча (накопление) или его съедает
+            # конкретный тяжёлый файл (много операций/участников).
+            # 100 файлов = 6 мин, 230 файлов = 45 мин — нелинейно, причина
+            # пока не локализована.
+            log_lines.append(
+                f"{_now()} | parsed {file_name}: {status} | "
+                f"{file_number}/{total_files} | "
+                f"размер={len(content) // 1024} КБ, "
+                f"операций={parse_result.total_operations}, "
+                f"строк={len(parse_result.rows)}, "
+                f"разбор={parse_seconds:.2f} с, "
+                f"запись={write_seconds:.2f} с, "
+                f"всего строк в отчёте={total_rows_written}"
+            )
+            if progress_callback is not None:
+                progress_callback(file_number, total_files, file_name)
 
         if not protocol_rows:
             protocol_rows.append(
@@ -216,6 +290,14 @@ def run_batch(
             )
             log_lines.append(f"{_now()} | empty batch")
 
+        files_loop_seconds = time.perf_counter() - files_loop_started
+        log_lines.append(
+            f"{_now()} | обработка файлов: {files_loop_seconds:.1f} с "
+            f"(разбор XML: {parse_total_seconds:.1f} с, "
+            f"запись строк отчёта: {write_total_seconds:.1f} с), "
+            f"строк в отчёте: {total_rows_written}"
+        )
+
         year, month, day, date_key = build_date_parts()
 
         with tempfile.TemporaryDirectory() as artifacts_dir:
@@ -224,16 +306,24 @@ def run_batch(
             local_protocol = artifacts_dir_path / "protocol.xlsx"
             local_log = artifacts_dir_path / "process.log"
 
+            stage_started = time.perf_counter()
             report_writer.save(local_report)
+            report_save_seconds = time.perf_counter() - stage_started
+            stage_started = time.perf_counter()
             write_protocol(protocol_rows, local_protocol)
+            protocol_seconds = time.perf_counter() - stage_started
+            log_lines.append(
+                f"{_now()} | сохранение отчёта: {report_save_seconds:.1f} с "
+                f"({local_report.stat().st_size // 1024} КБ), "
+                f"протокол: {protocol_seconds:.1f} с"
+            )
             local_log.write_text("\r\n".join(log_lines) + "\r\n", encoding="utf-8")
 
             try:
                 # Номер запуска для папки дня в OpenCloud определяем по
                 # самому OpenCloud — см. next_opencloud_run_number.
-                opencloud_space_id = resolve_personal_drive_id(
+                # opencloud_space_id получен в начале run_batch (проверка
-                    opencloud_username, opencloud_access_token
+                # доступа до обработки).
-                )
                 run_number = next_opencloud_run_number(
                     opencloud_space_id,
                     (DEFAULT_OUTPUT_DIR_NAME, year, month, day),
@@ -259,6 +349,14 @@ def run_batch(
                     run_number=run_number,
                 )
             except OpenCloudUploadError as exc:
+                # Файлы ещё лежат во временной папке — забираем их в память,
+                # чтобы пользователь мог скачать результат из интерфейса.
+                fallback_name_suffix = date_key
+                fallback_files = {
+                    f"report_{fallback_name_suffix}.xlsx": local_report.read_bytes(),
+                    f"protocol_{fallback_name_suffix}.xlsx": local_protocol.read_bytes(),
+                    f"process_{fallback_name_suffix}.log": local_log.read_bytes(),
+                }
                 return BatchResult(
                     status="error",
                     processed=0,
@@ -268,10 +366,13 @@ def run_batch(
                     protocol_path="",
                     log_path="",
                     messages=[
-                        *log_lines,
                         f"Файлы обработаны, но не удалось загрузить результат "
-                        f"в OpenCloud: {exc}",
+                        f"в OpenCloud: {_describe_opencloud_access_error(exc, opencloud_username)} "
+                        f"Отчёт можно скачать кнопками ниже.",
+                        *log_lines[-5:],
                     ],
+                    duration_seconds=round(time.perf_counter() - started_at, 2),
+                    fallback_files=fallback_files,
                 )
 
     processed = sum(1 for row in protocol_rows if row.status == "success")
@@ -376,6 +477,18 @@ def _upload_artifacts(
     return report_url, protocol_url, log_url
 
 
+def _describe_opencloud_access_error(exc: Exception, username: str) -> str:
+    text = str(exc)
+    if "HTTP 401" in text:
+        return (
+            f"OpenCloud отклонил App Token для пользователя «{username}» (HTTP 401). "
+            "Вероятно, токен истёк, отозван, введён с ошибкой или выпущен под "
+            "другой учётной записью. Создайте новый App Token в OpenCloud и "
+            "сохраните его через «Добавить токен»."
+        )
+    return f"Нет доступа к OpenCloud: {text}"
+
+
 def _now() -> str:
     return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 

main.py

@@ -48,14 +48,32 @@ def main() -> None:
         ):
             with tempfile.TemporaryDirectory() as tmp_dir:
                 input_paths = _save_attachments(uploaded_files, Path(tmp_dir))
-                with st.spinner("Выполняется обработка..."):
+                progress_bar = st.progress(0.0, text="Подготовка файлов...")
+                status_placeholder = st.empty()
+
+                def _on_progress(done: int, total: int, file_name: str) -> None:
+                    # Прогресс по файлам, а не один спиннер на весь батч —
+                    # иначе при большом количестве файлов (были жалобы на
+                    # батч в 230 штук) непонятно, идёт обработка или
+                    # приложение зависло, и пользователь закрывает вкладку
+                    # посреди обработки.
+                    fraction = done / total if total else 1.0
+                    progress_bar.progress(
+                        fraction, text=f"Обработано {done} из {total}: {file_name}"
+                    )
+                    status_placeholder.caption(
+                        f"Последний обработанный файл: {file_name}"
+                    )
+
                 launcher_name = resolve_opencloud_username() or resolve_user_id()
                 batch_result = run_batch(
                     input_paths=input_paths,
                     launcher=launcher_name,
                     opencloud_username=resolve_opencloud_username(),
                     opencloud_access_token=resolve_opencloud_access_token(),
+                    progress_callback=_on_progress,
                 )
+                progress_bar.progress(1.0, text="Готово")
             if batch_result.status == "success":
                 st.success("Обработка завершена успешно.")
             elif batch_result.status == "partial":
@@ -76,6 +94,16 @@ def main() -> None:
                 st.caption("Последние сообщения обработки:")
                 for message in batch_result.messages:
                     st.write(f"- {message}")
+            # Если обработка прошла, а загрузка в OpenCloud — нет, отдаём
+            # результат напрямую, чтобы не терять долгую обработку.
+            for file_name, payload in batch_result.fallback_files.items():
+                st.download_button(
+                    f"Скачать {file_name}",
+                    data=payload,
+                    file_name=file_name,
+                    use_container_width=True,
+                    key=f"fallback_{file_name}",
+                )
 
     if st.button("Выйти", type="secondary"):
         clear_session()