143 lines
7.2 KiB
Python
143 lines
7.2 KiB
Python
"""Локальное, детерминированное редактирование учебных сценариев.
|
||
|
||
Это копирование утверждённого шаблона, а не генерация нового сюжета ИИ.
|
||
Никаких сетевых вызовов в этом модуле нет.
|
||
"""
|
||
|
||
from copy import deepcopy
|
||
import re
|
||
|
||
from pydantic import ValidationError
|
||
|
||
from app.scenarios.loader import ScenarioError, _derive_ground_truth
|
||
from app.scenarios.schema import Scenario
|
||
|
||
DERIVED_TRUTH = {"incident_type", "dds", "incident_code", "notify", "required_facts"}
|
||
PLACEHOLDER = re.compile(r"\{\{?\s*[a-zA-Z_][\w.-]*\s*\}?\}")
|
||
|
||
|
||
def classifier_sign_gaps(signs: list[str], narrative: list[str]) -> list[str]:
|
||
"""Найти признаки ЕКП, которые не подтверждены текстом сценария.
|
||
|
||
Сравниваются трёхбуквенные основы, поэтому русские окончания
|
||
(``жилой дом`` / ``жилого дома``) не создают ложного расхождения. Это
|
||
намеренно строгая, детерминированная граница: модель не может заявить
|
||
один объект, а в карточку незаметно положить код другого.
|
||
"""
|
||
def stems(value: str) -> set[str]:
|
||
return {
|
||
word[:3]
|
||
for word in re.findall(r"[0-9a-zа-яё]+", value.casefold())
|
||
if len(word) >= 3
|
||
}
|
||
|
||
present = stems(" ".join(narrative))
|
||
gaps: list[str] = []
|
||
for sign in signs:
|
||
required = stems(sign)
|
||
if required and not required <= present:
|
||
gaps.append(sign)
|
||
return gaps
|
||
|
||
|
||
def template_copy(source: Scenario, draft_id: str, title: str | None = None) -> dict:
|
||
"""Сделать полностью самостоятельный шаблонный черновик из сценария."""
|
||
body = source.model_dump(mode="json")
|
||
body["id"] = draft_id
|
||
body["title"] = title or f"{source.title} — копия"
|
||
body["extends"] = None # чек-лист уже развёрнут загрузчиком
|
||
body["ticket"] = None
|
||
body["position"] = None
|
||
for key in DERIVED_TRUTH:
|
||
body["ground_truth"].pop(key, None)
|
||
return body
|
||
|
||
|
||
def merge_patch(body: dict, patch: dict) -> dict:
|
||
"""JSON Merge Patch: словари сливаются, списки заменяются целиком.
|
||
|
||
`id` — идентичность записи, а вычисляемые поля эталона не принимаются от
|
||
клиента. Некорректные обычные поля можно временно сохранить в черновике;
|
||
утверждение всё равно потребует полной проверки.
|
||
"""
|
||
if "id" in patch and patch["id"] != body["id"]:
|
||
raise ScenarioError("id сценария менять нельзя")
|
||
truth = patch.get("ground_truth")
|
||
if isinstance(truth, dict) and DERIVED_TRUTH & truth.keys():
|
||
raise ScenarioError("вычисляемые поля ground_truth нельзя задавать вручную")
|
||
|
||
result = deepcopy(body)
|
||
|
||
def apply(target: dict, changes: dict) -> None:
|
||
for key, value in changes.items():
|
||
if value is None:
|
||
target.pop(key, None)
|
||
elif isinstance(value, dict) and isinstance(target.get(key), dict):
|
||
apply(target[key], value)
|
||
else:
|
||
target[key] = deepcopy(value)
|
||
|
||
apply(result, patch)
|
||
result["id"] = body["id"]
|
||
return result
|
||
|
||
|
||
def validate(body: dict) -> Scenario:
|
||
"""Проверить схему и связи фактов, затем пересчитать эталон по ЕКП."""
|
||
declared = body.get("ground_truth") or {}
|
||
if not isinstance(declared, dict):
|
||
raise ScenarioError("ground_truth: ожидается объект")
|
||
forbidden = DERIVED_TRUTH & declared.keys()
|
||
if forbidden:
|
||
raise ScenarioError(
|
||
f"{', '.join(sorted(forbidden))} в ground_truth выводится кодом и руками не пишется"
|
||
)
|
||
try:
|
||
scenario = Scenario.model_validate(body)
|
||
except ValidationError as exc:
|
||
first = exc.errors()[0]
|
||
where = ".".join(str(part) for part in first["loc"])
|
||
raise ScenarioError(f"{where}: {first['msg']}") from exc
|
||
|
||
if not scenario.title.strip() or len(scenario.title) > 200:
|
||
raise ScenarioError("title: требуется название длиной от 1 до 200 символов")
|
||
if not scenario.first_line.strip():
|
||
raise ScenarioError("first_line: требуется первая реплика")
|
||
if not scenario.facts or not scenario.checklist:
|
||
raise ScenarioError("сценарию нужны факты и чек-лист для оценки занятия")
|
||
|
||
known = scenario.fact_ids()
|
||
question_ids = {item.id for item in scenario.checklist}
|
||
if len(known) != len(scenario.facts):
|
||
raise ScenarioError("id фактов должны быть уникальны")
|
||
if len(question_ids) != len(scenario.checklist):
|
||
raise ScenarioError("id пунктов чек-листа должны быть уникальны")
|
||
for item in scenario.checklist:
|
||
if not item.question:
|
||
raise ScenarioError(f"у пункта {item.id} нет текста вопроса")
|
||
if item.fact and item.fact not in known:
|
||
raise ScenarioError(f"пункт {item.id} ссылается на факт {item.fact}, которого нет")
|
||
for fact in scenario.facts:
|
||
if PLACEHOLDER.search(fact.value) or (fact.refined and PLACEHOLDER.search(fact.refined)):
|
||
raise ScenarioError(
|
||
f"факт {fact.id}: остался плейсхолдер вместо конкретного ответа заявителя"
|
||
)
|
||
question_id = fact.reveal_on.question if fact.reveal_on else None
|
||
if question_id and question_id not in question_ids:
|
||
raise ScenarioError(f"факт {fact.id} раскрывается несуществующим вопросом {question_id}")
|
||
if fact.refine_on and fact.refine_on not in question_ids:
|
||
raise ScenarioError(f"факт {fact.id} уточняется несуществующим вопросом {fact.refine_on}")
|
||
# Для старых карточек и безопасных вариаций классификатор уже утверждён
|
||
# заказчиком. Новый свободный сюжет обязан доказать свой путь ЕКП текстом.
|
||
if scenario.id.startswith("ai-full-"):
|
||
gaps = classifier_sign_gaps(
|
||
scenario.signs,
|
||
[scenario.title, scenario.first_line]
|
||
+ [fact.value for fact in scenario.facts]
|
||
+ [fact.refined for fact in scenario.facts if fact.refined],
|
||
)
|
||
if gaps:
|
||
raise ScenarioError(
|
||
"signs: сюжет не подтверждает путь ЕКП: " + ", ".join(gaps)
|
||
)
|
||
return _derive_ground_truth(scenario)
|