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
377 lines
16 KiB
Python
377 lines
16 KiB
Python
"""Versioned, resumable questionnaires. Disabled until the clinical release gate opens."""
|
|
|
|
import math
|
|
from dataclasses import asdict, replace
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app.core.intake_questions import (
|
|
content,
|
|
cssrs_answers,
|
|
diary_questions,
|
|
group_answers,
|
|
next_intake,
|
|
scale_question,
|
|
texts,
|
|
)
|
|
from app.core.intake_runtime import diary_flags, intake_enabled, load_runtime
|
|
from app.core.questionnaire import alcohol_expansion, load_policy, score_alcohol, score_cssrs_recent
|
|
from app.core.scales import load_scale
|
|
from app.core.types import AlertLevel
|
|
from app.db.intake import QuestionnaireRecord, QuestionnaireTask
|
|
from app.dialog.contracts import Button, Event, Reply
|
|
from app.dialog.crisis import crisis_open, open_crisis
|
|
from app.dialog.fsm import Step, on_enter, state, trigger
|
|
from app.services import questionnaire as service
|
|
|
|
|
|
def allowed(ctx):
|
|
return ctx.data.get("consent") is True and ctx.data.get("adult") is True and intake_enabled()
|
|
|
|
|
|
def next_question(row):
|
|
a = row.answers
|
|
if row.kind == "intake":
|
|
return next_intake(a, load_policy(), row.details.get("sex"))
|
|
if row.kind == "diary":
|
|
for q in diary_questions(
|
|
row.details["modules"], suicide_monitoring=row.details["suicide_monitoring"], weekday=row.day.weekday()
|
|
):
|
|
if q.id in a:
|
|
continue
|
|
if q.id == "medication_reason" and a.get("medication") not in ("partial", "none"):
|
|
continue
|
|
if q.id == "side_effects" and a.get("medication") == "not_prescribed":
|
|
continue
|
|
if q.id == "thought_intensity" and a.get("death_thoughts") != 1:
|
|
continue
|
|
return q
|
|
return None
|
|
if row.kind in ("phq9", "pcl5", "rus_audit_s"):
|
|
sid = row.kind
|
|
if sid == "rus_audit_s":
|
|
short = score_alcohol(
|
|
{k: a[k] for k in load_scale(sid).item_ids if k in a}, short=True, sex=row.details.get("sex")
|
|
)
|
|
if alcohol_expansion(short, load_policy()):
|
|
sid = "rus_audit"
|
|
for item in load_scale(sid).item_ids:
|
|
if item == "rus_audit_2" and a.get("rus_audit_1") == 0:
|
|
continue
|
|
if item not in a:
|
|
q = scale_question(sid, item)
|
|
if sid == "pcl5" and row.details["weekly_pcl"]:
|
|
q = replace(q, text=q.text.replace("последний месяц", "последнюю неделю"), period="past_week")
|
|
return q
|
|
if row.details.get("medication_active") and "side_effects" not in a:
|
|
return next(
|
|
q
|
|
for q in diary_questions([], suicide_monitoring=False, weekday=row.day.weekday())
|
|
if q.id == "side_effects"
|
|
)
|
|
return None
|
|
|
|
|
|
def question_reply(row, q):
|
|
t = texts()
|
|
# Nonce + item + option index rejects old/replayed callbacks without exposing raw answers.
|
|
prefix = f"iq:{row.id[:12]}:{q.id}:"
|
|
buttons = [[Button(label=label, data=prefix + str(i))] for i, (_, label) in enumerate(q.options)]
|
|
if q.optional:
|
|
buttons.append([Button(label=t["comment_skip"], data=prefix + "skip")])
|
|
buttons.extend([[Button(label=t["help"], data="intake:help")], [Button(label=t["pause"], data="intake:pause")]])
|
|
return Reply(text=q.text + "\n\n" + t["as_of"].format(day=row.day.isoformat()), buttons=buttons, kind="question")
|
|
|
|
|
|
async def urgent(ctx, row=None, *, code="help_now", categories=None):
|
|
if crisis_open(ctx.data):
|
|
return open_crisis(ctx, categories or [], source="questionnaire:" + code)
|
|
profile = await ctx.profile()
|
|
if profile:
|
|
profile.enhanced_monitoring = True
|
|
flags = list(service.decision(row).flags) if row else []
|
|
if row and (row.answers.get("death_thoughts") == 1 or row.answers.get("phq9_9", 0) > 0):
|
|
flags.append("cssrs_since_contact_required")
|
|
if row:
|
|
await service.finish(ctx, row, "interrupted", extra={"stop_reason": code, "flags": flags})
|
|
await service.alert(
|
|
ctx.session,
|
|
ctx.token,
|
|
f"{row.id if row else ctx.now.isoformat()}:{code}",
|
|
code,
|
|
ctx.now,
|
|
level=AlertLevel.ACUTE,
|
|
params={"record_id": row.id if row else None, "flags": flags},
|
|
)
|
|
step = open_crisis(ctx, categories or [], source="questionnaire:" + code)
|
|
step.data["questionnaire_record"] = None
|
|
step.data["intake_review_required"] = True
|
|
for reply in step.replies:
|
|
reply.buttons.append([Button(label=ctx.texts.safety_plan.title.rstrip(":"), data="safety_plan")])
|
|
return step
|
|
|
|
|
|
async def manual_review(ctx, row, code):
|
|
await service.finish(ctx, row, "review_required", extra={"stop_reason": code})
|
|
await service.alert(
|
|
ctx.session, ctx.token, f"{row.id}:{code}", code, ctx.now, level=AlertLevel.HIGH, params={"record_id": row.id}
|
|
)
|
|
return Step(
|
|
replies=[
|
|
Reply(
|
|
text=texts()["missing_followup"] if code == "cssrs_since_contact_required" else texts()["review"],
|
|
buttons=[[Button(label=texts()["help"], data="intake:help")]],
|
|
kind="info",
|
|
)
|
|
],
|
|
next_state="idle",
|
|
data={"questionnaire_record": None, "intake_review_required": True},
|
|
)
|
|
|
|
|
|
async def advance(ctx, row):
|
|
a = row.answers
|
|
c = score_cssrs_recent(cssrs_answers(a))
|
|
policy = load_policy()
|
|
if a.get("phq9_9", 0) > 0 or a.get("death_thoughts") == 1 or c.risk_lower_bound == "high":
|
|
return await urgent(ctx, row, code="suicide_signal")
|
|
if c.positive and policy.cssrs_positive_policy == "stop_any":
|
|
return await urgent(ctx, row, code="cssrs_positive")
|
|
if c.conflicts:
|
|
return await manual_review(ctx, row, "cssrs_conflict")
|
|
# Finish the risk questions before any routine question; a moderate result stops here.
|
|
if c.complete and c.risk == "moderate":
|
|
return await manual_review(ctx, row, "urgent_psychiatrist")
|
|
if c.complete and c.risk == "low":
|
|
profile = await ctx.profile()
|
|
if profile:
|
|
profile.enhanced_monitoring = True
|
|
await service.alert(
|
|
ctx.session,
|
|
ctx.token,
|
|
row.id + ":cssrs_low",
|
|
"cssrs_low_safety_plan",
|
|
ctx.now,
|
|
params={"record_id": row.id},
|
|
)
|
|
if row.kind == "cssrs_since_contact":
|
|
# Recent and Since Last Contact are different instruments. No silent substitution.
|
|
return await manual_review(ctx, row, "cssrs_since_contact_required")
|
|
try:
|
|
q = next_question(row)
|
|
# Validate preserved short answers against a newly answered full-form item 1.
|
|
score_alcohol(group_answers(a, "rus_audit"), short=False, sex=row.details.get("sex"))
|
|
except ValueError:
|
|
return await manual_review(ctx, row, "answer_conflict")
|
|
if q:
|
|
replies = [question_reply(row, q)]
|
|
if c.complete and c.risk == "low" and not row.details.get("plan_shown"):
|
|
from app.dialog.actions import safety_plan_reply
|
|
|
|
replies.insert(0, await safety_plan_reply(ctx))
|
|
row.details = {**row.details, "plan_shown": True}
|
|
return Step(replies=replies, next_state="intake.collect", data={"questionnaire_record": row.id})
|
|
extra = {}
|
|
if a.get("side_effects") == 1:
|
|
await service.alert(
|
|
ctx.session,
|
|
ctx.token,
|
|
row.id + ":side_effects_review",
|
|
"side_effects_review",
|
|
ctx.now,
|
|
params={"record_id": row.id},
|
|
)
|
|
if row.kind == "intake":
|
|
result = service.decision(row)
|
|
if result.status != "ready":
|
|
return await manual_review(ctx, row, "intake_incomplete_review")
|
|
extra = {"decision": asdict(result), "totals": service.summary(row)}
|
|
if a.get("recent_trauma") == 1:
|
|
extra["recent_trauma"] = True
|
|
if result.primary_route != "self_help" or result.flags or result.parallel_routes or a.get("recent_trauma"):
|
|
await service.alert(
|
|
ctx.session,
|
|
ctx.token,
|
|
row.id + ":route",
|
|
"intake_route_review",
|
|
ctx.now,
|
|
params={"record_id": row.id, **extra},
|
|
)
|
|
elif row.kind == "diary":
|
|
history = {
|
|
r.day: r.answers
|
|
for r in await service.records(ctx.session, ctx.token)
|
|
if r.kind == "diary" and r.status == "complete"
|
|
}
|
|
history[row.day] = a
|
|
flags = diary_flags(history, row.day)
|
|
extra["flags"] = flags
|
|
for flag in flags:
|
|
await service.alert(
|
|
ctx.session, ctx.token, row.id + ":" + flag, flag, ctx.now, params={"record_id": row.id}
|
|
)
|
|
if flag == "phq9_due_mood":
|
|
await service.queue_task(ctx.session, ctx.token, "phq9", row.id + ":phq9", ctx.now)
|
|
else:
|
|
extra["totals"] = service.summary(row)
|
|
if row.kind in ("phq9", "pcl5"):
|
|
sid = row.kind
|
|
total = extra["totals"][sid]["total"]
|
|
priority = getattr(policy, sid + "_priority_at")
|
|
specialist = getattr(policy, sid + "_specialist_at")
|
|
extra["route"] = (
|
|
"priority_psychiatrist"
|
|
if total >= priority
|
|
else ("psychiatrist_or_psychotherapist" if total >= specialist else "review_with_treating_clinician")
|
|
)
|
|
prior = next(
|
|
(
|
|
r
|
|
for r in reversed(await service.records(ctx.session, ctx.token))
|
|
if r.id != row.id
|
|
and r.status == "complete"
|
|
and r.version == row.version
|
|
and r.details.get("totals", {}).get(sid, {}).get("complete")
|
|
and (sid != "pcl5" or r.details.get("weekly_pcl", False) == row.details.get("weekly_pcl", False))
|
|
),
|
|
None,
|
|
)
|
|
if prior:
|
|
change = total - prior.details["totals"][sid]["total"]
|
|
extra["delta"] = {
|
|
"baseline_record": prior.id,
|
|
"value": change,
|
|
"retriage": change >= getattr(load_runtime(), sid + "_retriage_delta"),
|
|
}
|
|
if row.kind == "rus_audit_s" and not row.details.get("sex"):
|
|
return await manual_review(ctx, row, "alcohol_sex_unknown")
|
|
if extra["totals"]["rus_audit"].get("zone") == 4:
|
|
extra["parallel_routes"] = ["narcologist"]
|
|
retriage = extra.get("delta", {}).get("retriage", False)
|
|
await service.alert(
|
|
ctx.session,
|
|
ctx.token,
|
|
row.id + ":followup",
|
|
"followup_retriage_required" if retriage else "followup_review",
|
|
ctx.now,
|
|
level=AlertLevel.HIGH if retriage or extra.get("route") == "priority_psychiatrist" else AlertLevel.MODERATE,
|
|
params={"record_id": row.id, **extra},
|
|
)
|
|
await service.finish(ctx, row, "complete", extra=extra)
|
|
from app.dialog.flows.core import menu_reply
|
|
|
|
replies = [menu_reply(ctx, texts()["done"])]
|
|
alcohol_total = extra.get("totals", {}).get("rus_audit", {})
|
|
if alcohol_total.get("zone") in (2, 3):
|
|
replies.insert(0, Reply(text=texts()["alcohol_advice"], kind="info"))
|
|
return Step(replies=replies, next_state="idle", data={"questionnaire_record": None})
|
|
|
|
|
|
async def start(ctx, kind="intake", task=None):
|
|
if not allowed(ctx) or crisis_open(ctx.data):
|
|
if task:
|
|
return Step()
|
|
return Step(replies=[Reply(text=texts()["unavailable"], kind="info")])
|
|
existing = await service.records(ctx.session, ctx.token)
|
|
active = next((r for r in reversed(existing) if r.status in ("active", "paused")), None)
|
|
if active:
|
|
if task and task.record_id != active.id:
|
|
return Step()
|
|
if (
|
|
active.policy_version != load_policy().version
|
|
or active.version != content()["version"]
|
|
or active.details.get("runtime_version") != load_runtime().version
|
|
):
|
|
return await manual_review(ctx, active, "questionnaire_version_changed")
|
|
active.status = "active"
|
|
return await advance(ctx, active)
|
|
if ctx.data.get("intake_review_required") and kind != "cssrs_since_contact":
|
|
if task:
|
|
return Step()
|
|
return Step(replies=[Reply(text=texts()["review"], kind="info")])
|
|
row = await service.create_record(ctx, kind, task=task)
|
|
return await advance(ctx, row)
|
|
|
|
|
|
@on_enter("intake.start")
|
|
async def enter(ctx):
|
|
return await start(ctx)
|
|
|
|
|
|
@trigger("button:intake:start")
|
|
@trigger("button:intake:resume")
|
|
async def start_button(ctx, event):
|
|
return await start(ctx)
|
|
|
|
|
|
@trigger("button:menu:checkin", replace=True)
|
|
@trigger("system:daily_checkin", replace=True)
|
|
async def checkin(ctx, event):
|
|
if not intake_enabled():
|
|
return Step(replies=[Reply(text=ctx.texts.idle.stub_checkin)])
|
|
if event.params.get("questionnaire_task"):
|
|
task = await ctx.session.get(QuestionnaireTask, event.params["questionnaire_task"])
|
|
if task is None or task.token != ctx.token or task.status not in ("pending", "active"):
|
|
return Step()
|
|
return await start(ctx, task.kind, task)
|
|
existing = await service.records(ctx.session, ctx.token)
|
|
if not any(r.kind == "intake" and r.status == "complete" for r in existing):
|
|
return await start(ctx)
|
|
day = ctx.now.astimezone(ZoneInfo(load_runtime().timezone)).date()
|
|
if any(r.kind == "diary" and r.day == day and r.status == "complete" for r in existing):
|
|
return Step(replies=[Reply(text=texts()["done"], kind="info")])
|
|
risk_task = await service.queue_task(ctx.session, ctx.token, "cssrs_since_contact", f"cssrs:{day}", ctx.now)
|
|
if risk_task.status in ("pending", "active"):
|
|
await service.queue_task(ctx.session, ctx.token, "diary", f"diary:{day}", ctx.now)
|
|
return await start(ctx, "cssrs_since_contact", risk_task)
|
|
return await start(ctx, "diary")
|
|
|
|
|
|
@state("intake")
|
|
async def collect(ctx, event: Event):
|
|
if not allowed(ctx):
|
|
return Step(replies=[Reply(text=texts()["unavailable"], kind="info")], next_state="idle")
|
|
row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
|
|
if row is None or row.token != ctx.token or row.status != "active":
|
|
return Step(next_state="idle")
|
|
if (
|
|
row.policy_version != load_policy().version
|
|
or row.version != content()["version"]
|
|
or row.details.get("runtime_version") != load_runtime().version
|
|
):
|
|
return await manual_review(ctx, row, "questionnaire_version_changed")
|
|
if event.type == "system":
|
|
task_id = event.params.get("questionnaire_task")
|
|
if task_id and row.details.get("task_id") == task_id:
|
|
return await advance(ctx, row)
|
|
return Step()
|
|
if event.button == "intake:pause":
|
|
row.status = "paused"
|
|
return Step(
|
|
replies=[Reply(text=texts()["paused"], buttons=[[Button(label=texts()["resume"], data="intake:resume")]])],
|
|
next_state="idle",
|
|
)
|
|
q = next_question(row)
|
|
if q is None:
|
|
return await advance(ctx, row)
|
|
prefix = f"iq:{row.id[:12]}:{q.id}:"
|
|
value = None
|
|
valid = False
|
|
if event.type == "button" and (event.button or "").startswith(prefix):
|
|
index = event.button[len(prefix) :]
|
|
if index == "skip" and q.optional:
|
|
valid = True
|
|
elif index.isdigit() and int(index) < len(q.options):
|
|
value, valid = q.options[int(index)][0], True
|
|
elif event.type == "text" and q.numeric:
|
|
try:
|
|
value = float(str(ctx.text).replace(",", "."))
|
|
valid = math.isfinite(value) and 0 <= value <= 24
|
|
except ValueError:
|
|
pass
|
|
elif event.type == "text" and q.optional:
|
|
value, valid = str(ctx.text)[:2000], True
|
|
if not valid:
|
|
return Step(replies=[Reply(text=texts()["numeric" if q.numeric else "invalid"]), question_reply(row, q)])
|
|
service.save_answer(row, q, value, ctx.now)
|
|
return await advance(ctx, row)
|