atlasbot/atlasbot/snapshot/builder/summary_text.py

73 lines
2.7 KiB
Python

from __future__ import annotations
from typing import Any
from .core_a import *
from .core_b import *
from .format_a import *
from .format_b import *
from .format_c import *
def summary_text(snapshot: dict[str, Any] | None) -> str:
"""Render the snapshot summary into deterministic prompt text."""
summary = build_summary(snapshot)
if not summary:
return ""
lines: list[str] = []
lines.append("atlas_cluster: Titan Lab Atlas Kubernetes cluster (internal).")
collected_at = snapshot.get("collected_at") if isinstance(snapshot, dict) else None
snapshot_version = snapshot.get("snapshot_version") if isinstance(snapshot, dict) else None
if collected_at or snapshot_version:
bits = []
if collected_at:
bits.append(f"collected_at={collected_at}")
if snapshot_version:
bits.append(f"version={snapshot_version}")
lines.append("snapshot: " + ", ".join(bits))
_append_nodes(lines, summary)
_append_hardware(lines, summary)
_append_hardware_groups(lines, summary)
_append_lexicon(lines, summary)
_append_pressure(lines, summary)
_append_node_facts(lines, summary)
_append_node_ages(lines, summary)
_append_node_taints(lines, summary)
_append_capacity(lines, summary)
_append_pods(lines, summary)
_append_namespace_pods(lines, summary)
_append_namespace_nodes(lines, summary)
_append_node_pods(lines, summary)
_append_pod_issues(lines, summary)
_append_pod_issue_summary(lines, summary)
_append_workload_health(lines, summary)
_append_events(lines, summary)
_append_node_usage_stats(lines, summary)
_append_namespace_usage(lines, summary)
_append_namespace_requests(lines, summary)
_append_namespace_io_net(lines, summary)
_append_pod_usage(lines, summary)
_append_restarts(lines, summary)
_append_job_failures(lines, summary)
_append_jobs(lines, summary)
_append_postgres(lines, summary)
_append_hottest(lines, summary)
_append_pvc_usage(lines, summary)
_append_root_disk_headroom(lines, summary)
_append_namespace_capacity_summary(lines, summary)
_append_baseline_deltas(lines, summary)
_append_longhorn(lines, summary)
_append_workloads(lines, summary)
_append_topology(lines, summary)
_append_workloads_by_namespace(lines, summary)
_append_node_load_summary(lines, summary)
_append_cluster_watchlist(lines, summary)
_append_hardware_usage(lines, summary)
_append_cross_stats(lines, summary)
_append_flux(lines, summary)
_append_signals(lines, summary)
_append_profiles(lines, summary)
_append_units_windows(lines, summary)
return "\n".join(lines)