feat(consolidation): экспорт сводной ОСВ в xlsx с листом пробелов
GET /consolidations/versions/{id}/osv/export: лист «Сводная ОСВ» (15 колонок D-14, суммы — числовые
ячейки #,##0.00 из Decimal) и лист «Пробелы» по своду целиком; фильтры как у GET …/osv без пагинации;
Content-Disposition с filename* (чистый ASCII); audit_event(EXPORT) с filters и rows.
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
parent
e246531c70
commit
8e82c2c19f
@ -2,7 +2,7 @@
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from src.api.endpoints.consolidations import checklist, locks, mapping, members, osv, periods, status, versions
|
||||
from src.api.endpoints.consolidations import checklist, export, locks, mapping, members, osv, periods, status, versions
|
||||
|
||||
router = APIRouter(tags=["consolidations"])
|
||||
router.include_router(periods.router)
|
||||
@ -12,4 +12,5 @@ router.include_router(status.router)
|
||||
router.include_router(members.router)
|
||||
router.include_router(checklist.router)
|
||||
router.include_router(osv.router)
|
||||
router.include_router(export.router)
|
||||
router.include_router(mapping.router)
|
||||
|
||||
55
api/src/api/endpoints/consolidations/export.py
Normal file
55
api/src/api/endpoints/consolidations/export.py
Normal file
@ -0,0 +1,55 @@
|
||||
"""Экспорт сводной ОСВ версии в xlsx (spec consolidated-osv «Экспорт xlsx», audit-trail `EXPORT`; design D-14).
|
||||
|
||||
Право проверяется до видимости; версия ГК вне видимости — тот же 404, что у несуществующей; версия читается в любом
|
||||
статусе. Файл строится в памяти; единственная запись в БД — `audit_event(EXPORT)` в транзакции запроса."""
|
||||
|
||||
import uuid
|
||||
from typing import Annotated, Any
|
||||
from urllib.parse import quote
|
||||
|
||||
from fastapi import APIRouter, Depends, Query, Response, status
|
||||
|
||||
from src.api.dependencies.audit import RequestAudit, get_request_audit
|
||||
from src.api.dependencies.auth import require
|
||||
from src.api.dependencies.consolidations import ConsolidatedOsvServiceDep
|
||||
from src.schemas.common import ErrorDetail
|
||||
from src.schemas.consolidations import OsvFilters
|
||||
|
||||
router = APIRouter(prefix="/consolidations/versions")
|
||||
|
||||
XLSX_MEDIA_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||||
VIEW = [Depends(require("CONSOLIDATION_VIEW"))]
|
||||
RESPONSES: dict[int | str, dict[str, Any]] = {
|
||||
status.HTTP_200_OK: {
|
||||
"description": "файл xlsx: листы «Сводная ОСВ» и «Пробелы»",
|
||||
"content": {XLSX_MEDIA_TYPE: {"schema": {"type": "string", "format": "binary"}}},
|
||||
},
|
||||
status.HTTP_404_NOT_FOUND: {"model": ErrorDetail, "description": "нет или вне видимости"},
|
||||
}
|
||||
|
||||
|
||||
def content_disposition(filename: str) -> str:
|
||||
"""Заголовок вложения: ASCII-имя по умолчанию и `filename*` (RFC 5987) с полным percent-кодированием имени —
|
||||
пробел, кавычки, `;`, `/`, переводы строк и кириллица экранируются, заголовок остаётся чистым ASCII."""
|
||||
return f"attachment; filename=\"svod-osv.xlsx\"; filename*=UTF-8''{quote(filename, safe='')}"
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{version_id}/osv/export",
|
||||
response_class=Response,
|
||||
dependencies=VIEW,
|
||||
responses=RESPONSES,
|
||||
summary="Экспорт сводной ОСВ версии в xlsx",
|
||||
)
|
||||
async def export_osv(
|
||||
version_id: uuid.UUID,
|
||||
service: ConsolidatedOsvServiceDep,
|
||||
audit: Annotated[RequestAudit, Depends(get_request_audit)],
|
||||
filters: Annotated[OsvFilters, Query()],
|
||||
) -> Response:
|
||||
"""Файл `svod-osv_<ГК>_<год>-<квартал>_v<n>.xlsx`: строки свода после фильтров (как у `GET …/osv`, без
|
||||
пагинации, в том же порядке) и лист пробелов по своду целиком. Пишет `audit_event(EXPORT)`."""
|
||||
content, filename = await service.export(version_id, filters, audit)
|
||||
return Response(
|
||||
content=content, media_type=XLSX_MEDIA_TYPE, headers={"Content-Disposition": content_disposition(filename)}
|
||||
)
|
||||
@ -18,13 +18,14 @@ from typing import Literal
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.api.dependencies.audit import RequestAudit
|
||||
from src.common.exceptions import DomainValidationError, NotFoundError
|
||||
from src.domain import consolidated_osv as domain
|
||||
from src.domain.aliases import normalize_alias
|
||||
from src.domain.policy import VisibilityScope
|
||||
from src.models.consolidation import Consolidation, ConsolidationVersion
|
||||
from src.models.directories import CompanyGroup, LegalEntity, RelatedGroup
|
||||
from src.models.enums import DocumentType, RelatedGroupKind
|
||||
from src.models.enums import AuditAction, AuditSection, DocumentType, RelatedGroupKind
|
||||
from src.models.files import Document
|
||||
from src.models.osv import OsvEntry
|
||||
from src.repositories import consolidations as repo
|
||||
@ -40,6 +41,7 @@ from src.schemas.consolidations import (
|
||||
RefOut,
|
||||
)
|
||||
from src.services.consolidation_guard import VERSION_NOT_FOUND
|
||||
from src.services.osv_export import build_workbook
|
||||
|
||||
__all__ = ["MAX_CANDIDATES", "MIN_QUERY_ALL", "CandidateScope", "ConsolidatedOsvService", "domain_filter", "row_out"]
|
||||
|
||||
@ -168,6 +170,35 @@ class ConsolidatedOsvService:
|
||||
items=[row_out(row) for row in window], total=len(rows), limit=page.limit, offset=page.offset
|
||||
)
|
||||
|
||||
async def export(self, version_id: uuid.UUID, f: OsvFilters, audit: RequestAudit) -> tuple[bytes, str]:
|
||||
"""xlsx сводной ОСВ (spec «Экспорт xlsx», D-14): строки после тех же фильтров, что у `page`, без пагинации, в
|
||||
порядке D-16; лист пробелов — по своду целиком. Имя файла — `svod-osv_<ГК>_<год>-<квартал>_v<n>.xlsx` (без
|
||||
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={})
|
||||
group_name = (
|
||||
await self._session.execute(
|
||||
select(CompanyGroup.name).where(CompanyGroup.id == consolidation.company_group_id)
|
||||
)
|
||||
).scalar_one()
|
||||
await audit.record(
|
||||
action=AuditAction.EXPORT,
|
||||
section=AuditSection.EXPORT,
|
||||
object_type="consolidation_version",
|
||||
object_id=version.id,
|
||||
consolidation_version_id=version.id,
|
||||
after_value={
|
||||
"report": "consolidated_osv",
|
||||
"filters": f.model_dump(mode="json", exclude_none=True),
|
||||
"rows": len(rows),
|
||||
},
|
||||
)
|
||||
filename = (
|
||||
f"svod-osv_{group_name}_{consolidation.year}-{consolidation.quarter.value}_v{version.version_no}.xlsx"
|
||||
)
|
||||
return content, filename
|
||||
|
||||
async def summary(self, version_id: uuid.UUID) -> OsvSummaryOut:
|
||||
"""Сводка: строки, несопоставленные, автосопоставленные по имени/алиасу, ОСВ-документы версии и пробелы."""
|
||||
consolidated, _, _ = await self._build(version_id)
|
||||
|
||||
95
api/src/services/osv_export.py
Normal file
95
api/src/services/osv_export.py
Normal file
@ -0,0 +1,95 @@
|
||||
"""Книга xlsx сводной ОСВ (spec consolidated-osv «Экспорт xlsx»; design D-14): чистая функция без БД.
|
||||
|
||||
Лист «Сводная ОСВ» — строки свода в переданном порядке (порядок API, D-16), шесть сумм — числовые ячейки
|
||||
`#,##0.00` в рублях без деления. В ячейки пишется `Decimal` как есть: `float` в нашем коде не используется;
|
||||
`openpyxl` сам сериализует число в XML (формат xlsx хранит числа как IEEE-754 double, поэтому значения свыше
|
||||
~10^15 теряют младшие разряды — для сумм ОСВ в рублях запас на три порядка). Лист «Пробелы» — пробелы свода
|
||||
целиком (`Consolidated.gaps`), фильтры строк на него не влияют; без пробелов — одна ячейка «пробелов нет».
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from collections.abc import Mapping, Sequence
|
||||
from io import BytesIO
|
||||
|
||||
from openpyxl import Workbook
|
||||
from openpyxl.utils import get_column_letter
|
||||
from openpyxl.worksheet.worksheet import Worksheet
|
||||
|
||||
from src.domain.consolidated_osv import Consolidated, ConsolidatedRow
|
||||
|
||||
__all__ = ["GAPS_SHEET", "MAIN_SHEET", "MONEY_FORMAT", "NO_GAPS", "build_workbook"]
|
||||
|
||||
MAIN_SHEET = "Сводная ОСВ"
|
||||
GAPS_SHEET = "Пробелы"
|
||||
MONEY_FORMAT = "#,##0.00"
|
||||
NO_GAPS = "пробелов нет"
|
||||
MAIN_HEADER = (
|
||||
"Компания ГК", "ИНН компании", "Счёт", "Период", "Контрагент", "ИНН контрагента", "Сальдо начало Дт",
|
||||
"Сальдо начало Кт", "Обороты Дт", "Обороты Кт", "Сальдо конец Дт", "Сальдо конец Кт", "Тип контрагента",
|
||||
"Сопоставлено с", "Способ сопоставления",
|
||||
) # fmt: skip
|
||||
GAPS_HEADER = ("Компания ГК", "Счёт", "Нет ОСВ за кварталы")
|
||||
FIRST_MONEY_COLUMN, LAST_MONEY_COLUMN = 7, 12
|
||||
|
||||
|
||||
def _widths(sheet: Worksheet, widths: Sequence[int]) -> None:
|
||||
for index, width in enumerate(widths, start=1):
|
||||
sheet.column_dimensions[get_column_letter(index)].width = width
|
||||
|
||||
|
||||
def _main_sheet(sheet: Worksheet, rows: Sequence[ConsolidatedRow], entity_names: Mapping[uuid.UUID, str]) -> 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
|
||||
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:
|
||||
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])
|
||||
_widths(sheet, (32, 8, 24))
|
||||
|
||||
|
||||
def build_workbook(
|
||||
consolidated: Consolidated, rows: Sequence[ConsolidatedRow], *, entity_names: Mapping[uuid.UUID, str]
|
||||
) -> bytes:
|
||||
"""Книга из двух листов: `rows` (уже отфильтрованные строки свода) и пробелы `consolidated.gaps`.
|
||||
|
||||
`entity_names` — отображаемые имена ЮЛ по идентификатору (переопределяют имена из строк и пробелов; нет в
|
||||
словаре — имя из самой строки/пробела). Только память: `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)
|
||||
buffer = BytesIO()
|
||||
workbook.save(buffer)
|
||||
return buffer.getvalue()
|
||||
323
api/tests/test_api_osv_export.py
Normal file
323
api/tests/test_api_osv_export.py
Normal file
@ -0,0 +1,323 @@
|
||||
"""Экспорт сводной ОСВ в xlsx (spec consolidated-osv «Экспорт xlsx», audit-trail EXPORT; design D-14): файл читается
|
||||
обратно `openpyxl`, суммы — числа `#,##0.00` без `float` в нашем коде, лист «Пробелы», фильтры как у `GET …/osv`,
|
||||
заголовок `Content-Disposition`, аудит, права и видимость, никаких записей кроме `audit_event`."""
|
||||
|
||||
import io
|
||||
import uuid
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
from urllib.parse import unquote
|
||||
|
||||
import pytest
|
||||
from openpyxl import load_workbook
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models.access import AppUser
|
||||
from src.models.audit_ex import AuditEvent
|
||||
from src.models.consolidation import ConsolidationDocument, ConsolidationVersion
|
||||
from src.models.directories import CompanyGroup, LegalEntity, RegionalOffice
|
||||
from src.models.enums import AuditAction, AuditSection, ConsolidationStatus, CounterpartyMatchKind, PeriodQuarter
|
||||
from src.models.osv import OsvEntry
|
||||
from tests.conftest import AuthClientFactory
|
||||
from tests.factories import make_company_group, make_confirmed_osv, make_consolidation, make_legal_entity
|
||||
|
||||
HTTP_OK = 200
|
||||
HTTP_FORBIDDEN = 403
|
||||
HTTP_NOT_FOUND = 404
|
||||
XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||||
KA_RF = "ka_rf@armka.local"
|
||||
ROSTOV = "Ростовский региональный филиал"
|
||||
YEAR = 2024
|
||||
Q1, Q2, Q3 = PeriodQuarter.Q1, PeriodQuarter.Q2, PeriodQuarter.Q3
|
||||
D = Decimal
|
||||
MAIN, GAPS = "Сводная ОСВ", "Пробелы"
|
||||
HEADER = [
|
||||
"Компания ГК", "ИНН компании", "Счёт", "Период", "Контрагент", "ИНН контрагента", "Сальдо начало Дт",
|
||||
"Сальдо начало Кт", "Обороты Дт", "Обороты Кт", "Сальдо конец Дт", "Сальдо конец Кт", "Тип контрагента",
|
||||
"Сопоставлено с", "Способ сопоставления",
|
||||
] # fmt: skip
|
||||
GAPS_HEADER = ["Компания ГК", "Счёт", "Нет ОСВ за кварталы"]
|
||||
ROWS_TOTAL = 3
|
||||
|
||||
|
||||
class World:
|
||||
"""Версия Q3 2024, периметр {А, Б}. «А» по счёту 60 — Q1..Q3 (3 строки свода: «ООО Б» ГСК, «Неизвестный»
|
||||
без сопоставления, «ООО Внешний» — имя найдётся дорезолвом); «Б» по счёту 60 — Q1 и Q3 (пробел Q2)."""
|
||||
|
||||
version: ConsolidationVersion
|
||||
group: CompanyGroup
|
||||
a: LegalEntity
|
||||
b: LegalEntity
|
||||
external: LegalEntity
|
||||
|
||||
|
||||
async def _world(
|
||||
session: AsyncSession, *, group_name: str = "ГК Русский бисквит", gap: bool = True,
|
||||
status: ConsolidationStatus = ConsolidationStatus.DATA_CORRECTION,
|
||||
) -> World: # fmt: skip
|
||||
office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
|
||||
world = World()
|
||||
world.group = await make_company_group(session, name=group_name, office=office)
|
||||
world.a = await make_legal_entity(session, name="ООО А", group=world.group)
|
||||
world.b = await make_legal_entity(session, name="ООО Б", group=world.group)
|
||||
world.external = await make_legal_entity(session, name="ООО Внешний")
|
||||
user = (await session.execute(select(AppUser).where(AppUser.email == KA_RF))).scalar_one()
|
||||
_, world.version = await make_consolidation(
|
||||
session, group=world.group, year=YEAR, quarter=Q3, members=[world.a, world.b], initiator=user, status=status
|
||||
)
|
||||
documents = []
|
||||
for quarter in (Q1, Q2, Q3):
|
||||
step = {Q1: 0, Q2: 10, Q3: 20}[quarter]
|
||||
amounts = {
|
||||
"opening_debit": D(100 + step) + D("0.05"),
|
||||
"turnover_debit": D("10.10"),
|
||||
"closing_credit": D("7.07"),
|
||||
}
|
||||
documents.append(
|
||||
await make_confirmed_osv(
|
||||
session, entity=world.a, account="60", year=YEAR, quarter=quarter, is_cumulative=False,
|
||||
rows=[
|
||||
{"name": "ООО Б", "inn": world.b.inn, "counterparty_id": world.b.id,
|
||||
"counterparty_match_kind": CounterpartyMatchKind.INN, **amounts},
|
||||
{"name": "Неизвестный контрагент", **amounts},
|
||||
{"name": "ООО Внешний", **amounts},
|
||||
],
|
||||
)
|
||||
) # fmt: skip
|
||||
for quarter in (Q1, Q3) if gap else (Q1, Q2, Q3):
|
||||
documents.append(
|
||||
await make_confirmed_osv(
|
||||
session, entity=world.b, account="60", year=YEAR, quarter=quarter, is_cumulative=False,
|
||||
rows=[
|
||||
{"name": "ООО А", "counterparty_id": world.a.id,
|
||||
"counterparty_match_kind": CounterpartyMatchKind.MANUAL, "turnover_credit": D("5.00")},
|
||||
],
|
||||
)
|
||||
) # fmt: skip
|
||||
for document in documents:
|
||||
session.add(ConsolidationDocument(consolidation_version_id=world.version.id, document_id=document.id))
|
||||
await session.flush()
|
||||
return world
|
||||
|
||||
|
||||
def _url(world: World, query: str = "") -> str:
|
||||
return f"/consolidations/versions/{world.version.id}/osv/export?{query}"
|
||||
|
||||
|
||||
def _book(content: bytes) -> Any:
|
||||
return load_workbook(io.BytesIO(content))
|
||||
|
||||
|
||||
async def _export_events(session: AsyncSession) -> list[AuditEvent]:
|
||||
stmt = select(AuditEvent).where(AuditEvent.action == AuditAction.EXPORT).order_by(AuditEvent.occurred_at)
|
||||
return list((await session.execute(stmt)).scalars())
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_export_reads_back_with_amounts_equal_to_api(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||||
) -> None:
|
||||
world = await _world(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await client.get(_url(world))
|
||||
|
||||
assert response.status_code == HTTP_OK, response.text
|
||||
assert response.headers["content-type"] == XLSX
|
||||
book = _book(response.content)
|
||||
assert book.sheetnames == [MAIN, GAPS]
|
||||
sheet = book[MAIN]
|
||||
rows = list(sheet.iter_rows(values_only=True))
|
||||
assert list(rows[0]) == HEADER
|
||||
assert len(rows) == 1 + ROWS_TOTAL # заголовок и 3 строки: «Б» без строк (пробел)
|
||||
api_items = (await client.get(f"/consolidations/versions/{world.version.id}/osv")).json()["items"]
|
||||
assert len(api_items) == ROWS_TOTAL
|
||||
money = ("opening_debit", "opening_credit", "turnover_debit", "turnover_credit", "closing_debit", "closing_credit")
|
||||
for item, row in zip(api_items, rows[1:], strict=True):
|
||||
assert row[:6] == (item["legal_entity_name"], item["legal_entity_inn"], item["account_code"], item["period"],
|
||||
item["counterparty_name"], item["counterparty_inn"]) # fmt: skip
|
||||
assert [Decimal(str(value)) for value in row[6:12]] == [Decimal(item[key]) for key in money]
|
||||
assert row[12:] == (item["counterparty_type"], item["counterparty_entity_name"], item["match_kind"])
|
||||
assert sheet.cell(2, 7).value == pytest.approx(100.05) # 100 + 0.05 (Q1): число, не строка
|
||||
for cell in sheet[2][6:12]:
|
||||
assert cell.number_format == "#,##0.00" and cell.data_type == "n"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_gaps_sheet_lists_missing_quarters(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
world = await _world(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
book = _book((await client.get(_url(world))).content)
|
||||
|
||||
assert [list(row) for row in book[GAPS].iter_rows(values_only=True)] == [GAPS_HEADER, ["ООО Б", "60", "Q2"]]
|
||||
events = await _export_events(seeded)
|
||||
assert len(events) == 1
|
||||
event = events[0]
|
||||
assert (event.section, event.object_type, event.object_id) == (AuditSection.EXPORT, "consolidation_version",
|
||||
world.version.id) # fmt: skip
|
||||
assert event.consolidation_version_id == world.version.id
|
||||
assert event.after_value == {"report": "consolidated_osv", "filters": {}, "rows": ROWS_TOTAL}
|
||||
assert event.before_value is None and event.user_id is not None
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_gap_quarters_joined_and_no_gaps_cell(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
world = await _world(seeded, gap=False)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
book = _book((await client.get(_url(world))).content)
|
||||
|
||||
assert [list(row) for row in book[GAPS].iter_rows(values_only=True)] == [["пробелов нет"]]
|
||||
assert len(list(book[MAIN].iter_rows())) == 1 + ROWS_TOTAL + 1 # «Б» теперь даёт строку
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_unmapped_filter_after_reresolve_gives_header_only_and_filled_gaps(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||||
) -> None:
|
||||
"""Review Focus 5: у всех строк с `counterparty_id IS NULL` имя находится дорезолвом — несопоставленных нет."""
|
||||
world = await _world(seeded)
|
||||
for entry in (
|
||||
await seeded.execute(select(OsvEntry).where(OsvEntry.counterparty_name == "Неизвестный контрагент"))
|
||||
).scalars():
|
||||
entry.counterparty_name = "ООО Внешний"
|
||||
await seeded.flush()
|
||||
client = await auth_client(KA_RF)
|
||||
assert (await client.get(f"/consolidations/versions/{world.version.id}/osv?unmapped=true")).json()["total"] == 0
|
||||
|
||||
response = await client.get(_url(world, "unmapped=true"))
|
||||
|
||||
assert response.status_code == HTTP_OK
|
||||
book = _book(response.content)
|
||||
assert [list(row) for row in book[MAIN].iter_rows(values_only=True)] == [HEADER]
|
||||
assert [list(row) for row in book[GAPS].iter_rows(values_only=True)] == [GAPS_HEADER, ["ООО Б", "60", "Q2"]]
|
||||
(event,) = await _export_events(seeded)
|
||||
assert event.after_value == {"report": "consolidated_osv", "filters": {"unmapped": True}, "rows": 0}
|
||||
|
||||
|
||||
FILTERS = [
|
||||
"",
|
||||
"unmapped=true",
|
||||
"unmapped=false",
|
||||
"counterparty_type=GSK",
|
||||
"counterparty_type=EXTERNAL&account_root=60",
|
||||
"match_kind=INN&match_kind=MANUAL",
|
||||
"match_kind=NAME",
|
||||
"q=%D0%B2%D0%BD%D0%B5%D1%88",
|
||||
"account_root=62",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize("query", FILTERS)
|
||||
async def test_filters_match_page_endpoint_rows_and_order(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, query: str
|
||||
) -> None:
|
||||
world = await _world(seeded, gap=False)
|
||||
client = await auth_client(KA_RF)
|
||||
page = (await client.get(f"/consolidations/versions/{world.version.id}/osv?{query}&limit=2000")).json()
|
||||
|
||||
book = _book((await client.get(_url(world, query))).content)
|
||||
|
||||
exported = list(book[MAIN].iter_rows(min_row=2, values_only=True))
|
||||
assert [(row[0], row[2], row[4], row[12]) for row in exported] == [
|
||||
(item["legal_entity_name"], item["account_code"], item["counterparty_name"], item["counterparty_type"])
|
||||
for item in page["items"]
|
||||
]
|
||||
(*_, event) = await _export_events(seeded)
|
||||
assert event.after_value["rows"] == page["total"] == len(exported)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_filters_are_audited_as_json(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
world = await _world(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
query = f"legal_entity_id={world.a.id}&counterparty_type=GSK&match_kind=INN&match_kind=NAME&q=%D0%B1"
|
||||
|
||||
assert (await client.get(_url(world, query))).status_code == HTTP_OK
|
||||
|
||||
(event,) = await _export_events(seeded)
|
||||
assert event.after_value["filters"] == {
|
||||
"legal_entity_id": str(world.a.id), "counterparty_type": "GSK", "match_kind": ["INN", "NAME"], "q": "б",
|
||||
} # fmt: skip
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.parametrize("group_name", ["ГК Русский бисквит", 'ГК "Кавычки" / слэш; a=b\r\nX-Inj: 1', "ГК Ёлка ё'ё"])
|
||||
async def test_content_disposition_is_ascii_and_cannot_be_broken(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession, group_name: str
|
||||
) -> None:
|
||||
world = await _world(seeded, group_name=group_name)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await client.get(_url(world))
|
||||
|
||||
assert response.status_code == HTTP_OK
|
||||
header = response.headers["content-disposition"]
|
||||
header.encode("ascii")
|
||||
assert header.startswith("attachment; filename=\"svod-osv.xlsx\"; filename*=UTF-8''svod-osv_")
|
||||
encoded = header.split("filename*=UTF-8''", 1)[1]
|
||||
assert encoded.endswith("_2024-Q3_v1.xlsx")
|
||||
assert all(ch.isalnum() or ch in "%_.-~" for ch in encoded), encoded # пробел, кавычки, `;`, `/`, `'` экранированы
|
||||
assert len(response.headers.get_list("content-disposition")) == 1 and "x-inj" not in response.headers
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_filename_decodes_to_group_year_quarter_version(
|
||||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||||
) -> None:
|
||||
world = await _world(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
header = (await client.get(_url(world))).headers["content-disposition"]
|
||||
|
||||
assert unquote(header.split("filename*=UTF-8''", 1)[1]) == "svod-osv_ГК Русский бисквит_2024-Q3_v1.xlsx"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_cancelled_version_is_exportable(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
world = await _world(seeded, status=ConsolidationStatus.CANCELLED)
|
||||
client = await auth_client(KA_RF)
|
||||
|
||||
response = await client.get(_url(world))
|
||||
|
||||
assert response.status_code == HTTP_OK
|
||||
assert len(list(_book(response.content)[MAIN].iter_rows())) == 1 + ROWS_TOTAL
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_rights_and_visibility(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
world = await _world(seeded)
|
||||
curator = await auth_client("curator_go@armka.local")
|
||||
expert = await auth_client("expert@armka.local")
|
||||
foreign = await auth_client("client_manager@armka.local") # Краснодарский филиал
|
||||
admin = await auth_client("admin@armka.local") # права CONSOLIDATION_VIEW у роли нет
|
||||
|
||||
assert (await curator.get(_url(world))).status_code == HTTP_OK
|
||||
assert (await expert.get(_url(world))).status_code == HTTP_OK
|
||||
hidden = await foreign.get(_url(world))
|
||||
unknown = await foreign.get(f"/consolidations/versions/{uuid.uuid4()}/osv/export")
|
||||
assert hidden.status_code == unknown.status_code == HTTP_NOT_FOUND
|
||||
assert hidden.json() == unknown.json()
|
||||
assert (await admin.get(_url(world))).status_code == HTTP_FORBIDDEN
|
||||
events = await _export_events(seeded)
|
||||
assert len(events) == 2 # noqa: PLR2004 - curator и expert; отказы и 404 не аудируются
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_export_writes_only_the_audit_event(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||||
"""Дорезолв не пишется в `osv_entry`, свод не хранится: меняется только число `audit_event`."""
|
||||
world = await _world(seeded)
|
||||
client = await auth_client(KA_RF)
|
||||
before = (await seeded.execute(select(OsvEntry.id, OsvEntry.counterparty_id).order_by(OsvEntry.id))).all()
|
||||
others = select(func.count()).select_from(AuditEvent).where(AuditEvent.action != AuditAction.LOGIN)
|
||||
events_before = (await seeded.execute(others)).scalar_one() # вход (LOGIN) пишет клиент теста, не экспорт
|
||||
|
||||
assert (await client.get(_url(world))).status_code == HTTP_OK
|
||||
|
||||
after = (await seeded.execute(select(OsvEntry.id, OsvEntry.counterparty_id).order_by(OsvEntry.id))).all()
|
||||
assert after == before
|
||||
assert (await seeded.execute(others)).scalar_one() == events_before + 1
|
||||
assert len(await _export_events(seeded)) == 1
|
||||
@ -61,6 +61,7 @@ EXPECTED_ROUTES: dict[str, set[str]] = {
|
||||
"/api/v1/consolidations/versions/{version_id}/checklist": {"get"},
|
||||
"/api/v1/consolidations/versions/{version_id}/osv": {"get"},
|
||||
"/api/v1/consolidations/versions/{version_id}/osv/summary": {"get"},
|
||||
"/api/v1/consolidations/versions/{version_id}/osv/export": {"get"},
|
||||
"/api/v1/consolidations/versions/{version_id}/counterparty-candidates": {"get"},
|
||||
"/api/v1/consolidations/versions/{version_id}/osv/entries/{entry_id}/counterparty": {"post"},
|
||||
}
|
||||
@ -161,6 +162,17 @@ def test_consolidation_routes_declare_documented_responses(schema: dict[str, Any
|
||||
assert summary["200"]["content"]["application/json"]["schema"]["$ref"].endswith("/OsvSummaryOut")
|
||||
|
||||
|
||||
def test_osv_export_route_declares_binary_xlsx_response(schema: dict[str, Any]) -> None:
|
||||
"""Экспорт: 200 — файл xlsx (бинарный контент), 404/422; фильтры те же, что у страницы, но без пагинации."""
|
||||
export = schema["paths"]["/api/v1/consolidations/versions/{version_id}/osv/export"]["get"]
|
||||
assert {"200", "404", "422"} <= set(export["responses"])
|
||||
xlsx = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||||
assert export["responses"]["200"]["content"][xlsx]["schema"] == {"type": "string", "format": "binary"}
|
||||
names = {parameter["name"] for parameter in export["parameters"]}
|
||||
assert {"legal_entity_id", "account_root", "counterparty_type", "unmapped", "match_kind", "q"} <= names
|
||||
assert not {"limit", "offset"} & names
|
||||
|
||||
|
||||
def test_lock_and_status_routes_declare_documented_responses(schema: dict[str, Any]) -> None:
|
||||
"""Взятие — 201/200 с `LockOut`, 404/409/422; снятие — 204/404/409; смена статуса — карточка, 404/409/422."""
|
||||
lock = schema["paths"]["/api/v1/consolidations/versions/{version_id}/lock"]
|
||||
|
||||
@ -236,9 +236,9 @@
|
||||
- `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)}'})`.
|
||||
|
||||
- [ ] **Step 1: Тесты.** Сценарий «Экспорт читается обратно» (`openpyxl.load_workbook(BytesIO(resp.content))`: заголовки, 3 строки, `Decimal(str(cell.value))` равен API, лист «Пробелы» с `Q2`, аудит `rows = 3`); Review Focus 5 — пустой лист при `unmapped=true`; `Content-Disposition` содержит `filename*=UTF-8''`; `expert` (VIEW) → 200, без права → 403.
|
||||
- [ ] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_api_osv_export.py -q` → PASS.
|
||||
- [ ] **Step 5: Commit** `feat(consolidation): экспорт сводной ОСВ в xlsx с листом пробелов`.
|
||||
- [x] **Step 1: Тесты.** Сценарий «Экспорт читается обратно» (`openpyxl.load_workbook(BytesIO(resp.content))`: заголовки, 3 строки, `Decimal(str(cell.value))` равен API, лист «Пробелы» с `Q2`, аудит `rows = 3`); Review Focus 5 — пустой лист при `unmapped=true`; `Content-Disposition` содержит `filename*=UTF-8''`; `expert` (VIEW) → 200, без права → 403.
|
||||
- [x] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_api_osv_export.py -q` → PASS.
|
||||
- [x] **Step 5: Commit** `feat(consolidation): экспорт сводной ОСВ в xlsx с листом пробелов`.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -255,3 +255,9 @@ legal_entity_name, account_code, missing_quarters[]}}`. Фильтры `GET …/
|
||||
| 68 | 2026-10-06 | `alias_moved_from` (ответ и `after_value` `OSV_COUNTERPARTY_MAP`) — один `legal_entity_id`: первый владелец по `(created_at, id)`; при нескольких владельцах (возможны только из данных до общей блокировки) удаляются все их алиасы, каждый со своим `LEGAL_ENTITY_CHANGE`; `ALIAS_TAKEN.owners` — все владельцы в том же порядке | схема `MapCounterpartyOut.alias_moved_from: UUID \| None` (Task 3) | остальные прежние владельцы видны только в `LEGAL_ENTITY_CHANGE` |
|
||||
| 69 | 2026-10-06 | `MapCounterpartyOut.row` — `ConsolidatedRowOut \| null`: строка привязанного документа, которая в свод не попадает (пара с пробелом, квартальный документ при нарастающем), сопоставляется, но строки свода для ответа нет — `row = null` | спека разрешает сопоставлять любую строку привязанного ОСВ-документа; выдумывать строку свода из одной исходной нельзя | фронт обрабатывает `null` (из UI такая строка не выбирается — её нет в своде) |
|
||||
| 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 — точная |
|
||||
| 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 — действие над версией; запись без выгруженного файла была бы ложной | при сбое после коммита (обрыв соединения при отдаче) событие остаётся, а файл клиент мог не получить |
|
||||
|
||||
@ -51,7 +51,7 @@
|
||||
- [x] 4.3 **Кандидаты и сопоставление.** `GET …/counterparty-candidates`, `POST …/osv/entries/{id}/counterparty`
|
||||
(`services/counterparty_mapping.py`: алиас, `ALIAS_TAKEN`, перенос, `MANUAL`/`PROPAGATED`, ИНН-противоречие, аудит).
|
||||
Тесты: все сценарии «Кандидаты», «Ручное сопоставление», блокировка/статус.
|
||||
- [ ] 4.4 **Экспорт.** `services/osv_export.py` (openpyxl, два листа, `BytesIO`), `GET …/osv/export`, `Content-Disposition`,
|
||||
- [x] 4.4 **Экспорт.** `services/osv_export.py` (openpyxl, два листа, `BytesIO`), `GET …/osv/export`, `Content-Disposition`,
|
||||
аудит `EXPORT`. Тест: файл читается `openpyxl`, суммы равны API, лист «Пробелы».
|
||||
|
||||
## 5. Сквозной тест и документы
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user