70 lines
2.9 KiB
Python
70 lines
2.9 KiB
Python
"""Чтение правил элиминации для версии консолидации."""
|
||
|
||
import uuid
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.common.clock import moscow_date, utc_now
|
||
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, EliminationRulesOut, 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) -> EliminationRulesOut:
|
||
found = await consolidations_repo.get_version(self._session, self._scope, consolidation_version_id)
|
||
if found is None:
|
||
raise NotFoundError(VERSION_NOT_FOUND)
|
||
version, _ = found
|
||
is_version_fixed = version.elimination_rules_version is not None
|
||
if is_version_fixed:
|
||
as_of = None
|
||
version_no = version.elimination_rules_version
|
||
else:
|
||
as_of = moscow_date(utc_now())
|
||
version_no = await rules_repo.current_version_no(self._session, as_of)
|
||
rows = (
|
||
await rules_repo.list_rules_with_accounts(self._session, version_no, as_of=as_of)
|
||
if version_no is not None
|
||
else []
|
||
)
|
||
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(
|
||
{
|
||
"id": rule.id,
|
||
"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 EliminationRulesOut(
|
||
version_no=version_no,
|
||
is_version_fixed=is_version_fixed,
|
||
as_of=as_of,
|
||
rules=list(by_id.values()),
|
||
)
|