"""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() if row is None or not isinstance(row[0], bytes) or len(row[0]) != 32: raise RuntimeError("medication public identity key missing") return row[0] def _opaque_digest(key: bytes, domain: str, value: Any) -> str: return hmac.new( key, (domain + ":" + _canonical(value)).encode(), hashlib.sha256, ).hexdigest() def _value(row: sqlite3.Row | Mapping[str, Any], key: str) -> Any: return row[key] if key in row.keys() else None def medication_ref(row: sqlite3.Row | Mapping[str, Any], key: bytes) -> str: return "medrx_" + _opaque_digest( key, "health-medication-prescription-v1", [_value(row, "id"), _value(row, "medikament_name")], )[:24] def event_ref(row: sqlite3.Row | Mapping[str, Any], key: bytes) -> str: return "medevt_" + _opaque_digest( key, "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( prescription, planned, correction_target, planned_consumed=planned_consumed, ), ) def action_preview_revision( prescription: sqlite3.Row, planned: sqlite3.Row | None, correction_target: sqlite3.Row | None, payload: Mapping[str, Any], *, key: bytes, planned_consumed: bool = False, ) -> str: data = dict(payload.get("data") or {}) data.pop("preview_revision", None) bound_payload = { key: payload.get(key) for key in ( "version", "action", "capture_type", "request_version", "idempotency_key", "occurred_at", "ended_at", "attachments", "corrects_entry_id", "withdraws_entry_id", ) } bound_payload["data"] = data return _opaque_digest( key, "health-medication-action-preview-v1", { "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"] == "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: raise MedicationFieldError(field) from error def _validate_action_data_v1(data: Any) -> dict[str, Any]: required = { "contract", "status", "medication_ref", "planned_event_ref", "name", "planned_dose_value", "planned_dose_unit", "actual_dose_value", "actual_dose_unit", "route_original", "route_normalized", "injection_region", "injection_side", "injection_detail", "lot_number", "correction_target_ref", "corrected_target_status", "correction_reason", "note", "preview_revision", "plan_value_confirmed", "deviation_confirmed", "duplicate_confirmed", } if not isinstance(data, dict) or set(data) != required or data.get("contract") != ACTION_CONTRACT_VERSION_V1: raise ValueError("invalid medication action shape") status = data["status"] if status not in STATUSES: raise ValueError("invalid medication status") 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", } 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") 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"] != ("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 mode == "historical" and planned: raise ValueError("historical administration cannot reference a plan") 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 = _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) result = { "contract": ACTION_CONTRACT_VERSION, "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)