RSHB/arm-core
9
0
Fork 0
Code Issues 25 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

AURORA-1651. исправление one_sided, подсветки экспорта, исправление форматирования

Browse Source
This commit is contained in:
PotapovaA 2026-10-21 17:40:04 +03:00
parent 2c0a2ad939
commit 615aa65bab
12 changed files with 171 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
api/src/api/endpoints/directories/accounts.py
View File

@ -1,11 +1,13 @@
"""Плоский список счетов плана счетов для выбора статьи расходов."""
from fastapi import APIRouter, Depends
from typing import Annotated
from fastapi import APIRouter, Depends, Query
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
from src.schemas.directories import AccountOut, FiltersForChartOfAccount
router = APIRouter()
@ -16,6 +18,6 @@ router = APIRouter()
dependencies=[Depends(require("DIRECTORY_VIEW"))],
summary="Список счетов плана счетов",
)
async def list_accounts(session: SessionDep) -> list[AccountOut]:
async def list_accounts(session: SessionDep, filters: Annotated[FiltersForChartOfAccount, Query()]) -> list[AccountOut]:
"""Все счета одним списком"""
return [AccountOut.model_validate(account) for account in await repo.list_accounts(session)]
return [AccountOut.model_validate(account) for account in await repo.list_accounts(session, filters)]

13
api/src/repositories/directories.py
View File

@ -29,13 +29,22 @@ from src.models.directories import (
)
from src.models.enums import FileStatus, RelatedGroupKind
from src.models.files import Document, StoredFile
from src.schemas.directories import FiltersForChartOfAccount
LIKE_ESCAPE = "\\"
ALIAS_LOCK_PREFIX = "legal-entity-alias" # пространство ключей advisory-блокировки алиасов
async def list_accounts(session: AsyncSession) -> list[ChartOfAccount]:
accounts = (await session.scalars(select(ChartOfAccount))).all()
async def list_accounts(session: AsyncSession, filters: FiltersForChartOfAccount | None = None) -> list[ChartOfAccount]:
stmt = select(ChartOfAccount)
if filters is not None:
if filters.is_off_balance is not None:
stmt = stmt.where(ChartOfAccount.is_off_balance.is_(filters.is_off_balance))
if filters.kinds is not None:
stmt = stmt.where(ChartOfAccount.kind.in_(filters.kinds))
if filters.row_type is not None:
stmt = stmt.where(ChartOfAccount.row_type == filters.row_type)
accounts = (await session.scalars(stmt)).all()
return sorted(accounts, key=lambda account: (account_sort_key(account.code), account.code))

4
api/src/repositories/elimination_rules.py
View File

@ -14,9 +14,7 @@ async def list_rules_with_accounts(
) -> list[tuple[EliminationRule, EliminationRuleAccount | None]]:
"""Правила заданной версии либо все правила без `valid_to`."""
condition = (
EliminationRule.version_no == version_no
if version_no is not None
else EliminationRule.valid_to.is_(None)
EliminationRule.version_no == version_no if version_no is not None else EliminationRule.valid_to.is_(None)
)
stmt = (
select(EliminationRule, EliminationRuleAccount)

6
api/src/schemas/directories.py
View File

@ -449,3 +449,9 @@ class RelatedGroupDetail(RelatedGroupOut):
"""Карточка ГСК/ГСЗ со списком текущих членов (по имени ЮЛ)."""
members: list[RelatedGroupMemberOut] = []
class FiltersForChartOfAccount(BaseModel):
is_off_balance: bool | None = None
kinds: list[AccountKind] | None = None
row_type: AccountRowType | None = None

8
api/src/services/consolidated_osv.py
View File

@ -151,9 +151,7 @@ class ConsolidatedOsvService:
"""Свод версии и данные его расчёта; версия вне видимости — 404."""
return await self.build_with_members(version_id)
async def build_with_members(
self, version_id: uuid.UUID
) -> ConsolidatedOsvBuildResult:
async def build_with_members(self, version_id: uuid.UUID) -> ConsolidatedOsvBuildResult:
"""Свод и участники периметра, загруженные при его расчёте."""
found = await repo.get_version(self._session, self._scope, version_id)
if found is None:
@ -224,9 +222,7 @@ class ConsolidatedOsvService:
"rows": len(rows),
},
)
filename = (
f"svod-osv_{group_name}_{result.consolidation.year}-{result.consolidation.quarter.value}_v{result.version.version_no}.xlsx"
)
filename = f"svod-osv_{group_name}_{result.consolidation.year}-{result.consolidation.quarter.value}_v{result.version.version_no}.xlsx"
return content, filename
async def summary(self, version_id: uuid.UUID) -> OsvSummaryOut:

5
api/src/services/intercompany_matrix.py
View File

@ -29,6 +29,7 @@ from src.schemas.consolidations import (
IntercompanyResultTableOut,
UnpairedOsvOut,
)
from src.schemas.directories import FiltersForChartOfAccount
from src.services.consolidated_osv import ConsolidatedOsvService, row_out
Measure = Literal["closing", "turnover"]
@ -180,7 +181,7 @@ def calculate_matrix(
variance_amount=variance_amount,
variance_percent=variance_percent,
exceeds_threshold=exceeds_threshold,
one_sided=bool(debit_ids) != bool(credit_ids),
one_sided=min(amount_a, amount_b) == 0 and variance_amount > 0,
skipped_osv_entry_count=skipped_entries[(company_a, company_b)]
+ skipped_entries[(company_b, company_a)],
osv_entry_ids_debit=debit_ids,
@ -512,7 +513,7 @@ class IntercompanyMatrixService:
) -> MatrixBuildResult:
osv_result = await self._osv_service.build_with_members(version_id)
rows = osv_result.consolidated.rows
accounts = await directory_repo.list_accounts(self._session)
accounts = await directory_repo.list_accounts(self._session, FiltersForChartOfAccount(is_off_balance=False))
by_id, children = _make_account_tree(accounts)
rule: EliminationRule | None = None
if mode == "ALL":

21
api/src/services/intercompany_matrix_export.py
View File

@ -60,6 +60,21 @@ def _write_values(sheet: Worksheet, row: int, column: int, cell: IntercompanyMat
_skipped_comment(sheet, row, column, cell.skipped_osv_entry_count)
def _highlight_values(sheet: Worksheet, row: int, column: int, cell: IntercompanyMatrixCellOut) -> None:
def mark(offset: int) -> None:
marked = sheet.cell(row + offset, column)
marked.fill = THRESHOLD_FILL
marked.font = THRESHOLD_FONT
if cell.exceeds_threshold:
mark(3)
if cell.one_sided:
if cell.amount_a == 0 and cell.amount_b != 0:
mark(1)
elif cell.amount_b == 0 and cell.amount_a != 0:
mark(2)
def build_matrix_workbook(matrix: IntercompanyMatrixOut, info: MatrixExportInfo) -> bytes:
workbook = Workbook()
sheet = workbook.active
@ -107,11 +122,7 @@ def build_matrix_workbook(matrix: IntercompanyMatrixOut, info: MatrixExportInfo)
if cell is None:
continue
_write_values(sheet, start_row, column, cell)
if cell.exceeds_threshold:
for offset in range(len(METRICS)):
marked = sheet.cell(start_row + offset, column)
marked.fill = THRESHOLD_FILL
marked.font = THRESHOLD_FONT
_highlight_values(sheet, start_row, column, cell)
sheet.freeze_panes = f"C{DATA_START_ROW}"
buffer = BytesIO()

3
api/tests/test_api_accounts.py
View File

@ -7,6 +7,7 @@ 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
from src.schemas.directories import FiltersForChartOfAccount
async def test_accounts_are_flat_and_naturally_sorted() -> None:
@ -43,7 +44,7 @@ async def test_accounts_are_flat_and_naturally_sorted() -> None:
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)
response = await list_accounts(session, FiltersForChartOfAccount())
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

42
api/tests/test_api_intercompany_matrix.py
View File

@ -196,12 +196,16 @@ async def test_preset_unpaired_osv_returns_rows_and_total(auth_client: AuthClien
url = f"/consolidations/versions/{world.version.id}/intercompany-matrix/preset/{rule.id}/unpaired-osv"
response = await client.get(url)
assert response.status_code == OK, response.text
assert response.json()["total"] == 1
assert response.json()["total"] == 0
assert set(response.json()) == {"items", "total"}
assert response.json()["items"][0]["entry_ids"] == [str(debit_entry.id)]
assert response.json()["items"][0]["closing_debit"] == "0.00"
await _source(seeded, world, credit=True, amount="100", account="60.01")
unpaired = await client.get(url)
assert unpaired.status_code == OK, unpaired.text
assert unpaired.json()["total"] == 2 # noqa: PLR2004 - обе строки участвуют в односторонней сумме
assert any(item["entry_ids"] == [str(debit_entry.id)] for item in unpaired.json()["items"])
await _source(seeded, world, credit=False, amount="100", account="62.01")
paired = await client.get(url)
assert paired.status_code == OK, paired.text
assert paired.json()["items"] == []
@ -265,6 +269,20 @@ async def test_unmapped_account_source_is_visible_in_cell_count(
assert len(cell["osv_entry_ids_credit"]) == 1
async def test_all_excludes_off_balance_accounts(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
world = await _world(seeded)
assert world.accounts["009"].is_off_balance is True
await _source(seeded, world, credit=False, amount="500", account="009")
await _source(seeded, world, credit=False, amount="100", account="62.01")
await _source(seeded, world, credit=True, amount="90", account="60.01")
response = await _request(await auth_client(KA_RF), world.version.id, "all", world.selection)
assert response.status_code == OK, response.text
cell = next(c for c in response.json()["cells"] if c["company_a_id"] == str(world.members[0].id))
assert cell["amount_a"] == "100.00"
assert cell["amount_b"] == "90.00"
@pytest.mark.parametrize("credit", [False, True])
@pytest.mark.parametrize("amount", ["0", "100.25"])
async def test_pair_without_counterpart(
@ -277,13 +295,29 @@ async def test_pair_without_counterpart(
cell = next(c for c in response.json()["cells"] if c["company_a_id"] == str(world.members[0].id))
assert Decimal(cell["amount_a"]) == (Decimal(0) if credit else Decimal(amount))
assert Decimal(cell["amount_b"]) == (Decimal(amount) if credit else Decimal(0))
assert cell["one_sided"] is True
assert cell["one_sided"] is (Decimal(amount) != 0)
assert cell["variance_percent"] is None
assert cell["exceeds_threshold"] is False
assert bool(cell["osv_entry_ids_credit"]) is credit
assert bool(cell["osv_entry_ids_debit"]) is not credit
async def test_opposite_zero_balance_is_one_sided(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
world = await _world(seeded)
await _source(seeded, world, credit=False, amount="100000000")
await _source(seeded, world, credit=True, amount="0")
response = await _request(await auth_client(KA_RF), world.version.id, "manual", world.selection)
assert response.status_code == OK, response.text
cell = next(c for c in response.json()["cells"] if c["company_a_id"] == str(world.members[0].id))
assert cell["one_sided"] is True
assert cell["exceeds_threshold"] is False
assert cell["variance_percent"] is None
assert cell["amount_a"] == "100000000.00"
assert cell["amount_b"] == "0.00"
assert cell["osv_entry_ids_debit"] and cell["osv_entry_ids_credit"]
@pytest.mark.parametrize("suffix", ROUTES)
async def test_visibility_and_permission(auth_client: AuthClientFactory, seeded: AsyncSession, suffix: str) -> None:
world = await _world(seeded)

4
api/tests/test_elimination_rules.py
View File

@ -33,9 +33,7 @@ def _rule() -> SimpleNamespace:
(SimpleNamespace(version_no=3, valid_to="2026-12-31"), 3, True),
],
)
def test_rule_is_valid_for_version(
rule: SimpleNamespace | None, version_no: int | None, expected: bool
) -> None:
def test_rule_is_valid_for_version(rule: SimpleNamespace | None, version_no: int | None, expected: bool) -> None:
assert rules_repo.is_valid_for_version(rule, version_no) is expected

46
api/tests/test_intercompany_matrix.py
View File

@ -92,7 +92,7 @@ def test_routes_are_separate_and_matrix_expands_manual_accounts() -> None:
]
def test_unpaired_rows_uses_source_presence_and_preserves_osv_order() -> None:
def test_unpaired_rows_ignore_zero_only_sources_when_amounts_match() -> None:
company_a, company_b = uuid.uuid4(), uuid.uuid4()
debit_account, credit_account = uuid.uuid4(), uuid.uuid4()
paired_debit = _row(company_a, company_b, debit_account, debit="100")
@ -109,7 +109,28 @@ def test_unpaired_rows_uses_source_presence_and_preserves_osv_order() -> None:
rows, companies, matrix.MatrixSelection(frozenset({debit_account}), frozenset({credit_account}), "closing")
)
assert matrix.get_unpaired_rows(rows, result) == [unpaired_credit]
assert matrix.get_unpaired_rows(rows, result) == []
def test_unpaired_rows_include_both_sources_when_opposite_balance_is_zero() -> None:
company_a, company_b = uuid.uuid4(), uuid.uuid4()
debit_account, credit_account = uuid.uuid4(), uuid.uuid4()
debit = _row(company_a, company_b, debit_account, debit="100")
zero_credit = _row(company_b, company_a, credit_account, credit="0")
debit.entry_ids = (uuid.uuid4(),)
zero_credit.entry_ids = (uuid.uuid4(),)
rows = [zero_credit, debit]
companies = [
IntercompanyMatrixCompanyOut(id=company_a, name="А"),
IntercompanyMatrixCompanyOut(id=company_b, name="Б"),
]
result = matrix.calculate_matrix(
rows, companies, matrix.MatrixSelection(frozenset({debit_account}), frozenset({credit_account}), "closing")
)
assert result.cells[0].one_sided is True
assert result.cells[0].exceeds_threshold is False
assert matrix.get_unpaired_rows(rows, result) == rows
async def test_preset_applies_exclusions_to_its_own_side_and_rule_measure(monkeypatch: pytest.MonkeyPatch) -> None:
@ -237,7 +258,7 @@ async def test_preset_without_corrections_keeps_matrix(monkeypatch: pytest.Monke
assert adjusted is original
def test_one_sided_reflects_osv_sources_even_when_amount_is_zero() -> None:
def test_one_sided_reflects_amounts_even_when_both_sources_exist() -> None:
company_a, company_b = uuid.uuid4(), uuid.uuid4()
debit_account, credit_account = uuid.uuid4(), uuid.uuid4()
companies = [
@ -245,15 +266,30 @@ def test_one_sided_reflects_osv_sources_even_when_amount_is_zero() -> None:
IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"),
]
selection = matrix.MatrixSelection(frozenset({debit_account}), frozenset({credit_account}), "closing")
debit = _row(company_a, company_b, debit_account)
debit = _row(company_a, company_b, debit_account, debit="100000000")
credit = _row(company_b, company_a, credit_account)
debit.entry_ids = (uuid.uuid4(),)
credit.entry_ids = (uuid.uuid4(),)
for rows, expected in (([], False), ([debit], True), ([credit], True), ([debit, credit], False)):
for rows, expected in (([], False), ([debit], True), ([credit], False), ([debit, credit], True)):
cell = matrix.calculate_matrix(rows, companies, selection).cells[0]
assert cell.one_sided is expected
assert cell.model_dump(mode="json")["one_sided"] is expected
assert cell.variance_percent is None
assert cell.exceeds_threshold is False
assert bool(cell.osv_entry_ids_credit) is (credit in rows)
debit.closing_debit = Decimal("0")
both_zero = matrix.calculate_matrix([debit, credit], companies, selection).cells[0]
assert both_zero.one_sided is False
assert both_zero.variance_amount == Decimal("0")
assert both_zero.osv_entry_ids_debit and both_zero.osv_entry_ids_credit
debit.closing_debit = Decimal("100000000")
credit.closing_credit = Decimal("100000000")
matched = matrix.calculate_matrix([debit, credit], companies, selection).cells[0]
assert matched.one_sided is False
assert matched.variance_amount == Decimal("0")
def test_unmapped_account_entries_are_counted_for_both_directions_of_pair() -> None:

46
api/tests/test_intercompany_matrix_export.py
View File

@ -58,7 +58,7 @@ def test_matrix_export_has_three_rows_per_debit_company_and_highlights_threshold
variance_amount="10.25",
variance_percent="11.388888",
exceeds_threshold=True,
one_sided=True,
one_sided=False,
skipped_osv_entry_count=2,
osv_entry_ids_debit=[entry_id],
osv_entry_ids_credit=[],
@ -109,7 +109,9 @@ def test_matrix_export_has_three_rows_per_debit_company_and_highlights_threshold
assert all(sheet.cell(row, 4).fill.fgColor.rgb == DIAGONAL_FILL.fgColor.rgb for row in range(15, 19))
assert tuple(sheet.cell(row, 4).value for row in range(11, 15)) == (10.25, 100.25, 90, 11.39)
assert sheet["D14"].number_format == '0.00"%"'
assert all(sheet.cell(row, 4).fill.fgColor.rgb == THRESHOLD_FILL.fgColor.rgb for row in range(11, 15))
assert sheet["D14"].fill.fgColor.rgb == THRESHOLD_FILL.fgColor.rgb
assert sheet["D14"].font.bold
assert all(sheet.cell(row, 4).fill.patternType is None for row in range(11, 14))
assert sheet["D11"].comment is not None
assert "2" in sheet["D11"].comment.text
assert all(
@ -144,6 +146,45 @@ def test_matrix_export_has_three_rows_per_debit_company_and_highlights_threshold
assert preset_sheet["D13"].comment is None
@pytest.mark.parametrize(
("amount_a", "amount_b", "zero_row", "other_row"),
[("0", "25", 12, 13), ("25", "0", 13, 12)],
)
def test_matrix_export_highlights_only_zero_side_when_one_sided(
amount_a: str, amount_b: str, zero_row: int, other_row: int
) -> None:
company_a = UUID("11111111-1111-1111-1111-111111111111")
company_b = UUID("22222222-2222-2222-2222-222222222222")
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=amount_a,
amount_b=amount_b,
variance_amount="25",
variance_percent=None,
exceeds_threshold=False,
one_sided=True,
skipped_osv_entry_count=0,
osv_entry_ids_debit=[],
osv_entry_ids_credit=[],
)
],
)
sheet = load_workbook(BytesIO(build_matrix_workbook(matrix, _info()))).active
assert sheet is not None
assert sheet.cell(zero_row, 4).value == 0
assert sheet.cell(zero_row, 4).fill.fgColor.rgb == THRESHOLD_FILL.fgColor.rgb
assert sheet.cell(zero_row, 4).font.bold
assert all(sheet.cell(row, 4).fill.patternType is None for row in (11, other_row, 14))
async def test_export_routes_use_matrix_calculation() -> None:
version_id = UUID("11111111-1111-1111-1111-111111111111")
account_id = UUID("22222222-2222-2222-2222-222222222222")
@ -197,6 +238,7 @@ async def test_export_routes_use_matrix_calculation() -> None:
assert unquote(disposition.split("filename*=UTF-8''", 1)[1]) == expected_info.filename
book = load_workbook(BytesIO(response.body))
assert book.active["B6"].value in {"Все счета", "Выбранные счета", "Пресет"}
assert all(book.active.cell(row, 4).fill.patternType is None for row in range(11, 15))
preset_sheet = load_workbook(BytesIO(preset_response.body)).active
assert preset_sheet["B7"].value == "62.01"
assert preset_sheet["B8"].value == "60.01"