mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
ce5ff83eb4
- Fix kube context switching for k3s single-context kubeconfigs and k3d shorthand prefixes - Update common init/status scripts (registry, kerberos, cnpg backup, service layer, common services) - Add Gitea init script and installer ArgoCD screen - Add Supabase realtime probe patching plus regression tests - Extend installer core/UI test coverage
254 lines
8.4 KiB
Python
254 lines
8.4 KiB
Python
"""Common UI rendering helpers shared across all screens."""
|
|
|
|
import os
|
|
import subprocess
|
|
from datetime import datetime
|
|
import tkinter as tk
|
|
from tkinter import ttk, messagebox, filedialog
|
|
from installer.screen import TerminalConsole
|
|
from installer 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 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",
|
|
"init_password",
|
|
"init_db_build",
|
|
"init_cnpg_deploy",
|
|
"init_scripts",
|
|
"build",
|
|
"disk_selection",
|
|
"build_summary",
|
|
"supabase_config",
|
|
"ollama_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")
|