feat(consolidations): чек-лист в CSV — подписи макета, BOM, «;» (D-13)

Добавлен экспорт чек-листа в 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev 2026-10-07 13:23:36 +04:00
parent 6078532982
commit 0c203c1d91
5 changed files with 176 additions and 5 deletions

View File

@ -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

View File

@ -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"

View File

@ -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]]]]:

View File

@ -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

View File

@ -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"