"""Read-only, allowlisted data provider for Dashboard v5.""" from __future__ import annotations import json import re import sqlite3 from collections import defaultdict from datetime import date, datetime, timedelta, timezone from pathlib import Path from statistics import median from typing import Any from apple_health_analytics import daily_series as apple_daily from dashboard_v5.contracts import SCHEMA_VERSION from dashboard_v5.lab_registry import EXACT_LAB_NUMBER, LAB_ALLOWLIST from dashboard_v5.metric_registry import BY_ID, public_explorer, public_registry from dashboard_v5.nutrition_mapping_review import build_review as build_mapping_review TZ = "Europe/Zurich" SYMPTOM_LABELS = { "Aphthen/Mundulzera", "GI/Darm", "Müdigkeit/Fatigue", "Haut", "Augen", "Gelenke", "Vaskulär/Thrombose-Warnzeichen", } APPLE_SPECS = { "apple.sleep": ("sleep_analysis", "sum"), "apple.resting_heart_rate": ("resting_heart_rate", "avg"), "apple.hrv": ("heart_rate_variability", "avg"), "apple.steps": ("step_count", "sum"), } ADMINISTERED = { "administered", "verabreicht", "erste injektion / therapiebeginn", "planmäßige injektion", "planmässige injektion", "planmaessige injektion", } EXPLICIT_PLANS = {"planned", "scheduled", "geplant"} EXCLUDED_PLANS = {"cancelled", "canceled", "abgesagt", "missed", "verpasst"} PUBLIC_TEXT = re.compile(r"^[^\x00-\x1f\x7f]{1,160}$") IDENTIFIER = re.compile(r"^[a-z0-9_:.-]+$") def public_text(value: Any, maximum: int = 120, *, allow_empty: bool = False) -> str | None: text = " ".join(str(value or "").split()) if (not text and not allow_empty) or len(text) > maximum or not PUBLIC_TEXT.fullmatch(text or " "): return None if re.search(r"^(?:/|~/|[A-Za-z]:[\\/]|file:|https?://)|(?:^|/)\.hermes(?:/|$)", text, re.I): return None return text def connect(path: Path) -> sqlite3.Connection: connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True) connection.row_factory = sqlite3.Row return connection 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 slug(value: Any) -> str: normalized = str(value or "").casefold().translate( str.maketrans({"ä": "ae", "ö": "oe", "ü": "ue", "ß": "ss"}) ) return re.sub(r"[^a-z0-9]+", "_", normalized).strip("_") def unit_slug(value: Any) -> str: return slug(str(value or "").replace("µ", "u").replace("μ", "u").replace("%", " percent ")) def parse_day(value: Any) -> str | None: text = str(value or "").strip() if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", text): return None try: return date.fromisoformat(text).isoformat() except ValueError: return None def parse_score(value: Any) -> int | None: match = re.search(r"(?:^|\()([0-3])\)?\s*$", str(value or "").strip()) return int(match.group(1)) if match else None def rolling_baseline(points: list[dict[str, Any]], days: int, minimum: int = 7) -> list[dict[str, Any]]: observed = {date.fromisoformat(p["date"]): float(p["value"]) for p in points if p["value"] is not None} result: list[dict[str, Any]] = [] for point in points: current = date.fromisoformat(point["date"]) start = current - timedelta(days=days) values = [value for day, value in observed.items() if start <= day < current] result.append({"date": point["date"], "value": round(median(values), 3) if len(values) >= minimum else None}) 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}, [] 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 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 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 = 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) for row in rows: raw_key = (slug(row["parameter_name"]), unit_slug(row["einheit"])) canonical = LAB_ALLOWLIST.get(raw_key) day = parse_day(row["abnahme_datum"]) or parse_day(row["befund_datum"]) if canonical and day and (through_day is None or day <= through_day): canonical_parameter, canonical_unit = canonical candidates[(canonical_parameter, canonical_unit, day)].append(row) result = [] for (canonical_parameter, canonical_unit, day), same_day in candidates.items(): if len(same_day) != 1: continue row = same_day[0] if not EXACT_LAB_NUMBER.fullmatch(str(row["wert"] or "").strip()): continue result.append({ "parameter": canonical_parameter, "value": str(row["wert"]).strip(), "unit": canonical_unit, "date": day, "reference_min": row["reference_min"], "reference_max": row["reference_max"], "quality": "verified_original", "source_type": "scanned_original", }) result.sort(key=lambda item: item["date"], reverse=True) return result[:limit] def nutrition_data(connection: sqlite3.Connection) -> dict[str, Any]: daily: list[dict[str, Any]] = [] if table_exists(connection, "nutrition_daily_summary_v2"): for row in connection.execute("SELECT * FROM nutrition_daily_summary_v2 ORDER BY datum DESC LIMIT 180"): day = parse_day(row["datum"]) complete = ( day is not None and row["item_count"] is not None and int(row["item_count"]) > 0 and row["histamine_unknown_count"] is not None and int(row["histamine_unknown_count"]) == 0 and row["histamine_score"] is not None and row["histamine_label"] in {"classified", "green", "yellow", "orange", "red"} ) if day: daily.append({ "date": day, "kcal": row["kcal"], "item_count": row["item_count"], "histamine_score": row["histamine_score"] if complete else None, "histamine_status": "classified" if complete else "unknown", "unknown_items": row["histamine_unknown_count"], "quality": "complete" if complete else "incomplete", }) mapping = build_mapping_review(connection) state_order = { "suggestion_available": 0, "unassigned": 1, "review_required": 2, "conflict": 3, "deferred": 4, "mapped_unverified": 5, "confirmed": 6, "not_assignable": 7, "irrelevant": 8, } mapping_groups = sorted( mapping["groups"], key=lambda group: ( state_order.get(str(group["status"]), 99), -int(group["entry_count"]), str(group["name"]).casefold(), ), ) return { "daily": list(reversed(daily)), "mapping_review": mapping_groups, "mapping_summary": mapping["summary"], "mapping_catalog": mapping["catalog"], "mapping_contract_version": mapping["contract_version"], "mapping_open_count": mapping["summary"]["open_entries"], } def correlations(connection: sqlite3.Connection) -> dict[str, Any]: dimensions = { "lag_days": [0, 1, 2, 3], "medication_phases": ["baseline_pre_treatment", "early_treatment", "stable_treatment", "unknown"], "lag_direction": "predictor_d_to_target_d_plus_lag", } if not table_exists(connection, "multimodal_correlation_results"): return {"dimensions": dimensions, "results": []} import multimodal_correlations as engine results = [] rows = connection.execute( """SELECT predictor,target,lag_days,medication_phase,n,eligible_target_days, expected_target_days,target_coverage,missing_pairs,rho,p_value,q_value, status,method,quality_flags,interpretation FROM multimodal_correlation_results ORDER BY predictor,target,lag_days,medication_phase LIMIT 1000""" ) for row in rows: try: flags_raw = json.loads(str(row["quality_flags"])) if not isinstance(flags_raw, list) or not all(isinstance(flag, str) for flag in flags_raw): continue result = engine.CorrelationResult( predictor=str(row["predictor"]), target=str(row["target"]), lag_days=int(row["lag_days"]), medication_phase=str(row["medication_phase"]), n=int(row["n"]), eligible_target_days=int(row["eligible_target_days"]), expected_target_days=int(row["expected_target_days"]), target_coverage=float(row["target_coverage"]), missing_pairs=int(row["missing_pairs"]), rho=row["rho"], p_value=row["p_value"], q_value=row["q_value"], status=str(row["status"]), method=str(row["method"]), quality_flags=tuple(flags_raw), interpretation=str(row["interpretation"]), ) engine._validate_result_for_storage(result) except (TypeError, ValueError, json.JSONDecodeError): continue results.append({ "id": {"predictor": result.predictor, "target": result.target}, "dimensions": {"lag_days": result.lag_days, "medication_phase": result.medication_phase}, "sample": { "eligible_target_days": result.eligible_target_days, "expected_target_days": result.expected_target_days, "observed_pairs": result.n, "missing_pairs": result.missing_pairs, "target_coverage": result.target_coverage, }, "statistics": {"rho": result.rho, "p_value": result.p_value, "q_value": result.q_value}, "status": result.status, "method": result.method, "quality_flags": list(result.quality_flags), }) return {"dimensions": dimensions, "results": results} def build_bundle(connection: sqlite3.Connection, *, today: str, generated_at: str | None = None) -> dict[str, Any]: current = parse_day(today) if not current: raise ValueError("today must be a complete ISO date") symptoms, complete_days = symptom_series(connection) series: dict[str, list[dict[str, Any]]] = { "symptom.total": symptoms, "symptom.aphthae": symptom_dimension_series( connection, "Aphthen/Mundulzera", complete_days ), } for metric_id, (metric, mode) in APPLE_SPECS.items(): series[metric_id] = apple_series(connection, metric, mode) nutrition = nutrition_data(connection) nutrition["daily"] = [row for row in nutrition["daily"] if row["date"] <= current] series["nutrition.histamine"] = [ { "date": row["date"], "value": row["histamine_score"], "quality": row["quality"], "mapping_coverage": ( None if not row["item_count"] else round( (int(row["item_count"]) - int(row["unknown_items"] or 0)) / int(row["item_count"]), 4, ) ), "unknown_items": int(row["unknown_items"] or 0), "item_count": int(row["item_count"] or 0), } for row in nutrition["daily"] ] series = { metric_id: [point for point in points if point["date"] <= current] for metric_id, points in series.items() } for metric_id, points in series.items(): metric = BY_ID[metric_id] baseline = rolling_baseline( points, metric.baseline_days, minimum=metric.baseline_min_observations, ) by_day = {entry["date"]: entry["value"] for entry in baseline} for point in points: point["baseline"] = by_day.get(point["date"]) medication, medication_events = medication_data(connection, current) tasks = [] data_preparation = [] if current not in complete_days: tasks.append({"id": "symptom-checkin", "label": "Symptome vollständig erfassen", "action": "checkin"}) if nutrition["mapping_open_count"]: tasks.append({"id": "food-mapping", "label": "Lebensmittel-Zuordnungen prüfen", "count": nutrition["mapping_open_count"], "action": "nutrition"}) if table_exists(connection, "dokumente"): documents_pending = int(connection.execute("SELECT COUNT(*) FROM dokumente WHERE review_status='nicht_geprueft'").fetchone()[0]) if table_exists(connection,"document_reconciliation"): decisions=int(connection.execute("SELECT COALESCE(SUM(open_decisions),0) FROM document_reconciliation WHERE queue_bucket='now_reviewable'").fetchone()[0]) if decisions: tasks.insert(0,{"id":"document-guided-review","label":"Dokumente prüfen","count":decisions,"action":"document-review","queue_code":"now_reviewable"}) prepared=int(connection.execute("SELECT COUNT(*) FROM document_reconciliation WHERE queue_bucket IN ('automatically_prepared','original_missing','technical_blocked')").fetchone()[0]) if prepared:data_preparation.append({"id":"document-processing","label":"Dokumente technisch vorbereitet","count":prepared,"status":"system_work"}) elif documents_pending: tasks.append({"id":"document-content-pending","label":"Dokumentinhalt prüfen","count":documents_pending,"action":"documents"}) data_preparation.append({"id":"document-processing","label":"Dokument-OCR, Hash, Dubletten und FTS","count":documents_pending,"status":"system_work"}) return { "schema_version": SCHEMA_VERSION, "generated_at": generated_at or datetime.now(timezone.utc).isoformat(), "timezone": TZ, "today": current, "metrics": public_registry(), "explorer": public_explorer(), "series": series, "events": medication_events, "medication": medication, "labs": verified_labs(connection, through_day=current), "nutrition": nutrition, "tasks": tasks[:3], "data_preparation": data_preparation, "notices": [], "freshness": [], "correlations": correlations(connection), }