from __future__ import annotations

import json
import sqlite3
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
HEALTH = ROOT / "scripts" / "health"
if str(HEALTH) not in sys.path:
    sys.path.insert(0, str(HEALTH))

from dashboard_v5.capture_contract import validate_capture_payload  # noqa: E402
from dashboard_v5.metric_catalog_v2 import BY_ID_V2, catalog_search  # noqa: E402
from dashboard_v5.read_api import _series  # noqa: E402
from dashboard_v5.sprint6h_b_schema import apply_schema  # noqa: E402
import health_dashboard_action_worker as worker  # noqa: E402


def apple_db() -> sqlite3.Connection:
    connection = sqlite3.connect(":memory:")
    connection.row_factory = sqlite3.Row
    connection.execute(
        """CREATE TABLE apple_health_records(
               id INTEGER PRIMARY KEY, metric TEXT, value REAL, value_text TEXT,
               unit TEXT, start_date TEXT, end_date TEXT, source_name TEXT,
               file_name TEXT, file_hash TEXT, raw_json TEXT
           )"""
    )
    return connection


def insert_apple(
    connection: sqlite3.Connection,
    metric: str,
    day: str,
    value: float | None,
    unit: str,
    raw: dict | None = None,
) -> None:
    connection.execute(
        """INSERT INTO apple_health_records
           (metric,value,unit,start_date,end_date,source_name,file_name,file_hash,raw_json)
           VALUES(?,?,?,?,?,'Health Auto Export','daily.json','fixture',?)""",
        (metric, value, unit, f"{day}T08:00:00+01:00", f"{day}T09:00:00+01:00", json.dumps(raw or {})),
    )


def payload(kind: str, data: dict, key: str = "a" * 32) -> dict:
    return {
        "version": 1,
        "action": "capture_entry",
        "capture_type": "event",
        "request_version": 1,
        "idempotency_key": key,
        "occurred_at": "2026-01-10T10:00",
        "ended_at": "2026-01-10T10:30" if kind == "sauna" else "2026-01-10T11:00",
        "data": data,
        "attachments": [],
        "corrects_entry_id": None,
        "withdraws_entry_id": None,
    }


def sauna_data() -> dict:
    return {
        "title": "Sauna",
        "note": "synthetisch",
        "event_kind": "sauna",
        "duration_minutes": 30,
        "rounds": 2,
        "temperature_c": 80,
        "cooling": "kurze Kaltphase",
        "hydration_ml": 500,
    }


def training_data() -> dict:
    return {
        "title": "Gehen",
        "note": "synthetisch",
        "event_kind": "training",
        "activity_type": "Gehen",
        "duration_minutes": 60,
        "active_kcal": 250,
        "distance_km": 5.5,
    }


def test_priority_catalog_is_evidence_bound_and_has_five_search_groups() -> None:
    expected = {
        "apple.blood_pressure.systolic": ("mmHg", "last"),
        "apple.blood_pressure.diastolic": ("mmHg", "last"),
        "apple.weight": ("kg", "last"),
        "apple.bmi": ("kg/m²", "last"),
        "apple.respiratory_rate": ("Atemzüge/min", "avg"),
        "apple.oxygen_saturation": ("%", "avg"),
        "apple.active_energy": ("kcal", "sum"),
        "apple.exercise_time": ("min", "sum"),
        "apple.distance": ("km", "sum"),
        "apple.sleep.core": ("h", "sum"),
        "apple.sleep.deep": ("h", "sum"),
        "apple.sleep.rem": ("h", "sum"),
        "apple.sleep.awake": ("h", "sum"),
    }
    assert {metric_id: (BY_ID_V2[metric_id].unit, BY_ID_V2[metric_id].aggregation) for metric_id in expected} == expected
    assert not any("vo2" in metric_id or "wrist_temperature" in metric_id for metric_id in BY_ID_V2)
    connection = apple_db()
    try:
        assert {item["category"] for item in catalog_search(connection, "Vitalwerte")} == {
            "vital_signs", "respiratory", "body_measurement"
        }
    finally:
        connection.close()


def test_units_missingness_sparse_series_and_zero_one_many_points() -> None:
    connection = apple_db()
    insert_apple(connection, "active_energy", "2026-01-01", 418.4, "kJ")
    insert_apple(connection, "weight_body_mass", "2026-01-02", 70.0, "kg")
    try:
        energy = _series(connection, {"metric": "apple.active_energy", "from": "2026-01-01", "to": "2026-01-03", "resolution": "day"})
        weight = _series(connection, {"metric": "apple.weight", "from": "2026-01-01", "to": "2026-01-03", "resolution": "day"})
        empty = _series(connection, {"metric": "apple.exercise_time", "from": "2026-01-01", "to": "2026-01-03", "resolution": "day"})
        assert len(energy["points"]) == 1
        assert energy["points"][0]["date"] == "2026-01-01"
        assert energy["points"][0]["value"] == pytest.approx(100.0)
        assert energy["points"][0]["quality"] == "direct"
        assert [point["date"] for point in weight["points"]] == ["2026-01-02"]
        assert weight["coverage"]["expectation"] == "intermittent"
        assert weight["coverage"]["expected_days"] is None
        assert weight["coverage"]["gaps"] == "sparse_observations_not_interpolated"
        assert empty["points"] == []
        insert_apple(connection, "apple_exercise_time", "2026-01-01", 15, "min")
        one = _series(connection, {"metric": "apple.exercise_time", "from": "2026-01-01", "to": "2026-01-03", "resolution": "day"})
        insert_apple(connection, "apple_exercise_time", "2026-01-02", 20, "min")
        many = _series(connection, {"metric": "apple.exercise_time", "from": "2026-01-01", "to": "2026-01-03", "resolution": "day"})
        assert len(one["points"]) == 1
        assert len(many["points"]) == 2
        assert all(point["value"] != 0 for point in many["points"])
    finally:
        connection.close()


def test_sleep_phases_use_observed_payload_fields_without_inventing_missing_values() -> None:
    connection = apple_db()
    insert_apple(
        connection,
        "sleep_analysis",
        "2026-01-01",
        7.0,
        "hr",
        {"totalSleep": 7.0, "core": 4.0, "deep": 1.0, "rem": 1.5, "awake": 0.5},
    )
    insert_apple(
        connection,
        "sleep_analysis",
        "2026-01-02",
        6.5,
        "hr",
        {"totalSleep": 6.5, "core": 3.5, "deep": 1.0, "rem": 2.0},
    )
    try:
        deep = _series(connection, {"metric": "apple.sleep.deep", "from": "2026-01-01", "to": "2026-01-02", "resolution": "day"})
        awake = _series(connection, {"metric": "apple.sleep.awake", "from": "2026-01-01", "to": "2026-01-02", "resolution": "day"})
        assert [point["date"] for point in deep["points"]] == ["2026-01-01", "2026-01-02"]
        assert [point["date"] for point in awake["points"]] == ["2026-01-01"]
        assert all(point["value"] is not None for point in awake["points"])
    finally:
        connection.close()


def test_blood_pressure_pairs_same_timestamp_and_marks_unpaired_components() -> None:
    connection = apple_db()
    insert_apple(connection, "blood_pressure", "2026-01-01", None, "mmHg", {"systolic": 120, "diastolic": 80})
    insert_apple(connection, "blood_pressure", "2026-01-02", None, "mmHg", {"systolic": 121})
    insert_apple(connection, "blood_pressure", "2026-01-03", None, "mmHg", {"diastolic": 81})
    try:
        systolic = _series(connection, {"metric": "apple.blood_pressure.systolic", "from": "2026-01-01", "to": "2026-01-03", "resolution": "day"})
        diastolic = _series(connection, {"metric": "apple.blood_pressure.diastolic", "from": "2026-01-01", "to": "2026-01-03", "resolution": "day"})
        assert [point["quality"] for point in systolic["points"]] == ["paired", "incomplete_pair"]
        assert [point["quality"] for point in diastolic["points"]] == ["paired", "incomplete_pair"]
        assert systolic["points"][1]["pair_status"] == "incomplete"
        assert diastolic["points"][1]["pair_status"] == "incomplete"
    finally:
        connection.close()


def test_sauna_and_training_validation_is_strict_and_uses_existing_event_type() -> None:
    sauna = validate_capture_payload(payload("sauna", sauna_data()))
    training = validate_capture_payload(payload("training", training_data(), "b" * 32))
    assert sauna["capture_type"] == training["capture_type"] == "event"
    assert sauna["data"]["event_kind"] == "sauna"
    assert training["data"]["event_kind"] == "training"
    with pytest.raises(ValueError):
        validate_capture_payload(payload("sauna", {**sauna_data(), "duration_minutes": 0}))
    with pytest.raises(ValueError):
        validate_capture_payload({**payload("training", training_data()), "ended_at": "2026-01-10T10:59"})
    with pytest.raises(ValueError):
        validate_capture_payload(payload("training", {**training_data(), "activity_type": ""}))


def test_worker_keeps_manual_events_separate_and_idempotent_without_schema_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = tmp_path / "health.db"
    connection = sqlite3.connect(database)
    apply_schema(connection)
    connection.execute(
        """CREATE TABLE health_events(
               id INTEGER PRIMARY KEY, date TEXT, category TEXT, parameter TEXT,
               value REAL, unit TEXT, source TEXT, notes TEXT, occurred_at TEXT,
               intensity INTEGER
           )"""
    )
    connection.commit()
    connection.close()
    monkeypatch.setattr(worker, "DASHBOARD_DB", database)
    monkeypatch.setattr(worker, "CAPTURE_QUARANTINE", tmp_path / "quarantine")
    monkeypatch.setattr(worker, "CAPTURE_MEDIA", tmp_path / "media")
    first = payload("sauna", sauna_data())
    second = payload("training", training_data(), "b" * 32)
    first_id = worker.apply_capture_action(first)
    assert worker.apply_capture_action(first) == first_id
    worker.apply_capture_action(second)
    connection = sqlite3.connect(database)
    connection.row_factory = sqlite3.Row
    try:
        entries = connection.execute("SELECT capture_type,payload_json,source FROM capture_entries ORDER BY created_at,id").fetchall()
        events = connection.execute("SELECT category,parameter,value,unit,source FROM health_events ORDER BY id").fetchall()
        assert len(entries) == 2
        assert [row["capture_type"] for row in entries] == ["event", "event"]
        assert {json.loads(row["payload_json"])["event_kind"] for row in entries} == {"sauna", "training"}
        assert [row["category"] for row in events] == ["sauna_recovery", "manual_training"]
        assert all(row["unit"] == "min" and row["source"] == "dashboard_v5_manual_capture" for row in events)
        assert connection.execute("SELECT COUNT(*) FROM capture_action_log").fetchone()[0] == 2
    finally:
        connection.close()