"""Bounded read-only data API provider for Dashboard v5 Sprint 6B."""

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import math
import os
import re
import secrets
import sqlite3
import time
from statistics import median
from collections import defaultdict
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs
from zoneinfo import ZoneInfo

from apple_health_analytics import daily_points
from dashboard_v5.contracts import nutrition_queue_key
from dashboard_v5.data_provider import (
    ADMINISTERED,
    SYMPTOM_LABELS,
    parse_day,
    parse_score,
    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

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"}
)
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"(?<![A-Za-z0-9_-])[A-Za-z0-9_-]{25,}(?![A-Za-z0-9_-])")
API_PATHS = frozenset(
    {
        "/api/v1/metric-catalog",
        "/api/v1/series",
        "/api/v1/events",
        "/api/v1/labs",
        "/api/v1/calendar",
        "/api/v1/search",
        "/api/v1/nutrition/days",
        "/api/v1/nutrition/mapping-queue",
        "/api/v1/supplements",
        "/api/v1/associations/catalog",
        "/api/v1/associations",
        "/api/v1/explorer-comparison/catalog",
        "/api/v1/explorer-comparison",
        "/api/v1/observations",
        "/api/v1/observations/templates",
        "/api/v1/observations/active-today",
        "/api/v1/source-status",
    }
)
EXPECTED_ANALYTICS_UNITS = APPLE_ANALYTICS_UNITS


@dataclass
class APIError(Exception):
    status: int
    code: str

    def __str__(self) -> str:
        return self.code


def safe_metadata_text(
    value: Any, maximum: int, *, allow_empty: bool = False
) -> str | None:
    text = public_text(value, maximum, allow_empty=allow_empty)
    drive_id_like = bool(
        text
        and any(
            any(char.isalpha() for char in token)
            and any(char.isdigit() for char in token)
            for token in DRIVE_ID_TOKEN.findall(text)
        )
    )
    if text is None or FORBIDDEN_METADATA_TEXT.search(text) or drive_id_like:
        return None
    return text


def api_document_id(
    raw_id: Any, document_day: str | None, category: str, institution: str
) -> str:
    material = json.dumps(
        [int(raw_id), document_day, category, institution],
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    return f"api-document-{hashlib.sha256(b'health-api-document-v1:' + material).hexdigest()[:24]}"


def api_observation_id(prefix: str, raw_id: Any, *parts: Any) -> str:
    material = json.dumps(
        [int(raw_id), *parts],
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    return f"{prefix}-{hashlib.sha256(b'health-api-observation-v1:' + material).hexdigest()[:24]}"


def connect_read_only(database: Path) -> sqlite3.Connection:
    connection = sqlite3.connect(f"{database.resolve().as_uri()}?mode=ro", uri=True)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA query_only=ON")
    connection.execute("PRAGMA trusted_schema=OFF")
    deadline = time.monotonic() + QUERY_TIMEOUT_SECONDS
    connection.set_progress_handler(lambda: int(time.monotonic() > deadline), 1000)
    return connection


def table_exists(connection: sqlite3.Connection, table: str) -> bool:
    return (
        connection.execute(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)
        ).fetchone()
        is not None
    )


def parse_query(query: str, allowed: set[str]) -> dict[str, str]:
    if len(query.encode("utf-8")) > MAX_QUERY_BYTES:
        raise APIError(414, "query_too_large")
    try:
        parsed = (
            parse_qs(
                query,
                keep_blank_values=True,
                strict_parsing=True,
                max_num_fields=max(1, len(allowed) + 1),
            )
            if query
            else {}
        )
    except ValueError as error:
        raise APIError(400, "invalid_query") from error
    if not set(parsed) <= allowed:
        raise APIError(400, "unknown_parameter")
    if any(len(values) != 1 for values in parsed.values()):
        raise APIError(400, "duplicate_parameter")
    return {key: values[0] for key, values in parsed.items()}


def parse_iso_day(raw: str, *, code: str = "invalid_date") -> date:
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", raw):
        raise APIError(400, code)
    try:
        return date.fromisoformat(raw)
    except ValueError as error:
        raise APIError(400, code) from error


def local_today() -> date:
    synthetic_day = os.environ.get("HEALTH_DASHBOARD_TEST_TODAY", "").strip()
    if synthetic_day and os.environ.get("HEALTH_DASHBOARD_TEST_INSTANCE_ID", "").strip():
        try:
            return date.fromisoformat(synthetic_day)
        except ValueError as error:
            raise APIError(503, "invalid_test_today") from error
    return datetime.now(LOCAL_TZ).date()


def parse_range(
    params: dict[str, str], *, required: bool = False
) -> tuple[date | None, date | None]:
    raw_start = params.get("from")
    raw_end = params.get("to")
    if bool(raw_start) != bool(raw_end):
        raise APIError(400, "incomplete_range")
    if required and not raw_start:
        raise APIError(400, "range_required")
    if not raw_start:
        return None, None
    start = parse_iso_day(raw_start)
    end = parse_iso_day(raw_end or "")
    if end < start:
        raise APIError(400, "range_reversed")
    if start > local_today() or end > local_today():
        raise APIError(422, "future_range_not_allowed")
    if (end - start).days + 1 > MAX_RANGE_DAYS:
        raise APIError(422, "range_too_large")
    return start, end


def _apple_rows(
    connection: sqlite3.Connection,
    identifier: str,
    start: date | None,
    end: date | None,
    *,
    include_raw_json: bool = False,
) -> list[sqlite3.Row]:
    if include_raw_json:
        sql = """SELECT id,metric,value,unit,start_date,end_date,source_name,
                        file_name,file_hash,raw_json
                 FROM apple_health_records WHERE metric=? AND start_date IS NOT NULL"""
    else:
        sql = """SELECT id,metric,value,unit,start_date,end_date,source_name,file_name,file_hash
                 FROM apple_health_records WHERE metric=? AND start_date IS NOT NULL"""
    parameters: list[Any] = [identifier]
    # Keep the (metric,start_date) index usable with a deliberately coarse raw
    # candidate interval.  It includes one whole calendar day on either side;
    # the upper edge is exclusive so all lexical ISO values on end + 1 are kept.
    # daily_points remains the sole authority for Europe/Zurich assignment and
    # exact start/end filtering—this SQL clause never assigns a health day.
    if start is not None and start > date.min:
        sql += " AND start_date>=?"
        parameters.append((start - timedelta(days=1)).isoformat())
    # date.max cannot represent the exclusive edge after end + 1.  Omitting
    # that impossible upper bound preserves the exact canonical filter while
    # avoiding an arithmetic overflow at the ISO boundary.
    if end is not None and end < date.max - timedelta(days=1):
        sql += " AND start_date<?"
        parameters.append((end + timedelta(days=2)).isoformat())
    sql += " ORDER BY start_date,end_date,source_name,id LIMIT ?"
    parameters.append(MAX_RAW_APPLE_ROWS + 1)
    rows = list(connection.execute(sql, parameters))
    if len(rows) > MAX_RAW_APPLE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    return rows


def _numeric(value: float, precision: int, value_type: str) -> int | float:
    rounded = round(float(value), precision)
    if value_type == "integer" or (precision == 0 and rounded.is_integer()):
        return int(rounded)
    return rounded


def _apple_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    rows = _apple_rows(connection, metric.source_identifier, start, end)
    points = daily_points(
        metric.source_identifier,
        mode=metric.aggregation,
        stable_only=False,
        start_date=start.isoformat() if start else None,
        end_date=end.isoformat() if end else None,
        _rows=rows,
    )
    result = []
    for day, point in sorted(points.items()):
        if point.unit != EXPECTED_ANALYTICS_UNITS.get(metric.source_identifier):
            continue
        result.append(
            {
                "date": day,
                "value": _numeric(point.value, metric.precision, metric.value_type),
                "quality": point.quality,
                "source_class": point.source_class,
            }
        )
    return result


def _sleep_phase_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    phase = metric.id.rsplit(".", 1)[-1]
    transformed: list[dict[str, Any]] = []
    for row in _apple_rows(
        connection, "sleep_analysis", start, end, include_raw_json=True
    ):
        try:
            payload = json.loads(str(row["raw_json"] or ""))
            value = float(payload[phase])
        except (ValueError, TypeError, KeyError, json.JSONDecodeError):
            continue
        if not math.isfinite(value):
            continue
        item = dict(row)
        item["value"] = value
        transformed.append(item)
    points = daily_points(
        "sleep_analysis",
        mode="sum",
        stable_only=False,
        start_date=start.isoformat() if start else None,
        end_date=end.isoformat() if end else None,
        _rows=transformed,  # type: ignore[arg-type] -- dict rows preserve sqlite row keys
    )
    return [
        {
            "date": day,
            "value": _numeric(point.value, metric.precision, metric.value_type),
            "quality": point.quality,
            "source_class": point.source_class,
        }
        for day, point in sorted(points.items())
        if point.unit == "hr"
    ]


def _blood_pressure_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    component = "systolic" if metric.id.endswith(".systolic") else "diastolic"
    by_day: dict[str, tuple[datetime, int, float, str]] = {}
    for row in _apple_rows(
        connection, "blood_pressure", start, end, include_raw_json=True
    ):
        try:
            timestamp = datetime.fromisoformat(
                str(row["start_date"]).replace("Z", "+00:00")
            )
            if timestamp.tzinfo is None:
                timestamp = timestamp.replace(tzinfo=LOCAL_TZ)
            day = timestamp.astimezone(LOCAL_TZ).date()
            payload = json.loads(str(row["raw_json"] or ""))
            unit = re.sub(r"[^a-z]", "", str(row["unit"] or "").casefold())
            if unit != "mmhg":
                continue
            components: dict[str, float] = {}
            for key in ("systolic", "diastolic"):
                try:
                    candidate = float(payload[key])
                except (ValueError, TypeError, KeyError):
                    continue
                if math.isfinite(candidate):
                    components[key] = candidate
            if component not in components:
                continue
            value = components[component]
            pair_status = "paired" if len(components) == 2 else "incomplete"
        except (ValueError, TypeError, json.JSONDecodeError):
            continue
        if not math.isfinite(value):
            continue
        if start is not None and day < start:
            continue
        if end is not None and day > end:
            continue
        existing = by_day.get(day.isoformat())
        candidate = (timestamp, int(pair_status == "paired"), value, pair_status)
        if existing is None or candidate[:2] > existing[:2]:
            by_day[day.isoformat()] = candidate
    return [
        {
            "date": day,
            "value": _numeric(item[2], metric.precision, metric.value_type),
            "quality": "paired" if item[3] == "paired" else "incomplete_pair",
            "pair_status": item[3],
            "source_class": "apple_health",
        }
        for day, item in sorted(by_day.items())
    ]


def _symptom_dimension_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    sql = "SELECT datum,schwergrad FROM symptom_log WHERE kontext='daily_quick_score' AND symptom=?"
    parameters: list[Any] = [metric.source_identifier]
    if start:
        sql += " AND datum>=?"
        parameters.append(start.isoformat())
    if end:
        sql += " AND datum<=?"
        parameters.append(end.isoformat())
    sql += " ORDER BY datum,id LIMIT ?"
    parameters.append(MAX_DAILY_SOURCE_ROWS + 1)
    rows = list(connection.execute(sql, parameters))
    if len(rows) > MAX_DAILY_SOURCE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    grouped: dict[str, list[int | None]] = defaultdict(list)
    for row in rows:
        day = parse_day(row["datum"])
        if day:
            grouped[day].append(parse_score(row["schwergrad"]))
    return [
        {"date": day, "value": scores[0], "quality": "observed"}
        for day, scores in sorted(grouped.items())
        if len(scores) == 1 and scores[0] is not None
    ]


def _symptom_total_points(
    connection: sqlite3.Connection,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    sql = "SELECT datum,symptom,schwergrad FROM symptom_log WHERE kontext='daily_quick_score'"
    parameters: list[Any] = []
    if start:
        sql += " AND datum>=?"
        parameters.append(start.isoformat())
    if end:
        sql += " AND datum<=?"
        parameters.append(end.isoformat())
    sql += " ORDER BY datum,id LIMIT ?"
    parameters.append(MAX_DAILY_SOURCE_ROWS + 1)
    rows = list(connection.execute(sql, parameters))
    if len(rows) > MAX_DAILY_SOURCE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    grouped: dict[str, dict[str, list[int | None]]] = defaultdict(
        lambda: defaultdict(list)
    )
    for row in rows:
        day = parse_day(row["datum"])
        label = str(row["symptom"] or "")
        if day and label in SYMPTOM_LABELS:
            grouped[day][label].append(parse_score(row["schwergrad"]))
    result = []
    for day, dimensions in sorted(grouped.items()):
        if set(dimensions) != SYMPTOM_LABELS:
            continue
        if not all(
            len(scores) == 1 and scores[0] is not None for scores in dimensions.values()
        ):
            continue
        result.append(
            {
                "date": day,
                "value": sum(int(scores[0]) for scores in dimensions.values()),
                "quality": "complete",
            }
        )
    return result


def _nutrition_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    if metric.source_identifier in NUTRIENT_CONTRACTS:
        daily = _daily_nutrients(connection, start, end)
        return [
            {
                "date": day,
                "value": values[metric.source_identifier]["value"],
                "quality": "documented",
            }
            for day, values in sorted(daily.items())
            if metric.source_identifier in values
            and values[metric.source_identifier]["value"] is not None
        ]
    sql = """SELECT datum,histamine_score,histamine_max,item_count,
                    histamine_unknown_count,histamine_label
             FROM nutrition_daily_summary_v2 WHERE item_count>0"""
    parameters: list[Any] = []
    if start:
        sql += " AND datum>=?"
        parameters.append(start.isoformat())
    if end:
        sql += " AND datum<=?"
        parameters.append(end.isoformat())
    sql += " ORDER BY datum LIMIT ?"
    parameters.append(MAX_DAILY_SOURCE_ROWS + 1)
    rows = list(connection.execute(sql, parameters))
    if len(rows) > MAX_DAILY_SOURCE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    result = []
    for row in rows:
        day = parse_day(row["datum"])
        item_count = int(row["item_count"] or 0)
        unknown = int(row["histamine_unknown_count"] or 0)
        complete = (
            item_count > 0
            and unknown == 0
            and row["histamine_label"]
            in {"classified", "green", "yellow", "orange", "red"}
        )
        raw_value: Any = None
        if metric.source_identifier == "histamine_score" and complete:
            raw_value = row["histamine_score"]
        elif metric.source_identifier == "histamine_max" and item_count > unknown:
            raw_value = row["histamine_max"]
        elif metric.source_identifier == "mapping_coverage":
            raw_value = 100 * (item_count - unknown) / item_count
        value = _public_number(raw_value)
        if day and value is not None:
            result.append(
                {
                    "date": day,
                    "value": value,
                    "quality": "complete" if complete else "incomplete",
                }
            )
    return result


def _lab_key(parameter: Any, unit: Any) -> tuple[str, str]:
    def slug(value: Any) -> str:
        normalized = (
            str(value or "")
            .casefold()
            .replace("µ", "u")
            .replace("μ", "u")
            .replace("%", " percent ")
        )
        return re.sub(r"[^a-z0-9]+", "_", normalized).strip("_")

    return slug(parameter), slug(unit)


def _verified_lab_rows(
    connection: sqlite3.Connection,
    *,
    include_missing_reference: bool = False,
    start: date | None = None,
    end: date | None = None,
) -> list[dict[str, Any]]:
    if not table_exists(connection, "laborwerte"):
        return []
    date_expression = "COALESCE(NULLIF(l.abnahme_datum,''),l.befund_datum)"
    qualitative_values = sorted(LAB_QUALITATIVE)
    filters = [
        "lower(trim(COALESCE(l.validierungsstatus,'')))='validiert'",
        "l.verified_against_original=1",
        "l.reference_range_source='scanned_original'",
        "(trim(COALESCE(l.einheit,''))<>'' OR lower(trim(COALESCE(l.wert,''))) IN (" + ",".join("?" for _ in qualitative_values) + "))",
    ]
    values: list[Any] = list(qualitative_values)
    if start:
        filters.append(f"{date_expression}>=?")
        values.append(start.isoformat())
    if end:
        filters.append(f"{date_expression}<=?")
        values.append(end.isoformat())
    rows = list(
        connection.execute(
            """SELECT l.id,l.parameter_name,l.wert,l.einheit,l.bemerking,l.reference_min,l.reference_max,
                  l.abnahme_datum,l.befund_datum,l.reference_range_source,l.source_type,l.quelle,l.provenance_note,
                  l.dokument_id,l.canonical_document_id,d.id AS linked_document_record_id,
                  d.document_date,d.kategorie,d.institution,d.review_status
               FROM laborwerte l LEFT JOIN dokumente d ON d.id=l.canonical_document_id
               WHERE """
            + " AND ".join(filters)
            + f" ORDER BY {date_expression},l.id LIMIT ?",
            (*values, MAX_RAW_LAB_ROWS + 1),
        )
    )
    if len(rows) > MAX_RAW_LAB_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    grouped: dict[tuple[str, str, str, str], list[sqlite3.Row]] = defaultdict(list)
    for row in rows:
        canonical = LAB_ALLOWLIST.get(_lab_key(row["parameter_name"], row["einheit"]))
        if canonical is None and not str(row["einheit"] or "").strip():
            parameter_key = _lab_key(row["parameter_name"], "")[0]
            matches = {public[0] for (raw_parameter, _raw_unit), public in LAB_ALLOWLIST.items() if raw_parameter == parameter_key}
            if len(matches) == 1:
                canonical = (next(iter(matches)), "")
        day = parse_day(row["abnahme_datum"]) or parse_day(row["befund_datum"])
        if canonical and day and date.fromisoformat(day) <= local_today():
            source_identity = ":".join(
                (
                    str(row["source_type"] or "unknown"),
                    str(row["canonical_document_id"] or row["dokument_id"] or row["provenance_note"] or row["quelle"] or row["id"]),
                )
            )
            grouped[(canonical[0], canonical[1], day, source_identity)].append(row)
    result: list[dict[str, Any]] = []
    for (parameter, unit, day, _source_identity), same_day in sorted(grouped.items()):
        if len(same_day) != 1:
            continue
        row = same_day[0]
        raw_value = ("" if row["wert"] is None else str(row["wert"])).strip()
        result_match = LAB_RESULT_NUMBER.fullmatch(raw_value.replace(" ", ""))
        qualitative = raw_value.casefold() if result_match is None else None
        if result_match is None and qualitative not in LAB_QUALITATIVE:
            continue
        operator = (result_match.group("operator") or "").replace("≤", "<=").replace("≥", ">=") or None if result_match else None
        if result_match:
            numeric = float(result_match.group("number").replace(",", "."))
            if not math.isfinite(numeric):
                continue
            value: int | float | str = int(numeric) if numeric.is_integer() else numeric
            display_value = f"{operator or ''}{result_match.group('number')}"
            value_kind = "numeric"
        else:
            value = raw_value
            display_value = raw_value
            value_kind = "qualitative"
        reference_min = str(row["reference_min"] or "").strip()
        reference_max = str(row["reference_max"] or "").strip()
        reference_min = (
            reference_min
            if EXACT_LAB_NUMBER.fullmatch(reference_min)
            and math.isfinite(float(reference_min.replace(",", ".")))
            else None
        )
        reference_max = (
            reference_max
            if EXACT_LAB_NUMBER.fullmatch(reference_max)
            and math.isfinite(float(reference_max.replace(",", ".")))
            else None
        )
        item: dict[str, Any] = {
            "id": api_observation_id("lab-observation", row["id"], parameter, unit, day),
            "metric_id": LAB_CATALOG_SPECS[parameter].metric_id,
            "parameter": parameter,
            "value": value,
            "value_kind": value_kind,
            "operator": operator,
            "display_value": display_value,
            "unit": unit,
            "date": day,
            "quality": "verified_original",
            "document_link_status": (
                "linked" if row["linked_document_record_id"] is not None else "unlinked"
            ),
            "document_review_status": (
                "reviewed"
                if row["linked_document_record_id"] is not None
                and row["review_status"] == "geprueft"
                else "review_pending"
                if row["linked_document_record_id"] is not None
                else "not_linked"
            ),
        }
        if reference_min is None and reference_max is None:
            if not include_missing_reference:
                continue
            item["reference_status"] = "missing_reference"
        else:
            item["reference"] = {
                "min": reference_min,
                "max": reference_max,
                "source": "scanned_original",
            }
            source_reference = str(row["bemerking"])[len("Quell-Referenzbereich: "):] if str(row["bemerking"] or "").startswith("Quell-Referenzbereich: ") else None
            if source_reference:
                item["reference"]["raw"] = source_reference
            item["reference_status"] = "verified_observation_specific"
        if (
            row["canonical_document_id"] is not None
            and row["review_status"] == "geprueft"
        ):
            category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
            institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
            document_day = parse_day(row["document_date"])
            if category is not None and institution is not None:
                item["document"] = {
                    "id": api_document_id(
                        row["canonical_document_id"],
                        document_day,
                        category,
                        institution,
                    ),
                    "date": document_day,
                    "category": category,
                    "institution": institution,
                }
        result.append(item)
    return result


def _metric_points(
    connection: sqlite3.Connection,
    metric: MetricV2,
    start: date | None,
    end: date | None,
) -> list[dict[str, Any]]:
    if metric.value_type == "panel":
        raise APIError(400, "metric_not_series")
    if metric.source == "apple_health":
        if metric.source_identifier == "blood_pressure":
            return _blood_pressure_points(connection, metric, start, end)
        if metric.id.startswith("apple.sleep."):
            return _sleep_phase_points(connection, metric, start, end)
        return _apple_points(connection, metric, start, end)
    if metric.id == "symptom.total":
        return _symptom_total_points(connection, start, end)
    if metric.id == "symptom.aphthae":
        return _symptom_dimension_points(connection, metric, start, end)
    if metric.source == "nutrition_daily_summary_v2":
        return _nutrition_points(connection, metric, start, end)
    if metric.source == "laborwerte":
        return [
            {
                key: row[key]
                for key in ("date", "value", "operator", "display_value", "quality", "reference", "reference_status")
                if key in row
            }
            for row in _verified_lab_rows(connection, include_missing_reference=True)
            if row["metric_id"] == metric.id
            and row.get("value_kind") == "numeric"
            and row.get("operator") in {None, "="}
            and (start is None or row["date"] >= start.isoformat())
            and (end is None or row["date"] <= end.isoformat())
        ]
    return []


def _weekly_rule(metric: MetricV2) -> dict[str, Any]:
    if metric.aggregation == "observation" or metric.value_type == "panel":
        return {
            "id": "unsupported_for_metric",
            "minimum_observations": None,
            "requires_complete_calendar_week": False,
        }
    if metric.expected_frequency == "daily":
        return {
            "id": "complete_calendar_week_daily_metric",
            "minimum_observations": 7,
            "requires_complete_calendar_week": True,
        }
    if metric.expected_frequency == "intermittent":
        return {
            "id": "observed_measurements_within_iso_week",
            "minimum_observations": 1,
            "requires_complete_calendar_week": False,
        }
    return {
        "id": "unsupported_for_metric",
        "minimum_observations": None,
        "requires_complete_calendar_week": False,
    }


def _add_past_only_baselines(points: list[dict[str, Any]], metric: MetricV2) -> None:
    rule = metric.baseline_rule
    if not rule:
        return
    window = int(rule["window_days"])
    minimum = int(rule["minimum_observations"])
    for point in points:
        day = date.fromisoformat(point["date"])
        prior = [
            float(candidate["value"])
            for candidate in points
            if date.fromisoformat(candidate["date"]) < day
            and (day - date.fromisoformat(candidate["date"])).days <= window
        ]
        point["baseline"] = (
            _numeric(float(median(prior)), metric.precision, metric.value_type)
            if len(prior) >= minimum
            else None
        )


def _weekly(
    points: list[dict[str, Any]], metric: MetricV2, rule: dict[str, Any]
) -> list[dict[str, Any]]:
    if rule["id"] == "unsupported_for_metric":
        return []
    grouped: dict[tuple[int, int], list[dict[str, Any]]] = defaultdict(list)
    for point in points:
        day = date.fromisoformat(point["date"])
        iso = day.isocalendar()
        grouped[(iso.year, iso.week)].append(point)
    result = []
    for (year, week), rows in sorted(grouped.items()):
        distinct = {row["date"] for row in rows}
        if rule["requires_complete_calendar_week"] and len(distinct) != 7:
            continue
        values = [float(row["value"]) for row in rows]
        if metric.aggregation == "sum":
            value = sum(values)
        elif metric.aggregation == "last":
            value = float(max(rows, key=lambda row: row["date"])["value"])
        else:
            value = sum(values) / len(values)
        week_start = date.fromisocalendar(year, week, 1)
        week_end = week_start + timedelta(days=6)
        item = {
            "week": f"{year}-W{week:02d}",
            "date": min(distinct),
            "week_start": week_start.isoformat(),
            "week_end": week_end.isoformat(),
            "drilldown_date": week_start.isoformat(),
            "value": _numeric(value, metric.precision, metric.value_type),
            "observation_count": len(rows),
            "observed_days": len(distinct),
            "quality": (
                "complete_calendar_week"
                if rule["requires_complete_calendar_week"]
                else "observed_measurements"
            ),
        }
        result.append(item)
    return result


def _coverage(
    points: list[dict[str, Any]], start: date | None, end: date | None
) -> dict[str, Any]:
    observed = sorted({date.fromisoformat(point["date"]) for point in points})
    effective_start = start or (observed[0] if observed else None)
    effective_end = end or (observed[-1] if observed else None)
    if effective_start is not None and effective_end is not None:
        span = (effective_end - effective_start).days + 1
        if span > MAX_RANGE_DAYS:
            raise APIError(422, "range_too_large")
    else:
        span = 0
    return {
        "from": effective_start.isoformat() if effective_start else None,
        "to": effective_end.isoformat() if effective_end else None,
        "expected_days": span,
        "observed_days": len(observed),
        "observation_count": len(points),
        "missing_days": max(0, span - len(observed)),
        "gaps": "omitted_not_interpolated",
    }


def _series(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    metric_id = params.get("metric", "")
    metric = BY_ID_V2.get(metric_id)
    if metric is None:
        raise APIError(404, "metric_not_allowed")
    resolution = params.get("resolution", "day")
    if resolution not in ALLOWED_RESOLUTIONS:
        raise APIError(400, "resolution_not_allowed")
    start, end = parse_range(params)
    points = _metric_points(connection, metric, start, end)
    today = local_today().isoformat()
    points = [point for point in points if point["date"] <= today]
    _add_past_only_baselines(points, metric)
    if start is None and points:
        observed_start = date.fromisoformat(points[0]["date"])
        observed_end = date.fromisoformat(points[-1]["date"])
        if (observed_end - observed_start).days + 1 > MAX_RANGE_DAYS:
            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,
                ),
            )
        )
        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 "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 ?""",
                (end.isoformat(), start.isoformat(), MAX_EVENT_SOURCE_ROWS + 1),
            )
        )
        if len(rows) > MAX_EVENT_SOURCE_ROWS:
            raise APIError(422, "source_row_limit_exceeded")
        for row in rows:
            event_type = safe_metadata_text(row["event_type"], 80)
            label = safe_metadata_text(row["label"], 120, allow_empty=True)
            start_day = parse_day(row["start_date"])
            end_day = parse_day(row["end_date"]) if row["end_date"] else None
            if (
                start_day is None
                or (end_day is not None and end_day < start_day)
                or (end_day is not None and date.fromisoformat(end_day) > local_today())
            ):
                continue
            if event_type and label is not None:
                result.append(
                    {
                        "date": start_day,
                        "end_date": end_day,
                        "type": "health_period",
                        "category": event_type,
                        "label": label,
                    }
                )
    if "nutrition_day" in selected and table_exists(connection, "nutrition_daily_summary_v2"):
        for row in connection.execute("SELECT datum FROM nutrition_daily_summary_v2 WHERE datum>=? AND datum<=? AND item_count>0 ORDER BY datum LIMIT ?", (start.isoformat(), end.isoformat(), MAX_EVENT_SOURCE_ROWS + 1)):
            day = parse_day(row["datum"])
            if day:
                result.append({"date": day, "type": "nutrition_day", "label": "Ernährung dokumentiert"})
    if "laboratory" in selected:
        for item in _verified_lab_rows(connection, start=start, end=end):
            result.append({"date": item["date"], "type": "laboratory", "label": item["parameter"]})
    if "document" in selected and table_exists(connection, "dokumente"):
        for row in connection.execute("SELECT document_date,kategorie FROM dokumente WHERE document_date>=? AND document_date<=? AND review_status='geprueft' ORDER BY document_date,id LIMIT ?", (start.isoformat(), end.isoformat(), MAX_EVENT_SOURCE_ROWS + 1)):
            day = parse_day(row["document_date"])
            label = safe_metadata_text(row["kategorie"], 120, allow_empty=True)
            if day and label is not None:
                result.append({"date": day, "type": "document", "label": label or "Geprüftes Dokument"})
    if "appointment" in selected and table_exists(connection, "arztbesuche"):
        for row in connection.execute("SELECT datum,grund FROM arztbesuche WHERE 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["grund"], 120, allow_empty=True)
            if day and label is not None:
                result.append({"date": day, "type": "appointment", "label": label or "Arzttermin/Untersuchung"})
    result.sort(key=lambda item: (item["date"], item["type"], item.get("label", "")))
    if len(result) > MAX_EVENT_ROWS:
        raise APIError(422, "row_limit_exceeded")
    return {
        "from": start.isoformat(),
        "to": end.isoformat(),
        "types": sorted(selected),
        "timezone": TZ_NAME,
        "events": result,
    }


def _safe_documents(
    connection: sqlite3.Connection, needle: str
) -> list[dict[str, Any]]:
    if not table_exists(connection, "dokumente"):
        return []
    result = []
    for row in connection.execute(
        """SELECT id,document_date,kategorie,institution FROM dokumente
           WHERE review_status='geprueft' ORDER BY document_date DESC,id DESC LIMIT 200"""
    ):
        category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
        institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
        document_day = parse_day(row["document_date"])
        if category is None or institution is None:
            continue
        haystack = f"{category} {institution} {document_day or ''}".casefold()
        if needle.casefold() not in haystack:
            continue
        result.append(
            {
                "id": api_document_id(row["id"], document_day, category, institution),
                "date": document_day,
                "category": category,
                "institution": institution,
                "drill_down_target": "document_metadata",
            }
        )
        if len(result) >= MAX_SEARCH_ROWS_PER_GROUP:
            break
    return result


def _fts_search_documents(
    connection: sqlite3.Connection, query: str
) -> list[dict[str, Any]]:
    if not table_exists(connection, "health_document_fts"):
        return []
    page = _record_document_rows(
        connection,
        {"q": query, "limit": str(MAX_SEARCH_ROWS_PER_GROUP)},
    )
    return [
        {
            "id": row["id"],
            "date": row["document_date"],
            "category": row["category"],
            "institution": row["institution"],
            "snippet": row["snippet"],
            "drill_down_target": "document",
        }
        for row in page["documents"]
    ]


def _machine_search_documents(
    connection: sqlite3.Connection, query: str
) -> list[dict[str, Any]]:
    if not table_exists(connection, "health_document_machine_fts"):
        return []
    result = []
    for row in connection.execute(
        """SELECT d.id,d.document_date,d.kategorie,d.institution,
                  snippet(health_document_machine_fts,3,'','','…',12) AS snippet,
                  f.review_label
           FROM health_document_machine_fts f JOIN dokumente d ON d.id=f.document_id
           WHERE health_document_machine_fts MATCH ?
           ORDER BY bm25(health_document_machine_fts) LIMIT ?""",
        (_fts_literal_query(_fts_tokens(query)), MAX_SEARCH_ROWS_PER_GROUP),
    ):
        category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
        institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
        snippet = safe_metadata_text(row["snippet"], 300, allow_empty=True)
        review_label = safe_metadata_text(row["review_label"], 80)
        document_day = parse_day(row["document_date"])
        if category is None or institution is None or snippet is None or review_label is None:
            continue
        result.append({
            "id": api_document_id(row["id"], document_day, category, institution),
            "date": document_day,
            "category": category,
            "institution": institution,
            "snippet": snippet,
            "review_label": review_label,
            "drill_down_target": "document_machine_extraction",
        })
    return result


def _search(connection: sqlite3.Connection, query: str, *, include_machine: bool = False) -> dict[str, Any]:
    clean = " ".join(query.split())
    if len(clean) < 2 or len(clean) > 80:
        raise APIError(400, "query_length_invalid")
    matches = catalog_search(connection, clean)
    groups: dict[str, list[dict[str, Any]]] = {
        "metrics": [],
        "laboratory": [],
        "symptoms_events": [],
        "documents": [],
        "days": [],
    }
    for item in matches:
        target = (
            "laboratory"
            if item["category"] == "laboratory"
            else ("symptoms_events" if item["category"] == "symptoms" else "metrics")
        )
        if len(groups[target]) < MAX_SEARCH_ROWS_PER_GROUP:
            groups[target].append(item)
    normalized = clean.casefold()
    if normalized in {"medikamente", "medikament", "medikation", "arznei"}:
        groups["symptoms_events"].append(
            {
                "id": "event.medication",
                "label": "Medikamente",
                "category": "events",
                "drill_down_target": "day",
                "availability": {"status": "supported"},
            }
        )
    groups["documents"] = _fts_search_documents(connection, clean)
    if include_machine:
        machine_matches = _machine_search_documents(connection, clean)
        known = {item["id"] for item in groups["documents"]}
        groups["documents"].extend(item for item in machine_matches if item["id"] not in known)
        groups["documents"] = groups["documents"][:MAX_SEARCH_ROWS_PER_GROUP]
    if not groups["documents"]:
        groups["documents"] = _safe_documents(connection, clean)
    try:
        parsed = parse_iso_day(clean)
    except APIError:
        parsed = None
    if parsed is not None and parsed <= local_today():
        groups["days"] = [{"date": parsed.isoformat(), "drill_down_target": "day"}]
    return {"query": clean, "timezone": TZ_NAME, "groups": groups}


def _safe_day_text(
    value: Any, maximum: int = 160, *, allow_empty: bool = True
) -> str | None:
    return safe_metadata_text(value, maximum, allow_empty=allow_empty)


def _day_symptoms(connection: sqlite3.Connection, day: str) -> dict[str, Any]:
    dimensions = {
        label: {"value": None, "note": None, "status": "not_documented"}
        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),
        )
    )
    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)
    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}


def _day_documents(
    connection: sqlite3.Connection, requested: str
) -> list[dict[str, Any]]:
    if not table_exists(connection, "dokumente"):
        return []
    result = []
    rows = connection.execute(
        """SELECT id,document_date,kategorie,institution FROM dokumente
           WHERE document_date=? AND review_status='geprueft' ORDER BY id LIMIT 100""",
        (requested,),
    )
    for row in rows:
        category = _safe_day_text(row["kategorie"], 80)
        institution = _safe_day_text(row["institution"], 120)
        if category is None or institution is None:
            continue
        result.append(
            {
                "id": api_document_id(row["id"], requested, category, institution),
                "date": requested,
                "category": category,
                "institution": institution,
            }
        )
    return result


def _day_appointments(connection: sqlite3.Connection, requested: str) -> dict[str, Any]:
    if not table_exists(connection, "arztbesuche"):
        return {"status": "supported_no_data", "items": []}
    items = []
    for row in connection.execute(
        "SELECT arzt,klinik,grund,notizen FROM arztbesuche WHERE datum=? ORDER BY id LIMIT 100",
        (requested,),
    ):
        practitioner = _safe_day_text(row["arzt"], 120)
        institution = _safe_day_text(row["klinik"], 120)
        reason = _safe_day_text(row["grund"], 160)
        note = _safe_day_text(row["notizen"], 300)
        if all(
            value is not None for value in (practitioner, institution, reason, note)
        ):
            items.append(
                {
                    "date": requested,
                    "practitioner": practitioner,
                    "institution": institution,
                    "specialty_or_reason": reason,
                    "note": note,
                    # arztbesuche currently has no explicit status column.  Its date
                    # documents that an entry exists, but it never proves attendance,
                    # cancellation, or completion; retain that distinction explicitly.
                    "status": "documented_visit",
                }
            )
    return {"status": "supported" if items else "supported_no_data", "items": items}


def _day_nutrition(connection: sqlite3.Connection, requested: str) -> dict[str, Any]:
    if not table_exists(connection, "nutrition_daily_summary_v2"):
        return {"status": "supported_no_data", "date": requested}
    row = connection.execute(
        """SELECT kcal,protein_g,carb_g,fat_g,item_count,histamine_score,
                  histamine_unknown_count,histamine_label
           FROM nutrition_daily_summary_v2 WHERE datum=? LIMIT 1""",
        (requested,),
    ).fetchone()
    if row is None:
        return {"status": "supported_no_data", "date": requested}
    complete = (
        int(row["item_count"] or 0) > 0
        and int(row["histamine_unknown_count"] or 0) == 0
        and row["histamine_score"] is not None
    )
    return {
        "status": "documented",
        "date": requested,
        "kcal": row["kcal"],
        "protein_g": row["protein_g"],
        "carb_g": row["carb_g"],
        "fat_g": row["fat_g"],
        "item_count": row["item_count"],
        "histamine_score": row["histamine_score"] if complete else None,
        "histamine_status": row["histamine_label"] if complete else "not_documented",
        "unknown_items": row["histamine_unknown_count"],
        "complete": complete,
    }


def _nutrition_complete(row: sqlite3.Row) -> bool:
    return (
        int(row["item_count"] or 0) > 0
        and int(row["histamine_unknown_count"] or 0) == 0
        and row["histamine_score"] is not None
        and row["histamine_label"] in {"classified", "green", "yellow", "orange", "red"}
    )


def _public_number(value: Any) -> int | float | None:
    try:
        numeric = float(value)
    except (TypeError, ValueError):
        return None
    if not math.isfinite(numeric):
        return None
    rounded = round(numeric, 3)
    return int(rounded) if rounded.is_integer() else rounded


def _nutrition_reference_profile(connection: sqlite3.Connection) -> dict[str, Any]:
    """Read one explicit local profile; never infer demographics from health records."""
    if not table_exists(connection, "nutrition_reference_profile"):
        return {}
    columns = {
        str(row[1]) for row in connection.execute("PRAGMA table_info(nutrition_reference_profile)")
    }
    allowed = (
        "birth_date",
        "sex",
        "life_stage",
        "activity_level",
        "body_weight_kg",
        "phytate_mg_per_day",
        "pregnancy_trimester",
        "lactation_month",
    )
    selected = [name for name in allowed if name in columns]
    if not {"birth_date", "sex"} <= set(selected):
        return {}
    where = " WHERE active=1" if "active" in columns else ""
    rows = list(
        connection.execute(
            f"SELECT {','.join(selected)} FROM nutrition_reference_profile{where} LIMIT 2"
        )
    )
    if len(rows) != 1:
        return {}
    return {name: rows[0][name] for name in selected if rows[0][name] is not None}


def _daily_nutrients(
    connection: sqlite3.Connection, start: date | None, end: date | None
) -> dict[str, dict[str, dict[str, Any]]]:
    if not table_exists(connection, "nutrition_item_nutrients"):
        return {}
    nutrient_columns = {
        str(item[1]) for item in connection.execute("PRAGMA table_info(nutrition_item_nutrients)")
    }
    provenance_projection = ",".join(
        f"n.{name}" if name in nutrient_columns else f"NULL AS {name}"
        for name in (
            "raw_value",
            "raw_unit",
            "canonical_unit",
            "conversion_factor",
            "value_status",
            "conversion_contract_version",
        )
    )
    sql = f"""SELECT i.id,i.datum,i.amount,i.amount_unit,
                    n.nutrient_key,n.value,n.unit,{provenance_projection}
             FROM nutrition_item_nutrients n JOIN nutrition_items i ON i.id=n.item_id
             WHERE i.datum<=?"""
    values: list[Any] = [(end or local_today()).isoformat()]
    if start:
        sql += " AND i.datum>=?"
        values.append(start.isoformat())
    sql += " ORDER BY i.datum,n.nutrient_key LIMIT ?"
    values.append(MAX_DAILY_SOURCE_ROWS + 1)
    rows = list(connection.execute(sql, values))
    if len(rows) > MAX_DAILY_SOURCE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    item_sql = "SELECT datum,COUNT(*) AS count FROM nutrition_items WHERE datum<=?"
    item_values: list[Any] = [(end or local_today()).isoformat()]
    if start:
        item_sql += " AND datum>=?"
        item_values.append(start.isoformat())
    item_sql += " GROUP BY datum"
    item_counts = {
        str(row["datum"]): int(row["count"])
        for row in connection.execute(item_sql, item_values)
    }
    yazio_totals: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
    day_keys: dict[str, set[str]] = defaultdict(set)
    known_items: dict[str, dict[str, set[int]]] = defaultdict(
        lambda: defaultdict(set)
    )
    status_counts: dict[str, dict[str, dict[str, int]]] = defaultdict(
        lambda: defaultdict(lambda: defaultdict(int))
    )
    raw_units: dict[str, dict[str, set[str]]] = defaultdict(lambda: defaultdict(set))
    enrichment_totals: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
    enrichment_sources: dict[str, dict[str, set[str]]] = defaultdict(lambda: defaultdict(set))
    for row in rows:
        key = str(row["nutrient_key"] or "")
        contract = NUTRIENT_CONTRACTS.get(key)
        day = parse_day(row["datum"])
        if contract is None or day is None:
            continue
        day_keys[day].add(key)
        stored_status = str(row["value_status"] or "")
        if (
            stored_status in {"documented_value", "documented_zero"}
            and row["canonical_unit"] == contract.unit
            and row["conversion_contract_version"] == NUTRIENT_CONTRACT_VERSION
        ):
            normalized_status = {
                "status": stored_status,
                "value": _public_number(row["value"]),
                "unit": contract.unit,
            }
        elif row["raw_value"] is not None:
            # v2 stored the consumed raw mass without its source unit. This exact,
            # key-specific legacy rule is versioned; zero remains unknown because
            # old exports cannot prove whether it was explicit or a default.
            normalized_status = normalize_nutrient_status(
                key, row["raw_value"], contract.source_unit, explicit_zero=False
            )
        else:
            normalized_status = normalize_nutrient_status(
                key, row["value"], row["unit"], explicit_zero=True
            )
        status = str(normalized_status["status"])
        status_counts[day][key][status] += 1
        source_unit = row["raw_unit"] or normalized_status.get("raw_unit") or row["unit"]
        if source_unit:
            raw_units[day][key].add(str(source_unit))
        if status in {"documented_value", "documented_zero"} and normalized_status.get("value") is not None:
            yazio_totals[day][key] += float(normalized_status["value"])
            known_items[day][key].add(int(row["id"]))
    if table_exists(connection, "food_nutrient_enrichment"):
        enriched_sql = """SELECT i.id,i.datum,i.amount,i.amount_unit,e.nutrient_key,
                                   e.value_per_100g,e.unit,e.density,e.source_name,e.source_version
                            FROM nutrition_items i JOIN food_nutrient_enrichment e
                              ON e.yazio_name=i.name
                            WHERE i.datum<=? AND e.review_status IN ('auto_applied','approved')
                              AND NOT EXISTS(
                                SELECT 1 FROM nutrition_item_nutrients n
                                WHERE n.item_id=i.id AND n.nutrient_key=e.nutrient_key)
                         """
        enriched_values: list[Any] = [(end or local_today()).isoformat()]
        if start:
            enriched_sql += " AND i.datum>=?"
            enriched_values.append(start.isoformat())
        enriched_sql += " ORDER BY i.datum,i.id,e.nutrient_key LIMIT ?"
        enriched_values.append(MAX_DAILY_SOURCE_ROWS + 1)
        enriched_rows = list(connection.execute(enriched_sql, enriched_values))
        if len(enriched_rows) > MAX_DAILY_SOURCE_ROWS:
            raise APIError(422, "source_row_limit_exceeded")
        for row in enriched_rows:
            key = str(row["nutrient_key"] or "")
            contract = NUTRIENT_CONTRACTS.get(key)
            day = parse_day(row["datum"])
            amount = _public_number(row["amount"])
            raw = _public_number(row["value_per_100g"])
            if contract is None or day is None or amount is None or raw is None or row["unit"] != contract.unit:
                continue
            amount_unit = str(row["amount_unit"] or "").casefold()
            if amount_unit == "g":
                grams = float(amount)
            elif amount_unit == "ml" and _public_number(row["density"]) is not None:
                grams = float(amount) * float(row["density"])
            else:
                continue
            if grams < 0 or grams > 100_000:
                continue
            day_keys[day].add(key)
            enrichment_totals[day][key] += float(raw) * grams / 100.0
            status_counts[day][key]["estimated"] += 1
            known_items[day][key].add(int(row["id"]))
            enrichment_sources[day][key].add(f"{row['source_name']} {row['source_version']}")
    all_days = sorted(day_keys)
    return {
        day: {
            key: {
                "label": NUTRIENT_CONTRACTS[key].label,
                "group": NUTRIENT_CONTRACTS[key].group,
                "value": _public_number(
                    float(yazio_totals[day].get(key) or 0.0)
                    + float(enrichment_totals[day].get(key) or 0.0)
                ) if key in yazio_totals[day] or key in enrichment_totals[day] else None,
                "yazio_value": _public_number(yazio_totals[day].get(key)) if key in yazio_totals[day] else None,
                "enrichment_value": _public_number(enrichment_totals[day].get(key)) if key in enrichment_totals[day] else None,
                "unit": NUTRIENT_CONTRACTS[key].unit,
                "data_status": (
                    "documented_zero"
                    if key in yazio_totals[day] and yazio_totals[day][key] == 0
                    else "documented_value"
                    if key in yazio_totals[day]
                    else "estimated"
                    if key in enrichment_totals[day]
                    else "not_reported"
                    if status_counts[day][key].get("not_reported", 0)
                    and not status_counts[day][key].get("unknown", 0)
                    else "unknown"
                ),
                "status_counts": dict(status_counts[day][key]),
                "item_count": item_counts.get(day, 0),
                "known_item_count": len(known_items[day][key]),
                "unknown_item_count": max(
                    0, item_counts.get(day, 0) - len(known_items[day][key])
                ),
                "data_completeness": (
                    "complete"
                    if item_counts.get(day, 0) > 0
                    and len(known_items[day][key]) == item_counts.get(day, 0)
                    else "partial"
                ),
                "raw_units": sorted(raw_units[day][key]),
                "source": "YAZIO" if key not in enrichment_totals[day] else "YAZIO plus lokale Schätzung",
                "enrichment_sources": sorted(enrichment_sources[day].get(key, set())),
                "is_estimated": key in enrichment_totals[day],
                "conversion": NUTRIENT_CONTRACTS[key].public_conversion(),
                "contract_version": NUTRIENT_CONTRACT_VERSION,
                "null_contract_version": NULL_VALUE_CONTRACT_VERSION,
            }
            for key in sorted(day_keys[day])
        }
        for day in all_days
    }


def _nutrition_contributions(
    connection: sqlite3.Connection,
    start: date,
    end: date | None = None,
    *,
    divisor: int = 1,
) -> dict[str, list[dict[str, Any]]]:
    """Return bounded, allowlisted food contributions without source identifiers."""
    if not table_exists(connection, "nutrition_item_nutrients"):
        return {}
    totals: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
    effective_end = end or start
    rows = list(
        connection.execute(
            """SELECT i.name,n.nutrient_key,n.value,n.unit
                 FROM nutrition_item_nutrients n
                 JOIN nutrition_items i ON i.id=n.item_id
                WHERE i.datum>=? AND i.datum<=?
                ORDER BY i.id,n.nutrient_key LIMIT ?""",
            (start.isoformat(), effective_end.isoformat(), MAX_DAILY_SOURCE_ROWS + 1),
        )
    )
    if len(rows) > MAX_DAILY_SOURCE_ROWS:
        raise APIError(422, "source_row_limit_exceeded")
    for row in rows:
        key = str(row["nutrient_key"] or "")
        contract = NUTRIENT_CONTRACTS.get(key)
        normalized = normalize_nutrient(key, row["value"], row["unit"])
        name = safe_metadata_text(row["name"], 120, allow_empty=False)
        if contract is None or normalized is None or name is None:
            continue
        value, _unit = normalized
        totals[key][name] += float(value)
    result: dict[str, list[dict[str, Any]]] = {}
    for key, foods in totals.items():
        unit = NUTRIENT_CONTRACTS[key].unit
        result[key] = sorted(
            (
                {
                    "food": name,
                    "amount": _public_number(value / max(1, divisor)),
                    "unit": unit,
                }
                for name, value in foods.items()
            ),
            key=lambda item: float(item["amount"] or 0),
            reverse=True,
        )[:5]
    return dict(result)


def _nutrition_days(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    start, end = parse_range(params)
    if not table_exists(connection, "nutrition_daily_summary_v2"):
        return {"version": 1, "timezone": TZ_NAME, "definition": _histamine_definition(), "days": []}
    sql = """SELECT datum,kcal,protein_g,carb_g,fat_g,item_count,histamine_score,
                    histamine_max,histamine_unknown_count,histamine_label
             FROM nutrition_daily_summary_v2 WHERE datum<=?"""
    values: list[Any] = [local_today().isoformat()]
    if start:
        sql += " AND datum>=?"
        values.append(start.isoformat())
    if end:
        sql += " AND datum<=?"
        values.append(end.isoformat())
    sql += " ORDER BY datum DESC LIMIT ?"
    values.append(MAX_NUTRITION_DAYS + 1)
    rows = list(connection.execute(sql, values))
    if len(rows) > MAX_NUTRITION_DAYS:
        raise APIError(422, "row_limit_exceeded")
    daily_nutrients = _daily_nutrients(connection, start, end)
    effective_start = start or (
        date.fromisoformat(str(rows[-1]["datum"])) if rows else local_today()
    )
    effective_end = end or local_today()
    try:
        supplement_totals = supplement_nutrient_totals(
            connection, effective_start, effective_end
        )
    except ValueError as error:
        raise APIError(422, str(error)) from error
    days = []
    for row in reversed(rows):
        day = parse_day(row["datum"])
        if not day:
            continue
        complete = _nutrition_complete(row)
        mapped_items = max(
            0,
            int(row["item_count"] or 0)
            - int(row["histamine_unknown_count"] or 0),
        )
        food_nutrients = daily_nutrients.get(day, {})
        day_supplements = supplement_totals.get(day, {})
        combined_nutrients: dict[str, dict[str, Any]] = {}
        for key in sorted(set(food_nutrients) | set(day_supplements)):
            contract = NUTRIENT_CONTRACTS[key]
            food = food_nutrients.get(key, {})
            extra = day_supplements.get(key, {})
            food_value = food.get("value")
            yazio_value = food.get("yazio_value")
            enrichment_value = food.get("enrichment_value")
            supplement_value = extra.get("value")
            combined_values = [
                float(value)
                for value in (food_value, supplement_value)
                if value is not None
            ]
            total_value = _public_number(sum(combined_values)) if combined_values else None
            combined_nutrients[key] = {
                "label": contract.label,
                "group": contract.group,
                "unit": contract.unit,
                "data_status": food.get("data_status") or ("documented_value" if supplement_value is not None else "not_reported"),
                "status_counts": food.get("status_counts", {}),
                "item_count": food.get("item_count", int(row["item_count"] or 0)),
                "known_item_count": food.get("known_item_count", 0),
                "unknown_item_count": food.get(
                    "unknown_item_count", int(row["item_count"] or 0)
                ),
                "data_completeness": food.get("data_completeness", "partial"),
                "raw_units": food.get("raw_units", []),
                "conversion": contract.public_conversion(),
                "value": food_value,
                "food_value": food_value,
                "yazio_value": yazio_value,
                "enrichment_value": enrichment_value,
                "supplement_value": supplement_value,
                "total_value": total_value,
                "food_source": "YAZIO",
                "enrichment_sources": food.get("enrichment_sources", []),
                "supplement_source": "Dokumentierte Supplemente",
                "contains_estimate": bool(food.get("is_estimated")) or bool(extra.get("contains_estimate")),
                "contract_version": NUTRIENT_CONTRACT_VERSION,
            }
        days.append({
            "date": day,
            "macros": {
                "kcal": _public_number(row["kcal"]),
                "protein_g": _public_number(row["protein_g"]),
                "carb_g": _public_number(row["carb_g"]),
                "fat_g": _public_number(row["fat_g"]),
            },
            "item_count": int(row["item_count"] or 0),
            "nutrients": combined_nutrients,
            "histamine": {
                "score": _public_number(row["histamine_score"]) if complete else None,
                "mapped_score_sum": (
                    _public_number(row["histamine_score"]) if mapped_items else None
                ),
                "max_score": int(row["histamine_max"] or 0) if mapped_items else None,
                "mapped_items": mapped_items,
                "mapping_coverage": round((int(row["item_count"] or 0) - int(row["histamine_unknown_count"] or 0)) / int(row["item_count"] or 1), 3) if int(row["item_count"] or 0) else 0,
                "unknown_items": int(row["histamine_unknown_count"] or 0),
                "status": "complete" if complete else "incomplete",
                "definition_version": _histamine_definition()["version"],
            },
        })
    averages: dict[str, dict[str, Any]] = {}
    expected_days = (effective_end - effective_start).days + 1
    for key, contract in NUTRIENT_CONTRACTS.items():
        documented = [
            float(item["nutrients"][key]["food_value"])
            for item in days
            if key in item["nutrients"]
            and item["nutrients"][key]["food_value"] is not None
        ]
        combined_values = [
            float(item["nutrients"][key]["total_value"])
            for item in days
            if key in item["nutrients"]
            and item["nutrients"][key]["total_value"] is not None
        ]
        food_values = [
            float(item["nutrients"][key]["food_value"])
            for item in days
            if key in item["nutrients"]
            and item["nutrients"][key]["food_value"] is not None
        ]
        supplement_values = [
            float(item["nutrients"][key]["supplement_value"])
            for item in days
            if key in item["nutrients"]
            and item["nutrients"][key]["supplement_value"] is not None
        ]
        if documented:
            averages[key] = {
                "label": contract.label,
                "value": _public_number(sum(documented) / len(documented)),
                "arithmetic_mean": _public_number(sum(documented) / len(documented)),
                "median": _public_number(median(documented)),
                "food_mean": _public_number(sum(food_values) / len(food_values)) if food_values else None,
                "supplement_mean": _public_number(sum(supplement_values) / len(supplement_values)) if supplement_values else None,
                "combined_mean": _public_number(sum(combined_values) / len(combined_values)) if combined_values else None,
                "unit": contract.unit,
                "documented_days": len(documented),
                "present_days": len(documented),
                "expected_days": expected_days,
                "period": {"from": effective_start.isoformat(), "to": effective_end.isoformat()},
                "calculation": "arithmetic_mean_per_documented_day",
                "aggregation": "arithmetic_mean_per_documented_day",
                "reference_context": public_reference_context(BY_ID_V2[contract.metric_id]),
            }
    reference_nutrients: dict[str, dict[str, Any]] = {}
    for key, contract in NUTRIENT_CONTRACTS.items():
        average = averages.get(key, {})
        per_day = [item["nutrients"].get(key, {}) for item in days]
        reference_nutrients[key] = {
            "value": average.get("food_mean"),
            "food_value": average.get("food_mean"),
            "unit": contract.unit,
            "documented_days": int(average.get("documented_days") or 0),
            "item_count": sum(int(item.get("item_count") or 0) for item in per_day),
            "unknown_item_count": sum(
                int(item.get("unknown_item_count") or 0) for item in per_day
            ),
        }
    reference_overview = build_overview(
        on_date=effective_end,
        profile=_nutrition_reference_profile(connection),
        nutrients=reference_nutrients,
        energy_kcal=(averages.get("energy.energy") or {}).get("food_mean"),
        documented_days=len(days),
        period_from=effective_start.isoformat(),
        period_to=effective_end.isoformat(),
        contributions=_nutrition_contributions(
            connection,
            effective_start,
            effective_end,
            divisor=1,
        ),
    )
    nutrition_factor_keys = {
        "energy.energy",
        "nutrient.dietaryfiber",
        "nutrient.sugar",
        "nutrient.saturated",
    }
    return {
        "version": 2,
        "timezone": TZ_NAME,
        "definition": _histamine_definition(),
        "nutrient_contract_version": NUTRIENT_CONTRACT_VERSION,
        "nutrient_reference_overview": reference_overview,
        "days": days,
        "summary": {
            "documented_days": len(days),
            "fully_mapped_days": sum(
                1 for item in days if item["histamine"]["status"] == "complete"
            ),
            "days_with_open_mappings": sum(
                1 for item in days if item["histamine"]["unknown_items"] > 0
            ),
            "averages": averages,
        },
        "nutrition_factors": [
            averages[key] for key in sorted(nutrition_factor_keys) if key in averages
        ],
        "nutrition_factor_statement": (
            "Ernährungsfaktoren im gewählten Zeitraum. Durchschnitt pro dokumentiertem "
            "Tag im gewählten Zeitraum; kein validierter Entzündungsindex."
        ),
    }


def _histamine_definition() -> dict[str, Any]:
    return {
        "id": "sighi_mapping_load_v1",
        "version": "sighi_mapping_load_v2",
        "display_name": "Histamin-Zuordnungsindex",
        "components": ["mapped_item_score_sum", "mapped_item_score_max", "mapping_coverage", "unknown_item_count"],
        "coverage_semantics": "unknown_or_unmapped_items_make_day_incomplete",
        "insufficient_inputs": "no_index",
        "score_scale": "SIGHi 0-3 or unknown per mapped item",
        "complete_rule": "complete only when every documented nutrition item has a mapping",
        "method_note": (
            "Der Index fasst dokumentierte SIGHi-Zuordnungen zusammen. Er misst nicht "
            "den tatsächlichen Histamingehalt einer Mahlzeit und ist keine individuelle "
            "Verträglichkeitsaussage."
        ),
    }


def _nutrition_day_detail(connection: sqlite3.Connection, requested: date) -> dict[str, Any]:
    day = requested.isoformat()
    if not table_exists(connection, "nutrition_daily_summary_v2"):
        return {
            "version": 1,
            "date": day,
            "timezone": TZ_NAME,
            "status": "supported_no_data",
            "nutrient_reference_overview": build_overview(
                on_date=requested,
                profile=_nutrition_reference_profile(connection),
                nutrients={},
                energy_kcal=None,
                documented_days=0,
                period_from=day,
                period_to=day,
            ),
        }
    row = connection.execute(
        """SELECT datum,kcal,protein_g,carb_g,fat_g,item_count,nutrient_json,
                  histamine_score,histamine_max,histamine_unknown_count,histamine_label
           FROM nutrition_daily_summary_v2 WHERE datum=? LIMIT 1""",
        (day,),
    ).fetchone()
    if row is None:
        return {
            "version": 1,
            "date": day,
            "timezone": TZ_NAME,
            "status": "supported_no_data",
            "nutrient_reference_overview": build_overview(
                on_date=requested,
                profile=_nutrition_reference_profile(connection),
                nutrients={},
                energy_kcal=None,
                documented_days=0,
                period_from=day,
                period_to=day,
            ),
        }
    complete = _nutrition_complete(row)
    nutrients: dict[str, dict[str, Any]] = {}
    nutrient_values: dict[str, int | float | None] = {}
    if table_exists(connection, "nutrition_item_nutrients"):
        for nutrient in connection.execute(
            """SELECT n.nutrient_key,n.value,n.unit
               FROM nutrition_item_nutrients n JOIN nutrition_items i ON i.id=n.item_id
               WHERE i.datum=?""",
            (day,),
        ):
            key = str(nutrient["nutrient_key"] or "")
            allowed = NUTRIENT_ALLOWLIST.get(key)
            normalized = normalize_nutrient(key, nutrient["value"], nutrient["unit"])
            if not allowed or normalized is None:
                continue
            value, _unit = normalized
            nutrient_values[key] = _public_number(
                float(nutrient_values.get(key) or 0) + value
            )
    else:
        try:
            raw_nutrients = json.loads(row["nutrient_json"] or "{}")
        except json.JSONDecodeError:
            raw_nutrients = {}
        if isinstance(raw_nutrients, dict):
            for key in NUTRIENT_ALLOWLIST:
                value = _public_number(raw_nutrients.get(key))
                if value is not None:
                    nutrient_values[key] = value
    for key, (label, unit) in NUTRIENT_ALLOWLIST.items():
        if key in nutrient_values:
            nutrients[key] = {
                "label": label,
                "value": nutrient_values[key],
                "food_value": nutrient_values[key],
                "supplement_value": None,
                "total_value": nutrient_values[key],
                "unit": unit,
                "food_source": "YAZIO",
                "supplement_source": "Dokumentierte Supplemente",
                "contains_estimate": False,
            }
    projected_food = _daily_nutrients(connection, requested, requested).get(day, {})
    for key, projected in projected_food.items():
        contract = NUTRIENT_CONTRACTS[key]
        nutrients[key] = {
            "label": contract.label,
            "group": contract.group,
            "value": projected.get("value"),
            "food_value": projected.get("value"),
            "yazio_value": projected.get("yazio_value"),
            "enrichment_value": projected.get("enrichment_value"),
            "supplement_value": None,
            "total_value": projected.get("value"),
            "unit": contract.unit,
            "data_status": projected.get("data_status", "unknown"),
            "status_counts": projected.get("status_counts", {}),
            "item_count": projected.get("item_count", int(row["item_count"] or 0)),
            "known_item_count": projected.get("known_item_count", 0),
            "unknown_item_count": projected.get(
                "unknown_item_count", int(row["item_count"] or 0)
            ),
            "data_completeness": projected.get("data_completeness", "partial"),
            "raw_units": projected.get("raw_units", []),
            "conversion": contract.public_conversion(),
            "food_source": "YAZIO",
            "enrichment_sources": projected.get("enrichment_sources", []),
            "supplement_source": "Dokumentierte Supplemente",
            "contains_estimate": bool(projected.get("is_estimated")),
            "contract_version": NUTRIENT_CONTRACT_VERSION,
            "null_contract_version": NULL_VALUE_CONTRACT_VERSION,
        }
    try:
        supplement_values = supplement_nutrient_totals(connection, requested, requested).get(day, {})
    except ValueError as error:
        raise APIError(422, str(error)) from error
    for key, extra in supplement_values.items():
        contract = NUTRIENT_CONTRACTS[key]
        nutrient = nutrients.setdefault(
            key,
            {
                "label": contract.label,
                "group": contract.group,
                "value": None,
                "food_value": None,
                "supplement_value": None,
                "total_value": None,
                "unit": contract.unit,
                "data_status": "not_reported",
                "status_counts": {},
                "raw_units": [],
                "conversion": contract.public_conversion(),
                "food_source": "YAZIO",
                "supplement_source": "Dokumentierte Supplemente",
                "contains_estimate": False,
                "contract_version": NUTRIENT_CONTRACT_VERSION,
                "null_contract_version": NULL_VALUE_CONTRACT_VERSION,
            },
        )
        nutrient["supplement_value"] = extra["value"]
        combined_values = [
            float(value)
            for value in (nutrient["food_value"], extra["value"])
            if value is not None
        ]
        nutrient["total_value"] = (
            _public_number(sum(combined_values)) if combined_values else None
        )
        nutrient["contains_estimate"] = bool(extra["contains_estimate"])
    supplement_day = supplements(connection, requested, requested)
    meals = []
    if table_exists(connection, "nutrition_meal_summary"):
        for meal in connection.execute(
            """SELECT meal,kcal,protein_g,carb_g,fat_g,item_count,histamine_score,histamine_label
               FROM nutrition_meal_summary WHERE datum=?
               ORDER BY CASE meal WHEN 'breakfast' THEN 1 WHEN 'lunch' THEN 2 WHEN 'dinner' THEN 3 WHEN 'snack' THEN 4 ELSE 9 END LIMIT 12""",
            (day,),
        ):
            meal_id = str(meal["meal"] or "unknown")
            meals.append({
                "meal": meal_id if meal_id in MEAL_LABELS else "unknown",
                "label": MEAL_LABELS.get(meal_id, "Unbekannt"),
                "item_count": int(meal["item_count"] or 0),
                "macros": {
                    "kcal": _public_number(meal["kcal"]),
                    "protein_g": _public_number(meal["protein_g"]),
                    "carb_g": _public_number(meal["carb_g"]),
                    "fat_g": _public_number(meal["fat_g"]),
                },
                "histamine": {
                    "score": _public_number(meal["histamine_score"]) if meal["histamine_label"] in {"classified", "green", "yellow", "orange", "red"} else None,
                    "status": "complete" if meal["histamine_label"] in {"classified", "green", "yellow", "orange", "red"} else "incomplete",
                },
            })
    existing_meals = {meal["meal"] for meal in meals}
    for meal_id, label in MEAL_LABELS.items():
        if meal_id not in existing_meals:
            meals.append({"meal": meal_id, "label": label, "item_count": 0, "macros": {"kcal": None, "protein_g": None, "carb_g": None, "fat_g": None}, "histamine": {"score": None, "status": "incomplete"}})
    meals.sort(key=lambda item: ["breakfast", "lunch", "dinner", "snack", "unassigned"].index(item["meal"]))
    tolerance_by_food: dict[str, dict[str, Any]] = {}
    composition_review_foods: set[str] = set()
    if table_exists(connection, "nutrition_mapping_provenance"):
        provenance_rows = list(
            connection.execute(
                """SELECT canonical_food,personal_tolerance_status,
                          personal_tolerance_note,ingredient_review_required
                   FROM nutrition_mapping_provenance
                   WHERE canonical_food IS NOT NULL ORDER BY id DESC LIMIT 5000"""
            )
        )
        for provenance in provenance_rows:
            food = str(provenance["canonical_food"] or "")
            status = str(provenance["personal_tolerance_status"] or "unknown")
            if not food:
                continue
            if int(provenance["ingredient_review_required"] or 0):
                composition_review_foods.add(food)
            if food not in tolerance_by_food and status in {
                "unknown",
                "documented_tolerated",
                "documented_not_tolerated",
                "unclear",
            }:
                tolerance_by_food[food] = {
                    "status": status,
                    "note": safe_metadata_text(
                        provenance["personal_tolerance_note"], 160, allow_empty=True
                    )
                    or "",
                }
    items = []
    open_mapping_names: set[str] = set()
    if table_exists(connection, "nutrition_review_queue"):
        open_mapping_names = {
            str(queue_row["normalized_name"])
            for queue_row in connection.execute(
                """SELECT normalized_name FROM nutrition_review_queue
                   WHERE status='open' AND normalized_name IS NOT NULL"""
            )
        }
    if table_exists(connection, "nutrition_items"):
        for item in connection.execute(
            """SELECT i.meal,i.name,i.brand,i.amount,i.amount_unit,i.kcal,i.protein_g,i.carb_g,i.fat_g,i.source,
                      h.canonical_food,h.sighi_score,h.traffic_light,h.confidence
               FROM nutrition_items i LEFT JOIN nutrition_histamine_scores h ON h.item_id=i.id
               WHERE i.datum=? ORDER BY i.meal,i.id LIMIT ?""",
            (day, MAX_NUTRITION_ITEMS + 1),
        ):
            if len(items) >= MAX_NUTRITION_ITEMS:
                raise APIError(422, "row_limit_exceeded")
            name = safe_metadata_text(item["name"], 120, allow_empty=False)
            brand = safe_metadata_text(item["brand"], 80, allow_empty=True)
            canonical = safe_metadata_text(item["canonical_food"], 120, allow_empty=True)
            if name is None or brand is None or canonical is None:
                continue
            normalized_name = re.sub(
                r"\s+",
                " ",
                re.sub(
                    r"[^a-z0-9]+",
                    " ",
                    name.casefold().translate(
                        str.maketrans(
                            {"ä": "ae", "ö": "oe", "ü": "ue", "é": "e", "è": "e", "à": "a", "ß": "ss"}
                        )
                    ),
                ),
            ).strip()
            mapped = item["sighi_score"] in (0, 1, 2, 3)
            items.append({
                "meal": item["meal"] if item["meal"] in MEAL_LABELS else "unassigned",
                "name": name,
                "brand": brand,
                "amount": _public_number(item["amount"]),
                "amount_unit": item["amount_unit"] if item["amount_unit"] in {"g", "ml", "piece", "portion"} else "unknown",
                "macros": {
                    "kcal": _public_number(item["kcal"]),
                    "protein_g": _public_number(item["protein_g"]),
                    "carb_g": _public_number(item["carb_g"]),
                    "fat_g": _public_number(item["fat_g"]),
                },
                "histamine": {
                    "canonical_food": canonical or None,
                    "score": int(item["sighi_score"]) if mapped else None,
                    "status": "mapped" if mapped else "unmapped",
                    "confidence": item["confidence"] if item["confidence"] in {"low", "medium", "high"} else "low",
                },
                "personal_tolerance": tolerance_by_food.get(
                    canonical, {"status": "unknown", "note": ""}
                ),
                "composition_review_required": canonical in composition_review_foods,
                "mapping_queue_key": nutrition_queue_key(normalized_name)
                if not mapped and normalized_name in open_mapping_names
                else None,
                "source": "YAZIO"
                if item["source"] == "yazio_api"
                else "Weitere dokumentierte Ernährungsquelle",
            })
    for nutrient in nutrients.values():
        nutrient.setdefault("item_count", int(row["item_count"] or 0))
        nutrient.setdefault("known_item_count", 0)
        nutrient.setdefault("unknown_item_count", int(row["item_count"] or 0))
    reference_overview = build_overview(
        on_date=requested,
        profile=_nutrition_reference_profile(connection),
        nutrients=nutrients,
        energy_kcal=_public_number(row["kcal"]),
        documented_days=1,
        period_from=day,
        period_to=day,
        contributions=_nutrition_contributions(connection, requested),
    )
    provenance_row = connection.execute(
        """SELECT MAX(imported_at) AS last_import,
                  COUNT(DISTINCT CASE WHEN source='yazio_api' THEN 'YAZIO' ELSE 'other' END) AS source_count,
                  MIN(CASE WHEN source='yazio_api' THEN 1 ELSE 0 END) AS only_yazio
           FROM nutrition_items WHERE datum=?""",
        (day,),
    ).fetchone()
    source_label = (
        "YAZIO"
        if provenance_row and int(provenance_row["only_yazio"] or 0) == 1
        else "Mehrere dokumentierte Ernährungsquellen"
    )
    last_import = (
        safe_metadata_text(provenance_row["last_import"], 40, allow_empty=True)
        if provenance_row
        else None
    )
    return {
        "version": 2,
        "date": day,
        "timezone": TZ_NAME,
        "status": "documented",
        "definition": _histamine_definition(),
        "coverage": {
            "status": "complete" if complete else "incomplete",
            "item_count": int(row["item_count"] or 0),
            "unknown_items": int(row["histamine_unknown_count"] or 0),
            "mapped_items": max(0, int(row["item_count"] or 0) - int(row["histamine_unknown_count"] or 0)),
            "open_items": int(row["histamine_unknown_count"] or 0),
            "composite_products": sum(
                1 for item in items if item["composition_review_required"]
            ),
            "documented_nutrient_fields": sum(
                1
                for nutrient in nutrients.values()
                if nutrient.get("data_status") in {"documented_value", "documented_zero", "estimated"}
            ),
            "unknown_nutrient_fields": sum(
                1 for nutrient in nutrients.values() if nutrient.get("data_status") == "unknown"
            ),
            "released_nutrient_fields": len(NUTRIENT_CONTRACTS),
            "nutrient_coverage": round(
                sum(
                    1
                    for nutrient in nutrients.values()
                    if nutrient.get("data_status") in {"documented_value", "documented_zero", "estimated"}
                ) / len(NUTRIENT_CONTRACTS),
                3,
            ),
            "mapping_coverage": round((int(row["item_count"] or 0) - int(row["histamine_unknown_count"] or 0)) / int(row["item_count"] or 1), 3) if int(row["item_count"] or 0) else 0,
            "insufficient_inputs": None if complete else "no_index",
        },
        "provenance": {
            "source": source_label,
            "last_successful_import": last_import,
            "item_count": int(row["item_count"] or 0),
            "mapping_coverage": round((int(row["item_count"] or 0) - int(row["histamine_unknown_count"] or 0)) / int(row["item_count"] or 1), 3) if int(row["item_count"] or 0) else 0,
            "open_items": int(row["histamine_unknown_count"] or 0),
            "review_status": "technically_complete" if complete else "review_required",
        },
        "macros": {
            "kcal": _public_number(row["kcal"]),
            "protein_g": _public_number(row["protein_g"]),
            "carb_g": _public_number(row["carb_g"]),
            "fat_g": _public_number(row["fat_g"]),
        },
        "nutrients": nutrients,
        "nutrient_reference_overview": reference_overview,
        "histamine": {
            "name": "Histamin-Zuordnungsindex",
            "score": _public_number(row["histamine_score"]) if complete else None,
            "mapped_score_sum": (
                _public_number(row["histamine_score"])
                if int(row["item_count"] or 0) > int(row["histamine_unknown_count"] or 0)
                else None
            ),
            "max_score": (
                int(row["histamine_max"] or 0)
                if int(row["item_count"] or 0) > int(row["histamine_unknown_count"] or 0)
                else None
            ),
            "unknown_items": int(row["histamine_unknown_count"] or 0),
            "definition_version": _histamine_definition()["version"],
            "method_note": (
                "Der Index fasst dokumentierte SIGHi-Zuordnungen zusammen. Er misst "
                "nicht den tatsächlichen Histamingehalt einer Mahlzeit und ist keine "
                "individuelle Verträglichkeitsaussage."
            ),
        },
        "meals": meals,
        "items": items,
        "supplements": supplement_day,
        "supplements_documented": (
            "yes"
            if supplement_day.get("administered")
            else "no"
            if supplement_day.get("missed")
            else "unknown"
        ),
        "nutrient_contract_version": NUTRIENT_CONTRACT_VERSION,
        "null_contract_version": NULL_VALUE_CONTRACT_VERSION,
        "nutrition_statement": (
            "Dokumentierte Aufnahme aus YAZIO und Supplementen bleibt nach Quelle getrennt; "
            "eine dokumentierte Aufnahme ist keine Mangeldiagnose oder Dosierungsempfehlung."
        ),
    }


def _nutrition_mapping_queue(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    status_aliases = {
        "open": "open",
        "conflicts": "conflict",
        "deferred": "deferred",
        "confirmed": "confirmed",
        "all": "all",
    }
    status = status_aliases.get(params.get("status", "open"))
    if status is None:
        raise APIError(400, "status_not_allowed")
    search = params.get("search", "")
    if len(search) > 120 or re.search(r"[\x00-\x1f\x7f]", search):
        raise APIError(400, "invalid_search")
    try:
        result = build_mapping_review(connection, status=status, search=search)
    except ValueError as error:
        raise APIError(400, str(error)) from error
    except ReviewLimitError as error:
        raise APIError(422, "row_limit_exceeded") from error
    except UnsafeReviewMetadataError as error:
        raise APIError(422, "unsafe_mapping_metadata") from error
    if len(result["groups"]) > 5_000:
        raise APIError(422, "row_limit_exceeded")
    return {
        **result,
        "items": result["groups"],
        "timezone": TZ_NAME,
        "definition": _histamine_definition(),
        "action_contract": "nutrition_mapping_action_v2",
        "allowed_decisions": [
            "assign", "composite", "defer", "not_assignable",
            "irrelevant", "conflict", "reopen",
        ],
    }


def _day(connection: sqlite3.Connection, day: date) -> dict[str, Any]:
    requested = day.isoformat()
    today = local_today().isoformat()
    measurements = []
    for metric in BY_ID_V2.values():
        if metric.value_type == "panel" or metric.source in {
            "laborwerte",
            "symptom_log",
            "nutrition_daily_summary_v2",
        }:
            continue
        points = _metric_points(connection, metric, day, day)
        if not points:
            continue
        point = points[-1]
        measurements.append(
            {
                "metric": metric.id,
                "label": metric.label,
                "value": point["value"],
                "unit": metric.unit,
                "source": metric.source,
                "aggregation": metric.aggregation,
                "quality": point["quality"],
            }
        )
    symptoms = _day_symptoms(connection, requested)
    medications = _day_medications(connection, requested)
    try:
        day_supplements = supplements(connection, day, day)
    except ValueError as error:
        raise APIError(422, str(error)) from error
    day_events = _day_events(connection, requested)
    labs = [
        row
        for row in _verified_lab_rows(connection, include_missing_reference=True)
        if row["date"] == requested
    ]
    nutrition = _day_nutrition(connection, requested)
    documents = _day_documents(connection, requested)
    appointments = _day_appointments(connection, requested)
    counts = {
        "measurements": len(measurements),
        "symptoms": symptoms["documented_dimensions"] + symptoms["additional_count"],
        "medications": sum(len(items) for items in medications.values()),
        "supplements": sum(
            len(day_supplements.get(kind, []))
            for kind in ("planned", "administered", "missed", "corrected")
        ),
        "labs": len(labs),
        "events": len(day_events["events"]) + len(day_events["periods"]),
        "documents": len(documents),
        "nutrition": 1 if nutrition["status"] == "documented" else 0,
        "appointments": len(appointments["items"]),
    }
    categories = [name for name, count in counts.items() if count]
    metrics = {
        item["metric"]: {
            "value": item["value"],
            "unit": item["unit"],
            "aggregation": item["aggregation"],
            "quality": item["quality"],
        }
        for item in measurements
    }
    notices = []
    if not categories:
        notices.append("Für diesen Tag ist nichts dokumentiert.")
    if symptoms["documented_dimensions"] and not symptoms["complete"]:
        notices.append("Die Symptomdokumentation dieses Tages ist unvollständig.")
    if any(item.get("reference_status") == "missing_reference" for item in labs):
        notices.append(
            "Für mindestens eine Laborbeobachtung fehlt ein verifizierter Referenzbereich."
        )
    return {
        "version": 2,
        "date": requested,
        "today": today,
        "timezone": TZ_NAME,
        "is_today": requested == today,
        "is_future": requested > today,
        "coverage": {
            "status": "documented" if categories else "not_documented",
            "categories": categories,
            "counts": counts,
            "symptoms_complete": symptoms["complete"],
        },
        "measurements": measurements,
        "metrics": metrics,
        "symptoms": symptoms,
        "medications": medications,
        "supplements": day_supplements,
        "events": day_events,
        "labs": labs,
        "nutrition": nutrition,
        "documents": documents,
        "appointments": appointments,
        "notices": notices[:3],
    }


def _calendar(connection: sqlite3.Connection, params: dict[str, str]) -> dict[str, Any]:
    raw_start = params.get("from")
    raw_end = params.get("to")
    if not raw_start or not raw_end:
        raise APIError(400, "range_required")
    start = parse_iso_day(raw_start)
    end = parse_iso_day(raw_end)
    if end < start:
        raise APIError(400, "range_reversed")
    days = (end - start).days + 1
    if days > MAX_CALENDAR_DAYS:
        raise APIError(422, "calendar_range_too_large")
    today = local_today()
    if end > today + timedelta(days=MAX_FUTURE_DAYS):
        raise APIError(422, "future_range_too_large")
    dates = [(start + timedelta(days=offset)).isoformat() for offset in range(days)]
    counts_by_day = {
        day: {
            "measurements": 0,
            "symptoms": 0,
            "medications": 0,
            "supplements": 0,
            "labs": 0,
            "events": 0,
            "documents": 0,
            "nutrition": 0,
            "appointments": 0,
        }
        for day in dates
    }

    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)

    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
_RECORD_CURSOR_KEY = secrets.token_bytes(32)
_RECORD_CURSOR_VERSION = 1
_RECORD_SORT_SPECS = {
    "document_date_desc": ("document_date", "desc", "desc"),
    "document_date_asc": ("document_date", "asc", "asc"),
    "import_date_desc": ("upload_datum", "desc", "desc"),
    "category": ("kategorie", "asc", "desc"),
    "institution": ("institution", "asc", "desc"),
    "type": ("daten_typ", "asc", "desc"),
    "review_status": ("review_status", "asc", "desc"),
}


def _cursor_binding(params: dict[str, str], *, search: bool) -> str:
    bound: dict[str, Any] = {
        key: params.get(key, "")
        for key in (
            "from",
            "to",
            "category",
            "institution",
            "type",
            "review_status",
            "q",
            "sort",
        )
    }
    bound["search"] = search
    return hashlib.sha256(
        json.dumps(bound, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()


def _encode_record_cursor(payload: dict[str, Any]) -> str:
    raw = json.dumps(
        {"v": _RECORD_CURSOR_VERSION, **payload},
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    signature = hmac.new(_RECORD_CURSOR_KEY, raw, hashlib.sha256).digest()[:16]
    return "record-cursor-" + base64.urlsafe_b64encode(raw + signature).decode(
        "ascii"
    ).rstrip("=")


def _decode_record_cursor(value: str, binding: str) -> dict[str, Any]:
    if not value.startswith("record-cursor-") or len(value) > 600:
        raise APIError(400, "invalid_cursor")
    encoded = value.removeprefix("record-cursor-")
    try:
        packed = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
        raw, supplied = packed[:-16], packed[-16:]
        expected = hmac.new(_RECORD_CURSOR_KEY, raw, hashlib.sha256).digest()[:16]
        payload = json.loads(raw)
    except (ValueError, UnicodeError, json.JSONDecodeError) as error:
        raise APIError(400, "invalid_cursor") from error
    if not hmac.compare_digest(supplied, expected):
        raise APIError(400, "invalid_cursor")
    if payload.get("v") != _RECORD_CURSOR_VERSION or payload.get("b") != binding:
        raise APIError(400, "cursor_mismatch")
    if not isinstance(payload.get("id"), int) or payload["id"] < 1:
        raise APIError(400, "invalid_cursor")
    return payload


def _record_page_size(params: dict[str, str]) -> int:
    raw = params.get("limit", str(RECORD_PAGE_SIZE))
    if not raw.isdigit():
        raise APIError(400, "invalid_limit")
    size = int(raw)
    if not 1 <= size <= RECORD_MAX_PAGE_SIZE:
        raise APIError(400, "invalid_limit")
    return size


def _record_document_rows(
    connection: sqlite3.Connection, params: dict[str, str]
) -> dict[str, Any]:
    """Apply document filters and FTS/review join in SQLite before page limiting."""
    if not table_exists(connection, "dokumente"):
        return {
            "documents": [], "next_cursor": None, "truncated": False, "total": 0,
            "coverage": {"total": 0, "original_available": 0, "original_missing": 0,
                         "original_unsupported": 0, "original_blocked": 0,
                         "original_too_large": 0, "original_unavailable": 0,
                         "text_extracted": 0, "content_reviewed": 0,
                         "fulltext_searchable": 0},
        }
    fts_document_ids: set[int] = set()
    if table_exists(connection, "health_document_fts"):
        fts_document_ids = {
            int(row[0])
            for row in connection.execute(
                "SELECT DISTINCT document_id FROM health_document_fts"
            )
            if str(row[0]).isdigit()
        }
    coverage_rows = connection.execute(
        """SELECT id,review_status,extrahierte_inhalte,local_original_path,dateipfad
           FROM dokumente ORDER BY id LIMIT ?""",
        (RECORD_MAX_ROWS * 20 + 1,),
    ).fetchall()
    if len(coverage_rows) > RECORD_MAX_ROWS * 20:
        raise APIError(422, "source_row_limit_exceeded")
    original_counts = {
        "available": 0,
        "missing": 0,
        "unsupported": 0,
        "blocked": 0,
        "too_large": 0,
    }
    for coverage_row in coverage_rows:
        original = probe_original(
            coverage_row["local_original_path"] or coverage_row["dateipfad"]
        )
        try:
            if original.status in original_counts:
                original_counts[original.status] += 1
            else:
                original_counts["blocked"] += 1
        finally:
            original.close()
    document_coverage = {
        "total": len(coverage_rows),
        "original_available": original_counts["available"],
        "original_missing": original_counts["missing"],
        "original_unsupported": original_counts["unsupported"],
        "original_blocked": original_counts["blocked"],
        "original_too_large": original_counts["too_large"],
        "original_unavailable": sum(
            original_counts[key]
            for key in ("missing", "unsupported", "blocked", "too_large")
        ),
        "text_extracted": sum(
            bool(str(row["extrahierte_inhalte"] or "").strip())
            for row in coverage_rows
        ),
        "content_reviewed": sum(
            row["review_status"] == "geprueft" for row in coverage_rows
        ),
        "fulltext_searchable": sum(
            row["review_status"] == "geprueft"
            and int(row["id"]) in fts_document_ids
            for row in coverage_rows
        ),
    }
    start, end = parse_range(params)
    filters = ["1=1"]
    values: list[Any] = []
    if start:
        filters.append("d.document_date>=?")
        values.append(start.isoformat())
    if end:
        filters.append("d.document_date<=?")
        values.append(end.isoformat())
    for key, column, maximum in (
        ("category", "kategorie", 80),
        ("institution", "institution", 120),
        ("type", "daten_typ", 20),
        ("review_status", "review_status", 32),
    ):
        raw = params.get(key)
        if raw:
            clean = safe_metadata_text(raw, maximum, allow_empty=False)
            if clean is None:
                raise APIError(400, "filter_not_allowed")
            filters.append(f"d.{column}=?")
            values.append(clean)
    queue=params.get("queue")
    queue_clauses={
      "now_reviewable":"r.queue_bucket='now_reviewable'","original_missing":"r.queue_bucket='original_missing'",
      "conflict":"r.conflict_count>0","duplicate":"r.byte_duplicate_of IS NOT NULL","completed":"r.queue_bucket='no_action' AND r.byte_duplicate_of IS NULL",
      "laboratory":"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":"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":"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":"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'))",
    }
    if queue:
        if queue not in queue_clauses or not table_exists(connection,"document_reconciliation"):raise APIError(400,"filter_not_allowed")
        filters.append(f"EXISTS(SELECT 1 FROM document_reconciliation r WHERE r.document_id=d.id AND {queue_clauses[queue]})")
    sort = params.get("sort", "document_date_desc")
    if sort not in RECORD_DOCUMENT_SORTS:
        raise APIError(400, "sort_not_allowed")
    page_size = _record_page_size(params)
    query_text = params.get("q")
    search = bool(query_text)
    binding = _cursor_binding(params, search=search)
    cursor_payload = (
        _decode_record_cursor(params["cursor"], binding)
        if params.get("cursor")
        else None
    )
    join = ""
    select_snippet = "NULL AS snippet"
    select_rank = "NULL AS search_rank"
    order = RECORD_DOCUMENT_SORTS[sort].replace("id", "d.id")
    query_values = list(values)
    search_cursor_clause = ""
    search_cursor_values: list[Any] = []
    if search:
        if not table_exists(connection, "health_document_fts"):
            raise APIError(503, "fts_unavailable")
        tokens = _fts_tokens(query_text or "")
        join = " JOIN health_document_fts f ON CAST(d.id AS TEXT)=f.document_id"
        filters.append("d.review_status='geprueft'")
        filters.append("health_document_fts MATCH ?")
        query_values.append(_fts_literal_query(tokens))
        select_snippet = "snippet(health_document_fts, 6, '', '', '…', ?) AS snippet"
        select_rank = "bm25(health_document_fts) AS search_rank"
        order = "search_rank ASC, d.id DESC"
        if cursor_payload:
            rank = cursor_payload.get("k")
            if not isinstance(rank, (int, float)):
                raise APIError(400, "invalid_cursor")
            search_cursor_clause = "WHERE (search_rank>? OR (search_rank=? AND id<?))"
            search_cursor_values.extend(
                (float(rank), float(rank), cursor_payload["id"])
            )
    else:
        column, primary_direction, id_direction = _RECORD_SORT_SPECS[sort]
        if cursor_payload:
            primary = cursor_payload.get("k")
            if not isinstance(primary, str):
                raise APIError(400, "invalid_cursor")
            primary_operator = ">" if primary_direction == "asc" else "<"
            id_operator = ">" if id_direction == "asc" else "<"
            collation = " COLLATE NOCASE" if sort in {"category", "institution"} else ""
            filters.append(
                f"(COALESCE(d.{column},''){collation}{primary_operator}? OR "
                f"(COALESCE(d.{column},''){collation}=?{collation} AND d.id{id_operator}?))"
            )
            query_values.extend((primary, primary, cursor_payload["id"]))
    where = " AND ".join(filters)
    if search:
        sql = f"""WITH raw_hits AS MATERIALIZED (
                        SELECT d.id,d.document_date,d.upload_datum,d.kategorie,
                               d.institution,d.daten_typ,d.review_status,
                               d.processing_quality,d.groessekbytes,d.extrahierte_inhalte,
                               d.local_original_path,d.dateipfad,f.chunk_no,
                               {select_snippet},{select_rank}
                        FROM dokumente d{join}
                        WHERE {where}
                    ), best_rank AS (
                        SELECT id,MIN(search_rank) AS search_rank
                        FROM raw_hits GROUP BY id
                    ), selected AS (
                        SELECT r.* FROM raw_hits r
                        JOIN best_rank b ON b.id=r.id AND b.search_rank=r.search_rank
                        WHERE r.chunk_no=(
                            SELECT MIN(r2.chunk_no) FROM raw_hits r2
                            WHERE r2.id=r.id AND r2.search_rank=r.search_rank
                        )
                    )
                    SELECT * FROM selected {search_cursor_clause}
                    ORDER BY search_rank ASC,id DESC LIMIT ?"""
        parameters = [
            RECORD_MAX_SNIPPET_CHARS,
            *query_values,
            *search_cursor_values,
            page_size + 1,
        ]
    else:
        sql = f"""SELECT d.id,d.document_date,d.upload_datum,d.kategorie,d.institution,
                         d.daten_typ,d.review_status,d.processing_quality,d.groessekbytes,
                         d.extrahierte_inhalte,d.local_original_path,d.dateipfad,
                         {select_snippet},{select_rank}
                  FROM dokumente d{join}
                  WHERE {where}
                  ORDER BY {order} LIMIT ?"""
        parameters = [*query_values, page_size + 1]
    try:
        rows = connection.execute(sql, parameters).fetchall()
    except sqlite3.OperationalError as error:
        if search:
            raise APIError(400, "query_not_allowed") from error
        raise
    has_more = len(rows) > page_size
    result: list[dict[str, Any]] = []
    for row in rows[:page_size]:
        category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
        institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
        if category is None or institution is None:
            continue
        document_day = parse_day(row["document_date"])
        reviewed = row["review_status"] == "geprueft"
        content_status = (
            "extracted"
            if str(row["extrahierte_inhalte"] or "").strip()
            else "not_extracted"
        )
        original = probe_original(row["local_original_path"] or row["dateipfad"])
        original_status = original.status
        original.close()
        media_kind = None
        if table_exists(connection, "document_media"):
            media_row = connection.execute("SELECT media_kind FROM document_media WHERE document_id=?", (int(row["id"]),)).fetchone()
            media_kind = str(media_row[0]) if media_row else None
        result.append(
            {
                "id": api_document_id(row["id"], document_day, category, institution),
                "title": f"Dokument · {category or 'Ohne Kategorie'}",
                "document_date": document_day,
                "document_date_status": "known" if document_day else "unknown",
                "import_date": parse_day(row["upload_datum"]),
                "category": category,
                "institution": institution,
                "type": "video" if media_kind == "video" else (
                    row["daten_typ"] if row["daten_typ"] in {"pdf", "image"} else "unknown"
                ),
                "review_status": row["review_status"]
                if row["review_status"] in {"geprueft", "nicht_geprueft"}
                else "unknown",
                "content_status": content_status,
                "extraction_status": content_status,
                "original_status": original_status,
                "search_status": (
                    "reviewed_searchable"
                    if reviewed and int(row["id"]) in fts_document_ids
                    else "extracted_not_reviewed"
                    if content_status == "extracted" and not reviewed
                    else "not_searchable"
                ),
                "processing_quality": safe_metadata_text(
                    row["processing_quality"], 40, allow_empty=True
                ),
                "size_kb": int(row["groessekbytes"])
                if isinstance(row["groessekbytes"], int)
                and 0 <= row["groessekbytes"] <= 2_000_000
                else None,
                "snippet": str(row["snippet"])[:RECORD_MAX_SNIPPET_CHARS]
                if row["snippet"]
                else None,
            }
        )
        if table_exists(connection, "document_processing"):
            processing = connection.execute(
                """SELECT intake_id,original_status,extraction_status,content_status,search_status,
                          transfer_status,page_count,engine,engine_version,duplicate_document_id,
                          investigation_day,personal_title,last_error_code
                   FROM document_processing WHERE document_id=?""",
                (int(row["id"]),),
            ).fetchone()
            if processing is not None:
                item = result[-1]
                item.update({
                    "review_id": str(processing["intake_id"]),
                    "original_status": str(processing["original_status"]),
                    "extraction_status": str(processing["extraction_status"]),
                    "content_review_status": str(processing["content_status"]),
                    "search_status": str(processing["search_status"]),
                    "transfer_status": str(processing["transfer_status"]),
                    "page_count": processing["page_count"],
                    "engine": safe_metadata_text(processing["engine"], 80, allow_empty=True),
                    "engine_version": safe_metadata_text(processing["engine_version"], 80, allow_empty=True),
                    "duplicate_candidate": processing["duplicate_document_id"] is not None,
                    "investigation_day": parse_day(processing["investigation_day"]),
                    "personal_title": safe_metadata_text(processing["personal_title"], 120, allow_empty=True),
                    "retry_available": processing["extraction_status"] == "failed",
                })
                item["candidate_count"] = int(connection.execute(
                    "SELECT COUNT(*) FROM document_candidates WHERE document_id=? AND status IN ('open','conflicting')",
                    (int(row["id"]),),
                ).fetchone()[0])
        if table_exists(connection,"document_reconciliation"):
            reconciliation=connection.execute("SELECT queue_bucket,reason_code,priority,open_decisions,exact_match_count,conflict_count,byte_duplicate_of FROM document_reconciliation WHERE document_id=?",(int(row["id"]),)).fetchone()
            if reconciliation is not None:
                result[-1].update({"queue_bucket":str(reconciliation["queue_bucket"]),"queue_reason":str(reconciliation["reason_code"]),"priority":int(reconciliation["priority"]),"open_decisions":int(reconciliation["open_decisions"]),"exact_match_count":int(reconciliation["exact_match_count"]),"conflict_count":int(reconciliation["conflict_count"]),"byte_duplicate":reconciliation["byte_duplicate_of"] is not None})
    next_cursor = None
    if has_more and result:
        last_row = rows[page_size - 1]
        if search:
            cursor_key: str | float = float(last_row["search_rank"])
        else:
            cursor_column = _RECORD_SORT_SPECS[sort][0]
            cursor_key = str(last_row[cursor_column] or "")
        next_cursor = _encode_record_cursor(
            {"b": binding, "k": cursor_key, "id": int(last_row["id"])}
        )
    return {
        "documents": result,
        "next_cursor": next_cursor,
        "truncated": has_more,
        "total": None,
        "coverage": document_coverage,
    }


def _fts_tokens(raw: str) -> list[str]:
    clean = " ".join(raw.split())
    if not 2 <= len(clean) <= 80:
        raise APIError(400, "query_length_invalid")
    tokens = re.findall(r"[A-Za-zÀ-ÿ0-9]{2,32}", clean)
    if not tokens or len(tokens) > 8 or " ".join(tokens).casefold() != clean.casefold():
        raise APIError(400, "query_not_allowed")
    return tokens


def _fts_literal_query(tokens: list[str]) -> str:
    """Construct only quoted FTS phrases; user input never supplies FTS syntax."""
    return " AND ".join(f'"{token.replace(chr(34), chr(34) * 2)}"' for token in tokens)


def _resolve_record_document(
    connection: sqlite3.Connection, opaque: str, *, reviewed: bool = False
) -> sqlite3.Row:
    if not re.fullmatch(r"api-document-[a-f0-9]{24}", opaque):
        raise APIError(404, "document_not_found")
    sql = "SELECT id,document_date,kategorie,institution,daten_typ,review_status,processing_quality,groessekbytes,datei_hash,upload_datum,extrahierte_inhalte,local_original_path,dateipfad FROM dokumente"
    if reviewed:
        sql += " WHERE review_status='geprueft'"
    for row in connection.execute(sql + " ORDER BY id LIMIT 10000"):
        category = safe_metadata_text(row["kategorie"], 80, allow_empty=True)
        institution = safe_metadata_text(row["institution"], 120, allow_empty=True)
        if (
            category is not None
            and institution is not None
            and api_document_id(
                row["id"], parse_day(row["document_date"]), category, institution
            )
            == opaque
        ):
            return row
    raise APIError(404, "document_not_found")


def _record_document_detail(
    connection: sqlite3.Connection, opaque: str, params: dict[str, str]
) -> dict[str, Any]:
    row = _resolve_record_document(connection, opaque, reviewed=True)
    category = safe_metadata_text(row["kategorie"], 80, allow_empty=True) or ""
    institution = safe_metadata_text(row["institution"], 120, allow_empty=True) or ""
    document_day = parse_day(row["document_date"])
    all_sections = document_chunks(row["extrahierte_inhalte"])
    content_status = "available" if all_sections else "no_extracted_content"
    original = probe_original(
        row["local_original_path"] or row["dateipfad"], reviewed=True
    )
    original_status = original.status
    original.close()
    media_kind = None
    if table_exists(connection, "document_media"):
        media_row = connection.execute("SELECT media_kind FROM document_media WHERE document_id=?", (int(row["id"]),)).fetchone()
        media_kind = str(media_row[0]) if media_row else None
    try:
        after = decode_chunk_cursor(params["cursor"]) if params.get("cursor") else 0
    except ValueError as error:
        raise APIError(400, "invalid_cursor") from error
    page_size = _record_page_size(params)
    selected = [section for section in all_sections if section.number > after]
    has_more = len(selected) > page_size
    sections = [
        {"number": section.number, "text": section.text}
        for section in selected[:page_size]
    ]
    return {
        "id": opaque,
        "title": f"Dokument · {category or 'Ohne Kategorie'}",
        "document_date": document_day,
        "document_date_status": "known" if document_day else "unknown",
        "category": category,
        "institution": institution,
        "type": "video" if media_kind == "video" else (row["daten_typ"] if row["daten_typ"] in {"pdf", "image"} else "unknown"),
        "review_status": "geprueft",
        "content_status": content_status,
        "original_status": original_status,
        "sections": sections,
        "section_total": len(all_sections),
        "next_cursor": encode_chunk_cursor(sections[-1]["number"])
        if has_more and sections
        else None,
        "truncated": has_more,
        "original_available": original_status == "available",
    }


def _record_document_extracted_preview(
    connection: sqlite3.Connection, opaque: str, params: dict[str, str]
) -> dict[str, Any]:
    """Return machine-extracted text without granting review, FTS or report status."""
    row = _resolve_record_document(connection, opaque, reviewed=False)
    try:
        after = decode_chunk_cursor(params["cursor"]) if params.get("cursor") else 0
    except ValueError as error:
        raise APIError(400, "invalid_cursor") from error
    page_size = _record_page_size(params)
    all_sections = document_chunks(row["extrahierte_inhalte"])
    if not all_sections:
        raise APIError(404, "extracted_content_not_available")
    selected = [section for section in all_sections if section.number > after]
    has_more = len(selected) > page_size
    sections = [
        {"number": section.number, "text": section.text}
        for section in selected[:page_size]
    ]
    reviewed = row["review_status"] == "geprueft"
    return {
        "id": opaque,
        "extraction_status": "extracted",
        "review_status": "geprueft" if reviewed else "nicht_geprueft",
        "search_status": "reviewed_searchable" if reviewed else "extracted_not_reviewed",
        "notice": (
            "Maschinell extrahierter Inhalt – inhaltlich geprüft"
            if reviewed
            else "Maschinell extrahierter Inhalt – noch nicht inhaltlich geprüft"
        ),
        "sections": sections,
        "section_total": len(all_sections),
        "next_cursor": encode_chunk_cursor(sections[-1]["number"])
        if has_more and sections
        else None,
        "truncated": has_more,
    }


def _record_document_matches(
    connection: sqlite3.Connection, opaque: str, params: dict[str, str]
) -> dict[str, Any]:
    row = _resolve_record_document(connection, opaque, reviewed=True)
    tokens = _fts_tokens(params.get("q", ""))
    if not table_exists(connection, "health_document_fts"):
        raise APIError(503, "fts_unavailable")
    fts_query = _fts_literal_query(tokens)
    matches: list[dict[str, Any]] = []
    rows = connection.execute(
        """SELECT f.chunk_no,f.content
           FROM health_document_fts f
           JOIN dokumente d ON CAST(d.id AS TEXT)=f.document_id
           WHERE d.id=? AND d.review_status='geprueft'
             AND health_document_fts MATCH ?
           ORDER BY f.chunk_no LIMIT ?""",
        (int(row["id"]), fts_query, RECORD_MAX_MATCHES + 1),
    )
    for match in list(rows)[:RECORD_MAX_MATCHES]:
        text = str(match["content"])
        position = min(
            (
                text.casefold().find(token.casefold())
                for token in tokens
                if text.casefold().find(token.casefold()) >= 0
            ),
            default=-1,
        )
        if position >= 0:
            matches.append(
                {
                    "section": int(match["chunk_no"]),
                    "start": position,
                    "end": position + len(tokens[0]),
                    "snippet": text[max(0, position - 100) : position + 220],
                }
            )
    return {
        "id": opaque,
        "query": " ".join(tokens),
        "matches": matches,
        "next_cursor": None,
        "truncated": len(matches) == RECORD_MAX_MATCHES,
        "total": None,
    }


def _document_review_workspace(connection: sqlite3.Connection, opaque: str) -> dict[str, Any]:
    row = _resolve_record_document(connection, opaque, reviewed=False)
    if not table_exists(connection, "document_processing"):
        raise APIError(404, "review_workspace_not_available")
    processing = connection.execute(
        "SELECT * FROM document_processing WHERE document_id=?", (int(row["id"]),)
    ).fetchone()
    if processing is None:
        raise APIError(404, "review_workspace_not_available")
    pages = []
    if table_exists(connection, "document_pages"):
        latest = connection.execute(
            "SELECT COALESCE(MAX(text_version),0) FROM document_pages WHERE document_id=?",
            (int(row["id"]),),
        ).fetchone()[0]
        page_sql=("""SELECT p.page_number,p.original_text,p.confidence,p.repetition_status,p.reviewed_at,
                      r.compared_document_id,r.compared_page_number,r.relation,r.similarity
               FROM document_pages p LEFT JOIN document_section_relations r ON r.document_id=p.document_id AND r.page_number=p.page_number
               WHERE p.document_id=? AND p.text_version=? ORDER BY p.page_number LIMIT 200""" if table_exists(connection,"document_section_relations") else """SELECT page_number,original_text,confidence,repetition_status,reviewed_at,NULL AS compared_document_id,NULL AS compared_page_number,NULL AS relation,NULL AS similarity FROM document_pages WHERE document_id=? AND text_version=? ORDER BY page_number LIMIT 200""")
        for page in connection.execute(page_sql,
            (int(row["id"]), latest),
        ):
            differences=[]
            if page["relation"]=="near_match" and page["compared_document_id"] is not None:
                prior=connection.execute("SELECT original_text FROM document_pages WHERE document_id=? AND page_number=? AND text_version=(SELECT MAX(version) FROM document_text_versions WHERE document_id=?)",(int(page["compared_document_id"]),int(page["compared_page_number"]),int(page["compared_document_id"]))).fetchone()
                if prior:differences=changed_fragments(str(prior[0]),str(page["original_text"]))
            pages.append({
                "page": int(page["page_number"]),
                "section": int(page["page_number"]),
                "text": str(page["original_text"]),
                "confidence": page["confidence"],
                "repetition_status": str(page["repetition_status"]),
                "similarity": round(float(page["similarity"]),3) if page["similarity"] is not None else None,
                "differences": differences,
                "reviewed": page["reviewed_at"] is not None,
            })
    candidates = []
    if table_exists(connection, "document_candidates"):
        rich=table_exists(connection,"document_candidate_matches")
        candidate_sql=("""SELECT c.id,c.candidate_type,c.value_text,c.unit,c.page_number,c.section_number,c.context_text,c.engine,c.confidence,c.status,c.corrected_value_text,c.corrected_unit,
                      m.match_status,m.target_area,m.normalized_parameter,m.normalized_date,m.normalized_value,m.normalized_unit,m.reference_text,m.existing_database_value,m.existing_workbook_value,
                      s.status AS staging_status,s.operation AS staging_operation,s.old_value_json,s.new_value_json
               FROM document_candidates c LEFT JOIN document_candidate_matches m ON m.candidate_id=c.id LEFT JOIN document_transfer_staging s ON s.candidate_id=c.id
               WHERE c.document_id=? AND c.source_text_version=(SELECT MAX(version) FROM document_text_versions WHERE document_id=c.document_id) ORDER BY c.page_number,c.section_number,c.id LIMIT 500""" if rich else """SELECT id,candidate_type,value_text,unit,page_number,section_number,context_text,engine,confidence,status,corrected_value_text,corrected_unit,NULL AS match_status,NULL AS target_area,NULL AS normalized_parameter,NULL AS normalized_date,NULL AS normalized_value,NULL AS normalized_unit,NULL AS reference_text,NULL AS existing_database_value,NULL AS existing_workbook_value,NULL AS staging_status,NULL AS staging_operation,NULL AS old_value_json,NULL AS new_value_json FROM document_candidates WHERE document_id=? AND source_text_version=(SELECT MAX(version) FROM document_text_versions WHERE document_id=document_candidates.document_id) ORDER BY page_number,section_number,id LIMIT 500""")
        labels={"laboratory_value":"Laborwert","reference_range":"Referenzbereich","medication":"Medikament","dose":"Dosis","intake_status":"Einnahmestatus","diagnosis":"Dokumentierte Diagnoseaussage","symptom":"Dokumentierte Beschwerde","appointment":"Termin/Untersuchung","important_event":"Dokumentierte Aussage","document_date":"Dokumentdatum","institution":"Institution","vaccination":"Impfung"}
        match_labels={"exact_match":"Exakt bereits vorhanden","format_unit_match":"Nur Format-/Einheitenunterschied","value_conflict":"Wert widersprüchlich","not_present":"Noch nicht vorhanden","ambiguous":"Zuordnung mehrdeutig","repeated_exact":"Bereits in früherem Dokument enthalten","supporting_reference":"Zum Laborwert gehörender Referenzbereich","non_transferable":"Bleibt geprüfte Dokumentinformation"}
        for item in connection.execute(candidate_sql,(int(row["id"]),)):
            transfer=None
            if item["staging_status"]:
                stored_old=json.loads(str(item["old_value_json"]));stored_new=json.loads(str(item["new_value_json"]))
                old={key:stored_old.get(key) for key in ("database","workbook") if key in stored_old}
                new={key:stored_new.get(key) for key in ("parameter","value","unit","date","reference") if key in stored_new}
                transfer={"status":str(item["staging_status"]),"operation":str(item["staging_operation"]),"target":str(item["target_area"]),"old":old,"new":new,"source":{"page":int(item["page_number"])}}
            candidates.append({
                "id": str(item["id"]), "type": str(item["candidate_type"]),"type_label":labels.get(str(item["candidate_type"]),"Dokumenthinweis"),
                "value": str(item["value_text"]), "unit": item["unit"],
                "page": int(item["page_number"]), "section": int(item["section_number"]),
                "context": str(item["context_text"])[:320], "engine": public_engine_label(item["engine"]),
                "confidence": item["confidence"], "status": str(item["status"]),
                "corrected_value": item["corrected_value_text"], "corrected_unit": item["corrected_unit"],
                "comparison":{"status":item["match_status"],"label":match_labels.get(str(item["match_status"]),"Noch nicht abgeglichen"),"parameter":item["normalized_parameter"],"date":parse_day(item["normalized_date"]),"value":item["normalized_value"],"unit":item["normalized_unit"],"reference":item["reference_text"],"database_value":item["existing_database_value"],"workbook_value":item["existing_workbook_value"],"medication_state":("tatsächlich verabreicht/eingenommen" if str(item["candidate_type"])=="medication" and str(item["target_area"] or "")=="medication" else ("verordnet oder Status unklar" if str(item["candidate_type"])=="medication" else None))},
                "transfer_preview":transfer,
            })
    category = safe_metadata_text(row["kategorie"], 80, allow_empty=True) or ""
    institution = safe_metadata_text(row["institution"], 120, allow_empty=True) or ""
    original_reviewed = processing["original_reviewed_at"] is not None
    media = None
    if table_exists(connection, "document_media"):
        media_row = connection.execute("SELECT * FROM document_media WHERE document_id=?", (int(row["id"]),)).fetchone()
        if media_row:
            media = {
                "media_kind": str(media_row["media_kind"]),
                "type_label": "Video" if media_row["media_kind"] == "video" else "Foto",
                "preview_status": str(media_row["preview_status"]),
                "preview_url": f"/api/v1/documents/{opaque}/preview" if media_row["preview_name"] else None,
                "playback_url": f"/api/v1/documents/{opaque}/proxy" if media_row["proxy_name"] else None,
                "frame_count": int(media_row["frame_count"]),
                "auxiliary_count": int(media_row["auxiliary_count"]),
                "unconfirmed_captured_at": safe_metadata_text(media_row["unconfirmed_captured_at"], 40, allow_empty=True),
                "decoder": safe_metadata_text(media_row["decoder_name"], 80, allow_empty=False),
                "decoder_version": safe_metadata_text(media_row["decoder_version"], 160, allow_empty=False),
            }
    unresolved_candidates = sum(1 for item in candidates if item["status"] in {"open","conflicting"})
    reconciliation=None;navigation={"previous":None,"next":None,"position":None,"total":0};overall={"completed":0,"total":0,"decisions_open":0}
    if table_exists(connection,"document_reconciliation"):
        rec=connection.execute("SELECT queue_bucket,reason_code,priority,open_decisions,exact_match_count,conflict_count,byte_duplicate_of FROM document_reconciliation WHERE document_id=?",(int(row["id"]),)).fetchone()
        if rec:reconciliation={"bucket":str(rec["queue_bucket"]),"reason":str(rec["reason_code"]),"priority":int(rec["priority"]),"open_decisions":int(rec["open_decisions"]),"exact_matches":int(rec["exact_match_count"]),"conflicts":int(rec["conflict_count"]),"byte_duplicate":rec["byte_duplicate_of"] is not None}
        queued=connection.execute("""SELECT d.id,d.document_date,d.kategorie,d.institution FROM dokumente d JOIN document_reconciliation r ON r.document_id=d.id WHERE r.queue_bucket='now_reviewable' ORDER BY r.priority DESC,r.open_decisions DESC,d.document_date DESC,d.id DESC LIMIT 500""").fetchall()
        ids=[int(item["id"]) for item in queued]
        if int(row["id"]) in ids:
            index=ids.index(int(row["id"]));navigation={"previous":None,"next":None,"position":index+1,"total":len(ids)}
            for key,target in (("previous",queued[index-1] if index else None),("next",queued[index+1] if index+1<len(queued) else None)):
                if target:
                    target_category=safe_metadata_text(target["kategorie"],80,allow_empty=True) or "";target_institution=safe_metadata_text(target["institution"],120,allow_empty=True) or "";navigation[key]=api_document_id(target["id"],parse_day(target["document_date"]),target_category,target_institution)
        overall={"completed":int(connection.execute("SELECT COUNT(*) FROM document_reconciliation WHERE queue_bucket='no_action'").fetchone()[0]),"total":int(connection.execute("SELECT COUNT(*) FROM document_reconciliation").fetchone()[0]),"decisions_open":int(connection.execute("SELECT COALESCE(SUM(open_decisions),0) FROM document_reconciliation WHERE queue_bucket='now_reviewable'").fetchone()[0])}
    review_blockers: list[str] = []
    if not original_reviewed: review_blockers.append("Originaldatei noch nicht bestätigt")
    is_video = bool(media and media["media_kind"] == "video")
    if is_video and media is not None:
        if media["preview_status"] != "ready": review_blockers.append("Vorschau konnte lokal nicht erzeugt werden")
    else:
        if not pages or any(not page["reviewed"] for page in pages): review_blockers.append("Nicht alle Textseiten geprüft")
        if processing["extraction_status"] not in {"text_layer","ocr"}: review_blockers.append("Textextraktion nicht vollständig prüfbar")
    if unresolved_candidates: review_blockers.append("Medizinische Kandidaten noch offen oder widersprüchlich")
    return {
        "id": opaque, "review_id": str(processing["intake_id"]),
        "title": safe_metadata_text(processing["personal_title"],120,allow_empty=True) or f"Dokument · {category or 'Ohne Kategorie'}",
        "document_date": parse_day(row["document_date"]), "category": category,
        "institution": institution, "type": "video" if is_video else (row["daten_typ"] if row["daten_typ"] in {"pdf","image"} else "unknown"),
        "media": media,
        "original_preview_url": (f"/api/v1/documents/{opaque}/preview-original" if str(row["daten_typ"])=="pdf" and str(processing["original_status"])=="available" else (media.get("preview_url") if media else None)),
        "statuses": {"original":str(processing["original_status"]),"extraction":str(processing["extraction_status"]),
                     "content":str(processing["content_status"]),"search":str(processing["search_status"]),
                     "transfer":str(processing["transfer_status"])},
        "page_count": processing["page_count"], "engine": safe_metadata_text(processing["engine"],80,allow_empty=True),
        "engine_version": safe_metadata_text(processing["engine_version"],80,allow_empty=True),
        "investigation_day": parse_day(processing["investigation_day"]),
        "duplicate_candidate": processing["duplicate_document_id"] is not None,
        "original_reviewed": original_reviewed,
        "review_ready": not review_blockers,
        "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]
) -> dict[str, Any]:
    start, end = parse_range(params)
    result: dict[str, Any] = {
        "planned": [],
        "administered": [],
        "missed": [],
        "corrected": [],
    }
    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"),
    }
    truncated_sections: list[str] = []
    for bucket, events in groups.items():
        effective = (
            "COALESCE(NULLIF(scheduled_next_date,''),datum)"
            if bucket == "planned"
            else "datum"
        )
        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


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],
        "truncated": len(rows) > RECORD_MAX_ROWS,
        "next_cursor": None,
    }


def _lab_search_key(value: str) -> str:
    return re.sub(r"[^a-z0-9]+", "", value.casefold().replace("ß", "ss"))


LAB_SEARCH_ALIASES = {
    _lab_search_key("c-reaktives protein crp"): "CRP",
    _lab_search_key("crp"): "CRP",
    _lab_search_key("d-dimer"): "D-Dimer",
}


def _record_labs(
    connection: sqlite3.Connection, params: dict[str, str]
) -> dict[str, Any]:
    query = params.get("q", "")
    if query and (len(query) > 80 or safe_metadata_text(query, 80) is None):
        raise APIError(400, "query_not_allowed")
    search_key = _lab_search_key(query) if query else ""
    canonical_query = LAB_SEARCH_ALIASES.get(search_key, query.casefold())
    start, end = parse_range(params)
    today = local_today()
    history_floor = today - timedelta(days=MAX_RANGE_DAYS - 1)
    all_rows = _verified_lab_rows(
        connection,
        include_missing_reference=True,
        start=history_floor,
        end=today,
    )
    grouped_rows: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in all_rows:
        grouped_rows[str(row["metric_id"])].append(row)
    histories: dict[str, dict[str, Any]] = {}
    previous_by_id: dict[str, dict[str, Any]] = {}
    for metric_id, observations in grouped_rows.items():
        observations.sort(key=lambda row: (str(row["date"]), str(row["id"])))
        histories[metric_id] = {
            "metric_id": metric_id,
            "earliest_date": observations[0]["date"],
            "latest_date": observations[-1]["date"],
            "observation_count": len(observations),
            "verified_observation_count": len(observations),
            "missing_reference_count": sum(
                item.get("reference_status") == "missing_reference"
                for item in observations
            ),
            "linked_document_count": sum(
                item.get("document_link_status") == "linked" for item in observations
            ),
            "unlinked_document_count": sum(
                item.get("document_link_status") != "linked" for item in observations
            ),
            "range_from": observations[0]["date"],
            "range_to": today.isoformat(),
        }
        for index, observation in enumerate(observations):
            if index:
                previous_by_id[str(observation["id"])] = observations[index - 1]
    result = []
    for source in all_rows:
        if canonical_query and canonical_query.casefold() not in str(source["parameter"]).casefold():
            continue
        if start and source["date"] < start.isoformat():
            continue
        if end and source["date"] > end.isoformat():
            continue
        previous = previous_by_id.get(str(source["id"]))
        value = source.get("value")
        prior = previous.get("value") if previous else None
        delta = (
            round(float(value) - float(prior), 6)
            if isinstance(value, (int, float)) and isinstance(prior, (int, float))
            else None
        )
        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),
    )
    result = []
    for row in rows:
        recorded = parse_day(row["datum"])
        scheduled = parse_day(row["scheduled_next_date"])
        name = safe_metadata_text(row["medication_name"], 120)
        if recorded and scheduled and name:
            result.append(
                {
                    "date": scheduled,
                    "recorded_date": recorded,
                    "scheduled_next_date": scheduled,
                    "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),
                }
            )
    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,
    }


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": [],
            "undated": [],
            "medical_statement": "Dokumentierte Tageschronik ohne medizinische Bewertung.",
        }
    rows = connection.execute(
        """SELECT id,root_id,version,capture_type,occurred_at,ended_at,payload_json,status,
                         corrects_entry_id,withdraws_entry_id,source,created_at
                               FROM capture_entries WHERE substr(occurred_at,1,10)=? ORDER BY occurred_at,created_at,id LIMIT 500""",
        (day,),
    ).fetchall()
    items: list[dict[str, Any]] = []
    attachment_columns = {
        str(item[1]) for item in connection.execute("PRAGMA table_info(capture_attachments)")
    } if table_exists(connection, "capture_attachments") else set()
    media_v2 = "media_kind" in attachment_columns
    labels = {
        "symptom": "Beschwerde/Symptom",
        "medication": "Medikament",
        "supplement": "Supplement",
        "event": "Ereignis/Einfluss",
        "sauna": "Sauna/Erholung",
        "training": "Training · manuell erfasst",
        "measurement": "Körperwert",
        "photo": "Foto/Beobachtung",
    }
    for row in rows:
        try:
            data = json.loads(row["payload_json"])
        except (TypeError, json.JSONDecodeError):
            continue
        if not isinstance(data, dict):
            continue
        attachments = []
        if attachment_columns:
            selection = (
                "id,description,body_region,media_kind,review_status,preview_status,frame_count,auxiliary_count,unconfirmed_captured_at,proxy_name"
                if media_v2 else
                "id,description,body_region,'photo','unverified','ready',1,0,NULL,NULL"
            )
            for attachment in connection.execute(
                f"SELECT {selection} FROM capture_attachments WHERE entry_id=? ORDER BY created_at,id LIMIT 5",
                (row["id"],),
            ):
                media_id = str(attachment[0])
                media_kind = str(attachment[3])
                attachments.append({
                    "preview_url": f"/api/v1/capture/media/{media_id}/preview",
                    "thumbnail_url": f"/api/v1/capture/media/{media_id}/preview",
                    "playback_url": f"/api/v1/capture/media/{media_id}/proxy" if media_kind == "video" and attachment[9] else None,
                    "original_url": f"/api/v1/capture/media/{media_id}/original",
                    "media_kind": media_kind,
                    "type_label": "Video" if media_kind == "video" else "Foto",
                    "review_status": str(attachment[4]),
                    "verified": attachment[4] == "verified",
                    "preview_status": str(attachment[5]),
                    "frame_count": int(attachment[6]),
                    "auxiliary_count": int(attachment[7]),
                    "unconfirmed_captured_at": safe_metadata_text(attachment[8], 40, allow_empty=True),
                    "description": safe_metadata_text(attachment[1], 160, allow_empty=True),
                    "body_region": safe_metadata_text(attachment[2], 80, allow_empty=True),
                })
        occurred = safe_metadata_text(row["occurred_at"], 32, allow_empty=False)
        display_type = str(row["capture_type"])
        if display_type == "event" and data.get("event_kind") in {"sauna", "training"}:
            display_type = str(data["event_kind"])
        title = safe_metadata_text(
            data.get("title", ""), 100, allow_empty=True
        ) or labels.get(display_type, "Dokumentierter Eintrag")
        item = {
            "entry_ref": str(row["id"]),
            "root_ref": str(row["root_id"]),
            "version": int(row["version"]),
            "type": display_type,
            "category_label": labels.get(display_type, "Dokumentation"),
            "time": occurred[11:16] if occurred and len(occurred) >= 16 else None,
            "occurred_at": occurred,
            "ended_at": safe_metadata_text(row["ended_at"], 32, allow_empty=True),
            "title": title,
            "status": str(row["status"]),
            "source_label": (
                "Manuell in Dashboard V5 erfasst"
                if str(row["source"]) == "dashboard_v5_mobile_capture"
                else "Dokumentierter Eintrag"
            ),
            "captured_at": safe_metadata_text(row["created_at"], 40, allow_empty=True),
            "correction": bool(row["corrects_entry_id"]),
            "withdrawal": bool(row["withdraws_entry_id"]),
            "details": data,
            "attachments": attachments,
        }
        items.append(item)
    current_by_root: dict[str, dict[str, Any]] = {}
    history_by_root: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for item in items:
        history_by_root[item["root_ref"]].append(item)
        if item["status"] == "active":
            current_by_root[item["root_ref"]] = item
    visible = list(current_by_root.values())
    for root, history in history_by_root.items():
        if root not in current_by_root:
            visible.append(max(history, key=lambda item: item["version"]))
    for item in visible:
        item["history"] = [
            {
                "version": h["version"],
                "status": h["status"],
                "occurred_at": h["occurred_at"],
            }
            for h in history_by_root[item["root_ref"]]
        ]
        item.pop("root_ref", None)
    if table_exists(connection, "nutrition_daily_summary_v2"):
        nutrition = connection.execute(
            "SELECT item_count FROM nutrition_daily_summary_v2 WHERE datum=? AND item_count>0",
            (day,),
        ).fetchone()
        if nutrition:
            visible.append(
                {
                    "entry_ref": None,
                    "version": 1,
                    "type": "nutrition",
                    "category_label": "Ernährung",
                    "time": None,
                    "occurred_at": day,
                    "ended_at": None,
                    "title": "Ernährung dokumentiert",
                    "status": "active",
                    "correction": False,
                    "withdrawal": False,
                    "details": {"item_count": int(nutrition[0])},
                    "attachments": [],
                    "history": [],
                }
            )
    if table_exists(connection, "arztbesuche"):
        for appointment in connection.execute(
            "SELECT id,grund FROM arztbesuche WHERE datum=? ORDER BY id LIMIT 20",
            (day,),
        ):
            visible.append(
                {
                    "entry_ref": None,
                    "version": 1,
                    "type": "appointment",
                    "category_label": "Arzttermin/Untersuchung",
                    "time": None,
                    "occurred_at": day,
                    "ended_at": None,
                    "title": safe_metadata_text(appointment[1], 100, allow_empty=True)
                    or "Arzttermin",
                    "status": "reference",
                    "correction": False,
                    "withdrawal": False,
                    "details": {},
                    "attachments": [],
                    "history": [],
                }
            )
    timed = sorted(
        (item for item in visible if item.get("time")),
        key=lambda item: (item["time"], item["type"], item.get("entry_ref") or ""),
    )
    undated = sorted(
        (item for item in visible if not item.get("time")),
        key=lambda item: (item["type"], item["title"]),
    )
    media_counts = {
        "photos": sum(attachment.get("media_kind") == "photo" for item in visible for attachment in item.get("attachments", [])),
        "videos": sum(attachment.get("media_kind") == "video" for item in visible for attachment in item.get("attachments", [])),
    }
    return {
        "date": day,
        "timed": timed,
        "undated": undated,
        "media_counts": media_counts,
        "media_hint": " · ".join(part for part in ((f"{media_counts['photos']} Foto" if media_counts["photos"] == 1 else f"{media_counts['photos']} Fotos") if media_counts["photos"] else "", f"{media_counts['videos']} Video" if media_counts["videos"] == 1 else (f"{media_counts['videos']} Videos" if media_counts["videos"] else "")) if part),
        "medical_statement": "Dokumentierte Tageschronik ohne Diagnose-, Kausalitäts- oder Therapieaussage.",
    }

def _additional_symptoms(connection: sqlite3.Connection, start: date, end: date) -> list[dict[str, Any]]:
    if not table_exists(connection, "symptom_log"):
        return []
    columns = {row[1] for row in connection.execute("PRAGMA table_info(symptom_log)")}
    occurred = "occurred_at" if "occurred_at" in columns else "NULL"
    rows = connection.execute(
        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 = (
        _record_medications(
            connection, {"from": start.isoformat(), "to": end.isoformat()}
        )
        if "medications" in selected
        else {"planned": [], "administered": [], "missed": [], "corrected": []}
    )
    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"]
            ):
                symptoms.append({"date": day, "symptoms": symptom, "events": events})
            current += timedelta(days=1)
    additional_symptoms = _additional_symptoms(connection, start, end) if "symptoms" in selected else []
    nutrition = _nutrition_days(connection, {"from": start.isoformat(), "to": end.isoformat()}) if "nutrition" in selected else {"days": []}
    observations: list[dict[str, Any]] = []
    if "observations" in selected:
        for metric_id in ("apple.sleep", "apple.hrv", "apple.resting_heart_rate", "apple.steps", "apple.distance", "apple.active_energy"):
            try:
                observations.append(_series(connection, {"metric": metric_id, "from": start.isoformat(), "to": end.isoformat(), "resolution": "day"}))
            except APIError as error:
                if error.code not in {"metric_not_found", "metric_not_released"}:
                    raise
    timeline = _events(connection, {"from": start.isoformat(), "to": end.isoformat()})["events"] if "timeline" in selected else []
    temporal_observations: list[dict[str, Any]] = []
    symptom_days = sorted({item["date"] for item in additional_symptoms} | {item["date"] for item in symptoms})
    for symptom_day in symptom_days:
        anchor = date.fromisoformat(symptom_day)
        nearby = []
        for event in timeline:
            offset = (date.fromisoformat(event["date"]) - anchor).days
            if -3 <= offset <= 3 and event["date"] != symptom_day:
                nearby.append({"date": event["date"], "offset_days": offset, "type": event["type"], "label": event.get("label", "Dokumentiertes Ereignis")})
        if nearby:
            temporal_observations.append({"symptom_date": symptom_day, "nearby": nearby[:12], "statement": "Explorative zeitliche Nähe; keine Kausalitätsaussage."})
    nutrition_proximity: list[dict[str, Any]] = []
    if "nutrition" in selected and table_exists(connection, "nutrition_meal_summary"):
        meal_rows = list(
            connection.execute(
                """SELECT datum,meal,item_count FROM nutrition_meal_summary
                   WHERE datum>=? AND datum<=? AND item_count>0
                   ORDER BY datum,CASE meal WHEN 'breakfast' THEN 1 WHEN 'lunch' THEN 2
                   WHEN 'dinner' THEN 3 WHEN 'snack' THEN 4 ELSE 9 END LIMIT 1000""",
                (start.isoformat(), end.isoformat()),
            )
        )
        meals_by_day: dict[str, list[str]] = defaultdict(list)
        for meal_row in meal_rows:
            meal_day = parse_day(meal_row["datum"])
            meal = str(meal_row["meal"] or "unassigned")
            if meal_day and meal in MEAL_LABELS:
                meals_by_day[meal_day].append(MEAL_LABELS[meal])
        for symptom_day in symptom_days:
            anchor = date.fromisoformat(symptom_day)
            for nutrition_day, meals in sorted(meals_by_day.items()):
                offset = (date.fromisoformat(nutrition_day) - anchor).days
                if -3 <= offset <= 3:
                    nutrition_proximity.append(
                        {
                            "nutrition_date": nutrition_day,
                            "symptom_date": symptom_day,
                            "offset_days": offset,
                            "meals": meals,
                            "statement": (
                                "Explorative zeitliche Nähe; keine Kausalitätsaussage."
                            ),
                        }
                    )
                if len(nutrition_proximity) >= 30:
                    break
            if len(nutrition_proximity) >= 30:
                break
    overview = (
        _record_summary(connection, reviewed_documents_only=True)
        if "overview" in selected
        else None
    )
    personal_observations: list[dict[str, Any]] = []
    if "personal_observations" in selected and table_exists(connection, "personal_observations"):
        candidates = connection.execute(
            "SELECT id FROM personal_observations WHERE include_doctor=1 AND status IN ('active','completed') AND start_date<=? AND end_date>=? ORDER BY updated_at DESC LIMIT 5",
            (end.isoformat(), start.isoformat()),
        ).fetchall()
        for candidate in candidates:
            observation_id = str(candidate[0])

            def report_series(metric_id: str, range_start: date, range_end: date) -> dict[str, Any]:
                try:
                    return _series(connection, {"metric": metric_id, "from": range_start.isoformat(), "to": range_end.isoformat(), "resolution": "day"})
                except APIError as error:
                    if error.code in {"metric_not_found", "metric_not_released"}:
                        return {"points": [], "unit": "", "source": {"type": "not_documented"}}
                    raise

            analysis = evaluate_observation(connection, observation_id, report_series)
            chart_metrics: list[dict[str, Any]] = []
            seen: set[str] = set()
            for phase in analysis["phases"]:
                for metric_id, summary in phase["metrics"].items():
                    if metric_id not in seen and len(seen) < 2:
                        seen.add(metric_id)
                        chart_metrics.append({"metric_id": metric_id, "label": summary["label"], "unit": summary["unit"], "source": summary["source"], "phases": []})
                    target = next((item for item in chart_metrics if item["metric_id"] == metric_id), None)
                    if target is not None:
                        target["phases"].append({"name": phase["name"], "start_date": phase["start_date"], "end_date": phase["end_date"], "points": summary["points"], "documented_days": summary["documented_days"], "expected_days": summary["expected_days"]})
            personal_observations.append({"id": observation_id, "title": analysis["configuration"]["title"], "question": analysis["configuration"]["question"], "from": analysis["configuration"]["start_date"], "to": analysis["configuration"]["end_date"], "lag_min": analysis["configuration"]["lag_min"], "lag_max": analysis["configuration"]["lag_max"], "phases": [{"name": phase["name"], "phase_type": phase["phase_type"], "start_date": phase["start_date"], "end_date": phase["end_date"], "summary_label": phase["summary_label"], "expected_days": phase["expected_days"], "checkin_days": phase["checkin_days"], "event_count": phase["event_count"], "known_factor_count": len(phase["known_factors"])} for phase in analysis["phases"]], "charts": chart_metrics[:2], "medical_statement": analysis["medical_statement"]})
    payload = {
        "from": start.isoformat(),
        "to": end.isoformat(),
        "timezone": TZ_NAME,
        "selected_sections": sorted(selected),
        "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")
                ),
                "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 = {
        "/api/v1/record-summary",
        "/api/v1/record-labs",
        "/api/v1/lab-review",
        "/api/v1/medications",
        "/api/v1/appointments",
        "/api/v1/documents",
        "/api/v1/document-review-queue",
        "/api/v1/doctor-report",
    }
    if (
        path not in API_PATHS
        and path not in allowed_record
        and path != "/api/v1/capture/plans"
        and not observation_route
        and not observation_result_route
        and not document_route
        and not capture_timeline_route
        and not re.fullmatch(r"/api/v1/day/\d{4}-\d{2}-\d{2}", path)
        and not re.fullmatch(r"/api/v1/nutrition/day/\d{4}-\d{2}-\d{2}", path)
    ):
        raise APIError(404, "endpoint_not_found")
    connection = connect_read_only(database)
    try:
        if path == "/api/v1/source-status":
            parse_query(query, set())
            return build_source_status(connection, artifact_path=runtime_artifact)
        if path == "/api/v1/capture/plans":
            parse_query(query, set())
            return _capture_plans(connection)
        if capture_timeline_route:
            parse_query(query, set())
            return _capture_timeline(connection, capture_timeline_route.group(1))
        if path == "/api/v1/observations/templates":
            parse_query(query, set())
            return {
                "version": 1,
                "contract_version": PUBLIC_CONTRACT_VERSION,
                "templates": observation_templates(),
                "method_version": "personal-observation-phases-v1",
            }
        if path == "/api/v1/observations":
            params = parse_query(query, {"status"})
            requested = set(filter(None, params.get("status", "").split(",")))
            if requested and not requested <= OBSERVATION_STATUS:
                raise APIError(422, "observation_status_not_allowed")
            items = (
                list_observations(connection, requested or None)
                if table_exists(connection, "personal_observations")
                else []
            )
            return {"version": 1, "contract_version": PUBLIC_CONTRACT_VERSION, "items": items, "count": len(items)}
        if path == "/api/v1/observations/active-today":
            parse_query(query, set())
            return {
                "version": 1,
                "contract_version": PUBLIC_CONTRACT_VERSION,
                "task": None,
                "reason": "no_unambiguous_user_action",
            }
        if observation_result_route:
            observation_id, result_id = observation_result_route.groups()
            parse_query(query, set())
            if not table_exists(connection, "personal_observation_results"):
                raise APIError(404, "result_not_found")
            row = connection.execute(
                "SELECT id,result_version,configuration_json,phases_json,summary_json,method_version,configuration_hash,created_at FROM personal_observation_results WHERE id=? AND observation_id=?",
                (result_id, observation_id),
            ).fetchone()
            if row is None:
                raise APIError(404, "result_not_found")
            return {
                "id": str(row[0]),
                "result_version": int(row[1]),
                "configuration": json.loads(str(row[2])),
                "phases": json.loads(str(row[3])),
                "summary": json.loads(str(row[4])),
                "method_version": str(row[5]),
                "configuration_hash": str(row[6]),
                "created_at": str(row[7]),
                "immutable": True,
            }
        if observation_route:
            observation_id, suffix = observation_route.groups()
            parse_query(query, set())
            try:
                if suffix == "analysis":

                    def observation_series(
                        metric_id: str, start: date, end: date
                    ) -> dict[str, Any]:
                        try:
                            return _series(
                                connection,
                                {
                                    "metric": metric_id,
                                    "from": start.isoformat(),
                                    "to": end.isoformat(),
                                    "resolution": "day",
                                },
                            )
                        except APIError as error:
                            if error.code in {
                                "metric_not_found",
                                "metric_not_released",
                            }:
                                return {
                                    "points": [],
                                    "unit": "",
                                    "source": {"type": "not_documented"},
                                }
                            raise

                    detail = observation_detail(connection, observation_id)
                    if detail["influences"][0] in {"event.sauna", "event.training", "nutrition.profile", "nutrition.histamine"} and all(
                        not item.startswith("event.") for item in detail["outcomes"]
                    ):
                        return evaluate_plan(
                            connection,
                            observation_id,
                            observation_series,
                            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"}))
        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"}:
                parse_query(query, set())
                return _document_review_workspace(connection, opaque)
            if suffix == "compare":
                parse_query(query, set())
                return _document_compare(connection, opaque)
            return _record_document_detail(
                connection,
                opaque,
                parse_query(query, {"cursor", "limit"}),
            )
        if path == "/api/v1/doctor-report":
            return _doctor_report(
                connection, parse_query(query, {"from", "to", "sections"})
            )
        if path == "/api/v1/metric-catalog":
            params = parse_query(query, {"q"})
            query_text = params.get("q", "")
            metrics = (
                catalog_search(connection, query_text)
                if query_text
                else public_catalog(connection)
            )
            inventory = inventory_sources(connection)
            return {
                "version": 2,
                "timezone": TZ_NAME,
                "today": local_today().isoformat(),
                "groups": {"metrics": metrics},
                "inventory": {
                    "apple_observed_identifiers": len(inventory["apple_health"]),
                    "apple_released_identifiers": sum(
                        item["status"] == "released"
                        for item in inventory["apple_health"]
                    ),
                    "canonical_verified_laboratory_parameters": len(
                        inventory["laboratory"]
                    ),
                    "observed_vital_sign_parameters": len(inventory["vital_signs"]),
                },
            }
        if path == "/api/v1/explorer-comparison/catalog":
            parse_query(query, set())
            availability = {
                item["id"]: int(item.get("availability", {}).get("observations", 0) or 0)
                for item in public_catalog(connection)
            }
            return comparison_catalog(availability)
        if path == "/api/v1/explorer-comparison":
            params = parse_query(
                query, {"metrics", "influences", "from", "to"}
            )
            if not {"from", "to"} <= set(params):
                raise APIError(400, "comparison_parameters_required")
            start, end = parse_range(params, required=True)
            assert start is not None and end is not None
            if (end - start).days + 1 > MAX_COMPARISON_DAYS:
                raise APIError(422, "comparison_range_invalid")

            def load_comparison_series(
                metric_id: str, range_start: date, range_end: date
            ) -> dict[str, Any]:
                return _series(
                    connection,
                    {
                        "metric": metric_id,
                        "from": range_start.isoformat(),
                        "to": range_end.isoformat(),
                        "resolution": "day",
                    },
                )

            try:
                return build_comparison(connection, params, load_comparison_series)
            except (KeyError, ValueError) as error:
                code = str(error).strip("'") or "comparison_invalid"
                allowed_codes = {
                    "comparison_selection_invalid",
                    "comparison_range_invalid",
                    "comparison_response_too_large",
                }
                raise APIError(
                    422, code if code in allowed_codes else "comparison_invalid"
                ) from error
        if path == "/api/v1/associations/catalog":
            parse_query(query, set())
            result = association_catalog(BY_ID_V2)
            available = {
                item["id"]
                for item in public_catalog(connection)
                if item.get("availability", {}).get("observations", 0) > 0
            }
            event_start = local_today() - timedelta(days=MAX_ANALYSIS_DAYS - 1)
            for item in result["influences"] + result["targets"]:
                if item["kind"] == "event" and association_event_days(
                    connection, item["id"], event_start, local_today()
                ):
                    available.add(item["id"])
            result["presets"] = [
                preset
                for preset in result["presets"]
                if preset["influence"] in available and preset["target"] in available
            ]
            return result
        if path == "/api/v1/associations":
            params = parse_query(
                query,
                {"question", "influence", "target", "from", "to", "lag"},
            )
            required = {"question", "influence", "target", "from", "to"}
            if not required <= set(params):
                raise APIError(400, "association_parameters_required")
            start, end = parse_range(params, required=True)
            assert start is not None and end is not None
            if (end - start).days + 1 > MAX_ANALYSIS_DAYS:
                raise APIError(422, "association_range_too_large")

            def load_series(
                metric_id: str, range_start: date, range_end: date
            ) -> dict[str, Any]:
                return _series(
                    connection,
                    {
                        "metric": metric_id,
                        "from": range_start.isoformat(),
                        "to": range_end.isoformat(),
                        "resolution": "day",
                    },
                )

            try:
                return analyze_association(connection, params, BY_ID_V2, load_series)
            except (KeyError, ValueError) as error:
                code = str(error).strip("'") or "invalid_association"
                allowed_codes = {
                    "question_not_allowed",
                    "lag_out_of_range",
                    "range_too_large",
                    "influence_not_allowed",
                    "target_not_allowed",
                    "incompatible_metrics",
                    "event_not_allowed",
                    "event_row_limit",
                    "response_too_large",
                }
                raise APIError(
                    422, code if code in allowed_codes else "invalid_association"
                ) from error
        if path == "/api/v1/series":
            params = parse_query(query, {"metric", "from", "to", "resolution"})
            if not params.get("metric"):
                raise APIError(400, "metric_required")
            return _series(connection, params)
        if path.startswith("/api/v1/day/"):
            if query:
                raise APIError(400, "unknown_parameter")
            requested_day = parse_iso_day(path.rsplit("/", 1)[-1])
            if requested_day > local_today() + timedelta(days=MAX_FUTURE_DAYS):
                raise APIError(422, "future_date_not_allowed")
            return _day(connection, requested_day)
        if path == "/api/v1/supplements":
            params = parse_query(query, {"from", "to"})
            start, end = parse_range(params, required=True)
            assert start is not None and end is not None
            try:
                return supplements(connection, start, end)
            except ValueError as error:
                raise APIError(422, str(error)) from error
        if path == "/api/v1/nutrition/days":
            return _nutrition_days(connection, parse_query(query, {"from", "to"}))
        if path.startswith("/api/v1/nutrition/day/"):
            if query:
                raise APIError(400, "unknown_parameter")
            requested_day = parse_iso_day(path.rsplit("/", 1)[-1])
            if requested_day > local_today():
                raise APIError(422, "future_date_not_allowed")
            return _nutrition_day_detail(connection, requested_day)
        if path == "/api/v1/nutrition/mapping-queue":
            return _nutrition_mapping_queue(connection, parse_query(query, {"status", "search"}))
        if path == "/api/v1/calendar":
            params = parse_query(query, {"from", "to"})
            return _calendar(connection, params)
        if path == "/api/v1/events":
            params = parse_query(query, {"from", "to", "types"})
            return _events(connection, params)
        if path == "/api/v1/labs":
            parse_query(query, set())
            rows = _verified_lab_rows(connection)
            if len(rows) > MAX_LAB_ROWS:
                raise APIError(422, "row_limit_exceeded")
            return {
                "timezone": TZ_NAME,
                "reference_policy": "observation_specific_verified_original",
                "observations": rows,
            }
        if path == "/api/v1/search":
            params = parse_query(query, {"q", "include_machine"})
            include_machine = params.get("include_machine", "0")
            if include_machine not in {"0", "1"}:
                raise APIError(400, "include_machine_not_allowed")
            return _search(connection, params.get("q", ""), include_machine=include_machine == "1")
        raise APIError(404, "endpoint_not_found")
    except SourceInventoryLimitError as error:
        raise APIError(422, "source_inventory_limit_exceeded") from error
    except sqlite3.Error as error:
        raise APIError(503, "data_unavailable") from error
    finally:
        connection.close()

__HERMES_CWD_8d46a20096ed__/home/agent/.hermes/repos/HealthManager__HERMES_CWD_8d46a20096ed__