"""Репозиторий консолидаций (Срез 4; spec consolidations-api, consolidated-osv; design D-1, D-5, D-13).

Функции не делают `commit` и, кроме `get_version`, `get_consolidation_for_group` и `list_periods` (они получают
`VisibilityScope` и скрывают ГК чужого филиала), не применяют policy: проверку видимости делает вызывающий сервис.
Порядок блокировок — файл → документ → advisory ГК (`lock_group_bindings`) → версии; `lock_versions_for_entity` берёт
advisory ГК ЮЛ и версии пачкой по `id`.
"""

import uuid
from collections import defaultdict
from collections.abc import Collection, Sequence

from sqlalchemy import ColumnElement, Select, and_, case, exists, false, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from src.domain.checklist import EntityFilesState
from src.domain.osv_mapping import account_sort_key
from src.domain.policy import VisibilityScope, company_groups_visible
from src.models.access import AppUser
from src.models.accounting import ChartOfAccount
from src.models.audit_ex import RequiredDocumentRule
from src.models.consolidation import (
    Consolidation,
    ConsolidationDocument,
    ConsolidationLock,
    ConsolidationMember,
    ConsolidationStatusHistory,
    ConsolidationVersion,
)
from src.models.directories import CompanyGroup, LegalEntity, LegalEntityAlias, RelatedGroup, RelatedGroupMember
from src.models.enums import (
    ConsolidationApproach,
    ConsolidationStatus,
    DocumentStatus,
    DocumentType,
    FileStatus,
    OsvDataType,
    OsvGroupType,
    PeriodQuarter,
    RelatedGroupKind,
)
from src.models.files import Document, StoredFile
from src.models.osv import OsvEntry
from src.repositories.directories import LIKE_ESCAPE, escape_like
from src.repositories.files import on_desk

FILES_IN_PROGRESS = (FileStatus.UPLOADED, FileStatus.PROCESSING)
GROUP_BINDINGS_LOCK_PREFIX = "consolidation-group"  # пространство ключей advisory-блокировки привязок ГК (R-2)
SUMMARY_DATA_TYPES = (OsvDataType.BU, OsvDataType.SUM)
UNCONFIRMED_STATUSES = (DocumentStatus.RECOGNIZED, DocumentStatus.NEEDS_CHECK)
# Пробельные символы `str.strip()` (`str.isspace`): `btrim` в SQL срезает ровно их, как `.strip()` склейки свода (D-2)
PY_WHITESPACE = "".join(char for char in map(chr, range(0x3001)) if char.isspace())


def _visible_consolidations(scope: VisibilityScope) -> Select[tuple[Consolidation]]:
    """`SELECT consolidation` с присоединённой ГК и фильтром видимости (§10.1)."""
    stmt = select(Consolidation).join(CompanyGroup, CompanyGroup.id == Consolidation.company_group_id)
    return company_groups_visible(stmt, scope)


async def get_version(
    session: AsyncSession, scope: VisibilityScope, version_id: uuid.UUID
) -> tuple[ConsolidationVersion, Consolidation] | None:
    """Версия и её консолидация, если ГК видна пользователю (`consolidation` → `company_group`); иначе `None`."""
    stmt = (
        select(ConsolidationVersion, Consolidation)
        .join(Consolidation, Consolidation.id == ConsolidationVersion.consolidation_id)
        .join(CompanyGroup, CompanyGroup.id == Consolidation.company_group_id)
        .where(ConsolidationVersion.id == version_id)
    )
    row = (await session.execute(company_groups_visible(stmt, scope))).first()
    return None if row is None else (row[0], row[1])


async def get_consolidation(
    session: AsyncSession, scope: VisibilityScope, consolidation_id: uuid.UUID
) -> Consolidation | None:
    """Консолидация по id, если её ГК видна пользователю; чужая и несуществующая — `None` (v5.18, design D-3)."""
    stmt = _visible_consolidations(scope).where(Consolidation.id == consolidation_id)
    return (await session.execute(stmt)).scalar_one_or_none()


async def get_consolidation_for_group(
    session: AsyncSession, scope: VisibilityScope, group_id: uuid.UUID, year: int, quarter: PeriodQuarter
) -> Consolidation | None:
    """Консолидация ГК за период (уникальна по `(ГК, год, квартал)`), если ГК видна пользователю."""
    stmt = _visible_consolidations(scope).where(
        Consolidation.company_group_id == group_id, Consolidation.year == year, Consolidation.quarter == quarter
    )
    return (await session.execute(stmt)).scalar_one_or_none()


async def list_periods(
    session: AsyncSession, scope: VisibilityScope, group_id: uuid.UUID
) -> list[tuple[Consolidation, list[ConsolidationVersion]]]:
    """Периоды ГК по убыванию года и квартала; версии каждого — по убыванию `version_no`. Двумя запросами."""
    stmt = (
        _visible_consolidations(scope)
        .where(Consolidation.company_group_id == group_id)
        .order_by(Consolidation.year.desc(), Consolidation.quarter.desc())
    )
    consolidations = list((await session.execute(stmt)).scalars())
    if not consolidations:
        return []
    versions: dict[uuid.UUID, list[ConsolidationVersion]] = defaultdict(list)
    rows = await session.execute(
        select(ConsolidationVersion)
        .where(ConsolidationVersion.consolidation_id.in_([c.id for c in consolidations]))
        .order_by(ConsolidationVersion.version_no.desc())
    )
    for version in rows.scalars():
        versions[version.consolidation_id].append(version)
    return [(consolidation, versions[consolidation.id]) for consolidation in consolidations]


async def lock_version(session: AsyncSession, version_id: uuid.UUID) -> ConsolidationVersion | None:
    """Версия под `FOR UPDATE`; `populate_existing` — состояние строки в сессии заменяется свежим из БД."""
    stmt = (
        select(ConsolidationVersion)
        .where(ConsolidationVersion.id == version_id)
        .with_for_update()
        .execution_options(populate_existing=True)
    )
    return (await session.execute(stmt)).scalar_one_or_none()


async def lock_group_bindings(session: AsyncSession, group_id: uuid.UUID) -> None:
    """`pg_advisory_xact_lock` привязок документов ГК (R-2, design №30): сериализует создание версии консолидации ГК с
    подтверждением и автозаменой документов ЮЛ этой ГК. Новая версия невидима другим транзакциям до commit, поэтому
    `FOR UPDATE` версий её не ловит: создание берёт этот lock до чтения подтверждённых документов, а
    `lock_versions_for_entity` — до блокировки версий. Снимается в конце транзакции."""
    key = func.hashtextextended(f"{GROUP_BINDINGS_LOCK_PREFIX}:{group_id}", 0)
    await session.execute(select(func.pg_advisory_xact_lock(key)))


async def lock_versions_for_entity(
    session: AsyncSession, entity_id: uuid.UUID, statuses: Collection[ConsolidationStatus]
) -> list[ConsolidationVersion]:
    """Версии со статусом из `statuses`, где ЮЛ — член периметра; блокируются `FOR UPDATE ORDER BY id` (R-2).

    Первой берётся advisory-блокировка привязок ГК ЮЛ (`lock_group_bindings`; у ЮЛ без ГК её нет): иначе версия,
    создаваемая параллельно, не видна ни здесь, ни подтверждённый нами документ — там. ЮЛ без ГК (после выхода из
    группы или переноса) может оставаться членом версий «в работе» — его ловит ветка `is_member`; такое ЮЛ в периметре
    заставляет любой `PUT …/members` отвечать `ENTITY_NOT_IN_GROUP`, пока его не уберут из периметра.

    Блокируется НАДМНОЖЕСТВО: версии этих статусов, где ЮЛ — член ИЛИ `consolidation.company_group_id` равен ГК ЮЛ
    (`legal_entity.company_group_id`). Иначе подтверждение документа ∥ `PUT …/members` теряет привязку: подтверждение
    не видит ещё не зафиксированного нового члена и не ждёт версию, которую правит периметр. Блокировка версии ГК,
    где ЮЛ формально не член, заставляет такую правку дождаться нас и закоммитить периметр до того, как мы его
    перечитаем. Возвращаются только версии, где ЮЛ член, — отдельным вторым запросом: в READ COMMITTED он получает
    свежий снимок (уже видит периметры, закоммиченные ожидавшимися транзакциями); блокировка на нём не нужна, строки
    заблокированы первым запросом. `populate_existing` — состояние версий в сессии заменяется свежим.
    """
    is_member = exists().where(
        ConsolidationMember.consolidation_version_id == ConsolidationVersion.id,
        ConsolidationMember.legal_entity_id == entity_id,
    )
    group_id = (
        await session.execute(select(LegalEntity.company_group_id).where(LegalEntity.id == entity_id))
    ).scalar_one_or_none()
    if group_id is not None:
        await lock_group_bindings(session, group_id)
    in_group = Consolidation.company_group_id == group_id if group_id is not None else false()
    superset = (
        select(ConsolidationVersion.id)
        .join(Consolidation, Consolidation.id == ConsolidationVersion.consolidation_id)
        .where(ConsolidationVersion.status.in_(statuses), is_member | in_group)
        .order_by(ConsolidationVersion.id)
        .with_for_update(of=ConsolidationVersion)
    )
    locked_ids = list((await session.execute(superset)).scalars())
    if not locked_ids:
        return []
    members_only = (
        select(ConsolidationVersion)
        .where(ConsolidationVersion.id.in_(locked_ids), is_member)
        .order_by(ConsolidationVersion.id)
        .execution_options(populate_existing=True)
    )
    return list((await session.execute(members_only)).scalars())


async def versions_of(session: AsyncSession, consolidation_id: uuid.UUID) -> list[ConsolidationVersion]:
    """Все версии консолидации по возрастанию `version_no` (без блокировки; проверки до приёма файлов, D-3)."""
    stmt = (
        select(ConsolidationVersion)
        .where(ConsolidationVersion.consolidation_id == consolidation_id)
        .order_by(ConsolidationVersion.version_no)
    )
    return list((await session.execute(stmt)).scalars())


async def lock_versions_of_consolidation(
    session: AsyncSession,
    consolidation_id: uuid.UUID,
    entity_id: uuid.UUID,
    statuses: Collection[ConsolidationStatus],
) -> list[ConsolidationVersion]:
    """Версии консолидации со статусом из `statuses`, где ЮЛ — член периметра, под `FOR UPDATE ORDER BY id` (R-2;
    design D-4: переход после загрузки в консолидацию).

    Двухшаговая схема, как в `lock_versions_for_entity`: сначала блокируются ВСЕ версии консолидации с нужными
    статусами (`ORDER BY id FOR UPDATE`) — без условия членства: в READ COMMITTED после ожидания блокировки `EXISTS`
    в том же операторе не пересчитывается и пропустил бы члена, добавленного ожидавшейся правкой периметра. Затем
    вторым запросом, без блокировки (строки уже заблокированы), выбираются версии, где ЮЛ — член; он получает свежий
    снимок и видит закоммиченные правки. `populate_existing` — состояние версий в сессии заменяется свежим."""
    locked = (
        select(ConsolidationVersion.id)
        .where(
            ConsolidationVersion.consolidation_id == consolidation_id,
            ConsolidationVersion.status.in_(statuses),
        )
        .order_by(ConsolidationVersion.id)
        .with_for_update()
    )
    locked_ids = list((await session.execute(locked)).scalars())
    if not locked_ids:
        return []
    is_member = exists().where(
        ConsolidationMember.consolidation_version_id == ConsolidationVersion.id,
        ConsolidationMember.legal_entity_id == entity_id,
    )
    members_only = (
        select(ConsolidationVersion)
        .where(ConsolidationVersion.id.in_(locked_ids), is_member)
        .order_by(ConsolidationVersion.id)
        .execution_options(populate_existing=True)
    )
    return list((await session.execute(members_only)).scalars())


async def active_lock(session: AsyncSession, version_id: uuid.UUID) -> ConsolidationLock | None:
    """Активная блокировка версии (`released_at IS NULL`; не более одной — partial UNIQUE)."""
    stmt = select(ConsolidationLock).where(
        ConsolidationLock.consolidation_version_id == version_id, ConsolidationLock.released_at.is_(None)
    )
    return (await session.execute(stmt)).scalar_one_or_none()


async def active_locks(
    session: AsyncSession, version_ids: Collection[uuid.UUID]
) -> dict[uuid.UUID, tuple[ConsolidationLock, str]]:
    """Активные блокировки версий из `version_ids` одним запросом: версия → (блокировка, ФИО владельца)."""
    if not version_ids:
        return {}
    stmt = (
        select(ConsolidationLock, AppUser.full_name)
        .join(AppUser, AppUser.id == ConsolidationLock.locked_by)
        .where(ConsolidationLock.consolidation_version_id.in_(version_ids), ConsolidationLock.released_at.is_(None))
    )
    return {row[0].consolidation_version_id: (row[0], row[1]) for row in await session.execute(stmt)}


async def members_counts(session: AsyncSession, version_ids: Collection[uuid.UUID]) -> dict[uuid.UUID, int]:
    """Размер периметра каждой версии из `version_ids` одним запросом (версия без членов — 0)."""
    if not version_ids:
        return {}
    stmt = (
        select(ConsolidationMember.consolidation_version_id, func.count(ConsolidationMember.id))
        .where(ConsolidationMember.consolidation_version_id.in_(version_ids))
        .group_by(ConsolidationMember.consolidation_version_id)
    )
    counts = {row[0]: row[1] for row in await session.execute(stmt)}
    return {version_id: counts.get(version_id, 0) for version_id in version_ids}


async def members(session: AsyncSession, version_id: uuid.UUID) -> list[tuple[ConsolidationMember, LegalEntity]]:
    """Периметр версии: член и его ЮЛ, по имени ЮЛ (`id` — для устойчивости порядка)."""
    stmt = (
        select(ConsolidationMember, LegalEntity)
        .join(LegalEntity, LegalEntity.id == ConsolidationMember.legal_entity_id)
        .where(ConsolidationMember.consolidation_version_id == version_id)
        .order_by(LegalEntity.name, LegalEntity.id)
    )
    return [(row[0], row[1]) for row in await session.execute(stmt)]


async def entities_in_group(
    session: AsyncSession, group_id: uuid.UUID, entity_ids: Collection[uuid.UUID]
) -> set[uuid.UUID]:
    """Из `entity_ids` — те ЮЛ, что существуют и принадлежат ГК `group_id` (`legal_entity.company_group_id`)."""
    if not entity_ids:
        return set()
    stmt = select(LegalEntity.id).where(LegalEntity.id.in_(entity_ids), LegalEntity.company_group_id == group_id)
    return set((await session.execute(stmt)).scalars())


async def bound_documents(session: AsyncSession, version_id: uuid.UUID) -> list[Document]:
    """Документы, привязанные к версии (`consolidation_document`)."""
    stmt = (
        select(Document)
        .join(ConsolidationDocument, ConsolidationDocument.document_id == Document.id)
        .where(ConsolidationDocument.consolidation_version_id == version_id)
        .order_by(Document.legal_entity_id, Document.osv_account_code, Document.period_quarter, Document.id)
    )
    return list((await session.execute(stmt)).scalars())


async def document_versions(session: AsyncSession, document_id: uuid.UUID) -> list[uuid.UUID]:
    """Версии, к которым привязан документ (по `id`)."""
    stmt = (
        select(ConsolidationDocument.consolidation_version_id)
        .where(ConsolidationDocument.document_id == document_id)
        .order_by(ConsolidationDocument.consolidation_version_id)
    )
    return list((await session.execute(stmt)).scalars())


async def load_osv_rows(session: AsyncSession, version_id: uuid.UUID) -> list[tuple[OsvEntry, Document, LegalEntity]]:
    """Исходные строки свода: контрагентские (`group_type = COUNTERPARTY`) строки BU/SUM привязанных документов.

    `TOTAL` не хранится, строки `ACCOUNT` и данные `NU` отбрасываются. Порядок: имя ЮЛ, код счёта документа, квартал,
    `row_number` (а затем `OsvEntry.id` для устойчивости).
    """
    stmt = (
        select(OsvEntry, Document, LegalEntity)
        .join(Document, Document.id == OsvEntry.document_id)
        .join(ConsolidationDocument, ConsolidationDocument.document_id == Document.id)
        .join(LegalEntity, LegalEntity.id == Document.legal_entity_id)
        .where(
            ConsolidationDocument.consolidation_version_id == version_id,
            OsvEntry.group_type == OsvGroupType.COUNTERPARTY,
            OsvEntry.data_type.in_(SUMMARY_DATA_TYPES),
        )
        .order_by(
            LegalEntity.name, Document.osv_account_code, Document.period_quarter, OsvEntry.row_number, OsvEntry.id
        )
    )
    return [(row[0], row[1], row[2]) for row in await session.execute(stmt)]


async def history(session: AsyncSession, version_id: uuid.UUID) -> list[ConsolidationStatusHistory]:
    """Журнал смен статуса версии по времени (`changed_at`, затем `id`)."""
    stmt = (
        select(ConsolidationStatusHistory)
        .where(ConsolidationStatusHistory.version_id == version_id)
        .order_by(ConsolidationStatusHistory.changed_at, ConsolidationStatusHistory.id)
    )
    return list((await session.execute(stmt)).scalars())


async def entity_files_state(
    session: AsyncSession, entity_ids: Sequence[uuid.UUID], consolidation_id: uuid.UUID
) -> dict[uuid.UUID, EntityFilesState]:
    """Состояние файлов каждого ЮЛ из `entity_ids` на столе консолидации `consolidation_id` (чек-лист, D-13;
    `files.on_desk`, AURORA-1646): файл в работе и статус последнего файла.

    «В работе» — любой файл `UPLOADED`/`PROCESSING`; «последний» — по `uploaded_at` (при равенстве или NULL — по `id`),
    включая замещённые файлы. ЮЛ без файлов на столе — `EntityFilesState(False, None)`. Неподтверждённые
    документы-кандидаты сюда не входят (R-5).
    """
    states = {entity_id: EntityFilesState(has_in_progress=False, last_file_status=None) for entity_id in entity_ids}
    if not states:
        return states
    desk = on_desk(consolidation_id)
    in_progress = select(StoredFile.legal_entity_id).where(
        StoredFile.legal_entity_id.in_(states), StoredFile.status.in_(FILES_IN_PROGRESS), desk
    )
    busy = set((await session.execute(in_progress.distinct())).scalars())
    last = (
        select(StoredFile.legal_entity_id, StoredFile.status)
        .where(StoredFile.legal_entity_id.in_(states), desk)
        .distinct(StoredFile.legal_entity_id)
        .order_by(StoredFile.legal_entity_id, StoredFile.uploaded_at.desc().nulls_last(), StoredFile.id.desc())
    )
    last_status = {entity_id: status for entity_id, status in await session.execute(last)}
    return {
        entity_id: EntityFilesState(has_in_progress=entity_id in busy, last_file_status=last_status.get(entity_id))
        for entity_id in states
    }


async def rules_for(
    session: AsyncSession, approach: ConsolidationApproach
) -> list[tuple[RequiredDocumentRule, str | None]]:
    """Правила комплекта подхода (`required_document_rule`) и код счёта правила (`chart_of_account.code`; у ББ/ОФР —
    `None`). Порядок столбцов задаёт домен (`checklist.order_rules`); здесь — детерминированный по `id`."""
    stmt = (
        select(RequiredDocumentRule, ChartOfAccount.code)
        .outerjoin(ChartOfAccount, ChartOfAccount.id == RequiredDocumentRule.account_id)
        .where(RequiredDocumentRule.approach == approach)
        .order_by(RequiredDocumentRule.id)
    )
    return [(row[0], row[1]) for row in await session.execute(stmt)]


def _osv_of_consolidation(consolidation_id: uuid.UUID) -> ColumnElement[bool]:
    """ОСВ, файл которой загружен в консолидацию: в режиме «период из версии» (arm-core#103) — любого периода шапки."""
    return and_(Document.document_type == DocumentType.OSV, StoredFile.consolidation_id == consolidation_id)


async def file_consolidation_ids(
    session: AsyncSession, file_ids: Collection[uuid.UUID]
) -> dict[uuid.UUID, uuid.UUID | None]:
    """`stored_file.consolidation_id` по id файлов одним запросом (режим «период из версии», arm-core#103)."""
    if not file_ids:
        return {}
    stmt = select(StoredFile.id, StoredFile.consolidation_id).where(StoredFile.id.in_(list(file_ids)))
    return {file_id: consolidation_id for file_id, consolidation_id in await session.execute(stmt)}


async def unconfirmed_documents(
    session: AsyncSession,
    entity_ids: Collection[uuid.UUID],
    consolidation_id: uuid.UUID,
    year: int,
    *,
    osv_any_year: bool = False,
) -> list[Document]:
    """Неподтверждённые документы ЮЛ за год (`RECOGNIZED`/`NEEDS_CHECK`) из файлов не `SUPERSEDED` на столе консолидации
    `consolidation_id` (`files.on_desk`, AURORA-1646: документы других консолидаций и карточки ЮЛ не учитываются); любых
    типов (R-5). `osv_any_year` (режим «период из версии», arm-core#103) — плюс ОСВ, загруженные в эту консолидацию,
    любого года.

    Кандидаты чек-листа: «подходящесть» версии (`fits`) проверяет сервис. Порядок — по ЮЛ, типу, счёту, кварталу, `id`.
    """
    if not entity_ids:
        return []
    in_period: ColumnElement[bool] = Document.period_year == year
    if osv_any_year:
        in_period = or_(in_period, _osv_of_consolidation(consolidation_id))
    stmt = (
        select(Document)
        .join(StoredFile, StoredFile.id == Document.stored_file_id)
        .where(
            Document.legal_entity_id.in_(list(entity_ids)),
            on_desk(consolidation_id),
            in_period,
            Document.status.in_(UNCONFIRMED_STATUSES),
            StoredFile.status != FileStatus.SUPERSEDED,
        )
        .order_by(
            Document.legal_entity_id,
            Document.document_type,
            Document.osv_account_code,
            Document.period_quarter,
            Document.id,
        )
    )
    return list((await session.execute(stmt)).scalars())


async def osv_accounts(
    session: AsyncSession,
    entity_ids: Collection[uuid.UUID],
    year: int,
    quarters: Collection[PeriodQuarter],
    consolidation_id: uuid.UUID | None = None,
) -> dict[uuid.UUID, list[str]]:
    """Счета ОСВ, загруженные к периоду версии (колонка «ОСВ» панели [34]; front-api-gaps D-3): уникальные
    `osv_account_code` документов `OSV` ЮЛ за год `year` и кварталы `quarters`; документ не `ERROR`, файл не
    `SUPERSEDED`/`ERROR`, квартал и код счёта заданы; статус подтверждения и привязка к версии не важны. Порядок —
    натуральный (`account_sort_key`); ЮЛ без документов в словаре нет. `consolidation_id` (режим «период из версии»,
    arm-core#103) — плюс ОСВ, загруженные в эту консолидацию, любого периода."""
    if not entity_ids:
        return {}
    in_period: ColumnElement[bool] = and_(Document.period_year == year, Document.period_quarter.in_(list(quarters)))
    if consolidation_id is not None:
        in_period = or_(in_period, _osv_of_consolidation(consolidation_id))
    stmt = (
        select(Document.legal_entity_id, Document.osv_account_code)
        .join(StoredFile, StoredFile.id == Document.stored_file_id)
        .where(
            Document.legal_entity_id.in_(list(entity_ids)),
            Document.document_type == DocumentType.OSV,
            in_period,
            Document.status != DocumentStatus.ERROR,
            StoredFile.status.not_in([FileStatus.SUPERSEDED, FileStatus.ERROR]),
            Document.osv_account_code.is_not(None),
            Document.osv_account_code != "",
        )
        .distinct()
    )
    accounts: dict[uuid.UUID, list[str]] = defaultdict(list)
    for entity_id, code in await session.execute(stmt):
        accounts[entity_id].append(code)
    return {entity_id: sorted(codes, key=account_sort_key) for entity_id, codes in accounts.items()}


def _summary_entries(version_id: uuid.UUID) -> Select[tuple[OsvEntry]]:
    """Строки свода версии: контрагентские (`COUNTERPARTY`) строки BU/SUM ОСВ-документов, привязанных к версии."""
    return (
        select(OsvEntry)
        .join(Document, Document.id == OsvEntry.document_id)
        .join(ConsolidationDocument, ConsolidationDocument.document_id == Document.id)
        .where(
            ConsolidationDocument.consolidation_version_id == version_id,
            Document.document_type == DocumentType.OSV,
            OsvEntry.group_type == OsvGroupType.COUNTERPARTY,
            OsvEntry.data_type.in_(SUMMARY_DATA_TYPES),
        )
    )


async def entry_in_version(session: AsyncSession, entry_id: uuid.UUID, version_id: uuid.UUID) -> OsvEntry | None:
    """Строка `entry_id`, если она — строка свода версии (`_summary_entries`): документ ОСВ привязан к версии, строка
    `COUNTERPARTY` с данными BU/SUM. Иначе (непривязанный документ, `ACCOUNT`, `NU`, чужая строка) — `None`."""
    stmt = _summary_entries(version_id).where(OsvEntry.id == entry_id)
    return (await session.execute(stmt)).scalar_one_or_none()


async def lock_entries_same_name(
    session: AsyncSession, version_id: uuid.UUID, *, entry_id: uuid.UUID, name: str
) -> list[OsvEntry]:
    """Строка `entry_id` и строки свода версии с тем же `counterparty_name.strip()` = `name` под
    `FOR NO KEY UPDATE ORDER BY id` (ручное сопоставление, D-10); пустой `name` — только сама строка.

    Строки ОСВ общие для всех версий, к которым привязан документ (OQ-123): сопоставления в двух версиях блокируют
    пересекающиеся множества строк, порядок по `id` исключает взаимоблокировку, а `populate_existing` отдаёт
    состояние после ожидания (READ COMMITTED перечитывает заблокированную строку), поэтому `counterparty_id IS NULL`
    проверяется по свежим данным. Блокируется только `osv_entry` (документы и привязки не трогаются)."""
    same = OsvEntry.id == entry_id
    if name:
        same = or_(same, func.btrim(OsvEntry.counterparty_name, PY_WHITESPACE) == name)
    stmt = (
        _summary_entries(version_id)
        .where(same)
        .order_by(OsvEntry.id)
        .with_for_update(of=OsvEntry, key_share=True)
        .execution_options(populate_existing=True)
    )
    return list((await session.execute(stmt)).scalars())


async def search_entities(  # noqa: PLR0913 - фильтры поиска кандидатов передаются явными именованными аргументами
    session: AsyncSession,
    *,
    q: str | None,
    group_id: uuid.UUID | None,
    member_ids: Collection[uuid.UUID],
    recommended_ids: Collection[uuid.UUID],
    limit: int,
) -> list[tuple[LegalEntity, CompanyGroup | None]]:
    """Кандидаты для ручного сопоставления (D-11) без policy видимости, с ГК каждого ЮЛ.

    `group_id` задан — ЮЛ этой ГК и `member_ids` (периметр версии); `None` — весь справочник. `q` (уже без краевых
    пробелов): ИНН по префиксу (пробелы внутри `q` не учитываются), имя или алиас — подстрока без регистра; шаблоны
    LIKE в `q` экранируются. Порядок: `recommended_ids` первыми, затем имя и `id`; не больше `limit`.
    """
    stmt = select(LegalEntity, CompanyGroup).outerjoin(CompanyGroup, CompanyGroup.id == LegalEntity.company_group_id)
    if group_id is not None:
        in_scope = LegalEntity.company_group_id == group_id
        stmt = stmt.where(or_(in_scope, LegalEntity.id.in_(list(member_ids))) if member_ids else in_scope)
    if q:
        pattern = f"%{escape_like(q)}%"
        by_alias = exists().where(
            LegalEntityAlias.legal_entity_id == LegalEntity.id,
            LegalEntityAlias.alias.ilike(pattern, escape=LIKE_ESCAPE),
        )
        conditions = [LegalEntity.name.ilike(pattern, escape=LIKE_ESCAPE), by_alias]
        if inn := "".join(q.split()):
            conditions.append(LegalEntity.inn.like(f"{escape_like(inn)}%", escape=LIKE_ESCAPE))
        stmt = stmt.where(or_(*conditions))
    first = case((LegalEntity.id.in_(list(recommended_ids)), 0), else_=1) if recommended_ids else None
    order = [LegalEntity.name, LegalEntity.id] if first is None else [first, LegalEntity.name, LegalEntity.id]
    return [(row[0], row[1]) for row in await session.execute(stmt.order_by(*order).limit(limit))]


async def current_related_groups(
    session: AsyncSession, entity_ids: Collection[uuid.UUID]
) -> dict[uuid.UUID, dict[RelatedGroupKind, RelatedGroup]]:
    """Текущие ГСК/ГСЗ каждого ЮЛ из `entity_ids` одним запросом: ЮЛ → вид → группа (не больше одной каждого вида)."""
    if not entity_ids:
        return {}
    stmt = (
        select(RelatedGroupMember.legal_entity_id, RelatedGroup)
        .join(RelatedGroup, RelatedGroup.id == RelatedGroupMember.related_group_id)
        .where(RelatedGroupMember.legal_entity_id.in_(entity_ids), RelatedGroupMember.is_current.is_(True))
    )
    groups: dict[uuid.UUID, dict[RelatedGroupKind, RelatedGroup]] = defaultdict(dict)
    for entity_id, group in await session.execute(stmt):
        groups[entity_id][group.kind] = group
    return dict(groups)