Pavel Eliseev
c42ef621e5
CI / ruff · pytest (pull_request) Successful in 55s
Add gated TSQ, Russian C-SSRS Recent, durable responses and notifications, diaries, schedules and audited clinical review. Preserve manual assessment when the Russian follow-up source is unavailable. Refs #52
265 lines
10 KiB
Python
265 lines
10 KiB
Python
"""Transactional questionnaire storage and a durable doctor notification queue.
|
|
|
|
Notifier acceptance is not delivery or a doctor's acknowledgement. At-least-once
|
|
retries can duplicate delivery if a process dies after sending but before commit.
|
|
"""
|
|
|
|
import asyncio
|
|
from dataclasses import asdict
|
|
from datetime import timedelta
|
|
from uuid import uuid4
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.core import clock
|
|
from app.core.intake_questions import content, cssrs_answers, group_answers
|
|
from app.core.intake_runtime import load_runtime, month_before
|
|
from app.core.questionnaire import evaluate_intake, load_policy, score_alcohol, score_cssrs_recent
|
|
from app.core.scales import load_scale, score
|
|
from app.core.types import AlertLevel
|
|
from app.db import repo
|
|
from app.db.clinical import Alert, DialogState
|
|
from app.db.intake import QuestionnaireNotice, QuestionnaireRecord, QuestionnaireTask
|
|
from app.db.session import get_clinical_session
|
|
from app.integrations.base import MemoryNotifier, integrations
|
|
|
|
|
|
async def records(session, token):
|
|
return list(
|
|
(
|
|
await session.scalars(
|
|
select(QuestionnaireRecord)
|
|
.where(QuestionnaireRecord.token == token)
|
|
.order_by(QuestionnaireRecord.started_at, QuestionnaireRecord.id)
|
|
)
|
|
).all()
|
|
)
|
|
|
|
|
|
async def create_record(ctx, kind, *, task=None):
|
|
profile = await ctx.profile()
|
|
runtime = load_runtime()
|
|
prior = next(
|
|
(r for r in reversed(await records(ctx.session, ctx.token)) if r.kind == "intake" and r.status == "complete"),
|
|
None,
|
|
)
|
|
sex = profile.sex if profile else None
|
|
if sex is None and prior:
|
|
sex = prior.details.get("sex")
|
|
row = QuestionnaireRecord(
|
|
id=uuid4().hex,
|
|
token=ctx.token,
|
|
kind=kind,
|
|
version=content()["version"],
|
|
policy_version=load_policy().version,
|
|
started_at=ctx.now,
|
|
day=ctx.now.astimezone(ZoneInfo(runtime.timezone)).date(),
|
|
status="active",
|
|
answers={},
|
|
details={
|
|
"runtime_version": runtime.version,
|
|
"modules": list(profile.modules or []) if profile else [],
|
|
"sex": sex,
|
|
"medication_active": bool(
|
|
(profile and profile.therapy_start and profile.therapy_start <= ctx.now.date())
|
|
or (prior and prior.answers.get("psychotropics") == 1)
|
|
),
|
|
"suicide_monitoring": bool(profile and profile.enhanced_monitoring),
|
|
"weekly_pcl": kind == "pcl5" and bool((ctx.data.get("crisis") or {}).get("closed_at")),
|
|
"answer_metadata": {},
|
|
"task_id": task.id if task else None,
|
|
},
|
|
)
|
|
ctx.session.add(row)
|
|
await ctx.session.flush()
|
|
if task:
|
|
task.record_id = row.id
|
|
task.status = "active"
|
|
return row
|
|
|
|
|
|
def save_answer(row, question, value, now):
|
|
"""Anchor recall windows to administration, never to the day of a resumed answer."""
|
|
day = row.day
|
|
starts = {
|
|
"past_month": month_before(day),
|
|
"past_3_months": month_before(day, 3),
|
|
"past_week": day - timedelta(days=7),
|
|
"past_day": day - timedelta(days=1),
|
|
"последние 2 недели": day - timedelta(days=14),
|
|
"последний месяц": month_before(day),
|
|
"последняя неделя": day - timedelta(days=7),
|
|
"прошлая неделя": day - timedelta(days=7),
|
|
"последние 12 месяцев": month_before(day, 12),
|
|
}
|
|
row.answers = {**row.answers, question.id: value}
|
|
if question.id == "sex" and value in ("m", "f"):
|
|
row.details = {**row.details, "sex": value}
|
|
row.details = {
|
|
**row.details,
|
|
"answer_metadata": {
|
|
**row.details.get("answer_metadata", {}),
|
|
question.id: {
|
|
"answered_at": now.isoformat(),
|
|
"period": question.period,
|
|
"period_start": starts[question.period].isoformat() if question.period in starts else None,
|
|
"period_end": day.isoformat(),
|
|
"source": "patient",
|
|
"derived": False,
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def decision(row):
|
|
a = row.answers
|
|
sex = row.details.get("sex") or (a.get("sex") if a.get("sex") in ("m", "f") else None)
|
|
return evaluate_intake(
|
|
phq9=group_answers(a, "phq9"),
|
|
tsq=group_answers(a, "tsq"),
|
|
pcl5=group_answers(a, "pcl5"),
|
|
alcohol=group_answers(a, "rus_audit"),
|
|
cssrs=cssrs_answers(a),
|
|
sex=sex,
|
|
trauma_exposure=bool(a["trauma"]) if "trauma" in a else None,
|
|
traumatic_brain_injury=a.get("tbi") == 1,
|
|
psychotropics=a.get("psychotropics") == 1,
|
|
)
|
|
|
|
|
|
def summary(row):
|
|
a = row.answers
|
|
totals = {}
|
|
for sid in ("phq9", "tsq", "pcl5"):
|
|
result = score(sid, group_answers(a, sid))
|
|
totals[sid] = {
|
|
"complete": result.complete,
|
|
"total": result.total if result.complete else None,
|
|
"answered": result.answered,
|
|
}
|
|
sex = row.details.get("sex") or (a.get("sex") if a.get("sex") in ("m", "f") else None)
|
|
for sid in ("rus_audit_s", "rus_audit"):
|
|
values = {k: a[k] for k in load_scale(sid).item_ids if k in a}
|
|
totals[sid] = asdict(score_alcohol(values, short=sid.endswith("_s"), sex=sex))
|
|
totals["cssrs_recent"] = asdict(score_cssrs_recent(cssrs_answers(a)))
|
|
return totals
|
|
|
|
|
|
async def finish(ctx, row, status, *, extra=None):
|
|
row.status = status
|
|
row.completed_at = ctx.now if status == "complete" else None
|
|
row.details = {**row.details, **(extra or {})}
|
|
if row.details.get("task_id"):
|
|
task = await ctx.session.get(QuestionnaireTask, row.details["task_id"])
|
|
if task:
|
|
task.status = "done"
|
|
|
|
|
|
async def alert(session, token, key, code, now, *, level=AlertLevel.MODERATE, params=None):
|
|
existing = await session.scalar(
|
|
select(QuestionnaireNotice).where(QuestionnaireNotice.token == token, QuestionnaireNotice.event_key == key)
|
|
)
|
|
if existing:
|
|
return existing
|
|
row = await repo.add_alert(session, token=token, level=level, code=code, reason=code, ts=now, params=params)
|
|
notice = QuestionnaireNotice(token=token, event_key=key, alert_id=row.id, created_at=now)
|
|
session.add(notice)
|
|
await session.flush()
|
|
return notice
|
|
|
|
|
|
async def cancel_pending(session, token):
|
|
for row in await records(session, token):
|
|
if row.status in ("active", "paused"):
|
|
row.status = "cancelled"
|
|
tasks = await session.scalars(
|
|
select(QuestionnaireTask).where(
|
|
QuestionnaireTask.token == token, QuestionnaireTask.status.in_(("pending", "active"))
|
|
)
|
|
)
|
|
for task in tasks:
|
|
task.status = "cancelled"
|
|
task.delivery = []
|
|
notices = await session.scalars(
|
|
select(QuestionnaireNotice).where(QuestionnaireNotice.token == token, QuestionnaireNotice.status == "pending")
|
|
)
|
|
for notice in notices:
|
|
notice.status = "cancelled"
|
|
|
|
|
|
async def dispatch_notices(token=None):
|
|
now = clock.now()
|
|
async with get_clinical_session() as session:
|
|
stmt = (
|
|
select(QuestionnaireNotice.id, QuestionnaireNotice.token)
|
|
.where(QuestionnaireNotice.status == "pending")
|
|
.order_by(QuestionnaireNotice.id)
|
|
.limit(100)
|
|
)
|
|
if token:
|
|
stmt = stmt.where(QuestionnaireNotice.token == token)
|
|
candidates = (await session.execute(stmt)).all()
|
|
for notice_id, patient_token in candidates:
|
|
async with get_clinical_session() as session:
|
|
# Same lock order as withdrawal: patient first, then notice.
|
|
await lock_patient(session, patient_token)
|
|
notice = await session.scalar(
|
|
select(QuestionnaireNotice).where(QuestionnaireNotice.id == notice_id).with_for_update()
|
|
)
|
|
if notice.status != "pending":
|
|
continue
|
|
if notice.attempted_at and now < notice.attempted_at + timedelta(
|
|
minutes=min(60, 2 ** min(notice.attempts, 6))
|
|
):
|
|
continue
|
|
state = await repo.get_dialog_state(session, notice.token)
|
|
if state is None or state.data.get("consent") is not True:
|
|
notice.status = "cancelled"
|
|
continue
|
|
profile = await repo.get_profile(session, notice.token)
|
|
if not profile or not profile.doctor_id:
|
|
continue
|
|
item = await session.get(Alert, notice.alert_id)
|
|
notice.attempted_at, notice.attempts = now, notice.attempts + 1
|
|
try:
|
|
await asyncio.wait_for(
|
|
integrations.notifier.send(profile.doctor_id, item.level, item.code, token=notice.token), timeout=10
|
|
)
|
|
except Exception as exc:
|
|
notice.error_code = type(exc).__name__[:80]
|
|
else:
|
|
notice.error_code = None
|
|
notice.accepted_at = now
|
|
notice.status = (
|
|
"test_sink" if isinstance(integrations.notifier, MemoryNotifier) else "accepted_by_adapter"
|
|
)
|
|
|
|
|
|
async def queue_task(session, token, kind, key, now):
|
|
existing = await session.scalar(
|
|
select(QuestionnaireTask).where(QuestionnaireTask.token == token, QuestionnaireTask.task_key == key)
|
|
)
|
|
if existing:
|
|
return existing
|
|
# Only one outstanding administration per form, however many ticks were missed.
|
|
existing = await session.scalar(
|
|
select(QuestionnaireTask).where(
|
|
QuestionnaireTask.token == token,
|
|
QuestionnaireTask.kind == kind,
|
|
QuestionnaireTask.status.in_(("pending", "active")),
|
|
)
|
|
)
|
|
if existing:
|
|
return existing
|
|
row = QuestionnaireTask(token=token, kind=kind, task_key=key, due_at=now)
|
|
session.add(row)
|
|
await session.flush()
|
|
return row
|
|
|
|
|
|
async def lock_patient(session, token):
|
|
# PostgreSQL serializes callbacks from different workers. SQLite tests additionally
|
|
# use the engine's in-process lock. Consent and answers share this transaction.
|
|
await session.scalar(select(DialogState).where(DialogState.token == token).with_for_update())
|