mitai-jinkendo/backend/data_layer/history_overview_viz.py

"""
Layer 2b: Gesamtansicht «Verlauf» — komponiert nur Bundles aus body-, nutrition-, fitness-, recovery_viz.

Issue #53: keine parallele Business-Logik; ein Router-Endpoint liefert diese Zusammenfassung.
"""

from __future__ import annotations

from typing import Any, Dict, List, Optional

from data_layer.body_viz import get_body_history_viz_bundle
from data_layer.correlation_chart_payloads import (
    build_lbm_protein_correlation_chart_payload,
    build_load_vitals_correlation_chart_payload,
    build_recovery_performance_chart_payload,
    build_weight_energy_correlation_chart_payload,
)
from data_layer.correlations import calculate_lag_correlation, calculate_top_drivers
from data_layer.fitness_viz import get_fitness_dashboard_viz_bundle
from data_layer.nutrition_viz import get_nutrition_history_viz_bundle
from data_layer.recovery_viz import get_recovery_dashboard_viz_bundle
from data_layer.utils import safe_float


def _take_kpis(tiles: Any, max_n: int = 4) -> List[Dict[str, Any]]:
    if not isinstance(tiles, list):
        return []
    out: List[Dict[str, Any]] = []
    for t in tiles[:max_n]:
        if not isinstance(t, dict):
            continue
        out.append(
            {
                "key": t.get("key"),
                "category": t.get("category"),
                "icon": t.get("icon"),
                "value": t.get("value"),
                "sublabel": t.get("sublabel"),
                "status": t.get("status"),
                "verdict": t.get("verdict"),
            }
        )
    return out


def _short_body_interpretation_tiles(tiles: Any, max_n: int = 3) -> List[Dict[str, Any]]:
    """Körper-Interpretationskacheln (keine KPI-Kacheln)."""
    if not isinstance(tiles, list):
        return []
    out: List[Dict[str, Any]] = []
    for t in tiles[:max_n]:
        if not isinstance(t, dict):
            continue
        det = str(t.get("detail") or "")
        if len(det) > 140:
            det = det[:137] + "…"
        out.append(
            {
                "title": t.get("title") or t.get("category") or "Hinweis",
                "detail": det,
                "status": t.get("status"),
            }
        )
    return out


def _take_insights(items: Any, max_n: int = 2) -> List[Dict[str, Any]]:
    if not isinstance(items, list):
        return []
    out: List[Dict[str, Any]] = []
    for it in items[:max_n]:
        if not isinstance(it, dict):
            continue
        out.append(
            {
                "title": it.get("title") or it.get("title_de"),
                "body": it.get("body") or it.get("detail") or it.get("message"),
                "tone": it.get("tone") or it.get("status"),
            }
        )
    return out


def get_history_overview_viz_bundle(profile_id: str, days: int) -> Dict[str, Any]:
    """
    Kompakte Übersicht für den ersten Reiter «Gesamtansicht»: KPI-Kurzformen + Lag-Korrelationen (C1–C4).
    """
    eff = max(7, min(int(days), 9999))
    body = get_body_history_viz_bundle(profile_id, eff)
    nutr = get_nutrition_history_viz_bundle(profile_id, eff)
    fit = get_fitness_dashboard_viz_bundle(profile_id, eff)
    rec = get_recovery_dashboard_viz_bundle(profile_id, eff)

    c1 = calculate_lag_correlation(profile_id, "energy_balance", "weight", 14)
    c2 = calculate_lag_correlation(profile_id, "protein", "lbm", 14)
    c3_hrv = calculate_lag_correlation(profile_id, "load", "hrv", 14)
    c3_rhr = calculate_lag_correlation(profile_id, "load", "rhr", 14)
    c3 = None
    if c3_hrv and c3_rhr:
        a1 = abs(safe_float(c3_hrv.get("correlation"), 0.0))
        a2 = abs(safe_float(c3_rhr.get("correlation"), 0.0))
        c3 = c3_hrv if a1 >= a2 else c3_rhr
        if c3 is c3_hrv:
            c3 = dict(c3)
            c3["metric"] = "HRV"
        else:
            c3 = dict(c3_rhr)
            c3["metric"] = "RHR"
    elif c3_hrv:
        c3 = dict(c3_hrv)
        c3["metric"] = "HRV"
    elif c3_rhr:
        c3 = dict(c3_rhr)
        c3["metric"] = "RHR"

    drivers = calculate_top_drivers(profile_id)

    b_sum = body.get("summary") if isinstance(body.get("summary"), dict) else {}
    last_w = b_sum.get("weight_kg")

    fs = fit.get("summary") if isinstance(fit.get("summary"), dict) else {}
    if fit.get("has_activity_entries"):
        ac = int(fs.get("activity_count") or 0)
        fitness_line = f"{ac} Trainingseinheiten im gewählten Fenster"
    else:
        fitness_line = fit.get("message") or "Keine Trainingsdaten"

    drv_list = drivers if isinstance(drivers, list) else []

    return {
        "days_requested": days,
        "effective_window_days": eff,
        "confidence": _overview_confidence(body, nutr, fit, rec),
        "sections": [
            {
                "id": "body",
                "title": "Körper",
                "tab_id": "body",
                "summary_line": (
                    f"Letztes Gewicht: {last_w} kg"
                    if last_w is not None
                    else "Keine Gewichtsdaten im Fenster"
                ),
                "interpretation_short": _short_body_interpretation_tiles(body.get("interpretation_tiles"), 3),
            },
            {
                "id": "nutrition",
                "title": "Ernährung",
                "tab_id": "nutrition",
                "summary_line": (
                    f"Ø {round(float((nutr.get('summary') or {}).get('kcal_avg') or 0))} kcal/Tag"
                    if nutr.get("has_nutrition_entries")
                    else (nutr.get("message") or "Keine Ernährungsdaten")
                ),
                "kpi_short": _take_kpis(nutr.get("kpi_tiles"), 4),
                "heuristic_short": (nutr.get("nutrition_correlation_heuristics") or [])[:2],
            },
            {
                "id": "fitness",
                "title": "Fitness",
                "tab_id": "activity",
                "summary_line": fitness_line,
                "kpi_short": _take_kpis(fit.get("kpi_tiles"), 4),
                "insights_short": _take_insights(fit.get("progress_insights"), 2),
            },
            {
                "id": "recovery",
                "title": "Erholung",
                "tab_id": "activity",
                "summary_line": "Schlaf & Vitalwerte"
                if rec.get("has_recovery_data")
                else (rec.get("message") or "Keine Erholungsdaten"),
                "kpi_short": _take_kpis(rec.get("kpi_tiles"), 4),
                "insights_short": _take_insights(rec.get("progress_insights"), 2),
            },
        ],
        "lag_correlations": {
            "weight_energy": _compact_lag("C1 Energiebilanz ↔ Gewicht", c1),
            "protein_lbm": _compact_lag("C2 Protein ↔ Magermasse", c2),
            "load_vitals": _compact_lag(
                f"C3 Last ↔ {(c3 or {}).get('metric') or 'Vital'}",
                c3,
                extra_keys=("metric",),
            ),
            "recovery_performance": {
                "label": "C4 Top-Treiber (Einflussfaktoren)",
                "drivers": drv_list[:8],
            },
        },
        "chart_payloads": {
            "c1_weight_energy": build_weight_energy_correlation_chart_payload(profile_id, 14),
            "c2_protein_lbm": build_lbm_protein_correlation_chart_payload(profile_id, 14),
            "c3_load_vitals": build_load_vitals_correlation_chart_payload(profile_id, 14),
            "c4_recovery_performance": build_recovery_performance_chart_payload(profile_id),
        },
        "meta": {
            "layer_1": "composed_metrics",
            "layer_2b": "history_overview_viz",
            "issue": "53-history-overview",
            "sources": {
                "body": "body_viz",
                "nutrition": "nutrition_viz",
                "fitness": "fitness_viz",
                "recovery": "recovery_viz",
                "lag": "correlations.calculate_lag_correlation",
                "drivers": "correlations.calculate_top_drivers",
            },
        },
    }


def _overview_confidence(b: Dict, n: Dict, f: Dict, r: Dict) -> str:
    scores = []
    for x in (b, n, f, r):
        c = x.get("confidence")
        if c == "high":
            scores.append(3)
        elif c == "medium":
            scores.append(2)
        elif c == "low":
            scores.append(1)
        else:
            scores.append(0)
    s = sum(scores) / max(len(scores), 1)
    if s >= 2.5:
        return "high"
    if s >= 1.5:
        return "medium"
    return "low"


def _compact_lag(
    label: str,
    payload: Optional[Dict[str, Any]],
    extra_keys: tuple = (),
) -> Dict[str, Any]:
    if not payload:
        return {"label": label, "available": False}
    out: Dict[str, Any] = {
        "label": label,
        "available": payload.get("correlation") is not None,
        "correlation": payload.get("correlation"),
        "best_lag_days": payload.get("best_lag_days", payload.get("best_lag")),
        "confidence": payload.get("confidence"),
        "interpretation": payload.get("interpretation", ""),
        "data_points": payload.get("data_points"),
    }
    for k in extra_keys:
        if k in payload:
            out[k] = payload[k]
    return out