"""Clinical DB schema: pseudonymised health data. SPEC §5, ADR-002, ADR-003, ADR-006. Every row is keyed by `token` (32 hex chars of HMAC(PSEUDO_KEY, SNILS)), never by name/SNILS/phone/chat id. Free text is stored only after `scrub()` (`*_scrubbed` columns). This module must not import `app.db.identity` (guardrail #5, checked in tests/test_db.py); `audit_log` also lives in this DB. Business timestamps have no server defaults: callers pass time from `app.core.clock` (ADR-005), so the demo virtual clock is reflected consistently in data, rules and the dashboard. """ from datetime import date, datetime from typing import Any from sqlalchemy import ( BigInteger, Boolean, Date, Float, ForeignKey, Index, Integer, SmallInteger, String, Text, UniqueConstraint, ) from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from app.db.types import JSONType, TZDateTime TOKEN_LEN = 32 # BIGSERIAL on Postgres; SQLite autoincrements only INTEGER PRIMARY KEY. BigId = BigInteger().with_variant(Integer(), "sqlite") class ClinicalBase(DeclarativeBase): """Metadata of the clinical DB (incl. audit_log); never mixed with identity metadata.""" class PatientProfile(ClinicalBase): """Clinical view of a patient: no name, SNILS or contacts — only what monitoring needs.""" __tablename__ = "patient_profiles" token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True) doctor_id: Mapped[str] = mapped_column(String(64), index=True) modules: Mapped[list[str]] = mapped_column(JSONType, default=list) # veteran | depression | family therapy_start: Mapped[date | None] = mapped_column(Date) next_visit: Mapped[date | None] = mapped_column(Date) birth_year: Mapped[int | None] = mapped_column(SmallInteger) sex: Mapped[str | None] = mapped_column(String(1)) # Set by rules (e.g. after an acute episode): denser check-ins and lower silence thresholds. enhanced_monitoring: Mapped[bool] = mapped_column(Boolean, default=False) # Cache for the dashboard list; the source of truth is open alerts (colour is derived, ADR-006). status_color: Mapped[str | None] = mapped_column(String(8)) class ItemResponse(ClinicalBase): """Single scale item answer (PHQ-9, C-SSRS...) from buttons, LLM extraction or clinician.""" __tablename__ = "item_responses" __table_args__ = (Index("ix_item_responses_token_ts", "token", "ts"),) id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) token: Mapped[str] = mapped_column(String(TOKEN_LEN)) ts: Mapped[datetime] = mapped_column(TZDateTime) scale: Mapped[str] = mapped_column(String(16)) # phq9 | gad7 | pcl5 | cssrs | auditc ... item_id: Mapped[str] = mapped_column(String(32)) score: Mapped[int | None] = mapped_column(SmallInteger) confidence: Mapped[float | None] = mapped_column(Float) evidence: Mapped[str | None] = mapped_column(Text) # scrubbed phrase supporting an LLM-extracted score source: Mapped[str] = mapped_column(String(16)) # llm | button | clinician class ScaleResult(ClinicalBase): __tablename__ = "scale_results" __table_args__ = (Index("ix_scale_results_token_ts", "token", "ts"),) id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) token: Mapped[str] = mapped_column(String(TOKEN_LEN)) ts: Mapped[datetime] = mapped_column(TZDateTime) scale: Mapped[str] = mapped_column(String(16)) period_start: Mapped[date | None] = mapped_column(Date) period_end: Mapped[date | None] = mapped_column(Date) total: Mapped[int] = mapped_column(SmallInteger) severity: Mapped[str] = mapped_column(String(32)) complete: Mapped[bool] = mapped_column(Boolean) # PHQ-9 item 9 (suicidal ideation) kept separately: any value >= 1 escalates regardless of the total. item9: Mapped[int | None] = mapped_column(SmallInteger) # False = unscheduled administration triggered by a rule (e.g. mood <= 3 for three days, ADR-002). scheduled: Mapped[bool] = mapped_column(Boolean, default=True) flags: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict) class DailyCheckin(ClinicalBase): """One row per patient per (virtual) day, filled in several steps of the daily dialog (ADR-002).""" __tablename__ = "daily_checkins" __table_args__ = (UniqueConstraint("token", "day", name="uq_daily_checkins_token_day"),) id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) token: Mapped[str] = mapped_column(String(TOKEN_LEN)) day: Mapped[date] = mapped_column(Date) mood: Mapped[int | None] = mapped_column(SmallInteger) # 0..10 med_taken: Mapped[bool | None] = mapped_column(Boolean) sleep_bucket: Mapped[str | None] = mapped_column(String(16)) alcohol_bucket: Mapped[str | None] = mapped_column(String(16)) side_effects: Mapped[list[str]] = mapped_column(JSONType, default=list) quit_intent: Mapped[bool | None] = mapped_column(Boolean) # wants to stop medication — doctor flag nightmares: Mapped[bool | None] = mapped_column(Boolean) free_text_scrubbed: Mapped[str | None] = mapped_column(Text) updated_at: Mapped[datetime] = mapped_column(TZDateTime) class Prescription(ClinicalBase): __tablename__ = "prescriptions" id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) token: Mapped[str] = mapped_column(String(TOKEN_LEN), index=True) drug: Mapped[str] = mapped_column(String(128)) # INN (МНН) dose: Mapped[str] = mapped_column(String(64)) schedule: Mapped[dict[str, Any]] = mapped_column(JSONType) # {"times": ["08:00", "20:00"]} start_date: Mapped[date | None] = mapped_column(Date) end_date: Mapped[date | None] = mapped_column(Date) class Dose(ClinicalBase): __tablename__ = "doses" __table_args__ = (Index("ix_doses_token_planned", "token", "planned_at"),) id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) token: Mapped[str] = mapped_column(String(TOKEN_LEN)) prescription_id: Mapped[int] = mapped_column(ForeignKey("prescriptions.id")) planned_at: Mapped[datetime] = mapped_column(TZDateTime) status: Mapped[str] = mapped_column(String(16), default="pending") # taken | missed | skipped | pending answered_at: Mapped[datetime | None] = mapped_column(TZDateTime) side_effects: Mapped[str | None] = mapped_column(Text) class Alert(ClinicalBase): """Output of the deterministic rules engine (ADR-007). Level is stored, UI colour is derived (ADR-006).""" __tablename__ = "alerts" __table_args__ = (Index("ix_alerts_token_ts", "token", "ts"),) id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) token: Mapped[str] = mapped_column(String(TOKEN_LEN)) ts: Mapped[datetime] = mapped_column(TZDateTime) level: Mapped[str] = mapped_column(String(16)) # acute | high | moderate | info (AlertLevel name) code: Mapped[str] = mapped_column(String(64)) # rule id from config/rules.yaml reason: Mapped[str] = mapped_column(Text) params: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict) acked_by: Mapped[str | None] = mapped_column(String(64)) acked_at: Mapped[datetime | None] = mapped_column(TZDateTime) class Message(ClinicalBase): __tablename__ = "messages" __table_args__ = (Index("ix_messages_token_ts", "token", "ts"),) id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) token: Mapped[str] = mapped_column(String(TOKEN_LEN)) ts: Mapped[datetime] = mapped_column(TZDateTime) direction: Mapped[str] = mapped_column(String(3)) # in | out text_scrubbed: Mapped[str] = mapped_column(Text) fsm_state: Mapped[str | None] = mapped_column(String(64)) class DialogState(ClinicalBase): """FSM state of the dialog engine, persisted instead of Redis (ADR-003).""" __tablename__ = "dialog_state" token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True) state: Mapped[str] = mapped_column(String(64)) data: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict) updated_at: Mapped[datetime] = mapped_column(TZDateTime) class SafetyPlan(ClinicalBase): __tablename__ = "safety_plans" token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True) steps: Mapped[list[Any]] = mapped_column(JSONType, default=list) updated_at: Mapped[datetime] = mapped_column(TZDateTime) class ClockState(ClinicalBase): """Single row (id=1): virtual clock offset so the demo timeline survives restarts (ADR-005).""" __tablename__ = "clock_state" id: Mapped[int] = mapped_column(Integer, primary_key=True) offset_seconds: Mapped[float] = mapped_column(Float, default=0.0) class DemoSetting(ClinicalBase): """Presentation switches changed at runtime from the demo console (e.g. the initial screening on/off).""" __tablename__ = "demo_settings" key: Mapped[str] = mapped_column(String(64), primary_key=True) value: Mapped[Any] = mapped_column(JSONType)