5b3b1011cb
Строки пишутся по явному счётчику (sheet.max_row делал сборку квадратичной: 6 000 строк — 8.4 с → 0.5 с), сборка вынесена в run_in_threadpool. Тексты проходят _put_text: запрещённые XML-символы вырезаются, тип ячейки строковый — ведущие =, +, -, @ не становятся формулами (R-12, design #77). Параметр entity_names удалён (R-13, #78), порог точности double в #73 исправлен, тесты пустых сумм и 6 000 строк. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
286 lines
16 KiB
Python
286 lines
16 KiB
Python
"""Сводная ОСВ версии: страница и сводка (spec consolidated-osv «Страница свода и сводка»; design D-1, D-16).
|
||
|
||
Свод — чистая функция версии (DECISION-5), в БД не хранится: каждый запрос заново собирает его из периметра,
|
||
привязанных ОСВ-документов и их строк (`domain.consolidated_osv.build`) и отдаёт срез. Сервис только читает: без
|
||
блокировок и аудита, версия в любом статусе (включая `CANCELLED`) читается; видимость — через ГК консолидации
|
||
(чужая и несуществующая версии — один 404 `VERSION_NOT_FOUND`).
|
||
|
||
Запросов к БД — константа, не зависящая от числа строк и контрагентов (без N+1): `get_version`, периметр,
|
||
документы версии, строки, два запроса индекса контрагентов (`CounterpartyIndex.load`) и один `SELECT legal_entity` по
|
||
собранным идентификаторам — контрагенты строк и кандидаты дорезолва. Дорезолв на лету кормит `build` тем же
|
||
`CounterpartyIndex`, поэтому идентификаторы кандидатов собираются заранее тем же `resolve` по словарям в памяти.
|
||
"""
|
||
|
||
import uuid
|
||
from collections.abc import Collection, Sequence
|
||
from typing import Literal
|
||
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.api.dependencies.audit import RequestAudit
|
||
from src.common.exceptions import DomainValidationError, NotFoundError
|
||
from src.domain import consolidated_osv as domain
|
||
from src.domain.aliases import normalize_alias
|
||
from src.domain.policy import VisibilityScope
|
||
from src.models.consolidation import Consolidation, ConsolidationVersion
|
||
from src.models.directories import CompanyGroup, LegalEntity, RelatedGroup
|
||
from src.models.enums import AuditAction, AuditSection, DocumentType, RelatedGroupKind
|
||
from src.models.files import Document
|
||
from src.models.osv import OsvEntry
|
||
from src.repositories import consolidations as repo
|
||
from src.repositories.counterparties import CounterpartyIndex
|
||
from src.schemas.common import Page
|
||
from src.schemas.consolidations import (
|
||
CandidateOut,
|
||
ConsolidatedRowOut,
|
||
GapOut,
|
||
OsvFilters,
|
||
OsvPageParams,
|
||
OsvSummaryOut,
|
||
RefOut,
|
||
)
|
||
from src.services.consolidation_guard import VERSION_NOT_FOUND
|
||
from src.services.osv_export import build_workbook
|
||
|
||
__all__ = ["MAX_CANDIDATES", "MIN_QUERY_ALL", "CandidateScope", "ConsolidatedOsvService", "domain_filter", "row_out"]
|
||
|
||
CandidateScope = Literal["group", "all"]
|
||
MAX_CANDIDATES = 100 # кандидатов в ответе не больше (D-11)
|
||
MIN_QUERY_ALL = 3 # `scope=all`: поиск по всему справочнику — от трёх символов (решение №5)
|
||
|
||
|
||
def row_out(row: domain.ConsolidatedRow) -> ConsolidatedRowOut:
|
||
"""Строка свода → схема API: кортежи домена (`entry_ids`, `warnings`) — списки, суммы — `Decimal` (JSON-строка)."""
|
||
return ConsolidatedRowOut.model_validate(row, from_attributes=True)
|
||
|
||
|
||
def domain_filter(f: OsvFilters) -> domain.OsvFilter:
|
||
"""Query-модель фильтров → доменный `OsvFilter` (повторяемый `match_kind` → множество)."""
|
||
return domain.OsvFilter(
|
||
legal_entity_id=f.legal_entity_id,
|
||
account_root=f.account_root,
|
||
counterparty_type=f.counterparty_type,
|
||
unmapped=f.unmapped,
|
||
match_kinds=frozenset(f.match_kind or ()),
|
||
q=f.q,
|
||
)
|
||
|
||
|
||
def _entity(entity: LegalEntity) -> domain.Entity:
|
||
return domain.Entity(id=entity.id, name=entity.name, inn=entity.inn, company_group_id=entity.company_group_id)
|
||
|
||
|
||
def _ref(item: CompanyGroup | RelatedGroup | None) -> RefOut | None:
|
||
return None if item is None else RefOut(id=item.id, name=item.name)
|
||
|
||
|
||
def _source_document(document: Document) -> domain.SourceDocument:
|
||
return domain.SourceDocument(
|
||
id=document.id,
|
||
legal_entity_id=document.legal_entity_id,
|
||
osv_account_code=document.osv_account_code,
|
||
period_quarter=document.period_quarter,
|
||
is_cumulative=document.is_cumulative,
|
||
)
|
||
|
||
|
||
def _source_row(entry: OsvEntry) -> domain.SourceRow:
|
||
return domain.SourceRow(
|
||
id=entry.id,
|
||
document_id=entry.document_id,
|
||
row_number=entry.row_number,
|
||
account_code=entry.account_code,
|
||
account_root=entry.account_root,
|
||
counterparty_name=entry.counterparty_name,
|
||
counterparty_inn=entry.counterparty_inn,
|
||
counterparty_id=entry.counterparty_id,
|
||
match_kind=entry.counterparty_match_kind,
|
||
opening_debit=entry.opening_debit,
|
||
opening_credit=entry.opening_credit,
|
||
turnover_debit=entry.turnover_debit,
|
||
turnover_credit=entry.turnover_credit,
|
||
closing_debit=entry.closing_debit,
|
||
closing_credit=entry.closing_credit,
|
||
)
|
||
|
||
|
||
def _candidate_ids(rows: Sequence[domain.SourceRow], index: CounterpartyIndex) -> set[uuid.UUID]:
|
||
"""Идентификаторы контрагентов, имена которых понадобятся `build`: хранимые `counterparty_id` и результат дорезолва
|
||
для строк без него (`resolve` — по словарям индекса, без обращений к БД; ключ склейки включает имя и ИНН, поэтому
|
||
результат для представителя группы равен результату для любой её строки)."""
|
||
ids: set[uuid.UUID] = set()
|
||
for row in rows:
|
||
if row.counterparty_id is not None:
|
||
ids.add(row.counterparty_id)
|
||
elif (resolution := index.resolve(row.counterparty_inn, row.counterparty_name)) is not None:
|
||
ids.add(resolution.entity_id)
|
||
return ids
|
||
|
||
|
||
async def _entities(session: AsyncSession, ids: Collection[uuid.UUID]) -> dict[uuid.UUID, domain.Entity]:
|
||
"""ЮЛ справочника по `ids` одним запросом, без policy (контрагент может быть из любого филиала)."""
|
||
if not ids:
|
||
return {}
|
||
found = await session.execute(select(LegalEntity).where(LegalEntity.id.in_(ids)))
|
||
return {entity.id: _entity(entity) for entity in found.scalars()}
|
||
|
||
|
||
class ConsolidatedOsvService:
|
||
"""Чтение сводной ОСВ версии: страница с фильтрами и сводка с пробелами (без блокировок и аудита)."""
|
||
|
||
def __init__(self, session: AsyncSession, scope: VisibilityScope) -> None:
|
||
self._session = session
|
||
self._scope = scope
|
||
|
||
async def _build(self, version_id: uuid.UUID) -> tuple[domain.Consolidated, ConsolidationVersion, Consolidation]:
|
||
"""Свод версии (`domain.build`) вместе с версией и консолидацией; версия вне видимости — 404."""
|
||
found = await repo.get_version(self._session, self._scope, version_id)
|
||
if found is None:
|
||
raise NotFoundError(VERSION_NOT_FOUND)
|
||
version, consolidation = found
|
||
members = {member.id: _entity(member) for _, member in await repo.members(self._session, version.id)}
|
||
documents = [
|
||
_source_document(document)
|
||
for document in await repo.bound_documents(self._session, version.id)
|
||
if document.document_type == DocumentType.OSV
|
||
]
|
||
# ровно то, что отдал `load_osv_rows`: он оставляет `COUNTERPARTY`/`BU`+`SUM`, домен их не перефильтровывает
|
||
rows = [_source_row(entry) for entry, _, _ in await repo.load_osv_rows(self._session, version.id)]
|
||
index = await CounterpartyIndex.load(self._session)
|
||
entities = {**await _entities(self._session, _candidate_ids(rows, index) - members.keys()), **members}
|
||
consolidated = domain.build(
|
||
year=consolidation.year,
|
||
quarter=consolidation.quarter,
|
||
group_id=consolidation.company_group_id,
|
||
members=members,
|
||
entities=entities,
|
||
documents=documents,
|
||
rows=rows,
|
||
resolver=index,
|
||
)
|
||
return consolidated, version, consolidation
|
||
|
||
async def page(self, version_id: uuid.UUID, f: OsvFilters, page: OsvPageParams) -> Page[ConsolidatedRowOut]:
|
||
"""Страница свода после фильтров: `total` — число строк после фильтра, срез — в памяти (порядок D-16)."""
|
||
consolidated, _, _ = await self._build(version_id)
|
||
rows = domain.apply_filter(consolidated.rows, domain_filter(f))
|
||
window = rows[page.offset : page.offset + page.limit]
|
||
return Page[ConsolidatedRowOut](
|
||
items=[row_out(row) for row in window], total=len(rows), limit=page.limit, offset=page.offset
|
||
)
|
||
|
||
async def export(self, version_id: uuid.UUID, f: OsvFilters, audit: RequestAudit) -> tuple[bytes, str]:
|
||
"""xlsx сводной ОСВ (spec «Экспорт xlsx», D-14): строки после тех же фильтров, что у `page`, без пагинации, в
|
||
порядке D-16; лист пробелов — по своду целиком. Имя файла — `svod-osv_<ГК>_<год>-<квартал>_v<n>.xlsx` (без
|
||
percent-кодирования: оно в маршруте). `audit_event(EXPORT)` пишется после сборки книги, в транзакции запроса."""
|
||
consolidated, version, consolidation = await self._build(version_id)
|
||
rows = domain.apply_filter(consolidated.rows, domain_filter(f))
|
||
content = await run_in_threadpool(build_workbook, consolidated, rows)
|
||
group_name = (
|
||
await self._session.execute(
|
||
select(CompanyGroup.name).where(CompanyGroup.id == consolidation.company_group_id)
|
||
)
|
||
).scalar_one()
|
||
await audit.record(
|
||
action=AuditAction.EXPORT,
|
||
section=AuditSection.EXPORT,
|
||
object_type="consolidation_version",
|
||
object_id=version.id,
|
||
consolidation_version_id=version.id,
|
||
after_value={
|
||
"report": "consolidated_osv",
|
||
"filters": f.model_dump(mode="json", exclude_none=True),
|
||
"rows": len(rows),
|
||
},
|
||
)
|
||
filename = (
|
||
f"svod-osv_{group_name}_{consolidation.year}-{consolidation.quarter.value}_v{version.version_no}.xlsx"
|
||
)
|
||
return content, filename
|
||
|
||
async def summary(self, version_id: uuid.UUID) -> OsvSummaryOut:
|
||
"""Сводка: строки, несопоставленные, автосопоставленные по имени/алиасу, ОСВ-документы версии и пробелы."""
|
||
consolidated, _, _ = await self._build(version_id)
|
||
rows_total, unmapped_total, auto_matched_total = domain.summary(consolidated)
|
||
return OsvSummaryOut(
|
||
rows_total=rows_total,
|
||
unmapped_total=unmapped_total,
|
||
auto_matched_total=auto_matched_total,
|
||
documents_total=consolidated.documents_total,
|
||
gaps=[GapOut.model_validate(gap, from_attributes=True) for gap in consolidated.gaps],
|
||
)
|
||
|
||
async def row_for_entry(self, version_id: uuid.UUID, entry_id: uuid.UUID) -> ConsolidatedRowOut | None:
|
||
"""Строка свода, в `entry_ids` которой есть `entry_id`; `None` — строка в своде не участвует (пара с пробелом
|
||
или документ не выбран источником при наличии нарастающего)."""
|
||
consolidated, _, _ = await self._build(version_id)
|
||
row = next((row for row in consolidated.rows if entry_id in row.entry_ids), None)
|
||
return None if row is None else row_out(row)
|
||
|
||
async def candidates(
|
||
self, version_id: uuid.UUID, *, q: str | None, scope: CandidateScope, entry_id: uuid.UUID | None
|
||
) -> list[CandidateOut]:
|
||
"""Кандидаты для ручного сопоставления контрагента (spec consolidated-osv «Кандидаты для сопоставления», D-11).
|
||
|
||
Версия — через видимость (404). `group` — ЮЛ периметра и ГК консолидации, `q` необязателен; `all` — весь
|
||
справочник без policy, `q` без краевых пробелов не короче `MIN_QUERY_ALL` (иначе 422 `QUERY_TOO_SHORT`).
|
||
`recommended` — нормализованное имя или алиас ЮЛ совпали с `normalize_alias` имени строки `entry_id` (строка
|
||
свода этой версии; иначе рекомендаций нет); рекомендуемые первыми, затем по имени; не больше `MAX_CANDIDATES`.
|
||
Запросов — константа: версия, периметр, (индекс контрагентов при `entry_id`), поиск с ГК, членства ГСК/ГСЗ.
|
||
"""
|
||
found = await repo.get_version(self._session, self._scope, version_id)
|
||
if found is None:
|
||
raise NotFoundError(VERSION_NOT_FOUND)
|
||
version, consolidation = found
|
||
query = (q or "").strip() or None
|
||
if scope == "all" and (query is None or len(query) < MIN_QUERY_ALL):
|
||
raise DomainValidationError(
|
||
"QUERY_TOO_SHORT", f"Для поиска по всему справочнику нужно не меньше {MIN_QUERY_ALL} символов"
|
||
)
|
||
members = {entity.id: _entity(entity) for _, entity in await repo.members(self._session, version.id)}
|
||
recommended = await self._recommended_ids(version.id, entry_id)
|
||
found_entities = await repo.search_entities(
|
||
self._session,
|
||
q=query,
|
||
group_id=consolidation.company_group_id if scope == "group" else None,
|
||
member_ids=list(members),
|
||
recommended_ids=recommended,
|
||
limit=MAX_CANDIDATES,
|
||
)
|
||
related = await repo.current_related_groups(self._session, [entity.id for entity, _ in found_entities])
|
||
result = []
|
||
for entity, group in found_entities:
|
||
groups = related.get(entity.id, {})
|
||
kind = domain.classify(
|
||
entity.id, members=members, entities={entity.id: _entity(entity)},
|
||
group_id=consolidation.company_group_id,
|
||
) # fmt: skip
|
||
result.append(
|
||
CandidateOut(
|
||
id=entity.id,
|
||
inn=entity.inn,
|
||
name=entity.name,
|
||
company_group=_ref(group),
|
||
gsk=_ref(groups.get(RelatedGroupKind.gsk)),
|
||
gsz=_ref(groups.get(RelatedGroupKind.gsz)),
|
||
counterparty_type=kind,
|
||
recommended=entity.id in recommended,
|
||
)
|
||
)
|
||
return result
|
||
|
||
async def _recommended_ids(self, version_id: uuid.UUID, entry_id: uuid.UUID | None) -> set[uuid.UUID]:
|
||
"""ЮЛ, чьё нормализованное имя или алиас равны `normalize_alias` имени строки `entry_id` (строка свода версии).
|
||
|
||
Нормализация — функция Python (§3 п. 4), поэтому совпадения ищутся тем же `CounterpartyIndex`, что резолвит
|
||
строки свода (`by_normalized` — имена и алиасы всего справочника без policy)."""
|
||
if entry_id is None:
|
||
return set()
|
||
entry = await repo.entry_in_version(self._session, entry_id, version_id)
|
||
if entry is None or not (normalized := normalize_alias(entry.counterparty_name or "")):
|
||
return set()
|
||
index = await CounterpartyIndex.load(self._session)
|
||
return set(index.by_normalized.get(normalized, ()))
|