feat(directories): алиасы ЮЛ с нормализацией §3

normalize_alias (lower, ё→е, без пунктуации и ОПФ), POST/DELETE
/directories/legal-entities/{id}/aliases, конфликты ALIAS_EXISTS /
ALIAS_TAKEN (с legal_entity_id владельца), EMPTY_ALIAS, аудит
LEGAL_ENTITY_CHANGE по legal_entity_alias.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev 2026-10-04 21:52:47 +04:00
parent e03f97946f
commit 7007c51f63
8 changed files with 485 additions and 2 deletions

View File

@ -8,6 +8,7 @@ from src.api.dependencies.audit import RequestAudit, get_request_audit
from src.api.dependencies.auth import Principal, current_user
from src.api.dependencies.db import SessionDep
from src.domain.policy import VisibilityScope, scope_of
from src.services.aliases import AliasService
from src.services.company_groups import CompanyGroupService
from src.services.legal_entities import LegalEntityService
@ -37,3 +38,14 @@ def get_legal_entity_service(
LegalEntityServiceDep = Annotated[LegalEntityService, Depends(get_legal_entity_service)]
def get_alias_service(
session: SessionDep,
audit: Annotated[RequestAudit, Depends(get_request_audit)],
scope: Annotated[VisibilityScope, Depends(get_scope)],
) -> AliasService:
return AliasService(session, audit, scope)
AliasServiceDep = Annotated[AliasService, Depends(get_alias_service)]

View File

@ -1,5 +1,5 @@
"""Маршруты ЮЛ/ИП `/directories/legal-entities` (spec directories-api): список, карточка, создание, изменение,
перевод в другую ГК, вывод из состава, история.
перевод в другую ГК, вывод из состава, история, алиасы.
Право (`require`) проверяется до видимости объекта; чужое и несуществующее ЮЛ — одинаковый 404 (spec directory-policy).
"""
@ -10,9 +10,11 @@ from typing import Annotated
from fastapi import APIRouter, Depends, Query, status
from src.api.dependencies.auth import require
from src.api.dependencies.directories import LegalEntityServiceDep
from src.api.dependencies.directories import AliasServiceDep, LegalEntityServiceDep
from src.schemas.common import Page, PageParams
from src.schemas.directories import (
AliasCreate,
AliasOut,
LegalEntityCreate,
LegalEntityDetail,
LegalEntityFilter,
@ -27,6 +29,7 @@ router = APIRouter(prefix="/legal-entities")
VIEW = [Depends(require("DIRECTORY_VIEW"))]
EDIT = [Depends(require("DIRECTORY_EDIT"))]
DELETE = [Depends(require("DIRECTORY_DELETE"))]
class LegalEntityListQuery(LegalEntityFilter, PageParams):
@ -99,3 +102,25 @@ async def withdraw_legal_entity(
)
async def legal_entity_history(entity_id: uuid.UUID, service: LegalEntityServiceDep) -> list[LegalEntityHistoryOut]:
return [LegalEntityHistoryOut.model_validate(row) for row in await service.history(entity_id)]
@router.post(
"/{entity_id}/aliases",
response_model=AliasOut,
status_code=status.HTTP_201_CREATED,
dependencies=EDIT,
summary="Добавить алиас ЮЛ/ИП",
)
async def add_alias(entity_id: uuid.UUID, data: AliasCreate, service: AliasServiceDep) -> AliasOut:
"""Сохраняет `alias` как есть и `normalized_alias` по §3; повтор — 409 `ALIAS_EXISTS`, чужой — `ALIAS_TAKEN`."""
return AliasOut.model_validate(await service.add(entity_id, data))
@router.delete(
"/{entity_id}/aliases/{alias_id}",
status_code=status.HTTP_204_NO_CONTENT,
dependencies=DELETE,
summary="Удалить алиас ЮЛ/ИП",
)
async def delete_alias(entity_id: uuid.UUID, alias_id: uuid.UUID, service: AliasServiceDep) -> None:
await service.remove(entity_id, alias_id)

41
api/src/domain/aliases.py Normal file
View File

@ -0,0 +1,41 @@
"""Нормализация алиаса ЮЛ (§3 п. 4, ANL-028): ключ сопоставления контрагента из ОСВ с ЮЛ по тексту названия.
Чистая функция без обращения к БД; одно и то же название в разных написаниях (`ООО «Пекарня»`, `Пекарня, ООО`)
даёт один и тот же `normalized_alias`.
"""
import re
# Одиночные токены ОПФ (§3 п. 4: «ООО, АО, ПАО, ИП…»); убираются только целиком — «ооо» внутри слова не трогается.
LEGAL_FORMS: frozenset[str] = frozenset(
{
"ооо",
"ао",
"пао",
"зао",
"оао",
"нао",
"ип",
"гуп",
"муп",
"фгуп",
"ано",
"тсж",
"кфх",
"пк",
"спк",
"сх",
"сха",
}
)
_PUNCT_RE = re.compile(r"[^\w\s]", re.UNICODE) # кавычки «»"'„“ и прочая пунктуация → пробел
def normalize_alias(raw: str) -> str:
"""§3: lower → ё→е → пунктуация и кавычки → пробел → убрать токены ОПФ → схлопнуть пробелы → strip.
Строка из одних ОПФ и пунктуации нормализуется в `""` — вызывающий отвергает такой алиас.
"""
text = _PUNCT_RE.sub(" ", raw.lower().replace("ё", "е"))
return " ".join(token for token in text.split() if token not in LEGAL_FORMS)

View File

@ -244,6 +244,25 @@ async def list_aliases(session: AsyncSession, entity_id: uuid.UUID) -> list[Lega
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)

View File

@ -236,6 +236,12 @@ class MembershipOut(BaseModel):
valid_from: datetime.datetime
class AliasCreate(BaseModel):
"""Тело `POST …/aliases`: `alias` сохраняется как есть (как в ОСВ), `normalized_alias` считает сервис (§3)."""
alias: str = Field(min_length=1, max_length=NAME_MAX_LENGTH)
class AliasOut(BaseModel):
"""Алиас ЮЛ: `alias` — как в ОСВ, `normalized_alias` — по §3."""

View File

@ -0,0 +1,95 @@
"""Сервис алиасов ЮЛ (spec directories-api «Алиасы ЮЛ», audit-trail, §3 п. 4).
Порядок: право (`require` на маршруте) → видимость ЮЛ (`scope`, невидимое = несуществующее) → проверки → запись.
Аудит `LEGAL_ENTITY_CHANGE` / `object_type = legal_entity_alias` пишется в той же транзакции запроса.
"""
import datetime
import uuid
from sqlalchemy.ext.asyncio import AsyncSession
from src.api.dependencies.audit import RequestAudit
from src.common.exceptions import ConflictError, DomainValidationError, NotFoundError
from src.domain.aliases import normalize_alias
from src.domain.policy import VisibilityScope
from src.models.directories import LegalEntity, LegalEntityAlias
from src.models.enums import AuditAction, AuditSection
from src.repositories import directories as repo
from src.schemas.directories import AliasCreate
from src.services.legal_entities import LEGAL_ENTITY_NOT_FOUND
OBJECT_TYPE = "legal_entity_alias"
ALIAS_NOT_FOUND = "Алиас не найден"
class AliasService:
"""Добавление и удаление алиасов ЮЛ/ИП от имени текущего пользователя в его области видимости."""
def __init__(self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope) -> None:
self._session = session
self._audit = audit
self._scope = scope
async def add(self, entity_id: uuid.UUID, data: AliasCreate) -> LegalEntityAlias:
"""Новый алиас: `alias` как есть, `normalized_alias` по §3; конфликты — `ALIAS_EXISTS` / `ALIAS_TAKEN`."""
entity = await self._get_entity(entity_id)
normalized = normalize_alias(data.alias)
if not normalized:
raise DomainValidationError("EMPTY_ALIAS", "Алиас пуст после нормализации (одни ОПФ и знаки препинания)")
owner = await repo.find_alias_owner(self._session, normalized)
if owner is not None:
if owner.legal_entity_id == entity.id:
raise ConflictError("ALIAS_EXISTS", "У этого ЮЛ уже есть такой алиас")
# Владелец может быть вне видимости пользователя: спека требует `legal_entity_id`, остальное не раскрываем
raise ConflictError(
"ALIAS_TAKEN", "Такой алиас уже закреплён за другим ЮЛ", legal_entity_id=owner.legal_entity_id
)
alias = LegalEntityAlias(
legal_entity_id=entity.id,
alias=data.alias,
normalized_alias=normalized,
created_by=self._audit.user_id,
created_at=datetime.datetime.now(datetime.UTC),
)
self._session.add(alias)
await self._session.flush()
await self._audit.record(
action=AuditAction.LEGAL_ENTITY_CHANGE,
section=AuditSection.DIRECTORIES,
object_type=OBJECT_TYPE,
object_id=alias.id,
after_value=self._payload(alias),
)
return alias
async def remove(self, entity_id: uuid.UUID, alias_id: uuid.UUID) -> None:
"""Удаляет алиас этого ЮЛ (иначе 404); `before_value` аудита — удалённые значения."""
entity = await self._get_entity(entity_id)
alias = await repo.get_alias(self._session, entity.id, alias_id)
if alias is None:
raise NotFoundError(ALIAS_NOT_FOUND)
before = self._payload(alias)
await self._session.delete(alias)
await self._session.flush()
await self._audit.record(
action=AuditAction.LEGAL_ENTITY_CHANGE,
section=AuditSection.DIRECTORIES,
object_type=OBJECT_TYPE,
object_id=alias_id,
before_value=before,
)
async def _get_entity(self, entity_id: uuid.UUID) -> LegalEntity:
entity = await repo.get_legal_entity(self._session, self._scope, entity_id)
if entity is None:
raise NotFoundError(LEGAL_ENTITY_NOT_FOUND)
return entity
@staticmethod
def _payload(alias: LegalEntityAlias) -> dict[str, object]:
return {
"legal_entity_id": alias.legal_entity_id,
"alias": alias.alias,
"normalized_alias": alias.normalized_alias,
}

View File

@ -0,0 +1,233 @@
"""API алиасов ЮЛ (spec directories-api «Алиасы ЮЛ», audit-trail, directory-policy)."""
import uuid
from collections.abc import Sequence
from typing import Any
import httpx
import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.audit_ex import AuditEvent
from src.models.directories import LegalEntityAlias, RegionalOffice
from src.models.enums import AuditAction, AuditSection
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group, make_legal_entity
HTTP_OK = 200
HTTP_CREATED = 201
HTTP_NO_CONTENT = 204
HTTP_FORBIDDEN = 403
HTTP_NOT_FOUND = 404
HTTP_CONFLICT = 409
HTTP_UNPROCESSABLE = 422
ENTITIES = "/directories/legal-entities"
OBJECT_TYPE = "legal_entity_alias"
KRASNODAR = "Краснодарский региональный филиал"
async def _events(session: AsyncSession) -> Sequence[AuditEvent]:
stmt = (
select(AuditEvent).where(AuditEvent.object_type == OBJECT_TYPE).order_by(AuditEvent.occurred_at, AuditEvent.id)
)
return (await session.execute(stmt)).scalars().all()
async def _alias_count(session: AsyncSession) -> int:
return (await session.execute(select(func.count()).select_from(LegalEntityAlias))).scalar_one()
async def _add(client: httpx.AsyncClient, entity_id: Any, alias: str) -> httpx.Response:
return await client.post(f"{ENTITIES}/{entity_id}/aliases", json={"alias": alias})
@pytest.mark.integration
async def test_alias_is_normalized_and_repeat_is_conflict(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
"""Сценарий спеки «Нормализация»."""
entity = await make_legal_entity(seeded, name="ООО Пекарня")
client = await auth_client("ka_go@armka.local")
created = await _add(client, entity.id, "ООО «Пекарня Счастливочная»")
assert created.status_code == HTTP_CREATED, created.text
body = created.json()
assert (body["alias"], body["normalized_alias"]) == ("ООО «Пекарня Счастливочная»", "пекарня счастливочная")
assert body["created_at"]
repeat = await _add(client, entity.id, "Пекарня Счастливочная, ООО")
assert repeat.status_code == HTTP_CONFLICT
assert repeat.json()["detail"]["code"] == "ALIAS_EXISTS"
assert await _alias_count(seeded) == 1
assert len(await _events(seeded)) == 1 # неудачная попытка аудита не пишет
card = (await client.get(f"{ENTITIES}/{entity.id}")).json()
assert [alias["id"] for alias in card["aliases"]] == [body["id"]]
@pytest.mark.integration
async def test_create_alias_audit_and_created_by(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
entity = await make_legal_entity(seeded)
client = await auth_client("ka_go@armka.local")
body = (await _add(client, entity.id, 'АО "РУССКИЙ БИСКВИТ"')).json()
(event,) = await _events(seeded)
assert event.action == AuditAction.LEGAL_ENTITY_CHANGE
assert (event.section, str(event.object_id)) == (AuditSection.DIRECTORIES, body["id"])
assert event.before_value is None
assert event.after_value == {
"legal_entity_id": str(entity.id),
"alias": 'АО "РУССКИЙ БИСКВИТ"',
"normalized_alias": "русский бисквит",
}
row = (await seeded.execute(select(LegalEntityAlias))).scalar_one()
assert row.created_by == event.user_id
assert row.created_by is not None
@pytest.mark.integration
async def test_alias_of_another_entity_is_taken_with_owner_id_only(
auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
owner = await make_legal_entity(seeded, name="ООО Владелец")
other = await make_legal_entity(seeded, name="ООО Другое")
client = await auth_client("ka_go@armka.local")
assert (await _add(client, owner.id, "ООО «Пекарня»")).status_code == HTTP_CREATED
response = await _add(client, other.id, "пекарня, ао")
assert response.status_code == HTTP_CONFLICT
detail = response.json()["detail"]
assert detail["code"] == "ALIAS_TAKEN"
assert detail["legal_entity_id"] == str(owner.id)
assert set(detail) == {"code", "message", "legal_entity_id"} # ни имени, ни ИНН владельца
assert "Владелец" not in response.text
assert owner.inn not in response.text
assert await _alias_count(seeded) == 1
assert len(await _events(seeded)) == 1
@pytest.mark.integration
async def test_taken_alias_of_invisible_owner_discloses_only_id(
auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
"""Владелец в чужом филиале: пользователь филиала получает только его `legal_entity_id` (решение по спеке)."""
krasnodar = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name == KRASNODAR))).scalar_one()
foreign_group = await make_company_group(seeded, name="ГК Краснодар", office=krasnodar)
foreign = await make_legal_entity(seeded, name="ООО Чужое", group=foreign_group)
go_client = await auth_client("ka_go@armka.local")
assert (await _add(go_client, foreign.id, "Пекарня")).status_code == HTTP_CREATED
rostov = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name.like("Ростов%")))).scalar_one()
own_group = await make_company_group(seeded, name="ГК Ростов", office=rostov)
own = await make_legal_entity(seeded, name="ООО Своё", group=own_group)
rostov_client = await auth_client("ka_rf@armka.local")
response = await _add(rostov_client, own.id, "ООО Пекарня")
assert response.status_code == HTTP_CONFLICT
assert response.json()["detail"] == {
"code": "ALIAS_TAKEN",
"message": response.json()["detail"]["message"],
"legal_entity_id": str(foreign.id),
}
assert "Чужое" not in response.text
@pytest.mark.integration
async def test_empty_normalization_is_422(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
entity = await make_legal_entity(seeded)
client = await auth_client("ka_go@armka.local")
for alias in ("ООО", "«»", " "):
response = await _add(client, entity.id, alias)
assert response.status_code == HTTP_UNPROCESSABLE
assert response.json()["detail"]["code"] == "EMPTY_ALIAS"
assert await _alias_count(seeded) == 0
assert not await _events(seeded)
@pytest.mark.integration
async def test_alias_body_validation(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
entity = await make_legal_entity(seeded)
client = await auth_client("ka_go@armka.local")
assert (await _add(client, entity.id, "")).status_code == HTTP_UNPROCESSABLE
assert (await _add(client, entity.id, "а" * 501)).status_code == HTTP_UNPROCESSABLE
assert (await client.post(f"{ENTITIES}/{entity.id}/aliases", json={})).status_code == HTTP_UNPROCESSABLE
assert await _alias_count(seeded) == 0
@pytest.mark.integration
async def test_delete_alias_writes_audit_with_before_value(
auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
entity = await make_legal_entity(seeded)
editor = await auth_client("ka_go@armka.local")
alias_id = (await _add(editor, entity.id, "ООО «Пекарня»")).json()["id"]
# DIRECTORY_DELETE есть только у администратора; редактор справочников получает 403 и алиас остаётся
forbidden = await editor.delete(f"{ENTITIES}/{entity.id}/aliases/{alias_id}")
assert forbidden.status_code == HTTP_FORBIDDEN
assert await _alias_count(seeded) == 1
admin = await auth_client("admin@armka.local")
response = await admin.delete(f"{ENTITIES}/{entity.id}/aliases/{alias_id}")
assert response.status_code == HTTP_NO_CONTENT
assert response.content == b""
assert await _alias_count(seeded) == 0
created, deleted = await _events(seeded)
assert deleted.action == AuditAction.LEGAL_ENTITY_CHANGE
assert (deleted.section, str(deleted.object_id)) == (AuditSection.DIRECTORIES, alias_id)
assert deleted.before_value == {
"legal_entity_id": str(entity.id),
"alias": "ООО «Пекарня»",
"normalized_alias": "пекарня",
}
assert deleted.after_value is None
assert created.object_id == deleted.object_id # запись о создании остаётся (append-only)
# освободившийся алиас можно добавить снова
assert (await _add(editor, entity.id, "Пекарня")).status_code == HTTP_CREATED
@pytest.mark.integration
async def test_delete_unknown_or_foreign_alias_is_404(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
first = await make_legal_entity(seeded)
second = await make_legal_entity(seeded)
editor = await auth_client("ka_go@armka.local")
alias_id = (await _add(editor, first.id, "Первое")).json()["id"]
admin = await auth_client("admin@armka.local")
for entity_id, target in ((second.id, alias_id), (first.id, uuid.uuid4()), (uuid.uuid4(), alias_id)):
response = await admin.delete(f"{ENTITIES}/{entity_id}/aliases/{target}")
assert response.status_code == HTTP_NOT_FOUND
assert await _alias_count(seeded) == 1
assert len(await _events(seeded)) == 1
@pytest.mark.integration
async def test_invisible_entity_aliases_are_404_like_missing(
auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
krasnodar = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name == KRASNODAR))).scalar_one()
foreign_group = await make_company_group(seeded, name="ГК Краснодар", office=krasnodar)
foreign = await make_legal_entity(seeded, group=foreign_group)
go_client = await auth_client("ka_go@armka.local")
alias_id = (await _add(go_client, foreign.id, "Чужой алиас")).json()["id"]
client = await auth_client("ka_rf@armka.local")
missing = await _add(client, uuid.uuid4(), "Что угодно")
assert missing.status_code == HTTP_NOT_FOUND
response = await _add(client, foreign.id, "Что угодно")
assert response.status_code == HTTP_NOT_FOUND
assert response.json() == missing.json()
assert await _alias_count(seeded) == 1
manager = await auth_client("client_manager@armka.local") # Краснодар видит ЮЛ, но права DELETE нет
assert (await manager.delete(f"{ENTITIES}/{foreign.id}/aliases/{alias_id}")).status_code == HTTP_FORBIDDEN
@pytest.mark.integration
async def test_expert_cannot_add_alias(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
entity = await make_legal_entity(seeded)
client = await auth_client("expert@armka.local")
response = await _add(client, entity.id, "Пекарня")
assert response.status_code == HTTP_FORBIDDEN
assert await _alias_count(seeded) == 0
assert not await _events(seeded)

View File

@ -0,0 +1,52 @@
"""Нормализация алиаса ЮЛ (§3 п. 4): lower, ё→е, без пунктуации/кавычек/ОПФ, схлопнутые пробелы."""
import pytest
from src.domain.aliases import LEGAL_FORMS, normalize_alias
@pytest.mark.parametrize(
("raw", "expected"),
[
("ООО «Пекарня Счастливочная»", "пекарня счастливочная"),
("Пекарня Счастливочная, ООО", "пекарня счастливочная"),
('АО "РУССКИЙ БИСКВИТ"', "русский бисквит"),
("ИП Иванов И.И.", "иванов и и"),
("Пекарня Счастливочная ООО", "пекарня счастливочная"),
(" Пекарня «Счастливочная» ", "пекарня счастливочная"),
("Ёлка и Ёж", "елка и еж"),
("ЁЛКА", "елка"),
("„Пекарня“ ПАО", "пекарня"),
("Пекарня\tСчастливочная\n", "пекарня счастливочная"),
],
)
def test_normalize_alias(raw: str, expected: str) -> None:
assert normalize_alias(raw) == expected
@pytest.mark.parametrize("raw", ["ООО", "ООО «АО»", " ип, ооо ", "«»", "", " "])
def test_only_legal_forms_or_punctuation_gives_empty(raw: str) -> None:
assert normalize_alias(raw) == ""
@pytest.mark.parametrize(
("raw", "expected"),
[("Плооо", "плооо"), ("Оооптторг", "оооптторг"), ("Саоп ао", "саоп"), ("ТСЖ Ипподром", "ипподром")],
)
def test_legal_form_is_removed_only_as_whole_token(raw: str, expected: str) -> None:
assert normalize_alias(raw) == expected
def test_same_name_in_different_spelling_is_equal() -> None:
forms = [
"ООО «Пекарня Счастливочная»",
"Пекарня Счастливочная, ООО",
"пекарня счастливочная",
"ООО Пекарня Счастливочная",
]
assert len({normalize_alias(form) for form in forms}) == 1
def test_legal_forms_are_lowercase_tokens() -> None:
assert {"ооо", "ао", "пао", "зао", "ип"} <= LEGAL_FORMS
assert all(form == form.lower() and " " not in form for form in LEGAL_FORMS)