diff --git a/scripts/health/dashboard_v5/association_engine.py b/scripts/health/dashboard_v5/association_engine.py index d2b8c8b..7a64979 100644 --- a/scripts/health/dashboard_v5/association_engine.py +++ b/scripts/health/dashboard_v5/association_engine.py @@ -101,121 +101,131 @@ EVENTS = ( EVENT_BY_ID = {item.id: item for item in EVENTS} SYMPTOM_NAMES = { "event.aphthae": {"aphthen/mundulzera", "aphte/mundulzera", "aphthae"}, "event.headache": {"kopfschmerzen", "headache"}, "event.gi": {"gi/darm", "gi-beschwerden", "durchfall/gi-beschwerden", "gi"}, "event.fatigue": {"müdigkeit/fatigue", "fatigue"}, "event.skin": {"haut", "hautveränderung"}, "event.eyes": {"augen", "augenbeschwerden"}, "event.joints": {"gelenke", "gelenkbeschwerden"}, "event.vascular": { "vaskulär/thrombose-warnzeichen", "vaskuläre beschwerden", "vascular", }, } HEALTH_EVENT_CATEGORIES = { "event.stress": {"stress"}, "event.sleep_disruption": {"sleep_disruption", "schlafstörung"}, "event.infection": {"infection", "infekt"}, "event.heat": {"heat", "sauna", "hitze"}, "event.physical_load": {"physical_load", "besondere belastung"}, } def _table_exists(connection: sqlite3.Connection, name: str) -> bool: return ( connection.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,) ).fetchone() is not None ) def _day(raw: Any) -> str | None: text = str(raw or "")[:10] try: return date.fromisoformat(text).isoformat() except ValueError: return None def _score(raw: Any) -> float | None: try: value = float(str(raw).replace(",", ".")) except (TypeError, ValueError): return None return value if math.isfinite(value) else None def event_days( connection: sqlite3.Connection, metric_id: str, start: date, end: date ) -> list[str]: if metric_id not in EVENT_BY_ID: raise ValueError("event_not_allowed") found: set[str] = set() if metric_id == "event.medication" and _table_exists( connection, "medication_administrations" ): rows = list( connection.execute( - "SELECT datum FROM medication_administrations WHERE lower(trim(COALESCE(event_type,'')))='administered' AND datum>=? AND datum<=? ORDER BY datum LIMIT 1001", + """SELECT event.datum FROM medication_administrations AS event + WHERE lower(trim(COALESCE( + CASE WHEN lower(trim(COALESCE(event.event_type,'')))='corrected' + THEN event.corrected_target_status ELSE event.event_type END,'' + )))='administered' + AND event.datum>=? AND event.datum<=? + AND NOT EXISTS ( + SELECT 1 FROM medication_administrations AS correction + WHERE correction.corrects_event_id=event.id + ) + ORDER BY event.datum LIMIT 1001""", (start.isoformat(), end.isoformat()), ) ) if len(rows) > 1000: raise ValueError("event_row_limit") for row in rows: if observed := _day(row[0]): found.add(observed) elif metric_id == "event.supplement" and _table_exists( connection, "supplement_intakes" ): rows = list( connection.execute( "SELECT occurred_at FROM supplement_intakes WHERE status='administered' AND substr(occurred_at,1,10)>=? AND substr(occurred_at,1,10)<=? ORDER BY occurred_at LIMIT 1001", (start.isoformat(), end.isoformat()), ) ) if len(rows) > 1000: raise ValueError("event_row_limit") for row in rows: if observed := _day(row[0]): found.add(observed) elif metric_id in SYMPTOM_NAMES and _table_exists(connection, "symptom_log"): names = SYMPTOM_NAMES[metric_id] rows = list( connection.execute( "SELECT datum,symptom,schwergrad FROM symptom_log WHERE datum>=? AND datum<=? ORDER BY datum,id LIMIT 10001", (start.isoformat(), end.isoformat()), ) ) if len(rows) > 10000: raise ValueError("event_row_limit") for raw_day, raw_name, raw_severity in rows: if ( str(raw_name or "").strip().casefold() in names and (_score(raw_severity) or 0) > 0 ): if observed := _day(raw_day): found.add(observed) if metric_id == "event.aphthae" and _table_exists( connection, "health_event_periods" ): period_rows = list( connection.execute( "SELECT start_date,event_type FROM health_event_periods WHERE start_date>=? AND start_date<=? ORDER BY start_date LIMIT 1001", (start.isoformat(), end.isoformat()), ) ) if len(period_rows) > 1000: raise ValueError("event_row_limit") for row in period_rows: if "apht" in str(row[1] or "").casefold(): if observed := _day(row[0]): found.add(observed) elif metric_id in HEALTH_EVENT_CATEGORIES and _table_exists( connection, "health_events" ): allowed = HEALTH_EVENT_CATEGORIES[metric_id] rows = list( connection.execute( diff --git a/scripts/health/dashboard_v5/comparison_contract.py b/scripts/health/dashboard_v5/comparison_contract.py index a0cd27f..62433ef 100644 --- a/scripts/health/dashboard_v5/comparison_contract.py +++ b/scripts/health/dashboard_v5/comparison_contract.py @@ -236,128 +236,131 @@ def _nutrition_profile_events(connection: sqlite3.Connection, start: date, end: "source": "YAZIO-Import", "data_status": "partially_documented", "review_status": "complete_mapping" if unknown == 0 else "review_open", "mapping_coverage": coverage, "mapped_entries": mapped, "documented_entries": item_count, } ) return events, components def _histamine_events(connection: sqlite3.Connection, start: date, end: date) -> list[dict[str, Any]]: events = [] for row in _nutrition_rows(connection, start, end): item_count = int(row["item_count"] or 0) unknown = max(0, int(row["histamine_unknown_count"] or 0)) mapped = max(0, item_count - unknown) day = date.fromisoformat(str(row["datum"])).isoformat() maximum = _number(row["histamine_max"]) coverage = round(mapped / item_count, 4) if item_count else None details = [ {"label": "Bestätigt zugeordnete Einträge", "value": str(mapped)}, {"label": "Noch ungeprüfte Einträge", "value": str(unknown)}, {"label": "Mapping-Abdeckung", "value": f"{round(coverage * 100)} %" if coverage is not None else "Unbekannt"}, ] if maximum is not None: details.append({"label": "Höchste vorhandene Zuordnungskategorie", "value": f"Kategorie {maximum:g}"}) events.append( { "id": _opaque_event_id("nutrition.histamine", day), "influence_id": "nutrition.histamine", "date": day, "time": None, "category": ( "Histaminbezogene Ernährungseinträge" if mapped > 0 else "Offene Histaminzuordnung" ), "title": ( "Dokumentierte histaminbezogene Ernährungseinträge" if mapped > 0 else "Ernährung dokumentiert – Zuordnung offen" ), "details": details, "source": "Lokale, bestehende Lebensmittelzuordnungen", "data_status": "documented_mapping" if mapped > 0 else "unmapped", "review_status": "complete_mapping" if unknown == 0 else "review_open", "mapping_coverage": coverage, "mapped_entries": mapped, "documented_entries": item_count, } ) return events def _medication_events(connection: sqlite3.Connection, start: date, end: date) -> list[dict[str, Any]]: if not _table_exists(connection, "medication_administrations"): return [] events = [] columns = _table_columns(connection, "medication_administrations") - expressions = [name if name in columns else f"NULL AS {name}" for name in ("datum", "occurred_at", "medication_name", "dose", "route", "event_type", "source")] + expressions = [name if name in columns else f"NULL AS {name}" for name in ("datum", "occurred_at", "medication_name", "dose", "route", "event_type", "source", "corrected_target_status")] + latest_clause = " AND NOT EXISTS (SELECT 1 FROM medication_administrations correction WHERE correction.corrects_event_id=medication_administrations.id)" if "corrects_event_id" in columns else "" rows = connection.execute( f"""SELECT {','.join(expressions)} FROM medication_administrations - WHERE datum>=? AND datum<=? ORDER BY datum,id LIMIT ?""", + WHERE datum>=? AND datum<=?{latest_clause} ORDER BY datum,id LIMIT ?""", (start.isoformat(), end.isoformat(), MAX_EVENTS + 1), ) for row in rows: event_type = _text(row["event_type"], 80).casefold() + if event_type in {"corrected", "korrigiert", "correction"}: + event_type = _text(row["corrected_target_status"], 80).casefold() if event_type not in AFFIRMATIVE_MEDICATION_EVENTS: continue day, time = _local_timestamp(row["occurred_at"], row["datum"]) if day < start.isoformat() or day > end.isoformat(): continue details = [item for item in ( _detail("Dokumentierte Dosis", row["dose"]), _detail("Applikationsweg", row["route"]), ) if item] title = _text(row["medication_name"], 80) or "Medikament dokumentiert" events.append( { "id": _opaque_event_id("medication", day, time, title), "influence_id": "event.medication", "date": day, "time": time, "category": "Medikament", "title": title, "details": details, "source": "Dokumentierte Medikamentengabe", "data_status": "documented", "review_status": "none", "mapping_coverage": None, } ) return events def _capture_events(connection: sqlite3.Connection, kind: str, start: date, end: date) -> list[dict[str, Any]]: if not _table_exists(connection, "capture_entries"): return [] events = [] for row in connection.execute( """SELECT occurred_at,payload_json,source,created_at FROM capture_entries WHERE capture_type='event' AND status='active' AND substr(occurred_at,1,10)>=? AND substr(occurred_at,1,10)<=? ORDER BY occurred_at,created_at LIMIT ?""", ((start - timedelta(days=1)).isoformat(), (end + timedelta(days=1)).isoformat(), MAX_EVENTS + 1), ): try: payload = json.loads(str(row["payload_json"] or "{}")) except json.JSONDecodeError: continue if payload.get("event_kind") != kind: continue day, time = _local_timestamp(row["occurred_at"]) if day < start.isoformat() or day > end.isoformat(): continue details = [] fields = ( (("Dauer", "duration_minutes", "min"), ("Saunagänge", "rounds", ""), ("Temperatur", "temperature_c", "°C"), ("Abkühlung/Kaltphase", "cooling", ""), ("Flüssigkeitszufuhr", "hydration_ml", "ml")) if kind == "sauna" else (("Aktivitätsart", "activity_type", ""), ("Dauer", "duration_minutes", "min"), ("Aktive Energie", "active_kcal", "kcal"), ("Distanz", "distance_km", "km")) ) for label, key, unit in fields: item = _detail(label, payload.get(key), unit) if item: details.append(item) note = _detail("Notiz", payload.get("note")) diff --git a/scripts/health/dashboard_v5/data_provider.py b/scripts/health/dashboard_v5/data_provider.py index 35d81c5..9b53fc6 100644 --- a/scripts/health/dashboard_v5/data_provider.py +++ b/scripts/health/dashboard_v5/data_provider.py @@ -95,140 +95,159 @@ def rolling_baseline(points: list[dict[str, Any]], days: int, minimum: int = 7) return result def symptom_series(connection: sqlite3.Connection) -> tuple[list[dict[str, Any]], set[str]]: if not table_exists(connection, "symptom_log"): return [], set() grouped: dict[str, dict[str, list[int | None]]] = defaultdict(lambda: defaultdict(list)) for row in connection.execute( "SELECT datum,symptom,schwergrad FROM symptom_log WHERE kontext='daily_quick_score' ORDER BY datum,id" ): day = parse_day(row["datum"]) score = parse_score(row["schwergrad"]) label = str(row["symptom"] or "") if day and label in SYMPTOM_LABELS: grouped[day][label].append(score) complete = { day for day, values in grouped.items() if set(values) == SYMPTOM_LABELS and all(len(scores) == 1 and scores[0] is not None for scores in values.values()) } return [ {"date": day, "value": sum(score for scores in grouped[day].values() for score in scores if score is not None), "quality": "complete"} for day in sorted(complete) ], complete def symptom_dimension_series( connection: sqlite3.Connection, label: str, complete_days: set[str] ) -> list[dict[str, Any]]: if not complete_days: return [] result: list[dict[str, Any]] = [] for row in connection.execute( """SELECT datum,schwergrad FROM symptom_log WHERE kontext='daily_quick_score' AND symptom=? ORDER BY datum,id""", (label,), ): day = parse_day(row["datum"]) score = parse_score(row["schwergrad"]) if day in complete_days and score is not None: result.append({"date": day, "value": score, "quality": "complete"}) return result def apple_series(connection: sqlite3.Connection, metric: str, mode: str) -> list[dict[str, Any]]: if not table_exists(connection, "apple_health_records"): return [] rows = list(connection.execute( """SELECT id,metric,value,unit,start_date,end_date,source_name,file_name,file_hash FROM apple_health_records WHERE metric=? AND value IS NOT NULL AND start_date IS NOT NULL ORDER BY start_date,end_date,source_name,id""", (metric,) )) values = apple_daily(metric, mode, _rows=rows) return [{"date": day, "value": value, "quality": "observed"} for day, value in sorted(values.items())] def medication_data(connection: sqlite3.Connection, today: str) -> tuple[dict[str, Any], list[dict[str, Any]]]: if not table_exists(connection, "medication_administrations"): return {"last_administered": None, "next_planned": None}, [] - rows = list(connection.execute( - "SELECT datum,medication_name,dose,event_type,scheduled_next_date FROM medication_administrations ORDER BY datum" - )) + columns = {str(row[1]) for row in connection.execute("PRAGMA table_info(medication_administrations)")} + if {"corrects_event_id", "corrected_target_status"}.issubset(columns): + rows = list(connection.execute( + """SELECT event.datum,event.medication_name,event.dose,event.event_type, + event.scheduled_next_date,event.corrected_target_status + FROM medication_administrations AS event + WHERE NOT EXISTS ( + SELECT 1 FROM medication_administrations AS correction + WHERE correction.corrects_event_id=event.id + ) ORDER BY event.datum""" + )) + else: + rows = list(connection.execute( + """SELECT datum,medication_name,dose,event_type,scheduled_next_date, + NULL AS corrected_target_status + FROM medication_administrations ORDER BY datum""" + )) + def effective_status(row: sqlite3.Row) -> str: + raw = str(row["event_type"] or "").strip().casefold() + if raw in {"corrected", "korrigiert", "correction"}: + return str(row["corrected_target_status"] or "").strip().casefold() + return raw actual = [ row for row in rows - if str(row["event_type"] or "").strip().casefold() in ADMINISTERED + if effective_status(row) in ADMINISTERED and (administered_day := parse_day(row["datum"])) is not None and administered_day <= today ] last = actual[-1] if actual else None cancelled_plans = { (str(row["medication_name"] or "").strip().casefold(), day) for row in rows - if str(row["event_type"] or "").strip().casefold() in EXCLUDED_PLANS + if effective_status(row) in EXCLUDED_PLANS for day in (parse_day(row["datum"]), parse_day(row["scheduled_next_date"])) if day } candidates = [] for row in rows: - event_type = str(row["event_type"] or "").strip().casefold() + event_type = effective_status(row) medication_key = str(row["medication_name"] or "").strip().casefold() scheduled = parse_day(row["scheduled_next_date"]) explicit_day = parse_day(row["datum"]) planned = None if ( scheduled and event_type in ADMINISTERED and explicit_day is not None and explicit_day <= today ): planned = scheduled elif event_type in EXPLICIT_PLANS: planned = explicit_day if planned and planned > today and (medication_key, planned) not in cancelled_plans: candidates.append((planned, row)) next_item = min(candidates, key=lambda item: item[0]) if candidates else None def public_medication(row: sqlite3.Row | None, day: str | None, *, planned: bool) -> dict[str, Any] | None: if row is None or day is None: return None name = public_text(row["medication_name"], 120) dose = public_text(row["dose"], 40, allow_empty=True) if name is None or dose is None: return None item: dict[str, Any] = {"date": day, "name": name, "dose": dose} if planned: item["status"] = "planned" return item medication = { "last_administered": public_medication(last, parse_day(last["datum"]) if last else None, planned=False), "next_planned": public_medication(next_item[1], next_item[0], planned=True) if next_item else None, } events = [] for row in actual: label = public_text(row["medication_name"], 120) day = parse_day(row["datum"]) if label and day: events.append({"date": day, "type": "medication_administered", "label": label}) return medication, events def verified_labs( connection: sqlite3.Connection, limit: int = 12, *, through_day: str | None = None, ) -> list[dict[str, Any]]: if not table_exists(connection, "laborwerte"): return [] rows = list(connection.execute( """SELECT parameter_name,wert,einheit,reference_min,reference_max,abnahme_datum,befund_datum, reference_range_source,source_type FROM laborwerte WHERE lower(trim(COALESCE(validierungsstatus,'')))='validiert' AND verified_against_original=1 AND reference_range_source='scanned_original' AND trim(COALESCE(einheit,''))<>'' ORDER BY id""" )) candidates: dict[tuple[str, str, str], list[sqlite3.Row]] = defaultdict(list) diff --git a/scripts/health/dashboard_v5/read_api.py b/scripts/health/dashboard_v5/read_api.py index 602130e..3563737 100644 --- a/scripts/health/dashboard_v5/read_api.py +++ b/scripts/health/dashboard_v5/read_api.py @@ -31,120 +31,121 @@ from dashboard_v5.data_provider import ( public_text, ) from dashboard_v5.document_chunks import ( decode_chunk_cursor, document_chunks, encode_chunk_cursor, ) from dashboard_v5.document_originals import probe_original from dashboard_v5.document_reconciliation import changed_fragments from dashboard_v5.lab_registry import EXACT_LAB_NUMBER, LAB_ALLOWLIST, LAB_RESULT_NUMBER from dashboard_v5.lab_review import QUALITATIVE as LAB_QUALITATIVE, REVIEW_STATES as LAB_REVIEW_STATES, build_lab_review, public_engine_label from dashboard_v5.metric_catalog_v2 import ( APPLE_ANALYTICS_UNITS, APPLE_SOURCE_SPECS, BY_ID_V2, LAB_CATALOG_SPECS, MetricV2, SourceInventoryLimitError, catalog_search, inventory_sources, public_catalog, ) from dashboard_v5.nutrition_contract import ( NULL_VALUE_CONTRACT_VERSION, NUTRIENT_CONTRACT_VERSION, NUTRIENT_CONTRACTS, normalize_nutrient, normalize_nutrient_status, ) from dashboard_v5.nutrition_references import build_overview from dashboard_v5.nutrition_mapping_review import ( ReviewLimitError, UnsafeReviewMetadataError, build_review as build_mapping_review, ) from dashboard_v5.association_engine import ( MAX_ANALYSIS_DAYS, analyze as analyze_association, catalog as association_catalog, event_days as association_event_days, ) from dashboard_v5.comparison_contract import ( MAX_DAYS as MAX_COMPARISON_DAYS, build as build_comparison, catalog as comparison_catalog, ) from dashboard_v5.reference_contract import public_reference_context from dashboard_v5.supplement_read import supplement_nutrient_totals, supplements from dashboard_v5.source_status import build_source_status from dashboard_v5.observation_contract import PUBLIC_CONTRACT_VERSION, STATUS as OBSERVATION_STATUS, templates as observation_templates from dashboard_v5.observation_engine import evaluate_observation, evaluate_plan, list_observations, observation_detail from dashboard_v5.medication_schema import assert_schema as assert_medication_schema from dashboard_v5.medication_contract import ( PUBLIC_CONTRACT_VERSION as MEDICATION_HISTORY_CONTRACT, event_ref as medication_event_ref, list_events as medication_event_rows, list_prescriptions as medication_prescription_rows, medication_ref as medication_prescription_ref, prescription_preset as medication_prescription_preset, planned_event_preset as medication_planned_event_preset, + correction_event_preset as medication_correction_event_preset, public_action_context_token as medication_public_context_token, public_identity_key as medication_public_identity_key, plan_is_consumed as medication_plan_is_consumed, ) TZ_NAME = "Europe/Zurich" LOCAL_TZ = ZoneInfo(TZ_NAME) MAX_RANGE_DAYS = 3660 MAX_SERIES_ROWS = 3660 MAX_RAW_APPLE_ROWS = 100000 MAX_RAW_LAB_ROWS = 10000 MAX_EVENT_SOURCE_ROWS = 10000 MAX_DAILY_SOURCE_ROWS = 30000 MAX_EVENT_ROWS = 1000 MAX_LAB_ROWS = 100 MAX_SEARCH_ROWS_PER_GROUP = 20 MAX_CALENDAR_DAYS = 62 MAX_NUTRITION_DAYS = 180 MAX_NUTRITION_ITEMS = 200 MAX_NUTRITION_QUEUE = 50 MAX_FUTURE_DAYS = 366 MAX_QUERY_BYTES = 512 QUERY_TIMEOUT_SECONDS = 1.0 ALLOWED_RESOLUTIONS = frozenset({"day", "week"}) ALLOWED_EVENT_TYPES = frozenset( {"medication_administered", "medication", "supplement", "symptom_day", "health_event", "health_period", "nutrition_day", "laboratory", "document", "appointment"} ) NUTRIENT_ALLOWLIST = { key: (contract.label, contract.unit) for key, contract in NUTRIENT_CONTRACTS.items() } MEAL_LABELS = { "breakfast": "Frühstück", "lunch": "Mittagessen", "dinner": "Abendessen", "snack": "Snack", "unassigned": "Nicht zugeordnet", } FORBIDDEN_METADATA_TEXT = re.compile( r"(?:/home/|/tmp/|/Users/|/(?:var|etc|usr|opt|srv|root|run|mnt|media|private)/|" r"~/|file:|https?://|[A-Za-z]:[\\/]|" r"(?:^|[\s:|])/(?:[^/\s]+/)+[^/\s]+|\.\.[\\/]|\\\\[^\\/\s]+[\\/]|" r"\.hermes(?:/|$))", re.IGNORECASE, ) DRIVE_ID_TOKEN = re.compile(r"(? dict[str, raise APIError(422, "range_too_large") coverage = _coverage(points, start, end) if end is None: coverage["to"] = today if coverage["from"] is not None: coverage["expected_days"] = ( date.fromisoformat(today) - date.fromisoformat(coverage["from"]) ).days + 1 coverage["missing_days"] = max( 0, coverage["expected_days"] - coverage["observed_days"] ) coverage["expectation"] = metric.expected_frequency if metric.expected_frequency == "intermittent": coverage["expected_days"] = None coverage["missing_days"] = None coverage["gaps"] = "sparse_observations_not_interpolated" aggregation_rule = { "id": "daily_metric_points", "minimum_observations": 1, "requires_complete_calendar_week": False, } if resolution == "week": aggregation_rule = _weekly_rule(metric) points = _weekly(points, metric, aggregation_rule) if len(points) > MAX_SERIES_ROWS: raise APIError(422, "row_limit_exceeded") return { "metric": metric.id, "label": metric.label, "unit": metric.unit, "aggregation": metric.aggregation, "aggregation_rule": aggregation_rule, "resolution": resolution, "source": { "type": metric.source, "identifier": metric.source_identifier, "parser": metric.parser_contract, }, "reference_context": public_reference_context(metric), "coverage": coverage, "timezone": TZ_NAME, "today": today, "points": points, } def _events(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]: start, end = parse_range(params, required=True) assert start is not None and end is not None raw_types = params.get("types", "") if not raw_types: selected = set(ALLOWED_EVENT_TYPES) else: selected = set(raw_types.split(",")) if "" in selected or not selected <= ALLOWED_EVENT_TYPES: raise APIError(400, "event_type_not_allowed") result: list[dict[str, Any]] = [] if "medication_administered" in selected and table_exists( connection, "medication_administrations" ): - administered_values = tuple(sorted(ADMINISTERED)) - if len(administered_values) != 6: - raise APIError(503, "api_contract_invalid") - rows = list( - connection.execute( - """SELECT datum,medication_name,event_type FROM medication_administrations - WHERE datum>=? AND datum<=? - AND lower(trim(COALESCE(event_type,''))) IN (?,?,?,?,?,?) - ORDER BY datum,id LIMIT ?""", - ( - start.isoformat(), - end.isoformat(), - *administered_values, - MAX_EVENT_SOURCE_ROWS + 1, - ), + medication_record = _effective_medication_projection( + _record_medications( + connection, {"from": start.isoformat(), "to": end.isoformat()} ) ) - if len(rows) > MAX_EVENT_SOURCE_ROWS: - raise APIError(422, "source_row_limit_exceeded") - for row in rows: - label = safe_metadata_text(row["medication_name"], 120) - day = parse_day(row["datum"]) - if day and label: - result.append( - {"date": day, "type": "medication_administered", "label": label} - ) + for row in medication_record["administered"]: + result.append( + { + "date": row["date"], + "type": "medication_administered", + "label": row["name"], + } + ) if "medication" in selected and table_exists(connection, "medication_administrations"): - rows = list(connection.execute( - """SELECT datum,medication_name,event_type FROM medication_administrations - WHERE datum>=? AND datum<=? ORDER BY datum,id LIMIT ?""", - (start.isoformat(), end.isoformat(), MAX_EVENT_SOURCE_ROWS + 1), - )) - if len(rows) > MAX_EVENT_SOURCE_ROWS: - raise APIError(422, "source_row_limit_exceeded") - for row in rows: - raw = str(row["event_type"] or "").strip().casefold() - if raw in ADMINISTERED: - continue - category = ( - "planned" if raw in {"planned", "scheduled", "geplant"} - else "missed" if raw in {"missed", "verpasst", "ausgelassen"} - else "corrected" if raw in {"corrected", "korrigiert", "correction"} - else "unknown" + medication_record = _effective_medication_projection( + _record_medications( + connection, {"from": start.isoformat(), "to": end.isoformat()} ) - label = safe_metadata_text(row["medication_name"], 120) - day = parse_day(row["datum"]) - if day and label: - result.append({"date": day, "type": "medication", "category": category, "label": label}) + ) + for category in ("planned", "missed", "corrected", "unknown"): + for row in medication_record[category]: + result.append( + { + "date": row["date"], + "type": "medication", + "category": category, + "label": row["name"], + } + ) if "supplement" in selected: try: supplement_events = supplements(connection, start, end) except ValueError as error: raise APIError(422, str(error)) from error for kind in ("administered", "missed", "corrected"): for item in supplement_events[kind]: result.append( { "date": item["date"], "type": "supplement", "category": kind, "label": item["product"], } ) if "symptom_day" in selected and table_exists(connection, "symptom_log"): for point in _symptom_total_points(connection, start, end): result.append( { "date": point["date"], "type": "symptom_day", "label": "Symptome vollständig dokumentiert", } ) for row in connection.execute( "SELECT datum,symptom FROM symptom_log WHERE kontext='additional_symptom' AND datum>=? AND datum<=? ORDER BY datum,id LIMIT ?", (start.isoformat(), end.isoformat(), MAX_EVENT_SOURCE_ROWS + 1), ): day = parse_day(row["datum"]) label = safe_metadata_text(row["symptom"], 80, allow_empty=False) if day and label: result.append({"date": day, "type": "symptom_day", "category": "additional_symptom", "label": label}) if "health_event" in selected and table_exists(connection, "health_events"): rows = list( connection.execute( "SELECT date,category,parameter FROM health_events WHERE date>=? AND date<=? ORDER BY date,id LIMIT ?", (start.isoformat(), end.isoformat(), MAX_EVENT_SOURCE_ROWS + 1), ) ) if len(rows) > MAX_EVENT_SOURCE_ROWS: raise APIError(422, "source_row_limit_exceeded") for row in rows: category = safe_metadata_text(row["category"], 80, allow_empty=True) parameter = safe_metadata_text(row["parameter"], 120, allow_empty=True) day = parse_day(row["date"]) if day and category is not None and parameter is not None: result.append( { "date": day, "type": "health_event", "category": category, "label": parameter, } ) if "health_period" in selected and table_exists(connection, "health_event_periods"): rows = list( connection.execute( """SELECT start_date,end_date,event_type,label FROM health_event_periods WHERE start_date<=? AND COALESCE(end_date,start_date)>=? ORDER BY start_date,id LIMIT ?""", @@ -1306,158 +1288,145 @@ def _day_symptoms(connection: sqlite3.Connection, day: str) -> dict[str, Any]: for label in sorted(SYMPTOM_LABELS) } additional: list[dict[str, Any]] = [] if table_exists(connection, "symptom_log"): rows = list( connection.execute( "SELECT symptom,schwergrad,notizen FROM symptom_log WHERE datum=? AND kontext='daily_quick_score' ORDER BY id LIMIT 100", (day,), ) ) for row in rows: label = str(row["symptom"] or "") if label not in dimensions: continue score = parse_score(row["schwergrad"]) note = _safe_day_text(row["notizen"], 300) dimensions[label] = { "value": score, "note": note, "status": "observed" if score is not None else "not_documented", } for row in connection.execute( """SELECT symptom,schwergrad,notizen FROM symptom_log WHERE datum=? AND COALESCE(kontext,'')<>'daily_quick_score' ORDER BY id LIMIT 100""", (day,), ): label = _safe_day_text(row["symptom"], 80, allow_empty=False) severity = _safe_day_text(row["schwergrad"], 40) note = _safe_day_text(row["notizen"], 160) if label and severity is not None: additional.append( { "name": label, "severity": severity, "note": note or "", "status": "observed", } ) values = [entry["value"] for entry in dimensions.values()] complete = all(value is not None for value in values) notes = sorted({entry["note"] for entry in dimensions.values() if entry["note"]}) return { "dimensions": [{"name": name, **entry} for name, entry in dimensions.items()], "total": sum(values) if complete else None, "complete": complete, "documented_dimensions": sum(value is not None for value in values), "expected_dimensions": len(dimensions), "additional": additional, "additional_count": len(additional), "notes": notes, } def _day_medications( connection: sqlite3.Connection, requested: str ) -> dict[str, list[dict[str, Any]]]: result = {"planned": [], "administered": [], "missed": [], "corrected": []} if not table_exists(connection, "medication_administrations"): return result - rows = list( - connection.execute( - """SELECT datum,medication_name,dose,route,event_type,scheduled_next_date,notes,source - FROM medication_administrations - WHERE datum=? OR scheduled_next_date=? ORDER BY id LIMIT 200""", - (requested, requested), + record = _effective_medication_projection( + _record_medications( + connection, {"from": requested, "to": requested}, allow_future=True ) ) - missed = {"missed", "verpasst", "ausgelassen"} - corrected = {"corrected", "korrigiert", "correction"} - for row in rows: - event_type = str(row["event_type"] or "").strip().casefold() - name = _safe_day_text(row["medication_name"], 120, allow_empty=False) - dose = _safe_day_text(row["dose"], 40) - route = _safe_day_text(row["route"], 40) - note = _safe_day_text(row["notes"], 300) - source = _safe_day_text(row["source"], 80) - if not name or dose is None or route is None or note is None or source is None: - continue - item = { - "name": name, - "dose": dose, - "route": route, - "note": note, - "source": source, - } - datum = parse_day(row["datum"]) - scheduled = parse_day(row["scheduled_next_date"]) - if datum == requested and event_type in ADMINISTERED: - result["administered"].append(item) - elif datum == requested and event_type in missed: - result["missed"].append(item) - elif datum == requested and event_type in corrected: - result["corrected"].append(item) - elif ( - datum == requested and event_type in {"planned", "scheduled", "geplant"} - ) or scheduled == requested: - result["planned"].append(item) + for bucket in ("planned", "administered", "missed", "corrected"): + for row in record[bucket]: + if bucket == "planned": + values = [row.get("planned_quantity_value"), row.get("planned_dosage_form")] + strength = row.get("planned_strength") + else: + values = [row.get("actual_quantity_value"), row.get("actual_dosage_form")] + strength = row.get("actual_strength") + structured = " ".join(str(value) for value in values if value) + if strength: + structured += f" · {strength}" + result[bucket].append( + { + "name": row["name"], + "dose": structured or row.get("legacy_dose") or "", + "route": row.get("route_original") or row.get("route_normalized") or "", + "note": row.get("note") or "", + "source": row.get("source") or "", + } + ) return result def _day_events( connection: sqlite3.Connection, requested: str ) -> dict[str, list[dict[str, Any]]]: events: list[dict[str, Any]] = [] periods: list[dict[str, Any]] = [] if table_exists(connection, "health_events"): for row in connection.execute( "SELECT category,parameter,value,unit,source,notes FROM health_events WHERE date=? ORDER BY id LIMIT 200", (requested,), ): fields = { "category": _safe_day_text(row["category"], 80), "label": _safe_day_text(row["parameter"], 120), "value": _safe_day_text(row["value"], 80), "unit": _safe_day_text(row["unit"], 40), "source": _safe_day_text(row["source"], 80), # Notes are optional display metadata. An overlong or unsafe note # must not hide an otherwise safe, documented event. "note": _safe_day_text(row["notes"], 160) or "", } if fields["category"] and fields["label"] and fields["source"]: fields["value"] = fields["value"] or "" fields["unit"] = fields["unit"] or "" events.append(fields) if table_exists(connection, "health_event_periods"): for row in connection.execute( """SELECT start_date,end_date,event_type,label,source,notes FROM health_event_periods WHERE start_date<=? AND COALESCE(end_date,start_date)>=? ORDER BY start_date,id LIMIT 200""", (requested, requested), ): start = parse_day(row["start_date"]) end = parse_day(row["end_date"]) if row["end_date"] else start event_type = _safe_day_text(row["event_type"], 80, allow_empty=False) label = _safe_day_text(row["label"], 120) source = _safe_day_text(row["source"], 80) note = _safe_day_text(row["notes"], 300) if ( start and end and start <= requested <= end and event_type and label is not None and source is not None and note is not None ): periods.append( { "start_date": start, "end_date": end, "type": event_type, "label": label, "source": source, "note": note, } ) return {"events": events, "periods": periods} @@ -2639,134 +2608,134 @@ def _calendar(connection: sqlite3.Connection, params: dict[str, str]) -> dict[st def grouped_count( table: str, date_column: str, category: str, where: str = "1=1" ) -> None: if not table_exists(connection, table): return sql = ( f"SELECT substr({date_column},1,10) AS day,COUNT(*) AS count FROM {table} " f"WHERE {where} AND substr({date_column},1,10) BETWEEN ? AND ? GROUP BY substr({date_column},1,10)" ) for row in connection.execute(sql, (start.isoformat(), end.isoformat())): if row["day"] in counts_by_day: counts_by_day[row["day"]][category] += min(int(row["count"]), 999) # Apple Health uses the same allowlisted metric paths and canonical Zurich-day # normalization as /api/v1/day. A row is visible only if its source identifier # is released in APPLE_SOURCE_SPECS; unrecognized Apple rows never influence # calendar categories or counts. apple_metrics: dict[str, MetricV2] = {} for metric in BY_ID_V2.values(): if ( metric.source == "apple_health" and metric.value_type != "panel" and metric.source_identifier in APPLE_SOURCE_SPECS ): apple_metrics.setdefault(metric.source_identifier, metric) for metric in apple_metrics.values(): for point in _metric_points(connection, metric, start, end): observed_day = point["date"] if observed_day in counts_by_day: counts_by_day[observed_day]["measurements"] = min( counts_by_day[observed_day]["measurements"] + 1, 999 ) grouped_count("vitalzeichen", "datum", "measurements") grouped_count("symptom_log", "datum", "symptoms") grouped_count( "laborwerte", "abnahme_datum", "labs", "validierungsstatus='validiert' AND verified_against_original=1", ) grouped_count("health_events", "date", "events") grouped_count("dokumente", "document_date", "documents", "review_status='geprueft'") grouped_count("nutrition_daily_summary_v2", "datum", "nutrition") grouped_count("arztbesuche", "datum", "appointments") grouped_count( "capture_entries", "occurred_at", "supplements", "capture_type='supplement' AND status='active'", ) grouped_count( "capture_entries", "occurred_at", "events", "capture_type='photo' AND status='active'", ) if table_exists(connection, "medication_administrations"): - seen_medications: dict[str, set[int]] = {day: set() for day in dates} - for row in connection.execute( - """SELECT id,datum,scheduled_next_date FROM medication_administrations - WHERE datum BETWEEN ? AND ? OR scheduled_next_date BETWEEN ? AND ? LIMIT 10000""", - (start.isoformat(), end.isoformat(), start.isoformat(), end.isoformat()), - ): - for candidate in ( - parse_day(row["datum"]), - parse_day(row["scheduled_next_date"]), - ): - if candidate in seen_medications: - seen_medications[candidate].add(int(row["id"])) - for day, identifiers in seen_medications.items(): - counts_by_day[day]["medications"] = min(len(identifiers), 999) + medication_record = _effective_medication_projection( + _record_medications( + connection, + {"from": start.isoformat(), "to": end.isoformat()}, + allow_future=True, + ) + ) + for bucket in ("planned", "administered", "missed", "corrected", "unknown"): + for row in medication_record[bucket]: + candidate = row.get("date") + if candidate in counts_by_day: + counts_by_day[candidate]["medications"] = min( + counts_by_day[candidate]["medications"] + 1, 999 + ) if table_exists(connection, "health_event_periods"): for row in connection.execute( """SELECT start_date,end_date FROM health_event_periods WHERE start_date<=? AND COALESCE(end_date,start_date)>=? LIMIT 10000""", (end.isoformat(), start.isoformat()), ): period_start = parse_day(row["start_date"]) period_end = parse_day(row["end_date"]) if row["end_date"] else period_start if not period_start or not period_end: continue current = max(period_start, start.isoformat()) bounded_end = min(period_end, end.isoformat()) while current <= bounded_end: counts_by_day[current]["events"] = min( counts_by_day[current]["events"] + 1, 999 ) current = (date.fromisoformat(current) + timedelta(days=1)).isoformat() result = [] for current in dates: counts = counts_by_day[current] categories = [name for name, count in counts.items() if count] result.append( { "date": current, "is_today": current == today.isoformat(), "categories": categories, "counts": counts, "completeness": "documented" if categories else "not_documented", "codes": [f"has_{category}" for category in categories], } ) return { "from": start.isoformat(), "to": end.isoformat(), "today": today.isoformat(), "timezone": TZ_NAME, "days": result, } RECORD_TABS = frozenset( {"overview", "labs", "medications", "appointments", "documents", "report"} ) RECORD_DOCUMENT_SORTS = { "document_date_desc": "document_date DESC,id DESC", "document_date_asc": "document_date ASC,id ASC", "import_date_desc": "upload_datum DESC,id DESC", "category": "kategorie COLLATE NOCASE,id DESC", "institution": "institution COLLATE NOCASE,id DESC", "type": "daten_typ,id DESC", "review_status": "review_status,id DESC", } RECORD_MAX_ROWS = 100 RECORD_PAGE_SIZE = 25 RECORD_MAX_PAGE_SIZE = 50 RECORD_MAX_MATCHES = 50 RECORD_MAX_SECTIONS = 20 RECORD_MAX_SNIPPET_CHARS = 320 @@ -3454,123 +3423,128 @@ def _document_review_workspace(connection: sqlite3.Connection, opaque: str) -> d "review_blockers": review_blockers, "pages": pages, "candidates": candidates, "reconciliation":reconciliation,"navigation":navigation,"overall_progress":overall, "notice": "Maschinell extrahierter Inhalt – noch nicht gegen das Original geprüft" if processing["content_status"] != "reviewed" else "Inhalt durch Benutzer geprüft", } def _document_review_queue(connection: sqlite3.Connection) -> dict[str, Any]: if not table_exists(connection,"document_reconciliation"): return {"groups":[],"total_decisions":0,"progress":{"completed":0,"total":0}} definitions=[ ("now_reviewable","Jetzt prüfbar","r.queue_bucket='now_reviewable'"), ("original_missing","Original fehlt","r.queue_bucket='original_missing'"), ("conflict","Widerspruch","r.conflict_count>0"), ("laboratory","Laborwerte","EXISTS(SELECT 1 FROM document_candidates c WHERE c.document_id=d.id AND c.candidate_type IN ('laboratory_value','reference_range') AND c.status IN ('open','conflicting','confirmed','corrected_confirmed'))"), ("medication","Medikamente","EXISTS(SELECT 1 FROM document_candidates c WHERE c.document_id=d.id AND c.candidate_type IN ('medication','dose','intake_status') AND c.status IN ('open','conflicting','confirmed','corrected_confirmed'))"), ("statement","Diagnosen/Beschwerden","EXISTS(SELECT 1 FROM document_candidates c WHERE c.document_id=d.id AND c.candidate_type IN ('diagnosis','symptom','important_event') AND c.status IN ('open','conflicting'))"), ("appointment","Termine","EXISTS(SELECT 1 FROM document_candidates c WHERE c.document_id=d.id AND c.candidate_type='appointment' AND c.status IN ('open','conflicting','confirmed','corrected_confirmed'))"), ("duplicate","Dubletten","r.byte_duplicate_of IS NOT NULL"), ("completed","Abgeschlossen","r.queue_bucket='no_action' AND r.byte_duplicate_of IS NULL"), ] groups=[] for code,label,clause in definitions: rows=connection.execute(f"""SELECT d.id,d.document_date,d.kategorie,d.institution,r.open_decisions FROM dokumente d JOIN document_reconciliation r ON r.document_id=d.id WHERE {clause} ORDER BY r.priority DESC,r.open_decisions DESC,d.document_date DESC,d.id DESC LIMIT 101""").fetchall() if not rows:continue first=rows[0];category=safe_metadata_text(first["kategorie"],80,allow_empty=True) or "";institution=safe_metadata_text(first["institution"],120,allow_empty=True) or "" groups.append({"code":code,"label":label,"count":min(len(rows),100),"decision_count":sum(int(row["open_decisions"]) for row in rows[:100]),"truncated":len(rows)>100,"first_document":api_document_id(first["id"],parse_day(first["document_date"]),category,institution)}) total_decisions=int(connection.execute("SELECT COALESCE(SUM(open_decisions),0) FROM document_reconciliation WHERE queue_bucket='now_reviewable'").fetchone()[0]) total=int(connection.execute("SELECT COUNT(*) FROM document_reconciliation").fetchone()[0]);completed=int(connection.execute("SELECT COUNT(*) FROM document_reconciliation WHERE queue_bucket='no_action'").fetchone()[0]) return {"groups":groups,"total_decisions":total_decisions,"next_document":next((group["first_document"] for group in groups if group["code"]=="now_reviewable"),None),"progress":{"completed":completed,"total":total}} def _document_compare(connection: sqlite3.Connection, opaque: str) -> dict[str, Any]: row = _resolve_record_document(connection, opaque, reviewed=False) current = connection.execute("SELECT page_number,normalized_text FROM document_pages WHERE document_id=? AND text_version=(SELECT MAX(text_version) FROM document_pages WHERE document_id=?) ORDER BY page_number",(int(row["id"]),int(row["id"]))).fetchall() previous = connection.execute("""SELECT d.id,d.document_date,d.kategorie,d.institution FROM dokumente d JOIN document_processing p ON p.document_id=d.id WHERE d.id<>? AND d.kategorie=? AND d.document_date<=COALESCE(?,d.document_date) ORDER BY d.document_date DESC,d.id DESC LIMIT 1""",(int(row["id"]),row["kategorie"],row["document_date"])).fetchone() if previous is None: return {"id":opaque,"previous":None,"new":len(current),"changed":0,"removed":0,"identical":0,"sections":[]} old = connection.execute("SELECT page_number,normalized_text FROM document_pages WHERE document_id=? AND text_version=(SELECT MAX(text_version) FROM document_pages WHERE document_id=?) ORDER BY page_number",(int(previous["id"]),int(previous["id"]))).fetchall() sections=[]; identical=changed=new=0 used=set() for page in current: best=None; score=0.0 for candidate in old: if int(candidate["page_number"]) in used: continue left=set(str(page["normalized_text"]).casefold().split()); right=set(str(candidate["normalized_text"]).casefold().split()) value=len(left&right)/len(left|right) if left and right else 0.0 if value>score: score,best=value,candidate if best is not None and score==1.0: kind="identical"; identical+=1; used.add(int(best["page_number"])) elif best is not None and score>=.60: kind="changed"; changed+=1; used.add(int(best["page_number"])) else: kind="new"; new+=1 sections.append({"page":int(page["page_number"]),"status":kind,"similarity":round(score,3)}) category=safe_metadata_text(previous["kategorie"],80,allow_empty=True) or ""; institution=safe_metadata_text(previous["institution"],120,allow_empty=True) or "" return {"id":opaque,"previous":api_document_id(previous["id"],parse_day(previous["document_date"]),category,institution),"new":new,"changed":changed,"removed":max(0,len(old)-len(used)),"identical":identical,"sections":sections,"medical_evaluation":False} def _record_medications( - connection: sqlite3.Connection, params: dict[str, str] + connection: sqlite3.Connection, params: dict[str, str], *, allow_future: bool = False ) -> dict[str, Any]: - start, end = parse_range(params) + if allow_future and params.get("from") and params.get("to"): + start, end = parse_iso_day(params["from"]), parse_iso_day(params["to"]) + if end < start: + raise APIError(400, "range_reversed") + else: + start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: raise APIError(503, "medication_schema_unavailable") from error identity_key = medication_public_identity_key(connection) prescriptions = medication_prescription_rows(connection) prescription_by_id = {int(row["id"]): row for row in prescriptions} prescription_by_name: dict[str, list[sqlite3.Row]] = defaultdict(list) for row in prescriptions: prescription_by_name[str(row["medikament_name"])].append(row) public_prescriptions = [] for row in prescriptions: name = safe_metadata_text(row["medikament_name"], 120) dose = safe_metadata_text(row["dosierung"], 80, allow_empty=True) route = safe_metadata_text(row["anwendungsform"], 80, allow_empty=True) source = safe_metadata_text(row["prescription_status_source"], 80, allow_empty=True) provenance = safe_metadata_text(row["prescription_status_provenance"], 160, allow_empty=True) status = str(row["prescription_status"] or "unknown") trusted_status = bool( status in {"active", "ended", "paused"} and source and provenance and row["business_revision"] ) if not trusted_status or None in (name, dose, route): status = "unknown" item = { "id": medication_prescription_ref(row, identity_key), "name": name, "documented_dose": dose, "documented_route": route, "status": status, "status_source": source, @@ -3581,194 +3555,229 @@ def _record_medications( } public_prescriptions.append(item) empty["prescriptions"] = public_prescriptions empty["current_prescriptions"] = [item for item in public_prescriptions if item["status"] == "active"] empty["other_prescriptions"] = [item for item in public_prescriptions if item["status"] != "active"] rows = medication_event_rows(connection) corrected_origins = { int(row["corrects_event_id"]) for row in rows if str(row["event_type"] or "").strip().casefold() == "corrected" and row["corrects_event_id"] is not None } aliases = { "oral": "oral", "subkutan": "subcutaneous", "subcutaneous": "subcutaneous", "intravenös": "intravenous", "intravenous": "intravenous", "intramuskulär": "intramuscular", "intramuscular": "intramuscular", "äußerlich": "topical", "topical": "topical", "inhalativ": "inhaled", "inhaled": "inhaled", } sources: set[str] = set() truncated_sections: set[str] = set() for row in rows: raw_status = str(row["event_type"] or "").strip().casefold() if raw_status in {"planned", "scheduled", "geplant"}: bucket = "planned" elif raw_status in ADMINISTERED: bucket = "administered" elif raw_status in {"missed", "verpasst", "ausgelassen"}: bucket = "missed" elif raw_status in {"corrected", "korrigiert", "correction"}: bucket = "corrected" else: bucket = "unknown" recorded_day = parse_day(row["datum"]) scheduled = parse_day(row["scheduled_next_date"]) effective_day = scheduled or recorded_day if bucket == "planned" else recorded_day if not recorded_day or not effective_day: continue if start and effective_day < start.isoformat(): continue if end and effective_day > end.isoformat(): continue name = safe_metadata_text(row["medication_name"], 120) source = safe_metadata_text(row["source"], 80, allow_empty=True) if not name or source is None: continue if status_filter and bucket != status_filter: continue if source_filter and source != source_filter: continue candidates: list[sqlite3.Row] = [] if row["medication_id"] is not None and int(row["medication_id"]) in prescription_by_id: candidates = [prescription_by_id[int(row["medication_id"])]] elif len(prescription_by_name.get(name, [])) == 1: candidates = prescription_by_name[name] prescription = candidates[0] if len(candidates) == 1 else None if medication_filter and ( prescription is None or medication_prescription_ref(prescription, identity_key) != medication_filter ): continue - route_original = safe_metadata_text(row["route_original"] or row["route"], 60, allow_empty=True) + route_original = safe_metadata_text(row["route_original"], 60, allow_empty=True) + route_display_fallback = safe_metadata_text(row["route"], 60, allow_empty=True) route_normalized = safe_metadata_text(row["route_normalized"], 24, allow_empty=True) - if not route_normalized and route_original: - route_normalized = aliases.get(route_original.casefold(), "unknown") + if not route_normalized and route_display_fallback: + route_normalized = aliases.get(route_display_fallback.casefold(), "unknown") planned_context = None if row["planned_event_id"] is not None: planned_context = next((candidate for candidate in rows if int(candidate["id"]) == int(row["planned_event_id"])), None) correction_context = None if row["corrects_event_id"] is not None: correction_context = next((candidate for candidate in rows if int(candidate["id"]) == int(row["corrects_event_id"])), None) plan_consumed = ( medication_plan_is_consumed(connection, int(row["id"])) if bucket == "planned" else False ) + capture_entry = ( + connection.execute( + "SELECT entry_id FROM capture_action_log WHERE action_hash=?", + (str(row["business_revision"]),), + ).fetchone() + if row["business_revision"] and table_exists(connection, "capture_action_log") + else None + ) item = { "id": medication_event_ref(row, identity_key), "date": effective_day, "recorded_date": recorded_day, "occurred_at": safe_metadata_text(row["occurred_at"], 40, allow_empty=True), "scheduled_next_date": scheduled, "name": name, "status": bucket, "effective_status": safe_metadata_text(row["corrected_target_status"], 20, allow_empty=True) if bucket == "corrected" else bucket, "legacy_dose": safe_metadata_text(row["dose"], 80, allow_empty=True), "planned_dose_value": safe_metadata_text(row["planned_dose_value"], 40, allow_empty=True), "planned_dose_unit": safe_metadata_text(row["planned_dose_unit"], 30, allow_empty=True), "actual_dose_value": safe_metadata_text(row["actual_dose_value"], 40, allow_empty=True), "actual_dose_unit": safe_metadata_text(row["actual_dose_unit"], 30, allow_empty=True), "planned_quantity_value": safe_metadata_text(row["planned_quantity_value"], 40, allow_empty=True), "planned_dosage_form": safe_metadata_text(row["planned_dosage_form"], 40, allow_empty=True), "planned_strength": safe_metadata_text(row["planned_strength"], 80, allow_empty=True), "actual_quantity_value": safe_metadata_text(row["actual_quantity_value"], 40, allow_empty=True), "actual_dosage_form": safe_metadata_text(row["actual_dosage_form"], 40, allow_empty=True), "actual_strength": safe_metadata_text(row["actual_strength"], 80, allow_empty=True), "administration_preset": medication_planned_event_preset(row, identity_key) if bucket == "planned" else None, + "correction_administration_preset": medication_correction_event_preset(row, identity_key) if int(row["id"]) not in corrected_origins and (bucket == "administered" or (bucket == "corrected" and str(row["corrected_target_status"] or "").strip().casefold() == "administered")) else None, + "capture_entry_ref": str(capture_entry[0]) if capture_entry is not None else None, "route_original": route_original, "route_normalized": route_normalized or "unknown", "injection_region": safe_metadata_text(row["injection_region"], 80, allow_empty=True), "injection_side": safe_metadata_text(row["injection_side"], 20, allow_empty=True), "injection_detail": safe_metadata_text(row["injection_detail"], 120, allow_empty=True), "lot_number": safe_metadata_text(row["lot_number"], 80, allow_empty=True), "note": safe_metadata_text(row["notes"], 300, allow_empty=True), "source": source, "prescription_id": medication_prescription_ref(prescription, identity_key) if prescription is not None else None, "planned_event_id": medication_event_ref(planned_context, identity_key) if planned_context is not None else None, "corrects_event_id": medication_event_ref(correction_context, identity_key) if correction_context is not None else None, "correction_reason": safe_metadata_text(row["correction_reason"], 300, allow_empty=True), "superseded_by_correction": int(row["id"]) in corrected_origins, "plan_consumed": plan_consumed, "administration_preview_revision": medication_public_context_token( identity_key, prescription, row, None, planned_consumed=plan_consumed, ) if prescription is not None and bucket == "planned" and not plan_consumed else None, "correction_preview_revision": medication_public_context_token(identity_key, prescription, None, row) if prescription is not None and int(row["id"]) not in corrected_origins else None, } if len(empty[bucket]) >= RECORD_MAX_ROWS: truncated_sections.add(bucket) continue empty[bucket].append(item) if source: sources.add(source) for bucket in ("administered", "missed", "corrected", "unknown"): empty[bucket].sort(key=lambda item: (item["date"], item["id"]), reverse=True) empty["planned"].sort(key=lambda item: (item["date"], item["id"])) empty["sources"] = sorted(sources) empty["truncated"] = bool(truncated_sections) empty["truncated_sections"] = sorted(truncated_sections) return empty +def _effective_medication_projection(record: dict[str, Any]) -> dict[str, Any]: + """Project only terminal correction revisions into their effective clinical state.""" + projected = { + key: value for key, value in record.items() + if key not in {"planned", "administered", "missed", "corrected", "unknown"} + } + for key in ("planned", "administered", "missed", "corrected", "unknown"): + projected[key] = [] + for source_bucket in ("planned", "administered", "missed", "corrected", "unknown"): + for item in record.get(source_bucket, []): + if item.get("superseded_by_correction"): + continue + if source_bucket == "corrected": + target_bucket = item.get("effective_status") + if target_bucket not in {"planned", "administered", "missed"}: + target_bucket = "corrected" + else: + target_bucket = source_bucket + if target_bucket not in {"planned", "administered", "missed", "corrected"}: + target_bucket = "unknown" + projected[target_bucket].append(item) + return projected + + def _record_appointments( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) order = params.get("order", "desc") if order not in {"asc", "desc"}: raise APIError(400, "sort_not_allowed") institution_filter = ( safe_metadata_text(params["institution"], 120) if params.get("institution") else None ) filters = ["1=1"] values: list[Any] = [] if start: filters.append("datum>=?") values.append(start.isoformat()) if end: filters.append("datum<=?") values.append(end.isoformat()) if institution_filter: filters.append("klinik=?") values.append(institution_filter) rows = ( connection.execute( "SELECT datum,arzt,klinik,grund,notizen FROM arztbesuche WHERE " + " AND ".join(filters) + f" ORDER BY datum {order.upper()},id {order.upper()} LIMIT ?", (*values, RECORD_MAX_ROWS + 1), ).fetchall() if table_exists(connection, "arztbesuche") else [] ) items = [] for row in rows: day = parse_day(row["datum"]) institution = safe_metadata_text(row["klinik"], 120, allow_empty=True) if ( not day or (start and day < start.isoformat()) or (end and day > end.isoformat()) or (institution_filter and institution != institution_filter) ): continue practitioner = safe_metadata_text(row["arzt"], 120, allow_empty=True) reason = safe_metadata_text(row["grund"], 160, allow_empty=True) note = safe_metadata_text(row["notizen"], 300, allow_empty=True) if None not in (institution, practitioner, reason, note): items.append( { "date": day, "practitioner": practitioner, "institution": institution, "reason": reason, "note": note, "status": "documented_visit", } ) return { "items": items[:RECORD_MAX_ROWS], @@ -3877,121 +3886,121 @@ def _record_labs( return { "reference_policy": "observation_specific_verified_original", "today": today.isoformat(), "histories": visible_histories, "observations": result, "truncated": False, "completeness": "complete_page", } def _next_planned_medications( connection: sqlite3.Connection, today: date, *, limit: int = 3 ) -> list[dict[str, Any]]: if not table_exists(connection, "medication_administrations"): return [] rows = connection.execute( """SELECT p.id,p.datum,p.medication_name,p.dose,p.route,p.scheduled_next_date,p.notes,p.source, COALESCE(NULLIF(p.scheduled_next_date,''),NULLIF(substr(p.occurred_at,1,10),''),p.datum) AS effective_plan_date FROM medication_administrations p WHERE lower(trim(COALESCE(p.event_type,''))) IN ('planned','scheduled','geplant') AND COALESCE(NULLIF(p.scheduled_next_date,''),NULLIF(substr(p.occurred_at,1,10),''),p.datum)>? ORDER BY effective_plan_date ASC,p.id ASC""", (today.isoformat(),), ) result = [] for row in rows: if medication_plan_is_consumed(connection, int(row["id"])): continue recorded = parse_day(row["datum"]) effective = parse_day(row["effective_plan_date"]) explicit_schedule = parse_day(row["scheduled_next_date"]) name = safe_metadata_text(row["medication_name"], 120) if recorded and effective and name: result.append( { "date": effective, "recorded_date": recorded, "scheduled_next_date": explicit_schedule, "name": name, "dose": safe_metadata_text(row["dose"], 40, allow_empty=True), "route": safe_metadata_text(row["route"], 40, allow_empty=True), "note": safe_metadata_text(row["notes"], 300, allow_empty=True), "source": safe_metadata_text(row["source"], 80, allow_empty=True), } ) if len(result) >= limit: break return result def _record_summary( connection: sqlite3.Connection, *, reviewed_documents_only: bool = False ) -> dict[str, Any]: document_params = {"sort": "document_date_desc", "limit": "3"} if reviewed_documents_only: document_params["review_status"] = "geprueft" document_page = _record_document_rows( connection, document_params ) today = local_today() - medications = _record_medications(connection, {}) + medications = _effective_medication_projection(_record_medications(connection, {})) next_planned = _next_planned_medications(connection, today) appointments = _record_appointments(connection, {"order": "desc"}) labs = _record_labs(connection, {}) not_reviewed = 0 if table_exists(connection, "dokumente"): not_reviewed = int( connection.execute( "SELECT COUNT(*) FROM dokumente WHERE review_status<>'geprueft'" ).fetchone()[0] ) latest_labs: list[dict[str, Any]] = [] seen_lab_parameters: set[tuple[str, str]] = set() for observation in labs["observations"]: key = (str(observation.get("parameter")), str(observation.get("unit"))) if key not in seen_lab_parameters: seen_lab_parameters.add(key) latest_labs.append(observation) return { "labs": latest_labs[:3], "labs_truncated": labs["truncated"], "latest_administered": medications["administered"][:3], "next_planned": next_planned, "appointments": appointments["items"][:3], "documents": document_page["documents"], "documents_not_reviewed": not_reviewed, } def _capture_plans(connection: sqlite3.Connection) -> dict[str, Any]: medications: list[dict[str, Any]] = [] if table_exists(connection, "medication_administrations"): seen: set[str] = set() for row in connection.execute( "SELECT medication_name,dose,route FROM medication_administrations WHERE trim(COALESCE(medication_name,''))<>'' ORDER BY id DESC LIMIT 500" ): name = str(row["medication_name"]) if name in seen: continue seen.add(name) medications.append({"name": name, "plan_value": row["dose"], "route": row["route"], "source": "existing_plan"}) supplements: list[dict[str, Any]] = [] if table_exists(connection, "supplement_plans"): columns = {str(row[1]) for row in connection.execute("PRAGMA table_info(supplement_plans)")} if {"product", "amount", "unit"}.issubset(columns): status_clause = "AND status='active'" if "status" in columns else "" for row in connection.execute( f"SELECT product,amount,unit FROM supplement_plans WHERE trim(COALESCE(product,''))<>'' {status_clause} ORDER BY id DESC LIMIT 200" ): supplements.append({"name": str(row["product"]), "plan_value": row["amount"], "unit": row["unit"], "source": "existing_plan"}) return {"medications": medications, "supplements": supplements, "dose_semantics": "documented_plan_value_not_actual"} def _capture_timeline(connection: sqlite3.Connection, day: str) -> dict[str, Any]: try: date.fromisoformat(day) except ValueError as exc: raise APIError(400, "invalid_date") from exc if not table_exists(connection, "capture_entries"): return { "date": day, "timed": [], __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/repos/HealthManager__HERMES_CWD_8d46a20096ed__