app/tools.py
Rama main · réplica de lectura del repositorio original.
import asyncio
import os
import re
import httpx
def _service_urls() -> dict[str, str]:
"""Services surveillés : config via VEKTOR_SERVICES (format
`nom=url,nom=url`). Aucune IP privée n'est codée en dur."""
raw = os.environ.get("VEKTOR_SERVICES", "")
services: dict[str, str] = {}
for item in raw.split(","):
if "=" in item:
name, url = item.split("=", 1)
services[name.strip()] = url.strip()
return services
SERVICE_URLS = _service_urls()
PVE_API_URL = os.environ.get("PVE_API_URL", "https://PVE_HOST:8006")
PVE_API_TOKEN = os.environ.get("PVE_API_TOKEN", "")
PVE_VERIFY_SSL = os.environ.get("PVE_VERIFY_SSL", "0") == "1"
PVE_SSH_HOST = os.environ.get("PVE_SSH_HOST", "PVE_HOST")
PVE_SSH_USER = os.environ.get("PVE_SSH_USER", "root")
PVE_SSH_KEY = os.environ.get("PVE_SSH_KEY", "/app/secrets/vektor_pve_ed25519")
OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://172.20.0.1:11434")
OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "qwen2.5:14b")
# Latences LLM réelles (secondes), in-memory, fenêtre glissante :
# alimenté par graph.py à chaque vraie inférence, lu par model_card().
LLM_LATENCIES: list[float] = []
_LLM_LATENCY_WINDOW = 20
def _fmt_gib(num_bytes: float | None) -> str:
if num_bytes is None:
return "?"
return f"{num_bytes / 2**30:.1f}G"
def record_llm_latency(seconds: float) -> None:
"""Mémorise une latence d'inférence réelle (fenêtre glissante)."""
LLM_LATENCIES.append(seconds)
del LLM_LATENCIES[:-_LLM_LATENCY_WINDOW]
def _host_hw_facts() -> dict:
"""Lecture /proc : le noyau et la RAM visibles sont ceux du host."""
facts: dict = {"cores": "?", "ram_gib": "?", "cpu_pct": None}
try:
facts["cores"] = str(os.cpu_count() or "?")
with open("/proc/meminfo") as fh:
for line in fh:
if line.startswith("MemTotal:"):
facts["ram_gib"] = f"{int(line.split()[1]) / 2**20:.0f}"
break
def _cpu_times() -> tuple[int, int] | None:
with open("/proc/stat") as fh:
for line in fh:
if line.startswith("cpu "):
vals = [int(v) for v in line.split()[1:8]]
return sum(vals), vals[3] + vals[4]
return None
first = _cpu_times()
if first:
import time as _time
_time.sleep(0.3)
second = _cpu_times()
if second and second[0] > first[0]:
idle_ratio = (second[1] - first[1]) / (second[0] - first[0])
facts["cpu_pct"] = max(0.0, (1 - idle_ratio) * 100)
except (OSError, ValueError):
pass
return facts
async def model_card() -> str:
"""Fiche technique réelle : modèle Ollama, hardware, latence mesurée.
Données brutes des API Ollama et de /proc — aucun appel LLM."""
lines: list[str] = ["🧠 **VEKTOR — Fiche technique**", f"\n🤖 **Modèle** : {OLLAMA_MODEL} via Ollama"]
try:
async with httpx.AsyncClient(timeout=8) as client:
show_response = await client.post(f"{OLLAMA_BASE_URL}/api/show", json={"name": OLLAMA_MODEL})
ps_response = await client.get(f"{OLLAMA_BASE_URL}/api/ps")
show = show_response.json() if show_response.status_code == 200 else {}
running = ps_response.json() if ps_response.status_code == 200 else {}
except (httpx.HTTPError, ValueError):
return "\n".join(lines) + "\n\n🔴 Ollama injoignable — fiche détaillée indisponible."
details = show.get("details", {}) or {}
if details:
lines.append(
f"• famille {details.get('family', '?')}, "
f"{details.get('parameter_size', '?')} paramètres, "
f"quantification {details.get('quantization_level', '?')}"
)
loaded = next(
(m for m in running.get("models", []) if str(m.get("name", "")).startswith(OLLAMA_MODEL.split(":")[0])),
None,
)
if loaded:
size_gib = (loaded.get("size") or 0) / 2**30
lines.append(f"• état : **chargé en RAM** ({size_gib:.1f}G), réponses rapides")
else:
lines.append("• état : non chargé (rechargement ~1 min au prochain message LLM)")
hw = _host_hw_facts()
cpu_now = f", usage CPU {hw['cpu_pct']:.0f}% à l'instant" if hw["cpu_pct"] is not None else ""
lines.append(f"\n💻 **Hardware (host)** : {hw['cores']} cœurs, {hw['ram_gib']}G RAM{cpu_now}")
if LLM_LATENCIES:
avg = sum(LLM_LATENCIES) / len(LLM_LATENCIES)
lines.append(
f"\n⏱️ **Latence LLM** : {avg:.1f}s en moyenne sur les {len(LLM_LATENCIES)} "
"dernières réponses générées (les contrôles live restent instantanés)"
)
else:
lines.append("\n⏱️ **Latence LLM** : pas encore de mesure depuis le dernier redémarrage")
lines.append("\n📡 **Canaux** : Telegram (texte, actions avec confirmation OUI) · Alexa (vocal, lecture seule)")
lines.append("🔒 **Périmètre** : lecture seule sur l'infra, écriture uniquement après OUI explicite")
return "\n".join(lines)
async def _pve_get(path: str):
if not PVE_API_TOKEN:
return None
try:
async with httpx.AsyncClient(
timeout=40,
verify=PVE_VERIFY_SSL,
headers={"Authorization": f"PVEAPIToken={PVE_API_TOKEN}"},
) as client:
response = await client.get(f"{PVE_API_URL}/api2/json{path}")
if response.status_code != 200:
return None
return response.json().get("data")
except httpx.HTTPError:
return None
async def _pve_node() -> str | None:
nodes = await _pve_get("/nodes")
if not nodes:
return None
for node in nodes:
if node.get("status") == "online":
return node["node"]
return nodes[0]["node"] if nodes else None
async def pve_summary() -> str:
node = await _pve_node()
if not node:
return "PVE : API injoignable ou non configurée."
status = await _pve_get(f"/nodes/{node}/status")
if not status:
return "PVE : status du noeud indisponible."
mem = status.get("memory", {}) or {}
swap = status.get("swap", {}) or {}
mem_pct = mem.get("used", 0) / max(mem.get("total", 1), 1) * 100
swap_pct = swap.get("used", 0) / max(swap.get("total", 1), 1) * 100
# loadavg PVE = liste de strings ; on garde la charge 1 min en nombre propre
try:
load1 = float((status.get("loadavg") or ["0"])[0])
except (ValueError, TypeError, IndexError):
load1 = 0.0
lines = [
f"Noeud {node} : CPU {(status.get('cpu') or 0) * 100:.0f}%, charge {load1:.1f}",
f"RAM {_fmt_gib(mem.get('used'))}/{_fmt_gib(mem.get('total'))} ({mem_pct:.0f}%)",
f"Swap {_fmt_gib(swap.get('used'))}/{_fmt_gib(swap.get('total'))} ({swap_pct:.0f}%)",
f"Uptime {(status.get('uptime') or 0) / 86400:.1f} jours",
]
return "\n".join(lines)
async def pve_lxc_status() -> str:
node = await _pve_node()
if not node:
return "PVE : API injoignable ou non configurée."
cts = await _pve_get(f"/nodes/{node}/lxc")
if not cts:
return "PVE : liste des CTs indisponible."
lines = ["CTs Proxmox :"]
for ct in sorted(cts, key=lambda item: item.get("vmid", 0)):
mem = ct.get("mem", 0)
maxmem = ct.get("maxmem", 0)
mem_pct = f"{mem / maxmem * 100:.0f}%" if maxmem else "?"
lines.append(
f"CT {ct.get('vmid')} {ct.get('name', '')} : {ct.get('status')}, "
f"RAM {_fmt_gib(mem)}/{_fmt_gib(maxmem)} ({mem_pct}), "
f"uptime {(ct.get('uptime') or 0) / 3600:.0f} h"
)
return "\n".join(lines)
async def pve_storage_status() -> str:
node = await _pve_node()
if not node:
return "PVE : API injoignable ou non configurée."
storages = await _pve_get(f"/nodes/{node}/storage")
if not storages:
return "PVE : liste des stockages indisponible."
lines = ["Stockages PVE :"]
for storage in sorted(storages, key=lambda item: item.get("storage", "")):
total = storage.get("total") or 0
used = storage.get("used") or 0
if total <= 0:
continue
pct = used / total * 100
flag = " ⚠️" if pct >= 90 else ""
lines.append(
f"{storage.get('storage')} : {_fmt_gib(used)}/{_fmt_gib(total)} ({pct:.0f}%){flag}"
)
return "\n".join(lines)
async def _status_channel(remote_cmd: str) -> str:
"""Canal SSH lecture seule du PVE : commande fermee (vektor-status [*]).
La cle ne peut lancer que /usr/local/bin/vektor-status, qui refuse tout
argument autre que ceux prevus."""
if not os.path.exists(PVE_SSH_KEY):
return "Canal SSH lecture seule non configuré."
try:
proc = await asyncio.create_subprocess_exec(
"ssh",
"-i", PVE_SSH_KEY,
"-o", "BatchMode=yes",
"-o", "StrictHostKeyChecking=accept-new",
"-o", "UserKnownHostsFile=/tmp/known_hosts",
"-o", "ConnectTimeout=15",
f"{PVE_SSH_USER}@{PVE_SSH_HOST}",
remote_cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=100)
except (asyncio.TimeoutError, OSError):
return "Canal SSH lecture seule indisponible pour le moment."
text = stdout.decode(errors="replace").strip()
return text if text else "Sortie vide du canal lecture seule."
async def docker_inventory() -> str:
return await _status_channel("vektor-status")
async def seeds_report() -> str:
"""Rapport seeds : stats qBittorrent live + staging récupérable (cache).
Lecture seule, aucun LLM, aucune écriture."""
return await _status_channel("vektor-status seeds")
async def jellyfin_search(query: str) -> str:
"""Recherche un film ou une série dans la bibliothèque Jellyfin.
Lecture seule : /Items avec searchTerm. Clé via VEKTOR_JELLYFIN_TOKEN,
URL du service via VEKTOR_SERVICES (jellyfin=...)."""
url = SERVICE_URLS.get("jellyfin")
token = os.environ.get("VEKTOR_JELLYFIN_TOKEN", "")
if not url or not token:
return "Recherche Jellyfin non configurée (VEKTOR_JELLYFIN_TOKEN)."
term = query.strip()
if not term:
return "Recherche Jellyfin : titre vide."
try:
async with httpx.AsyncClient(timeout=8) as client:
response = await client.get(
f"{url.rstrip('/')}/Items",
# Jellyfin 12+ refuse l'ancien header X-Emby-Token (401) :
# la forme Authorization MediaBrowser est la voie valide.
headers={"Authorization": f'MediaBrowser Token="{token}"'},
params={
"searchTerm": term,
"Recursive": "true",
"IncludeItemTypes": "Movie,Series",
"Limit": "5",
},
)
if response.status_code != 200:
return f"Jellyfin : recherche impossible (HTTP {response.status_code})."
items = response.json().get("Items") or []
except httpx.HTTPError as exc:
return f"Jellyfin : injoignable ({exc.__class__.__name__})."
if not items:
return f"Aucun titre trouvé pour « {term} » dans la bibliothèque Jellyfin."
lines = [f"Bibliothèque Jellyfin — {len(items)} résultat(s) pour « {term} » :"]
for item in items:
kind = "Film" if item.get("Type") == "Movie" else "Série"
year = item.get("ProductionYear") or "?"
lines.append(f"- {kind} : {item.get('Name', '?')} ({year})")
return "\n".join(lines)
async def check_service(name: str) -> str:
url = SERVICE_URLS.get(name.lower())
if not url:
return f"Service inconnu ou non autorisé : {name}"
try:
async with httpx.AsyncClient(timeout=5, follow_redirects=False) as client:
response = await client.get(url)
return f"{name}: HTTP {response.status_code} depuis le contrôle live."
except httpx.HTTPError as exc:
return f"{name}: indisponible ({exc.__class__.__name__})."
_STATE_WORDS = ["état", "etat", "status", "statut", "tourne", "tournent", "liste",
"inventaire", "up", "down", "marche", "santé", "sante"]
# Mots qui signalent une demande de donnée chiffrée ("quel est", "combien de")
# plutot qu'une explication : déclenchent aussi la collecte live.
_METRIC_WORDS = ["quel", "quelles", "quels", "combien", "montre", "donne"]
_EXPLAIN_WORDS = ["explique", "pourquoi", "comment", "c'est quoi", "qu'est-ce", "qu est ce", "difference", "différence", "définis", "define"]
async def full_status_report() -> str:
"""Rapport homelab complet formaté Telegram, sans LLM, données live."""
lines: list[str] = ["🏠 **VEKTOR — Rapport homelab**"]
node = await _pve_node()
if node:
status = await _pve_get(f"/nodes/{node}/status")
if status:
mem = status.get("memory", {}) or {}
swap = status.get("swap", {}) or {}
mem_pct = mem.get("used", 0) / max(mem.get("total", 1), 1) * 100
swap_pct = swap.get("used", 0) / max(swap.get("total", 1), 1) * 100
load = status.get("loadavg") or ["?"]
if isinstance(load, str):
load = load.split()
lines.append(
f"\n⚙️ **PVE** : charge {load[0] if load else '?'} | "
f"RAM {mem_pct:.0f}% | swap {swap_pct:.0f}% | "
f"up {(status.get('uptime') or 0) / 86400:.1f} j"
)
cts = await _pve_get(f"/nodes/{node}/lxc")
if cts:
running = [c for c in cts if c.get("status") == "running"]
lines.append(f"\n📦 **CTs** : {len(running)}/{len(cts)} running")
for ct in sorted(cts, key=lambda item: item.get("vmid", 0)):
icon = "🟢" if ct.get("status") == "running" else "🔴"
maxmem = ct.get("maxmem") or 0
mem_pct_ct = (ct.get("mem") or 0) / maxmem * 100 if maxmem else 0
lines.append(
f"{icon} {ct.get('vmid')} {ct.get('name', '')} : "
f"{ct.get('status')}, RAM {mem_pct_ct:.0f}%"
)
storages = await _pve_get(f"/nodes/{node}/storage")
if storages:
lines.append("\n💾 **Stockages**")
for storage in sorted(storages, key=lambda item: item.get("storage", "")):
total = storage.get("total") or 0
used = storage.get("used") or 0
if total <= 0:
continue
pct = used / total * 100
icon = "⚠️" if pct >= 90 else "•"
lines.append(f"{icon} {storage.get('storage')} : {pct:.0f}% ({_fmt_gib(total - used)} libres)")
else:
lines.append("\n🔴 **PVE** : API injoignable")
inv = await docker_inventory()
if not inv.startswith("Canal") and not inv.startswith("Sortie"):
up_count = inv.count("Up ")
exited = inv.lower().count("exited")
lines.append(f"\n🐳 **Docker** : {up_count} conteneurs Up" + (f", {exited} exited ⚠️" if exited else ""))
dead = [l for l in inv.splitlines() if "exited" in l.lower() or "restarting" in l.lower()]
for entry in dead[:5]:
lines.append(f"⚠️ {entry.strip()}")
else:
lines.append("\n🐳 **Docker** : canal indisponible")
lines.append("\n🌐 **Services**")
for name in ["jellyfin", "sonarr", "radarr", "qbittorrent"]:
result = await check_service(name)
ok = "HTTP" in result and not result.startswith(name + ": indisponible")
icon = "🟢" if ok else "🔴"
lines.append(f"{icon} {name} : {result.split(': ', 1)[1] if ': ' in result else result}")
from datetime import datetime, timezone, timedelta
now = datetime.now(timezone(timedelta(hours=2)))
lines.append(f"\n_ Rapport généré le {now.strftime('%d/%m à %H:%M')} — lecture seule_")
return "\n".join(lines)
async def infra_live_report(text: str) -> str | None:
low = text.lower()
parts: list[str] = []
# Multi-tâches : « check l'état de jellyfin et dis-moi si j'ai le film
# cars » — un court-circuit ici ne traiterait que la 1re partie et la
# 2e question serait perdue. Toute demande combinée va au LLM, qui
# orchestre toutes les parties avec ses outils.
if re.search(
r"\b(dis[- ]?moi|cherche|trouve|est[- ]ce que|si j'ai|si j ai|puis|ensuite|aussi|peux[- ]tu|fais[- ]moi)\b",
low,
):
return None
services = [name for name in SERVICE_URLS if name in low]
if services:
for name in services[:3]:
parts.append(await check_service(name))
# Info externe (v1.8.0) : cloudflare / ip publique / horloge — sans LLM
if re.search(r"\b(cloudflare|cf\s+status|ip\s+publique|adresse\s+ip\s+publique|d[ée]rive\s+horloge)\b", low):
from . import extinfo
if re.search(r"\bcloudflare|cf\s+status\b", low):
parts.append(await extinfo.cloudflare_status())
if re.search(r"\bip\s+publique|adresse\s+ip\s+publique\b", low):
parts.append(await extinfo.public_ip())
if re.search(r"\bd[ée]rive\s+horloge\b", low):
parts.append(await extinfo.clock_check())
if parts:
return "\n\n".join(p for p in parts if p)
is_explanation = any(word in low for word in _EXPLAIN_WORDS)
full = any(k in low for k in ["homelab", "état général", "etat general", "vue d'ensemble"])
# Alexa n'envoie que le slot extrait (ex: "proxmox", sans "état") :
# un sujet nu et court est traite comme une demande d'etat.
bare_subject = len(low.split()) <= 5 and any(re.search(
r"\b(proxmox|pve|docker|conteneur|container|stack|arr|conteneurs?|cts|lxc|disques?|stockage|espace|ram|swap|cpu|charge)\b", low
) or name in low for name in SERVICE_URLS)
state = (any(word in low for word in _STATE_WORDS) or bare_subject) and not is_explanation
# "quel est / combien de / quelle est" : une demande de donnée chiffrée,
# pas une explication -> déclenche la collecte live comme un mot d'état.
wants_metric = any(re.search(rf"\b{w}\b", low) for w in _METRIC_WORDS)
want_node = (re.search(
r"\b(proxmox|pve|hyperviseur|noeud|node|charge|cpu|swap|uptime)\b", low
) or re.search(r"\bram\b|\bm[ée]moire\b", low)) and (state or wants_metric or full)
want_cts = bool(
re.search(r"\bct\s?\d+\b|\bcts\b|\blxc\b", low)
or ("conteneur" in low and "proxmox" in low)
) and (state or wants_metric or full)
want_docker = ("docker" in low and state) or (
re.search(r"\b(conteneurs?|containers?|stacks?|arr)\b", low) and (state or wants_metric) and not services
)
want_storage = re.search(r"\b(disque|disques|stockage|espace)\b", low) and (state or wants_metric or full)
if full:
parts.append(await pve_summary())
parts.append(await pve_lxc_status())
parts.append(await docker_inventory())
parts.append(await pve_storage_status())
else:
if want_node:
parts.append(await pve_summary())
if want_cts:
parts.append(await pve_lxc_status())
if want_docker:
parts.append(await docker_inventory())
if want_storage:
parts.append(await pve_storage_status())
if not parts:
return None
return "\n\n".join(p for p in parts if p)