diff --git a/api/src/api/endpoints/consolidations/intercompany_matrix.py b/api/src/api/endpoints/consolidations/intercompany_matrix.py index 04d12f5..5ba3fa8 100644 --- a/api/src/api/endpoints/consolidations/intercompany_matrix.py +++ b/api/src/api/endpoints/consolidations/intercompany_matrix.py @@ -1,22 +1,40 @@ -"""Расчёт шахматки по выбранным счетам или правилу элиминации.""" +"""Расчёт шахматки по всем или выбранным счетам.""" import uuid from typing import Any -from fastapi import APIRouter, Depends, status +from fastapi import APIRouter, Depends, Response, status +from starlette.concurrency import run_in_threadpool from src.api.dependencies.auth import require from src.api.dependencies.consolidations import IntercompanyMatrixServiceDep +from src.common.filenames import content_disposition from src.schemas.common import ErrorDetail from src.schemas.consolidations import IntercompanyMatrixManualIn, IntercompanyMatrixOut, IntercompanyResultTableOut +from src.services.intercompany_matrix_export import build_matrix_workbook router = APIRouter(prefix="/consolidations/versions") VIEW = [Depends(require("CONSOLIDATION_VIEW"))] ERRORS: dict[int | str, dict[str, Any]] = { - status.HTTP_404_NOT_FOUND: {"model": ErrorDetail, "description": "Версия или правило не найдены либо недоступны"}, - status.HTTP_422_UNPROCESSABLE_CONTENT: {"model": ErrorDetail, "description": "Неверный выбор счетов или правило"}, + status.HTTP_404_NOT_FOUND: {"model": ErrorDetail, "description": "Версия не найдена либо недоступна"}, + status.HTTP_422_UNPROCESSABLE_CONTENT: {"model": ErrorDetail, "description": "Неверный выбор счетов"}, } +XLSX_MEDIA_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" +EXPORT_RESPONSES = { + **ERRORS, + status.HTTP_200_OK: {"content": {XLSX_MEDIA_TYPE: {"schema": {"type": "string", "format": "binary"}}}}, +} + + +async def _xlsx_response(matrix: IntercompanyMatrixOut) -> Response: + content = await run_in_threadpool(build_matrix_workbook, matrix) + filename = "intercompany-matrix.xlsx" + return Response( + content=content, + media_type=XLSX_MEDIA_TYPE, + headers={"Content-Disposition": content_disposition(filename, ascii_fallback=filename)}, + ) @router.get( @@ -34,6 +52,21 @@ async def build_intercompany_matrix_all( return await service.build_accounts(version_id, mode="ALL") +@router.get( + "/{version_id}/intercompany-matrix/all/export", + response_class=Response, + dependencies=VIEW, + responses=EXPORT_RESPONSES, + summary="Скачать шахматку по всем счетам в XLSX", +) +async def export_intercompany_matrix_all( + version_id: uuid.UUID, + service: IntercompanyMatrixServiceDep, +) -> Response: + matrix = await service.build_accounts(version_id, mode="ALL") + return await _xlsx_response(matrix) + + @router.post( "/{version_id}/intercompany-matrix/manual", response_model=IntercompanyMatrixOut, @@ -55,6 +88,27 @@ async def build_intercompany_matrix_manual( ) +@router.post( + "/{version_id}/intercompany-matrix/manual/export", + response_class=Response, + dependencies=VIEW, + responses=EXPORT_RESPONSES, + summary="Скачать шахматку по выбранным счетам в XLSX", +) +async def export_intercompany_matrix_manual( + version_id: uuid.UUID, + selection: IntercompanyMatrixManualIn, + service: IntercompanyMatrixServiceDep, +) -> Response: + matrix = await service.build_accounts( + version_id, + mode="MANUAL", + debit_account_ids=selection.debit_account_ids, + credit_account_ids=selection.credit_account_ids, + ) + return await _xlsx_response(matrix) + + @router.get( "/{version_id}/intercompany-matrix/result-table", response_model=IntercompanyResultTableOut, @@ -68,19 +122,3 @@ async def get_intercompany_result_table( ) -> IntercompanyResultTableOut: """Пять фиксированных статей: суммы по компаниям и итоги по группе.""" return await service.build_result_table(version_id) - - -@router.get( - "/{version_id}/intercompany-matrix/presets/{rule_no}", - response_model=IntercompanyMatrixOut, - dependencies=VIEW, - responses=ERRORS, - summary="Рассчитать шахматку по пресету", -) -async def build_intercompany_matrix_preset( - version_id: uuid.UUID, - rule_no: int, - service: IntercompanyMatrixServiceDep, -) -> IntercompanyMatrixOut: - """Правило выбирается по номеру из версии правил консолидации.""" - return await service.build_preset(version_id, rule_no) diff --git a/api/src/api/endpoints/directories/__init__.py b/api/src/api/endpoints/directories/__init__.py index 9454158..d8972aa 100644 --- a/api/src/api/endpoints/directories/__init__.py +++ b/api/src/api/endpoints/directories/__init__.py @@ -2,10 +2,18 @@ from fastapi import APIRouter -from src.api.endpoints.directories import company_groups, import_, legal_entities, regional_offices, related_groups +from src.api.endpoints.directories import ( + accounts, + company_groups, + import_, + legal_entities, + regional_offices, + related_groups, +) router = APIRouter(prefix="/directories", tags=["directories"]) router.include_router(regional_offices.router) +router.include_router(accounts.router) router.include_router(company_groups.router) router.include_router(legal_entities.router) router.include_router(related_groups.router) diff --git a/api/src/api/endpoints/directories/accounts.py b/api/src/api/endpoints/directories/accounts.py new file mode 100644 index 0000000..a00f3e6 --- /dev/null +++ b/api/src/api/endpoints/directories/accounts.py @@ -0,0 +1,21 @@ +"""Плоский список счетов плана счетов для выбора статьи расходов.""" + +from fastapi import APIRouter, Depends + +from src.api.dependencies.auth import require +from src.api.dependencies.db import SessionDep +from src.repositories import directories as repo +from src.schemas.directories import AccountOut + +router = APIRouter() + + +@router.get( + "/accounts", + response_model=list[AccountOut], + dependencies=[Depends(require("DIRECTORY_VIEW"))], + summary="Список счетов плана счетов", +) +async def list_accounts(session: SessionDep) -> list[AccountOut]: + """ Все счета одним списком """ + return [AccountOut.model_validate(account) for account in await repo.list_accounts(session)] diff --git a/api/src/repositories/directories.py b/api/src/repositories/directories.py index 3b6a7ef..fb14d20 100644 --- a/api/src/repositories/directories.py +++ b/api/src/repositories/directories.py @@ -14,7 +14,9 @@ from sqlalchemy import Select, Sequence, and_, delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from src.domain.directory_import import DirectorySnapshot, EntitySnap, GroupSnap, OfficeSnap +from src.domain.osv_mapping import account_sort_key from src.domain.policy import VisibilityScope, company_groups_visible, legal_entities_visible +from src.models.accounting import ChartOfAccount from src.models.directories import ( CompanyGroup, CompanyGroupHistory, @@ -32,6 +34,11 @@ LIKE_ESCAPE = "\\" ALIAS_LOCK_PREFIX = "legal-entity-alias" # пространство ключей advisory-блокировки алиасов +async def list_accounts(session: AsyncSession) -> list[ChartOfAccount]: + accounts = (await session.scalars(select(ChartOfAccount))).all() + return sorted(accounts, key=lambda account: (account_sort_key(account.code), account.code)) + + def escape_like(text: str) -> str: """Экранирует `\\`, `%` и `_`, чтобы пользовательская строка искалась как подстрока, а не как шаблон LIKE.""" return text.replace(LIKE_ESCAPE, LIKE_ESCAPE * 2).replace("%", f"{LIKE_ESCAPE}%").replace("_", f"{LIKE_ESCAPE}_") diff --git a/api/src/schemas/consolidations.py b/api/src/schemas/consolidations.py index 55e4ccb..8e0e609 100644 --- a/api/src/schemas/consolidations.py +++ b/api/src/schemas/consolidations.py @@ -343,6 +343,9 @@ class IntercompanyMatrixCellOut(BaseModel): amount_b: Decimal variance_amount: Decimal variance_percent: Decimal | None + is_more_5_percent: bool + entry_osv_ids_debit: list[uuid.UUID] + entry_osv_ids_credit: list[uuid.UUID] class IntercompanyMatrixCompanyOut(BaseModel): diff --git a/api/src/schemas/directories.py b/api/src/schemas/directories.py index f0e60cb..3b6c892 100644 --- a/api/src/schemas/directories.py +++ b/api/src/schemas/directories.py @@ -6,7 +6,7 @@ from typing import Annotated, Literal, Self from pydantic import BaseModel, ConfigDict, Field, computed_field, field_validator, model_validator -from src.models.enums import EntityType, LeHistoryEvent, RelatedGroupKind +from src.models.enums import AccountKind, AccountRowType, EntityType, LeHistoryEvent, RelatedGroupKind NAME_MAX_LENGTH = 500 MAX_ALIASES = 100 # алиасов в одном теле формы ЮЛ (#84): граница входа, 422 до обращения к БД @@ -38,6 +38,32 @@ class RegionalOfficeOut(RegionalOfficeBrief): is_active: bool +class AccountOut(BaseModel): + """Одна строка плана счетов; parent_id сохраняет иерархию в плоском списке.""" + + model_config = ConfigDict(from_attributes=True) + + id: uuid.UUID + code: str + parent_id: uuid.UUID | None + name: str + kind: AccountKind | None + row_type: AccountRowType + is_off_balance: bool + + @computed_field + @property + def is_debit(self) -> bool: + """Счёт можно выбрать для дебетовой стороны: активный или активно-пассивный.""" + return self.kind in (AccountKind.A, AccountKind.AP) + + @computed_field + @property + def is_credit(self) -> bool: + """Счёт можно выбрать для кредитовой стороны: пассивный или активно-пассивный.""" + return self.kind in (AccountKind.P, AccountKind.AP) + + class CompanyGroupCreate(BaseModel): """Создание ГК: `code` и `valid_from` назначает сервис.""" diff --git a/api/src/services/intercompany_matrix.py b/api/src/services/intercompany_matrix.py index f6a95cb..38b4c28 100644 --- a/api/src/services/intercompany_matrix.py +++ b/api/src/services/intercompany_matrix.py @@ -5,18 +5,17 @@ from collections import defaultdict from collections.abc import Iterable, Sequence from dataclasses import dataclass from decimal import Decimal -from typing import Literal, cast +from typing import Literal from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from src.common.exceptions import DomainValidationError, NotFoundError +from src.common.exceptions import DomainValidationError from src.domain.consolidated_osv import ConsolidatedRow from src.domain.policy import VisibilityScope -from src.models.accounting import ChartOfAccount, EliminationRuleAccount +from src.models.accounting import ChartOfAccount from src.models.enums import AccountKind from src.repositories import consolidations as repo -from src.repositories import elimination_rules as rules_repo from src.schemas.consolidations import ( IntercompanyMatrixCellOut, IntercompanyMatrixCompanyOut, @@ -30,6 +29,7 @@ from src.services.consolidated_osv import ConsolidatedOsvService Measure = Literal["closing", "turnover"] Side = Literal["D", "K"] ZERO = Decimal("0") +VARIANCE_THRESHOLD_PERCENT = Decimal("5") DEBIT_KINDS = frozenset({AccountKind.A, AccountKind.AP}) CREDIT_KINDS = frozenset({AccountKind.P, AccountKind.AP}) @@ -66,12 +66,11 @@ RESULT_ARTICLES = ( @dataclass(frozen=True) class MatrixSelection: - """Конечные наборы счетов двух сторон и показатель выбранного режима или правила.""" + """Счета дебетовой и кредитовой сторон и показатель выбранного режима.""" company_1_account_ids: frozenset[uuid.UUID] company_2_account_ids: frozenset[uuid.UUID] measure: Measure - side_1: Side def _amount(row: ConsolidatedRow, measure: Measure, side: Side) -> Decimal: @@ -87,21 +86,19 @@ def calculate_matrix( member_companies: Sequence[IntercompanyMatrixCompanyOut], selection: MatrixSelection, ) -> IntercompanyMatrixOut: - """Суммы по парам компаний; счета уже раскрыты и исключения учтены. + """Суммы по парам компаний; - Сторона 1: ОСВ компании A по контрагенту Б. Сторона 2: ОСВ компании Б по контрагенту A. + Сторона 1: дебет ОСВ компании A по контрагенту Б. + Сторона 2: кредит ОСВ компании Б по контрагенту A. """ if selection.measure not in ("closing", "turnover"): raise ValueError(f"неизвестный показатель правила: {selection.measure}") - if selection.side_1 not in ("D", "K"): - raise ValueError(f"неизвестная сторона правила: {selection.side_1}") company_info = tuple({company.id: company for company in member_companies}.values()) company_ids = tuple(company.id for company in company_info) members = set(company_ids) - side_2: Side = "K" if selection.side_1 == "D" else "D" - amounts_1: dict[tuple[uuid.UUID, uuid.UUID], Decimal] = defaultdict(lambda: ZERO) - amounts_2: dict[tuple[uuid.UUID, uuid.UUID], Decimal] = defaultdict(lambda: ZERO) + amounts_1: dict[tuple[uuid.UUID, uuid.UUID], dict] = defaultdict(lambda: {"amount": ZERO, "entry_osv_ids": set()}) + amounts_2: dict[tuple[uuid.UUID, uuid.UUID], dict] = defaultdict(lambda: {"amount": ZERO, "entry_osv_ids": set()}) for row in rows: owner = row.legal_entity_id @@ -110,20 +107,29 @@ def calculate_matrix( continue pair = (owner, counterparty) if row.account_id in selection.company_1_account_ids: - amounts_1[pair] += _amount(row, selection.measure, selection.side_1) + item = amounts_1[pair] + item["amount"] += _amount(row, selection.measure, "D") + item["entry_osv_ids"].update(row.entry_ids) if row.account_id in selection.company_2_account_ids: - amounts_2[pair] += _amount(row, selection.measure, side_2) + item = amounts_2[pair] + item["amount"] += _amount(row, selection.measure, "K") + item["entry_osv_ids"].update(row.entry_ids) cells: list[IntercompanyMatrixCellOut] = [] for company_a in company_ids: for company_b in company_ids: if company_a == company_b: continue - amount_a = amounts_1.get((company_a, company_b), ZERO) - amount_b = amounts_2.get((company_b, company_a), ZERO) + item_a = amounts_1.get((company_a, company_b)) + item_b = amounts_2.get((company_b, company_a)) + + amount_a = item_a["amount"] if item_a is not None else ZERO + amount_b = item_b["amount"] if item_b is not None else ZERO + variance_amount = abs(amount_a - amount_b) base = min(amount_a, amount_b) variance_percent = None if base == ZERO else variance_amount / base * 100 + is_more_5_percent = False if base == ZERO else variance_percent > VARIANCE_THRESHOLD_PERCENT cells.append( IntercompanyMatrixCellOut( company_a_id=company_a, @@ -132,6 +138,9 @@ def calculate_matrix( amount_b=amount_b, variance_amount=variance_amount, variance_percent=variance_percent, + is_more_5_percent=is_more_5_percent, + entry_osv_ids_debit=item_a["entry_osv_ids"] if item_a is not None else (), + entry_osv_ids_credit=item_b["entry_osv_ids"] if item_b is not None else (), ) ) return IntercompanyMatrixOut(companies=list(company_info), cells=cells) @@ -149,7 +158,7 @@ def _account_tree( def _expand(roots: Iterable[uuid.UUID], children: dict[uuid.UUID, list[uuid.UUID]]) -> set[uuid.UUID]: - """Выбранные счета и все их потомки; пересечения веток учитываются один раз.""" + """Выбранные счета и все их потомки.""" result: set[uuid.UUID] = set() pending = list(roots) while pending: @@ -174,24 +183,14 @@ def _selected_accounts( roots = set(included) missing = roots - by_id.keys() if missing: - raise DomainValidationError("ACCOUNT_NOT_FOUND", "Выбранный счёт не найден в плане счетов") + missing_ids = ", ".join(map(str, sorted(missing))) + raise DomainValidationError("ACCOUNT_NOT_FOUND", f"Выбранного счёта нет ({missing_ids})") if any(by_id[account_id].kind not in _kinds(side) for account_id in roots): raise DomainValidationError("ACCOUNT_KIND_MISMATCH", "Сторона счёта не соответствует направлению расчёта") selected = _expand(roots, children) - _expand(excluded, children) return frozenset(account_id for account_id in selected if by_id[account_id].kind in _kinds(side)) -def _preset_accounts( - accounts: Sequence[EliminationRuleAccount], - company_no: int, -) -> tuple[set[uuid.UUID], set[uuid.UUID]]: - included = { - account.account_id for account in accounts if account.company_no == company_no and not account.is_excluded - } - excluded = {account.account_id for account in accounts if account.company_no == company_no and account.is_excluded} - return included, excluded - - def calculate_result_table( rows: Iterable[ConsolidatedRow], member_companies: Sequence[IntercompanyMatrixCompanyOut], @@ -262,7 +261,7 @@ class IntercompanyMatrixService: version_id: uuid.UUID, selection: MatrixSelection, ) -> IntercompanyMatrixOut: - """Счета принимает готовыми; выбор режима ALL/MANUAL/PRESET выполняется вызывающим кодом.""" + """Счета принимает готовыми; выбор режима ALL/MANUAL выполняется вызывающим кодом.""" consolidated, version, _ = await self._osv_service.build(version_id) return await self._calculate(consolidated.rows, version.id, selection) @@ -275,7 +274,7 @@ class IntercompanyMatrixService: return calculate_matrix(rows, companies, selection) async def build_result_table(self, version_id: uuid.UUID) -> IntercompanyResultTableOut: - """Статичная таблица по пяти статьям, независимо от режима и пресета шахматки.""" + """Статичная таблица по пяти статьям""" consolidated, version, _ = await self._osv_service.build(version_id) members = await repo.members(self._session, version.id) accounts = list((await self._session.scalars(select(ChartOfAccount))).all()) @@ -304,35 +303,9 @@ class IntercompanyMatrixService: raise DomainValidationError("ACCOUNTS_REQUIRED", "Выберите дебетовые и кредитовые счета") debit = _selected_accounts(debit_account_ids, (), "D", by_id, children) credit = _selected_accounts(credit_account_ids, (), "K", by_id, children) + print(debit) + print(credit) else: raise DomainValidationError("INVALID_MODE", "Неизвестный режим выбора счетов") - selection = MatrixSelection(debit, credit, "closing", "D") - return await self._calculate(consolidated.rows, version.id, selection) - - async def build_preset(self, version_id: uuid.UUID, rule_no: int) -> IntercompanyMatrixOut: - """Пресет выбирается по номеру из версии правил; учитываются показатель, направление и исключения.""" - consolidated, version, _ = await self._osv_service.build(version_id) - if version.elimination_rules_version is None: - raise DomainValidationError("RULES_VERSION_NOT_SET", "Для версии консолидации не задана версия правил") - rule_rows = await rules_repo.list_rules_with_accounts(self._session, version_id) - selected_rows = [(rule, account) for rule, account in rule_rows if rule.rule_no == rule_no] - if not selected_rows: - raise NotFoundError("правило элиминации не найдено") - if len({rule.id for rule, _ in selected_rows}) > 1: - raise DomainValidationError("AMBIGUOUS_RULE_NO", "Номер правила не уникален в версии правил") - rule = selected_rows[0][0] - if rule.measure not in ("closing", "turnover") or rule.side_1 not in ("D", "K"): - raise DomainValidationError("INVALID_RULE", "У правила неверный показатель или направление") - rule_accounts = [account for _, account in selected_rows if account is not None] - accounts = list((await self._session.scalars(select(ChartOfAccount))).all()) - by_id, children = _account_tree(accounts) - side_1 = cast(Side, rule.side_1) - side_2: Side = "K" if side_1 == "D" else "D" - included_1, excluded_1 = _preset_accounts(rule_accounts, 1) - included_2, excluded_2 = _preset_accounts(rule_accounts, 2) - accounts_1 = _selected_accounts(included_1, excluded_1, side_1, by_id, children) - accounts_2 = _selected_accounts(included_2, excluded_2, side_2, by_id, children) - if not accounts_1 or not accounts_2: - raise DomainValidationError("RULE_ACCOUNTS_EMPTY", "У правила нет счетов одной из сторон") - selection = MatrixSelection(accounts_1, accounts_2, cast(Measure, rule.measure), side_1) + selection = MatrixSelection(debit, credit, "closing") return await self._calculate(consolidated.rows, version.id, selection) diff --git a/api/src/services/intercompany_matrix_export.py b/api/src/services/intercompany_matrix_export.py new file mode 100644 index 0000000..be92abf --- /dev/null +++ b/api/src/services/intercompany_matrix_export.py @@ -0,0 +1,82 @@ +from io import BytesIO + +from openpyxl import Workbook +from openpyxl.cell.cell import ILLEGAL_CHARACTERS_RE +from openpyxl.styles import Alignment, Border, Font, PatternFill, Side +from openpyxl.utils import get_column_letter +from openpyxl.worksheet.worksheet import Worksheet + +from src.schemas.consolidations import IntercompanyMatrixOut +from src.services.osv_export import MONEY_FORMAT + +METRICS = ("Разница", "Дебет", "Кредит") +THRESHOLD_FILL = PatternFill("solid", fgColor="FCE8E6") +THRESHOLD_FONT = Font(color="9C0006", bold=True) +HEADER_FILL = PatternFill("solid", fgColor="E7EDF5") +DIAGONAL_FILL = PatternFill("solid", fgColor="F2F2F2") +OUTLINE_SIDE = Side(style="thin", color="A6A6A6") + + +def _text(sheet: Worksheet, row: int, column: int, value: str) -> None: + """Записать текст без XML-символов и без превращения ведущего `=` в формулу.""" + cell = sheet.cell(row, column, ILLEGAL_CHARACTERS_RE.sub("", value)) + cell.data_type = "s" + + +def build_matrix_workbook(matrix: IntercompanyMatrixOut) -> bytes: + workbook = Workbook() + sheet = workbook.active + assert sheet is not None + sheet.title = "Шахматка" + _text(sheet, 1, 1, "Дебет ↓ / Кредит →") + _text(sheet, 1, 2, "Показатель") + sheet.column_dimensions["A"].width = 32 + sheet.column_dimensions["B"].width = 18 + + for column, company in enumerate(matrix.companies, start=3): + _text(sheet, 1, column, company.name) + sheet.column_dimensions[get_column_letter(column)].width = 26 + sheet.row_dimensions[1].height = 36 + for header in sheet[1]: + header.font = Font(bold=True) + header.fill = HEADER_FILL + header.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) + + cells = {(cell.company_a_id, cell.company_b_id): cell for cell in matrix.cells} + for company_index, debit_company in enumerate(matrix.companies): + start_row = 2 + company_index * len(METRICS) + end_row = start_row + len(METRICS) - 1 + sheet.merge_cells(start_row=start_row, start_column=1, end_row=end_row, end_column=1) + _text(sheet, start_row, 1, debit_company.name) + sheet.cell(start_row, 1).alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) + for offset, metric in enumerate(METRICS): + _text(sheet, start_row + offset, 2, metric) + sheet.cell(start_row + offset, 2).alignment = Alignment(horizontal="center", vertical="center") + + for column, credit_company in enumerate(matrix.companies, start=3): + cell = cells.get((debit_company.id, credit_company.id)) + for offset in range(len(METRICS)): + sheet.cell(start_row + offset, column).border = Border( + left=OUTLINE_SIDE, + right=OUTLINE_SIDE, + top=OUTLINE_SIDE if offset == 0 else Side(), + bottom=OUTLINE_SIDE if offset == len(METRICS) - 1 else Side(), + ) + if debit_company.id == credit_company.id: + for offset in range(len(METRICS)): + sheet.cell(start_row + offset, column).fill = DIAGONAL_FILL + continue + if cell is None: + continue + for offset, amount in enumerate((cell.variance_amount, cell.amount_a, cell.amount_b)): + sheet.cell(start_row + offset, column, amount).number_format = MONEY_FORMAT + if cell.is_more_5_percent: + for offset in range(len(METRICS)): + marked = sheet.cell(start_row + offset, column) + marked.fill = THRESHOLD_FILL + marked.font = THRESHOLD_FONT + + sheet.freeze_panes = "C2" + buffer = BytesIO() + workbook.save(buffer) + return buffer.getvalue() diff --git a/api/tests/test_api_accounts.py b/api/tests/test_api_accounts.py new file mode 100644 index 0000000..4052303 --- /dev/null +++ b/api/tests/test_api_accounts.py @@ -0,0 +1,37 @@ +"""Плоский справочник счетов: все уровни и порядок кодов.""" + +import uuid +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from src.api.endpoints.directories.accounts import list_accounts +from src.models.enums import AccountKind, AccountRowType +from src.repositories.directories import list_accounts as query_accounts + + +async def test_accounts_are_flat_and_naturally_sorted() -> None: + root_id, sub_id = uuid.uuid4(), uuid.uuid4() + rows = [ + SimpleNamespace( + id=sub_id, code="60.10", parent_id=root_id, name="Авансы", kind=AccountKind.A, + row_type=AccountRowType.subaccount, is_off_balance=False, + ), + SimpleNamespace( + id=root_id, code="60", parent_id=None, name="Расчёты", kind=AccountKind.AP, + row_type=AccountRowType.account, is_off_balance=False, + ), + SimpleNamespace( + id=uuid.uuid4(), code="60.2", parent_id=root_id, name="Поставщики", kind=AccountKind.P, + row_type=AccountRowType.subaccount, is_off_balance=False, + ), + ] + session = SimpleNamespace(scalars=AsyncMock(return_value=SimpleNamespace(all=lambda: rows))) + + assert [item.code for item in await query_accounts(session)] == ["60", "60.2", "60.10"] + response = await list_accounts(session) + assert [item.code for item in response] == ["60", "60.2", "60.10"] + assert response[0].parent_id is None + assert response[2].parent_id == root_id + assert response[2].id == sub_id + assert [item.model_dump(mode="json")["is_debit"] for item in response] == [True, False, True] + assert [item.model_dump(mode="json")["is_credit"] for item in response] == [True, True, False] diff --git a/api/tests/test_intercompany_matrix.py b/api/tests/test_intercompany_matrix.py index a11c274..9141e46 100644 --- a/api/tests/test_intercompany_matrix.py +++ b/api/tests/test_intercompany_matrix.py @@ -1,15 +1,11 @@ -"""Выбор счетов и показателя в режимах шахматки без обращения к БД.""" +"""Выбор счетов и расчёт шахматки без обращения к БД.""" import uuid from decimal import Decimal from types import SimpleNamespace -from unittest.mock import AsyncMock - -import pytest from src.api.endpoints.consolidations.intercompany_matrix import router -from src.common.exceptions import DomainValidationError -from src.models.enums import AccountKind, StatementType +from src.models.enums import AccountKind from src.schemas.consolidations import IntercompanyMatrixCompanyOut, IntercompanyMatrixOut, IntercompanyResultTableOut from src.services import intercompany_matrix as matrix @@ -34,6 +30,7 @@ def _row( closing_credit=Decimal(credit), turnover_debit=Decimal("0"), turnover_credit=Decimal("0"), + entry_ids=(), ) @@ -42,10 +39,7 @@ def test_routes_are_separate_and_matrix_expands_manual_accounts() -> None: assert ("/consolidations/versions/{version_id}/intercompany-matrix/all", ("GET",)) in paths assert ("/consolidations/versions/{version_id}/intercompany-matrix/manual", ("POST",)) in paths assert ("/consolidations/versions/{version_id}/intercompany-matrix/result-table", ("GET",)) in paths - assert ( - "/consolidations/versions/{version_id}/intercompany-matrix/presets/{rule_no}", - ("GET",), - ) in paths + assert not any("/intercompany-matrix/presets/" in path for path, _ in paths) debit_parent = _account(AccountKind.A) debit_child = _account(AccountKind.A, debit_parent.id) @@ -67,7 +61,7 @@ def test_routes_are_separate_and_matrix_expands_manual_accounts() -> None: IntercompanyMatrixCompanyOut(id=company_a, name="Компания А"), IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"), ], - matrix.MatrixSelection(debit, credit, "closing", "D"), + matrix.MatrixSelection(debit, credit, "closing"), ) assert [(company.id, company.name) for company in result.companies] == [ (company_a, "Компания А"), @@ -131,55 +125,3 @@ def test_result_table_uses_fixed_articles_and_group_totals() -> None: assert isinstance(result, IntercompanyResultTableOut) response = result assert response.articles[0].amounts[0].company_id == company_a - - -@pytest.mark.asyncio -async def test_preset_uses_rule_version_measure_side_and_exclusions(monkeypatch: pytest.MonkeyPatch) -> None: - version_id, company_a, company_b = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() - debit_root = _account(AccountKind.A) - debit_excluded = _account(AccountKind.A, debit_root.id) - credit_root = _account(AccountKind.P) - rows = [ - _row(company_a, company_b, credit_root.id, credit="999"), - _row(company_b, company_a, debit_root.id, debit="999"), - _row(company_b, company_a, debit_excluded.id, debit="999"), - ] - rows[0].turnover_credit = Decimal("80") - rows[1].turnover_debit = Decimal("70") - rows[2].turnover_debit = Decimal("20") - version = SimpleNamespace(id=version_id, elimination_rules_version=2) - osv = SimpleNamespace(build=AsyncMock(return_value=(SimpleNamespace(rows=rows), version, None))) - session = SimpleNamespace( - scalars=AsyncMock(return_value=SimpleNamespace(all=lambda: [debit_root, debit_excluded, credit_root])) - ) - rule = SimpleNamespace( - id=uuid.uuid4(), statement_type=StatementType.income, rule_no=3, measure="turnover", side_1="K" - ) - rule_accounts = [ - SimpleNamespace(company_no=1, account_id=credit_root.id, is_excluded=False), - SimpleNamespace(company_no=2, account_id=debit_root.id, is_excluded=False), - SimpleNamespace(company_no=2, account_id=debit_excluded.id, is_excluded=True), - ] - list_rules = AsyncMock(return_value=[(rule, account) for account in rule_accounts]) - members = AsyncMock( - return_value=[ - (None, SimpleNamespace(id=company_a, name="Компания А")), - (None, SimpleNamespace(id=company_b, name="Компания Б")), - ] - ) - monkeypatch.setattr(matrix.rules_repo, "list_rules_with_accounts", list_rules) - monkeypatch.setattr(matrix.repo, "members", members) - - service = matrix.IntercompanyMatrixService(session, SimpleNamespace(), osv) - result = await service.build_preset(version_id, 3) - - list_rules.assert_awaited_once_with(session, version_id) - assert result.cells[0].amount_a == Decimal("80") - assert result.cells[0].amount_b == Decimal("70") - assert result.cells[0].variance_amount == Decimal("10") - assert [company.name for company in result.companies] == ["Компания А", "Компания Б"] - - duplicate = SimpleNamespace(id=uuid.uuid4(), statement_type=StatementType.balance, rule_no=3) - list_rules.return_value = [(rule, rule_accounts[0]), (duplicate, None)] - with pytest.raises(DomainValidationError, match="Номер правила не уникален"): - await service.build_preset(version_id, 3) diff --git a/api/tests/test_intercompany_matrix_export.py b/api/tests/test_intercompany_matrix_export.py new file mode 100644 index 0000000..82fe9a7 --- /dev/null +++ b/api/tests/test_intercompany_matrix_export.py @@ -0,0 +1,122 @@ +"""XLSX-выгрузка шахматки из готового результата расчёта.""" + +from io import BytesIO +from types import SimpleNamespace +from uuid import UUID + +from openpyxl import load_workbook + +from src.api.endpoints.consolidations.intercompany_matrix import ( + export_intercompany_matrix_all, + export_intercompany_matrix_manual, + router, +) +from src.schemas.consolidations import ( + IntercompanyMatrixCellOut, + IntercompanyMatrixCompanyOut, + IntercompanyMatrixManualIn, + IntercompanyMatrixOut, +) +from src.services.intercompany_matrix_export import DIAGONAL_FILL, METRICS, THRESHOLD_FILL, build_matrix_workbook + + +def test_matrix_export_has_three_rows_per_debit_company_and_highlights_threshold() -> None: + company_a = UUID("11111111-1111-1111-1111-111111111111") + company_b = UUID("22222222-2222-2222-2222-222222222222") + entry_id = UUID("33333333-3333-3333-3333-333333333333") + matrix = IntercompanyMatrixOut( + companies=[ + IntercompanyMatrixCompanyOut(id=company_a, name="=Компания А"), + IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"), + ], + cells=[ + IntercompanyMatrixCellOut( + company_a_id=company_a, + company_b_id=company_b, + amount_a="100.25", + amount_b="90.00", + variance_amount="10.25", + variance_percent="11.388888", + is_more_5_percent=True, + entry_osv_ids_debit=[entry_id], + entry_osv_ids_credit=[], + ), + IntercompanyMatrixCellOut( + company_a_id=company_b, + company_b_id=company_a, + amount_a="20", + amount_b="20", + variance_amount="0", + variance_percent="0", + is_more_5_percent=False, + entry_osv_ids_debit=[], + entry_osv_ids_credit=[], + ), + ], + ) + + workbook = load_workbook(BytesIO(build_matrix_workbook(matrix))) + sheet = workbook.active + assert sheet is not None + assert sheet.title == "Шахматка" + assert [cell.value for cell in sheet[1]] == ["Дебет ↓ / Кредит →", "Показатель", "=Компания А", "Компания Б"] + assert all(cell.alignment.horizontal == "center" and cell.alignment.vertical == "center" for cell in sheet[1]) + assert sheet["C1"].data_type == "s" + assert sheet["A2"].value == "=Компания А" + assert sheet["A2"].data_type == "s" + assert sheet["A2"].alignment.horizontal == "center" + assert {str(merged) for merged in sheet.merged_cells.ranges} == {"A2:A4", "A5:A7"} + assert [sheet.cell(row, 2).value for row in (2, 3, 4)] == list(METRICS) + assert [sheet.cell(row, 2).value for row in (5, 6, 7)] == list(METRICS) + assert all(sheet.cell(row, 2).alignment.horizontal == "center" for row in (2, 3, 4, 5, 6, 7)) + assert sheet["A5"].value == "Компания Б" + assert [sheet.cell(row, 3).value for row in (2, 3, 4)] == [None, None, None] + assert all(sheet.cell(row, 3).fill.fgColor.rgb == DIAGONAL_FILL.fgColor.rgb for row in (2, 3, 4)) + assert all(sheet.cell(row, 4).fill.fgColor.rgb == DIAGONAL_FILL.fgColor.rgb for row in (5, 6, 7)) + expected_ab = (10.25, 100.25, 90) + assert tuple(sheet.cell(row, 4).value for row in (2, 3, 4)) == expected_ab + assert all(sheet.cell(row, 4).fill.fgColor.rgb == THRESHOLD_FILL.fgColor.rgb for row in (2, 3, 4)) + assert all(sheet.cell(row, column).comment is None for row in range(2, 8) for column in (3, 4)) + assert [sheet.cell(row, 3).value for row in (5, 6, 7)] == [0, 20, 20] + assert all(sheet.cell(row, 3).fill.fgColor.rgb != THRESHOLD_FILL.fgColor.rgb for row in (5, 6, 7)) + for column, first_row in ((3, 2), (4, 2), (3, 5), (4, 5)): + top, middle, bottom = (sheet.cell(first_row + offset, column) for offset in range(3)) + assert top.border.top.style == "thin" + assert top.border.bottom.style is None + assert middle.border.top.style is None + assert middle.border.bottom.style is None + assert bottom.border.bottom.style == "thin" + assert all( + cell.border.left.style == "thin" and cell.border.right.style == "thin" + for cell in (top, middle, bottom) + ) + + +async def test_both_export_routes_use_matrix_calculation() -> None: + version_id = UUID("11111111-1111-1111-1111-111111111111") + account_id = UUID("22222222-2222-2222-2222-222222222222") + matrix = IntercompanyMatrixOut(companies=[], cells=[]) + calls = [] + + async def build_accounts(*args: object, **kwargs: object) -> IntercompanyMatrixOut: + calls.append((args, kwargs)) + return matrix + + service = SimpleNamespace(build_accounts=build_accounts) + selection = IntercompanyMatrixManualIn(debit_account_ids=[account_id], credit_account_ids=[account_id]) + + all_response = await export_intercompany_matrix_all(version_id, service) + manual_response = await export_intercompany_matrix_manual(version_id, selection, service) + + assert calls == [ + ((version_id,), {"mode": "ALL"}), + ((version_id,), {"mode": "MANUAL", "debit_account_ids": [account_id], "credit_account_ids": [account_id]}), + ] + for response in (all_response, manual_response): + assert response.media_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + assert "attachment" in response.headers["content-disposition"] + assert load_workbook(BytesIO(response.body)).active is not None + + paths = {(route.path, tuple(sorted(route.methods))) for route in router.routes} + assert ("/consolidations/versions/{version_id}/intercompany-matrix/all/export", ("GET",)) in paths + assert ("/consolidations/versions/{version_id}/intercompany-matrix/manual/export", ("POST",)) in paths diff --git a/api/tests/test_openapi_routes.py b/api/tests/test_openapi_routes.py index 8bc79b4..cdcf63a 100644 --- a/api/tests/test_openapi_routes.py +++ b/api/tests/test_openapi_routes.py @@ -20,6 +20,7 @@ EXPECTED_ROUTES: dict[str, set[str]] = { "/api/v1/me": {"get"}, # directories-api: филиалы, ГК, ГСК/ГСЗ, ЮЛ/ИП f"{D}/regional-offices": {"get"}, + f"{D}/accounts": {"get"}, f"{D}/company-groups": {"get", "post"}, f"{D}/company-groups/{{group_id}}": {"get", "patch"}, f"{D}/company-groups/{{group_id}}/archive": {"post"}, diff --git a/openspec/specs/directories-api/spec.md b/openspec/specs/directories-api/spec.md index 844f8fc..4880fb0 100644 --- a/openspec/specs/directories-api/spec.md +++ b/openspec/specs/directories-api/spec.md @@ -16,6 +16,17 @@ API справочников ядра: группы компаний (ГК), ю - **WHEN** пользователь с `DIRECTORY_VIEW` запрашивает список после `seed --dev` - **THEN** ответ 200 содержит Головной офис с `is_head_office = true` и два dev-филиала +### Requirement: Плоский список счетов +`GET /api/v1/directories/accounts` SHALL возвращать пользователю с правом `DIRECTORY_VIEW` все строки +`chart_of_account` одним массивом в естественном порядке кодов. Каждая строка содержит `id`, `code`, +`parent_id`, `name`, `kind`, `row_type`, `is_off_balance`, `is_debit`, `is_credit`; корневые счета имеют `parent_id = null`. +`is_debit = true` для видов `A` и `AP`, `false` для `P` и незаданного вида. +`is_credit = true` для видов `P` и `AP`, `false` для `A` и незаданного вида. + +#### Scenario: Счета разных уровней +- **WHEN** в плане счетов есть счёт, субсчёт и аналитика +- **THEN** ответ содержит все три строки без вложенности, а `parent_id` указывает на родительскую строку + ### Requirement: Группа компаний — создание, чтение, изменение, архив, история `POST /api/v1/directories/company-groups` (право `DIRECTORY_EDIT`) SHALL создавать ГК с полями `name` (обязательно), `crm_id` (необязательно, уникально), `base_regional_office_id` (необязательно), генерируя `code` вида `GK-NNNNNN` и