"""Сервис консолидаций ГК: создание, список периодов, карточка версии (spec consolidations-api «Создание консолидации», «Список консолидаций ГК и карточка версии»; directory-policy «Видимость консолидаций»; audit-trail `CONSOLIDATION_CREATE`; design D-15, D-16). Порядок операции: право (`require` на маршруте) → видимость (ГК/версия — через ГК, невидимое = несуществующее, 404 с тем же телом) → проверки → запись. Перимeтр версии при создании — текущие члены ГСК, `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 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.consolidation_periods import 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 ( ConsolidationCreate, ConsolidationPeriodOut, HistoryOut, LockOut, MemberOut, RefOut, VersionBrief, VersionCard, ) from src.services import consolidation_documents from src.services.company_groups import COMPANY_GROUP_NOT_FOUND from src.services.related_groups import RELATED_GROUP_NOT_FOUND CONSOLIDATION_NOT_FOUND = "консолидация не найдена" # одинаково для чужой и несуществующей (spec directory-policy) VERSION_NOT_FOUND = "версия консолидации не найдена" # одинаково для чужой и несуществующей OBJECT_TYPE = "consolidation_version" MIN_PERIMETER = 2 # в периметре консолидации не меньше двух ЮЛ (иначе нечего сопоставлять) FIRST_VERSION_NO = 1 def _lock_out(lock: ConsolidationLock, locked_by_name: str) -> LockOut: 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`) → в ГСК не меньше двух текущих членов (422 `PERIMETER_TOO_SMALL`) → периода ещё нет (409 `CONSOLIDATION_EXISTS`; гонку закрывает UNIQUE). Всё — в транзакции запроса: любая ошибка откатывает всё. """ 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) 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", "ГСК должна принадлежать выбранной ГК") entities = [entity for _, entity in await directories_repo.current_members(self._session, related.id)] if len(entities) < MIN_PERIMETER: raise DomainValidationError( "PERIMETER_TOO_SMALL", f"В ГСК должно быть не меньше {MIN_PERIMETER} ЮЛ, сейчас {len(entities)}" ) 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, entities) 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.id for entity in entities), }, ) 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 _insert( self, group_id: uuid.UUID, data: ConsolidationCreate, entities: list[LegalEntity] ) -> 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 in entities ) 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, [entity.id for entity in entities]) 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: """Карточка версии: версия, консолидация, ГК, инициатор, периметр, история, активная блокировка. Чек-лист у членов — пустой список до Task 8 (`domain.checklist.build`). """ 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)) 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, is_head_entity=member.is_head_entity, checklist=[], # Task 8: ячейки по правилам подхода ) for member, entity in 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 ], )