"""ГРЛС stand-in: short excerpts of the official drug instruction, per section (SPEC §9, SPEC §7 п.5). Each `data/meds/grls/.yaml` holds one drug: inn: сертралин aliases: [sertraline] sections: side_effects: {text: "...", source: "ГРЛС, инструкция по медицинскому применению, раздел «...»"} Section codes are the ones `app/llm/med_qa.py` understands (side_effects, onset, alcohol, discontinuation, administration, interactions, overdose). The bot answers drug questions ONLY from these texts, so they are the single place to verify against the current ГРЛС instruction (see the header of each file). """ import logging from pathlib import Path import yaml from pydantic import BaseModel, Field, ValidationError from app.config import get_settings from app.integrations.base import DrugSection log = logging.getLogger(__name__) class _SectionFile(BaseModel): text: str source: str class _DrugFile(BaseModel): inn: str aliases: list[str] = Field(default_factory=list) sections: dict[str, _SectionFile] def default_grls_dir() -> Path: return get_settings().data_dir / "meds" / "grls" def normalize_inn(name: str) -> str: """Lookup key: «Сертралин 50 мг», «сертралин», «Sertraline» -> first word, lower case, ё -> е.""" words = name.strip().lower().replace("ё", "е").split() return words[0] if words else "" class GrlsExcerpts: """`DrugInfo` over local YAML excerpts. Files are read lazily on the first lookup and cached.""" def __init__(self, directory: Path | None = None) -> None: self._dir = directory self._index: dict[str, list[DrugSection]] | None = None def _load(self) -> dict[str, list[DrugSection]]: if self._index is not None: return self._index directory = self._dir or default_grls_dir() index: dict[str, list[DrugSection]] = {} for path in sorted(directory.glob("*.yaml")): try: drug = _DrugFile.model_validate(yaml.safe_load(path.read_text(encoding="utf-8"))) except (OSError, yaml.YAMLError, ValidationError) as exc: log.warning("grls file skipped file=%s error=%s", path.name, type(exc).__name__) continue sections = [ DrugSection(inn=drug.inn, section=code, text=" ".join(s.text.split()), source=s.source) for code, s in drug.sections.items() ] for key in (drug.inn, *drug.aliases): index[normalize_inn(key)] = sections self._index = index return index def known(self) -> list[str]: """INNs with excerpts (for the dashboard / tests).""" return sorted({s[0].inn for s in self._load().values() if s}) async def sections(self, inn: str) -> list[DrugSection]: return [s.model_copy() for s in self._load().get(normalize_inn(inn), [])]