arm-core/api/src/services/consolidations.py
Andrey Epifancev 1ae6b8d2da feat(consolidation): временный режим «период ОСВ из версии консолидации» (#103)
Настройка CONSOLIDATION_PERIOD_FROM_VERSION (по умолчанию false; в контуре —
OPENBAO__APP__CONSOLIDATION_PERIOD_FROM_VERSION): ОСВ, загруженная в консолидацию
(stored_file.consolidation_id), считается нарастающей за год и квартал консолидации,
какой бы период ни стоял в шапке. Нужен для тестов на произвольных ОСВ: без него ОСВ
за 1 кв. 2025 в консолидации 2026 Q1 молча не привязывается и свод пуст.

- domain/version_period.py: представление документа с периодом версии (первичка не меняется)
- consolidation_documents.for_consolidation: единая точка подмены; её используют привязка
  (fits, ключ R-1, bind_confirmed, хуки подтверждения и маппинга), свод, чек-лист,
  колонка «ОСВ» периметра (osv_accounts)
- ББ/ОФР и файлы с рабочего стола режим не затрагивает; склейки кварталов в режиме нет
- тесты: unit на чистые функции, интеграционные — сценарий 2025 Q1 → 2026 Q1 с режимом
  и без, файл вне консолидации, чек-лист и карточка, подхват «один слот на счёт»

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 19:17:01 +04:00

452 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Сервис консолидаций ГК: создание, список периодов, карточка версии (spec consolidations-api «Создание
консолидации», «Список консолидаций ГК и карточка версии»; directory-policy «Видимость консолидаций»;
audit-trail `CONSOLIDATION_CREATE`; design D-15, D-16).
Порядок операции: право (`require` на маршруте) → видимость (ГК/версия — через ГК, невидимое = несуществующее, 404
с тем же телом) → проверки → запись. Перимeтр версии при создании — текущие члены ГСК или выбранные ЮЛ ГК
(`legal_entity_ids`, front-api-gaps D-1), `is_head_entity = false`; ГСК в `consolidation.related_group_id` не
сохраняется (фиксируется при верификации, Срез 7). Блокировка на создателя не ставится. Параллельное создание того
же периода ловит UNIQUE `uq_consolidation_group_year_quarter` → `IntegrityError` → 409 `CONSOLIDATION_EXISTS`
(`common/constraint_conflicts.py`).
"""
import uuid
from collections.abc import Sequence
from sqlalchemy.ext.asyncio import AsyncSession
from src.api.dependencies.audit import RequestAudit
from src.common.clock import utc_now
from src.common.exceptions import ConflictError, DomainValidationError, NotFoundError
from src.domain import checklist as checklist_domain
from src.domain.consolidation_periods import quarters_through, reporting_date
from src.domain.policy import FINISHED_STATUSES, VisibilityScope
from src.models.access import AppUser
from src.models.consolidation import (
Consolidation,
ConsolidationLock,
ConsolidationMember,
ConsolidationStatusHistory,
ConsolidationVersion,
)
from src.models.directories import LegalEntity
from src.models.enums import (
AuditAction,
AuditSection,
ConsolidationApproach,
ConsolidationScenario,
ConsolidationStatus,
RelatedGroupKind,
)
from src.repositories import consolidations as repo
from src.repositories import directories as directories_repo
from src.repositories.users import UserWithAccess
from src.schemas.consolidations import (
ChecklistCellOut,
ChecklistOut,
ChecklistRowOut,
ChecklistRuleOut,
ConsolidationCreate,
ConsolidationPeriodOut,
HistoryOut,
LockOut,
MemberOut,
MembersIn,
RefOut,
VersionBrief,
VersionCard,
)
from src.services import consolidation_documents
from src.services.company_groups import COMPANY_GROUP_NOT_FOUND
from src.services.consolidation_guard import VERSION_NOT_FOUND, lock_editable_version
from src.services.related_groups import RELATED_GROUP_NOT_FOUND
OBJECT_TYPE = "consolidation_version"
MIN_PERIMETER = 2 # в периметре консолидации не меньше двух ЮЛ (иначе нечего сопоставлять)
FIRST_VERSION_NO = 1
def lock_out(lock: ConsolidationLock, locked_by_name: str) -> LockOut:
"""Активная блокировка в ответе API (список, карточка, `POST …/lock`)."""
return LockOut(
locked_by=lock.locked_by,
locked_by_name=locked_by_name,
locked_at=lock.locked_at,
taken_over_from=lock.taken_over_from,
takeover_reason=lock.takeover_reason,
)
class ConsolidationService:
"""Консолидации ГК от имени текущего пользователя в его области видимости."""
def __init__(
self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, principal: UserWithAccess
) -> None:
self._session = session
self._audit = audit
self._scope = scope
self._principal = principal
async def create(self, data: ConsolidationCreate) -> VersionCard:
"""Консолидация ГК за период с версией №1 (`INITIATED`), периметром (ГСК или выбранные ЮЛ), историей и аудитом.
Проверки по порядку: ГК видима (404) → ГСК видима (404), принадлежит ГК и `kind = gsk` (422
`RELATED_GROUP_MISMATCH`) → периметр: без `legal_entity_ids` — текущие члены ГСК, не меньше двух (422
`PERIMETER_TOO_SMALL`); со списком — `_validate_perimeter` (422 `ENTITY_NOT_IN_GROUP` / `PERIMETER_TOO_SMALL`) →
периода ещё нет (409 `CONSOLIDATION_EXISTS`; гонку закрывает UNIQUE). Всё — в транзакции запроса: любая ошибка
откатывает всё.
После проверки ГК берётся advisory-блокировка привязок ГК (R-2): подхват документов (`bind_confirmed`) не
разминётся с параллельным подтверждением или автозаменой документа ЮЛ этой ГК.
"""
group = await directories_repo.get_company_group(self._session, self._scope, data.company_group_id)
if group is None:
raise NotFoundError(COMPANY_GROUP_NOT_FOUND)
# R-2 (design №30): до чтения подтверждённых документов — подтверждение и автозамена документов ЮЛ ГК ждут нас
# (новая версия им невидима до commit), а мы — их
await repo.lock_group_bindings(self._session, group.id)
related = await directories_repo.get_related_group(self._session, self._scope, data.related_group_id)
if related is None:
raise NotFoundError(RELATED_GROUP_NOT_FOUND)
if related.kind != RelatedGroupKind.gsk or related.company_group_id != group.id:
raise DomainValidationError("RELATED_GROUP_MISMATCH", "ГСК должна принадлежать выбранной ГК")
if data.legal_entity_ids is None:
entity_ids = [entity.id for _, entity in await directories_repo.current_members(self._session, related.id)]
if len(entity_ids) < MIN_PERIMETER:
raise DomainValidationError(
"PERIMETER_TOO_SMALL", f"В ГСК должно быть не меньше {MIN_PERIMETER} ЮЛ, сейчас {len(entity_ids)}"
)
else:
entity_ids = await self._validate_perimeter(group.id, data.legal_entity_ids)
if await repo.get_consolidation_for_group(self._session, self._scope, group.id, data.year, data.quarter):
raise ConflictError("CONSOLIDATION_EXISTS", "Консолидация ГК за этот период уже создана")
consolidation, version = await self._insert(group.id, data, entity_ids)
await self._audit.record(
action=AuditAction.CONSOLIDATION_CREATE,
section=AuditSection.CONSOLIDATION,
object_type=OBJECT_TYPE,
object_id=version.id,
consolidation_version_id=version.id,
after_value={
"consolidation_id": consolidation.id,
"year": consolidation.year,
"quarter": consolidation.quarter,
"members": sorted(entity_ids),
},
)
return await self._card(version, consolidation)
async def list_periods(self, group_id: uuid.UUID, *, in_work_only: bool) -> list[ConsolidationPeriodOut]:
"""Периоды ГК по убыванию года и квартала, в каждом — версии по убыванию `version_no`.
`in_work_only` оставляет версии со статусом не из {`VERIFIED`, `CANCELLED`, `ARCHIVED`, `REJECTED`}; период
без оставшихся версий в ответ не попадает. ГК вне видимости и несуществующая — один и тот же 404.
"""
if await directories_repo.get_company_group(self._session, self._scope, group_id) is None:
raise NotFoundError(COMPANY_GROUP_NOT_FOUND)
periods = await repo.list_periods(self._session, self._scope, group_id)
if in_work_only:
periods = [
(consolidation, [v for v in versions if v.status not in FINISHED_STATUSES])
for consolidation, versions in periods
]
periods = [(consolidation, versions) for consolidation, versions in periods if versions]
version_ids = [version.id for _, versions in periods for version in versions]
locks = await repo.active_locks(self._session, version_ids)
counts = await repo.members_counts(self._session, version_ids)
return [
ConsolidationPeriodOut(
consolidation_id=consolidation.id,
year=consolidation.year,
quarter=consolidation.quarter,
reporting_date=consolidation.reporting_date,
versions=[self._brief(version, locks.get(version.id), counts[version.id]) for version in versions],
)
for consolidation, versions in periods
]
async def card(self, version_id: uuid.UUID) -> VersionCard:
"""Карточка версии; версия ГК вне видимости и несуществующая — один и тот же 404."""
found = await repo.get_version(self._session, self._scope, version_id)
if found is None:
raise NotFoundError(VERSION_NOT_FOUND)
return await self._card(*found)
async def checklist(self, version_id: uuid.UUID) -> ChecklistOut:
"""Чек-лист комплекта версии: правила подхода (`BALANCE`, `INCOME`, ОСВ по коду счёта) и строки по ЮЛ периметра
(по имени) с ячейками в порядке правил; версия вне видимости и несуществующая — один и тот же 404."""
found = await repo.get_version(self._session, self._scope, version_id)
if found is None:
raise NotFoundError(VERSION_NOT_FOUND)
version, consolidation = found
entities = [entity for _, entity in await repo.members(self._session, version.id)]
rules, rows = await self._checklist_cells(version, consolidation, entities)
return ChecklistOut(
rules=[
ChecklistRuleOut(document_type=rule.document_type, account=rule.account, is_mandatory=rule.is_mandatory)
for rule in rules
],
rows=[
ChecklistRowOut(
legal_entity_id=entity.id,
legal_entity_name=entity.name,
legal_entity_inn=entity.inn,
cells=[self._cell_out(rule, cell) for rule, cell in zip(rules, cells, strict=True)],
)
for entity, cells in rows
],
)
async def version_title(self, version_id: uuid.UUID) -> tuple[str, int, str]:
"""Название ГК, год и квартал версии для имени файла экспорта (CSV чек-листа).
Версия вне видимости и несуществующая — один и тот же 404 как в `checklist`.
"""
found = await repo.get_version(self._session, self._scope, version_id)
if found is None:
raise NotFoundError(VERSION_NOT_FOUND)
version, consolidation = found
group = await directories_repo.get_company_group(self._session, self._scope, consolidation.company_group_id)
if group is None:
raise NotFoundError(VERSION_NOT_FOUND)
return group.name, consolidation.year, consolidation.quarter.value
async def _checklist_cells(
self, version: ConsolidationVersion, consolidation: Consolidation, entities: list[LegalEntity]
) -> tuple[list[checklist_domain.Rule], list[tuple[LegalEntity, list[checklist_domain.Cell]]]]:
"""Правила подхода версии и ячейки по каждому ЮЛ `entities` (D-13): привязанные документы версии,
неподтверждённые подходящие (`fits`) и состояние файлов ЮЛ. Кандидаты — `repo.unconfirmed_documents`,
фильтр `fits` — здесь."""
rules = checklist_domain.order_rules(
checklist_domain.Rule(rule.document_type, account, rule.is_mandatory)
for rule, account in await repo.rules_for(self._session, version.approach)
)
entity_ids = [entity.id for entity in entities]
candidates = await consolidation_documents.for_consolidation(
self._session,
await repo.unconfirmed_documents(
self._session, entity_ids, consolidation.year, self._period_from_version(consolidation)
),
consolidation,
)
unbound = [doc for doc in candidates if consolidation_documents.fits(doc, version, consolidation, entity_ids)]
rows = checklist_domain.build(
rules,
version_quarter=consolidation.quarter,
members=entities,
bound=await consolidation_documents.for_consolidation(
self._session, await repo.bound_documents(self._session, version.id), consolidation
),
unbound=unbound,
files_state=await repo.entity_files_state(self._session, entity_ids),
)
return rules, rows
@staticmethod
def _period_from_version(consolidation: Consolidation) -> uuid.UUID | None:
"""`consolidation_id` для выборок кандидатов в режиме «период из версии» (arm-core#103); иначе `None`."""
return consolidation.id if consolidation_documents.period_from_version() else None
@staticmethod
def _cell_out(rule: checklist_domain.Rule, cell: checklist_domain.Cell) -> ChecklistCellOut:
"""Ячейка ответа с правилом (D-2); `rule` — правило того же столбца, что и `cell`."""
return ChecklistCellOut(
document_type=rule.document_type,
account=rule.account,
status=cell.status,
missing_quarters=list(cell.missing_quarters),
document_ids=list(cell.document_ids),
)
async def set_members(self, version_id: uuid.UUID, data: MembersIn) -> VersionCard:
"""Заменяет периметр версии списком `data.legal_entity_ids`; ответ — карточка версии.
Порядок: `FOR UPDATE` версии → видимость (404) → редактируемость и своя блокировка (409, D-8, D-9) → дубли
схлопываются → все ЮЛ существуют и принадлежат ГК консолидации (422 `ENTITY_NOT_IN_GROUP`,
`legal_entity_ids` — в порядке запроса) → не меньше двух (422 `PERIMETER_TOO_SMALL`) → diff: ушедшие
теряют `consolidation_member` и привязки документов (`unbind_entity`), пришедшие получают член
(`is_head_entity = false`) и подхват подтверждённых документов (`bind_confirmed`); оставшиеся не трогаются →
`MEMBERSHIP_CHANGE` с полными отсортированными списками ЮЛ до/после. Состав не изменился — 200 без записи и
без события. `related_group_member` не меняется никогда.
"""
user = self._principal.user
version, consolidation = await lock_editable_version(self._session, self._scope, version_id, user.id)
requested = await self._validate_perimeter(consolidation.company_group_id, data.legal_entity_ids)
current = {member.legal_entity_id: member for member, _ in await repo.members(self._session, version.id)}
added = [entity_id for entity_id in requested if entity_id not in current]
removed = [entity_id for entity_id in current if entity_id not in set(requested)]
if not added and not removed:
return await self._card(version, consolidation)
for entity_id in removed:
await self._session.delete(current[entity_id])
self._session.add_all(
ConsolidationMember(consolidation_version_id=version.id, legal_entity_id=entity_id, is_head_entity=False)
for entity_id in added
)
version.updated_by = user.id
version.updated_at = utc_now()
await self._session.flush()
for entity_id in removed:
await consolidation_documents.unbind_entity(self._session, version_id=version.id, entity_id=entity_id)
if added:
await consolidation_documents.bind_confirmed(
self._session, version=version, consolidation=consolidation, entity_ids=added
)
await self._audit.record(
action=AuditAction.MEMBERSHIP_CHANGE,
section=AuditSection.CONSOLIDATION,
object_type=OBJECT_TYPE,
object_id=version.id,
consolidation_version_id=version.id,
before_value={"legal_entity_ids": sorted(current)},
after_value={"legal_entity_ids": sorted(requested)},
)
return await self._card(version, consolidation)
async def _validate_perimeter(
self, company_group_id: uuid.UUID, entity_ids: Sequence[uuid.UUID]
) -> list[uuid.UUID]:
"""Периметр консолидации — одно правило для создания и `PUT …/members` (front-api-gaps D-1): дубли схлопываются
в порядке первого вхождения → все ЮЛ существуют и принадлежат ГК (422 `ENTITY_NOT_IN_GROUP`, `legal_entity_ids`
— чужие в порядке запроса) → не меньше двух (422 `PERIMETER_TOO_SMALL`)."""
requested = list(dict.fromkeys(entity_ids))
valid = await repo.entities_in_group(self._session, company_group_id, requested)
foreign = [entity_id for entity_id in requested if entity_id not in valid]
if foreign:
raise DomainValidationError(
"ENTITY_NOT_IN_GROUP", "ЮЛ не принадлежат ГК консолидации", legal_entity_ids=foreign
)
if len(requested) < MIN_PERIMETER:
raise DomainValidationError(
"PERIMETER_TOO_SMALL", f"В периметре должно быть не меньше {MIN_PERIMETER} ЮЛ, сейчас {len(requested)}"
)
return requested
async def _insert(
self, group_id: uuid.UUID, data: ConsolidationCreate, entity_ids: list[uuid.UUID]
) -> tuple[Consolidation, ConsolidationVersion]:
"""Записывает консолидацию, версию №1, периметр и первую запись истории; подхватывает документы."""
now = utc_now()
user = self._principal.user
consolidation = Consolidation(
company_group_id=group_id,
year=data.year,
quarter=data.quarter,
reporting_date=reporting_date(data.year, data.quarter),
created_at=now,
)
self._session.add(consolidation)
await self._session.flush() # гонку за (ГК, год, квартал) ловит UNIQUE → IntegrityError → 409
version = ConsolidationVersion(
consolidation_id=consolidation.id,
version_no=FIRST_VERSION_NO,
status=ConsolidationStatus.INITIATED,
scenario=ConsolidationScenario.CREATE,
approach=ConsolidationApproach.BASIC,
goodwill_enabled=False,
initiator_id=user.id,
initiator_office_id=user.regional_office_id,
is_in_work=True,
created_at=now,
created_by=user.id,
)
self._session.add(version)
await self._session.flush()
self._session.add_all(
ConsolidationMember(consolidation_version_id=version.id, legal_entity_id=entity_id, is_head_entity=False)
for entity_id in entity_ids
)
self._session.add(
ConsolidationStatusHistory(
version_id=version.id,
from_status=None,
to_status=ConsolidationStatus.INITIATED,
changed_by=user.id,
changed_at=now,
)
)
await self._session.flush()
await consolidation_documents.bind_confirmed(
self._session, version=version, consolidation=consolidation, entity_ids=entity_ids
)
return consolidation, version
@staticmethod
def _brief(
version: ConsolidationVersion, lock: tuple[ConsolidationLock, str] | None, members_count: int
) -> VersionBrief:
return VersionBrief(
id=version.id,
version_no=version.version_no,
status=version.status,
scenario=version.scenario,
approach=version.approach,
lock=lock_out(*lock) if lock is not None else None,
created_at=version.created_at,
members_count=members_count,
)
async def _card(self, version: ConsolidationVersion, consolidation: Consolidation) -> VersionCard:
"""Карточка версии: версия, консолидация, ГК, инициатор, периметр, история, активная блокировка.
Чек-лист у членов — те же ячейки, что у `GET …/checklist`, в порядке правил подхода; у членов — тип ЮЛ и счета
ОСВ к периоду версии (front-api-gaps D-3).
"""
group = await directories_repo.get_company_group(self._session, self._scope, consolidation.company_group_id)
initiator = await self._session.get(AppUser, version.initiator_id)
office = await directories_repo.get_regional_office(self._session, version.initiator_office_id)
if group is None or initiator is None or office is None: # FK гарантируют наличие; группа видна по построению
raise NotFoundError(VERSION_NOT_FOUND)
members = await repo.members(self._session, version.id)
locks = await repo.active_locks(self._session, [version.id])
history = await repo.history(self._session, version.id)
brief = self._brief(version, locks.get(version.id), len(members))
rules, cells = await self._checklist_cells(version, consolidation, [entity for _, entity in members])
accounts = await repo.osv_accounts(
self._session,
[entity.id for _, entity in members],
consolidation.year,
quarters_through(consolidation.quarter),
self._period_from_version(consolidation),
)
return VersionCard(
**brief.model_dump(),
consolidation_id=consolidation.id,
year=consolidation.year,
quarter=consolidation.quarter,
reporting_date=consolidation.reporting_date,
company_group=RefOut(id=group.id, name=group.name),
initiator=RefOut(id=initiator.id, name=initiator.full_name),
initiator_office=RefOut(id=office.id, name=office.name),
goodwill_enabled=version.goodwill_enabled,
is_in_work=version.is_in_work,
previous_version_id=version.previous_version_id,
members=[
MemberOut(
id=entity.id,
name=entity.name,
inn=entity.inn,
entity_type=entity.entity_type,
is_head_entity=member.is_head_entity,
checklist=[
self._cell_out(rule, cell) for rule, cell in zip(rules, cells[position][1], strict=True)
],
osv_accounts=accounts.get(entity.id, []),
)
for position, (member, entity) in enumerate(members)
],
history=[
HistoryOut(
from_status=record.from_status,
to_status=record.to_status,
changed_by=record.changed_by,
changed_at=record.changed_at,
comment=record.comment,
)
for record in history
],
)