"""Сервис ГК (spec directories-api, directory-policy, audit-trail): создание, изменение со снимком истории, архив. Порядок операции: право (`require` на маршруте) → видимость (`get`) → изменение. Все записи — в транзакции запроса, `commit` делает `SessionDep`; `audit_event` пишется в той же транзакции через `RequestAudit`. """ import datetime import uuid from collections.abc import Mapping from typing import Any 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.policy import VisibilityScope, office_for_create from src.models.directories import COMPANY_GROUP_CODE_SEQ, CompanyGroup, CompanyGroupHistory from src.models.enums import AuditAction, AuditSection from src.repositories import directories as repo from src.schemas.common import PageParams from src.schemas.directories import CompanyGroupCreate, CompanyGroupFilter, CompanyGroupUpdate COMPANY_GROUP_NOT_FOUND = "ГК не найдена" # одинаково для чужой и несуществующей (spec directory-policy) COMPANY_GROUP_CODE_PREFIX = "GK" OBJECT_TYPE = "company_group" TRACKED_FIELDS = ("name", "crm_id", "base_regional_office_id") # поля ГК, изменяемые PATCH и попадающие в историю async def apply_group_update( # noqa: PLR0913 - сессия, аудит, ГК, изменения и момент — без значений по умолчанию session: AsyncSession, audit: RequestAudit, group: CompanyGroup, changes: Mapping[str, Any], *, now: datetime.datetime, description: str | None = None, ) -> None: """Изменение полей ГК со снимком истории (SCD4, D-6) и аудитом; общее для PATCH и импорта (R-14). `changes` — только фактически изменившиеся поля (проверки — у вызывающего). Снимок ПРЕЖНИХ значений уходит в `company_group_history` (`valid_to = now`), у живой записи `valid_from = now`; аудит — `ASSIGNMENT_CHANGE` при смене филиала, иначе `GROUP_CHANGE`. """ before = {field: getattr(group, field) for field in changes} session.add( CompanyGroupHistory( company_group_id=group.id, crm_id=group.crm_id, name=group.name, base_regional_office_id=group.base_regional_office_id, valid_from=group.valid_from, valid_to=now, changed_by=audit.user_id, ) ) for field, value in changes.items(): setattr(group, field, value) group.valid_from = now group.updated_at = now group.updated_by = audit.user_id await session.flush() await audit.record( action=AuditAction.ASSIGNMENT_CHANGE if "base_regional_office_id" in changes else AuditAction.GROUP_CHANGE, section=AuditSection.DIRECTORIES, object_type=OBJECT_TYPE, object_id=group.id, description=description, before_value=before, after_value=dict(changes), changed_field=next(iter(changes)) if len(changes) == 1 else None, ) class CompanyGroupService: """Операции над ГК от имени текущего пользователя в его области видимости.""" def __init__(self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope) -> None: self._session = session self._audit = audit self._scope = scope async def create(self, data: CompanyGroupCreate) -> CompanyGroup: """Новая ГК с кодом `GK-NNNNNN`; филиал — по policy (`office_for_create`); аудит `GROUP_CHANGE`.""" office_id = office_for_create(self._scope, data.base_regional_office_id) await self._check_office_exists(office_id) if data.crm_id is not None: await self._check_crm_id_free(data.crm_id) now = utc_now() group = CompanyGroup( code=await repo.next_code(self._session, COMPANY_GROUP_CODE_SEQ, COMPANY_GROUP_CODE_PREFIX), crm_id=data.crm_id, name=data.name, base_regional_office_id=office_id, valid_from=now, created_at=now, updated_at=now, updated_by=self._audit.user_id, ) self._session.add(group) await self._session.flush() await self._audit.record( action=AuditAction.GROUP_CHANGE, section=AuditSection.DIRECTORIES, object_type=OBJECT_TYPE, object_id=group.id, after_value={ "code": group.code, "name": group.name, "crm_id": group.crm_id, "base_regional_office_id": group.base_regional_office_id, }, ) return group async def update(self, group_id: uuid.UUID, data: CompanyGroupUpdate) -> CompanyGroup: """Изменение `name`/`crm_id`/`base_regional_office_id`; без фактических изменений ничего не пишется (D-6).""" group = await self.get(group_id) changes = await self._changes(group, data) if not changes: return group await apply_group_update(self._session, self._audit, group, changes, now=utc_now()) return group async def set_archived(self, group_id: uuid.UUID, archived: bool) -> CompanyGroup: # noqa: FBT001 """Архив/разархив ГК; повтор в том же состоянии ничего не пишет.""" group = await self.get(group_id) if group.is_archived == archived: return group group.is_archived = archived group.updated_at = utc_now() group.updated_by = self._audit.user_id await self._session.flush() await self._audit.record( action=AuditAction.GROUP_CHANGE, section=AuditSection.DIRECTORIES, object_type=OBJECT_TYPE, object_id=group.id, before_value={"is_archived": not archived}, after_value={"is_archived": archived}, changed_field="is_archived", ) return group async def get(self, group_id: uuid.UUID) -> CompanyGroup: """ГК в области видимости; невидимая и несуществующая — один и тот же 404.""" group = await repo.get_company_group(self._session, self._scope, group_id) if group is None: raise NotFoundError(COMPANY_GROUP_NOT_FOUND) return group async def legal_entities_count(self, group_id: uuid.UUID) -> int: """Число ЮЛ ГК (для карточки).""" return await repo.count_legal_entities(self._session, group_id) async def history(self, group_id: uuid.UUID) -> list[CompanyGroupHistory]: """Снимки ГК по убыванию `valid_from`; видимость проверяется как для карточки.""" group = await self.get(group_id) return await repo.list_company_group_history(self._session, group.id) async def page(self, params: CompanyGroupFilter, page: PageParams) -> tuple[list[tuple[CompanyGroup, int]], int]: """Страница видимых ГК с числом ЮЛ и общее число по фильтру.""" return await repo.list_company_groups( self._session, self._scope, q=params.q, is_archived=params.is_archived, limit=page.limit, offset=page.offset, ) async def _changes(self, group: CompanyGroup, data: CompanyGroupUpdate) -> dict[str, Any]: """Только реально изменившиеся поля из переданных в PATCH (проверки — по ним же).""" changes: dict[str, Any] = {} for field in TRACKED_FIELDS: if field not in data.model_fields_set: continue value = getattr(data, field) if field == "base_regional_office_id": value = office_for_create(self._scope, value) if value != getattr(group, field): changes[field] = value if changes.get("base_regional_office_id") is not None: await self._check_office_exists(changes["base_regional_office_id"]) if changes.get("crm_id") is not None: await self._check_crm_id_free(changes["crm_id"], except_group_id=group.id) return changes async def _check_office_exists(self, office_id: uuid.UUID | None) -> None: if office_id is not None and await repo.get_regional_office(self._session, office_id) is None: raise DomainValidationError("OFFICE_NOT_FOUND", "Филиал не найден") async def _check_crm_id_free(self, crm_id: str, *, except_group_id: uuid.UUID | None = None) -> None: if await repo.crm_id_taken(self._session, crm_id, except_group_id=except_group_id): raise ConflictError("CRM_ID_EXISTS", "ГК с таким crm_id уже существует")