diff --git a/scripts/health/dashboard_v5/capture_contract.py b/scripts/health/dashboard_v5/capture_contract.py index b0fc9e2..6d9640c 100644 --- a/scripts/health/dashboard_v5/capture_contract.py +++ b/scripts/health/dashboard_v5/capture_contract.py @@ -154,160 +154,165 @@ def validate_capture_payload(payload: Any) -> dict[str, Any]: required = { "version", "action", "capture_type", "request_version", "idempotency_key", "occurred_at", "ended_at", "data", "attachments", "corrects_entry_id", "withdraws_entry_id", } if set(payload) != required: raise ValueError("invalid capture shape") capture_type = payload["capture_type"] if capture_type not in CAPTURE_TYPES: raise ValueError("invalid capture type") request_version = payload["request_version"] if type(request_version) is not int or not 1 <= request_version <= 1_000_000: raise ValueError("invalid request version") key = payload["idempotency_key"] if not isinstance(key, str) or not re.fullmatch(r"[a-f0-9]{32}", key): raise ValueError("invalid idempotency key") attachments = payload["attachments"] if ( not isinstance(attachments, list) or len(attachments) > 5 or any( not isinstance(token, str) or not TOKEN_RE.fullmatch(token) for token in attachments ) or len(set(attachments)) != len(attachments) ): raise ValueError("invalid attachments") corrects = payload["corrects_entry_id"] withdraws = payload["withdraws_entry_id"] if corrects and ( not isinstance(corrects, str) or not OPAQUE_RE.fullmatch(corrects) ): raise ValueError("invalid correction target") if withdraws and ( not isinstance(withdraws, str) or not OPAQUE_RE.fullmatch(withdraws) ): raise ValueError("invalid withdrawal target") if corrects and withdraws: raise ValueError("ambiguous version action") data = payload["data"] if not isinstance(data, dict): raise ValueError("invalid capture data") occurred = _when(payload["occurred_at"]) ended = _when(payload["ended_at"], optional=True) if ended and ended < occurred: raise ValueError("end before start") common = {"note", "title"} if capture_type == "symptom": expected = common | {"category", "intensity", "count", "body_region", "ongoing"} if ( set(data) != expected or data["category"] not in SYMPTOM_CATEGORIES or data["ongoing"] not in {"yes", "no", "unknown"} ): raise ValueError("invalid symptom") intensity = data["intensity"] if intensity != "unknown" and ( type(intensity) is not int or intensity not in range(4) ): raise ValueError("invalid intensity") count = data["count"] if count is not None and (type(count) is not int or not 0 <= count <= 999): raise ValueError("invalid count") normalized = { "category": data["category"], "intensity": intensity, "count": count, "body_region": _text(data["body_region"], 80), "ongoing": data["ongoing"], } elif capture_type == "medication" and data.get("contract") in ACTION_CONTRACT_VERSIONS: normalized = validate_action_data(data) + if data.get("contract") == "health.medication_action.v2": + if normalized.get("mode") == "correction" and not corrects: + raise ValueError("structured correction requires capture target") + if normalized.get("mode") != "correction" and corrects: + raise ValueError("capture correction target on non-correction") elif capture_type in {"medication", "supplement"}: expected = common | { "status", "plan_id", "name", "amount", "unit", "route", "plan_value_confirmed", "deviation_confirmed", } if set(data) != expected or data["status"] not in STATUSES: raise ValueError("invalid administration") amount = data["amount"] if amount not in {None, ""}: try: amount = float(amount) except (TypeError, ValueError) as exc: raise ValueError("invalid amount") from exc if not math.isfinite(amount) or amount < 0 or amount > 1_000_000: raise ValueError("invalid amount") if ( type(data["plan_value_confirmed"]) is not bool or type(data["deviation_confirmed"]) is not bool ): raise ValueError("invalid confirmation") normalized = { "status": data["status"], "plan_id": _text(data["plan_id"], 80), "name": _text(data["name"], 120, required=True), "amount": amount, "unit": _text(data["unit"], 30), "route": _text(data["route"], 60), "plan_value_confirmed": data["plan_value_confirmed"], "deviation_confirmed": data["deviation_confirmed"], } elif capture_type == "event": event_kind = data.get("event_kind", "generic") if event_kind == "sauna": expected = common | { "event_kind", "duration_minutes", "rounds", "temperature_c", "cooling", "hydration_ml", } if set(data) != expected: raise ValueError("invalid sauna") normalized = { "event_kind": "sauna", "duration_minutes": _bounded_number( data["duration_minutes"], 1, 720, integer=True ), "rounds": _bounded_number( data["rounds"], 1, 20, optional=True, integer=True ), "temperature_c": _bounded_number( data["temperature_c"], 30, 130, optional=True ), "cooling": _text(data["cooling"], 80), "hydration_ml": _bounded_number( data["hydration_ml"], 0, 10_000, optional=True, integer=True ), } elif event_kind == "training": expected = common | { "event_kind", "activity_type", "duration_minutes", "active_kcal", "distance_km", } if set(data) != expected: raise ValueError("invalid training") normalized = { "event_kind": "training", "activity_type": _text(data["activity_type"], 80, required=True), "duration_minutes": _bounded_number( data["duration_minutes"], 1, 1_440, integer=True diff --git a/scripts/health/dashboard_v5/medication_contract.py b/scripts/health/dashboard_v5/medication_contract.py index 9f8c6c6..d5237d7 100644 --- a/scripts/health/dashboard_v5/medication_contract.py +++ b/scripts/health/dashboard_v5/medication_contract.py @@ -1,108 +1,110 @@ """Versioned, fail-closed medication action and read-model helpers.""" from __future__ import annotations import hashlib import hmac import json import re import sqlite3 +from decimal import Decimal, InvalidOperation from typing import Any, Mapping PUBLIC_CONTRACT_VERSION = "health.medication_history.v1" ACTION_CONTRACT_VERSION_V1 = "health.medication_action.v1" ACTION_CONTRACT_VERSION = "health.medication_action.v2" ACTION_CONTRACT_VERSIONS = frozenset({ACTION_CONTRACT_VERSION_V1, ACTION_CONTRACT_VERSION}) ROUTES = frozenset( { "oral", "subcutaneous", "intravenous", "intramuscular", "topical", "inhaled", "other", "unknown", } ) INJECTION_ROUTES = frozenset({"subcutaneous", "intravenous", "intramuscular", "other"}) SIDES = frozenset({"left", "right", "unspecified"}) +DOSAGE_FORMS = frozenset({"Spritze", "Tablette"}) STATUSES = frozenset({"administered", "missed", "corrected"}) CORRECTION_STATUSES = frozenset({"planned", "administered", "missed"}) OPAQUE_RE = re.compile(r"med(?:rx|evt)_[a-f0-9]{24}") REVISION_RE = re.compile(r"[a-f0-9]{64}") EVENT_FIELDS = ( "id", "datum", "medication_name", "dose", "route", "event_type", "scheduled_next_date", "notes", "source", "created_at", "occurred_at", "medication_id", "planned_event_id", "planned_dose_value", "planned_dose_unit", "actual_dose_value", "actual_dose_unit", "route_original", "route_normalized", "injection_region", "injection_side", "injection_detail", "lot_number", "corrects_event_id", "corrected_target_status", "correction_reason", "business_revision", "planned_quantity_value", "planned_dosage_form", "planned_strength", "actual_quantity_value", "actual_dosage_form", "actual_strength", ) PRESCRIPTION_FIELDS = ( "id", "medikament_name", "dosierung", "anwendungsform", "erhaltungsform", "prescription_status", "prescription_status_source", "prescription_status_provenance", "business_revision", "administration_preset_quantity_value", "administration_preset_dosage_form", "administration_preset_strength", "administration_preset_route_original", "administration_preset_route", "administration_preset_source", "administration_preset_provenance", "administration_preset_revision", ) class MedicationFieldError(ValueError): def __init__(self, field: str, code: str = "invalid") -> None: super().__init__(f"invalid medication field: {field}") self.field = field self.code = code def _canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) def _digest(domain: str, value: Any) -> str: return hashlib.sha256((domain + ":" + _canonical(value)).encode()).hexdigest() def public_identity_key(connection: sqlite3.Connection) -> bytes: row = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() @@ -137,160 +139,187 @@ def event_ref(row: sqlite3.Row | Mapping[str, Any], key: bytes) -> str: "health-medication-event-v1", [_value(row, "id"), _value(row, "datum"), _value(row, "medication_name")], )[:24] def prescription_preset( row: sqlite3.Row | Mapping[str, Any], key: bytes ) -> dict[str, str] | None: fields = ( "administration_preset_quantity_value", "administration_preset_dosage_form", "administration_preset_strength", "administration_preset_source", "administration_preset_provenance", "administration_preset_revision", ) if any(not _value(row, field) for field in fields): return None route = str(_value(row, "administration_preset_route") or "") route_original = str(_value(row, "administration_preset_route_original") or "") if route and (route not in ROUTES or not route_original): return None revision = str(_value(row, "administration_preset_revision")) if not REVISION_RE.fullmatch(revision): return None revision_values = ( str(_value(row, "administration_preset_quantity_value")), str(_value(row, "administration_preset_dosage_form")), str(_value(row, "administration_preset_strength")), route_original, route, str(_value(row, "administration_preset_source")), str(_value(row, "administration_preset_provenance")), ) expected_revision = hashlib.sha256( (str(_value(row, "medikament_name")) + "\0" + "\0".join(revision_values)).encode("utf-8") ).hexdigest() if not hmac.compare_digest(revision, expected_revision): return None return { "quantity_value": str(_value(row, "administration_preset_quantity_value")), "dosage_form": str(_value(row, "administration_preset_dosage_form")), "strength": str(_value(row, "administration_preset_strength")), "route_original": route_original, "route_normalized": route, "preset_revision": _opaque_digest( key, "health-medication-administration-preset-v1", [revision, *revision_values, _value(row, "id"), _value(row, "medikament_name")], ), } def planned_event_preset( row: sqlite3.Row | Mapping[str, Any], key: bytes ) -> dict[str, str] | None: values = tuple( str(_value(row, field) or "") for field in ("planned_quantity_value", "planned_dosage_form", "planned_strength") ) if not all(values): return None route = str(_value(row, "route_normalized") or "") route_original = str(_value(row, "route_original") or "") if route and route not in ROUTES: return None return { "quantity_value": values[0], "dosage_form": values[1], "strength": values[2], "route_original": route_original, "route_normalized": route, "preset_revision": _opaque_digest( key, "health-medication-planned-administration-preset-v1", [event_revision(row), event_ref(row, key)], ), } +def correction_event_preset( + row: sqlite3.Row | Mapping[str, Any], key: bytes +) -> dict[str, str] | None: + values = tuple( + str(_value(row, field) or "") + for field in ("actual_quantity_value", "actual_dosage_form", "actual_strength") + ) + if not all(values): + return None + route = str(_value(row, "route_normalized") or "unknown") + route_original = str(_value(row, "route_original") or "") + if route not in ROUTES: + return None + return { + "quantity_value": values[0], + "dosage_form": values[1], + "strength": values[2], + "route_original": route_original, + "route_normalized": route, + "preset_revision": _opaque_digest( + key, + "health-medication-correction-preset-v1", + [event_revision(row), event_ref(row, key)], + ), + } + + def _available_fields(connection: sqlite3.Connection, table: str, fields: tuple[str, ...]) -> tuple[str, ...]: actual = {str(row[1]) for row in connection.execute(f'PRAGMA table_info("{table}")')} return tuple(field for field in fields if field in actual) def _mapping(row: sqlite3.Row | Mapping[str, Any], fields: tuple[str, ...]) -> dict[str, Any]: keys = set(row.keys()) return {field: row[field] if field in keys else None for field in fields} def prescription_revision(row: sqlite3.Row | Mapping[str, Any]) -> str: return _digest("health-medication-prescription-revision-v1", _mapping(row, PRESCRIPTION_FIELDS)) def event_revision(row: sqlite3.Row | Mapping[str, Any]) -> str: return _digest("health-medication-event-revision-v1", _mapping(row, EVENT_FIELDS)) def list_prescriptions(connection: sqlite3.Connection) -> list[sqlite3.Row]: fields = _available_fields(connection, "medikamente", PRESCRIPTION_FIELDS) return list(connection.execute(f"SELECT {','.join(fields)} FROM medikamente ORDER BY id")) def list_events(connection: sqlite3.Connection) -> list[sqlite3.Row]: fields = _available_fields(connection, "medication_administrations", EVENT_FIELDS) return list(connection.execute(f"SELECT {','.join(fields)} FROM medication_administrations ORDER BY id")) def resolve_prescription(connection: sqlite3.Connection, opaque: str) -> sqlite3.Row: if not isinstance(opaque, str) or not OPAQUE_RE.fullmatch(opaque) or not opaque.startswith("medrx_"): raise RuntimeError("invalid medication reference") key = public_identity_key(connection) matches = [row for row in list_prescriptions(connection) if medication_ref(row, key) == opaque] if len(matches) != 1: raise RuntimeError("medication reference is not unique") return matches[0] def resolve_event(connection: sqlite3.Connection, opaque: str) -> sqlite3.Row: if not isinstance(opaque, str) or not OPAQUE_RE.fullmatch(opaque) or not opaque.startswith("medevt_"): raise RuntimeError("invalid medication event reference") key = public_identity_key(connection) matches = [row for row in list_events(connection) if event_ref(row, key) == opaque] if len(matches) != 1: raise RuntimeError("medication event reference is not unique") return matches[0] def action_context_revision( prescription: sqlite3.Row, planned: sqlite3.Row | None = None, correction_target: sqlite3.Row | None = None, *, planned_consumed: bool = False, ) -> str: return _digest( "health-medication-action-context-v1", { "prescription": prescription_revision(prescription), "planned": event_revision(planned) if planned is not None else None, "correction_target": event_revision(correction_target) if correction_target is not None else None, "planned_consumed": planned_consumed, }, ) def public_action_context_token( key: bytes, prescription: sqlite3.Row, planned: sqlite3.Row | None = None, correction_target: sqlite3.Row | None = None, *, planned_consumed: bool = False, ) -> str: return _opaque_digest( key, "health-medication-public-action-context-v1", action_context_revision( @@ -328,171 +357,188 @@ def action_preview_revision( { "context_revision": action_context_revision( prescription, planned, correction_target, planned_consumed=planned_consumed, ), "payload": bound_payload, }, ) def effective_event_status(connection: sqlite3.Connection, event_id: int) -> str: current_id = event_id seen: set[int] = set() row: sqlite3.Row | tuple[Any, ...] | None = None while current_id not in seen: seen.add(current_id) row = connection.execute( "SELECT id,event_type,corrected_target_status FROM medication_administrations WHERE id=?", (current_id,), ).fetchone() if row is None: return "unknown" child = connection.execute( """SELECT id FROM medication_administrations WHERE corrects_event_id=? AND business_revision IS NOT NULL LIMIT 1""", (current_id,), ).fetchone() if child is None: status = str(row[2] if str(row[1] or "").strip().casefold() == "corrected" else row[1] or "unknown") normalized = status.strip().casefold() if normalized in {"planned", "scheduled", "geplant"}: return "planned" return normalized current_id = int(child[0]) return "unknown" def plan_is_consumed(connection: sqlite3.Connection, planned_id: int) -> bool: if effective_event_status(connection, planned_id) != "planned": return True linked = connection.execute( """SELECT id FROM medication_administrations WHERE planned_event_id=? AND business_revision IS NOT NULL AND lower(trim(COALESCE(event_type,''))) IN ('administered','missed')""", (planned_id,), ) return any(effective_event_status(connection, int(row[0])) in {"administered", "missed"} for row in linked) def resolve_action_preview( connection: sqlite3.Connection, payload: Mapping[str, Any], ) -> tuple[str, sqlite3.Row, sqlite3.Row | None, sqlite3.Row | None]: data = payload["data"] prescription = resolve_prescription(connection, data["medication_ref"]) if str(prescription["medikament_name"]) != data["name"]: raise RuntimeError("medication reference changed") prescription_status = str(prescription["prescription_status"] or "unknown").strip().casefold() trusted_active = bool( prescription_status == "active" and prescription["prescription_status_source"] and prescription["prescription_status_provenance"] and prescription["business_revision"] ) is_v2 = data.get("contract") == ACTION_CONTRACT_VERSION if is_v2: if data["mode"] == "planned" and not trusted_active: raise RuntimeError("planned administration requires an explicitly active prescription") elif data["status"] != "corrected" and not trusted_active: raise RuntimeError("medication prescription is not explicitly active") planned = resolve_event(connection, data["planned_event_ref"]) if data["planned_event_ref"] else None target_ref = data.get("correction_target_ref", "") target = resolve_event(connection, target_ref) if target_ref else None if is_v2: if data["mode"] == "historical" and planned is not None: raise RuntimeError("historical administration cannot consume a plan") if data["mode"] == "planned" and planned is None: raise RuntimeError("planned administration requires a real plan") + if data["mode"] == "correction" and (planned is not None or target is None): + raise RuntimeError("structured correction requires exactly one target") if planned is not None and str(planned["event_type"] or "").strip().casefold() != "planned": raise RuntimeError("planned event reference changed") consumed = plan_is_consumed(connection, int(planned["id"])) if planned is not None else False if consumed: raise RuntimeError("planned medication event is already consumed") if is_v2: key = public_identity_key(connection) preset = prescription_preset(prescription, key) baseline: tuple[str, str, str, str, str] | None = None expected_revision = "" - if data["mode"] == "planned" and planned is not None: + if data["mode"] == "correction" and target is not None: + if effective_event_status(connection, int(target["id"])) != "administered": + raise RuntimeError("structured correction target is not administered") + target_occurred = str(_value(target, "occurred_at") or "") + if not target_occurred or target_occurred != payload["occurred_at"]: + raise RuntimeError("structured correction must preserve occurrence time") + target_preset = correction_event_preset(target, key) + if target_preset is None: + raise RuntimeError("structured correction target lacks complete fields") + baseline = ( + target_preset["quantity_value"], target_preset["dosage_form"], + target_preset["strength"], target_preset["route_original"], + target_preset["route_normalized"], + ) + expected_revision = target_preset["preset_revision"] + elif data["mode"] == "planned" and planned is not None: baseline_values = tuple( str(_value(planned, field) or "") for field in ( "planned_quantity_value", "planned_dosage_form", "planned_strength", "route_original", "route_normalized", ) ) if not all(baseline_values[:3]): raise RuntimeError("planned medication event lacks a structured administration preset") baseline = ( baseline_values[0], baseline_values[1], baseline_values[2], baseline_values[3], baseline_values[4], ) expected_revision = _opaque_digest( key, "health-medication-planned-administration-preset-v1", [event_revision(planned), event_ref(planned, key)], ) elif preset is not None: baseline = ( preset["quantity_value"], preset["dosage_form"], preset["strength"], preset["route_original"], preset["route_normalized"], ) expected_revision = preset["preset_revision"] if baseline is None: if data["preset_revision"]: raise RuntimeError("administration preset is no longer available") else: if not hmac.compare_digest(expected_revision, data["preset_revision"]): raise RuntimeError("administration preset changed") actual = ( data["quantity_value"], data["dosage_form"], data["strength"], data["route_original"], data["route_normalized"], ) if actual != baseline and not data["deviation_confirmed"]: raise RuntimeError("administration differs from verified preset") if target is not None and connection.execute( "SELECT 1 FROM medication_administrations WHERE corrects_event_id=? LIMIT 1", (int(target["id"]),), ).fetchone() is not None: raise RuntimeError("correction target is no longer the latest revision") medication_id = int(prescription["id"]) for related in (planned, target): if related is None: continue related_id = related["medication_id"] if related_id is not None and int(related_id) != medication_id: raise RuntimeError("medication relation changed") if related_id is None and str(related["medication_name"]) != data["name"]: raise RuntimeError("legacy medication relation changed") revision = action_preview_revision( prescription, planned, target, payload, key=public_identity_key(connection), planned_consumed=consumed, ) return revision, prescription, planned, target def _text(value: Any, maximum: int, *, required: bool = False) -> str: if not isinstance(value, str): raise ValueError("invalid medication text") cleaned = " ".join(value.split()) if ( (required and not cleaned) or len(cleaned) > maximum or re.search(r"[\x00-\x1f\x7f]|https?://|\\|\.hermes", cleaned, re.I) or cleaned.startswith("/") or re.search(r"(?:^|/)\.\.(?:/|$)", cleaned) ): raise ValueError("invalid medication text") return cleaned def _field_text(field: str, value: Any, maximum: int, *, required: bool = False) -> str: try: return _text(value, maximum, required=required) except ValueError as error: @@ -533,151 +579,188 @@ def _validate_action_data_v1(data: Any) -> dict[str, Any]: medication = data["medication_ref"] planned = data["planned_event_ref"] target = data["correction_target_ref"] revision = data["preview_revision"] if not isinstance(medication, str) or not OPAQUE_RE.fullmatch(medication) or not medication.startswith("medrx_"): raise ValueError("invalid medication reference") for value in (planned, target): if value and (not isinstance(value, str) or not OPAQUE_RE.fullmatch(value) or not value.startswith("medevt_")): raise ValueError("invalid medication event reference") if not isinstance(revision, str) or not REVISION_RE.fullmatch(revision): raise ValueError("invalid preview revision") route = data["route_normalized"] side = data["injection_side"] if route not in ROUTES or side not in (SIDES | {""}): raise ValueError("invalid medication route or side") injection_region = _text(data["injection_region"], 80) injection_detail = _text(data["injection_detail"], 120) if (injection_region or side or injection_detail) and route not in INJECTION_ROUTES: raise ValueError("injection details require injection route") corrected_status = data["corrected_target_status"] correction_reason = _text(data["correction_reason"], 300) if status == "corrected": if not target or corrected_status not in CORRECTION_STATUSES or not correction_reason: raise ValueError("correction target, status and reason required") elif target or corrected_status or correction_reason: raise ValueError("correction fields on non-correction") for flag in ("plan_value_confirmed", "deviation_confirmed", "duplicate_confirmed"): if type(data[flag]) is not bool: raise ValueError("invalid medication confirmation") planned_value = _text(data["planned_dose_value"], 40) planned_unit = _text(data["planned_dose_unit"], 30) actual_value = _text(data["actual_dose_value"], 40) actual_unit = _text(data["actual_dose_unit"], 30) if status == "administered": if not (data["plan_value_confirmed"] or data["deviation_confirmed"]): raise ValueError("administration requires conscious dose confirmation") if not actual_value or not actual_unit: raise ValueError("administration requires explicitly documented actual dose and unit") elif status == "planned": if actual_value or actual_unit or not planned_value or not planned_unit: raise ValueError("planned event requires planned dose only") elif status == "missed": if actual_value or actual_unit or planned_value or planned_unit: raise ValueError("missed event cannot assert dose values") elif status == "corrected": if corrected_status == "administered": if not actual_value or not actual_unit or planned_value or planned_unit: raise ValueError("administered correction requires actual dose only") elif corrected_status == "planned": if actual_value or actual_unit or not planned_value or not planned_unit: raise ValueError("planned correction requires planned dose only") elif actual_value or actual_unit or planned_value or planned_unit: raise ValueError("missed or unknown correction cannot assert dose values") return { "contract": ACTION_CONTRACT_VERSION_V1, "status": status, "medication_ref": medication, "planned_event_ref": planned, "name": _text(data["name"], 120, required=True), "planned_dose_value": planned_value, "planned_dose_unit": planned_unit, "actual_dose_value": actual_value, "actual_dose_unit": actual_unit, "route_original": _text(data["route_original"], 60), "route_normalized": route, "injection_region": injection_region, "injection_side": side, "injection_detail": injection_detail, "lot_number": _text(data["lot_number"], 80), "correction_target_ref": target, "corrected_target_status": corrected_status, "correction_reason": correction_reason, "note": _text(data["note"], 300), "preview_revision": revision, "plan_value_confirmed": data["plan_value_confirmed"], "deviation_confirmed": data["deviation_confirmed"], "duplicate_confirmed": data["duplicate_confirmed"], } +def _quantity(value: Any) -> str: + text = _field_text("quantity_value", value, 40, required=True) + if not re.fullmatch(r"(?:0|[1-9][0-9]{0,5})(?:[.,][0-9]{1,4})?", text): + raise MedicationFieldError("quantity_value", "not_numeric") + try: + number = Decimal(text.replace(",", ".")) + except InvalidOperation as error: + raise MedicationFieldError("quantity_value", "not_numeric") from error + if number <= 0: + raise MedicationFieldError("quantity_value", "not_positive") + normalized = format(number.normalize(), "f") + return normalized.rstrip("0").rstrip(".") if "." in normalized else normalized + + def _validate_action_data_v2(data: Any) -> dict[str, Any]: required = { "contract", "mode", "status", "medication_ref", "planned_event_ref", "name", "quantity_value", "dosage_form", "strength", "route_original", "route_normalized", "injection_region", "injection_side", "injection_detail", "note", "preset_revision", "preview_revision", "deviation_confirmed", "duplicate_confirmed", } - if not isinstance(data, dict) or set(data) != required or data.get("contract") != ACTION_CONTRACT_VERSION: + correction_fields = {"correction_target_ref", "correction_reason", "bind_verified_preset"} + if not isinstance(data, dict) or data.get("contract") != ACTION_CONTRACT_VERSION: raise ValueError("invalid medication action shape") - if data["mode"] not in {"historical", "planned"}: + mode = data.get("mode") + if set(data) != (required | correction_fields if mode == "correction" else required): + raise ValueError("invalid medication action shape") + if mode not in {"historical", "planned", "correction"}: raise MedicationFieldError("capture_mode") - if data["status"] != "administered": + if data["status"] != ("corrected" if mode == "correction" else "administered"): raise MedicationFieldError("capture_mode") medication = data["medication_ref"] planned = data["planned_event_ref"] if not isinstance(medication, str) or not OPAQUE_RE.fullmatch(medication) or not medication.startswith("medrx_"): raise MedicationFieldError("medication") if planned and (not isinstance(planned, str) or not OPAQUE_RE.fullmatch(planned) or not planned.startswith("medevt_")): raise MedicationFieldError("capture_mode") - if data["mode"] == "historical" and planned: + if mode == "historical" and planned: raise ValueError("historical administration cannot reference a plan") - if data["mode"] == "planned" and not planned: + if mode == "planned" and not planned: raise ValueError("planned administration requires a real plan") + target = data.get("correction_target_ref", "") + if mode == "correction": + if planned or not isinstance(target, str) or not OPAQUE_RE.fullmatch(target) or not target.startswith("medevt_"): + raise MedicationFieldError("correction_target") + if type(data["bind_verified_preset"]) is not bool: + raise MedicationFieldError("preset") for field in ("preview_revision", "preset_revision"): value = data[field] if value and (not isinstance(value, str) or not REVISION_RE.fullmatch(value)): raise MedicationFieldError("preview" if field == "preview_revision" else "preset") route = data["route_normalized"] side = data["injection_side"] if route not in ROUTES: raise MedicationFieldError("route_normalized") if side not in (SIDES | {""}): raise MedicationFieldError("injection_side") - quantity = _field_text("quantity_value", data["quantity_value"], 40, required=True) + quantity = _quantity(data["quantity_value"]) dosage_form = _field_text("dosage_form", data["dosage_form"], 40, required=True) + if dosage_form not in DOSAGE_FORMS: + raise MedicationFieldError("dosage_form", "not_controlled") strength = _field_text("strength", data["strength"], 80, required=True) injection_region = _field_text("injection_region", data["injection_region"], 80) injection_detail = _field_text("injection_detail", data["injection_detail"], 120) if (injection_region or side or injection_detail) and route not in INJECTION_ROUTES: raise MedicationFieldError("route_normalized") for flag in ("deviation_confirmed", "duplicate_confirmed"): if type(data[flag]) is not bool: raise MedicationFieldError(flag) - return { + result = { "contract": ACTION_CONTRACT_VERSION, - "mode": data["mode"], - "status": "administered", + "mode": mode, + "status": "corrected" if mode == "correction" else "administered", "medication_ref": medication, "planned_event_ref": planned, "name": _field_text("medication", data["name"], 120, required=True), "quantity_value": quantity, "dosage_form": dosage_form, "strength": strength, "route_original": _field_text("route_original", data["route_original"], 60), "route_normalized": route, "injection_region": injection_region, "injection_side": side, "injection_detail": injection_detail, "note": _field_text("note", data["note"], 300), "preset_revision": data["preset_revision"], "preview_revision": data["preview_revision"], "deviation_confirmed": data["deviation_confirmed"], "duplicate_confirmed": data["duplicate_confirmed"], } + if mode == "correction": + result.update( + { + "correction_target_ref": target, + "correction_reason": _field_text( + "correction_reason", data["correction_reason"], 300, required=True + ), + "bind_verified_preset": data["bind_verified_preset"], + } + ) + return result def validate_action_data(data: Any) -> dict[str, Any]: if isinstance(data, dict) and data.get("contract") == ACTION_CONTRACT_VERSION_V1: return _validate_action_data_v1(data) return _validate_action_data_v2(data) diff --git a/scripts/health/dashboard_v5/medication_schema.py b/scripts/health/dashboard_v5/medication_schema.py index 2e8dd38..6fd26ff 100644 --- a/scripts/health/dashboard_v5/medication_schema.py +++ b/scripts/health/dashboard_v5/medication_schema.py @@ -100,160 +100,207 @@ MANAGED_TRIGGER_SHA256 = { def _normalized_sql(value: object) -> str: return re.sub(r"\s+", " ", str(value or "")).strip().casefold() def _columns(connection: sqlite3.Connection, table: str) -> set[str]: return {str(row[1]) for row in connection.execute(f'PRAGMA table_info("{table}")')} def _require_base(connection: sqlite3.Connection) -> None: tables = { str(row[0]) for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'") } missing = {"medikamente", "medication_administrations"} - tables if missing: raise RuntimeError("base medication schema missing") def _add_columns( connection: sqlite3.Connection, table: str, expected: tuple[tuple[str, str], ...] ) -> list[str]: actual = _columns(connection, table) changes: list[str] = [] for name, kind in expected: if name in actual: continue connection.execute(f'ALTER TABLE "{table}" ADD COLUMN "{name}" {kind}') changes.append(f"{table}.{name}") return changes def _install_verified_presets(connection: sqlite3.Connection) -> list[str]: """Install only the exact preset explicitly verified for this hotfix.""" rows = connection.execute( """SELECT id,anwendungsform, administration_preset_quantity_value, administration_preset_dosage_form, administration_preset_strength, administration_preset_route_original, administration_preset_route, administration_preset_source, administration_preset_provenance, administration_preset_revision FROM medikamente WHERE medikament_name=? ORDER BY id""", (HYRIMOZ_PRESET_NAME,), ).fetchall() if not rows: return [] if len(rows) != 1: raise RuntimeError("verified Hyrimoz preset identity is ambiguous") row = rows[0] if str(row[1] or "").strip().casefold() != "subkutan": raise RuntimeError("verified Hyrimoz route source changed") revision = hashlib.sha256( (HYRIMOZ_PRESET_NAME + "\0" + "\0".join(HYRIMOZ_PRESET)).encode("utf-8") ).hexdigest() expected = (*HYRIMOZ_PRESET, revision) current = tuple(row[index] for index in range(2, 10)) if all(value is None for value in current): connection.execute( """UPDATE medikamente SET administration_preset_quantity_value=?, administration_preset_dosage_form=?, administration_preset_strength=?, administration_preset_route_original=?, administration_preset_route=?, administration_preset_source=?, administration_preset_provenance=?, administration_preset_revision=? WHERE id=?""", (*expected, int(row[0])), ) return ["medikamente.verified_hyrimoz_administration_preset"] if current != expected: raise RuntimeError("verified Hyrimoz preset conflicts with existing data") return [] +def bind_verified_hyrimoz_preset( + connection: sqlite3.Connection, medication_id: int, expected_business_revision: str +) -> None: + """Bind the approved preset to one explicitly resolved prescription row only.""" + row = connection.execute( + """SELECT id,medikament_name,business_revision, + administration_preset_quantity_value, + administration_preset_dosage_form, + administration_preset_strength, + administration_preset_route_original, + administration_preset_route, + administration_preset_source, + administration_preset_provenance, + administration_preset_revision + FROM medikamente WHERE id=?""", + (medication_id,), + ).fetchone() + if ( + row is None + or str(row[1]) != HYRIMOZ_PRESET_NAME + or (expected_business_revision and str(row[2] or "") != expected_business_revision) + ): + raise RuntimeError("verified preset prescription binding changed") + revision = hashlib.sha256( + (HYRIMOZ_PRESET_NAME + "\0" + "\0".join(HYRIMOZ_PRESET)).encode("utf-8") + ).hexdigest() + expected = (*HYRIMOZ_PRESET, revision) + current = tuple(row[index] for index in range(3, 11)) + if all(value is None for value in current): + binding_clause = "business_revision=?" if expected_business_revision else "business_revision IS NULL" + parameters = (*expected, medication_id, expected_business_revision) if expected_business_revision else (*expected, medication_id) + connection.execute( + f"""UPDATE medikamente SET + administration_preset_quantity_value=?,administration_preset_dosage_form=?, + administration_preset_strength=?,administration_preset_route_original=?, + administration_preset_route=?,administration_preset_source=?, + administration_preset_provenance=?,administration_preset_revision=? + WHERE id=? AND {binding_clause}""", + parameters, + ) + if connection.execute("SELECT changes()").fetchone()[0] != 1: + raise RuntimeError("verified preset prescription binding changed") + return + if current != expected: + raise RuntimeError("verified preset conflicts with selected prescription") + + def _has_legacy_event_uniqueness(connection: sqlite3.Connection) -> bool: for row in connection.execute("PRAGMA index_list('medication_administrations')"): if not int(row[2]): continue columns = [ str(column[2]) for column in connection.execute(f'PRAGMA index_info("{row[1]}")') ] if columns == ["datum", "medication_name", "event_type"]: return True return False def _rebuild_event_table_without_legacy_uniqueness( connection: sqlite3.Connection, ) -> bool: """Relax the obsolete tuple uniqueness in the disposable candidate DB.""" if not _has_legacy_event_uniqueness(connection): return False existing = _columns(connection, "medication_administrations") preserved_objects: dict[str, str] = {} for object_type, object_name, object_sql in connection.execute( """SELECT type,name,sql FROM sqlite_master WHERE tbl_name='medication_administrations' AND type IN ('index','trigger') AND sql IS NOT NULL""" ): object_name = str(object_name) if object_name.startswith("trg_medication_") or object_name in MANAGED_INDEX_NAMES: continue if object_type == "index": columns = [ str(column[2]) for column in connection.execute(f'PRAGMA index_info("{object_name}")') ] if columns == ["datum", "medication_name", "event_type"]: continue preserved_objects[object_name] = str(object_sql) connection.execute("PRAGMA defer_foreign_keys=ON") connection.execute("DROP TABLE IF EXISTS medication_administrations_7cf_new") connection.execute( """CREATE TABLE medication_administrations_7cf_new ( id INTEGER PRIMARY KEY AUTOINCREMENT, datum TEXT NOT NULL, medication_name TEXT NOT NULL, dose TEXT, route TEXT, event_type TEXT, scheduled_next_date TEXT, notes TEXT, source TEXT DEFAULT 'manual', created_at TEXT DEFAULT CURRENT_TIMESTAMP, occurred_at TEXT, medication_id INTEGER REFERENCES medikamente(id), planned_event_id INTEGER REFERENCES medication_administrations_7cf_new(id), planned_dose_value TEXT, planned_dose_unit TEXT, actual_dose_value TEXT, actual_dose_unit TEXT, route_original TEXT, route_normalized TEXT, injection_region TEXT, injection_side TEXT, injection_detail TEXT, lot_number TEXT, corrects_event_id INTEGER REFERENCES medication_administrations_7cf_new(id), corrected_target_status TEXT, correction_reason TEXT, business_revision TEXT )""" ) target_columns = [ "id", "datum", "medication_name", "dose", "route", "event_type", "scheduled_next_date", "notes", "source", "created_at", "occurred_at", *(name for name, _ in EVENT_COLUMNS), ] copied = [name for name in target_columns if name in existing] quoted = ",".join(f'"{name}"' for name in copied) connection.execute( f"INSERT INTO medication_administrations_7cf_new({quoted}) " f"SELECT {quoted} FROM medication_administrations ORDER BY id" diff --git a/scripts/health/health_dashboard_action_worker.py b/scripts/health/health_dashboard_action_worker.py index a77eb66..2626066 100644 --- a/scripts/health/health_dashboard_action_worker.py +++ b/scripts/health/health_dashboard_action_worker.py @@ -1,173 +1,179 @@ #!/usr/bin/env python3 """Consume locally queued Dashboard actions outside the network-facing service.""" from __future__ import annotations import json import os import re import stat import subprocess import sys import sqlite3 import hashlib import hmac import secrets import shutil import tempfile from datetime import date, datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo try: from dashboard_v5.patient_action_schema import REQUIRED_PATIENT_ACTION_COLUMNS from dashboard_v5.sprint6f_b_schema import assert_supplement_schema from dashboard_v5.supplement_contract import validate_supplement_payload from dashboard_v5.observation_contract import ( TRANSITIONS, validate_action as validate_observation_action, ) from dashboard_v5.sprint6f_c_schema import ( METHOD_VERSION, assert_schema as assert_observation_schema, ) from dashboard_v5.capture_contract import validate_capture_payload - from dashboard_v5.medication_schema import assert_schema as assert_medication_schema + from dashboard_v5.medication_schema import ( + assert_schema as assert_medication_schema, + bind_verified_hyrimoz_preset, + ) from dashboard_v5.medication_contract import ( ACTION_CONTRACT_VERSIONS as MEDICATION_ACTION_CONTRACTS, resolve_action_preview, ) from dashboard_v5.capture_media import cleanup_expired, promote_attachment from dashboard_v5.media_validation import MediaInfo, create_safe_derivatives, media_runtime_self_test from dashboard_v5.sprint6i_b_schema import ( apply_schema as apply_media_schema, assert_schema as assert_media_schema, ) from dashboard_v5.sprint6h_b_schema import ( apply_schema as apply_capture_schema, assert_schema as assert_capture_schema, ) from dashboard_v5.document_review import ( CANDIDATE_ID_RE, ID_RE, TOKEN_RE, candidate_rows, extract_pages, load_quarantine, open_quarantine_source, normalize_search_text, section_similarity, validate_metadata, ) from dashboard_v5.document_originals import probe_original from dashboard_v5.sprint6i_a_schema import ( apply_schema as apply_document_schema, assert_schema as assert_document_schema, ) from dashboard_v5.sprint6i_c_schema import assert_schema as assert_reconciliation_schema from dashboard_v5.document_reconciliation import canonical_lab_pair, normalize_parameter, normalize_unit, normalize_value, reconcile_documents, transfer_preview from dashboard_v5.lab_review import QUALITATIVE, candidate_catalog_match, explicit_candidate_date, preview_revision_digest from dashboard_v5.metric_catalog_v2 import LAB_CATALOG_SPECS from dashboard_v5.nutrition_mapping_review import ( aggressive_identity as _normalize_food_identity, assignment_evidence_token, exact_identity as _exact_food_identity, resolve_catalog_assignment, resolve_group as resolve_mapping_group, ) except ModuleNotFoundError: # direct importlib fixture execution sys.path.insert(0, str(Path(__file__).resolve().parent)) from dashboard_v5.patient_action_schema import REQUIRED_PATIENT_ACTION_COLUMNS from dashboard_v5.sprint6f_b_schema import assert_supplement_schema from dashboard_v5.supplement_contract import validate_supplement_payload from dashboard_v5.observation_contract import ( TRANSITIONS, validate_action as validate_observation_action, ) from dashboard_v5.sprint6f_c_schema import ( METHOD_VERSION, assert_schema as assert_observation_schema, ) from dashboard_v5.capture_contract import validate_capture_payload - from dashboard_v5.medication_schema import assert_schema as assert_medication_schema + from dashboard_v5.medication_schema import ( + assert_schema as assert_medication_schema, + bind_verified_hyrimoz_preset, + ) from dashboard_v5.medication_contract import ( ACTION_CONTRACT_VERSIONS as MEDICATION_ACTION_CONTRACTS, resolve_action_preview, ) from dashboard_v5.capture_media import cleanup_expired, promote_attachment from dashboard_v5.media_validation import MediaInfo, create_safe_derivatives, media_runtime_self_test from dashboard_v5.sprint6i_b_schema import ( apply_schema as apply_media_schema, assert_schema as assert_media_schema, ) from dashboard_v5.sprint6h_b_schema import ( apply_schema as apply_capture_schema, assert_schema as assert_capture_schema, ) from dashboard_v5.document_review import ( CANDIDATE_ID_RE, ID_RE, TOKEN_RE, candidate_rows, extract_pages, load_quarantine, open_quarantine_source, normalize_search_text, section_similarity, validate_metadata, ) from dashboard_v5.document_originals import probe_original from dashboard_v5.sprint6i_a_schema import ( apply_schema as apply_document_schema, assert_schema as assert_document_schema, ) from dashboard_v5.sprint6i_c_schema import assert_schema as assert_reconciliation_schema from dashboard_v5.document_reconciliation import canonical_lab_pair, normalize_parameter, normalize_unit, normalize_value, reconcile_documents, transfer_preview from dashboard_v5.lab_review import QUALITATIVE, candidate_catalog_match, explicit_candidate_date, preview_revision_digest from dashboard_v5.metric_catalog_v2 import LAB_CATALOG_SPECS from dashboard_v5.nutrition_mapping_review import ( aggressive_identity as _normalize_food_identity, assignment_evidence_token, exact_identity as _exact_food_identity, resolve_catalog_assignment, resolve_group as resolve_mapping_group, ) BASE = Path.home() / ".hermes" / "assets" / "Gesundheit" DEFAULT_DASHBOARD_DB = (BASE / "health_data.db").resolve() DASHBOARD_DB = ( Path(os.environ["HEALTH_DASHBOARD_DB"]).resolve() if os.environ.get("HEALTH_DASHBOARD_DB") else None ) ACTION_INBOX = Path( os.environ.get( "HEALTH_DASHBOARD_ACTION_INBOX", str(BASE / "runtime" / "dashboard-actions"), ) ).resolve() CAPTURE_MEDIA = Path( os.environ.get( "HEALTH_DASHBOARD_CAPTURE_MEDIA", str(BASE / "private-media" / "capture") ) ).resolve() CAPTURE_QUARANTINE = Path( os.environ.get( "HEALTH_DASHBOARD_CAPTURE_QUARANTINE", str(BASE / "runtime" / "capture-quarantine"), ) ).resolve() DOCUMENT_QUARANTINE = Path( os.environ.get( "HEALTH_DASHBOARD_DOCUMENT_QUARANTINE", str(BASE / "runtime" / "document-quarantine"), ) ).resolve() DOCUMENT_STORAGE = Path( os.environ.get( "HEALTH_DASHBOARD_DOCUMENT_STORAGE", str(BASE / "private-media" / "documents"), ) ).resolve() @@ -973,203 +979,213 @@ def apply_supplement_action(payload: dict[str, Any]) -> None: payload["amount"], payload["unit"], payload["status"], payload["occurred_at"], payload["composition_source"], payload["assignment_reliability"], payload["notes"] or None, ), ) connection.commit() except Exception: connection.rollback() raise finally: connection.close() def apply_capture_action(payload: dict[str, Any]) -> str: if DASHBOARD_DB is None: raise RuntimeError("HEALTH_DASHBOARD_DB is required") payload = validate_capture_payload(payload) action_hash = hashlib.sha256(_canonical_json(payload).encode()).hexdigest() connection = sqlite3.connect(DASHBOARD_DB) connection.row_factory = sqlite3.Row promoted: list[dict[str, Any]] = [] try: connection.execute("PRAGMA foreign_keys=ON") apply_capture_schema(connection) assert_capture_schema(connection) apply_media_schema(connection) assert_media_schema(connection) connection.commit() connection.execute("BEGIN IMMEDIATE") existing = connection.execute( "SELECT action_hash,entry_id,request_version FROM capture_action_log WHERE idempotency_key=?", (payload["idempotency_key"],), ).fetchone() if existing: if ( existing["action_hash"] != action_hash or int(existing["request_version"]) != payload["request_version"] ): raise RuntimeError("conflicting idempotency key") connection.rollback() return str(existing["entry_id"]) now = datetime.now(LOCAL_TIMEZONE).isoformat(timespec="seconds") target_id = payload["corrects_entry_id"] or payload["withdraws_entry_id"] if target_id: target = connection.execute( "SELECT id,root_id,version,status,capture_type FROM capture_entries WHERE id=?", (target_id,), ).fetchone() if ( target is None or target["status"] != "active" or target["capture_type"] != payload["capture_type"] ): raise RuntimeError("version target is not active") root_id, version = str(target["root_id"]), int(target["version"]) + 1 connection.execute( "UPDATE capture_entries SET status=? WHERE id=?", ( "corrected" if payload["corrects_entry_id"] else "withdrawn", target_id, ), ) else: root_id, version = "", 1 entry_id = _opaque("cap_") if not root_id: root_id = entry_id data = payload["data"] medication_context: dict[str, Any] | None = None known_tables = { str(row[0]) for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'") } if payload["capture_type"] == "medication" and data.get("contract") in MEDICATION_ACTION_CONTRACTS: assert_medication_schema(connection) expected_revision, prescription, planned, correction_target = resolve_action_preview( connection, payload ) if not hmac.compare_digest(expected_revision, data["preview_revision"]): raise RuntimeError("stale medication preview revision") + if data["contract"] == "health.medication_action.v2" and data["mode"] == "correction": + if correction_target is None: + raise RuntimeError("structured correction target missing") + linked_capture = connection.execute( + "SELECT entry_id FROM capture_action_log WHERE action_hash=?", + (str(correction_target["business_revision"] or ""),), + ).fetchone() + if linked_capture is None or str(linked_capture[0]) != payload["corrects_entry_id"]: + raise RuntimeError("structured correction capture target changed") medication_id = int(prescription["id"]) if data["contract"] == "health.medication_action.v2": duplicate = connection.execute( """SELECT 1 FROM medication_administrations WHERE business_revision IS NOT NULL AND medication_id=? AND COALESCE(planned_event_id,-1)=COALESCE(?,-1) AND COALESCE(occurred_at,'')=? AND lower(trim(COALESCE(event_type,'')))=? AND COALESCE(actual_quantity_value,'')=? AND COALESCE(actual_dosage_form,'')=? AND COALESCE(actual_strength,'')=? LIMIT 1""", ( medication_id, int(planned["id"]) if planned is not None else None, payload["occurred_at"], data["status"], data["quantity_value"], data["dosage_form"], data["strength"], ), ).fetchone() else: duplicate = connection.execute( """SELECT 1 FROM medication_administrations WHERE business_revision IS NOT NULL AND medication_id=? AND COALESCE(planned_event_id,-1)=COALESCE(?,-1) AND COALESCE(occurred_at,'')=? AND lower(trim(COALESCE(event_type,'')))=? AND COALESCE(actual_dose_value,'')=? AND COALESCE(actual_dose_unit,'')=? LIMIT 1""", ( medication_id, int(planned["id"]) if planned is not None else None, payload["occurred_at"], data["status"], data["actual_dose_value"], data["actual_dose_unit"], ), ).fetchone() if duplicate and not data["duplicate_confirmed"]: raise RuntimeError("possible duplicate medication event") medication_context = { "medication_id": medication_id, "planned_event_id": int(planned["id"]) if planned is not None else None, "planned_quantity_value": str(planned["planned_quantity_value"] or "") if planned is not None else "", "planned_dosage_form": str(planned["planned_dosage_form"] or "") if planned is not None else "", "planned_strength": str(planned["planned_strength"] or "") if planned is not None else "", "corrects_event_id": int(correction_target["id"]) if correction_target is not None else None, + "prescription_business_revision": str(prescription["business_revision"] or ""), } elif payload["capture_type"] == "medication" and "medication_administrations" in known_tables: known_medications = { str(row[0]) for row in connection.execute( "SELECT DISTINCT medication_name FROM medication_administrations WHERE trim(COALESCE(medication_name,''))<>''" ) } if data["name"] not in known_medications: raise RuntimeError("medication is not in the known exact catalog") if payload["capture_type"] == "supplement" and "supplement_plans" in known_tables: known_supplements = { str(row[0]) for row in connection.execute( "SELECT DISTINCT product FROM supplement_plans WHERE trim(COALESCE(product,''))<>''" ) } if data["name"] not in known_supplements: raise RuntimeError("supplement is not in the known exact plan") if ( payload["capture_type"] in {"medication", "supplement"} and data["status"] == "administered" and data.get("contract") != "health.medication_action.v2" ): if not (data["plan_value_confirmed"] or data["deviation_confirmed"]): raise RuntimeError("planned administration requires conscious confirmation") if payload["capture_type"] == "measurement": soft = { "systolic": (70, 220), "diastolic": (40, 140), "pulse": (35, 200), "weight": (30, 300), "temperature": (34, 42), "oxygen_saturation": (70, 100), "blood_glucose": (2, 30), } unusual = any( not soft[key][0] <= float(value) <= soft[key][1] for key, value in data["values"].items() ) if unusual and not data["plausibility_confirmed"]: raise RuntimeError("plausible outlier requires confirmation") duplicate = connection.execute( "SELECT 1 FROM capture_entries WHERE capture_type='measurement' AND occurred_at=? AND payload_json=? AND status='active'", (payload["occurred_at"], _canonical_json(data)), ).fetchone() if duplicate and not data["duplicate_confirmed"]: raise RuntimeError("duplicate measurement requires confirmation") for token in payload["attachments"]: attachment = promote_attachment(CAPTURE_QUARANTINE, CAPTURE_MEDIA, token) duplicate_media = connection.execute( """SELECT media_name,thumbnail_name,proxy_name,mime_type,media_kind,byte_size,width,height, frame_count,auxiliary_count,duration_seconds,container,codec,preview_sha256,proxy_sha256, preview_status,decoder_name,decoder_version,probe_version,unconfirmed_captured_at FROM capture_attachments WHERE sha256=? ORDER BY created_at LIMIT 1""", (attachment["sha256"],), ).fetchone() if duplicate_media: for duplicate_name in (attachment["media_name"], attachment["thumbnail_name"], attachment.get("proxy_name")): if duplicate_name: (CAPTURE_MEDIA / duplicate_name).unlink(missing_ok=True) attachment.update(dict(duplicate_media)) attachment["reused"] = True else: attachment["reused"] = False promoted.append(attachment) if sum(item["media_kind"] == "photo" for item in promoted) > 4 or sum(item["media_kind"] == "video" for item in promoted) > 1: raise RuntimeError("capture media count exceeded") entry_status = "withdrawn" if payload["withdraws_entry_id"] else "active" connection.execute( "INSERT INTO capture_entries(id,root_id,version,capture_type,occurred_at,ended_at,payload_json,status,corrects_entry_id,withdraws_entry_id,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)", ( entry_id, root_id, version, payload["capture_type"], payload["occurred_at"], payload["ended_at"], _canonical_json(data), entry_status, @@ -1178,176 +1194,194 @@ def apply_capture_action(payload: dict[str, Any]) -> str: now, ), ) for attachment in promoted: connection.execute( """INSERT INTO capture_attachments( id,entry_id,sha256,mime_type,media_kind,byte_size,width,height,frame_count,auxiliary_count, duration_seconds,container,codec,media_name,thumbnail_name,proxy_name,preview_sha256,proxy_sha256, preview_status,decoder_name,decoder_version,probe_version,unconfirmed_captured_at,review_status, description,body_region,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( attachment["id"], entry_id, attachment["sha256"], attachment["mime_type"], attachment["media_kind"], attachment["byte_size"], attachment["width"], attachment["height"], attachment["frame_count"], attachment["auxiliary_count"], attachment.get("duration_seconds"), attachment.get("container"), attachment.get("codec"), attachment["media_name"], attachment["thumbnail_name"], attachment.get("proxy_name"), attachment.get("preview_sha256"), attachment.get("proxy_sha256"), attachment["preview_status"], attachment["decoder_name"], attachment["decoder_version"], attachment.get("probe_version"), attachment.get("unconfirmed_captured_at"), "unverified", data.get("description") or None, data.get("body_region") or None, now, ), ) if payload["corrects_entry_id"] and not promoted: columns = "sha256,mime_type,media_kind,byte_size,width,height,frame_count,auxiliary_count,duration_seconds,container,codec,media_name,thumbnail_name,proxy_name,preview_sha256,proxy_sha256,preview_status,decoder_name,decoder_version,probe_version,unconfirmed_captured_at,review_status,explicit_pair_id,description,body_region" for old in connection.execute(f"SELECT {columns} FROM capture_attachments WHERE entry_id=?", (payload["corrects_entry_id"],)).fetchall(): prefix = "vid_" if old[2] == "video" else "img_" connection.execute( f"INSERT INTO capture_attachments(id,entry_id,{columns},created_at) VALUES(?,?{',?' * len(old)},?)", (_opaque(prefix), entry_id, *old, now), ) day = payload["occurred_at"].split("T", 1)[0] available_tables = { str(row[0]) for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'") } if not payload["withdraws_entry_id"]: if payload["capture_type"] == "symptom" and "symptom_log" in available_tables: severity = None if data["intensity"] == "unknown" else data["intensity"] label = data["title"] or data["category"] context = _canonical_json( { "source": "mobile_capture", "category": data["category"], "count": data["count"], "body_region": data["body_region"], "ongoing": data["ongoing"], "entry_id": entry_id, } ) symptom_columns = { str(row[1]) for row in connection.execute("PRAGMA table_info(symptom_log)") } if {"occurred_at", "onset_at", "duration_minutes"}.issubset(symptom_columns): connection.execute( "INSERT INTO symptom_log(datum,symptom,schwergrad,kontext,notizen,occurred_at,onset_at,duration_minutes) VALUES(?,?,?,?,?,?,?,NULL)", ( day, label, severity, context, data["note"] or None, payload["occurred_at"], payload["occurred_at"], ), ) else: connection.execute( "INSERT INTO symptom_log(datum,symptom,schwergrad,kontext,notizen) VALUES(?,?,?,?,?)", (day, label, severity, context, data["note"] or None), ) elif payload["capture_type"] == "medication" and "medication_administrations" in available_tables: if medication_context is not None: if data["contract"] == "health.medication_action.v2": legacy_dose = f"{data['quantity_value']} {data['dosage_form']} · {data['strength']}" legacy_route = data["route_original"] or data["route_normalized"] connection.execute( """INSERT INTO medication_administrations( datum,medication_name,dose,route,event_type,scheduled_next_date,notes,source,occurred_at, medication_id,planned_event_id,planned_quantity_value,planned_dosage_form, planned_strength,route_original,route_normalized,injection_region, injection_side,injection_detail,business_revision,actual_quantity_value, - actual_dosage_form,actual_strength) - VALUES(?,?,?,?,?,NULL,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + actual_dosage_form,actual_strength,corrects_event_id,corrected_target_status, + correction_reason) + VALUES(?,?,?,?,?,NULL,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( day, data["name"], legacy_dose, legacy_route or None, data["status"], data["note"] or None, "dashboard_medication_action_version_two", payload["occurred_at"], medication_context["medication_id"], medication_context["planned_event_id"], medication_context["planned_quantity_value"] or None, medication_context["planned_dosage_form"] or None, medication_context["planned_strength"] or None, data["route_original"] or None, data["route_normalized"] or None, data["injection_region"] or None, data["injection_side"] or None, data["injection_detail"] or None, action_hash, data["quantity_value"], data["dosage_form"], data["strength"], + medication_context["corrects_event_id"], + "administered" if data["mode"] == "correction" else None, + data.get("correction_reason") or None, ), ) + if data.get("bind_verified_preset"): + if ( + data["mode"] != "correction" + or ( + data["quantity_value"], data["dosage_form"], + data["strength"], data["route_normalized"], + ) != ("1", "Spritze", "40 mg/0,4 ml", "subcutaneous") + ): + raise RuntimeError("verified preset confirmation changed") + bind_verified_hyrimoz_preset( + connection, + medication_context["medication_id"], + str(medication_context["prescription_business_revision"]), + ) else: actual_dose = " ".join( part for part in (data["actual_dose_value"], data["actual_dose_unit"]) if part ) legacy_route = data["route_original"] or data["route_normalized"] connection.execute( """INSERT INTO medication_administrations( datum,medication_name,dose,route,event_type,scheduled_next_date,notes,source,occurred_at, medication_id,planned_event_id,planned_dose_value,planned_dose_unit,actual_dose_value, actual_dose_unit,route_original,route_normalized,injection_region,injection_side, injection_detail,lot_number,corrects_event_id,corrected_target_status,correction_reason, business_revision) VALUES(?,?,?,?,?,NULL,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( day, data["name"], actual_dose or None, legacy_route or None, data["status"], data["note"] or None, "dashboard_v5_medication_action", payload["occurred_at"], medication_context["medication_id"], medication_context["planned_event_id"], data["planned_dose_value"] or None, data["planned_dose_unit"] or None, data["actual_dose_value"] or None, data["actual_dose_unit"] or None, data["route_original"] or None, data["route_normalized"] or None, data["injection_region"] or None, data["injection_side"] or None, data["injection_detail"] or None, data["lot_number"] or None, medication_context["corrects_event_id"], data["corrected_target_status"] or None, data["correction_reason"] or None, action_hash, ), ) else: dose = " ".join( str(part) for part in (data["amount"], data["unit"]) if part not in {None, ""} ) medication_columns = {str(row[1]) for row in connection.execute("PRAGMA table_info(medication_administrations)")} if "occurred_at" in medication_columns: connection.execute( "INSERT INTO medication_administrations(datum,medication_name,dose,route,event_type,notes,source,occurred_at) VALUES(?,?,?,?,?,?,?,?)", (day, data["name"], dose or None, data["route"] or None, data["status"], data["note"] or None, "dashboard_v5_mobile_capture", payload["occurred_at"]), ) else: connection.execute( "INSERT INTO medication_administrations(datum,medication_name,dose,route,event_type,notes,source) VALUES(?,?,?,?,?,?,?)", (day, data["name"], dose or None, data["route"] or None, data["status"], data["note"] or None, "dashboard_v5_mobile_capture"), ) elif payload["capture_type"] == "supplement" and "supplement_intakes" in available_tables: connection.execute( "INSERT INTO supplement_intakes(plan_id,product,brand_variant,nutrient_key,amount,unit,status,occurred_at,composition_source,assignment_reliability,notes) VALUES(NULL,?,NULL,'other',?,?,?,?, 'user_documented','documented',?)", ( data["name"], data["amount"], data["unit"], data["status"], payload["occurred_at"], data["note"] or None, ), ) elif payload["capture_type"] == "event" and "health_events" in available_tables: event_columns = { str(row[1]) for row in connection.execute("PRAGMA table_info(health_events)") } event_kind = data.get("event_kind") if event_kind == "sauna": category = "sauna_recovery" parameter = "Sauna" value = data["duration_minutes"] unit = "min" source = "dashboard_v5_manual_capture" intensity = None elif event_kind == "training": category = "manual_training" parameter = data["activity_type"] value = data["duration_minutes"] unit = "min" source = "dashboard_v5_manual_capture" intensity = None else: category = data["category"] parameter = data["title"] intensity = ( None if data["intensity"] == "unknown" else data["intensity"] diff --git a/scripts/health/health_dashboard_server.py b/scripts/health/health_dashboard_server.py index e849486..fc5122e 100644 --- a/scripts/health/health_dashboard_server.py +++ b/scripts/health/health_dashboard_server.py @@ -1142,192 +1142,198 @@ def write_action_payload(payload: dict[str, object]) -> str: descriptor = os.open( temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, ) try: with os.fdopen(descriptor, "wb") as handle: handle.write(packed) handle.flush() os.fsync(handle.fileno()) temporary.replace(destination) directory = os.open(ACTION_INBOX, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(directory) finally: os.close(directory) except Exception: temporary.unlink(missing_ok=True) raise return token def validate_document_review_submission(form: dict[str, list[str]]) -> tuple[dict[str, object], str, str]: if set(form) != {"csrf_token", "payload", "return_to", "return_document"} or any(len(values) != 1 for values in form.values()) or form["return_to"][0] not in {"v5", "v5_labs"}: raise ValueError("invalid document review submission") return_document=form["return_document"][0] if not re.fullmatch(r"api-document-[a-f0-9]{24}",return_document):raise ValueError("invalid return document") payload = json.loads(form["payload"][0], object_pairs_hook=reject_duplicate_object_pairs) required = {"version","action","operation","document_id","target_id","value","unit","decision","metadata"} optional = {"expected_candidate_revision", "preview_revision"} if not isinstance(payload, dict) or not required.issubset(payload) or not set(payload).issubset(required | optional) or payload.get("version") != 1 or payload.get("action") != "document_review": raise ValueError("invalid document review payload") if not isinstance(payload.get("document_id"), str) or not re.fullmatch(r"doc_[a-f0-9]{24}", payload["document_id"]): raise ValueError("invalid document review id") if payload.get("operation") not in {"metadata_review","text_correction","candidate_decision","candidate_remap","candidate_transfer","content_review","discard_document","retry_extraction","original_review","page_review"}: raise ValueError("invalid document operation") if not isinstance(payload.get("target_id"), str) or len(payload["target_id"]) > 64: raise ValueError("invalid document target") if payload["operation"] in {"text_correction","page_review"} and not re.fullmatch(r"page_[1-9][0-9]{0,2}",payload["target_id"]): raise ValueError("invalid document page target") if payload["operation"] in {"candidate_decision","candidate_remap","candidate_transfer"} and not re.fullmatch(r"cand_[a-f0-9]{24}",payload["target_id"]): raise ValueError("invalid document candidate target") expected_revision = payload.get("expected_candidate_revision") preview_revision = payload.get("preview_revision") if form["return_to"][0] == "v5_labs" and payload["operation"] in {"candidate_decision","candidate_remap","candidate_transfer"} and (expected_revision is None or preview_revision is None): raise ValueError("candidate preview binding required") if form["return_to"][0] == "v5_labs" and payload["operation"] == "candidate_decision" and payload.get("decision") == "corrected_confirmed": raise ValueError("laboratory correction requires a regenerated preview") if expected_revision is not None and (type(expected_revision) is not int or not 1 <= expected_revision <= 2_147_483_647): raise ValueError("invalid candidate revision") if preview_revision is not None and (not isinstance(preview_revision, str) or not re.fullmatch(r"[a-f0-9]{64}", preview_revision)): raise ValueError("invalid preview revision") if not isinstance(payload.get("value"), str) or len(payload["value"]) > 4000 or not isinstance(payload.get("unit"), str) or len(payload["unit"]) > 40 or not isinstance(payload.get("decision"), str) or len(payload["decision"]) > 30: raise ValueError("invalid document review value") metadata = payload.get("metadata") if payload["operation"] == "metadata_review": payload["metadata"] = validate_metadata(metadata) elif not isinstance(metadata, dict): raise ValueError("invalid document metadata") return payload, form["return_to"][0], return_document def write_symptom_checkin(day: str, scores: dict[str, int], notes: str) -> str: return write_action_payload( { "version": 1, "action": "symptom_checkin", "date": day, "scores": scores, "notes": notes, } ) def _medication_preview_error(error: Exception) -> tuple[int, dict[str, Any]]: field = getattr(error, "field", None) allowed_fields = { "quantity_value", "dosage_form", "strength", "route_original", "injection_region", "injection_side", "injection_detail", "note", "medication", "route_normalized", "capture_mode", "deviation_confirmed", - "duplicate_confirmed", "preset", "preview", + "duplicate_confirmed", "preset", "preview", "correction_target", "correction_reason", } if isinstance(error, ValueError) and field in allowed_fields: messages = { "quantity_value": "Bitte eine gültige Menge eingeben.", "dosage_form": "Bitte eine gültige Darreichungsform eingeben.", "strength": "Bitte eine gültige Wirkstoffstärke eingeben.", "route_original": "Bitte den dokumentierten Applikationsweg prüfen.", "injection_region": "Bitte die Injektionsregion prüfen.", "injection_side": "Bitte die Seite der Injektionsstelle prüfen.", "injection_detail": "Bitte die Injektionsstelle prüfen.", "route_normalized": "Bitte den Applikationsweg prüfen.", "capture_mode": "Bitte den Erfassungsmodus oder Plantermin prüfen.", "deviation_confirmed": "Bitte die Abweichung bewusst bestätigen.", "duplicate_confirmed": "Bitte einen möglichen Doppeleintrag bewusst bestätigen.", "preset": "Die Vorauswahl ist nicht mehr aktuell. Bitte neu laden.", "preview": "Die Vorschau ist nicht mehr aktuell. Bitte erneut prüfen.", "note": "Bitte die Notiz prüfen.", "medication": "Bitte das ausgewählte Medikament prüfen.", + "correction_target": "Der zu korrigierende Eintrag ist nicht mehr aktuell. Bitte neu laden.", + "correction_reason": "Bitte eine kurze Korrekturbegründung eingeben.", } return 422, {"error": {"kind": "validation", "code": f"{field}_invalid", "field": field, "message": messages.get(field, "Bitte die markierte Angabe prüfen.")}} reason = str(error) rules = ( ("invalid medication action shape", 422, "payload_invalid", "medication", "Die Medikationsangaben sind unvollständig oder ungültig."), ("invalid medication text", 422, "medication_text_invalid", "medication", "Name, Stärke oder Notiz enthält eine unzulässige Angabe."), ("requires a real plan", 422, "planned_event_required", "capture_mode", "Für diesen Modus ist kein belastbarer Plantermin verknüpft."), ("lacks a structured administration preset", 422, "planned_event_unstructured", "capture_mode", "Der Plantermin enthält keine revisionssichere Mengenangabe."), ("differs from verified preset", 422, "preset_deviation_unconfirmed", "deviation_confirmed", "Die Eingabe weicht von der verifizierten Vorauswahl ab und muss bewusst bestätigt werden."), ("administration preset changed", 409, "preset_changed", "preset", "Die verifizierte Vorauswahl hat sich geändert. Bitte Vorschau neu laden."), ("administration preset is no longer available", 409, "preset_unavailable", "preset", "Die verifizierte Vorauswahl ist nicht mehr verfügbar. Bitte neu laden."), ("stale medication preview", 409, "preview_changed", "preview", "Die Vorschau ist nicht mehr aktuell. Bitte erneut prüfen."), ("already consumed", 409, "planned_event_consumed", "capture_mode", "Der Plantermin wurde bereits dokumentiert."), + ("correction target is no longer the latest revision", 409, "correction_target_changed", "correction_target", "Der zu korrigierende Eintrag wurde bereits geändert. Bitte neu laden."), + ("structured correction must preserve occurrence time", 422, "correction_time_changed", "capture_mode", "Eine Korrektur muss den ursprünglichen Zeitpunkt unverändert übernehmen."), + ("structured correction target lacks complete fields", 422, "correction_target_unstructured", "correction_target", "Dieser ältere Eintrag kann nicht mit dem strukturierten Korrekturvertrag bearbeitet werden."), + ("structured correction target", 409, "correction_target_changed", "correction_target", "Der zu korrigierende Eintrag ist nicht mehr verfügbar. Bitte neu laden."), ("explicitly active prescription", 422, "prescription_not_active_for_plan", "capture_mode", "Ein Plantermin kann nur gegen eine ausdrücklich aktive Verordnung dokumentiert werden."), ) for fragment, status_code, code, field, message in rules: if fragment in reason: return status_code, {"error": {"kind": "validation", "code": code, "field": field, "message": message}} if isinstance(error, ValueError): return 422, {"error": {"kind": "validation", "code": "medication_validation_failed", "field": "medication", "message": "Die Medikationsangaben sind ungültig. Bitte die Felder prüfen."}} return 503, {"error": {"kind": "technical", "code": "medication_preview_unavailable", "field": "", "message": "Die Erfassung ist technisch nicht verfügbar. Es wurde nichts vorgemerkt."}} class Handler(BaseHTTPRequestHandler): def send_error( self, code: int, message: str | None = None, explain: str | None = None, ) -> None: request_path = urlparse(getattr(self, "path", "")).path if request_path.startswith("/api/v1/"): send_body = getattr(self, "command", "") != "HEAD" if not host_is_allowed(self.headers.get("Host", "")): self._send_api_error(421, "invalid_host", send_body) return status = 405 if code == 501 else code error_code = "read_only_endpoint" if code == 501 else "request_rejected" self._send_api_error(status, error_code, send_body) return super().send_error(code, message, explain) def _reject_unsupported(self) -> None: if not host_is_allowed(self.headers.get("Host", "")): if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(421, "host_not_allowed", self.command != "HEAD") else: self.send_error(421) return if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(405, "read_only_endpoint", self.command != "HEAD") return self.send_error(405) do_OPTIONS = _reject_unsupported # noqa: N815 do_PUT = _reject_unsupported # noqa: N815 do_DELETE = _reject_unsupported # noqa: N815 do_PATCH = _reject_unsupported # noqa: N815 do_TRACE = _reject_unsupported # noqa: N815 do_CONNECT = _reject_unsupported # noqa: N815 def do_HEAD(self) -> None: # noqa: N802 if not host_is_allowed(self.headers.get("Host", "")): if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(421, "host_not_allowed", False) else: self.send_error(421) return self._handle(send_body=False) def do_GET(self) -> None: # noqa: N802 if not host_is_allowed(self.headers.get("Host", "")): if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(421, "host_not_allowed", True) else: self.send_error(421) return self._handle(send_body=True) def do_POST(self) -> None: # noqa: N802 if not host_is_allowed(self.headers.get("Host", "")): if urlparse(self.path).path.startswith("/api/v1/"): self._send_api_error(421, "host_not_allowed", True) else: self.send_error(421) return api_path = urlparse(self.path).path if api_path.startswith("/api/v1/"): if api_path == "/api/v1/browser-session": self._handle_browser_session(True) elif api_path == CAPTURE_UPLOAD_ROUTE: self._handle_capture_upload() elif api_path == DOCUMENT_UPLOAD_ROUTE: __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/repos/HealthManager__HERMES_CWD_8d46a20096ed__