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

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

SQL без бизнес-правил; видимость — `src.domain.policy`.

Функции не делают `commit`: транзакцией запроса управляет `SessionDep`.
"""

import datetime
import uuid
from typing import Any

from sqlalchemy import Select, Sequence, delete, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

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 RelatedGroupKind

LIKE_ESCAPE = "\\"


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 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 get_alias(session: AsyncSession, entity_id: uuid.UUID, alias_id: uuid.UUID) -> LegalEntityAlias | None:
    """Алиас по id, только если он принадлежит ЮЛ `entity_id`."""
    stmt = select(LegalEntityAlias).where(
        LegalEntityAlias.id == alias_id, LegalEntityAlias.legal_entity_id == entity_id
    )
    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 list_regional_offices(session: AsyncSession) -> list[RegionalOffice]:
    """Все филиалы по имени."""
    return list((await session.execute(select(RegionalOffice).order_by(RegionalOffice.name))).scalars().all())