fix: отказ kio.patch возвращает серверные значения, каталог служб проверяется на входе, без дублей с notify

This commit is contained in:
GGlamer 2026-09-27 22:23:15 +03:00
commit f9d2c2cc77
5 changed files with 144 additions and 35 deletions

View file

@ -10,7 +10,7 @@ from enum import StrEnum
from typing import Any
from uuid import UUID, uuid4
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import BaseModel, ConfigDict, Field
from app.domain.classifiers import DDSCode, IncidentType
@ -119,7 +119,9 @@ class KIO(BaseModel):
# в боевом АРМ оператор может добавить службу вручную, но не удалить.
notify: list[str] = Field(default_factory=list, description="Список оповещения")
#: Службы, добавленные оператором вручную из каталога. Автоматический список
#: не заменяют и в оценку по эталону не входят.
#: не заменяют и в оценку по эталону не входят. Каталог проверяется на входе
#: (apply_patch), а не здесь: иначе правка названия в каталоге сломала бы
#: загрузку сохранённых снимков занятий и сданных карточек.
notify_extra: list[str] = Field(
default_factory=list, max_length=20, description="Дополнительно оповестить"
)
@ -135,19 +137,6 @@ class KIO(BaseModel):
medical: MedicalDetails | None = None
utility: UtilityDetails | None = None
@field_validator("notify_extra")
@classmethod
def known_catalog_services(cls, names: list[str]) -> list[str]:
"""Только службы каталога, без повторов: свободный текст в адресатах
карточки на АРМ ДДС выглядел бы как настоящая служба."""
from app.domain import ekp
known = ekp.catalog_names()
unknown = [name for name in names if name not in known]
if unknown:
raise ValueError(f"нет в каталоге служб: {', '.join(unknown[:3])}")
return list(dict.fromkeys(names))
#: Поля, которые курсант не редактирует: их проставляет система.
READ_ONLY_FIELDS: frozenset[str] = frozenset(
@ -241,11 +230,40 @@ def derive_incident(card: KIO) -> KIO:
return KIO.model_validate(data)
class PatchRejected(ValueError):
"""Дельта `kio.patch` не принята целиком: карточка не изменилась."""
def _check_catalog(names: Any) -> None:
"""Только службы каталога: свободный текст в адресатах карточки
на АРМ ДДС выглядел бы как настоящая служба."""
from app.domain import ekp
if not isinstance(names, list):
return # тип и лимит проверит модель
known = ekp.catalog_names()
unknown = [name for name in names if name not in known]
if unknown:
raise PatchRejected(f"нет в каталоге служб: {', '.join(map(str, unknown[:3]))}")
def _without_auto_services(card: KIO) -> KIO:
"""Добавки без повторов и без служб, которые уже есть в списке оповещения:
иначе на АРМ ДДС одна служба окажется в двух строках."""
extra = [name for name in dict.fromkeys(card.notify_extra) if name not in card.notify]
if extra == card.notify_extra:
return card
return card.model_copy(update={"notify_extra": extra})
def apply_patch(card: KIO, fields: dict[str, Any]) -> KIO:
"""Применить дельту `kio.patch`. Служебные поля игнорируются.
Вложенные поля приходят плоским путём: {"fire.floors": 5}.
Недопустимое значение — `PatchRejected` или `ValidationError`.
"""
if "notify_extra" in fields:
_check_catalog(fields["notify_extra"])
data = card.model_dump()
for path, value in fields.items():
if path in READ_ONLY_FIELDS:
@ -261,4 +279,4 @@ def apply_patch(card: KIO, fields: dict[str, Any]) -> KIO:
# Пересчитываем не только на смену признаков: пострадавшие поднимают скорую,
# газификация — МОСГАЗ, и список оповещения обязан это отразить сразу.
touched = MODIFIER_FIELDS & set(fields)
return derive_incident(updated) if touched else updated
return _without_auto_services(derive_incident(updated) if touched else updated)