"""Репозиторий справочников: ГК, ЮЛ, их история, членства ГСК/ГСЗ, алиасы, филиалы. SQL без бизнес-правил; видимость — `src.domain.policy`. Функции не делают `commit`: транзакцией запроса управляет `SessionDep`. """ import datetime import uuid from collections.abc import Iterable from typing import Any from sqlalchemy import Select, Sequence, and_, delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from src.domain.directory_import import DirectorySnapshot, EntitySnap, GroupSnap, OfficeSnap from src.domain.osv_mapping import account_sort_key from src.domain.policy import VisibilityScope, company_groups_visible, legal_entities_visible from src.models.accounting import ChartOfAccount from src.models.directories import ( CompanyGroup, CompanyGroupHistory, LegalEntity, LegalEntityAlias, LegalEntityHistory, RegionalOffice, RelatedGroup, RelatedGroupMember, ) from src.models.enums import FileStatus, RelatedGroupKind from src.models.files import Document, StoredFile LIKE_ESCAPE = "\\" ALIAS_LOCK_PREFIX = "legal-entity-alias" # пространство ключей advisory-блокировки алиасов async def list_accounts(session: AsyncSession) -> list[ChartOfAccount]: accounts = (await session.scalars(select(ChartOfAccount))).all() return sorted(accounts, key=lambda account: (account_sort_key(account.code), account.code)) def escape_like(text: str) -> str: """Экранирует `\\`, `%` и `_`, чтобы пользовательская строка искалась как подстрока, а не как шаблон LIKE.""" return text.replace(LIKE_ESCAPE, LIKE_ESCAPE * 2).replace("%", f"{LIKE_ESCAPE}%").replace("_", f"{LIKE_ESCAPE}_") async def next_code(session: AsyncSession, seq: Sequence, prefix: str) -> str: """Следующий код из последовательности PostgreSQL: `-NNNNNN` (6 цифр с нулями, D-5).""" number = (await session.execute(select(seq.next_value()))).scalar_one() return f"{prefix}-{number:06d}" def _filtered_groups(stmt: Select[Any], scope: VisibilityScope, q: str | None, is_archived: bool | None) -> Select[Any]: stmt = company_groups_visible(stmt, scope) if q is not None and q.strip(): pattern = f"%{escape_like(q.strip())}%" stmt = stmt.where( or_( CompanyGroup.name.ilike(pattern, escape=LIKE_ESCAPE), CompanyGroup.crm_id.ilike(pattern, escape=LIKE_ESCAPE), CompanyGroup.code.ilike(pattern, escape=LIKE_ESCAPE), ) ) if is_archived is not None: stmt = stmt.where(CompanyGroup.is_archived.is_(is_archived)) return stmt async def list_company_groups( # noqa: PLR0913 - фильтры и пагинация списка передаются явными именованными аргументами session: AsyncSession, scope: VisibilityScope, *, q: str | None, is_archived: bool | None, limit: int, offset: int, ) -> tuple[list[tuple[CompanyGroup, int]], int]: """Страница видимых ГК (по `name`) со счётчиком ЮЛ и общее число записей по фильтру (отдельный `count()`).""" legal_entities_count = ( select(func.count(LegalEntity.id)) .where(LegalEntity.company_group_id == CompanyGroup.id) .correlate(CompanyGroup) .scalar_subquery() ) page_stmt = ( _filtered_groups(select(CompanyGroup, legal_entities_count), scope, q, is_archived) .order_by(CompanyGroup.name, CompanyGroup.id) .limit(limit) .offset(offset) ) total_stmt = _filtered_groups(select(func.count(CompanyGroup.id)), scope, q, is_archived) rows = (await session.execute(page_stmt)).all() total = (await session.execute(total_stmt)).scalar_one() return [(group, count) for group, count in rows], total async def get_company_group(session: AsyncSession, scope: VisibilityScope, group_id: uuid.UUID) -> CompanyGroup | None: """ГК по id; `None` и для несуществующей, и для невидимой (неразличимы для вызывающего).""" stmt = company_groups_visible(select(CompanyGroup).where(CompanyGroup.id == group_id), scope) return (await session.execute(stmt)).scalar_one_or_none() async def crm_id_taken(session: AsyncSession, crm_id: str, *, except_group_id: uuid.UUID | None = None) -> bool: """Есть ли у какой-либо ГК (кроме `except_group_id`) такой `crm_id`; без фильтра видимости (уникален глобально).""" stmt = select(CompanyGroup.id).where(CompanyGroup.crm_id == crm_id) if except_group_id is not None: stmt = stmt.where(CompanyGroup.id != except_group_id) return (await session.execute(stmt.limit(1))).first() is not None async def related_group_crm_id_taken( session: AsyncSession, crm_id: str, *, except_group_id: uuid.UUID | None = None ) -> bool: """Есть ли у ГСК/ГСЗ (кроме `except_group_id`) такой `crm_id`; без фильтра видимости (уникален глобально).""" stmt = select(RelatedGroup.id).where(RelatedGroup.crm_id == crm_id) if except_group_id is not None: stmt = stmt.where(RelatedGroup.id != except_group_id) return (await session.execute(stmt.limit(1))).first() is not None async def count_legal_entities(session: AsyncSession, group_id: uuid.UUID) -> int: """Число ЮЛ/ИП с `company_group_id = group_id`.""" stmt = select(func.count(LegalEntity.id)).where(LegalEntity.company_group_id == group_id) return (await session.execute(stmt)).scalar_one() async def list_company_group_history(session: AsyncSession, group_id: uuid.UUID) -> list[CompanyGroupHistory]: """Снимки ГК по убыванию `valid_from` (новые первыми).""" stmt = ( select(CompanyGroupHistory) .where(CompanyGroupHistory.company_group_id == group_id) .order_by(CompanyGroupHistory.valid_from.desc(), CompanyGroupHistory.id) ) return list((await session.execute(stmt)).scalars().all()) def _filtered_entities( # noqa: PLR0913 - фильтры списка ЮЛ передаются явными именованными аргументами stmt: Select[Any], scope: VisibilityScope, *, company_group_id: uuid.UUID | None, outside_group: bool | None, inn: str | None, q: str | None, closed: bool | None, ) -> Select[Any]: stmt = legal_entities_visible(stmt, scope) if company_group_id is not None: stmt = stmt.where(LegalEntity.company_group_id == company_group_id) if outside_group is not None: group_column = LegalEntity.company_group_id stmt = stmt.where(group_column.is_(None) if outside_group else group_column.is_not(None)) if inn is not None: stmt = stmt.where(LegalEntity.inn == inn) if q is not None and q.strip(): pattern = f"%{escape_like(q.strip())}%" stmt = stmt.where( or_( LegalEntity.name.ilike(pattern, escape=LIKE_ESCAPE), LegalEntity.inn.ilike(pattern, escape=LIKE_ESCAPE), LegalEntity.kpp.ilike(pattern, escape=LIKE_ESCAPE), ) ) if closed is not None: stmt = stmt.where(LegalEntity.close_date.is_not(None) if closed else LegalEntity.close_date.is_(None)) return stmt async def list_legal_entities( # noqa: PLR0913 - фильтры и пагинация списка передаются явными именованными аргументами session: AsyncSession, scope: VisibilityScope, *, company_group_id: uuid.UUID | None, outside_group: bool | None, inn: str | None, q: str | None, closed: bool | None, limit: int, offset: int, ) -> tuple[list[LegalEntity], int]: """Страница видимых ЮЛ (по `name`, `id`) и общее число записей по фильтру. `inn` — точное равенство (вызывающий уже сделал `strip`); `q` — подстрока `name`/`inn`/`kpp` без регистра (R-4). """ filters: dict[str, Any] = { "company_group_id": company_group_id, "outside_group": outside_group, "inn": inn, "q": q, "closed": closed, } page_stmt = ( _filtered_entities(select(LegalEntity), scope, **filters) .order_by(LegalEntity.name, LegalEntity.id) .limit(limit) .offset(offset) ) total_stmt = _filtered_entities(select(func.count(LegalEntity.id)), scope, **filters) items = list((await session.execute(page_stmt)).scalars().all()) total = (await session.execute(total_stmt)).scalar_one() return items, total async def get_legal_entity(session: AsyncSession, scope: VisibilityScope, entity_id: uuid.UUID) -> LegalEntity | None: """ЮЛ по id; `None` и для несуществующего, и для невидимого (неразличимы для вызывающего).""" stmt = legal_entities_visible(select(LegalEntity).where(LegalEntity.id == entity_id), scope) return (await session.execute(stmt)).scalar_one_or_none() async def get_legal_entity_by_inn(session: AsyncSession, inn: str) -> LegalEntity | None: """ЮЛ по ИНН без фильтра видимости — для проверки глобальной уникальности `inn`.""" return (await session.execute(select(LegalEntity).where(LegalEntity.inn == inn))).scalar_one_or_none() async def documents_count_by_entity(session: AsyncSession, entity_ids: list[uuid.UUID]) -> dict[uuid.UUID, int]: """Число документов файлов каждой ЮЛ со статусом файла не SUPERSEDED (D-13). Возвращает словарь {entity_id: count}; отсутствующих ЮЛ — не в словаре. """ if not entity_ids: return {} stmt = ( select(Document.legal_entity_id, func.count(Document.id).label("count")) .join(StoredFile, Document.stored_file_id == StoredFile.id) .where(Document.legal_entity_id.in_(entity_ids), StoredFile.status != FileStatus.SUPERSEDED) .group_by(Document.legal_entity_id) ) rows = (await session.execute(stmt)).all() return {entity_id: count for entity_id, count in rows} async def legal_entity_crm_id_taken( session: AsyncSession, crm_id: str, *, except_entity_id: uuid.UUID | None = None ) -> bool: """Есть ли у какого-либо ЮЛ (кроме `except_entity_id`) такой `crm_id` — без фильтра видимости.""" stmt = select(LegalEntity.id).where(LegalEntity.crm_id == crm_id) if except_entity_id is not None: stmt = stmt.where(LegalEntity.id != except_entity_id) return (await session.execute(stmt.limit(1))).first() is not None async def list_legal_entity_history(session: AsyncSession, entity_id: uuid.UUID) -> list[LegalEntityHistory]: """Снимки ЮЛ по убыванию `valid_from` (новые первыми, R-3).""" stmt = ( select(LegalEntityHistory) .where(LegalEntityHistory.legal_entity_id == entity_id) .order_by(LegalEntityHistory.valid_from.desc(), LegalEntityHistory.id) ) return list((await session.execute(stmt)).scalars().all()) async def current_memberships_by_entities( session: AsyncSession, entity_ids: Iterable[uuid.UUID] ) -> dict[uuid.UUID, list[tuple[RelatedGroupMember, RelatedGroup]]]: """Текущие членства ЮЛ в ГСК/ГСЗ вместе с группой по набору ЮЛ одним запросом (#109); у каждого ЮЛ — по виду и коду группы; ЮЛ без членств в словаре нет.""" ids = set(entity_ids) if not ids: return {} stmt = ( select(RelatedGroupMember, RelatedGroup) .join(RelatedGroup, RelatedGroup.id == RelatedGroupMember.related_group_id) .where(RelatedGroupMember.legal_entity_id.in_(ids), RelatedGroupMember.is_current.is_(True)) .order_by(RelatedGroupMember.kind, RelatedGroup.code) ) result: dict[uuid.UUID, list[tuple[RelatedGroupMember, RelatedGroup]]] = {} for member, group in (await session.execute(stmt)).all(): result.setdefault(member.legal_entity_id, []).append((member, group)) return result async def close_memberships( session: AsyncSession, entity_id: uuid.UUID, now: datetime.datetime ) -> list[RelatedGroupMember]: """Закрывает все текущие членства ЮЛ (SCD2: `valid_to = now`, `is_current = false`); возвращает закрытые.""" stmt = select(RelatedGroupMember).where( RelatedGroupMember.legal_entity_id == entity_id, RelatedGroupMember.is_current.is_(True) ) members = list((await session.execute(stmt)).scalars().all()) for member in members: member.valid_to = now member.is_current = False await session.flush() return members def _current_members_count() -> Any: """Скалярный подзапрос: число текущих членов группы `RelatedGroup` внешнего запроса.""" return ( select(func.count(RelatedGroupMember.id)) .where(RelatedGroupMember.related_group_id == RelatedGroup.id, RelatedGroupMember.is_current.is_(True)) .correlate(RelatedGroup) .scalar_subquery() ) async def list_related_groups(session: AsyncSession, company_group_id: uuid.UUID) -> list[tuple[RelatedGroup, int]]: """ГСК/ГСЗ ГК (по `name`, `id`) с числом текущих членов; видимость ГК проверяет вызывающий.""" stmt = ( select(RelatedGroup, _current_members_count()) .where(RelatedGroup.company_group_id == company_group_id) .order_by(RelatedGroup.name, RelatedGroup.id) ) return [(group, count) for group, count in (await session.execute(stmt)).all()] async def get_related_group(session: AsyncSession, scope: VisibilityScope, group_id: uuid.UUID) -> RelatedGroup | None: """ГСК/ГСЗ по id; видимость — через ГК (§10.1); `None` и для несуществующей, и для невидимой. `CompanyGroup` присоединяется явно: `company_groups_visible` фильтрует по колонке ГК и без JOIN дала бы декартово произведение (T4.1). """ stmt = ( select(RelatedGroup) .join(CompanyGroup, CompanyGroup.id == RelatedGroup.company_group_id) .where(RelatedGroup.id == group_id) ) return (await session.execute(company_groups_visible(stmt, scope))).scalar_one_or_none() async def count_current_members(session: AsyncSession, related_group_id: uuid.UUID) -> int: """Число текущих членов группы.""" stmt = select(func.count(RelatedGroupMember.id)).where( RelatedGroupMember.related_group_id == related_group_id, RelatedGroupMember.is_current.is_(True) ) return (await session.execute(stmt)).scalar_one() async def has_current_members(session: AsyncSession, related_group_id: uuid.UUID) -> bool: """Есть ли у группы хотя бы один текущий член.""" return await count_current_members(session, related_group_id) > 0 async def current_membership_of_kind( session: AsyncSession, entity_id: uuid.UUID, kind: RelatedGroupKind ) -> RelatedGroupMember | None: """Текущее членство ЮЛ в группе вида `kind` (не больше одного — частичный уникальный индекс v5.14).""" stmt = select(RelatedGroupMember).where( RelatedGroupMember.legal_entity_id == entity_id, RelatedGroupMember.kind == kind, RelatedGroupMember.is_current.is_(True), ) return (await session.execute(stmt)).scalar_one_or_none() async def current_membership_in_group( session: AsyncSession, related_group_id: uuid.UUID, entity_id: uuid.UUID ) -> RelatedGroupMember | None: """Текущее членство ЮЛ в конкретной группе.""" stmt = select(RelatedGroupMember).where( RelatedGroupMember.related_group_id == related_group_id, RelatedGroupMember.legal_entity_id == entity_id, RelatedGroupMember.is_current.is_(True), ) return (await session.execute(stmt)).scalar_one_or_none() async def current_members( session: AsyncSession, related_group_id: uuid.UUID ) -> list[tuple[RelatedGroupMember, LegalEntity]]: """Текущие члены группы вместе с ЮЛ, по имени ЮЛ.""" stmt = ( select(RelatedGroupMember, LegalEntity) .join(LegalEntity, LegalEntity.id == RelatedGroupMember.legal_entity_id) .where(RelatedGroupMember.related_group_id == related_group_id, RelatedGroupMember.is_current.is_(True)) .order_by(LegalEntity.name, LegalEntity.id) ) return [(member, entity) for member, entity in (await session.execute(stmt)).all()] async def list_group_member_rows(session: AsyncSession, related_group_id: uuid.UUID) -> list[RelatedGroupMember]: """Все строки состава группы (текущие и закрытые), по `valid_from`.""" stmt = ( select(RelatedGroupMember) .where(RelatedGroupMember.related_group_id == related_group_id) .order_by(RelatedGroupMember.valid_from, RelatedGroupMember.id) ) return list((await session.execute(stmt)).scalars().all()) async def delete_group_member_rows(session: AsyncSession, related_group_id: uuid.UUID) -> None: """Удаляет все строки состава группы (вызывать только для группы без текущих членов).""" await session.execute(delete(RelatedGroupMember).where(RelatedGroupMember.related_group_id == related_group_id)) async def list_aliases(session: AsyncSession, entity_id: uuid.UUID) -> list[LegalEntityAlias]: """Алиасы ЮЛ по времени добавления.""" stmt = ( select(LegalEntityAlias) .where(LegalEntityAlias.legal_entity_id == entity_id) .order_by(LegalEntityAlias.created_at, LegalEntityAlias.id) ) return list((await session.execute(stmt)).scalars().all()) async def find_alias_owner(session: AsyncSession, normalized: str) -> LegalEntityAlias | None: """Алиас с таким `normalized_alias` у любого ЮЛ — без фильтра видимости: уникальность глобальная (ANL-028).""" stmt = ( select(LegalEntityAlias) .where(LegalEntityAlias.normalized_alias == normalized) .order_by(LegalEntityAlias.created_at, LegalEntityAlias.id) .limit(1) ) return (await session.execute(stmt)).scalar_one_or_none() async def find_alias_owners(session: AsyncSession, normalized: str) -> list[tuple[LegalEntityAlias, str]]: """Все алиасы с таким `normalized_alias` (у любых ЮЛ, без фильтра видимости) и имена их владельцев — по времени добавления; строки алиасов — `FOR UPDATE` (параллельное удаление алиаса в справочнике дождётся нас или мы — его и не увидим удалённого). Уникальность в БД — только в пределах ЮЛ, поэтому владельцев может быть несколько.""" stmt = ( select(LegalEntityAlias, LegalEntity.name) .join(LegalEntity, LegalEntity.id == LegalEntityAlias.legal_entity_id) .where(LegalEntityAlias.normalized_alias == normalized) .order_by(LegalEntityAlias.created_at, LegalEntityAlias.id) .with_for_update(of=LegalEntityAlias) .execution_options(populate_existing=True) ) return [(row[0], row[1]) for row in await session.execute(stmt)] async def lock_alias_key(session: AsyncSession, normalized: str) -> None: """`pg_advisory_xact_lock` по нормализованному алиасу: сериализует проверку «алиас свободен» и вставку у всех писателей алиасов (`aliases.add`, ручное сопоставление контрагента). Глобальной уникальности `normalized_alias` в БД нет (индекс — в пределах ЮЛ), без блокировки два параллельных запроса закрепили бы алиас за двумя ЮЛ. Снимается в конце транзакции.""" key = func.hashtextextended(f"{ALIAS_LOCK_PREFIX}:{normalized}", 0) await session.execute(select(func.pg_advisory_xact_lock(key))) async def get_alias( session: AsyncSession, entity_id: uuid.UUID, alias_id: uuid.UUID, *, refresh: bool = False ) -> LegalEntityAlias | None: """Алиас по id, только если он принадлежит ЮЛ `entity_id`; `refresh` — состояние строки заменяется свежим из БД.""" stmt = select(LegalEntityAlias).where( LegalEntityAlias.id == alias_id, LegalEntityAlias.legal_entity_id == entity_id ) if refresh: stmt = stmt.execution_options(populate_existing=True) return (await session.execute(stmt)).scalar_one_or_none() async def get_regional_office(session: AsyncSession, office_id: uuid.UUID) -> RegionalOffice | None: return await session.get(RegionalOffice, office_id) async def regional_offices_by_ids( session: AsyncSession, office_ids: Iterable[uuid.UUID | None] ) -> dict[uuid.UUID, RegionalOffice]: """Филиалы по набору id одним запросом (включая неактивные); `None` и дубли игнорируются.""" ids = {office_id for office_id in office_ids if office_id is not None} if not ids: return {} rows = (await session.execute(select(RegionalOffice).where(RegionalOffice.id.in_(ids)))).scalars().all() return {office.id: office for office in rows} async def company_group_names_by_ids( session: AsyncSession, group_ids: Iterable[uuid.UUID | None] ) -> dict[uuid.UUID, str]: """Имена ГК по набору id одним запросом (включая архивные); `None` и дубли игнорируются.""" ids = {group_id for group_id in group_ids if group_id is not None} if not ids: return {} rows = (await session.execute(select(CompanyGroup.id, CompanyGroup.name).where(CompanyGroup.id.in_(ids)))).all() return {group_id: name for group_id, name in rows} async def list_regional_offices(session: AsyncSession) -> list[RegionalOffice]: """Все филиалы по имени.""" return list((await session.execute(select(RegionalOffice).order_by(RegionalOffice.name))).scalars().all()) async def load_snapshot(session: AsyncSession) -> DirectorySnapshot: """Снимок справочника для плана импорта (§13.2) тремя запросами, без фильтра видимости (импорт — администратор). Филиалы — по имени. ГК — только с `crm_id` (импорт сопоставляет по `iD ГК`), включая архивные: иначе план создал бы ГК с занятым `crm_id`; ГСК/ГСЗ ГК — первая по `code` группа каждого вида. ЮЛ — по ИНН с текущими членствами (`is_current`) каждого вида. """ offices = { office.name: OfficeSnap(id=office.id, name=office.name) for office in (await session.execute(select(RegionalOffice))).scalars() } groups: dict[str, GroupSnap] = {} group_stmt = ( select( CompanyGroup.id, CompanyGroup.crm_id, CompanyGroup.name, CompanyGroup.base_regional_office_id, RelatedGroup.id, RelatedGroup.kind, ) .outerjoin(RelatedGroup, RelatedGroup.company_group_id == CompanyGroup.id) .where(CompanyGroup.crm_id.is_not(None)) .order_by(CompanyGroup.id, RelatedGroup.code) ) for group_id, crm_id, name, office_id, related_id, kind in (await session.execute(group_stmt)).all(): snap = groups.get(crm_id) if snap is None: snap = GroupSnap(id=group_id, crm_id=crm_id, name=name, office_id=office_id, gsk_id=None, gsz_id=None) groups[crm_id] = snap if kind is RelatedGroupKind.gsk and snap.gsk_id is None: snap.gsk_id = related_id elif kind is RelatedGroupKind.gsz and snap.gsz_id is None: snap.gsz_id = related_id entities: dict[str, EntitySnap] = {} entity_stmt = select( LegalEntity.id, LegalEntity.inn, LegalEntity.name, LegalEntity.entity_type, LegalEntity.company_group_id, LegalEntity.kpp, RelatedGroupMember.related_group_id, RelatedGroupMember.kind, ).outerjoin( RelatedGroupMember, and_(RelatedGroupMember.legal_entity_id == LegalEntity.id, RelatedGroupMember.is_current.is_(True)), ) for entity_id, inn, name, entity_type, group_id, kpp, related_id, kind in ( await session.execute(entity_stmt) ).all(): entity = entities.get(inn) if entity is None: entity = EntitySnap( id=entity_id, inn=inn, name=name, entity_type=entity_type, group_id=group_id, gsk_group_id=None, gsz_group_id=None, kpp=kpp, ) entities[inn] = entity if kind is RelatedGroupKind.gsk: entity.gsk_group_id = related_id elif kind is RelatedGroupKind.gsz: entity.gsz_group_id = related_id return DirectorySnapshot(offices_by_name=offices, groups_by_crm=groups, entities_by_inn=entities)