#!/usr/bin/env python3 # -*- coding: utf-8 -*- import json import os import subprocess import sys import threading import tkinter as tk from tkinter import filedialog, messagebox, ttk from mass_deploy import ( DeployBackend, SCRIPT_DIR, build_runtime_config, ensure_admin, list_windows_interfaces, load_config, ) APP_TITLE = "ButtonTask SSH Deploy" CONFIG_PATH = os.path.join(SCRIPT_DIR, "config.json") README_PATH = os.path.join(SCRIPT_DIR, "README.md") LOGS_DIR = os.path.join(SCRIPT_DIR, "logs") def read_text(path, default=""): try: with open(path, "r", encoding="utf-8") as f: return f.read() except Exception: return default class DeployGui(tk.Tk): def __init__(self): super().__init__() self.title(APP_TITLE) self.geometry("1100x760") self.minsize(950, 680) self.backend = None self.interfaces = [] self.base_cfg = load_config(CONFIG_PATH) self.nic_var = tk.StringVar() self.bundle_var = tk.StringVar(value=self.base_cfg.get("bundle_path", "")) self.summary_var = tk.StringVar(value="Готово к запуску.") self.stats_var = tk.StringVar(value="DHCP: rx=0 offer=0 ack=0") self._build_ui() self.refresh_interfaces() self.load_instruction() def _build_ui(self): top = ttk.Frame(self, padding=10) top.pack(fill="x") top.columnconfigure(1, weight=1) ttk.Label(top, text="Интерфейс").grid(row=0, column=0, sticky="w", padx=(0, 8), pady=4) self.nic_combo = ttk.Combobox(top, textvariable=self.nic_var, state="readonly") self.nic_combo.grid(row=0, column=1, sticky="ew", pady=4) ttk.Button(top, text="Обновить", command=self.refresh_interfaces).grid(row=0, column=2, padx=6, pady=4) ttk.Label(top, text="Бандл .tgz").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=4) ttk.Entry(top, textvariable=self.bundle_var).grid(row=1, column=1, sticky="ew", pady=4) ttk.Button(top, text="Выбрать", command=self.pick_bundle).grid(row=1, column=2, padx=6, pady=4) actions = ttk.Frame(top) actions.grid(row=2, column=0, columnspan=3, sticky="w", pady=(10, 0)) self.start_btn = ttk.Button(actions, text="Start", command=self.start_backend) self.start_btn.pack(side="left") self.stop_btn = ttk.Button(actions, text="Stop", command=self.stop_backend, state="disabled") self.stop_btn.pack(side="left", padx=6) ttk.Button(actions, text="Open logs", command=self.open_logs).pack(side="left", padx=6) status = ttk.LabelFrame(self, text="Состояние", padding=10) status.pack(fill="x", padx=10, pady=(0, 10)) ttk.Label(status, textvariable=self.summary_var).pack(anchor="w") ttk.Label(status, textvariable=self.stats_var).pack(anchor="w", pady=(4, 0)) center = ttk.Panedwindow(self, orient="vertical") center.pack(fill="both", expand=True, padx=10, pady=(0, 10)) table_wrap = ttk.Labelframe(center, text="Устройства", padding=8) center.add(table_wrap, weight=3) self.tree = ttk.Treeview( table_wrap, columns=("mac", "ip", "state", "ts", "detail"), show="headings", height=14, ) self.tree.heading("mac", text="MAC") self.tree.heading("ip", text="IP") self.tree.heading("state", text="Статус") self.tree.heading("ts", text="Время") self.tree.heading("detail", text="Детали") self.tree.column("mac", width=150, anchor="w") self.tree.column("ip", width=120, anchor="w") self.tree.column("state", width=120, anchor="w") self.tree.column("ts", width=80, anchor="center") self.tree.column("detail", width=580, anchor="w") self.tree["displaycolumns"] = ("mac", "ip", "state", "ts", "detail") self.tree.tag_configure("ok", background="#dff6dd") self.tree.tag_configure("warn", background="#fff4ce") self.tree.tag_configure("err", background="#fde7e9") self.tree.tag_configure("work", background="#e5f1fb") self.tree.pack(fill="both", expand=True) yscroll = ttk.Scrollbar(table_wrap, orient="vertical", command=self.tree.yview) self.tree.configure(yscrollcommand=yscroll.set) yscroll.place(relx=1.0, rely=0, relheight=1.0, anchor="ne") note_wrap = ttk.Labelframe(center, text="Инструкция для оператора", padding=8) center.add(note_wrap, weight=2) self.note = tk.Text(note_wrap, wrap="word", height=14) self.note.pack(fill="both", expand=True) self.note.configure(state="disabled") def load_instruction(self): short = [ "Подключение", "1. Используйте изолированную сеть через отдельный свитч.", "2. Настройте сетевой адаптер ПК и подключите его к этому свитчу.", "3. Вставьте в свитч одно или несколько пустых устройств.", "", "Запуск", "1. Выберите сетевой интерфейс.", "2. Выберите .tgz бандл.", "3. Нажмите Start.", "", "Что делает программа", "- выдаёт устройствам IP по DHCP;", "- подключается по SSH;", "- копирует бандл;", "- запускает установку пакетов и обновление файлов;", "- применяет sudoers и служебные скрипты из бандла;", "- записывает итоговую сеть устройства;", "- выключает устройство, чтобы избежать конфликтов после установки.", "", "Операторский цикл", "Настроить -> включить -> вставить -> дождаться ГОТОВО -> вынуть -> повторить.", "", "Статусы", "- синий: процесс идёт;", "- зелёный: устройство готово;", "- жёлтый: промежуточное состояние или ожидание;", "- красный: ошибка, детали смотрите в таблице и logs.", ] self.note.configure(state="normal") self.note.delete("1.0", "end") self.note.insert("1.0", "\n".join(short)) self.note.configure(state="disabled") def refresh_interfaces(self): self.interfaces = list_windows_interfaces() labels = [x["label"] for x in self.interfaces] self.nic_combo["values"] = labels if labels and self.nic_var.get() not in labels: preferred = next((x["label"] for x in self.interfaces if x["ip"] == self.base_cfg.get("server_ip")), labels[0]) self.nic_var.set(preferred) def pick_bundle(self): path = filedialog.askopenfilename( title="Выберите .tgz бандл", filetypes=[("TGZ bundle", "*.tgz"), ("All files", "*.*")], initialdir=SCRIPT_DIR, ) if path: self.bundle_var.set(path) def selected_interface(self): label = self.nic_var.get().strip() for item in self.interfaces: if item["label"] == label: return item return None def start_backend(self): if self.backend and self.backend.running: return iface = self.selected_interface() if not iface: messagebox.showerror(APP_TITLE, "Выберите сетевой интерфейс.") return bundle = self.bundle_var.get().strip() if not bundle: messagebox.showerror(APP_TITLE, "Выберите .tgz бандл.") return cfg = build_runtime_config(self.base_cfg, interface_ip=iface["ip"], bundle_path=bundle) try: ensure_admin() self.backend = DeployBackend(CONFIG_PATH, cfg=cfg) except Exception as e: messagebox.showerror(APP_TITLE, str(e)) return self.backend.add_observer(lambda snapshot: self.after(0, self.apply_snapshot, snapshot)) self.backend.start() self.start_btn.configure(state="disabled") self.stop_btn.configure(state="normal") self.summary_var.set(f"Запущено: {iface['label']}") def stop_backend(self): if self.backend: self.backend.stop() self.start_btn.configure(state="normal") self.stop_btn.configure(state="disabled") self.summary_var.set("Остановлено.") def open_logs(self): os.makedirs(LOGS_DIR, exist_ok=True) if sys.platform.startswith("win"): os.startfile(LOGS_DIR) else: subprocess.Popen(["xdg-open", LOGS_DIR]) def apply_snapshot(self, snapshot): stats = snapshot["dhcp"] self.stats_var.set( f"DHCP: rx={stats['rx']} offer={stats['offer']} ack={stats['ack']}" + (f" err={stats['last_err']}" if stats.get("last_err") else "") ) if snapshot.get("error"): self.summary_var.set(snapshot["error"]) if self.backend and not self.backend.running: self.start_btn.configure(state="normal") self.stop_btn.configure(state="disabled") rows = snapshot["rows"] existing = set(self.tree.get_children()) for mac, row in sorted(rows.items()): values = (mac, row.get("ip", ""), row.get("state", ""), row.get("ts", ""), row.get("detail", "")) state = (row.get("state") or "").lower() if state in {"готово"}: tags = ("ok",) elif state in {"ошибка"}: tags = ("err",) elif state in {"halt", "ребут", "discover", "acked"}: tags = ("warn",) else: tags = ("work",) if mac in existing: self.tree.item(mac, values=values, tags=tags) existing.remove(mac) else: self.tree.insert("", "end", iid=mac, text=mac, values=values, tags=tags) for stale in existing: self.tree.delete(stale) def main(): app = DeployGui() app.mainloop() if __name__ == "__main__": main()