mirror of
https://github.com/dredx/prole.git
synced 2026-09-28 11:04:30 +00:00
93f7259722
- update Kong/Supabase ingress and service manifests for k3s - expand init scripts and runtime config/topology handling - refresh KNØE UI screens and Supabase deployment/render logic - include related env helper test adjustments Co-authored-by: Junie <junie@jetbrains.com>
371 lines
12 KiB
Python
371 lines
12 KiB
Python
"""Common UI rendering helpers shared across all screens."""
|
|
|
|
import os
|
|
import subprocess
|
|
import tkinter as tk
|
|
from datetime import datetime
|
|
from tkinter import ttk, messagebox, filedialog
|
|
|
|
from knoe import screen as ui
|
|
|
|
|
|
class ScreenBaseMixin:
|
|
"""Common UI rendering helpers shared across all screens."""
|
|
|
|
def _render_title(self, text, y=40):
|
|
ui.render_title(self, text, y)
|
|
|
|
def _render_paragraph(self, text, y, wrap=600):
|
|
ui.render_paragraph(self, text, y, wrap)
|
|
|
|
def _create_console_output(self, y=280, title="Scan Output", width=880, height=450, x=48):
|
|
"""Create a standardized console output area with a label and scrollable text."""
|
|
# Section title/label for the console
|
|
ui.canvas_text(
|
|
self, x, y, title, fill="#1d1d1f", font=("SF Pro Text", 12, "bold")
|
|
)
|
|
|
|
y_console = y + 30
|
|
# Use a background frame to ensure NO borders are visible around the console
|
|
console_bg = tk.Frame(self.bg_canvas, bg="white", highlightthickness=0, bd=0)
|
|
console = ui.TerminalConsole(console_bg, highlightthickness=0, bd=0)
|
|
console.pack(fill="both", expand=True, padx=1, pady=1)
|
|
|
|
console_window = self.bg_canvas.create_window(
|
|
x, y_console, window=console_bg, anchor="nw", width=width, height=height
|
|
)
|
|
self._canvas_items.append(console_window)
|
|
self._overlay_widgets.append(console_bg)
|
|
self._overlay_widgets.append(console)
|
|
return console
|
|
|
|
def _is_k3s_mode_active(self):
|
|
"""Return True when current mode resolves to k3s/service cluster."""
|
|
mode = (os.environ.get("PROLE_MODE") or "").strip().lower()
|
|
if mode == "k3s":
|
|
return True
|
|
try:
|
|
return (self.cluster_env.get() or "").strip().lower() == "service"
|
|
except Exception:
|
|
return False
|
|
|
|
def _fetch_k3s_nodes(self):
|
|
"""Return a list of node names from kubectl get nodes in k3s mode."""
|
|
cmd = ["kubectl", "get", "nodes", "-o", "name"]
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=20,
|
|
)
|
|
names = []
|
|
for line in (proc.stdout or "").splitlines():
|
|
line = (line or "").strip()
|
|
if not line:
|
|
continue
|
|
if line.startswith("node/"):
|
|
line = line.split("/", 1)[1]
|
|
names.append(line)
|
|
if not names:
|
|
raise ValueError("no nodes returned")
|
|
return names
|
|
except Exception:
|
|
fallback = []
|
|
for candidate in ("gandalf.prole.org", "myrddin.prole.org"):
|
|
if candidate not in fallback:
|
|
fallback.append(candidate)
|
|
return fallback
|
|
|
|
def _is_tainted_node(self, node_name):
|
|
"""Return True if the node has taints and should be visually disabled."""
|
|
cmd = [
|
|
"kubectl",
|
|
"get",
|
|
"node",
|
|
node_name,
|
|
"-o",
|
|
"jsonpath={.spec.taints[*].key}",
|
|
]
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
)
|
|
return bool((proc.stdout or "").strip())
|
|
except Exception:
|
|
return node_name == "myrddin.prole.org"
|
|
|
|
def _render_k3s_node_selector(
|
|
self,
|
|
*,
|
|
y,
|
|
target_var,
|
|
title="Node selector:",
|
|
x_label=48,
|
|
x_options=190,
|
|
option_spacing=175,
|
|
):
|
|
"""Render a reusable k3s node selector and return the next y offset."""
|
|
if not self._is_k3s_mode_active():
|
|
return y
|
|
|
|
item = ui.canvas_text(
|
|
self,
|
|
x_label,
|
|
y,
|
|
title,
|
|
fill="black",
|
|
font=("SF Pro Text", 12),
|
|
)
|
|
self._canvas_items.append(item)
|
|
|
|
selected = (target_var.get() or "").strip()
|
|
nodes = self._fetch_k3s_nodes()
|
|
x = x_options
|
|
y_radio = y - 2
|
|
for node in nodes:
|
|
disabled = self._is_tainted_node(node)
|
|
rb = tk.Radiobutton(
|
|
self.bg_canvas,
|
|
text=node,
|
|
variable=target_var,
|
|
value=node,
|
|
bg="white",
|
|
fg="#8c8c8c" if disabled else "black",
|
|
activebackground="white",
|
|
selectcolor="white",
|
|
state="disabled" if disabled else "normal",
|
|
font=("SF Pro Text", 10),
|
|
)
|
|
window = self.bg_canvas.create_window(x, y_radio, window=rb, anchor="nw")
|
|
self._canvas_items.append(window)
|
|
self._overlay_widgets.append(rb)
|
|
x += option_spacing
|
|
|
|
if (not selected or selected not in nodes or self._is_tainted_node(selected)) and nodes:
|
|
for node in nodes:
|
|
if not self._is_tainted_node(node):
|
|
target_var.set(node)
|
|
break
|
|
|
|
return y + 36
|
|
|
|
def safe_after(self, func, delay=0):
|
|
"""Run a function in the main thread if the root window still exists."""
|
|
if not self.root or not self.root.winfo_exists():
|
|
return
|
|
|
|
def wrapper():
|
|
try:
|
|
if self.root and self.root.winfo_exists():
|
|
func()
|
|
except (tk.TclError, RuntimeError):
|
|
pass
|
|
|
|
try:
|
|
self.root.after(delay, wrapper)
|
|
except (tk.TclError, RuntimeError):
|
|
pass
|
|
|
|
def _clear_canvas_page(self):
|
|
# Unbind common events to prevent "echo" or persistent behavior from previous pages
|
|
try:
|
|
self.bg_canvas.unbind("<Button-1>")
|
|
self.root.unbind("<Return>")
|
|
self.root.unbind("<KP_Enter>")
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
self._stop_k3s_service_status_updates()
|
|
except Exception:
|
|
pass
|
|
|
|
# Reset slide_area background to white for safety
|
|
try:
|
|
self.slide_area.configure(bg="white")
|
|
except Exception:
|
|
pass
|
|
|
|
# Remove any small overlay widgets from the previous page
|
|
try:
|
|
if getattr(self, "_overlay_widgets", None):
|
|
for w in list(self._overlay_widgets):
|
|
try:
|
|
w.destroy()
|
|
except Exception:
|
|
try:
|
|
w.place_forget()
|
|
except Exception:
|
|
pass
|
|
self._overlay_widgets.clear()
|
|
except Exception:
|
|
pass
|
|
|
|
# Clear page-specific canvas drawings (keep the background image)
|
|
if self._canvas_items:
|
|
for item in self._canvas_items:
|
|
try:
|
|
self.bg_canvas.delete(item)
|
|
except Exception:
|
|
pass
|
|
self._canvas_items.clear()
|
|
|
|
# ---------------- Environment setup helpers ----------------
|
|
def _register_canvas_renderer(self, page_id, func):
|
|
self.canvas_renderers[page_id] = func
|
|
|
|
def _register_page(self, page_id, frame):
|
|
self.pages.append((page_id, frame))
|
|
self.page_frames[page_id] = frame
|
|
|
|
def show_page(self, index_or_id):
|
|
self._showing_service_overlay = False
|
|
# Resolve index and page_id
|
|
old_idx = self.page_index
|
|
if isinstance(index_or_id, int):
|
|
idx = max(0, min(index_or_id, len(self.pages) - 1))
|
|
else:
|
|
# Find index by page_id
|
|
idx = -1
|
|
for i, (pid, _) in enumerate(self.pages):
|
|
if pid == index_or_id:
|
|
idx = i
|
|
break
|
|
|
|
if idx == -1:
|
|
msg = f"Navigation error: page ID '{index_or_id}' not found."
|
|
print(f"[ERROR] {msg}")
|
|
try:
|
|
messagebox.showerror("Navigation Error", msg)
|
|
except Exception:
|
|
pass
|
|
return
|
|
|
|
print(f"[DEBUG] Navigating from {old_idx} to {idx} (requested: {index_or_id})")
|
|
self.page_index = idx
|
|
pid, frame = self.pages[self.page_index]
|
|
|
|
# Clear any previously drawn canvas content
|
|
self._clear_canvas_page()
|
|
self._update_nav_highlight(pid)
|
|
|
|
# If the page has overlay widgets, we need to show the slide_area.
|
|
# Otherwise, we hide it so the canvas content is visible.
|
|
# Pages that use canvas drawing and need to show the background should have slide_area hidden.
|
|
canvas_only_pages = (
|
|
"welcome",
|
|
"network_scan",
|
|
"env_setup",
|
|
"kerberos_config",
|
|
"deps_summary",
|
|
"init_cluster",
|
|
"cluster_nodes",
|
|
"common_services",
|
|
"init_password",
|
|
"init_db_build",
|
|
"init_cnpg_deploy",
|
|
"init_scripts",
|
|
"build",
|
|
"disk_selection",
|
|
"build_summary",
|
|
"supabase_config",
|
|
"gitops_config",
|
|
"argocd_config",
|
|
"database_options",
|
|
"create_installer",
|
|
)
|
|
if pid in canvas_only_pages or pid.startswith("dep_"):
|
|
self.slide_area.place_forget()
|
|
else:
|
|
self.slide_area.place(relx=0, rely=0, relwidth=1, relheight=1)
|
|
self.slide_area.lift()
|
|
|
|
# Render the page directly on the canvas if we have a renderer
|
|
if pid in self.canvas_renderers:
|
|
try:
|
|
self.canvas_renderers[pid]()
|
|
except Exception as e:
|
|
# Fallback: show an error message on canvas
|
|
ui.canvas_text(
|
|
self,
|
|
32,
|
|
32,
|
|
f"Error rendering page '{pid}': {e}",
|
|
fill="black",
|
|
font=("SF Pro Text", 11),
|
|
)
|
|
elif frame is not None:
|
|
# Legacy fallback (should not be used)
|
|
frame.place(
|
|
relx=0.5, rely=0.5, anchor="center", relwidth=0.94, relheight=0.9
|
|
)
|
|
self.update_footer()
|
|
|
|
# ---------------- Canvas page rendering ----------------
|
|
def show_screen(self, screen_id):
|
|
self.show_page(screen_id)
|
|
|
|
def _run_cmd_capture(self, cmd, env=None, stdin_text=None, timeout=None):
|
|
output_lines = []
|
|
try:
|
|
# Ensure PYTHONUNBUFFERED=1
|
|
run_env = (env or os.environ).copy()
|
|
run_env["PYTHONUNBUFFERED"] = "1"
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
stdin=subprocess.PIPE if stdin_text is not None else None,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
env=run_env,
|
|
)
|
|
except Exception as e:
|
|
return 1, f"Failed to start command: {e}"
|
|
|
|
if stdin_text is not None and proc.stdin:
|
|
try:
|
|
proc.stdin.write(stdin_text)
|
|
proc.stdin.close()
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
for line in proc.stdout:
|
|
output_lines.append(line)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
rc = proc.wait(timeout=timeout)
|
|
except Exception:
|
|
try:
|
|
proc.kill()
|
|
except Exception:
|
|
pass
|
|
rc = 1
|
|
return rc, "".join(output_lines)
|
|
|
|
def _parse_rfc3339(self, ts: str):
|
|
if not ts:
|
|
return None
|
|
try:
|
|
if ts.endswith("Z"):
|
|
ts = ts[:-1] + "+00:00"
|
|
return datetime.fromisoformat(ts)
|
|
except Exception:
|
|
return None
|
|
|
|
def _format_pod_time(self, ts: str) -> str:
|
|
dt = self._parse_rfc3339(ts)
|
|
if not dt:
|
|
return ts or "—"
|
|
try:
|
|
return dt.astimezone().strftime("%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
return dt.strftime("%Y-%m-%d %H:%M:%S")
|