Merge pull request 'feat(integrations): адаптеры МИС, код подключения, выдержки ГРЛС, уведомления врача' (#67) from feat/14-integrations into dev

app/integrations/__init__.py

@@ -0,0 +1,51 @@
+"""Demo adapters behind the frozen integration Protocols (SPEC §9, #14).
+
+`setup()` runs when the package is first imported. Every consumer imports `app.integrations.base`, which
+initialises this package first, and `app/main.py` is a frozen contract — so the registry is wired before
+any request or job without a startup hook. Setup only constructs objects: data files and DB sessions are
+opened lazily on the first call, so importing the package never fails on I/O.
+
+Registry after setup:
+- `prescriptions` — `MisMock`: clinical DB `prescriptions`, then `data/meds/mis_prescriptions.json`;
+- `identity` — `ConnectCodeIdentity`: one-time connect code from the doctor (ЕСИА stand-in);
+- `drug_info` — `GrlsExcerpts`: `data/meds/grls/*.yaml`;
+- `notifier` — `CompositeNotifier`: in-memory dashboard feed + optional Telegram doctor chat;
+- `outbox` — unchanged `MemoryOutbox` (the web console drains it; the Telegram channel wraps it at start).
+"""
+
+from app.integrations.base import (
+    EmptyPrescriptions,
+    MemoryNotifier,
+    NoDrugInfo,
+    RejectIdentity,
+    integrations,
+)
+from app.integrations.connect_code import ConnectCodeIdentity
+from app.integrations.grls import GrlsExcerpts
+from app.integrations.mis_mock import MisMock
+from app.integrations.notify import CompositeNotifier, NotificationFeed
+
+__all__ = ["CompositeNotifier", "ConnectCodeIdentity", "GrlsExcerpts", "MisMock", "NotificationFeed", "setup"]
+
+_done = False
+
+
+def setup(*, force: bool = False) -> None:
+    """Register the demo adapters. Idempotent; never replaces an adapter someone already swapped in
+    (a test fake, the Telegram outbox), only the in-memory defaults from `base.py` — unless `force`."""
+    global _done
+    if _done and not force:
+        return
+    if force or isinstance(integrations.prescriptions, EmptyPrescriptions):
+        integrations.prescriptions = MisMock()
+    if force or isinstance(integrations.identity, RejectIdentity):
+        integrations.identity = ConnectCodeIdentity()
+    if force or isinstance(integrations.drug_info, NoDrugInfo):
+        integrations.drug_info = GrlsExcerpts()
+    # Exact type check: MemoryNotifier subclasses used as test sinks must stay in place.
+    if force or type(integrations.notifier) is MemoryNotifier:
+        integrations.notifier = CompositeNotifier()
+    _done = True
+
+
+setup()

app/integrations/connect_code.py

@@ -0,0 +1,58 @@
+"""ЕСИА stand-in: the patient proves identity with the one-time connect code issued by the doctor.
+
+The doctor creates the code in «Подключить пациента» (identity DB `connect_codes`); the patient opens the
+QR deep link / enters the code. In prod this is ЕСИА OAuth; the `IdentityProvider` Protocol stays the same.
+"""
+
+import logging
+from datetime import UTC, timedelta
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.core import clock
+from app.integrations.base import VerifiedPerson
+
+log = logging.getLogger(__name__)
+
+# One connect flow verifies the same code twice: the Telegram channel on `/start <code>` (to bind the chat)
+# and then the dialog engine's `new` state (to check the code belongs to this token). A short grace window
+# lets the second check pass; after it a used code is rejected like an unknown one.
+REDEEM_GRACE = timedelta(minutes=15)
+
+
+class ConnectCodeIdentity:
+    """`IdentityProvider` over identity-DB connect codes."""
+
+    def __init__(self, grace: timedelta = REDEEM_GRACE) -> None:
+        self.grace = grace
+
+    async def verify(self, auth_code: str) -> VerifiedPerson | None:
+        code = auth_code.strip()
+        if not code:
+            return None
+        # Imported lazily: the integrations package is imported by clinical-side modules, which must not
+        # pull the identity schema into their import graph (guardrail #5).
+        from app.db import identity_repo
+        from app.db.identity import ConnectCode
+        from app.db.session import get_identity_session
+
+        now = clock.now()
+        try:
+            async with get_identity_session() as s:
+                row = await s.get(ConnectCode, code)
+                if row is None:
+                    log.info("connect code rejected: unknown")
+                    return None
+                person = VerifiedPerson(token=row.token, doctor_id=row.doctor_id)
+                if row.used_at is not None:
+                    used_at = row.used_at if row.used_at.tzinfo else row.used_at.replace(tzinfo=UTC)
+                    if now - used_at <= self.grace:
+                        return person
+                    log.info("connect code rejected: already used")
+                    return None
+                if await identity_repo.redeem_connect_code(s, code=code, now=now) is None:
+                    return None
+                return person
+        except (OSError, SQLAlchemyError) as exc:
+            log.warning("connect code check failed error=%s", type(exc).__name__)
+            return None

app/integrations/grls.py

@@ -0,0 +1,80 @@
+"""ГРЛС 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), [])]

app/integrations/mis_mock.py

@@ -0,0 +1,97 @@
+"""МИС / ЕГИСЗ stand-in: the patient's prescriptions (SPEC §9).
+
+Source order: the clinical DB table `prescriptions` (written when the doctor connects a patient or by the
+demo seed), then the synthetic catalogue `data/meds/mis_prescriptions.json` keyed by token. In prod this is
+a regional МИС query; the Protocol stays the same.
+"""
+
+import json
+import logging
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, Field, ValidationError
+from sqlalchemy import select
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.config import get_settings
+from app.integrations.base import Prescription
+
+log = logging.getLogger(__name__)
+
+
+class _CatalogPatient(BaseModel):
+    persona: str
+    tokens: list[str] = Field(default_factory=list)
+    prescriptions: list[Prescription]
+
+
+class _Catalog(BaseModel):
+    patients: list[_CatalogPatient]
+
+
+def _from_row(drug: str, dose: str, schedule: dict[str, Any], start: date | None, end: date | None) -> Prescription:
+    # The table has no instructions column; the doctor's note («утром после еды») may ride in the schedule JSON.
+    times = {k: [str(x) for x in v] for k, v in schedule.items() if isinstance(v, list)}
+    return Prescription(
+        drug=drug,
+        dose=dose,
+        schedule=times or {"times": ["08:00"]},
+        instructions=str(schedule.get("instructions") or ""),
+        start_date=start,
+        end_date=end,
+    )
+
+
+def default_catalog_path() -> Path:
+    return get_settings().data_dir / "meds" / "mis_prescriptions.json"
+
+
+class MisMock:
+    """`PrescriptionSource`: clinical DB first, synthetic JSON catalogue as the fallback."""
+
+    def __init__(self, catalog_path: Path | None = None, *, use_db: bool = True) -> None:
+        self._path = catalog_path
+        self._use_db = use_db
+        self._by_token: dict[str, list[Prescription]] | None = None
+
+    def _catalog(self) -> dict[str, list[Prescription]]:
+        # Lazy: the package is imported before startup, a missing/broken file must not break the import.
+        if self._by_token is None:
+            path = self._path or default_catalog_path()
+            index: dict[str, list[Prescription]] = {}
+            try:
+                catalog = _Catalog.model_validate(json.loads(path.read_text(encoding="utf-8")))
+            except (OSError, ValueError, ValidationError) as exc:
+                log.warning("mis_mock catalogue not loaded error=%s", type(exc).__name__)
+                catalog = _Catalog(patients=[])
+            for patient in catalog.patients:
+                for token in patient.tokens:
+                    index[token] = patient.prescriptions
+            self._by_token = index
+        return self._by_token
+
+    async def _from_db(self, token: str) -> list[Prescription]:
+        from app.db.clinical import Prescription as PrescriptionRow
+        from app.db.session import get_clinical_session
+
+        async with get_clinical_session() as s:
+            rows = (
+                await s.scalars(
+                    select(PrescriptionRow).where(PrescriptionRow.token == token).order_by(PrescriptionRow.id)
+                )
+            ).all()
+        return [_from_row(r.drug, r.dose, r.schedule or {}, r.start_date, r.end_date) for r in rows]
+
+    async def get_prescriptions(self, token: str) -> list[Prescription]:
+        if self._use_db:
+            try:
+                found = await self._from_db(token)
+            except (OSError, SQLAlchemyError) as exc:
+                # DB down or not migrated: the demo still shows the catalogue prescription.
+                log.warning("mis_mock db lookup failed error=%s", type(exc).__name__)
+                found = []
+            if found:
+                return found
+        return [p.model_copy(deep=True) for p in self._catalog().get(token, [])]

app/integrations/notify.py

@@ -0,0 +1,141 @@
+"""Doctor notifications (SPEC §9): in-memory feed for the dashboard + optional Telegram doctor chat.
+
+The alert itself is already committed to the clinical DB before `send()` is called (the dashboard reads
+alerts from there via the API), so every channel here is best effort: failures are logged, never raised.
+Messages carry only the level, the rule reason/code and a short pseudonymous patient id — no PII
+(guardrail #8); the reason texts come from `config/rules.yaml`, not from patient messages.
+"""
+
+import asyncio
+import itertools
+import logging
+from collections import deque
+from collections.abc import Awaitable, Callable
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel
+
+from app.config import get_settings
+from app.core import clock
+from app.integrations.base import AlertLevelName
+
+log = logging.getLogger(__name__)
+
+FEED_SIZE = 500
+SHORT_ID_LEN = 6
+MAX_REASON_CHARS = 200
+TELEGRAM_TIMEOUT_S = 5.0
+# Doctor-facing labels (not patient texts); acute alerts are visually distinct in the chat.
+LEVEL_LABELS: dict[str, str] = {
+    "acute": "КРИТИЧНО",
+    "high": "Высокий",
+    "moderate": "Умеренный",
+    "info": "Инфо",
+}
+
+ChatSender = Callable[[int, str], Awaitable[Any]]
+
+
+def short_id(token: str | None) -> str:
+    return token[:SHORT_ID_LEN] if token else "—"
+
+
+def format_message(level: str, text: str, token: str | None) -> str:
+    reason = " ".join(text.split())
+    if len(reason) > MAX_REASON_CHARS:
+        reason = reason[: MAX_REASON_CHARS - 1] + "…"
+    return f"[{LEVEL_LABELS.get(level, level)}] {reason}\nПациент #{short_id(token)}"
+
+
+class Notification(BaseModel):
+    id: int
+    ts: datetime
+    doctor_id: str
+    level: str
+    text: str
+    token: str | None = None
+
+
+class NotificationFeed:
+    """Bounded in-memory feed; ids increase monotonically so a poller asks only for `recent(since_id)`."""
+
+    def __init__(self, size: int = FEED_SIZE) -> None:
+        self._items: deque[Notification] = deque(maxlen=size)
+        self._ids = itertools.count(1)
+
+    def add(self, doctor_id: str, level: str, text: str, token: str | None, ts: datetime) -> Notification:
+        item = Notification(id=next(self._ids), ts=ts, doctor_id=doctor_id, level=level, text=text, token=token)
+        self._items.append(item)
+        return item
+
+    def recent(self, since_id: int = 0, doctor_id: str | None = None) -> list[Notification]:
+        return [n for n in self._items if n.id > since_id and (doctor_id is None or n.doctor_id == doctor_id)]
+
+
+class TelegramDoctorChat:
+    """Sends to the doctor chat `DOCTOR_NOTIFY_CHAT_ID` through the bot `TG_BOT_TOKEN`.
+
+    The aiogram `Bot` is created on the first message only; with either setting empty the channel is off.
+    """
+
+    def __init__(self, chat_id: int | None = None, bot_token: str | None = None, sender: ChatSender | None = None):
+        settings = get_settings()
+        self.chat_id = chat_id if chat_id is not None else settings.doctor_notify_chat_id
+        self._bot_token = bot_token if bot_token is not None else settings.tg_bot_token
+        self._sender = sender
+        self._bot: Any = None
+
+    @property
+    def enabled(self) -> bool:
+        return self.chat_id is not None and (self._sender is not None or bool(self._bot_token))
+
+    def _get_sender(self) -> ChatSender:
+        if self._sender is None:
+            from aiogram import Bot
+
+            self._bot = Bot(self._bot_token)
+            self._sender = self._bot.send_message
+        return self._sender
+
+    async def send(self, level: str, text: str, token: str | None) -> None:
+        if not self.enabled or self.chat_id is None:
+            return
+        try:
+            await asyncio.wait_for(
+                self._get_sender()(self.chat_id, format_message(level, text, token)), timeout=TELEGRAM_TIMEOUT_S
+            )
+        except Exception as exc:  # noqa: BLE001 — any channel error; the alert is already in the DB
+            err = type(exc).__name__
+            log.warning("doctor chat notify failed level=%s tok=%s error=%s", level, short_id(token), err)
+
+    async def aclose(self) -> None:
+        if self._bot is not None:
+            await self._bot.session.close()
+            self._bot = None
+            self._sender = None
+
+
+class CompositeNotifier:
+    """`Notifier`: dashboard feed + Telegram doctor chat (when configured)."""
+
+    def __init__(self, feed: NotificationFeed | None = None, chat: TelegramDoctorChat | None = None) -> None:
+        self.feed = feed or NotificationFeed()
+        self._chat = chat
+
+    @property
+    def chat(self) -> TelegramDoctorChat:
+        # Lazy: settings are read on the first notification, not when the package is imported.
+        if self._chat is None:
+            self._chat = TelegramDoctorChat()
+        return self._chat
+
+    def recent(self, since_id: int = 0, doctor_id: str | None = None) -> list[Notification]:
+        return self.feed.recent(since_id, doctor_id)
+
+    async def send(self, doctor_id: str, level: AlertLevelName, text: str, token: str | None = None) -> None:
+        try:
+            self.feed.add(doctor_id, level, text, token, clock.now())
+        except Exception as exc:  # noqa: BLE001 — never break the caller
+            log.warning("notify feed failed error=%s", type(exc).__name__)
+        await self.chat.send(level, text, token)