diff --git a/scripts/health/dashboard_v5/read_api.py b/scripts/health/dashboard_v5/read_api.py index 94af38b..62ef74c 100644 --- a/scripts/health/dashboard_v5/read_api.py +++ b/scripts/health/dashboard_v5/read_api.py @@ -42,102 +42,113 @@ from dashboard_v5.lab_review import QUALITATIVE as LAB_QUALITATIVE, REVIEW_STATE 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, + 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", "supplement", "symptom_day", "health_event", "health_period", "nutrition_day", "laboratory", "document", "appointment"} + {"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, 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, ), ) ) 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} ) + 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" + ) + 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}) 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: @@ -3442,145 +3475,267 @@ def _document_review_queue(connection: sqlite3.Connection) -> dict[str, Any]: 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] ) -> dict[str, Any]: start, end = parse_range(params) - result: dict[str, Any] = { + 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"): - return result - groups = { - "planned": ("planned", "scheduled", "geplant"), - "administered": tuple(sorted(ADMINISTERED)), - "missed": ("missed", "verpasst", "ausgelassen"), - "corrected": ("corrected", "korrigiert", "correction"), + 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, + "status_provenance": provenance, + "preview_revision": medication_public_context_token(identity_key, row), + } + 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", } - truncated_sections: list[str] = [] - for bucket, events in groups.items(): - effective = ( - "COALESCE(NULLIF(scheduled_next_date,''),datum)" + 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_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") + 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 "datum" + else False ) - placeholders = ",".join("?" for _ in events) - filters = [f"lower(trim(COALESCE(event_type,''))) IN ({placeholders})"] - values: list[Any] = list(events) - if start: - filters.append(f"{effective}>=?") - values.append(start.isoformat()) - if end: - filters.append(f"{effective}<=?") - values.append(end.isoformat()) - direction = "ASC" if bucket == "planned" else "DESC" - rows = connection.execute( - "SELECT datum,medication_name,dose,route,event_type," - "scheduled_next_date,notes,source FROM medication_administrations WHERE " - + " AND ".join(filters) - + f" ORDER BY {effective} {direction},id {direction} LIMIT ?", - (*values, RECORD_MAX_ROWS + 1), - ).fetchall() - if len(rows) > RECORD_MAX_ROWS: - truncated_sections.append(bucket) - for row in rows[:RECORD_MAX_ROWS]: - 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 - ) - name = safe_metadata_text(row["medication_name"], 120) - if not recorded_day or not effective_day or not name: - continue - result[bucket].append( - { - "date": effective_day, - "recorded_date": recorded_day, - "scheduled_next_date": scheduled if bucket == "planned" else None, - "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), - } - ) - result["truncated"] = bool(truncated_sections) - result["truncated_sections"] = truncated_sections - result["next_cursor"] = None - return result + 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), + "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 _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 ( @@ -3687,105 +3842,111 @@ def _record_labs( item = { **source, "history": histories[source["metric_id"]], "previous": prior, "absolute_change": delta, "percent_change": round(delta / float(prior) * 100, 4) if delta is not None and float(prior) != 0 else None, "document_id": source.get("document", {}).get("id") if source.get("document") else None, } result.append(item) result.sort(key=lambda item: (item["date"], item["id"]), reverse=True) visible_histories = [ history for history in histories.values() if not canonical_query or canonical_query.casefold() in str(next(item["parameter"] for item in grouped_rows[history["metric_id"]])).casefold() ] visible_histories.sort(key=lambda item: item["metric_id"]) if len(result) > MAX_SERIES_ROWS: raise APIError(422, "row_limit_exceeded") 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 datum,medication_name,dose,route,scheduled_next_date,notes,source - FROM medication_administrations - WHERE lower(trim(COALESCE(event_type,''))) IN ('planned','scheduled','geplant') - AND scheduled_next_date>? - ORDER BY scheduled_next_date ASC,id ASC LIMIT ?""", - (today.isoformat(), limit), + """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"]) - scheduled = parse_day(row["scheduled_next_date"]) + 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 scheduled and name: + if recorded and effective and name: result.append( { - "date": scheduled, + "date": effective, "recorded_date": recorded, - "scheduled_next_date": scheduled, + "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, {}) 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, } @@ -4011,87 +4172,104 @@ def _additional_symptoms(connection: sqlite3.Connection, start: date, end: date) f"""SELECT datum,symptom,schwergrad,notizen,{occurred} AS occurred_at FROM symptom_log WHERE kontext='additional_symptom' AND datum>=? AND datum<=? ORDER BY datum,id LIMIT 500""", (start.isoformat(), end.isoformat()), ) result = [] for row in rows: day = parse_day(row["datum"]) symptom = safe_metadata_text(row["symptom"], 80, allow_empty=False) note = safe_metadata_text(row["notizen"], 300, allow_empty=True) severity = parse_score(row["schwergrad"]) if day and symptom and note is not None: result.append({"date": day, "occurred_at": safe_metadata_text(row["occurred_at"], 32, allow_empty=True), "symptom": symptom, "severity": severity, "note": note}) return result REPORT_SECTIONS = frozenset( {"overview", "labs", "medications", "supplements", "symptoms", "appointments", "documents", "nutrition", "observations", "timeline", "personal_observations"} ) def _doctor_report( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params, required=True) assert start and end selected = set(filter(None, params.get("sections", "").split(","))) or set( REPORT_SECTIONS ) if not selected <= REPORT_SECTIONS: raise APIError(400, "section_not_allowed") lab_page = ( _record_labs( connection, {"from": start.isoformat(), "to": end.isoformat()}, ) if "labs" in selected else {"observations": [], "truncated": False} ) labs = lab_page["observations"] - medications = ( + medication_record = ( _record_medications( connection, {"from": start.isoformat(), "to": end.isoformat()} ) if "medications" in selected - else {"planned": [], "administered": [], "missed": [], "corrected": []} + else {"planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "truncated": False} + ) + medication_report_fields = ( + "date", "name", "status", "effective_status", "scheduled_next_date", "planned_dose_value", + "planned_dose_unit", "actual_dose_value", "actual_dose_unit", "legacy_dose", + "superseded_by_correction", ) + medications: dict[str, Any] = { + kind: [ + { + **{field: item.get(field) for field in medication_report_fields}, + "is_correction": kind == "corrected", + "is_latest_effective": not bool(item.get("superseded_by_correction")), + } + for item in medication_record.get(kind, []) + ] + for kind in ("planned", "administered", "missed", "corrected", "unknown") + } + medications["truncated"] = bool(medication_record.get("truncated")) supplement_report = ( supplements(connection, start, end) if "supplements" in selected else {"planned": [], "administered": [], "missed": [], "corrected": []} ) appointments = ( _record_appointments( connection, {"from": start.isoformat(), "to": end.isoformat(), "order": "asc"}, )["items"] if "appointments" in selected else [] ) document_page = ( _record_document_rows( connection, { "from": start.isoformat(), "to": end.isoformat(), "sort": "document_date_asc", "limit": str(RECORD_MAX_PAGE_SIZE), "review_status": "geprueft", }, ) if "documents" in selected else {"documents": [], "truncated": False} ) documents = document_page["documents"] symptoms: list[dict[str, Any]] = [] if "symptoms" in selected: current = start while current <= end: day = current.isoformat() symptom = _day_symptoms(connection, day) events = _day_events(connection, day) if ( symptom["documented_dimensions"] or events["events"] or events["periods"] ): @@ -4198,81 +4376,81 @@ def _doctor_report( "overview": overview, "labs": labs, "medications": medications, "supplements": supplement_report, "symptoms": symptoms, "additional_symptoms": additional_symptoms, "appointments": appointments, "documents": documents, "nutrition": nutrition.get("days", []), "nutrition_summary": nutrition.get("summary", {}), "nutrition_factors": nutrition.get("nutrition_factors", []), "nutrition_factor_statement": nutrition.get("nutrition_factor_statement", ""), "observations": observations, "personal_observations": personal_observations, "timeline": timeline, "temporal_observations": temporal_observations, "nutrition_proximity": nutrition_proximity, "created_at": datetime.now(ZoneInfo(TZ_NAME)).isoformat(timespec="seconds"), "completeness": { name: { "status": "documented" if section_payload else "unknown", "truncated": { "overview": False, "labs": bool(lab_page.get("truncated")), "medications": bool(medications.get("truncated")), "supplements": bool(supplement_report.get("truncated")), "symptoms": False, "appointments": False, "documents": bool(document_page.get("truncated")), "nutrition": False, "observations": False, "personal_observations": False, "timeline": False, }[name], } for name, section_payload in { "overview": overview, "labs": labs, "medications": sum( len(medications.get(kind, [])) - for kind in ("planned", "administered", "missed", "corrected") + for kind in ("planned", "administered", "missed", "corrected", "unknown") ), "supplements": sum( len(supplement_report.get(kind, [])) for kind in ("planned", "administered", "missed", "corrected") ), "symptoms": symptoms, "appointments": appointments, "documents": documents, "nutrition": nutrition.get("days", []), "observations": observations, "personal_observations": personal_observations, "timeline": timeline, }.items() if name in selected }, "medical_statement": "Dokumentierte Daten ohne Diagnose-, Kausalitäts- oder Therapieaussage.", } return payload def dispatch_api( database: Path, path: str, query: str, runtime_artifact: Path | None = None, ) -> dict[str, Any]: capture_timeline_route = re.fullmatch( r"/api/v1/day/(\d{4}-\d{2}-\d{2})/timeline", path ) observation_route = re.fullmatch( r"/api/v1/observations/(obs_[a-f0-9]{24})(?:/(analysis|results))?", path ) observation_result_route = re.fullmatch( r"/api/v1/observations/(obs_[a-f0-9]{24})/results/(result_[a-f0-9]{24})", path ) document_route = re.fullmatch( r"/api/v1/documents/(api-document-[a-f0-9]{24})(?:/(matches|extracted-preview|review|candidates|compare))?", path, ) allowed_record = { @@ -4399,81 +4577,81 @@ def dispatch_api( today=local_today(), ) return evaluate_observation(connection, observation_id, observation_series) detail = observation_detail(connection, observation_id) if suffix == "results": return { "version": 1, "observation_id": observation_id, "items": detail["results"], } return detail except KeyError as error: raise APIError(404, "observation_not_found") from error if path == "/api/v1/record-summary": parse_query(query, set()) return _record_summary(connection) if path == "/api/v1/record-labs": return _record_labs(connection, parse_query(query, {"q", "from", "to"})) if path == "/api/v1/lab-review": params = parse_query(query, {"status", "q", "from", "to", "source"}) requested_status = params.get("status", "open") or "open" if requested_status not in set(LAB_REVIEW_STATES) | {"open", "all"}: raise APIError(422, "lab_review_status_not_allowed") if len(params.get("q", "")) > 80: raise APIError(422, "lab_review_query_too_long") for name in ("from", "to"): if params.get(name) and parse_day(params[name]) is None: raise APIError(422, "invalid_date") if params.get("source") and not re.fullmatch(r"api-document-[a-f0-9]{24}", params["source"]): raise APIError(422, "invalid_document_id") try: return build_lab_review( connection, params, verified_rows=_verified_lab_rows(connection, include_missing_reference=True), document_id_factory=api_document_id, ) except ValueError as error: raise APIError(422, str(error)) from error if path == "/api/v1/medications": - return _record_medications(connection, parse_query(query, {"from", "to"})) + return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) if path == "/api/v1/appointments": return _record_appointments( connection, parse_query(query, {"from", "to", "order", "institution"}) ) if path == "/api/v1/document-review-queue": parse_query(query, set()) return _document_review_queue(connection) if path == "/api/v1/documents": return _record_document_rows( connection, parse_query( query, { "from", "to", "category", "institution", "type", "review_status", "queue", "sort", "q", "cursor", "limit", }, ), ) if document_route: opaque, suffix = document_route.groups() if suffix == "matches": return _record_document_matches( connection, opaque, parse_query(query, {"q"}) ) if suffix == "extracted-preview": return _record_document_extracted_preview( connection, opaque, parse_query(query, {"cursor", "limit"}), ) if suffix in {"review", "candidates"}: __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/repos/HealthManager__HERMES_CWD_8d46a20096ed__