fix(consolidation): линейная сборка xlsx, защита от формул и управляющих символов в экспорте

Строки пишутся по явному счётчику (sheet.max_row делал сборку квадратичной: 6 000 строк — 8.4 с → 0.5 с),
сборка вынесена в run_in_threadpool. Тексты проходят _put_text: запрещённые XML-символы вырезаются, тип
ячейки строковый — ведущие =, +, -, @ не становятся формулами (R-12, design #77). Параметр entity_names
удалён (R-13, #78), порог точности double в #73 исправлен, тесты пустых сумм и 6 000 строк.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev 2026-10-06 07:12:45 +04:00
parent 8e82c2c19f
commit 5b3b1011cb
6 changed files with 196 additions and 43 deletions

View File

@ -15,6 +15,7 @@ import uuid
from collections.abc import Collection, Sequence
from typing import Literal
from fastapi.concurrency import run_in_threadpool
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
@ -176,7 +177,7 @@ class ConsolidatedOsvService:
percent-кодирования: оно в маршруте). `audit_event(EXPORT)` пишется после сборки книги, в транзакции запроса."""
consolidated, version, consolidation = await self._build(version_id)
rows = domain.apply_filter(consolidated.rows, domain_filter(f))
content = build_workbook(consolidated, rows, entity_names={})
content = await run_in_threadpool(build_workbook, consolidated, rows)
group_name = (
await self._session.execute(
select(CompanyGroup.name).where(CompanyGroup.id == consolidation.company_group_id)

View File

@ -2,16 +2,24 @@
Лист «Сводная ОСВ» — строки свода в переданном порядке (порядок API, D-16), шесть сумм — числовые ячейки
`#,##0.00` в рублях без деления. В ячейки пишется `Decimal` как есть: `float` в нашем коде не используется;
`openpyxl` сам сериализует число в XML (формат xlsx хранит числа как IEEE-754 double, поэтому значения свыше
~10^15 теряют младшие разряды — для сумм ОСВ в рублях запас на три порядка). Лист «Пробелы» — пробелы свода
`openpyxl` сам сериализует число в XML через `%.16g` (формат xlsx хранит числа как IEEE-754 double): значения до
10^13 с копейками читаются обратно точно, выше ~10^13 младшая копейка может потеряться (`96729051052621.71` →
`…621.7`), а `9999999999999999.99` станет `1e+16`. Для сумм ОСВ в рублях вывод прежний: сам Excel держит 15 цифр.
Лист «Пробелы» — пробелы свода
целиком (`Consolidated.gaps`), фильтры строк на него не влияют; без пробелов — одна ячейка «пробелов нет».
Тексты приходят из загруженных внешних файлов (R-12): в книгу они попадают только через `_text` / `_put_text` —
запрещённые XML-символы вырезаются (иначе `IllegalCharacterError` → 500), тип ячейки принудительно строковый (`"s"`),
поэтому ведущие `=`, `+`, `-`, `@` не превращаются в формулы Excel, а текст читается обратно без изменений.
Строки пишутся по явному счётчику (`sheet.max_row` в `openpyxl` — O(n) на вызов, цикл по нему квадратичен).
"""
import uuid
from collections.abc import Mapping, Sequence
from collections.abc import Sequence
from decimal import Decimal
from io import BytesIO
from openpyxl import Workbook
from openpyxl.cell.cell import ILLEGAL_CHARACTERS_RE
from openpyxl.utils import get_column_letter
from openpyxl.worksheet.worksheet import Worksheet
@ -37,59 +45,68 @@ def _widths(sheet: Worksheet, widths: Sequence[int]) -> None:
sheet.column_dimensions[get_column_letter(index)].width = width
def _main_sheet(sheet: Worksheet, rows: Sequence[ConsolidatedRow], entity_names: Mapping[uuid.UUID, str]) -> None:
def _text(value: str | None) -> str | None:
"""Текст для ячейки: запрещённые XML-символы вырезаны (`IllegalCharacterError` → 500), `None` остаётся пустым."""
return None if value is None else ILLEGAL_CHARACTERS_RE.sub("", value)
def _put_text(sheet: Worksheet, row: int, column: int, value: str | None) -> None:
"""Строковая ячейка: тип `"s"` принудительно — значение с ведущим `=` не становится формулой (R-12)."""
text = _text(value)
if text is None:
return
cell = sheet.cell(row=row, column=column, value=text)
cell.data_type = "s"
def _put_money(sheet: Worksheet, row: int, column: int, value: Decimal | None) -> None:
"""Числовая ячейка `#,##0.00`; `None` (сумма не задана) — пустая ячейка, не ноль."""
if value is None:
return
sheet.cell(row=row, column=column, value=value).number_format = MONEY_FORMAT
def _main_sheet(sheet: Worksheet, rows: Sequence[ConsolidatedRow]) -> None:
sheet.title = MAIN_SHEET
sheet.append(MAIN_HEADER)
for row in rows:
sheet.append(
[
entity_names.get(row.legal_entity_id, row.legal_entity_name),
row.legal_entity_inn,
row.account_code,
row.period,
row.counterparty_name,
row.counterparty_inn,
row.opening_debit,
row.opening_credit,
row.turnover_debit,
row.turnover_credit,
row.closing_debit,
row.closing_credit,
row.counterparty_type.value,
row.counterparty_entity_name,
None if row.match_kind is None else row.match_kind.value,
]
)
for cell in sheet[sheet.max_row][FIRST_MONEY_COLUMN - 1 : LAST_MONEY_COLUMN]:
cell.number_format = MONEY_FORMAT
for line, row in enumerate(rows, start=2): # явный счётчик строк: `sheet.max_row` — O(n) на вызов
texts = (row.legal_entity_name, row.legal_entity_inn, row.account_code, row.period, row.counterparty_name,
row.counterparty_inn) # fmt: skip
for column, text in enumerate(texts, start=1):
_put_text(sheet, line, column, text)
amounts = (row.opening_debit, row.opening_credit, row.turnover_debit, row.turnover_credit, row.closing_debit,
row.closing_credit) # fmt: skip
for column, amount in enumerate(amounts, start=FIRST_MONEY_COLUMN):
_put_money(sheet, line, column, amount)
_put_text(sheet, line, LAST_MONEY_COLUMN + 1, row.counterparty_type.value)
_put_text(sheet, line, LAST_MONEY_COLUMN + 2, row.counterparty_entity_name)
_put_text(sheet, line, LAST_MONEY_COLUMN + 3, None if row.match_kind is None else row.match_kind.value)
sheet.freeze_panes = "A2"
_widths(sheet, (32, 14, 8, 10, 36, 14, 16, 16, 16, 16, 16, 16, 12, 32, 14))
def _gaps_sheet(sheet: Worksheet, consolidated: Consolidated, entity_names: Mapping[uuid.UUID, str]) -> None:
def _gaps_sheet(sheet: Worksheet, consolidated: Consolidated) -> None:
sheet.title = GAPS_SHEET
if not consolidated.gaps:
sheet.append([NO_GAPS])
return
sheet.append(GAPS_HEADER)
for gap in consolidated.gaps:
quarters = ", ".join(quarter.value for quarter in gap.missing_quarters)
sheet.append([entity_names.get(gap.legal_entity_id, gap.legal_entity_name), gap.account_code, quarters])
for line, gap in enumerate(consolidated.gaps, start=2):
_put_text(sheet, line, 1, gap.legal_entity_name)
_put_text(sheet, line, 2, gap.account_code)
_put_text(sheet, line, 3, ", ".join(quarter.value for quarter in gap.missing_quarters))
_widths(sheet, (32, 8, 24))
def build_workbook(
consolidated: Consolidated, rows: Sequence[ConsolidatedRow], *, entity_names: Mapping[uuid.UUID, str]
) -> bytes:
def build_workbook(consolidated: Consolidated, rows: Sequence[ConsolidatedRow]) -> bytes:
"""Книга из двух листов: `rows` (уже отфильтрованные строки свода) и пробелы `consolidated.gaps`.
`entity_names` — отображаемые имена ЮЛ по идентификатору (переопределяют имена из строк и пробелов; нет в
словаре — имя из самой строки/пробела). Только память: `BytesIO`, S3 и БД не используются."""
Чистая функция: только память (`BytesIO`), ни БД, ни S3; время линейно по числу строк."""
workbook = Workbook()
main = workbook.active
assert main is not None # новая книга всегда с активным листом
_main_sheet(main, rows, entity_names)
_gaps_sheet(workbook.create_sheet(), consolidated, entity_names)
_main_sheet(main, rows)
_gaps_sheet(workbook.create_sheet(), consolidated)
buffer = BytesIO()
workbook.save(buffer)
return buffer.getvalue()

View File

@ -321,3 +321,25 @@ async def test_export_writes_only_the_audit_event(auth_client: AuthClientFactory
assert after == before
assert (await seeded.execute(others)).scalar_one() == events_before + 1
assert len(await _export_events(seeded)) == 1
@pytest.mark.integration
async def test_formula_like_and_control_characters_in_external_names(
auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
"""R-12: имена приходят из загруженных файлов — формула не вычисляется, `\x0b` вырезается, ответ 200."""
world = await _world(seeded)
names = {"Неизвестный контрагент": '=HYPERLINK("http://x","y")', "ООО Внешний": "a\x0bb"}
for entry in (await seeded.execute(select(OsvEntry).where(OsvEntry.counterparty_name.in_(list(names))))).scalars():
entry.counterparty_name = names[entry.counterparty_name]
await seeded.flush()
client = await auth_client(KA_RF)
response = await client.get(_url(world))
assert response.status_code == HTTP_OK, response.text
cells = {
(cell.data_type, cell.value)
for (cell,) in _book(response.content)[MAIN].iter_rows(min_row=2, min_col=5, max_col=5)
}
assert ("s", '=HYPERLINK("http://x","y")') in cells and ("s", "ab") in cells

View File

@ -0,0 +1,111 @@
"""`build_workbook` без БД (spec consolidated-osv «Экспорт xlsx»; design D-14, #77): линейное время, пустые суммы,
защита от формул Excel и запрещённых символов."""
import io
import time
import uuid
from decimal import Decimal
import pytest
from openpyxl import load_workbook
from src.domain.consolidated_osv import Consolidated, ConsolidatedRow, CounterpartyType, GapItem
from src.models.enums import CounterpartyMatchKind, PeriodQuarter
from src.services.osv_export import GAPS_SHEET, MAIN_SHEET, build_workbook
D = Decimal
BIG_ROWS = 6000
BIG_MAX_SECONDS = 3.0 # квадратичная версия на 6 000 строках — 8.4 с, линейная — 0.5 с (запас на загруженный CI)
ENTITY = uuid.uuid4()
INJECTIONS = ['=HYPERLINK("http://x","y")', "=A1+1", "@SUM(1)", "+1", "-1+2"]
def _row(name: str = "ООО Б", **overrides: object) -> ConsolidatedRow:
fields: dict[str, object] = {
"entry_ids": (uuid.uuid4(),),
"legal_entity_id": ENTITY,
"legal_entity_name": "ООО А",
"legal_entity_inn": "1234567890",
"account_code": "60",
"account_root": "60",
"period": "2024 Q3",
"is_cumulative": False,
"row_number": 1,
"counterparty_name": name,
"counterparty_inn": "0987654321",
"counterparty_id": None,
"counterparty_entity_name": None,
"counterparty_type": CounterpartyType.EXTERNAL,
"match_kind": None,
"match_stored": False,
"opening_debit": D("100.05"),
"opening_credit": D("0.00"),
"turnover_debit": D("10.10"),
"turnover_credit": D("0.00"),
"closing_debit": D("110.15"),
"closing_credit": D("0.00"),
}
fields.update(overrides)
return ConsolidatedRow(**fields) # type: ignore[arg-type]
def _consolidated(rows: tuple[ConsolidatedRow, ...], gaps: tuple[GapItem, ...] = ()) -> Consolidated:
return Consolidated(rows=rows, gaps=gaps, documents_total=1)
def _sheet(content: bytes, name: str = MAIN_SHEET): # noqa: ANN202 - тип листа openpyxl не типизирован
return load_workbook(io.BytesIO(content))[name]
def test_build_is_linear_for_thousands_of_rows() -> None:
rows = tuple(_row(f"Контрагент {n}") for n in range(BIG_ROWS))
started = time.perf_counter()
content = build_workbook(_consolidated(rows), rows)
elapsed = time.perf_counter() - started
print(f"\nPERF build_workbook {BIG_ROWS} rows: {elapsed:.2f}s") # noqa: T201
assert elapsed < BIG_MAX_SECONDS
sheet = _sheet(content)
assert sheet.max_row == BIG_ROWS + 1
assert sheet.cell(BIG_ROWS + 1, 5).value == f"Контрагент {BIG_ROWS - 1}"
def test_none_amounts_are_empty_cells_not_zero() -> None:
row = _row(opening_debit=None, opening_credit=D("1.00"), turnover_debit=None, turnover_credit=None,
closing_debit=D("0.00"), closing_credit=None) # fmt: skip
sheet = _sheet(build_workbook(_consolidated((row,)), (row,)))
assert [sheet.cell(2, column).value for column in range(7, 13)] == [None, 1, None, None, 0, None]
assert sheet.cell(2, 8).number_format == "#,##0.00" and sheet.cell(2, 8).data_type == "n"
assert sheet.cell(2, 13).value == "EXTERNAL" and sheet.cell(2, 14).value is None and sheet.cell(2, 15).value is None
@pytest.mark.parametrize("text", INJECTIONS)
def test_formula_like_text_stays_plain_string(text: str) -> None:
gap = GapItem(ENTITY, text, text, (PeriodQuarter.Q1, PeriodQuarter.Q2))
row = _row(
text, legal_entity_name=text, legal_entity_inn=text, account_code=text, counterparty_inn=text,
counterparty_entity_name=text, counterparty_type=CounterpartyType.GSK, match_kind=CounterpartyMatchKind.NAME,
) # fmt: skip
content = build_workbook(_consolidated((row,), (gap,)), (row,))
main = _sheet(content)
for column in (1, 2, 3, 5, 6, 14):
cell = main.cell(2, column)
assert (cell.data_type, cell.value) == ("s", text), column
gaps = _sheet(content, GAPS_SHEET)
assert [(cell.data_type, cell.value) for cell in gaps[2][:2]] == [("s", text), ("s", text)]
assert gaps.cell(2, 3).value == "Q1, Q2"
def test_control_characters_are_stripped_not_500() -> None:
row = _row("a\x0bb\x00c", legal_entity_name="ООО\x1f А", counterparty_entity_name="\x08Эф")
gap = GapItem(ENTITY, "Б\x0b", "6\x0b0", (PeriodQuarter.Q2,))
content = build_workbook(_consolidated((row,), (gap,)), (row,))
main = _sheet(content)
assert (main.cell(2, 5).value, main.cell(2, 1).value, main.cell(2, 14).value) == ("abc", "ООО А", "Эф")
assert [cell.value for cell in _sheet(content, GAPS_SHEET)[2]] == ["Б", "60", "Q2"]

View File

@ -232,7 +232,7 @@
- Modify: `api/requirements-dev.txt` (ничего, openpyxl есть), `api/pyproject.toml` (mypy override `openpyxl.*` `ignore_missing_imports`, если стабы не установлены)
**Interfaces:**
- `def build_workbook(c: Consolidated, rows: Sequence[ConsolidatedRow], *, entity_names: Mapping[uuid.UUID, str]) -> bytes` — `openpyxl.Workbook()`, лист «Сводная ОСВ» (15 колонок D-14, `number_format = "#,##0.00"`, суммы `float(Decimal)` недопустимы → писать `Decimal` напрямую: openpyxl принимает `Decimal` как число), лист «Пробелы»; `BytesIO`.
- `def build_workbook(c: Consolidated, rows: Sequence[ConsolidatedRow]) -> bytes` — `openpyxl.Workbook()`, лист «Сводная ОСВ» (15 колонок D-14, `number_format = "#,##0.00"`, суммы `float(Decimal)` недопустимы → писать `Decimal` напрямую: openpyxl принимает `Decimal` как число), лист «Пробелы»; `BytesIO`.
- `ConsolidatedOsvService.export(version_id, f: OsvFilters, audit: RequestAudit) -> tuple[bytes, str filename]`; аудит `EXPORT` (`section = EXPORT`, `after = {report: "consolidated_osv", filters: f.model_dump(exclude_none=True), rows: len(rows)}`).
- Маршрут `GET …/osv/export` → `Response(content, media_type=…, headers={"Content-Disposition": f'attachment; filename="svod-osv.xlsx"; filename*=UTF-8\'\'{quote(name)}'})`.

View File

@ -257,7 +257,9 @@ legal_entity_name, account_code, missing_quarters[]}}`. Фильтры `GET …/
| 70 | 2026-10-06 | Кандидаты: `entry_id` не строки свода версии (чужой, непривязанный, несуществующий) или имя с пустой нормализацией — рекомендаций нет, не 404; `recommended` ищется по `CounterpartyIndex.by_normalized` (имена и алиасы всего справочника; загружается только при `entry_id`); `q` обрезается по краям (`scope=all`: пустой или короче 3 — 422 `QUERY_TOO_SHORT`; `max_length` 500); ИНН — префикс без пробелов внутри `q`; алиас — подстрока `legal_entity_alias.alias`; `scope=group` = ГК консолидации ∪ периметр версии; признак — `domain.classify` | D-11 говорит «если передан»; нормализация — функция Python, в SQL её нет, а индекс — тот же, что у резолва свода | лишняя загрузка справочника на запрос с `entry_id` (как у `GET …/osv`) |
| 71 | 2026-10-06 | Экспорт: лист «Пробелы» строится по своду целиком (`Consolidated.gaps`) и от фильтров строк не зависит; при `unmapped=true` без совпадений лист «Сводная ОСВ» — только заголовок, «Пробелы» заполнен, `audit.rows = 0` (Review Focus 5) | D-14 и спека описывают лист пробелов отдельно от фильтров; пробел — свойство версии, а не строк, и фильтр не должен прятать неполноту данных | файл с фильтром показывает пробелы, не относящиеся к выбранным строкам; читателю это видно по листу |
| 72 | 2026-10-06 | Колонки листа: «Контрагент»/«ИНН контрагента» — как в файле ОСВ (`counterparty_name`/`counterparty_inn`), «Сопоставлено с» — имя ЮЛ справочника (`counterparty_entity_name`, пусто без сопоставления), «Тип контрагента» — код `GSK`/`GK`/`EXTERNAL`, «Способ сопоставления» — код `match_kind` (пусто без него); пустые значения — пустые ячейки | D-14 перечисляет колонки без значений; коды совпадают с API, и выгрузка сверяется со страницей свода построчно | если банк захочет русские подписи типа — правка одного словаря в `osv_export` |
| 73 | 2026-10-06 | Суммы пишутся `Decimal` напрямую; `openpyxl` сериализует число в XML через `float`-представление (формат xlsx — IEEE-754 double), значения свыше ~10^15 теряют младшие разряды; защит в коде нет | суммы ОСВ в рублях на три порядка меньше предела (`numeric(18,2)` — запас, не потребность); тест сверяет `Decimal(str(cell.value))` с API | сумма от 10^15 руб. в выгрузке будет округлена, в БД и API — точная |
| 73 | 2026-10-06 | Суммы пишутся `Decimal` напрямую; `openpyxl` сериализует число в XML через `%.16g` (формат xlsx — IEEE-754 double): значения до 10^13 с копейками читаются обратно точно, выше ~10^13 младшая копейка может потеряться (`96729051052621.71` → `…621.7`; ≈1,3 % случайных значений в 10^13–10^14), `9999999999999999.99` станет `1e+16`; защит в коде нет (порог уточнён ревью T12.3) | суммы ОСВ в рублях на практике ниже 10^13, а сам Excel держит 15 значащих цифр; тест сверяет `Decimal(str(cell.value))` с API | сумма от 10^13 руб. в выгрузке может потерять копейку, в БД и API — точная |
| 74 | 2026-10-06 | Имя файла: `svod-osv_<ГК>_<год>-<Qn>_v<n>.xlsx`, имя ГК берётся как есть и целиком кодируется `quote(safe="")` (пробел, кавычки, `;`, `/`, `'`, переводы строк, кириллица) — заголовок `Content-Disposition` чистый ASCII, `filename="svod-osv.xlsx"` фиксирован; квартал — `Q1..Q4` | D-14: `<год>-<квартал>`; фиксированное ASCII-имя защищает от инъекции в заголовок | клиент без поддержки `filename*` сохранит файл как `svod-osv.xlsx` |
| 75 | 2026-10-06 | `build_workbook(c, rows, *, entity_names)`: параметр `entity_names` сохранён по плану как переопределение отображаемых имён ЮЛ по id, без записи в словаре — имя из строки/пробела; сервис передаёт `{}`, так как строки и пробелы уже несут имена | интерфейс зафиксирован планом; убирать параметр — менять контракт задачи | параметр сейчас не используется сервисом; кандидат на удаление при ревью |
| 76 | 2026-10-06 | Аудит `EXPORT` пишется после сборки книги в той же транзакции запроса: `filters = f.model_dump(mode="json", exclude_none=True)` (enum → значение, UUID → строка; пустые фильтры — `{}`), `rows` — число строк листа после фильтров; ошибка сборки откатывает и аудит; отказы (403/404) не аудируются | spec audit-trail: EXPORT — действие над версией; запись без выгруженного файла была бы ложной | при сбое после коммита (обрыв соединения при отдаче) событие остаётся, а файл клиент мог не получить |
| 75 | 2026-10-06 | `build_workbook(c, rows, *, entity_names)`: параметр `entity_names` сохранён по плану как переопределение отображаемых имён ЮЛ по id, без записи в словаре — имя из строки/пробела; сервис передаёт `{}`, так как строки и пробелы уже несут имена | интерфейс зафиксирован планом; убирать параметр — менять контракт задачи | параметр сейчас не используется сервисом; кандидат на удаление при ревью (снято решением #78) |
| 76 | 2026-10-06 | Аудит `EXPORT` пишется после сборки книги в той же транзакции запроса: `filters = f.model_dump(mode="json", exclude_none=True)` (enum → значение, UUID → строка; пустые фильтры — `{}`), `rows` — число строк листа после фильтров; ошибка сборки откатывает и аудит; отказы (403/404) не аудируются | spec audit-trail: EXPORT — действие над версией; запись без выгруженного файла была бы ложной | при сбое после коммита (обрыв соединения при отдаче) событие остаётся, а файл клиент мог не получить; `model_dump(mode="json", exclude_none=True)` отличается от буквального `model_dump` плана (M-5): UUID и enum должны быть JSON-сериализуемы для `audit_event` |
| 77 | 2026-10-06 | R-12: все текстовые ячейки экспорта (компания, ИНН, счёт, период, контрагент, ИНН контрагента, «Сопоставлено с», тип, способ, лист «Пробелы») проходят `_put_text`: запрещённые XML-символы (`ILLEGAL_CHARACTERS_RE`) вырезаются, тип ячейки принудительно `"s"`, поэтому `=`, `+`, `-`, `@` в начале не становятся формулой, а текст читается обратно без изменений | файлы внешние, аналитики открывают выгрузку в Excel: без этого `=HYPERLINK(...)` из чужого файла выполнился бы в их книге, а управляющий символ давал `IllegalCharacterError` → 500 | текст ячейки с ведущим `=` больше не вычисляется — это желаемо; вырезанный управляющий символ из выгрузки пропадает |
| 78 | 2026-10-06 | R-13: параметр `entity_names` убран из `build_workbook` — сигнатура `build_workbook(c, rows) -> bytes`; сборка книги выполняется в `run_in_threadpool` (линейная по строкам, явный счётчик строк вместо `sheet.max_row`) | строки и пробелы уже несут имена ЮЛ, параметр не использовался (YAGNI); план предписывал параметр — Interfaces Task 12 исправлены | если фронту понадобится словарь переопределения имён — вернуть параметр |