prole/knoe/ui/screens/environment.py

688 lines
26 KiB
Python

"""System environment setup screen and env.sh management."""
import os
import shlex
import shutil
import subprocess
from pathlib import Path
import tkinter as tk
from tkinter import ttk, messagebox, filedialog
from knoe import screen as ui
from knoe.core.env import PROJECT_ROOT, _expand_cfg_value, resolve_prole_home
class EnvironmentScreenMixin:
"""System environment setup screen and env.sh management."""
def _env_setup_defaults_raw(self) -> dict[str, str]:
"""Default values to present to the operator.
These are intentionally variable-friendly for readability in the UI and
`env.sh` (e.g. `$HOME`, `$PROLE_HOME`). Filesystem operations should
always use `_resolve_env_paths_for_fs()`.
"""
return {
"PROLE_HOME": "$HOME/.prole",
"PROLE_CONF": "$PROLE_HOME/conf",
"PROLE_DATA": "$PROLE_HOME/data",
"PROLE_LOGS": "$PROLE_HOME/logs",
"PROLE_SERVICE": "$PROLE_HOME/etc",
}
def _apply_env_entry_style(self, key: str) -> None:
ent = (getattr(self, "_env_entries", {}) or {}).get(key)
if not ent:
return
try:
is_default = bool((getattr(self, "_env_entry_is_default", {}) or {}).get(key))
except Exception:
is_default = False
try:
ent.configure(fg="#6e6e73" if is_default else "black")
except Exception:
pass
def _sync_env_default_state(self, key: str) -> None:
"""Recompute whether an env entry is at its default and restyle it."""
ent = (getattr(self, "_env_entries", {}) or {}).get(key)
if not ent:
return
try:
cur = (ent.get() or "").strip()
except Exception:
cur = ""
defaults = getattr(self, "_env_defaults_raw", None) or {}
default_val = (defaults.get(key) or "").strip()
if not hasattr(self, "_env_entry_is_default") or self._env_entry_is_default is None:
self._env_entry_is_default = {}
self._env_entry_is_default[key] = bool(cur and default_val and cur == default_val)
self._apply_env_entry_style(key)
def _expand_shell_path(self, value: str | Path, env: dict[str, str] | None = None) -> str:
raw = str(value or "").strip()
if not raw:
return ""
env_map = env or os.environ
expanded = _expand_cfg_value(raw, env=env_map, max_depth=10)
return os.path.expanduser(expanded)
def _resolve_env_paths_for_fs(self, values: dict) -> dict[str, str]:
"""Expand shell variables in PROLE_* paths for filesystem operations.
Keeps the caller's original `values` intact so env.sh can preserve
user-friendly expressions like `$HOME/...`.
"""
raw = {k: str(v or "").strip() for k, v in (values or {}).items()}
env_map = dict(os.environ)
env_map.update({k: v for k, v in raw.items() if v})
resolved: dict[str, str] = {}
home = self._expand_shell_path(raw.get("PROLE_HOME", ""), env=env_map)
if home:
resolved["PROLE_HOME"] = home
env_map["PROLE_HOME"] = home
for k in ("PROLE_CONF", "PROLE_DATA", "PROLE_LOGS", "PROLE_SERVICE"):
v = raw.get(k, "")
if v:
resolved[k] = self._expand_shell_path(v, env=env_map)
return resolved
def _render_env_setup_page(self):
# Letterhead at top right
content_width = self.bg_canvas.winfo_width() or 975
right_margin = content_width - 48
ui.canvas_text(
self,
right_margin,
40,
"knoe.dev",
fill="#6e6e73",
font=("SF Pro Text", 32, "bold"),
anchor="ne",
)
ui.canvas_text(
self,
right_margin,
85,
"infrastructure.auto()",
fill="#6e6e73",
font=("SF Pro Text", 18),
anchor="ne",
)
# Task heading for clarity
self._render_title("Environment", y=150)
self._render_paragraph(
"Configure your Prole environment. You can accept defaults or choose custom locations. Changing PROLE_HOME will update the defaults for other paths.",
y=200,
)
defaults_raw = self._env_setup_defaults_raw()
self._env_defaults_raw = dict(defaults_raw)
defaults_resolved = self._resolve_env_paths_for_fs(defaults_raw)
existing = {}
try:
existing = self._read_existing_env() or {}
except Exception:
existing = {}
# Prefer showing the default expressions in the UI even when the existing
# env.sh contains literal paths that resolve to the same locations.
values = dict(defaults_raw)
try:
for k, v in dict(existing).items():
v = str(v or "").strip()
if not v:
continue
tmp = dict(values)
tmp[k] = v
resolved_v = (self._resolve_env_paths_for_fs(tmp) or {}).get(k, "")
if resolved_v and resolved_v == (defaults_resolved.get(k) or ""):
continue
values[k] = v
except Exception:
# Best-effort merging; fall back to defaults.
values = dict(defaults_raw)
self._env_entry_is_default = {}
try:
self._last_prole_home_resolved = (
self._resolve_env_paths_for_fs(values).get("PROLE_HOME")
or defaults_resolved.get("PROLE_HOME")
or ""
)
except Exception:
self._last_prole_home_resolved = ""
# Keep entry widgets to read values on Next
self._env_entries = {}
labels = [
("PROLE_HOME", "Base directory for data, configs, logs, and scripts"),
("PROLE_CONF", "Configuration directory"),
("PROLE_DATA", "Data directory"),
("PROLE_LOGS", "Logs directory"),
("PROLE_SERVICE", "System scripts directory"),
]
x_label = 48
x_field = 300
x_btn = 850
y = 280
row_h = 75
def browse(var_name: str):
from tkinter import filedialog
initial = self._env_entries[var_name].get() or Path.home()
path = filedialog.askdirectory(
title=f"Select {var_name}",
initialdir=self._expand_shell_path(initial, env=os.environ),
)
if path:
self._env_entries[var_name].delete(0, "end")
self._env_entries[var_name].insert(0, path)
self._env_entry_is_default[var_name] = False
self._apply_env_entry_style(var_name)
if var_name == "PROLE_HOME":
# Update dependent defaults if they still match old pattern
self._update_dependent_env_paths()
# Draw rows
for key, help_text in labels:
self._canvas_items.append(
ui.canvas_text(
self,
x_label,
y,
key,
fill="black",
font=("SF Pro Text", 12, "bold"),
anchor="w",
)
)
# Use tk.Entry and place on canvas via create_window
entry = tk.Entry(
self.bg_canvas,
bg="white",
fg="black",
insertbackground="black",
highlightbackground="#CCCCCC",
highlightthickness=1,
relief="flat",
font=("SF Pro Text", 11),
)
entry.insert(0, values[key])
entry_window = self.bg_canvas.create_window(
x_field, y, window=entry, anchor="w", width=500, height=32
)
self._canvas_items.append(entry_window)
self._overlay_widgets.append(entry)
self._env_entries[key] = entry
# Default-state styling
self._env_entry_is_default[key] = bool(
(values.get(key) or "").strip() == (defaults_raw.get(key) or "").strip()
)
self._apply_env_entry_style(key)
try:
entry.bind("<KeyRelease>", lambda _e, k=key: self._sync_env_default_state(k))
entry.bind("<FocusOut>", lambda _e, k=key: self._sync_env_default_state(k))
except Exception:
pass
# Use tk.Button and place on canvas via create_window
btn = tk.Button(
self.bg_canvas,
text="Browse…",
command=lambda k=key: browse(k),
bg="#F5F5DC",
fg="black",
activebackground="#E5E5D5",
highlightbackground="#F5F5DC",
highlightthickness=0,
relief="flat",
font=("SF Pro Text", 11),
padx=10,
pady=4,
)
btn_window = self.bg_canvas.create_window(
x_btn, y, window=btn, anchor="w", width=120
)
self._canvas_items.append(btn_window)
self._overlay_widgets.append(btn)
# hint text
self._canvas_items.append(
ui.canvas_text(
self,
x_field,
y + 24,
help_text,
fill="#6e6e73",
font=("SF Pro Text", 10),
anchor="nw",
)
)
y += row_h
# When PROLE_HOME changes, update dependent paths that still follow default pattern
def on_home_change(_evt=None):
self._update_dependent_env_paths()
try:
self._env_entries["PROLE_HOME"].bind("<FocusOut>", on_home_change)
except Exception:
pass
def _after_env_saved(self):
"""Reload the installer process environment after saving env.sh."""
try:
self.reload_env_from_shell()
self._save_prole_cfg()
except Exception as e:
try:
messagebox.showwarning(
"Environment", f"Environment saved, but reload failed: {e}"
)
except Exception:
pass
def resolve_prole_home(self) -> Path:
"""Return PROLE_HOME or default to `$HOME/.prole` as a Path (expanded)."""
return resolve_prole_home()
def ensure_prole_env(self) -> Path:
"""Ensure $PROLE_HOME and env.sh exist and are readable. Create minimal env if missing.
Returns the resolved PROLE_HOME Path. Raises Exception on failure.
"""
home = self.resolve_prole_home()
try:
home.mkdir(parents=True, exist_ok=True)
except Exception as e:
raise Exception(f"Failed to create PROLE_HOME at {home}: {e}")
env_file = home / "env.sh"
if not env_file.exists():
# Construct minimal values based on UI defaults.
# NOTE: Values are variable-friendly for readability; filesystem
# operations always resolve literals.
vals = dict(self._env_setup_defaults_raw())
try:
default_home = (self._resolve_env_paths_for_fs(vals) or {}).get("PROLE_HOME", "")
except Exception:
default_home = ""
if default_home and str(home) != default_home:
# Keep dependent defaults relative to PROLE_HOME, but set
# PROLE_HOME itself to the resolved literal location.
vals["PROLE_HOME"] = str(home)
self._save_env_to_file(vals)
# Validate readability
try:
data = env_file.read_text()
if not isinstance(data, str) or len(data) == 0:
raise Exception("env.sh is empty")
except Exception as e:
raise Exception(f"env.sh not readable at {env_file}: {e}")
return home
def reload_env_from_shell(self) -> None:
"""Reload environment by sourcing $PROLE_HOME/env.sh in a login shell and merging into os.environ."""
home = self.ensure_prole_env()
env_file = home / "env.sh"
# Use bash to source and print env in null-delimited form
cmd = (
f"export PROLE_HOME={shlex.quote(str(home))}; "
f"source {shlex.quote(str(env_file))}; "
"env -0"
)
try:
out = subprocess.check_output(["bash", "-lc", cmd])
except subprocess.CalledProcessError as e:
raise Exception(f"Failed to reload environment: {e}")
except Exception as e:
raise Exception(f"Failed to run bash to reload environment: {e}")
# Merge variables
try:
items = out.split(b"\x00")
for raw in items:
if not raw:
continue
try:
kv = raw.decode("utf-8", errors="ignore")
except Exception:
continue
if "=" not in kv:
continue
k, v = kv.split("=", 1)
# Avoid clobbering Python internals we rely on; safe list only
if k in ("PYTHONPATH", "PYTHONHOME"):
continue
os.environ[k] = v
except Exception:
# Best-effort; ignore merge errors
pass
def _resolve_prole_conf_dir(self) -> Path:
return self._resolve_env_dir("PROLE_CONF", "conf")
def _resolve_prole_logs_dir(self) -> Path:
return self._resolve_env_dir("PROLE_LOGS", "logs")
def _registry_log_path(self) -> Path:
logs_dir = self._resolve_prole_logs_dir()
try:
logs_dir.mkdir(parents=True, exist_ok=True)
except Exception:
pass
path = logs_dir / "local_registry.log"
self._last_registry_log_path = path
return path
def _common_services_log_path(self) -> Path:
logs_dir = self._resolve_prole_logs_dir()
try:
logs_dir.mkdir(parents=True, exist_ok=True)
except Exception:
pass
path = logs_dir / "init_common_services.log"
self._last_common_services_log_path = path
return path
def _read_log_tail(self, path: Path, max_lines: int = 400) -> str:
try:
from collections import deque
with path.open("r", encoding="utf-8", errors="ignore") as handle:
tail = deque(handle, maxlen=max_lines)
return "".join(tail)
except Exception:
return ""
def _restore_common_services_log(self, console) -> bool:
path = getattr(self, "_last_common_services_log_path", None)
if not path:
candidate = self._common_services_log_path()
if candidate.exists():
path = candidate
if not path or not path.exists():
return False
content = self._read_log_tail(path)
if not content:
return False
try:
console.clear()
console.write(f"Log file: {path}\n\n")
console.write(content)
except Exception:
return False
return True
def _ensure_prole_directories(self) -> None:
paths = {
"PROLE_HOME": self._resolve_env_value(
"PROLE_HOME", str(resolve_prole_home(env={}))
),
"PROLE_CONF": str(self._resolve_prole_conf_dir()),
"PROLE_DATA": str(self._resolve_env_dir("PROLE_DATA", "data")),
"PROLE_LOGS": str(self._resolve_prole_logs_dir()),
"PROLE_SERVICE": str(self._resolve_env_dir("PROLE_SERVICE", "etc")),
}
env_map = dict(os.environ)
env_map.update({k: str(v or "").strip() for k, v in paths.items() if v})
for key, raw in paths.items():
try:
expanded = self._expand_shell_path(raw, env=env_map)
if expanded:
Path(expanded).mkdir(parents=True, exist_ok=True)
except Exception:
pass
if key not in os.environ and raw:
# Prefer expanded values inside the installer process to avoid
# accidental creation of literal `$HOME`/`$PROLE_HOME` paths.
os.environ[key] = self._expand_shell_path(raw, env=env_map)
def _record_install_log(self, path) -> None:
if not path:
return
try:
p = Path(path)
except Exception:
return
if not hasattr(self, "_install_run_log_paths"):
self._install_run_log_paths = []
sp = str(p)
if sp not in self._install_run_log_paths:
self._install_run_log_paths.append(sp)
def _update_env_namespace(self, namespace: str):
# Namespace is persisted to prole.cfg only — never to env.sh or os.environ.
pass
def _save_env_to_file(self, values: dict):
resolved = self._resolve_env_paths_for_fs(values)
home = Path(resolved.get("PROLE_HOME") or values["PROLE_HOME"]).expanduser()
# ensure base dir exists
home.mkdir(parents=True, exist_ok=True)
# ensure subdirs exist
for key in ("PROLE_CONF", "PROLE_DATA", "PROLE_LOGS", "PROLE_SERVICE"):
try:
target = resolved.get(key) or values.get(key, "")
if target:
Path(target).expanduser().mkdir(parents=True, exist_ok=True)
except Exception:
pass
# write env.sh atomically
content = []
# Executable wrapper + sourceable config
content.append("#!/usr/bin/env bash")
content.append("# Prole environment configuration")
content.append(
"# This file is generated by the installer. Source it in new shells, or execute as a wrapper:"
)
content.append('# "$PROLE_HOME/env.sh" <command> [args…]')
content.append("# shellcheck shell=bash")
# Core PROLE_* directories
for k in (
"PROLE_HOME",
"PROLE_CONF",
"PROLE_DATA",
"PROLE_LOGS",
"PROLE_SERVICE",
):
content.append(f'export {k}="{values[k]}"')
# NAMESPACE/PROLE_NAMESPACE must never be written to env.sh — only prole.cfg is the source of truth
content.append("")
# Ensure PATH contains common locations and $PROLE_HOME/bin (POSIX sh compatible)
content.append("# Ensure PATH works for GUI-launched shells (Docker, etc.)")
content.append(
'_prole_add_path() { case ":${PATH}:" in *":$1:"*) ;; *) PATH="$1:${PATH:-}" ;; esac; }'
)
content.append('_prole_add_path "$PROLE_HOME/bin"')
content.append('_prole_add_path "/opt/homebrew/bin"')
content.append('_prole_add_path "/usr/local/bin"')
content.append('_prole_add_path "/usr/bin"')
content.append('_prole_add_path "/bin"')
content.append('_prole_add_path "/usr/sbin"')
content.append('_prole_add_path "/sbin"')
content.append("export PATH")
content.append("")
content.append("# Add custom paths below if needed (examples):")
content.append("")
content.append(
"# If executed with arguments (and not sourced), run them under this environment"
)
content.append('if [[ "${BASH_SOURCE[0]}" == "${0}" ]] && [ "$#" -gt 0 ]; then')
content.append(' exec "$@"')
content.append("fi")
new_content = "\n".join(content) + "\n"
out = home / "env.sh"
should_write = True
if out.exists():
if out.read_text() == new_content:
should_write = False
# print(f"[OK] {out} is up to date.")
if should_write:
tmp = home / "env.sh.tmp"
tmp.write_text(new_content)
tmp.replace(out)
try:
os.chmod(out, 0o755)
except Exception:
pass
# print(f"[OK] Wrote {out}")
# After creating env.sh, deploy additional resources as requested:
# 1) Deploy init-port-forward.sh into $PROLE_HOME (and ensure it's executable)
# 2) Copy contents of src/prole/etc (or fallback to top-level etc) into $PROLE_SERVICE
try:
prole_home = Path(resolved.get("PROLE_HOME") or values["PROLE_HOME"]).expanduser()
prole_service = Path(resolved.get("PROLE_SERVICE") or values["PROLE_SERVICE"]).expanduser()
# Determine source for init-port-forward.sh
# Preferred location under repo: PROJECT_ROOT/src/prole/etc/init-port-forward.sh
init_pf_src_candidates = [
PROJECT_ROOT / "src" / "prole" / "etc" / "init-port-forward.sh",
PROJECT_ROOT / "etc" / "init-port-forward.sh",
]
init_pf_src = next((p for p in init_pf_src_candidates if p.exists()), None)
if init_pf_src is not None:
init_pf_dst = prole_home / "init-port-forward.sh"
try:
data = init_pf_src.read_bytes()
init_pf_dst.write_bytes(data)
os.chmod(init_pf_dst, 0o755)
except Exception:
# best effort, ignore errors
pass
# Determine etc directory source
etc_src_candidates = [
PROJECT_ROOT / "src" / "prole" / "etc",
PROJECT_ROOT / "etc",
]
etc_src = next(
(p for p in etc_src_candidates if p.exists() and p.is_dir()), None
)
if etc_src is not None:
try:
prole_service.mkdir(parents=True, exist_ok=True)
except Exception:
pass
# Copy files (non-recursive: contents of etc root). If subdirectories exist, copy them recursively.
import shutil
def _copy_item(src_path: Path, dst_path: Path):
try:
if src_path.is_dir():
# copy directory tree
if dst_path.exists():
# remove then copy to keep in sync
shutil.rmtree(dst_path, ignore_errors=True)
shutil.copytree(src_path, dst_path)
else:
shutil.copy2(src_path, dst_path)
except Exception:
pass
try:
for item in etc_src.iterdir():
_copy_item(item, prole_service / item.name)
except Exception:
pass
except Exception:
# overall best-effort; do not fail env creation if copies fail
pass
def _update_dependent_env_paths(self):
defaults = getattr(self, "_env_defaults_raw", None) or self._env_setup_defaults_raw()
# Capture previous resolved home so we can detect dependent fields that
# were implicitly tied to the old home via literal paths.
old_home = (getattr(self, "_last_prole_home_resolved", "") or "").strip()
raw_now = {}
try:
raw_now = {k: (e.get() or "").strip() for k, e in (self._env_entries or {}).items()}
except Exception:
raw_now = {}
try:
new_home = (self._resolve_env_paths_for_fs(raw_now) or {}).get("PROLE_HOME", "")
except Exception:
new_home = ""
if new_home:
self._last_prole_home_resolved = new_home
mapping = {
"PROLE_CONF": (defaults.get("PROLE_CONF") or "").strip(),
"PROLE_DATA": (defaults.get("PROLE_DATA") or "").strip(),
"PROLE_LOGS": (defaults.get("PROLE_LOGS") or "").strip(),
"PROLE_SERVICE": (defaults.get("PROLE_SERVICE") or "").strip(),
}
suffixes = {
"PROLE_CONF": "conf",
"PROLE_DATA": "data",
"PROLE_LOGS": "logs",
"PROLE_SERVICE": "etc",
}
for key, default_expr in mapping.items():
ent = (self._env_entries or {}).get(key)
if not ent:
continue
try:
cur = (ent.get() or "").strip()
except Exception:
cur = ""
# If empty, fill with the default expression.
if not cur and default_expr:
try:
ent.delete(0, "end")
ent.insert(0, default_expr)
except Exception:
pass
self._env_entry_is_default[key] = True
self._apply_env_entry_style(key)
continue
# If the current value is a literal that matches the previous
# resolved-home default, switch it back to the default expression so
# it follows the new PROLE_HOME.
try:
if old_home and default_expr and "$" not in cur:
resolved_cur = self._expand_shell_path(cur, env=os.environ)
if resolved_cur == str(Path(old_home) / suffixes.get(key, "")):
ent.delete(0, "end")
ent.insert(0, default_expr)
self._env_entry_is_default[key] = True
self._apply_env_entry_style(key)
except Exception:
pass
# Finally, restyle PROLE_HOME itself.
try:
self._sync_env_default_state("PROLE_HOME")
except Exception:
pass