добавление ручки для получения правил шахматки

api/src/api/dependencies/consolidations.py

@@ -14,6 +14,7 @@ from src.services.consolidation_locks import ConsolidationLockService
 from src.services.consolidation_status import ConsolidationStatusService
 from src.services.consolidations import ConsolidationService
 from src.services.counterparty_mapping import CounterpartyMappingService
+from src.services.elimination_rules import EliminationRuleService
 
 
 def get_consolidation_service(
@@ -73,3 +74,13 @@ def get_counterparty_mapping_service(
 
 
 CounterpartyMappingServiceDep = Annotated[CounterpartyMappingService, Depends(get_counterparty_mapping_service)]
+
+
+def get_elimination_rule_service(
+    session: SessionDep, scope: Annotated[VisibilityScope, Depends(get_scope)]
+) -> EliminationRuleService:
+    """Сервис чтения правил элиминации из БД для версии консолидации."""
+    return EliminationRuleService(session, scope)
+
+
+EliminationRuleServiceDep = Annotated[EliminationRuleService, Depends(get_elimination_rule_service)]

api/src/api/endpoints/consolidations/__init__.py

@@ -2,7 +2,18 @@
 
 from fastapi import APIRouter
 
-from src.api.endpoints.consolidations import checklist, export, locks, mapping, members, osv, periods, status, versions
+from src.api.endpoints.consolidations import (
+    checklist,
+    elimination_rule,
+    export,
+    locks,
+    mapping,
+    members,
+    osv,
+    periods,
+    status,
+    versions,
+)
 
 router = APIRouter(tags=["consolidations"])
 router.include_router(periods.router)
@@ -14,3 +25,4 @@ router.include_router(checklist.router)
 router.include_router(osv.router)
 router.include_router(export.router)
 router.include_router(mapping.router)
+router.include_router(elimination_rule.router)

api/src/api/endpoints/consolidations/elimination_rule.py

@@ -0,0 +1,33 @@
+"""Правила элиминации для версии консолидации."""
+
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, Depends, status
+
+from src.api.dependencies.auth import require
+from src.api.dependencies.consolidations import EliminationRuleServiceDep
+from src.schemas.common import ErrorDetail
+from src.schemas.consolidations import EliminationRuleOut
+
+router = APIRouter(prefix="/consolidations/versions")
+
+VIEW = [Depends(require("CONSOLIDATION_VIEW"))]
+NOT_FOUND: dict[int | str, dict[str, Any]] = {
+    status.HTTP_404_NOT_FOUND: {"model": ErrorDetail, "description": "нет или вне видимости"}
+}
+
+
+@router.get(
+    "/{version_id}/elimination-rules",
+    response_model=list[EliminationRuleOut],
+    dependencies=VIEW,
+    responses=NOT_FOUND,
+    summary="Правила для шахматки",
+)
+async def list_rules(
+    version_id: uuid.UUID,
+    service: EliminationRuleServiceDep,
+) -> list[EliminationRuleOut]:
+    """Все правила выбранной версии со счетами и исключениями обеих сторон."""
+    return await service.list_rules(version_id)

api/src/repositories/elimination_rules.py

@@ -0,0 +1,53 @@
+"""Чтение версионированных правил элиминации и счетов их сторон."""
+
+import uuid
+from collections.abc import Collection
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.models.accounting import ChartOfAccount, EliminationRule, EliminationRuleAccount
+from src.models.consolidation import ConsolidationVersion
+
+
+async def list_rules_with_accounts(
+    session: AsyncSession, consolidation_version_id: uuid.UUID
+) -> list[tuple[EliminationRule, EliminationRuleAccount | None]]:
+    """Правила версии консолидации по `elimination_rules_version` и счета их сторон."""
+    stmt = (
+        select(EliminationRule, EliminationRuleAccount)
+        .join(ConsolidationVersion, ConsolidationVersion.elimination_rules_version == EliminationRule.version_no)
+        .outerjoin(EliminationRuleAccount, EliminationRuleAccount.rule_id == EliminationRule.id)
+        .where(ConsolidationVersion.id == consolidation_version_id)
+        .order_by(
+            EliminationRule.statement_type,
+            EliminationRule.rule_no,
+            EliminationRuleAccount.company_no,
+        )
+    )
+    return [(rule, account) for rule, account in (await session.execute(stmt)).all()]
+
+
+async def list_rules(session: AsyncSession, version_no: int) -> list[EliminationRule]:
+    """Все правила заданной версии, включая ББ и ОФР, в порядке типа отчётности и номера."""
+    stmt = (
+        select(EliminationRule)
+        .where(EliminationRule.version_no == version_no)
+        .order_by(EliminationRule.statement_type, EliminationRule.rule_no)
+    )
+    return list((await session.scalars(stmt)).all())
+
+
+async def list_rule_accounts(
+    session: AsyncSession, rule_ids: Collection[uuid.UUID]
+) -> list[tuple[EliminationRuleAccount, str]]:
+    """Счета правил с кодами из плана счетов; `company_no` задаёт сторону, `is_excluded` — исключение."""
+    if not rule_ids:
+        return []
+    stmt = (
+        select(EliminationRuleAccount, ChartOfAccount.code)
+        .join(ChartOfAccount, ChartOfAccount.id == EliminationRuleAccount.account_id)
+        .where(EliminationRuleAccount.rule_id.in_(rule_ids))
+        .order_by(EliminationRuleAccount.rule_id, EliminationRuleAccount.company_no, ChartOfAccount.code)
+    )
+    return [(account, code) for account, code in (await session.execute(stmt)).all()]

api/src/schemas/consolidations.py

@@ -9,6 +9,7 @@
 import datetime
 import uuid
 from decimal import Decimal
+from typing import Literal
 
 from pydantic import BaseModel, Field, field_validator
 
@@ -22,6 +23,7 @@ from src.models.enums import (
     DocumentType,
     EntityType,
     PeriodQuarter,
+    StatementType,
 )
 
 YEAR_MIN = 2000
@@ -304,3 +306,22 @@ class MapCounterpartyOut(BaseModel):
     propagated: int
     alias_created: bool
     alias_moved_from: uuid.UUID | None
+
+
+class IntercompanyMatrixRuleSideOut(BaseModel):
+    """Счета одной стороны правила intercompany-matrix (шахматки) до раскрытия иерархии."""
+
+    accounts: list[uuid.UUID]
+    excluded: list[uuid.UUID]
+
+
+class EliminationRuleOut(BaseModel):
+    """Правило элиминации из БД; ключ — тип отчётности, версия и номер."""
+
+    rule_no: int
+    version_no: int
+    statement_type: StatementType
+    side_1: Literal["D", "K"]
+    measure: Literal["closing", "turnover"]
+    company_1: IntercompanyMatrixRuleSideOut
+    company_2: IntercompanyMatrixRuleSideOut

api/src/services/elimination_rules.py

@@ -0,0 +1,51 @@
+"""Чтение правил элиминации для версии консолидации."""
+
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.common.exceptions import NotFoundError
+from src.domain.policy import VisibilityScope
+from src.repositories import consolidations as consolidations_repo
+from src.repositories import elimination_rules as rules_repo
+from src.schemas.consolidations import EliminationRuleOut, IntercompanyMatrixRuleSideOut
+
+VERSION_NOT_FOUND = "версия консолидации не найдена"
+
+
+class EliminationRuleService:
+    """Правила выбранной версии из БД с учетом видимости консолидации."""
+
+    def __init__(self, session: AsyncSession, scope: VisibilityScope) -> None:
+        self._session = session
+        self._scope = scope
+
+    async def list_rules(self, consolidation_version_id: uuid.UUID) -> list[EliminationRuleOut]:
+        found = await consolidations_repo.get_version(self._session, self._scope, consolidation_version_id)
+        if found is None:
+            raise NotFoundError(VERSION_NOT_FOUND)
+
+        rows = await rules_repo.list_rules_with_accounts(self._session, consolidation_version_id)
+        by_id: dict[uuid.UUID, EliminationRuleOut] = {}
+        for rule, account in rows:
+            result = by_id.get(rule.id)
+            if result is None:
+                result = EliminationRuleOut.model_validate(
+                    {
+                        "rule_no": rule.rule_no,
+                        "version_no": rule.version_no,
+                        "statement_type": rule.statement_type,
+                        "side_1": rule.side_1,
+                        "measure": rule.measure,
+                        "company_1": IntercompanyMatrixRuleSideOut(accounts=[], excluded=[]),
+                        "company_2": IntercompanyMatrixRuleSideOut(accounts=[], excluded=[]),
+                    }
+                )
+                by_id[rule.id] = result
+
+            if account is None:
+                continue
+            side = result.company_1 if account.company_no == 1 else result.company_2
+            (side.excluded if account.is_excluded else side.accounts).append(account.account_id)
+
+        return list(by_id.values())