b4522addc0
- consolidation_version.elimination_rules_version NOT NULL: миграция 0010 проставляет существующим версиям номер правил, действовавших на дату создания (по Москве), иначе последних открытых; без правил — отказ; создание версии без действующих правил — 409 ELIMINATION_RULES_MISSING; ветки «правила не зафиксированы» убраны (rules, preset, is_version_fixed всегда true); - GET …/elimination-totals: все тройки «правило ББ × упорядоченная пара периметра» с ΣA, ΣB, расхождением, суммой элиминации, статусом AUTO/DISCREPANCY/ONE_SIDED/MANUAL и osv_entry_ids; сводная ОСВ строится один раз; сбор строк сторон вынесен в _collect_rule_sides — общий с упаковкой тройки. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
60 lines
2.8 KiB
Python
60 lines
2.8 KiB
Python
"""Чтение правил элиминации для версии консолидации."""
|
||
|
||
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, 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:
|
||
"""Правила §4 версии консолидации с проверкой видимости и основанием выбора."""
|
||
found = await consolidations_repo.get_version(self._session, self._scope, consolidation_version_id)
|
||
if found is None:
|
||
raise NotFoundError(VERSION_NOT_FOUND)
|
||
version, _ = found
|
||
# с v5.23 версия правил фиксируется при создании версии консолидации и не бывает пустой
|
||
version_no = version.elimination_rules_version
|
||
rows = await rules_repo.list_rules_with_accounts(self._session, version_no)
|
||
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=True,
|
||
rules=list(by_id.values()),
|
||
)
|