feat(web): two-stage questionnaires, screening result and diary by result

- Sign-in: centred ЕМИАС card without the sidebar; anonymous entry asks «Депрессия / ПТСР».
- Stage 1 is the same for everyone and follows the fixed order of the MVP questionnaire:
  PHQ-2 → trauma filter/TSQ → RUS-AUDIT-S → C-SSRS → context; stage 2 only where stage 1
  is positive. Starts without a profile.
- After stage 1 the patient sees a preliminary result (screening, not a diagnosis), stored
  with the record; it switches on the diary modules (depression / PTSD).
- Stage 2 diary: morning (sleep, sleep quality, alcohol) and evening (mood, medication,
  3 PHQ-9 items in rotation, short PCL-5); C-SSRS since last contact only under risk;
  flags on weekly means against the personal norm; scheduled full scales with change
  (remission, response, worsening).
- Questionnaire log: GET /me/{token}/results and the «Мои результаты» screen.
- Questionnaires centred without the sidebar and without scale names; «Продолжить позже»
  goes home; confirmations are toasts; free-text comments are optional.
- Demo policy config/questionnaire_demo.yaml (INTAKE_DEMO only, not a clinical approval).

ADR-012.

Refs #77

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-04 15:14:09 +03:00
co-authored by Claude Opus 5.5
parent 6e7c89900b
commit 290b0d6151
36 changed files with 1900 additions and 691 deletions
+8 -7
View File
@@ -9,12 +9,11 @@ the versioned sources in config/.
from functools import lru_cache
from typing import Any, Literal
import yaml
from pydantic import BaseModel
from app.config import get_settings
from app.core.cssrs_content import load_cssrs_russian
from app.core.intake_questions import content, texts
from app.core.intake_questions import DIARY_PARTS, content, diary_config, texts
from app.core.scales import load_scale
Section = Literal["safety", "mood", "trauma", "alcohol", "about", "diary", "scale"]
@@ -113,13 +112,15 @@ def catalog() -> Catalog:
add(FormInfo(id="trauma", title="Травматическое событие", section="trauma"), ["trauma", "recent_trauma"])
add(FormInfo(id="context", title="Общие вопросы", section="about"), ["sex", "tbi", "psychotropics"])
diary = yaml.safe_load((get_settings().config_dir / "instruments/diary.yaml").read_text(encoding="utf-8"))
diary_items = [item for key in ("core", "ptsd", "depression", "suicide") for item in diary.get(key, [])]
diary = diary_config()
core = [item for part in DIARY_PARTS for item in diary[part]]
depression = [f"dphq_{i['id'].rsplit('_', 1)[1]}" for i in intake_forms[diary["depression"]["form"]]["items"]]
ptsd = [f"dpcl_{i.rsplit('_', 1)[1]}" for i in diary["ptsd"]["items"]]
add(
FormInfo(id="diary", title="Дневник самочувствия", section="diary"),
[i["id"] for i in diary_items],
optional={i["id"] for i in diary_items if i.get("optional")},
numeric={i["id"] for i in diary_items if i.get("numeric")},
[i["id"] for i in core] + depression + ptsd,
optional={i["id"] for i in core if i.get("optional")},
numeric={i["id"] for i in core if i.get("numeric")},
)
# Scales of the onboarding / scheduled sessions (`scale:<id>:...` buttons): the form header only.
+90 -41
View File
@@ -18,6 +18,7 @@ class Question:
period: str = "context"
numeric: bool = False
optional: bool = False
maximum: float = 24
def content() -> dict:
@@ -53,18 +54,10 @@ def cssrs_answers(answers: dict) -> dict[str, int]:
def next_intake(answers: dict, policy: IntakePolicy, sex: str | None) -> Question | None:
c = score_cssrs_recent(cssrs_answers(answers))
if c.missing:
form = load_cssrs_russian()
item = next(i for i in form.items if i.id == c.missing[0])
period = texts()["periods"][item.recall_period]
# Guidance is interviewer-facing; retained in the source and record version.
return Question(
"cssrs_" + item.id,
period + "\n\n" + item.text,
tuple((o.value, o.label) for o in form.options),
item.recall_period,
)
"""Stage 1 in the fixed order of «Опросники для MVP»: PHQ-2, trauma filter → TSQ, RUS-AUDIT-S, C-SSRS screener,
then the context questions; stage 2 (rest of PHQ-9, PCL-5, full RUS-AUDIT) only where stage 1 is positive.
Items inside a scale keep the order of the source. The risk stop rules run after every answer (flows/intake.py),
so a positive answer interrupts the questionnaire wherever it comes."""
for item in ("phq9_1", "phq9_2"):
if item not in answers:
return scale_question("phq9", item)
@@ -81,6 +74,18 @@ def next_intake(answers: dict, policy: IntakePolicy, sex: str | None) -> Questio
for item in load_scale("rus_audit_s").item_ids:
if item not in answers:
return scale_question("rus_audit_s", item)
c = score_cssrs_recent(cssrs_answers(answers))
if c.missing:
form = load_cssrs_russian()
item = next(i for i in form.items if i.id == c.missing[0])
period = texts()["periods"][item.recall_period]
# Guidance is interviewer-facing; retained in the source and record version.
return Question(
"cssrs_" + item.id,
period + "\n\n" + item.text,
tuple((o.value, o.label) for o in form.options),
item.recall_period,
)
if sex is None and "sex" not in answers:
return Question("sex", texts()["sex"], (("f", "Женский"), ("m", "Мужской"), ("unknown", "Не указывать")))
for key in ("tbi", "psychotropics"):
@@ -108,34 +113,78 @@ def next_intake(answers: dict, policy: IntakePolicy, sex: str | None) -> Questio
return None
def diary_questions(
modules: list[str], *, suicide_monitoring: bool, weekday: int, day_index: int | None = None
) -> tuple[Question, ...]:
config = yaml.safe_load((get_settings().config_dir / "instruments/diary.yaml").read_text())
selected = ["core"]
detailed = [name for name in ("veteran", "depression") if name in modules]
if len(detailed) == 2:
detailed = [detailed[(day_index if day_index is not None else weekday) % 2]]
if "veteran" in detailed:
selected.append("ptsd")
if "depression" in detailed:
selected.append("depression")
if suicide_monitoring or "depression" in modules:
selected.append("suicide")
questions = []
for section in selected:
for item in config[section]:
if item.get("weekly") and weekday != 0:
continue
questions.append(
Question(
item["id"],
item["text"],
tuple((o["value"], o["label"]) for o in item.get("options", [])),
"past_week" if item.get("weekly") else "past_day",
item.get("numeric", False),
item.get("optional", False),
def cssrs_since_contact_question(answers: dict) -> Question | None:
"""Daily suicide-risk module of stage 2: the C-SSRS screener asked «since the last contact».
The official Russian Since Last Contact form is not in the repository yet (docs/clinical/README.md, C06), so the
items of the Recent form are asked verbatim under the since-last-contact period. Behaviour since the last contact
is recent by definition: the answer to q6 also fills q6_recent (derived), so the recent triage applies.
"""
c = score_cssrs_recent(cssrs_answers(answers))
missing = [k for k in c.missing if k != "q6_recent"]
if not missing:
return None
form = load_cssrs_russian()
item = next(i for i in form.items if i.id == missing[0])
return Question(
"cssrs_" + item.id,
texts()["periods"]["since_last_contact"] + "\n\n" + item.text,
tuple((o.value, o.label) for o in form.options),
"since_last_contact",
)
DIARY_PARTS = ("morning", "evening")
DIARY_KINDS = ("diary_morning", "diary_evening")
def diary_config() -> dict:
return yaml.safe_load((get_settings().config_dir / "instruments/diary.yaml").read_text(encoding="utf-8"))
def diary_questions(part: str, modules: list[str], *, day_index: int = 0) -> tuple[Question, ...]:
"""Stage 2 diary of one part of the day: the core block plus, in the evening, the modules of the profile.
PTSD («veteran») and depression run side by side when both are present. The depression module asks three
PHQ-9 items a day in rotation by `day_index` (days since the first diary). Ids are prefixed (`dphq_`, `dpcl_`)
so diary answers never mix with the full scales.
"""
if part not in DIARY_PARTS:
raise ValueError("unknown diary part")
config = diary_config()
questions = [
Question(
item["id"],
item["text"],
tuple((o["value"], o["label"]) for o in item.get("options", [])),
"past_day",
item.get("numeric", False),
item.get("optional", False),
item.get("max", 24),
)
for item in config[part]
]
if part == "evening":
forms = content()["forms"]
if "depression" in modules:
module = config["depression"]
items = forms[module["form"]]["items"]
per_day = module["items_per_day"]
start = (day_index % (len(items) // per_day)) * per_day
options = tuple((o["value"], o["label"]) for o in module["options"])
for item in items[start : start + per_day]:
number = item["id"].rsplit("_", 1)[1]
questions.append(
Question(f"dphq_{number}", module["question"] + "\n\n" + item["text"], options, "today")
)
if "veteran" in modules:
module = config["ptsd"]
form = forms[module["form"]]
by_id = {i["id"]: i["text"] for i in form["items"]}
options = tuple((o["value"], o["label"]) for o in form["options"])
for item_id in module["items"]:
number = item_id.rsplit("_", 1)[1]
questions.append(
Question(f"dpcl_{number}", module["question"] + "\n\n" + by_id[item_id], options, "today")
)
)
questions.sort(key=lambda q: q.id != "death_thoughts")
return tuple(questions)
+73 -31
View File
@@ -1,6 +1,5 @@
"""Versioned release gate and deterministic follow-up planning."""
import os
from calendar import monthrange
from datetime import date, timedelta
from pathlib import Path
@@ -11,7 +10,7 @@ import yaml
from pydantic import BaseModel, ConfigDict, Field, model_validator
from app.config import get_settings
from app.core.questionnaire import Approval, load_policy
from app.core.questionnaire import Approval, demo_mode, load_policy
class RuntimePolicy(BaseModel):
@@ -22,13 +21,20 @@ class RuntimePolicy(BaseModel):
content_review: Approval | None = None
contact_review: Approval | None = None
timezone: str
daily_hour: int = Field(ge=0, le=23)
mood_baseline_days: int = Field(ge=2)
mood_bad_days: int = Field(ge=1)
mood_drop: int = Field(ge=1, le=10)
morning_hour: int = Field(ge=0, le=23)
evening_hour: int = Field(ge=0, le=23)
norm_days: int = Field(ge=2)
week_days: int = Field(ge=2)
min_norm_values: int = Field(ge=1)
min_week_values: int = Field(ge=1)
mood_drop: float = Field(gt=0, le=10)
depression_item_rise: float = Field(gt=0, le=3)
ptsd_sum_rise: float = Field(gt=0, le=16)
sleep_bad_days: int = Field(ge=1)
sleep_hours_below: float = Field(gt=0, le=24)
nightmares_at: int = Field(ge=0, le=4)
alcohol_heavy_portions: float = Field(gt=0)
alcohol_heavy_days: int = Field(ge=1)
adherence_below: float = Field(gt=0, le=1)
missed_medication_days: int = Field(ge=1)
high_risk_silence_days: int = Field(ge=1)
medication_days: tuple[int, ...]
@@ -45,6 +51,8 @@ class RuntimePolicy(BaseModel):
set(row) != {"phq9", "pcl5"} for row in self.route_intervals.values()
):
raise ValueError("all route and scale intervals must be defined")
if self.morning_hour >= self.evening_hour:
raise ValueError("the morning diary must come before the evening one")
if not self.medication_days or tuple(sorted(set(self.medication_days))) != self.medication_days:
raise ValueError("medication days must be nonempty and strictly increasing")
if self.medication_days[0] < 1 or any(v < 1 for row in self.route_intervals.values() for v in row.values()):
@@ -71,7 +79,7 @@ def demo_intake() -> bool:
It never touches the clinical release gate in config/intake_runtime.yaml; production runs keep the forms off
until the approvals are recorded there.
"""
return os.environ.get("INTAKE_DEMO", "").lower() in ("1", "true", "yes") and get_settings().demo_clock
return demo_mode()
def intake_enabled() -> bool:
@@ -125,39 +133,73 @@ def due_scales(
return tuple(sorted(due))
def diary_flags(history: dict[date, dict], today: date, policy: RuntimePolicy | None = None) -> tuple[str, ...]:
"""Consecutive calendar dates; missing data never become zero or missed doses.
def _mean(values: list[float]) -> float:
return sum(values) / len(values)
Baseline is the seven dates BEFORE the three-day deterioration window;
every baseline value is required. Nightmares OR short sleep must hold on
each of three nights. These candidate definitions require signed review.
def diary_flags(
history: dict[date, dict],
today: date,
policy: RuntimePolicy | None = None,
*,
medication_active: bool = False,
) -> tuple[str, ...]:
"""Stage 2 diary flags («Этап 2», p.4): weekly means against the personal norm, never single days.
`history` holds the merged morning + evening answers per date. The norm is the first `norm_days` of
observation; trends are checked only after it. Missing data never become zero or missed doses.
"""
p = policy or load_runtime()
flags = []
flags: list[str] = []
current = history.get(today, {})
if current.get("death_thoughts") == 1:
flags.append("suicide_assessment_now")
recent = [history.get(today - timedelta(days=i), {}) for i in range(p.mood_bad_days)]
baseline = [
history.get(today - timedelta(days=i), {}).get("mood")
for i in range(p.mood_bad_days, p.mood_bad_days + p.mood_baseline_days)
]
if all(x is not None for x in baseline) and all(x.get("mood") is not None for x in recent):
mean = sum(baseline) / len(baseline)
if all(mean - x["mood"] >= p.mood_drop for x in recent):
flags.append("phq9_due_mood")
start = min(history) if history else today
norm = [history[d] for d in sorted(history) if (d - start).days < p.norm_days]
week_start = today - timedelta(days=p.week_days - 1)
week = [history[d] for d in sorted(history) if week_start <= d <= today]
trend = (week_start - start).days >= p.norm_days
def values(rows: list[dict], key) -> list[float]:
return [v for r in rows if (v := key(r)) is not None]
def risen(key, by: float, *, min_norm: int, min_week: int, drop: bool = False) -> bool:
before, now = values(norm, key), values(week, key)
if not trend or len(before) < min_norm or len(now) < min_week:
return False
change = _mean(before) - _mean(now) if drop else _mean(now) - _mean(before)
return change >= by
if risen(lambda r: r.get("mood"), p.mood_drop, min_norm=p.min_norm_values, min_week=p.min_week_values, drop=True):
flags.append("phq9_due_mood")
# Depression module: each PHQ-9 item comes every third day, so a weekly mean has ~2 values per item.
for item in range(1, 10):
if risen(lambda r, i=item: r.get(f"dphq_{i}"), p.depression_item_rise, min_norm=2, min_week=2):
flags.append("phq9_due_depression")
break
# PTSD module: only the sum of the four items is interpreted, never single clusters.
ptsd_items = ("dpcl_1", "dpcl_7", "dpcl_10", "dpcl_18")
def ptsd_sum(r):
return sum(r[k] for k in ptsd_items) if all(k in r for k in ptsd_items) else None
if risen(ptsd_sum, p.ptsd_sum_rise, min_norm=p.min_norm_values, min_week=p.min_week_values):
flags.append("pcl5_due_ptsd")
nights = [history.get(today - timedelta(days=i), {}) for i in range(p.sleep_bad_days)]
if all(
(r.get("nightmares") is not None and r["nightmares"] >= p.nightmares_at)
or (r.get("sleep_hours") is not None and r["sleep_hours"] < p.sleep_hours_below)
(r.get("sleep_hours") is not None and r["sleep_hours"] < p.sleep_hours_below) or r.get("sleep_quality") == 0
for r in nights
):
flags.append("sleep_review")
heavy = [r for r in week if (r.get("alcohol_portions") or 0) >= p.alcohol_heavy_portions]
if medication_active and (current.get("alcohol_portions") or 0) >= p.alcohol_heavy_portions:
flags.append("alcohol_on_medication")
if len(heavy) >= p.alcohol_heavy_days:
flags.append("rus_audit_due_alcohol")
doses = [history.get(today - timedelta(days=i), {}) for i in range(p.missed_medication_days)]
if all(r.get("medication") == "none" for r in doses):
if all(r.get("medication") == "no" for r in doses):
flags.append("medication_contact")
if current.get("side_effects") == 1:
taken = [r["medication"] for r in week if r.get("medication") in ("yes", "no")]
if len(taken) >= p.min_week_values and taken.count("yes") / len(taken) < p.adherence_below:
flags.append("adherence_review")
if current.get("medication_reason") == "side_effects":
flags.append("side_effects_review")
if current.get("medication") == "partial":
flags.append("partial_medication_review")
return tuple(flags)
+165
View File
@@ -0,0 +1,165 @@
"""Patient-facing interpretation of questionnaire results: deterministic rules, texts from config/texts/results.yaml.
Stage 1 («Опросники для MVP»): screening levels per scale → a preliminary conclusion and the diary modules of stage 2.
Stage 2 («Этап 2»): full scales on schedule (level + change: response, remission, worsening) and diary flags.
Nothing here diagnoses: the conclusion is a screening result, the diagnosis is the doctor's (ADR-007, guardrail 3).
"""
from typing import Any
import yaml
from app.config import get_settings
from app.core.intake_questions import content
from app.core.questionnaire import IntakeDecision, IntakePolicy
# Stage 1 finding → diary module of stage 2 (PTSD is the «veteran» module of the profile).
FINDING_MODULE = {"depression_suspected": "depression", "ptsd_suspected": "veteran"}
# Remission and improvement cut-offs of the full scales («Этап 2», p.5: Kroenke 2001; Marx et al., 2022).
PHQ9_REMISSION_BELOW = 5
PHQ9_IMPROVED_BY = 5
PCL5_RESPONSE_BY = 10
PCL5_IMPROVED_BELOW = 28
def texts() -> dict[str, Any]:
return yaml.safe_load((get_settings().config_dir / "texts/results.yaml").read_text(encoding="utf-8"))
def render(interpretation: dict[str, Any]) -> str:
"""Plain text of an interpretation for the dialog reply (title, lines, then the disclaimer if any)."""
parts = [interpretation["title"], *interpretation.get("lines", [])]
if interpretation.get("disclaimer"):
parts.append(interpretation["disclaimer"])
return "\n\n".join(p for p in parts if p)
def _alcohol_zone(totals: dict) -> int | None:
full, short = totals.get("rus_audit") or {}, totals.get("rus_audit_s") or {}
return full.get("zone") if full.get("complete") else short.get("zone")
def intake_findings(totals: dict, answers: dict, policy: IntakePolicy) -> list[str]:
"""Screening findings of stage 1, in the order they are shown."""
found = []
phq = totals.get("phq9") or {}
if phq.get("complete"):
if phq["total"] >= policy.phq9_specialist_at:
found.append("depression_suspected")
elif phq["total"] >= policy.phq9_psychologist_at:
found.append("depression_mild")
pcl, tsq = totals.get("pcl5") or {}, totals.get("tsq") or {}
if pcl.get("complete") and pcl["total"] >= policy.pcl5_specialist_at:
found.append("ptsd_suspected")
elif answers.get("trauma") == 1 and tsq.get("complete") and tsq["total"] >= policy.tsq_expand_at:
found.append("ptsd_signs")
zone = _alcohol_zone(totals)
if zone == 4:
found.append("alcohol_high")
elif zone in (2, 3):
found.append("alcohol_risky")
if (totals.get("cssrs_recent") or {}).get("risk") == "low":
found.append("suicide_low")
if answers.get("tbi") == 1:
found.append("tbi")
return found
def intake_interpretation(totals: dict, answers: dict, decision: IntakeDecision, policy: IntakePolicy) -> dict:
t = texts()
found = intake_findings(totals, answers, policy)
modules = [FINDING_MODULE[f] for f in found if f in FINDING_MODULE]
head = t["intake"]["attention" if found else "ok"]
lines = [head["text"]]
lines += [f"{t['findings'][f]['title']}. {t['findings'][f]['text']}" for f in found]
if decision.primary_route in t["routes"]:
lines.append(t["routes"][decision.primary_route])
names = ", ".join(t["modules"]["names"][m] for m in modules)
lines.append(t["modules"]["diary"].format(modules=t["modules"]["with"].format(names=names) if names else ""))
return {
"kind": "intake",
"title": head["title"],
"lines": lines,
"codes": found or ["ok"],
"findings": [{"code": f, "title": t["findings"][f]["title"]} for f in found],
"route": decision.primary_route,
"modules": modules,
"disclaimer": t["disclaimer"],
}
def _phq9_level(total: int) -> str:
band = next(b for b in content()["forms"]["phq9"]["severity"] if b["min"] <= total <= b["max"])
return band["code"]
def scale_interpretation(
kind: str,
totals: dict,
*,
baseline: int | None,
previous: int | None,
policy: IntakePolicy,
retriage_delta: int,
) -> dict:
"""A scheduled full scale: level, then the change against the start of observation and the last measurement."""
t = texts()["scales"]
if kind == "rus_audit_s":
zone = _alcohol_zone(totals)
result = totals.get("rus_audit") if (totals.get("rus_audit") or {}).get("complete") else totals["rus_audit_s"]
level = t[kind]["levels"][f"zone_{zone}" if zone else "unknown"]
line = t["score"].format(title=t[kind]["title"], total=result.get("total"), level=level)
return {
"kind": kind,
"title": t[kind]["title"],
"lines": [line],
"codes": [f"zone_{zone}" if zone else "unknown"],
}
total = totals[kind]["total"]
pcl5_level = "probable" if total >= policy.pcl5_specialist_at else "below"
code = _phq9_level(total) if kind == "phq9" else pcl5_level
lines = [t["score"].format(title=t[kind]["title"], total=total, level=t[kind]["levels"][code])]
change = None
if previous is not None and total - previous >= retriage_delta:
change = "worse"
elif baseline is not None:
if kind == "phq9":
if total < PHQ9_REMISSION_BELOW:
change = "remission"
elif baseline > 0 and total <= baseline / 2:
change = "response"
elif baseline - total >= PHQ9_IMPROVED_BY:
change = "improved"
else:
change = "stable"
else:
if baseline - total >= PCL5_RESPONSE_BY:
change = "response"
elif total < PCL5_IMPROVED_BELOW <= baseline:
change = "improved"
else:
change = "stable"
if change:
lines.append(t["change"][change])
return {"kind": kind, "title": t[kind]["title"], "lines": lines, "codes": [code, *([change] if change else [])]}
def diary_interpretation(part: str, flags: tuple[str, ...] | list[str]) -> dict:
t = texts()["diary"]
shown = [f for f in flags if f in t["flags"]]
return {
"kind": f"diary_{part}",
"title": t["saved"].format(part=t["parts"][part]),
"lines": [t["flags"][f] for f in shown],
"codes": list(flags),
}
def cssrs_interpretation(risk: str | None) -> dict:
t = texts()["cssrs"]
return {
"kind": "cssrs_since_contact",
"title": t["low" if risk == "low" else "none"],
"lines": [],
"codes": [risk or "none"],
}
+11 -1
View File
@@ -8,6 +8,7 @@ the returned status. Pending clinical policy never authorizes routine continuati
from __future__ import annotations
import os
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
@@ -69,8 +70,17 @@ class IntakePolicy(BaseModel):
return self
def demo_mode() -> bool:
"""Demo-only switch (INTAKE_DEMO=1 with the virtual clock): questionnaires on synthetic data.
It selects `config/questionnaire_demo.yaml` (marked as NOT a clinical approval) instead of the real policy; the
real policy and the release gate in config/intake_runtime.yaml are never touched.
"""
return os.environ.get("INTAKE_DEMO", "").lower() in ("1", "true", "yes") and get_settings().demo_clock
def load_policy(path: Path | None = None) -> IntakePolicy:
path = path or get_settings().config_dir / "questionnaire.yaml"
path = path or get_settings().config_dir / ("questionnaire_demo.yaml" if demo_mode() else "questionnaire.yaml")
return IntakePolicy.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))