CI / ruff · pytest (pull_request) Successful in 1m26s
- setup() at package import (idempotent, replaces only base.py defaults): MisMock, ConnectCodeIdentity, GrlsExcerpts, CompositeNotifier; outbox stays MemoryOutbox - MisMock: clinical DB prescriptions, fallback data/meds/mis_prescriptions.json (synthetic, by token) - ConnectCodeIdentity: one-time connect code via identity_repo, 15 min grace for the double check in one connect flow (Telegram /start + engine `new` state) - GrlsExcerpts: synthetic paraphrased instruction excerpts for sertraline, paroxetine, escitalopram, fluoxetine, mirtazapine, prazosin - CompositeNotifier: in-memory feed (recent(since_id)) + Telegram doctor chat, PII-free, never raises - test_llm_med_qa: default-registry test now patches NoDrugInfo explicitly Refs #14 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
"""ГРЛС stand-in: short excerpts of the official drug instruction, per section (SPEC §9, SPEC §7 п.5).
|
||
|
||
Each `data/meds/grls/<name>.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), [])]
|