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


async def ghostfolio_block(client: httpx.AsyncClient) -> str:
    base = settings.ghostfolio_url.rstrip("/")
    token = settings.ghostfolio_token
    if not token:
        return "Données indisponibles : jeton Ghostfolio non configuré côté 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"))
        # quelques positions principales (par valeur), si présentes
        holdings = data.get("holdings", {}) or {}
        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 : {round(float(val)):,} (devise de base).".replace(",", " "))
        for h in tops:
            name = h.get("name") or h.get("symbol") or "?"
            hv = h.get("valueInBaseCurrency")
            if hv:
                lines.append(f"- {name} : {round(float(hv)):,}".replace(",", " "))
        return "\n".join(lines) if lines else "Aucune donnée de portefeuille renvoyée."
    except httpx.HTTPError:
        return "Données indisponibles : Ghostfolio injoignable."
    except (ValueError, KeyError):
        return "Données indisponibles : réponse Ghostfolio inattendue."


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, ...]) -> list[tuple[str, str]]:
    """Récupère les blocs live demandés (hors « docs », géré par le RAG). Best-effort."""
    import asyncio

    wanted = [s for s in sources if s in _FETCHERS]
    if not wanted:
        return []
    results = await asyncio.gather(
        *(_FETCHERS[s](client) 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