"""Deterministic drug-safety lookup: composition, contraindications, interactions, patient-specific findings. The LLM never decides whether a drug is safe for the patient (ADR-007 spirit): it calls `check_patient` and retells the findings. Matching is by tags shared with the ЕМИАС record (`data/emias/records.yaml`): condition tags → `cautions[].when_condition`, allergy tags → `composition.excipient_tags`, other medication classes → `interactions[].with_class`. Source data: `data/meds/drug_safety.yaml` (prototype retelling of ГРЛС). """ from functools import lru_cache from pathlib import Path from typing import Literal import yaml from pydantic import BaseModel, Field from app.config import get_settings from app.integrations.emias_record import PatientRecord Severity = Literal["high", "medium", "low"] class Caution(BaseModel): when_condition: str severity: Severity text: str class Interaction(BaseModel): with_class: str severity: Severity text: str class Composition(BaseModel): form: str active: str excipients: list[str] = Field(default_factory=list) excipient_tags: list[str] = Field(default_factory=list) note: str = "" class DrugEntry(BaseModel): inn: str aliases: list[str] = Field(default_factory=list) drug_class: str = "" classes: list[str] = Field(default_factory=list) # classes this drug belongs to (e.g. nsaid, ssri) contraindications: list[str] = Field(default_factory=list) cautions: list[Caution] = Field(default_factory=list) interactions: list[Interaction] = Field(default_factory=list) composition: Composition | None = None pregnancy: str = "" source: str = "" class Finding(BaseModel): severity: Severity kind: Literal["condition", "allergy", "interaction", "pregnancy", "contraindication"] text: str class SafetyDb(BaseModel): drugs: dict[str, DrugEntry] allergy_labels: dict[str, str] = Field(default_factory=dict) def _path() -> Path: return get_settings().data_dir / "meds" / "drug_safety.yaml" @lru_cache def load_db(path: Path | None = None) -> SafetyDb: raw = yaml.safe_load((path or _path()).read_text(encoding="utf-8")) classes_of: dict[str, list[str]] = {} for cls, members in (raw.get("drug_classes") or {}).items(): for m in members: classes_of.setdefault(m, []).append(cls) drugs: dict[str, DrugEntry] = {} for inn, d in (raw.get("drugs") or {}).items(): entry = DrugEntry( inn=inn, aliases=d.get("aliases", []), drug_class=d.get("class", ""), classes=sorted(set(d.get("classes", [])) | set(classes_of.get(inn, []))), contraindications=d.get("contraindications", []), cautions=d.get("cautions", []), interactions=d.get("interactions", []), composition=d.get("composition"), pregnancy=d.get("pregnancy", ""), source=d.get("source", ""), ) drugs[inn] = entry return SafetyDb(drugs=drugs, allergy_labels=raw.get("allergy_labels") or {}) def find_drug(name: str, db: SafetyDb | None = None) -> DrugEntry | None: """By INN or alias, case-insensitive; "сертралин 50 мг" → сертралин.""" db = db or load_db() words = [w.strip(".,!?;:«»()") for w in name.lower().replace("ё", "е").split()] for entry in db.drugs.values(): names = {entry.inn, *(a.lower() for a in entry.aliases)} # Russian case endings: "сертралине", "ибупрофена" — match on the stem (name minus the last letter). if any(w in names or any(len(n) > 4 and w.startswith(n[:-1]) for n in names) for w in words): return entry return None def check_patient(drug: DrugEntry, record: PatientRecord, db: SafetyDb | None = None) -> list[Finding]: """Patient-specific findings for `drug`, most severe first. Empty list = nothing found in the reference — which is NOT a statement that the drug is safe (the answer must say so).""" db = db or load_db() out: list[Finding] = [] cond_tags = {t for c in record.conditions for t in c.tags} for c in drug.cautions: if c.when_condition in cond_tags: out.append(Finding(severity=c.severity, kind="condition", text=c.text)) # Excipients vs allergies / intolerances. allergy_tags = {t for a in record.allergies for t in a.tags} if drug.composition is not None: for tag in sorted(allergy_tags & set(drug.composition.excipient_tags)): label = db.allergy_labels.get(tag, tag) out.append( Finding( severity="high", kind="allergy", text=f"В составе есть {label}, а в карте отмечена непереносимость." ) ) # Interactions with the patient's other medicines and with the current psychiatric prescription. other_classes: dict[str, str] = {} for m in record.other_medications: for cls in m.classes: other_classes[cls] = m.drug for rx in record.current_prescriptions: entry = find_drug(rx.drug, db) if entry is not None and entry.inn != drug.inn: for cls in entry.classes: other_classes.setdefault(cls, rx.drug) for i in drug.interactions: if i.with_class in other_classes: out.append( Finding( severity=i.severity, kind="interaction", text=f"{i.text} (в карте: {other_classes[i.with_class]})." ) ) if record.pregnancy and drug.pregnancy: out.append(Finding(severity="medium", kind="pregnancy", text=drug.pregnancy)) order = {"high": 0, "medium": 1, "low": 2} out.sort(key=lambda f: order[f.severity]) return out