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(",", " ")


async def _ghostfolio_fetch(client: httpx.AsyncClient, token: str | None):
    """Récupère le portefeuille → (value, currency, holdings) ou une str d'erreur.

    `token` : jeton fourni par le client (prioritaire) ; à défaut, jeton serveur (env).
    """
    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()
        data = r.json() or {}
        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)
    except httpx.HTTPError:
        return "Données indisponibles : Ghostfolio injoignable."
    except (ValueError, KeyError):
        return "Données indisponibles : réponse Ghostfolio inattendue."


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 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"}')
        not_ready = await _prom_query(
            client, 'kube_pod_status_ready{condition="false"} == 1'
        )
        ready = await _prom_query(client, 'count(kube_pod_status_ready{condition="true"} == 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 ready:
        lines.append(f"Pods prêts (cluster) : {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 scrutés sont prêts.")
    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)')
    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 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