From 0c203c1d91636a28f90a7ea4dc51e7d6c179d905 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 7 Oct 2026 13:23:36 +0400 Subject: [PATCH] =?UTF-8?q?feat(consolidations):=20=D1=87=D0=B5=D0=BA-?= =?UTF-8?q?=D0=BB=D0=B8=D1=81=D1=82=20=D0=B2=20CSV=20=E2=80=94=20=D0=BF?= =?UTF-8?q?=D0=BE=D0=B4=D0=BF=D0=B8=D1=81=D0=B8=20=D0=BC=D0=B0=D0=BA=D0=B5?= =?UTF-8?q?=D1=82=D0=B0,=20BOM,=20=C2=AB;=C2=BB=20(D-13)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Добавлен экспорт чек-листа в CSV-формате при параметре ?format=csv: - Реализована функция render_checklist_csv в src/domain/checklist_csv.py - UTF-8 кодировка с BOM, разделитель «;», окончание строк CRLF - Подписи статусов: CONFIRMED→«Загружен», PROCESSING→«Обработка», NEEDS_CHECK/RECOGNIZED→«Требуется проверка», ERROR→«Ошибка загрузки», MISSING→«Отсутствует» - При missing_quarters добавляется подстрока « (нет: Q1, Q2)» - Имя файла: checklist-<ГК>-<год>-<квартал>.csv с поддержкой RFC 5987 - Используется существующая функция content_disposition из filenames.py - Добавлена функция version_title в ConsolidationService для получения названия ГК, года и квартала версии Тесты: - test_domain_checklist_csv.py: юнит-тесты функций render и filename - test_api_consolidation_checklist.py: интеграционный тест CSV-ответа Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- .../api/endpoints/consolidations/checklist.py | 35 +++++++-- api/src/domain/checklist_csv.py | 76 +++++++++++++++++++ api/src/services/consolidations.py | 14 ++++ api/tests/test_api_consolidation_checklist.py | 19 +++++ api/tests/test_domain_checklist_csv.py | 37 +++++++++ 5 files changed, 176 insertions(+), 5 deletions(-) create mode 100644 api/src/domain/checklist_csv.py create mode 100644 api/tests/test_domain_checklist_csv.py diff --git a/api/src/api/endpoints/consolidations/checklist.py b/api/src/api/endpoints/consolidations/checklist.py index 92ab0bb..474f178 100644 --- a/api/src/api/endpoints/consolidations/checklist.py +++ b/api/src/api/endpoints/consolidations/checklist.py @@ -2,12 +2,14 @@ «Чек-лист комплекта»; design D-13). Право проверяется до видимости; версия ГК вне видимости — тот же 404.""" import uuid -from typing import Any +from typing import Any, Literal -from fastapi import APIRouter, Depends, status +from fastapi import APIRouter, Depends, Query, Response, status from src.api.dependencies.auth import require from src.api.dependencies.consolidations import ConsolidationServiceDep +from src.common.filenames import content_disposition +from src.domain.checklist_csv import checklist_filename, render_checklist_csv from src.schemas.common import ErrorDetail from src.schemas.consolidations import ChecklistOut @@ -26,6 +28,29 @@ NOT_FOUND: dict[int | str, dict[str, Any]] = { responses=NOT_FOUND, summary="Чек-лист комплекта документов версии", ) -async def get_checklist(version_id: uuid.UUID, service: ConsolidationServiceDep) -> ChecklistOut: - """Правила комплекта подхода версии и строки по ЮЛ периметра: ячейки `{status, missing_quarters, document_ids}`.""" - return await service.checklist(version_id) +async def get_checklist( + version_id: uuid.UUID, + service: ConsolidationServiceDep, + format: Literal["json", "csv"] = Query("json"), +) -> ChecklistOut | Response: + """Правила комплекта подхода версии и строки по ЮЛ периметра: ячейки `{status, missing_quarters, document_ids}`. + + При `?format=csv` возвращает чек-лист файлом CSV (UTF-8 с BOM, разделитель `;`, CRLF). + """ + checklist = await service.checklist(version_id) + + if format == "csv": + group_name, year, quarter = await service.version_title(version_id) + filename = checklist_filename(group_name, year, quarter) + csv_content = render_checklist_csv(checklist) + return Response( + content=csv_content, + media_type="text/csv; charset=utf-8", + headers={ + "Content-Disposition": content_disposition( + filename, disposition="attachment", ascii_fallback="checklist.csv" + ) + }, + ) + + return checklist diff --git a/api/src/domain/checklist_csv.py b/api/src/domain/checklist_csv.py new file mode 100644 index 0000000..487a910 --- /dev/null +++ b/api/src/domain/checklist_csv.py @@ -0,0 +1,76 @@ +"""Экспорт чек-листа в CSV (spec consolidations-api «Чек-лист комплекта»; D-13). + +`render_checklist_csv` — чистая функция без БД: строит CSV буфер, кодирует в UTF-8 с BOM. +`checklist_filename` — формирует имя файла для Content-Disposition. +""" + +import csv +import io + +from src.domain.checklist import CellStatus +from src.models.enums import DocumentType +from src.schemas.consolidations import ChecklistOut + +# Подписи статусов ячеек (макет UI) +_CELL_STATUS_LABELS = { + CellStatus.CONFIRMED: "Загружен", + CellStatus.PROCESSING: "Обработка", + CellStatus.NEEDS_CHECK: "Требуется проверка", + CellStatus.RECOGNIZED: "Требуется проверка", + CellStatus.ERROR: "Ошибка загрузки", + CellStatus.MISSING: "Отсутствует", +} + + +# Подписи правил +def _rule_label(document_type: DocumentType, account: str | None) -> str: + """Подпись правила для заголовка CSV.""" + if document_type == DocumentType.BALANCE: + return "Бух. баланс" + elif document_type == DocumentType.INCOME: + return "ОФР" + elif document_type == DocumentType.OSV: + return f"ОСВ {account}" if account else "ОСВ" + else: + return document_type.value + + +def render_checklist_csv(checklist: ChecklistOut) -> bytes: + """Чек-лист в CSV: UTF-8 с BOM, разделитель `;`, CRLF. + + Заголовок: Организация;ИНН;<правило>… + Строки: имя ЮЛ, ИНН, ячейки по порядку правил. + Статус ячейки: подпись из CELL_STATUS_LABELS, с " (нет: Q1, Q2)" при missing_quarters. + """ + buffer = io.StringIO() + writer = csv.writer(buffer, delimiter=";", lineterminator="\r\n") + + # Заголовок + rule_labels = [_rule_label(rule.document_type, rule.account) for rule in checklist.rules] + header = ["Организация", "ИНН"] + rule_labels + writer.writerow(header) + + # Строки + for row in checklist.rows: + cell_values = [] + for cell, _rule in zip(row.cells, checklist.rules, strict=True): + # Получить подпись статуса + label = _CELL_STATUS_LABELS.get(cell.status, cell.status.value) + + # Добавить missing_quarters если есть + if cell.missing_quarters: + quarters_str = ", ".join(str(q.value) for q in cell.missing_quarters) + label = f"{label} (нет: {quarters_str})" + + cell_values.append(label) + + writer.writerow([row.legal_entity_name, row.legal_entity_inn] + cell_values) + + # Кодировать в UTF-8 с BOM + csv_text = buffer.getvalue() + return ("" + csv_text).encode("utf-8") + + +def checklist_filename(group_name: str, year: int, quarter: str) -> str: + """Имя файла для Content-Disposition: checklist-<ГК>-<год>-<квартал>.csv""" + return f"checklist-{group_name}-{year}-{quarter}.csv" diff --git a/api/src/services/consolidations.py b/api/src/services/consolidations.py index 5080738..01e54aa 100644 --- a/api/src/services/consolidations.py +++ b/api/src/services/consolidations.py @@ -191,6 +191,20 @@ class ConsolidationService: ], ) + async def version_title(self, version_id: uuid.UUID) -> tuple[str, int, str]: + """Название ГК, год и квартал версии для имени файла экспорта (CSV чек-листа). + + Версия вне видимости и несуществующая — один и тот же 404 как в `checklist`. + """ + found = await repo.get_version(self._session, self._scope, version_id) + if found is None: + raise NotFoundError(VERSION_NOT_FOUND) + version, consolidation = found + group = await directories_repo.get_company_group(self._session, self._scope, consolidation.company_group_id) + if group is None: + raise NotFoundError(VERSION_NOT_FOUND) + return group.name, consolidation.year, consolidation.quarter.value + async def _checklist_cells( self, version: ConsolidationVersion, consolidation: Consolidation, entities: list[LegalEntity] ) -> tuple[list[checklist_domain.Rule], list[tuple[LegalEntity, list[checklist_domain.Cell]]]]: diff --git a/api/tests/test_api_consolidation_checklist.py b/api/tests/test_api_consolidation_checklist.py index d37a093..5924a05 100644 --- a/api/tests/test_api_consolidation_checklist.py +++ b/api/tests/test_api_consolidation_checklist.py @@ -241,3 +241,22 @@ async def test_checklist_unconfirmed_of_entity_outside_perimeter_has_no_effect( assert [row["legal_entity_name"] for row in body["rows"]] == ["ООО А", "ООО Б"] assert all({cell["status"] for cell in row["cells"]} == {"MISSING"} for row in body["rows"]) + + +@pytest.mark.integration +async def test_checklist_csv_format(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: + """Чек-лист в CSV: format=csv → text/csv, BOM, Content-Disposition с filename*.""" + version, (first, _) = await _bound_world(seeded) + client = await auth_client(KA_RF) + + response = await client.get(f"/consolidations/versions/{version.id}/checklist?format=csv") + + assert response.status_code == HTTP_OK, response.text + assert response.headers["content-type"].startswith("text/csv") + assert response.content.startswith("".encode()) + lines = response.content.decode("utf-8-sig").splitlines() + assert lines[0] == "Организация;ИНН;Бух. баланс;ОФР;ОСВ 01;ОСВ 58;ОСВ 60;ОСВ 62;ОСВ 66;ОСВ 67;ОСВ 76;ОСВ 90" + assert "Загружен" in lines[1] # первого ЮЛ подтверждён ОСВ 60 + disposition = response.headers.get("content-disposition", "") + assert "filename*=UTF-8''" in disposition + assert "checklist-" in disposition diff --git a/api/tests/test_domain_checklist_csv.py b/api/tests/test_domain_checklist_csv.py new file mode 100644 index 0000000..d5c554a --- /dev/null +++ b/api/tests/test_domain_checklist_csv.py @@ -0,0 +1,37 @@ +"""CSV чек-листа (spec consolidations-api «Чек-лист комплекта»; D-13).""" + +import uuid + +from src.domain.checklist import CellStatus +from src.domain.checklist_csv import checklist_filename, render_checklist_csv +from src.models.enums import DocumentType, PeriodQuarter +from src.schemas.consolidations import ChecklistCellOut, ChecklistOut, ChecklistRowOut, ChecklistRuleOut + + +def test_render() -> None: + checklist = ChecklistOut( + rules=[ + ChecklistRuleOut(document_type=DocumentType.BALANCE, account=None, is_mandatory=True), + ChecklistRuleOut(document_type=DocumentType.OSV, account="60", is_mandatory=True), + ], + rows=[ + ChecklistRowOut( + legal_entity_id=uuid.uuid4(), + legal_entity_name='ООО "Альфа; Бета"', + legal_entity_inn="7700000000", + cells=[ + ChecklistCellOut(status=CellStatus.CONFIRMED, missing_quarters=[], document_ids=[]), + ChecklistCellOut(status=CellStatus.MISSING, missing_quarters=[PeriodQuarter.Q2], document_ids=[]), + ], + ) + ], + ) + content = render_checklist_csv(checklist) + assert content.startswith("".encode()) + lines = content.decode("utf-8-sig").splitlines() + assert lines[0] == "Организация;ИНН;Бух. баланс;ОСВ 60" + assert lines[1] == '"ООО ""Альфа; Бета""";7700000000;Загружен;Отсутствует (нет: Q2)' + + +def test_filename() -> None: + assert checklist_filename("Русский бисквит", 2025, "Q3") == "checklist-Русский бисквит-2025-Q3.csv"