arm-core/api/src/repositories/directories.py

"""Репозиторий справочников: ГК, ЮЛ, их история, членства ГСК/ГСЗ, алиасы, филиалы.

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.policy import VisibilityScope, company_groups_visible, legal_entities_visible
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-блокировки алиасов


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: `<prefix>-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 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))
        )
    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` без учёта регистра (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(
    session: AsyncSession, entity_id: uuid.UUID
) -> list[tuple[RelatedGroupMember, RelatedGroup]]:
    """Текущие членства ЮЛ в ГСК/ГСЗ вместе с группой (для карточки), по виду и коду группы."""
    stmt = (
        select(RelatedGroupMember, RelatedGroup)
        .join(RelatedGroup, RelatedGroup.id == RelatedGroupMember.related_group_id)
        .where(RelatedGroupMember.legal_entity_id == entity_id, RelatedGroupMember.is_current.is_(True))
        .order_by(RelatedGroupMember.kind, RelatedGroup.code)
    )
    return [(member, group) for member, group in (await session.execute(stmt)).all()]


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 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,
        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, 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,
            )
            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)