ElebobaandClaude Opus 5.5
0bbe1e0932
CI / ruff · pytest (pull_request) Successful in 2m44s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""ЕМИАС medical record mock: diagnoses, comorbidities, allergies and other specialists' prescriptions.
|
|
|
|
Prod: the regional МИС / ЕГИСЗ adapter replaces `EmiasRecordSource` (same `get(token)` contract). Demo: synthetic
|
|
records from `data/emias/records.yaml` keyed by synthetic SNILS, converted to tokens with `pseudo_id()` at load
|
|
time — so the record handed to the chat agent carries no SNILS, name or phone (guardrail #1). The current
|
|
psychiatric prescription comes from the МИС mock (`integrations.prescriptions`).
|
|
"""
|
|
|
|
import logging
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.config import get_settings
|
|
from app.core.privacy import pseudo_id
|
|
from app.integrations.base import integrations
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class Diagnosis(BaseModel):
|
|
code: str
|
|
name: str
|
|
|
|
|
|
class Condition(BaseModel):
|
|
name: str
|
|
tags: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class OtherMedication(BaseModel):
|
|
drug: str
|
|
how: str = ""
|
|
prescribed_by: str = ""
|
|
classes: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class CurrentPrescription(BaseModel):
|
|
drug: str
|
|
dose: str
|
|
instructions: str = ""
|
|
|
|
|
|
class PatientRecord(BaseModel):
|
|
"""Clinical-only view of the record (no identifiers): what the chat agent may see."""
|
|
|
|
diagnoses: list[Diagnosis] = Field(default_factory=list)
|
|
conditions: list[Condition] = Field(default_factory=list)
|
|
allergies: list[Condition] = Field(default_factory=list)
|
|
other_medications: list[OtherMedication] = Field(default_factory=list)
|
|
pregnancy: dict[str, str] | None = None
|
|
current_prescriptions: list[CurrentPrescription] = Field(default_factory=list)
|
|
|
|
|
|
def default_records_path() -> Path:
|
|
return get_settings().data_dir / "emias" / "records.yaml"
|
|
|
|
|
|
@lru_cache
|
|
def _load(path: Path) -> dict[str, PatientRecord]:
|
|
raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
out: dict[str, PatientRecord] = {}
|
|
for snils, rec in (raw.get("records") or {}).items():
|
|
aliases = rec.pop("tokens", []) if isinstance(rec, dict) else []
|
|
record = PatientRecord.model_validate(rec)
|
|
out[pseudo_id(snils)] = record
|
|
for token in aliases: # demo tokens used by the patient web app (e.g. tok-andrey) share the record
|
|
out[token] = record
|
|
return out
|
|
|
|
|
|
class EmiasRecordSource:
|
|
def __init__(self, path: Path | None = None) -> None:
|
|
self._path = path
|
|
|
|
async def get(self, token: str) -> PatientRecord | None:
|
|
try:
|
|
base = _load(self._path or default_records_path()).get(token)
|
|
except (OSError, yaml.YAMLError, ValueError) as exc:
|
|
log.error("emias records unavailable: %s", type(exc).__name__)
|
|
base = None
|
|
record = base.model_copy(deep=True) if base is not None else PatientRecord()
|
|
try:
|
|
rx = await integrations.prescriptions.get_prescriptions(token)
|
|
except Exception as exc: # noqa: BLE001 — the МИС being down must not break the chat
|
|
log.error("prescriptions unavailable: %s", type(exc).__name__)
|
|
rx = []
|
|
record.current_prescriptions = [
|
|
CurrentPrescription(drug=p.drug, dose=p.dose, instructions=p.instructions) for p in rx
|
|
]
|
|
if base is None and not rx:
|
|
return None
|
|
return record
|
|
|
|
|
|
emias_records = EmiasRecordSource()
|