корректировка расчета шахматки, добавление эндпоинтов для получения статей, добавления эндпоинта для выгрузки xlsx
This commit is contained in:
PotapovaA
parent
d44a4ff007
commit
db19b3a176
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
21
api/src/api/endpoints/directories/accounts.py
Normal file
21
api/src/api/endpoints/directories/accounts.py
Normal file
@ -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)]
|
||||
@ -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}_")
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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` назначает сервис."""
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
82
api/src/services/intercompany_matrix_export.py
Normal file
82
api/src/services/intercompany_matrix_export.py
Normal file
@ -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()
|
||||
37
api/tests/test_api_accounts.py
Normal file
37
api/tests/test_api_accounts.py
Normal file
@ -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]
|
||||
@ -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)
|
||||
|
||||
122
api/tests/test_intercompany_matrix_export.py
Normal file
122
api/tests/test_intercompany_matrix_export.py
Normal file
@ -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
|
||||
@ -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"},
|
||||
|
||||
@ -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` и
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user