Arxip222andClaude Opus 5.5
0106aefc33
CI / ruff · pytest (pull_request) Successful in 2m9s
Conflict in app/api/routes.py: /demo/personas keeps the dev version (demo console, DB patients with scenarios); the patient app gets its own /patient/demo-personas (synthetic ЕМИАС accounts). Anonymous sign-in now always creates a doctor-less pseudonymous profile, since the dialog API of #12 serves known profiles only. Refs #77 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
407 lines
16 KiB
Python
407 lines
16 KiB
Python
"""Patient web app: sign-in, the patient's own plan and public reference data (`/api/v1/auth/*`, `/me/*`,
|
|
`/crisis/resources`, `/instruments/catalog`, `/meds/reference`).
|
|
|
|
The web app is one more channel of the same dialog engine: after sign-in it talks to `/dialog/{token}/...`
|
|
exactly like the web console, and consent, onboarding, scales and check-ins come from the engine.
|
|
|
|
Two ways in:
|
|
- **ЕМИАС** (demo): the doctor's one-time connect code via `EmiasMock`, or a synthetic demo persona. The
|
|
patient is linked to the attending doctor, who sees the data on the dashboard.
|
|
- **Anonymous**: a random pseudonymous token and nothing else — no identity record, no doctor. The engine
|
|
supports tokens without an identity record (consent is kept in the dialog state), so no personal data is
|
|
stored at all. Crisis safety still works: the fixed crisis script does not depend on the doctor.
|
|
|
|
The token returned here is the session credential of the prototype. Production needs short-lived sessions
|
|
issued by the server instead of handing out the pseudonymous token itself.
|
|
"""
|
|
|
|
import logging
|
|
import secrets
|
|
from collections import defaultdict
|
|
from datetime import date, datetime, time, timedelta
|
|
from typing import Any, Literal
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sqlalchemy import select
|
|
|
|
from app.api import fixtures
|
|
from app.api.schemas import DemoPersona
|
|
from app.core import clock
|
|
from app.core.instrument_catalog import Catalog, catalog
|
|
from app.core.intake_questions import DIARY_KINDS
|
|
from app.core.intake_runtime import intake_enabled, load_runtime
|
|
from app.core.interpretation import texts as results_texts
|
|
from app.core.meds_reference import find_drug, indication_for, load_reference
|
|
from app.core.safety import CrisisResource, crisis_reply
|
|
from app.db import repo
|
|
from app.db.clinical import DailyCheckin, Dose, Prescription
|
|
from app.db.identity import TOKEN_LEN
|
|
from app.db.intake import QuestionnaireRecord
|
|
from app.db.session import get_clinical_session
|
|
from app.integrations.emias_mock import CODE_MAX_LEN, EmiasMock
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(tags=["patient-web"])
|
|
_emias = EmiasMock()
|
|
|
|
|
|
class AuthOut(BaseModel):
|
|
token: str
|
|
mode: Literal["emias", "anonymous"]
|
|
display_name: str | None = None # only for ЕМИАС sign-in; anonymous has none by design
|
|
|
|
|
|
class EmiasCodeIn(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
code: str = Field(min_length=1, max_length=CODE_MAX_LEN)
|
|
|
|
|
|
class AnonymousIn(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
focus: Literal["depression", "ptsd"] | None = None # what the person suspects; picks the module
|
|
|
|
|
|
# Self-selected focus of an anonymous person → clinical module (PTSD is the veteran module, ADR-010).
|
|
_FOCUS_MODULE = {"depression": "depression", "ptsd": "veteran"}
|
|
|
|
|
|
class EmiasDemoIn(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
persona_token: str = Field(min_length=1, max_length=64)
|
|
|
|
|
|
async def _attach_doctor(token: str, doctor_id: str) -> None:
|
|
async with get_clinical_session() as s:
|
|
if await repo.get_profile(s, token) is None:
|
|
await repo.upsert_profile(s, token=token, doctor_id=doctor_id)
|
|
|
|
|
|
async def _seed_demo_persona(token: str) -> None:
|
|
"""Synthetic clinical start for a demo persona (no `make seed` yet): profile, prescription, today's doses.
|
|
|
|
Idempotent: only missing pieces are created. Data mirror the dashboard fixtures, so the patient app and the
|
|
doctor's card tell the same story.
|
|
"""
|
|
card = fixtures.card(token)
|
|
item = next((p for p in fixtures.PATIENTS if p.token == token), None)
|
|
if card is None or item is None:
|
|
return
|
|
now = clock.now()
|
|
today = now.date()
|
|
async with get_clinical_session() as s:
|
|
if await repo.get_profile(s, token) is None:
|
|
await repo.upsert_profile(
|
|
s,
|
|
token=token,
|
|
doctor_id=fixtures.DOCTOR_ID,
|
|
modules=list(item.modules),
|
|
therapy_start=min((p.start_date for p in card.prescriptions if p.start_date), default=today),
|
|
next_visit=today + timedelta(days=15),
|
|
)
|
|
if await s.scalar(select(Prescription.id).where(Prescription.token == token)) is not None:
|
|
return
|
|
for p in card.prescriptions:
|
|
rx = Prescription(
|
|
token=token, drug=p.drug, dose=p.dose, schedule=p.schedule, start_date=p.start_date or today
|
|
)
|
|
s.add(rx)
|
|
await s.flush()
|
|
# The dose history of the dashboard story (days before today), so the week marks match the doctor's card.
|
|
for d in card.doses:
|
|
if d.drug == p.drug and d.planned_at.astimezone(now.tzinfo).date() < today:
|
|
s.add(Dose(token=token, prescription_id=rx.id, planned_at=d.planned_at, status=d.status))
|
|
for t in p.schedule.get("times", []):
|
|
hh, mm = (int(x) for x in t.split(":"))
|
|
s.add(
|
|
Dose(
|
|
token=token, prescription_id=rx.id, planned_at=datetime.combine(today, time(hh, mm), now.tzinfo)
|
|
)
|
|
)
|
|
|
|
|
|
@router.post("/auth/emias", response_model=AuthOut)
|
|
async def auth_emias(body: EmiasCodeIn) -> AuthOut:
|
|
"""Sign-in through ЕМИАС (demo: one-time connect code from the doctor)."""
|
|
person = await _emias.verify(body.code)
|
|
if person is None:
|
|
log.info("web auth: emias code rejected")
|
|
raise HTTPException(401, "code not accepted")
|
|
await _attach_doctor(person.token, person.doctor_id)
|
|
log.info("web auth: emias tok=%s", person.token[:6])
|
|
return AuthOut(token=person.token, mode="emias")
|
|
|
|
|
|
@router.get("/patient/demo-personas", response_model=list[DemoPersona])
|
|
async def patient_demo_personas() -> list[DemoPersona]:
|
|
"""Synthetic ЕМИАС accounts for the patient app sign-in (the demo console has its own `/demo/personas`)."""
|
|
return fixtures.DEMO_PERSONAS
|
|
|
|
|
|
@router.post("/auth/emias/demo", response_model=AuthOut)
|
|
async def auth_emias_demo(body: EmiasDemoIn) -> AuthOut:
|
|
"""Demo ЕМИАС account: signs in as one of the synthetic personas (`/patient/demo-personas`)."""
|
|
persona = next((p for p in fixtures.DEMO_PERSONAS if p.token == body.persona_token), None)
|
|
if persona is None:
|
|
raise HTTPException(404, "persona not found")
|
|
await _seed_demo_persona(persona.token)
|
|
log.info("web auth: emias demo tok=%s", persona.token[:6])
|
|
return AuthOut(token=persona.token, mode="emias", display_name=persona.display_name)
|
|
|
|
|
|
@router.post("/auth/anonymous", response_model=AuthOut)
|
|
async def auth_anonymous(body: AnonymousIn | None = None) -> AuthOut:
|
|
"""Anonymous sign-in: a fresh random token; no identity record, no doctor.
|
|
|
|
The clinical side gets a doctor-less pseudonymous profile (the dialog API serves known profiles only). It holds
|
|
no personal data — only the module of the chosen `focus`; the stage 1 screening then sets the modules itself.
|
|
"""
|
|
token = secrets.token_hex(TOKEN_LEN // 2)
|
|
focus = body.focus if body is not None else None
|
|
async with get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=token, doctor_id="", modules=[_FOCUS_MODULE[focus]] if focus else [])
|
|
log.info("web auth: anonymous tok=%s", token[:6])
|
|
return AuthOut(token=token, mode="anonymous")
|
|
|
|
|
|
class DoseItem(BaseModel):
|
|
id: int
|
|
drug: str
|
|
dose: str
|
|
time: str # HH:MM, patient's local time
|
|
status: Literal["taken", "missed", "skipped", "pending"]
|
|
|
|
|
|
class WeekDay(BaseModel):
|
|
day: date
|
|
status: Literal["taken", "missed", "pending", "none"]
|
|
|
|
|
|
class PrescriptionItem(BaseModel):
|
|
drug: str
|
|
dose: str
|
|
times: list[str]
|
|
week: list[WeekDay] # Mon..Sun of the current week
|
|
ref_id: str | None = None # entry of /meds/reference, if the drug is covered
|
|
indication: Literal["depression", "ptsd"] | None = None
|
|
start_date: date | None = None
|
|
therapy_day: int | None = None # 1 on the start date (spec: M5, «ожидаемый эффект» cards on days 1/7/14)
|
|
|
|
|
|
class Questionnaires(BaseModel):
|
|
"""Questionnaire state for the home screen. Answers and scores are not exposed here."""
|
|
|
|
enabled: bool
|
|
intake: Literal["none", "active", "paused", "complete", "review"]
|
|
diary_done: bool # both parts of today's diary
|
|
diary_next: Literal["morning", "evening"] | None = None # the part still open today
|
|
|
|
|
|
class TodayOut(BaseModel):
|
|
today: date
|
|
now: str # HH:MM on the (demo) clock — the client never uses its own clock
|
|
has_doctor: bool
|
|
checkin_done: bool
|
|
next_visit: date | None
|
|
doses: list[DoseItem]
|
|
prescriptions: list[PrescriptionItem]
|
|
questionnaires: Questionnaires
|
|
|
|
|
|
def _intake_state(records: list[QuestionnaireRecord]) -> Literal["none", "active", "paused", "complete", "review"]:
|
|
intake = [r for r in records if r.kind == "intake"]
|
|
if any(r.status == "complete" for r in intake):
|
|
return "complete"
|
|
if not intake:
|
|
return "none"
|
|
last = max(intake, key=lambda r: r.started_at).status
|
|
if last in ("active", "paused"):
|
|
return last
|
|
return "review" if last in ("review_required", "interrupted") else "none"
|
|
|
|
|
|
def _day_status(statuses: list[str]) -> Literal["taken", "missed", "pending", "none"]:
|
|
if not statuses:
|
|
return "none"
|
|
if any(s in ("missed", "skipped") for s in statuses):
|
|
return "missed"
|
|
return "taken" if all(s == "taken" for s in statuses) else "pending"
|
|
|
|
|
|
@router.get("/me/{token}/today", response_model=TodayOut)
|
|
async def my_today(token: str) -> TodayOut:
|
|
"""The patient's own plan for today (home screen) and this week's dose marks (medication screen).
|
|
|
|
Only the patient's own prescription data; no scores, no alerts. Anonymous tokens have no profile → empty plan.
|
|
"""
|
|
now = clock.now()
|
|
today = now.date()
|
|
week_start = today - timedelta(days=today.weekday())
|
|
async with get_clinical_session() as s:
|
|
profile = await repo.get_profile(s, token)
|
|
prescriptions = (await s.scalars(select(Prescription).where(Prescription.token == token))).all()
|
|
doses = (
|
|
await s.scalars(
|
|
select(Dose)
|
|
.where(Dose.token == token, Dose.planned_at >= datetime.combine(week_start, time.min, now.tzinfo))
|
|
.order_by(Dose.planned_at)
|
|
)
|
|
).all()
|
|
checkin = await s.scalar(select(DailyCheckin).where(DailyCheckin.token == token, DailyCheckin.day == today))
|
|
records = list((await s.scalars(select(QuestionnaireRecord).where(QuestionnaireRecord.token == token))).all())
|
|
|
|
by_rx: dict[int, Prescription] = {p.id: p for p in prescriptions}
|
|
today_doses: list[DoseItem] = []
|
|
marks: dict[tuple[int, date], list[str]] = defaultdict(list)
|
|
for d in doses:
|
|
local = d.planned_at.astimezone(now.tzinfo)
|
|
marks[(d.prescription_id, local.date())].append(d.status)
|
|
rx = by_rx.get(d.prescription_id)
|
|
if local.date() == today and rx is not None:
|
|
today_doses.append(
|
|
DoseItem(id=d.id, drug=rx.drug, dose=rx.dose, time=local.strftime("%H:%M"), status=d.status)
|
|
)
|
|
week = [week_start + timedelta(days=i) for i in range(7)]
|
|
modules = list(profile.modules or []) if profile else []
|
|
diary_day = now.astimezone(ZoneInfo(load_runtime().timezone)).date()
|
|
done_parts = {r.kind for r in records if r.day == diary_day and r.status == "complete"}
|
|
open_parts = [k.removeprefix("diary_") for k in DIARY_KINDS if k not in done_parts]
|
|
diary_done = not open_parts
|
|
|
|
def item(p: Prescription) -> PrescriptionItem:
|
|
ref = find_drug(p.drug)
|
|
return PrescriptionItem(
|
|
drug=p.drug,
|
|
dose=p.dose,
|
|
times=list((p.schedule or {}).get("times", [])),
|
|
week=[WeekDay(day=d, status=_day_status(marks[(p.id, d)])) for d in week],
|
|
ref_id=ref["id"] if ref else None,
|
|
indication=indication_for(modules, ref) if ref else None,
|
|
start_date=p.start_date,
|
|
therapy_day=(today - p.start_date).days + 1 if p.start_date else None,
|
|
)
|
|
|
|
return TodayOut(
|
|
today=today,
|
|
now=now.strftime("%H:%M"),
|
|
has_doctor=profile is not None and bool(profile.doctor_id),
|
|
checkin_done=(checkin is not None and checkin.mood is not None) or diary_done,
|
|
next_visit=profile.next_visit if profile else None,
|
|
doses=today_doses,
|
|
prescriptions=[
|
|
item(p)
|
|
for p in prescriptions
|
|
if (p.end_date is None or p.end_date >= today) and (p.start_date is None or p.start_date <= today)
|
|
],
|
|
questionnaires=Questionnaires(
|
|
# Stage 1 needs no profile: the screening creates it (anonymous people have none before it).
|
|
enabled=intake_enabled(),
|
|
intake=_intake_state(records),
|
|
diary_done=diary_done,
|
|
diary_next=open_parts[0] if open_parts else None,
|
|
),
|
|
)
|
|
|
|
|
|
class ResultFinding(BaseModel):
|
|
code: str
|
|
title: str
|
|
|
|
|
|
class Conclusion(BaseModel):
|
|
"""Preliminary conclusion of stage 1 — a screening result, not a diagnosis."""
|
|
|
|
day: date
|
|
title: str
|
|
lines: list[str]
|
|
findings: list[ResultFinding]
|
|
modules: list[str]
|
|
disclaimer: str
|
|
|
|
|
|
class HistoryItem(BaseModel):
|
|
id: str
|
|
kind: str
|
|
kind_title: str
|
|
status: str
|
|
status_title: str
|
|
day: date
|
|
started_at: datetime
|
|
completed_at: datetime | None
|
|
title: str | None # interpretation shown to the patient when the form was finished
|
|
lines: list[str]
|
|
|
|
|
|
class ResultsOut(BaseModel):
|
|
conclusion: Conclusion | None
|
|
history: list[HistoryItem]
|
|
|
|
|
|
@router.get("/me/{token}/results", response_model=ResultsOut)
|
|
async def my_results(token: str) -> ResultsOut:
|
|
"""The patient's own questionnaire log: every administration with its date, status and the interpretation that
|
|
was shown when it was finished (stored with the record, so it never changes afterwards). No raw answers."""
|
|
async with get_clinical_session() as s:
|
|
records = list(
|
|
(
|
|
await s.scalars(
|
|
select(QuestionnaireRecord)
|
|
.where(QuestionnaireRecord.token == token)
|
|
.order_by(QuestionnaireRecord.started_at.desc(), QuestionnaireRecord.id)
|
|
)
|
|
).all()
|
|
)
|
|
t = results_texts()
|
|
intake = next((r for r in records if r.kind == "intake" and r.status == "complete"), None)
|
|
conclusion = None
|
|
if intake and intake.details.get("interpretation"):
|
|
i = intake.details["interpretation"]
|
|
conclusion = Conclusion(
|
|
day=intake.day,
|
|
title=i["title"],
|
|
lines=i["lines"],
|
|
findings=[ResultFinding(**f) for f in i.get("findings", [])],
|
|
modules=i.get("modules", []),
|
|
disclaimer=i.get("disclaimer", t["disclaimer"]),
|
|
)
|
|
history = [
|
|
HistoryItem(
|
|
id=r.id,
|
|
kind=r.kind,
|
|
kind_title=t["kinds"].get(r.kind, r.kind),
|
|
status=r.status,
|
|
status_title=t["status"].get(r.status, r.status),
|
|
day=r.day,
|
|
started_at=r.started_at,
|
|
completed_at=r.completed_at,
|
|
title=(r.details.get("interpretation") or {}).get("title"),
|
|
lines=(r.details.get("interpretation") or {}).get("lines", []),
|
|
)
|
|
for r in records
|
|
]
|
|
return ResultsOut(conclusion=conclusion, history=history)
|
|
|
|
|
|
@router.get("/crisis/resources", response_model=list[CrisisResource])
|
|
async def crisis_resources() -> list[CrisisResource]:
|
|
"""Help lines for the always-visible «help now» button — the same list as the fixed crisis script."""
|
|
return crisis_reply([], "patient").resources
|
|
|
|
|
|
@router.get("/instruments/catalog", response_model=Catalog)
|
|
async def instruments_catalog() -> Catalog:
|
|
"""Form, position and recall period of every questionnaire item, for the question screen header."""
|
|
return catalog()
|
|
|
|
|
|
@router.get("/meds/reference")
|
|
async def meds_reference() -> dict[str, Any]:
|
|
"""Patient drug reference (draft from the Minzdrav guidelines): expected effect, course, side effects, red flags.
|
|
|
|
The internal rule parameters stay on the server.
|
|
"""
|
|
ref = load_reference()
|
|
return {k: v for k, v in ref.items() if k != "rules"}
|