Funk-lab/stt/server/stt_server/sources.py

"""Sources de données live injectées dans le contexte d'Asa (lecture seule).

Chaque fonction renvoie un bloc de texte court (ou une note d'indisponibilité) à
injecter dans le prompt. Tout est best-effort : une source injoignable renvoie une
note explicative, jamais une exception (on n'ajoute jamais de latence fatale au tour).

URLs et jeton via variables d'environnement (cf. config.Settings) → surchargeables
selon la résolution réseau in-cluster.
"""

from __future__ import annotations

import httpx

from stt_server.config import settings

# Alerte toujours active (test du pipeline) — exclue des résumés.
_ALWAYS_ON = {"Watchdog"}


async def _prom_query(client: httpx.AsyncClient, expr: str) -> list[tuple[dict, float]]:
    r = await client.get(
        f"{settings.prometheus_url.rstrip('/')}/api/v1/query",
        params={"query": expr}, timeout=settings.sources_timeout,
    )
    r.raise_for_status()
    data = r.json()
    out: list[tuple[dict, float]] = []
    for s in data.get("data", {}).get("result", []):
        try:
            out.append((s.get("metric", {}), float(s["value"][1])))
        except (KeyError, IndexError, ValueError, TypeError):
            pass
    return out


def _money(v) -> str:
    return f"{round(float(v)):,}".replace(",", " ")


def _pct(v) -> float | None:
    """Normalise un pourcentage Ghostfolio → toujours en POURCENT.

    Ghostfolio rend ses ratios tantôt en fraction (0.12 = 12 %), tantôt déjà en
    pourcent (12) selon la version/le champ. Heuristique : |v| ≤ 1.5 ⇒ fraction.
    """
    try:
        v = float(v)
    except (TypeError, ValueError):
        return None
    return v * 100 if -1.5 <= v <= 1.5 else v


async def _ghostfolio_data(client: httpx.AsyncClient, token: str | None):
    """Auth + GET /portfolio/details → dict brut Ghostfolio, ou une str d'erreur.

    `token` : jeton fourni par le client (prioritaire) ; à défaut, jeton serveur (env).
    Source UNIQUE de l'appel Ghostfolio (réutilisée par le bloc contexte, la phrase
    vocale et l'aperçu structuré du portail).
    """
    base = settings.ghostfolio_url.rstrip("/")
    token = token or settings.ghostfolio_token
    if not token:
        return "Données indisponibles : aucun jeton Ghostfolio (ni client, ni serveur)."
    try:
        r = await client.post(
            f"{base}/api/v1/auth/anonymous",
            json={"accessToken": token}, timeout=settings.sources_timeout,
        )
        if r.status_code >= 400:
            return "Données indisponibles : authentification Ghostfolio refusée (jeton ?)."
        jwt = (r.json() or {}).get("authToken")
        if not jwt:
            return "Données indisponibles : jeton Ghostfolio invalide."
        r = await client.get(
            f"{base}/api/v1/portfolio/details",
            headers={"Authorization": f"Bearer {jwt}"}, timeout=settings.sources_timeout,
        )
        r.raise_for_status()
        return r.json() or {}
    except httpx.HTTPError:
        return "Données indisponibles : Ghostfolio injoignable."
    except (ValueError, KeyError):
        return "Données indisponibles : réponse Ghostfolio inattendue."


async def _ghostfolio_fetch(client: httpx.AsyncClient, token: str | None):
    """Récupère le portefeuille → (value, currency, holdings) ou une str d'erreur."""
    data = await _ghostfolio_data(client, token)
    if isinstance(data, str):
        return data
    summary = data.get("summary", {}) or {}
    val = (summary.get("currentValueInBaseCurrency")
           or summary.get("currentValue") or summary.get("netWorth"))
    cur = summary.get("baseCurrency") or "EUR"
    holdings = data.get("holdings", {}) or {}
    return (val, cur, holdings)


async def ghostfolio_block(client: httpx.AsyncClient, token: str | None = None) -> str:
    """Bloc de contexte (valeur + positions principales)."""
    res = await _ghostfolio_fetch(client, token)
    if isinstance(res, str):
        return res
    val, cur, holdings = res
    tops = sorted(
        (h for h in holdings.values() if isinstance(h, dict)),
        key=lambda h: h.get("valueInBaseCurrency") or 0, reverse=True,
    )[:5]
    lines = []
    if val is not None:
        lines.append(f"Valeur actuelle du portefeuille : {_money(val)} {cur}.")
    for h in tops:
        name = h.get("name") or h.get("symbol") or "?"
        hv = h.get("valueInBaseCurrency")
        if hv:
            lines.append(f"- {name} : {_money(hv)}")
    return "\n".join(lines) if lines else "Aucune donnée de portefeuille renvoyée."


async def ghostfolio_phrase(client: httpx.AsyncClient, token: str | None = None) -> str:
    """Phrase parlable (intent vocal « combien sur mon ghostfolio »)."""
    res = await _ghostfolio_fetch(client, token)
    if isinstance(res, str):
        return res
    val, cur, _ = res
    if val is None:
        return "Je n'ai pas trouvé la valeur du portefeuille dans la réponse de Ghostfolio."
    return f"Ton portefeuille Ghostfolio vaut environ {_money(val)} {cur}."


async def ghostfolio_details(client: httpx.AsyncClient, token: str | None = None) -> dict:
    """Aperçu STRUCTURÉ du portefeuille (page détail du portail HUD).

    Renvoie {value, currency, performance_pct, holdings:[{name, symbol, value,
    allocation_pct, performance_pct}]} ou {error: …}. Lecture seule, best-effort.
    """
    data = await _ghostfolio_data(client, token)
    if isinstance(data, str):
        return {"error": data}
    summary = data.get("summary", {}) or {}
    val = (summary.get("currentValueInBaseCurrency")
           or summary.get("currentValue") or summary.get("netWorth"))
    cur = summary.get("baseCurrency") or "EUR"
    perf = _pct(
        summary.get("currentNetPerformancePercent")
        or summary.get("netPerformancePercent")
        or summary.get("currentNetPerformancePercentWithCurrencyEffect")
    )
    holdings = data.get("holdings", {}) or {}
    rows = sorted(
        (h for h in holdings.values() if isinstance(h, dict)),
        key=lambda h: h.get("valueInBaseCurrency") or 0, reverse=True,
    )
    out = []
    for h in rows[:12]:
        hv = h.get("valueInBaseCurrency")
        if not hv:
            continue
        out.append({
            "name": h.get("name") or h.get("symbol") or "?",
            "symbol": h.get("symbol") or "",
            "value": round(float(hv)),
            "allocation_pct": _pct(h.get("allocationInPercentage")),
            "performance_pct": _pct(h.get("netPerformancePercent")),
        })
    return {
        "value": round(float(val)) if val is not None else None,
        "currency": cur,
        "performance_pct": perf,
        "holdings": out,
    }


async def alerts_block(client: httpx.AsyncClient) -> str:
    base = settings.alertmanager_url.rstrip("/")
    try:
        r = await client.get(
            f"{base}/api/v2/alerts",
            params={"active": "true", "silenced": "false"}, timeout=settings.sources_timeout,
        )
        r.raise_for_status()
        raw = r.json()
    except (httpx.HTTPError, ValueError):
        return "Données indisponibles : Alertmanager injoignable."
    alerts = []
    for a in raw if isinstance(raw, list) else []:
        labels = a.get("labels", {}) or {}
        if labels.get("alertname") in _ALWAYS_ON:
            continue
        ann = a.get("annotations", {}) or {}
        alerts.append(
            f"- {labels.get('alertname', 'alerte')} "
            f"({labels.get('severity', 'none')}) : "
            f"{ann.get('summary') or ann.get('description') or ''}".strip()
        )
    if not alerts:
        return "Aucune alerte active actuellement."
    return f"{len(alerts)} alerte(s) active(s) :\n" + "\n".join(alerts)


async def cluster_block(client: httpx.AsyncClient) -> str:
    try:
        nodes = await _prom_query(client, 'up{job=~"storage-01|gpu-01-node"}')
        knodes = await _prom_query(
            client, 'count(kube_node_status_condition{condition="Ready",status="true"} == 1)'
        )
        ready = await _prom_query(client, 'count(kube_pod_status_ready{condition="true"} == 1)')
        # « Non prêt » = ready false ET pod ENCORE actif (Running/Pending). Sans le filtre de
        # phase, les pods de CronJob terminés (Succeeded/Failed — ex. sacrifice-assign-renfort)
        # comptent comme « non prêts » → fausse alarme. Le join exclut les pods terminés.
        not_ready = await _prom_query(
            client,
            'kube_pod_status_ready{condition="false"} == 1'
            ' and on(namespace,pod) kube_pod_status_phase{phase=~"Running|Pending"} == 1',
        )
        crash = await _prom_query(
            client, 'kube_pod_container_status_waiting_reason{reason="CrashLoopBackOff"} == 1'
        )
    except httpx.HTTPError:
        return "Données indisponibles : Prometheus injoignable."
    hosts_up = sum(1 for _, v in nodes if v == 1)
    lines = [f"Hôtes hors-cluster joignables : {hosts_up}/{len(nodes) or 2} (storage-01, gpu-01)."]
    if knodes:
        lines.append(f"Nœuds k8s Ready : {int(knodes[0][1])}.")
    if ready:
        lines.append(f"Pods prêts : {int(ready[0][1])}.")
    if not_ready:
        names = ", ".join(m.get("pod", "?") for m, _ in not_ready[:6] if m.get("pod"))
        lines.append(f"Pods NON prêts : {names or len(not_ready)}.")
    else:
        lines.append("Tous les pods actifs sont prêts.")
    if crash:
        cnames = ", ".join(sorted({m.get("pod", "?") for m, _ in crash[:6]}))
        lines.append(f"⚠️ CrashLoopBackOff : {cnames}.")
    return "\n".join(lines)


async def metrics_block(client: httpx.AsyncClient) -> str:
    try:
        llama = await _prom_query(client, 'up{job="llama-server-gpu"}')
        targets_up = await _prom_query(client, 'count(up == 1)')
        targets_all = await _prom_query(client, 'count(up)')
        gpu_util = await _prom_query(client, 'rocm_gpu_utilization_percent')
        gpu_temp = await _prom_query(client, 'rocm_gpu_temperature_celsius')
        gpu_vram = await _prom_query(client, '100*rocm_vram_used_bytes/rocm_vram_total_bytes')
    except httpx.HTTPError:
        return "Données indisponibles : Prometheus injoignable."
    lines = []
    if llama:
        lines.append("llama-server (GPU) : " + ("en ligne" if llama[0][1] == 1 else "hors ligne") + ".")
    if gpu_util or gpu_temp or gpu_vram:
        g = []
        if gpu_util:
            g.append(f"utilisation {round(gpu_util[0][1])}%")
        if gpu_temp:
            g.append(f"{round(gpu_temp[0][1])}°C")
        if gpu_vram:
            g.append(f"VRAM {round(gpu_vram[0][1])}%")
        lines.append("GPU RX 6700XT : " + ", ".join(g) + ".")
    if targets_up and targets_all:
        lines.append(f"Cibles Prometheus UP : {int(targets_up[0][1])}/{int(targets_all[0][1])}.")
    return "\n".join(lines) if lines else "Aucune métrique disponible."


# Aiguillage source_id → fetcher
_FETCHERS = {
    "ghostfolio": ghostfolio_block,
    "alerts": alerts_block,
    "cluster": cluster_block,
    "metrics": metrics_block,
}


async def fetch_blocks(
    client: httpx.AsyncClient,
    sources: tuple[str, ...],
    secrets: dict | None = None,
) -> list[tuple[str, str]]:
    """Récupère les blocs live demandés (hors « docs », géré par le RAG). Best-effort.

    `secrets` : jetons fournis par le client (ex. {"ghostfolio_token": …}) → priorité
    sur les jetons serveur (env). Permet d'unifier la source du jeton côté client.
    """
    import asyncio

    secrets = secrets or {}
    wanted = [s for s in sources if s in _FETCHERS]
    if not wanted:
        return []

    def _call(s):
        if s == "ghostfolio":
            return ghostfolio_block(client, secrets.get("ghostfolio_token"))
        return _FETCHERS[s](client)

    results = await asyncio.gather(
        *(_call(s) for s in wanted), return_exceptions=True
    )
    out: list[tuple[str, str]] = []
    for src, res in zip(wanted, results):
        out.append((src, res if isinstance(res, str) else ""))
    return out