ElebobaandClaude Opus 5.5
9362eb6ecc
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>
98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
"""МИС / ЕГИСЗ 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, [])]
|