Amc/Blueforce-Linux
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Blueforce-Linux/bfos/ui_gui.py
emirhanmamak-blueforce 503f4df2ae Enhance WireGuard restart playbook with operation wave validation 
- Updated `bf-wireguard-restart.yml` to include dynamic host targeting based on `operation_wave`.
- Added pre-tasks to assert that the operation wave is approved and confirmed, ensuring proper execution context.
- Modified the service restart task to use a more structured command format.
- Adjusted `operations.py` and `ui_gui.py` to reflect changes in operation scope from "simple" to "wave" for relevant operations.
- Improved UI handling in `ui_gui.py` and `ui_surfaces.py` for better output management and user interaction.
- Updated log rotation configuration to target specific log files.
- Enhanced systemd service configurations for WireGuard to improve restart behavior.
- Refined firstboot scripts to ensure proper handling of dealer ID and state management.
- Improved error handling and rollback mechanisms in identity and installation scripts.
2026-09-17 17:06:58 +03:00

4374 lines
186 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""bfos/ui_gui.py -- Blueforce Field OS Turkce masaustu arayuzu (tkinter).
Ne yapar
--------
Python TUI konsolunun yaptigi isi ayni is mantigi uzerinden bir
gercek pencereye tasir:
* basit mod : uc buyuk buton -- Kur / Durum / Cikis (+ "Detayli mod"),
* detayli mod: kategori sekmeleri -- Kurulum, Cihaz islemleri, Merkez
hazirligi, Araclar, Cihaz bilgisi,
* hedef secimi: bayi no alani (^[0-9]{8}$ dogrulamali) + dalga grubu listesi,
* her komut CALISTIRILMADAN ONCE gosterilir ve onay ister; yikici islemlerde
AYRICA ikinci onay istenir, production dalgasi icin "production" yazilir,
* komut ciktisi canli olarak kaydirilabilir bir metin kutusuna akar,
* ekran yoksa (SSH/headless) veya tkinter kurulu degilse net TURKCE hata
verir ve terminal menusunu onerir.
======================================================================
SOZLESME -- backend arareyuzu (bfos/operations.py ve bfos/runner.py icin)
======================================================================
Bu dosya is mantigini TAKLIT ETMEZ; onu bir "backend adapter" uzerinden
kullanir. Gercek moduller (bfos.operations, bfos.runner) yazildiginda asagidaki
sozlesmeyi saglamalari yeterlidir; ui_gui.py onlari otomatik olarak adapte eder.
Sozlesme (bu dosyada tanimli veri tipleriyle ayni alan adlari):
1) Operasyon listesi
list_operations() -> Sequence[Operation]
ya da OPERATIONS / OPS / OPERATION_CATALOGUE isimli bir dizi.
Operation alanlari: id (str, tekil), label (str, TURKCE ad),
category (str, su bes kategoriden biri: "Kurulum", "Cihaz islemleri",
"Merkez hazirligi", "Araclar", "Cihaz bilgisi"), destructive (bool),
scope ("local"|"simple"|"wave"|"release"), description (str),
playbook (str, sadece filo islemleri icin "bf-*.yml" adi).
2) Hedef secimi
Target(kind="dealer"|"group"|"all", value="12010193"|"pilot_1"|"all")
- dealer -> Ansible --limit bf-<8 hane>, insan kimligi BF-<8 hane>
- group -> dalga/grup adi; wave/release kapsamli islemler SADECE onayli
bir dalga grubu kabul eder (lab, pilot_1, pilot_2, wave_1, wave_2,
production); release kapsamli islemler tek cihaz veya "all" KABUL ETMEZ.
3) Komut uretimi (calistirmadan)
build_command(operation, target, options, grants=None, check_mode=False)
-> CommandPlan
CommandPlan:
* commands : her biri bir argv olan demet (satir listesi
`CommandPlan.lines` uzerinden shlex ile uretilir),
* report_lines: SALT-OKUNUR rapor satirlari; commands bos ise hicbir sey
calistirilmaz, satirlar dogrudan gosterilir,
* destructive / requires_root / operation_id / target_label / notes,
* dry_run / check_mode.
options : {secenek_adi: str} (bkz. Operation.options)
grants : {"wave_gate_confirmed"|"production_override"|
"allow_production_reboot"|"fleet_wide": bool} -- operatorun
ACIK onaylari. Bir kapi bayragi yalniz onay verildiginde ve
playbook o degiskeni kullaniyorsa komuta eklenir.
check_mode: True ise komut --check ile uretilir (degisiklik uygulanmaz).
4) Kapi onaylari (acik onay adimlari)
gate_steps(operation, target) -> Sequence[GateStepInfo]
GateStepInfo(kind="token"|"approval", key, prompt, message, token, unlocks)
- "token" : operator kutulara `token` sozcugunu yazmak zorundadir
(dalga adi, "production", "GATE", "all"),
- "approval" : evet/hayir onayi (yikici ikinci onay dahil),
- unlocks : onay verilince build_command'a gecirilecek grant adlari.
Arayuz onaylarin TAMAMINI sorar; hicbir kapi sessizce gecilmez.
5) Komut calistirma (cikti + cikis kodu)
run(plan, on_line=None, cancel=None) -> RunResult(exit_code, ok, output)
on_line(str) her cikti satirinda cagrilir (canli akis icin);
cancel threading.Event ile calisan komut sonlandirilir.
Uyumluluk icin adapter su adlari da dener (sirayla):
katalog : catalogue_rows | ALL_OPERATIONS | list_operations | operations |
get_operations | OPERATIONS | OPS
komut : build_command | build_plan | command_for | build_ansible_cmd
kosum : run | run_plan | run_command | execute
ve donus degerlerini (dict / dataclass / object) normalize eder.
bfos.operations ve bfos.runner MEVCUTSA (bu repodaki gercek moduller) arayuz
dogrudan onlari kullanir; hicbir kural kopyalanmaz:
katalog : catalogue_rows() / ALL_OPERATIONS
envanter/hedef : Repository + Inventory.load_repository + resolve_target
komut uretimi : build_ansible_command / build_install_command /
build_tool_command / build_extra_vars
kapilar : confirmation_plan + GatesGranted
denetim kaydi : runner.AuditLog (+ runner.redact)
salt-okunur ekranlar: targets_lines / central_*_lines / identity_lines /
state_lines / environment_lines
Gercek moduller yoksa dosya kendi YERLESIK KATALOG'unu kullanir: bu katalog
kanonik işlem kataloğuyla (id, playbook, kapsam, yıkıcı, ek değişken
kurallari) birebir ayni kurallara ve ayni komut uretimine dayanir.
======================================================================
Kullanim:
python3 -m bfos.ui_gui # basit mod
python3 -m bfos.ui_gui --full # dogrudan detayli mod
python3 -m bfos.ui_gui --list # pencere acmadan katalogu yazdir
python3 -m bfos.ui_gui --dry-run # komutlari goster, calistirma
Notlar:
* tkinter Ubuntu'da hazir gelir (paket: python3-tk); indirme gerekmez.
* Bu dosya tkinter olmadan da ice aktarilabilir: yardimci fonksiyonlar
(dogrulama, komut onizleme, yikici siniflama, headless tespiti) tkinter'siz
calisir, bu yuzden statik testler pencere ACMADAN kosar.
"""
from __future__ import annotations
import argparse
import json
import os
import queue
import re
import shlex
import shutil
import subprocess
import sys
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence
# ---------------------------------------------------------------------------
# Sabitler -- kullanicinin gordugu TUM metinler Turkce
# ---------------------------------------------------------------------------
WINDOW_TITLE = "Blueforce Field OS — Yönetim"
APP_SUBTITLE = "Blueforce saha cihazı yönetimi (Kur / Durum / Bakım)"
# Basit modda gorunen uc buyuk buton (sira onemli).
SIMPLE_MODE_BUTTONS: tuple[str, ...] = ("Kur", "Durum", "Çıkış")
SIMPLE_MODE_LABEL = "Detaylı mod"
SIMPLE_MODE_BACK_LABEL = "Basit moda dön"
#: Birincil eylemin metni (docs/33 §9.2 ile ayni kalir).
RUN_BUTTON_LABEL = "Komutu göster ve çalıştır"
#: Islem listesindeki tur ve durum sutunlari: renk + SOZCUK (renksiz de okunur).
TYPE_WRITE_LABEL = "YIKICI"
TYPE_READ_LABEL = "salt-okunur"
AVAILABLE_LABEL = "açık"
BLOCKED_LABEL = "kapalı — sudo bf"
#: Komut cikti panelini gizle/goster (kanit erisilebilir kalir, ekrani kaplamaz).
OUTPUT_HIDE_LABEL = "Çıktı panelini gizle"
OUTPUT_SHOW_LABEL = "Çıktı panelini göster"
# Detayli moddaki kategori sekmeleri (docs/33 §9.2 ile ayni bes kategori).
CATEGORIES: tuple[str, ...] = (
"Kurulum",
"Cihaz işlemleri",
"Merkez hazırlığı",
"Araçlar",
"Cihaz bilgisi",
)
#: Kenar cubugu sayfalari: bes katalog kategorisi + oturum etkinlik sayfasi.
#: Kategoriler backend sozlesmesidir (testler sabitler); "Etkinlik" yalnizca
#: bu oturumda calisanlari listeler, denetim kaydini acmaz.
ACTIVITY_PAGE = "Etkinlik"
SIDEBAR_PAGES: tuple[str, ...] = CATEGORIES + (ACTIVITY_PAGE,)
#: Sayfa -> cizgi ikon adi (bfos/ui_widgets.py cizim sozlugu).
PAGE_ICONS: Mapping[str, str] = {
"Kurulum": "install",
"Cihaz işlemleri": "device",
"Merkez hazırlığı": "wave",
"Araçlar": "tool",
"Cihaz bilgisi": "info",
ACTIVITY_PAGE: "activity",
}
#: Hedef kapsami secimi: parca anahtari -> Turkce etiket.
SCOPE_OPTIONS: tuple[tuple[str, str], ...] = (
("device", "Cihaz"),
("wave", "Dalga"),
("all", "Tüm filo"),
)
# Onayli dalga sirasi (docs/10; admin/lib/targets.sh BF_WAVES ile ayni).
WAVES: tuple[str, ...] = ("lab", "pilot_1", "pilot_2", "wave_1", "wave_2", "production")
PRODUCTION_WAVE = "production"
GATE_TOKEN_RELEASE = "GATE"
DEALER_PATTERN = re.compile(r"^[0-9]{8}$")
SCOPE_LABELS: Mapping[str, str] = {
"local": "yerel — bu cihaz",
"simple": "filo — tek cihaz/grup (salt-okunur)",
"wave": "filo — dalga kapılı",
"release": "filo — sürüm kapılı (onaylı dalga)",
}
# ortak kimlik ve hedef kurallarıyla ayni kurallar.
SERVICE_ALLOWLIST: tuple[str, ...] = (
"blueforce-agent",
"docker",
"ssh",
"sshd",
"systemd-journald",
"cron",
)
SCRIPT_DIRS: tuple[str, ...] = ("/opt/blueforce/bin", "/usr/local/sbin")
MODULE_PATTERN = re.compile(r"^[0-9]{2}-[a-z0-9][a-z0-9._-]*$")
PACKAGE_PIN_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9+_.:-]*=.+$")
HOST_PATTERN = re.compile(r"^bf-[0-9]{8}$")
STATE_JSON_PATH = "/var/lib/blueforce/state.json"
STATE_ALLOWED_KEYS: tuple[str, ...] = (
"phase",
"enrollment_status",
"fleet_status",
"device_id",
"provisioning_id",
)
INSTALL_STATE_PATH = "/var/lib/blueforce/install-state"
INSTALL_LOG_PATH = "/var/log/blueforce-install.log"
#: --check eklenebilecek yerel komutlar (kurulum betigi ve kapi araclari).
CHECK_CAPABLE_LOCAL_OPS: tuple[str, ...] = (
"install-preflight",
"install-local",
"install-resume",
"install-module",
"tools-gate-local",
"tools-gate-enrollment",
"tools-gate-ready",
)
DEFAULT_CONSOLE_LOG = "/var/log/blueforce-console.log"
INSTALLER_REL = "scripts/install/blueforce-install.sh"
PLAYBOOK_DIR_REL = "ansible/playbooks"
TK_PACKAGE_HINT = "sudo apt install -y python3-tk"
# Grafik arayüz açılamadığında kullanılacak yol: aynı bfos uygulamasının
# terminal konsolu. GUI ve TUI aynı Python backend'ini kullanır.
TERMINAL_MENU_HINT = "sudo bf --tui"
# ---------------------------------------------------------------------------
# tkinter -- opsiyonel ice aktarma
# ---------------------------------------------------------------------------
class _MissingTk:
"""tkinter yoksa: kullanildigi anda Turkce hata veren yer tutucu.
Amac iki yonlu: (1) modul tkinter'siz de ice aktarilabilir kalsin,
(2) GUI kodundaki `tk.X`/`ttk.X` tip denetimi gecerli bir nesne gordugu
icin None uzerinden gelen yanlis uyarilar uretilmesin.
"""
def __getattr__(self, name: str):
raise RuntimeError(
"tkinter kurulu değil (" + TK_IMPORT_ERROR + "). "
f"Kurmak için: {TK_PACKAGE_HINT}"
)
try: # pragma: no cover - ortama bagli
import tkinter as tk
from tkinter import messagebox, ttk
HAVE_TKINTER = True
TK_IMPORT_ERROR = ""
except Exception as _tk_exc: # pragma: no cover - ortama bagli
HAVE_TKINTER = False
TK_IMPORT_ERROR = str(_tk_exc)
tk = ttk = messagebox = _MissingTk() # type: ignore[assignment]
# ---------------------------------------------------------------------------
# Tasarim belirtecleri -- tek kaynak: bfos/ui_theme.py, o da
# web/static/tokens.css'i birebir yansitir. Bu dosya renk/secim/bosluk
# degeri URETMEZ; hepsi temadan gelir (docs/DESIGN.md "Rules").
# ---------------------------------------------------------------------------
try:
from bfos import ui_theme as theme
except ImportError: # pragma: no cover - dosya paket disinda yuklendiginde
_THEME_DIR = str(Path(__file__).resolve().parent)
if _THEME_DIR not in sys.path:
sys.path.insert(0, _THEME_DIR)
import ui_theme as theme # type: ignore[no-redef]
try:
from bfos import ui_widgets as widgets
except ImportError: # pragma: no cover - dosya paket disinda yuklendiginde
_WIDGETS_DIR = str(Path(__file__).resolve().parent)
if _WIDGETS_DIR not in sys.path:
sys.path.insert(0, _WIDGETS_DIR)
import ui_widgets as widgets # type: ignore[no-redef]
# ---------------------------------------------------------------------------
# Durum sozlugu (docs/DESIGN.md "Status vocabulary")
# ---------------------------------------------------------------------------
#: Bir cihaz durumunun TUM sozcukleri. Renk tek basina anlam tasimaz: her
#: rozet sozcugunu da yazar, liste renksiz de okunur.
DEVICE_STATES: tuple[str, ...] = ("KURULDU", "KAYITLI", "HAZIR", "HATA", "YOK")
#: Sozcuk -> token anahtari (tema ile ayni esleme).
STATE_KEYS_BY_WORD: Mapping[str, str] = {
word: key for key, word in theme.STATE_WORDS.items()
}
#: state.json'da bir denetimin dustugunu gosteren isaretler.
FAILURE_MARKERS: tuple[str, ...] = ("FAIL", "HATA", "ERROR", "BOZUK")
#: Yetki seviyeleri: tam yetki ve salt-okunur GORUNUR birer durumdur.
PRIVILEGE_FULL = "full"
PRIVILEGE_READ_ONLY = "read_only"
PRIVILEGE_WORDS: Mapping[str, str] = {
PRIVILEGE_FULL: "TAM YETKİ",
PRIVILEGE_READ_ONLY: "SALT-OKUNUR",
}
def read_state_values(path: str = STATE_JSON_PATH) -> dict[str, str]:
"""state.json'u BEYAZ LISTE ile okur (jeton/anahtar alanlari okunmaz).
Okunamayan/yok dosya bos sozluk doner: cihaz kurulmamis olabilir ve bu bir
hata degil, ``YOK`` durumudur.
"""
try:
with open(path, "r", encoding="utf-8") as handle:
data = json.load(handle)
except (OSError, ValueError):
return {}
if not isinstance(data, dict):
return {}
values: dict[str, str] = {}
for key in STATE_ALLOWED_KEYS:
value = data.get(key)
if value is None or isinstance(value, bool):
continue
values[key] = str(value)
return values
def state_word_from_values(values: Mapping[str, str]) -> str:
"""state.json alanlarini sozlukteki TEK sozcuge indirger.
Sira onemli: once HATA (bir denetim dustu), sonra HAZIR (her kanal
dogrulandi), sonra KAYITLI (kaydedildi, kanallar eksik), sonra KURULDU
(kuruldu, henuz kaydedilmedi). Alan yoksa YOK: hic gorulmemis cihaz.
Acik olumsuzlama ("NOT_ENROLLED") kayit SAYILMAZ: o cihaz kaydedilmemistir.
"""
if not values:
return "YOK"
phase = str(values.get("phase", "")).upper()
fleet = str(values.get("fleet_status", "")).upper()
enrollment = str(values.get("enrollment_status", "")).upper()
if any(marker in field for marker in FAILURE_MARKERS
for field in (phase, fleet, enrollment)):
return "HATA"
if any(field.endswith("READY") and not field.startswith(("NOT_", "UN"))
for field in (phase, fleet, enrollment)):
return "HAZIR"
if any(field.endswith("ENROLLED") and not field.startswith(("NOT_", "UN"))
for field in (phase, fleet, enrollment)):
return "KAYITLI"
if phase or fleet or enrollment:
return "KURULDU"
return "YOK"
def device_state_word(path: str = STATE_JSON_PATH) -> str:
"""Bu cihazin yerel kurulum/kayit durumu (sozluk sozcugu)."""
return state_word_from_values(read_state_values(path))
def state_badge_style(word: str) -> str:
"""Sozcukten rozet stilini cozer; bilinmeyen sozcuk notr rozete duser."""
key = STATE_KEYS_BY_WORD.get(word)
if key is None:
return theme.STATE_BADGE_STYLE["absent"]
return theme.STATE_BADGE_STYLE[key]
def privilege_level(root: bool | None = None) -> str:
"""'full' ya da 'read_only'; yetki bir durumdur, satir degil."""
granted = is_root() if root is None else bool(root)
return PRIVILEGE_FULL if granted else PRIVILEGE_READ_ONLY
def privilege_word(level: str) -> str:
"""Yetki seviyesinin ekranda gorunen sozcugu."""
return PRIVILEGE_WORDS.get(level, PRIVILEGE_WORDS[PRIVILEGE_READ_ONLY])
def write_blocked_reason(operation) -> str | None:
"""Yazma islemi kapaliysa NEDENINI doner; islem calisabilirse None.
Yazma islemleri yetkisiz oturumda GIZLENMEZ: listede kalir, dugu kapali
cizilir ve nedenin metni dugmenin kendi uzerinde yazar
(docs/DESIGN.md "Privilege").
"""
if not operation.needs_root:
return None
return None if write_operations_enabled() else NON_ROOT_DISABLED_HINT
def operation_is_write(operation) -> bool:
"""True when the operation changes the device (catalogue rule: root)."""
return bool(operation.needs_root)
# ---------------------------------------------------------------------------
# Veri tipleri (backend sozlesmesi)
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class OptionSpec:
"""Bir işlemin ek parametresi (Ansible -e değişkenleri)."""
name: str
label: str
kind: str = "text" # text | int | bool | choice
default: str = ""
choices: tuple[str, ...] = ()
hint: str = ""
required: bool = False
validator: Callable[[str], "str | None"] | None = None
@dataclass(frozen=True)
class Operation:
"""Operasyon katalogu kaydi (sozlesme: bkz. dosya basligi)."""
id: str
label: str
category: str
destructive: bool = False
scope: str = "local" # local | simple | wave | release
description: str = ""
playbook: str = ""
options: tuple[OptionSpec, ...] = ()
# None means "not stated": the value is then derived from `destructive` by
# the property below, so a write operation cannot be marked read-only by
# omission. An operation that genuinely differs states the value.
requires_root: bool | None = None
needs_dealer: bool = False
@property
def needs_root(self) -> bool:
"""Whether this operation must run as root to be allowed."""
if self.requires_root is None:
return bool(self.destructive)
return bool(self.requires_root)
@property
def is_fleet(self) -> bool:
return self.scope in ("simple", "wave", "release")
@dataclass(frozen=True)
class Target:
"""Hedef secimi: bayı no, dalga/grup adi ya da tum filo."""
kind: str # dealer | group | all
value: str
@property
def limit(self) -> str:
if self.kind == "dealer":
return dealer_to_host(self.value)
return self.value
@property
def label(self) -> str:
if self.kind == "dealer":
return f"cihaz {dealer_to_device_id(self.value)} ({dealer_to_host(self.value)})"
if self.kind == "all":
return "tüm filo (envanterdeki her cihaz)"
if is_wave(self.value):
return f"dalga '{self.value}'"
return f"grup '{self.value}'"
@dataclass(frozen=True)
class GateStepInfo:
"""Calistirmadan once gereken ACIK operator onayi (sozlesme eki).
kind : "token" (yazilarak) | "approval" (evet/hayir)
token: kind == "token" ise yazilmasi gereken sozcuk
unlocks: onay verilince backend'e gecirilecek bayrak adlari
(or. "wave_gate_confirmed", "production_override")
"""
kind: str
key: str
prompt: str
message: str = ""
token: str = ""
unlocks: tuple[str, ...] = ()
@dataclass(frozen=True)
class CommandPlan:
"""Calistirilacak komutlar + ekranda gosterilecek satir listesi.
`report_lines` dolu ve `commands` bos oldugunda plan HICBIR SEY calistirmaz:
rapor satirlari (bfos.operations icindeki salt-okunur ekranlar) dogrudan
gosterilir.
"""
commands: tuple[tuple[str, ...], ...] = ()
destructive: bool = False
requires_root: bool | None = None
operation_id: str = ""
target_label: str = ""
notes: tuple[str, ...] = ()
report_lines: tuple[str, ...] = ()
dry_run: bool = False
check_mode: bool = False
@property
def needs_root(self) -> bool:
"""Whether running this plan requires root (derived when not stated)."""
if self.requires_root is None:
return bool(self.destructive)
return bool(self.requires_root)
@property
def lines(self) -> tuple[str, ...]:
"""Kopyala/yapıştır edilebilir komut satırları."""
return tuple(shlex.join(argv) for argv in self.commands if argv)
@property
def is_empty(self) -> bool:
return not any(self.commands) and not self.report_lines
@property
def is_report_only(self) -> bool:
return not any(self.commands) and bool(self.report_lines)
@dataclass
class RunResult:
"""Komut calistirma sonucu: cikti + cikis kodu (sozlesme maddesi 4)."""
exit_code: int
ok: bool
output: str = ""
class BackendUnavailable(RuntimeError):
"""Gercek backend modulu var ama sozlesmeye uymuyor."""
class CommandError(RuntimeError):
"""Komut uretilemedi / secenek gecersiz (TURKCE mesaj tasir)."""
def _as_int(value: Any, default: int = 1) -> int:
"""Farkli backend'lerin cikis kodu tiplerini (int/str/None) guvenle cevirir."""
try:
return int(value)
except (TypeError, ValueError):
return default
# ---------------------------------------------------------------------------
# Saf yardimcilar: birim testleri bunlari pencere ACMADAN kullanir
# ---------------------------------------------------------------------------
def validate_dealer(value: str) -> bool:
"""Bayi no kapisi: ^[0-9]{8}$ (docs/02, K-12)."""
return bool(DEALER_PATTERN.match(value or ""))
def dealer_error(value: str) -> str | None:
"""Gecerli ise None, degilse Turkce hata metni."""
if validate_dealer(value):
return None
if not value:
return "Bayi numarası boş olamaz. Tam 8 hane, yalnız rakam girin (örn. 12010193)."
if not value.isdigit():
return "Bayi numarası yalnız rakam içermelidir (harf, boşluk ve işaret kullanılamaz)."
return (
f"Geçersiz bayi numarası '{value}': tam 8 hane olmalıdır "
f"(verilen uzunluk: {len(value)}). Örnek: 12010193"
)
def dealer_to_host(dealer: str) -> str:
return f"bf-{dealer}"
def dealer_to_device_id(dealer: str) -> str:
return f"BF-{dealer}"
def host_to_dealer(host: str) -> str | None:
return host[3:] if HOST_PATTERN.match(host or "") else None
def is_wave(name: str) -> bool:
return name in WAVES
def validate_wave(value: str) -> str | None:
"""Dalga adini dogrular; gecerliyse None doner."""
if value in WAVES:
return None
return (
f"Geçersiz dalga '{value}'. Onaylı dalgalar (sırayla): {', '.join(WAVES)}."
)
def validate_disk_pct(value: str) -> str | None:
if value == "":
return None
if not value.isdigit() or not (1 <= int(value) <= 100):
return "Eşik 1 ile 100 arasında bir tam sayı olmalıdır."
return None
def validate_grace(value: str) -> str | None:
if value == "":
return None
if not value.isdigit():
return "Bekleme süresi saniye cinsinden bir tam sayı olmalıdır (örn. 300)."
return None
def validate_log_lines(value: str) -> str | None:
if value == "":
return None
if not value.isdigit() or int(value) < 1:
return "Satır sayısı pozitif bir tam sayı olmalıdır (örn. 500)."
return None
def validate_service_name(value: str) -> str | None:
if value in SERVICE_ALLOWLIST:
return None
return (
f"'{value}' izinli servis listesinde değil. İzinli olanlar: "
f"{', '.join(SERVICE_ALLOWLIST)}."
)
def validate_script_path(value: str) -> str | None:
if not value:
return "Script yolu boş olamaz."
if not value.startswith("/"):
return "Script yolu mutlak olmalıdır (örn. /opt/blueforce/bin/kontrol.sh)."
if not any(value == d or value.startswith(d + "/") for d in SCRIPT_DIRS):
return (
"Script yolu izinli dizinlerin dışında. İzinli dizinler: "
f"{', '.join(SCRIPT_DIRS)}."
)
return None
def validate_package_pins(value: str) -> str | None:
if not value.strip():
return "En az bir paket sürümü gerekir (biçim: paket=sürüm)."
for pin in value.split(","):
pin = pin.strip()
if not pin:
continue
if not PACKAGE_PIN_PATTERN.match(pin):
return (
f"Geçersiz pin '{pin}': 'paket=sürüm' biçiminde olmalıdır "
"(örn. docker-ce=5:27.0.3-1~ubuntu.24.04~noble)."
)
return None
def validate_module_name(value: str) -> str | None:
if not value:
return "Modül numarası boş olamaz (örn. 01-precheck)."
if not MODULE_PATTERN.match(value):
return (
f"Geçersiz modül '{value}': '01-precheck' biçiminde olmalıdır "
"(iki hane, tire, küçük harfli ad)."
)
return None
def validate_expected_version(value: str) -> str | None:
if value == "":
return None
if any(ch.isspace() for ch in value):
return "Sürüm boşluk içermemelidir (örn. 1.0.0)."
return None
# ---------------------------------------------------------------------------
# Headless / tkinter hatalari -- net TURKCE yonlendirme
# ---------------------------------------------------------------------------
def headless_error(env: Mapping[str, str] | None = None) -> str | None:
"""Ekran yoksa Turkce aciklama doner; ekran varsa None.
Sadece DISPLAY / WAYLAND_DISPLAY ortam degiskenlerine bakar, boylece
pencere acmadan test edilebilir.
"""
data = os.environ if env is None else env
if data.get("DISPLAY") or data.get("WAYLAND_DISPLAY"):
return None
return (
"Grafik ekran bulunamadı (DISPLAY ve WAYLAND_DISPLAY tanımlı değil).\n"
"\n"
"SSH veya başsız (headless) bir oturumda masaüstü penceresi açılamaz.\n"
"Bu durumda terminal menüsünü kullanın:\n"
f"\n {TERMINAL_MENU_HINT}\n"
"\n"
"Yerel ekranda çalıştırmak için oturum açık bir makinede deneyin:\n"
" python3 -m bfos.ui_gui\n"
)
def tkinter_missing_message(error: str = "") -> str:
"""tkinter kurulu degilse Turkce hata + kurulum ipucu."""
detail = f" (teknik ayrıntı: {error})" if error else ""
return (
"tkinter bulunamadı; masaüstü arayüzü başlatılamıyor"
f"{detail}.\n"
"\n"
"Bu, minimal kurulumlarda sık görülür: tkinter ayrı bir sistem paketidir.\n"
"Ubuntu/Debian üzerinde kurmak için:\n"
f"\n {TK_PACKAGE_HINT}\n"
" sudo apt update && sudo apt install -y python3-tk\n"
"\n"
f"tkinter kurmak istemiyorsanız terminal menüsünü kullanın: {TERMINAL_MENU_HINT}\n"
)
def window_start_error(exc: BaseException) -> str:
"""Tk() acilamazsa (DISPLAY var ama X/Wayland erisilemiyor) Turkce mesaj."""
return (
"Pencere açılamadı; grafik oturumuna erişilemiyor.\n"
f"Teknik ayrıntı: {exc}\n"
"\n"
"Muhtemel nedenler:\n"
" * SSH oturumundasınız ve X yönlendirmesi (ssh -X) yok,\n"
" * ekran kilitli veya grafik oturum kapalı,\n"
" * başka bir kullanıcının X sunucusuna bağlanmaya çalışıyorsunuz.\n"
f"\nTerminal menüsünü kullanın: {TERMINAL_MENU_HINT}\n"
)
# ---------------------------------------------------------------------------
# Onizleme metni (onay pencereleri ve kayit icin tek kaynak)
# ---------------------------------------------------------------------------
def preview_text(
plan: CommandPlan,
*,
operation: Operation | None = None,
target: Target | None = None,
) -> str:
"""Calistirilacak komutu gosteren Turkce onay metni."""
parts: list[str] = []
if plan.is_report_only:
parts.append("Bu ekran komut çalıştırmaz; salt-okunur rapor gösterir:")
parts.append("")
for line in plan.report_lines:
parts.append(f" {line}")
parts.append("")
else:
parts.append("Çalıştırılacak komut(lar):")
parts.append("")
if plan.is_empty:
parts.append(" (komut üretilemedi)")
for line in plan.lines:
parts.append(f" {line}")
parts.append("")
if operation is not None:
parts.append(f"İşlem : {operation.label} [{operation.id}]")
parts.append(f"Kapsam : {SCOPE_LABELS.get(operation.scope, operation.scope)}")
if operation.description:
parts.append(f"Açıklama: {operation.description}")
if target is not None:
parts.append(f"Hedef : {target.label}")
elif plan.target_label:
parts.append(f"Hedef : {plan.target_label}")
if plan.needs_root:
parts.append("Yetki : root gerekir — konsolu sudo ile açın.")
if plan.check_mode:
parts.append("Kip : --check (deneme) — hiçbir değişiklik uygulanmaz.")
if plan.dry_run:
parts.append("Kip : DENEME (--dry-run) — komut çalıştırılmaz, yalnız gösterilir.")
if plan.destructive:
parts.append("")
parts.append("UYARI: Bu YIKICI bir işlemdir; seçili hedefte değişiklik yapar.")
for note in plan.notes:
parts.append(f"Not : {note}")
return "\n".join(parts)
def destructive_operations(
operations: Sequence[Operation] | None = None,
) -> tuple[str, ...]:
"""Yikici isaretli operasyonlarin id'leri."""
ops = OPERATIONS if operations is None else operations
return tuple(op.id for op in ops if op.destructive)
def operation_is_destructive(operation_id: str) -> bool:
op = operation_by_id(operation_id)
return bool(op and op.destructive)
# ---------------------------------------------------------------------------
# Yerleşik katalog -- ortak Python katalog kuralları.
# ---------------------------------------------------------------------------
_FLEET_OPTIONS: Mapping[str, tuple[OptionSpec, ...]] = {
"disk": (
OptionSpec(
"disk_warn_pct",
"Uyarı eşiği (%)",
kind="int",
default="",
hint="Boş = playbook varsayılanı (80)",
validator=validate_disk_pct,
),
),
"fieldos-version": (
OptionSpec(
"fieldos_expected_version",
"Beklenen Field OS sürümü",
default="",
hint="Boş = drift denetimi atlanır",
validator=validate_expected_version,
),
),
"service-restart": (
OptionSpec(
"service_name",
"Servis adı",
kind="choice",
choices=SERVICE_ALLOWLIST,
required=True,
hint="Yalnız izinli servis adları",
validator=validate_service_name,
),
),
"run-script": (
OptionSpec(
"script_path",
"Script yolu",
required=True,
hint=f"İzinli dizinler: {', '.join(SCRIPT_DIRS)}",
validator=validate_script_path,
),
OptionSpec("script_args", "Script argümanları", default="", hint="İsteğe bağlı"),
),
"collect-logs": (
OptionSpec(
"log_lines",
"Cihaz başına journal satırı",
kind="int",
default="500",
validator=validate_log_lines,
),
),
"reboot": (
OptionSpec(
"reboot_grace_seconds",
"Yeniden başlatma bekleme süresi (saniye)",
kind="int",
default="300",
validator=validate_grace,
),
),
"deploy-update": (
OptionSpec(
"update_packages",
"Paket pinleri (paket=sürüm, virgülle)",
required=True,
hint="Örn. docker-ce=5:27.0.3-1~ubuntu.24.04~noble",
validator=validate_package_pins,
),
),
}
OPERATIONS: tuple[Operation, ...] = (
# --- Kurulum (bu cihaz; filo cihazina DOKUNMAZ) ------------------------
Operation(
"install-preflight",
"Kurulum ön kontrolü (--check)",
"Kurulum",
destructive=False,
scope="local",
description="Hiçbir şey yazmadan kurulum ön koşullarını raporlar.",
needs_dealer=True,
),
Operation(
"install-local",
"Bu cihazı kur",
"Kurulum",
destructive=True,
scope="local",
description="Cihazı kurar, hostname'i bf-<bayi-no> yapar.",
requires_root=True,
needs_dealer=True,
options=(
OptionSpec("offline", "Yerel çevrimdışı depodan kur (--offline)", kind="bool"),
OptionSpec("resume", "Zaten OK olan modülleri atla (--resume)", kind="bool"),
),
),
Operation(
"install-resume",
"Başarısız kuruluma devam et (--resume)",
"Kurulum",
destructive=True,
scope="local",
description="Yarıda kalan kurulumu idempotent biçimde sürdürür.",
requires_root=True,
needs_dealer=True,
),
Operation(
"install-module",
"Tek modül çalıştır (--only)",
"Kurulum",
destructive=True,
scope="local",
description="Kurulumun tek bir modülünü çalıştırır.",
requires_root=True,
needs_dealer=True,
options=(
OptionSpec(
"only",
"Modül",
required=True,
hint="Örn. 01-precheck",
validator=validate_module_name,
),
),
),
Operation(
"install-state",
"Kurulum modül durumu",
"Kurulum",
destructive=False,
scope="local",
description="Kurulumun hangi modülde kaldığını gösterir (salt-okunur).",
),
Operation(
"install-log",
"Kurulum günlüğü (son 40 satır)",
"Kurulum",
destructive=False,
scope="local",
description="Kurulum günlüğünün sonunu gösterir (salt-okunur).",
),
# --- Cihaz islemleri: salt-okunur raporlar -----------------------------
Operation("ping", "Ping / erişilebilirlik", "Cihaz işlemleri", False, "simple",
"SSH erişilebilirliği (salt-okunur).", "bf-ping.yml"),
Operation("uptime", "Çalışma süresi ve yük", "Cihaz işlemleri", False, "simple",
"Uptime ve yük ortalaması (salt-okunur).", "bf-uptime.yml"),
Operation("disk", "Disk kullanımı", "Cihaz işlemleri", False, "simple",
"Disk kullanımı; eşik üstünde uyarır.", "bf-disk-check.yml",
options=_FLEET_OPTIONS["disk"]),
Operation("docker", "Docker durumu", "Cihaz işlemleri", False, "simple",
"Docker servisi ve konteynerler (salt-okunur).", "bf-docker-status.yml"),
Operation("packages", "Kurulu paket sürümleri", "Cihaz işlemleri", False, "simple",
"Pinlenmiş paketlerin kurulu sürümleri.", "bf-package-check.yml"),
Operation("security-updates", "Bekleyen güvenlik güncellemeleri", "Cihaz işlemleri",
False, "simple", "Yalnız rapor; hiçbir güncelleme kurmaz.",
"bf-security-updates-check.yml"),
Operation("provisioning-status", "Kurulum fazı raporu", "Cihaz işlemleri", False,
"wave", "Cihaz başına state.json fazı.", "bf-provisioning-status.yml"),
Operation("enrollment-status", "Kayıt (enrollment) ve merkez kanıtı",
"Cihaz işlemleri", False, "wave",
"ENROLLED/READY iddialarını merkezi kanıta karşı denetler.",
"bf-enrollment-status.yml"),
Operation("fieldos-version", "Field OS sürüm raporu", "Cihaz işlemleri", False, "wave",
"Sürüm bilgisi; istenirse drift denetimi.", "bf-fieldos-version.yml",
options=_FLEET_OPTIONS["fieldos-version"]),
Operation("remote-channels", "Uzak erişim kanalları", "Cihaz işlemleri", False, "wave",
"SSH/xRDP/RustDesk/WireGuard hazırlığı.", "bf-remote-channels-check.yml"),
Operation("offline-ready", "Çevrimdışı hazırlık raporu", "Cihaz işlemleri", False, "wave",
"Offline provisioning yerel ön koşulları.", "bf-offline-ready-check.yml"),
# --- Cihaz islemleri: bakim (yikici olanlar isaretli) -------------------
Operation("rdp-restart", "xRDP servisini yeniden başlat", "Cihaz işlemleri", True,
"wave", "xRDP servisini yeniden başlatır (RDP açık kalır).",
"bf-rdp-restart.yml"),
Operation("rustdesk-restart", "RustDesk servisini yeniden başlat", "Cihaz işlemleri",
True, "wave", "RustDesk servisini yeniden başlatır.", "bf-rustdesk-restart.yml"),
Operation("wireguard-restart", "WireGuard (wg0) yeniden başlat", "Cihaz işlemleri", True,
"wave", "wg-quick@wg0 servisini yeniden başlatır.", "bf-wireguard-restart.yml"),
Operation("service-restart", "İzinli servisi yeniden başlat", "Cihaz işlemleri", True,
"wave", "Yalnız izinli servis adlarını yeniden başlatır.",
"bf-service-restart.yml", options=_FLEET_OPTIONS["service-restart"]),
Operation("gui-off", "Yerel grafik oturumu kapat (headless)", "Cihaz işlemleri", True,
"wave", "Dalgayı multi-user.target'e alır; xRDP'yi durdurmaz.", "bf-gui-off.yml"),
Operation("gui-on", "Yerel grafik oturumu aç (bakım)", "Cihaz işlemleri", True, "wave",
"Grafik oturumu yeniden açar (bakım için).", "bf-gui-on.yml"),
Operation("collect-logs", "Günlük topla / destek paketi", "Cihaz işlemleri", False,
"wave", "Zaman damgalı journal + destek paketi üretir.", "bf-collect-logs.yml",
options=_FLEET_OPTIONS["collect-logs"]),
Operation("run-script", "Onaylı script çalıştır", "Cihaz işlemleri", True, "wave",
"Yalnız onaylı dizinlerdeki scriptleri çalıştırır.", "bf-run-script.yml",
options=_FLEET_OPTIONS["run-script"]),
Operation("reboot", "Kontrollü yeniden başlatma", "Cihaz işlemleri", True, "release",
"Seri yeniden başlatma + sağlık kapısı.", "bf-reboot.yml",
options=_FLEET_OPTIONS["reboot"]),
Operation("deploy-update", "Onaylı paket güncellemesi", "Cihaz işlemleri", True,
"release", "package=sürüm pinleri + dalga ve sağlık kapıları.",
"bf-deploy-update.yml", options=_FLEET_OPTIONS["deploy-update"]),
# --- Merkez hazirligi (salt-okunur denetim; merkez SUNUCUSU kurulmaz) ---
Operation("central-targets", "Envanter hedefleri (dalga/grup/cihaz sayıları)",
"Merkez hazırlığı", False, "local",
"Envanteri ve dalga gruplarını listeler; hiçbir cihaza dokunmaz."),
Operation("central-placeholders", "Doldurulmamış PLACEHOLDER denetimi",
"Merkez hazırlığı", False, "local",
"group_vars/all.yml içinde kalan PLACEHOLDER değerlerini gösterir."),
Operation("central-endpoints", "Yapılandırılmış merkez uç noktaları",
"Merkez hazırlığı", False, "local",
"WireGuard hub, izleme ve depo uç noktalarını (secret hariç) gösterir."),
Operation("central-checklist", "Merkez hazırlık kontrol listesi",
"Merkez hazırlığı", False, "local",
"Kayıttan önce merkezde var olması gerekenleri hatırlatır."),
# --- Araclar -----------------------------------------------------------
Operation("tools-catalogue", "İşlem kataloğunu göster (--list)", "Araçlar", False,
"local", "Konsolun işlem kataloğunu listeler."),
Operation("tools-git-pull", "Bu kopyayı güncelle (git pull --ff-only)", "Araçlar", True,
"local", "Depo kopyasını ileri sarar; yerel değişikliği silmez."),
Operation("tools-gate-local", "Yerel çevrimdışı kapısı (bf-check-local)", "Araçlar",
False, "local", "Cihazın çevrimdışı kurulum ön koşullarını denetler."),
Operation("tools-gate-enrollment", "Kayıt kapısı (bf-check-enrollment)", "Araçlar",
False, "local", "Kayıt (enrollment) ön koşullarını denetler."),
Operation("tools-gate-ready", "Teslim kapısı (bf-check-ready)", "Araçlar", False,
"local", "Cihazın READY olabilmesi için gerekenleri denetler."),
Operation("tools-diagnostics", "Derin teşhis (bf-diagnostics)", "Araçlar", False,
"local", "Kapsamlı teşhis raporu üretir (salt-okunur)."),
Operation("tools-release", "Field OS sürüm kimliği (bf-release)", "Araçlar", False,
"local", "Cihazdaki Field OS sürümünü gösterir."),
Operation("tools-support-bundle", "Destek paketi oluştur", "Araçlar", False, "local",
"Parola/anahtar içermeyen destek paketi üretir."),
Operation("tools-console-log", "Konsol denetim günlüğü (son 40 satır)", "Araçlar",
False, "local", "Operatör konsolunun denetim günlüğünü gösterir."),
# --- Cihaz bilgisi (salt-okunur) ---------------------------------------
Operation("info-summary", "Durum özeti (sağlık + kurulum durumu)", "Cihaz bilgisi",
False, "local",
"bf-status, whitelist'li state.json alanları ve envanter hedefleri."),
Operation("info-identity", "Kimlik (hostname, cihaz no)", "Cihaz bilgisi", False,
"local", "Makine adı ve cihaz kimliği."),
Operation("info-network", "Ağ (adres, rota, DNS, WireGuard)", "Cihaz bilgisi", False,
"local", "Ağ arayüzleri, varsayılan rota, DNS ve WireGuard durumu."),
Operation("info-resources", "Disk, bellek, yük", "Cihaz bilgisi", False, "local",
"Dosya sistemleri, bellek ve yük ortalaması."),
Operation("info-packages", "Kurulu pinlenmiş paketler", "Cihaz bilgisi", False, "local",
"docker-ce, wireguard, rustdesk, xrdp, openssh-server sürümleri."),
Operation("info-services", "Servis durumu", "Cihaz bilgisi", False, "local",
"ssh, xrdp, rustdesk, docker, wg-quick@wg0, cron durumu."),
Operation("info-journal", "Son journal hataları", "Cihaz bilgisi", False, "local",
"journalctl -p err son 20 satır."),
Operation("info-state", "Kurulum/kayıt durumu (state.json beyaz liste)", "Cihaz bilgisi",
False, "local",
"Yalnız kimlik/durum alanları okunur; token ve anahtar asla gösterilmez."),
)
_OPERATION_INDEX: Mapping[str, Operation] = {op.id: op for op in OPERATIONS}
def operation_by_id(operation_id: str) -> Operation | None:
return _OPERATION_INDEX.get(operation_id)
def operations_by_category(category: str) -> tuple[Operation, ...]:
return tuple(op for op in OPERATIONS if op.category == category)
# ---------------------------------------------------------------------------
# Yardımcı: yol çözümleme.
# ---------------------------------------------------------------------------
def repo_root_from_file() -> Path:
"""bfos/ui_gui.py -> depo koku (bir ust dizin)."""
return Path(__file__).resolve().parents[1]
def resolve_inventory(repo_root: Path, explicit: str | None = None) -> Path | None:
if explicit:
candidate = Path(explicit).expanduser()
return candidate if candidate.is_file() else None
base = repo_root / "ansible" / "inventory"
for name in ("hosts.yml", "hosts.yaml", "hosts", "hosts.ini"):
candidate = base / name
if candidate.is_file():
return candidate
return None
def resolve_tool(name: str, repo_root: Path | None = None) -> str | None:
"""Araç yolu sırası: /usr/local/bin, sonra depo."""
system = Path("/usr/local/bin") / name
if system.is_file() and os.access(system, os.X_OK):
return str(system)
root = repo_root or repo_root_from_file()
for sub in ("scripts/diagnostics", "scripts/maintenance"):
candidate = root / sub / name
if candidate.is_file() and os.access(candidate, os.X_OK):
return str(candidate)
return None
def playbook_path(repo_root: Path, playbook: str) -> Path:
return repo_root / PLAYBOOK_DIR_REL / playbook
def playbook_declares_var(path: Path, var: str) -> bool:
"""Playbook dosyasi VAR'i (yorum satirlari haric) kullaniyor mu?
admin/bf-menu -> bf_playbook_var_present ile ayni fikir: playbook kendi
kapisinin kaynagi kalir, konsol onu varsaymaz.
"""
try:
text = path.read_text(encoding="utf-8", errors="replace")
except OSError:
return False
pattern = re.compile(rf"\b{re.escape(var)}\b")
for line in text.splitlines():
stripped = line.lstrip()
if stripped.startswith("#"):
continue
if pattern.search(line):
return True
return False
def playbook_scope(path: Path) -> str:
"""Playbook'un kendi beyan ettigi kapsam: release | wave | simple."""
if playbook_declares_var(path, "release_wave"):
return "release"
if playbook_declares_var(path, "operation_wave"):
return "wave"
return "simple"
def parse_inventory(path: Path) -> dict[str, list[str]]:
"""Salt-okunur envanter okuma: {grup: [host, ...]}.
admin/lib/targets.sh -> bf_inventory_pairs awk okuyucusunun Python
karsiligi: `hosts:` bloklarini iceren en yakin grup anahtari kullanilir.
Amac yalnizca hedef listesi/dalga listesi doldurmak; hicbir sey yazilmaz.
"""
groups: dict[str, list[str]] = {}
stack: list[tuple[int, str]] = []
hosts_indent: int | None = None
hosts_step: int | None = None
hosts_group = ""
try:
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
except OSError:
return groups
for raw in lines:
if raw.lstrip().startswith("#") or not raw.strip() or raw.lstrip().startswith("-"):
continue
if ":" not in raw:
continue
indent = len(raw) - len(raw.lstrip(" "))
key, _, value = raw.partition(":")
key = key.strip().strip("\"'")
value = value.strip().strip("\"'")
if not key:
continue
while stack and stack[-1][0] >= indent:
stack.pop()
if key == "hosts" and value == "":
hosts_indent = indent
hosts_step = None
hosts_group = stack[-1][1] if stack else ""
stack.append((indent, key))
continue
if hosts_indent is not None and indent <= hosts_indent:
hosts_indent = None
hosts_step = None
if hosts_indent is not None and indent > hosts_indent:
if hosts_step is None:
hosts_step = indent
if indent == hosts_step and hosts_group:
groups.setdefault(hosts_group, []).append(key)
stack.append((indent, key))
continue
if value == "":
stack.append((indent, key))
return groups
def waves_in_inventory(groups: Mapping[str, Sequence[str]]) -> tuple[str, ...]:
"""Envanterde gercekten bulunan onayli dalgalar (yayilim sirasiyla)."""
return tuple(wave for wave in WAVES if groups.get(wave))
def device_wave(host: str, groups: Mapping[str, Sequence[str]]) -> str | None:
for wave in WAVES:
if host in (groups.get(wave) or ()):
return wave
return None
# ---------------------------------------------------------------------------
# Komut calistirma (sozlesme maddesi 4)
# ---------------------------------------------------------------------------
def run_command(
argv: Sequence[str],
on_line: Callable[[str], None] | None = None,
cancel: threading.Event | None = None,
cwd: str | None = None,
) -> RunResult:
"""Tek komutu calistirir; ciktisi satir satir `on_line`'a akar."""
argv = [str(a) for a in argv]
if not argv:
return RunResult(2, False, "Boş komut çalıştırılamaz.")
emit = on_line or (lambda _line: None)
try:
proc = subprocess.Popen( # noqa: S603 - argv listesi, shell yok
argv,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
cwd=cwd,
)
except FileNotFoundError:
message = (
f"Komut bulunamadı: '{argv[0]}' bu cihazda kurulu değil. "
"Araçları bağlamak için: sudo admin/bf-bootstrap.sh"
)
emit(message)
return RunResult(127, False, message + "\n")
except OSError as exc:
message = f"Komut başlatılamadı: {' '.join(argv)} ({exc})"
emit(message)
return RunResult(126, False, message + "\n")
collected: list[str] = []
assert proc.stdout is not None
for line in proc.stdout:
collected.append(line)
emit(line.rstrip("\n"))
if cancel is not None and cancel.is_set():
proc.terminate()
note = "Kullanıcı iptal etti; komut sonlandırıldı."
collected.append(note + "\n")
emit(note)
break
exit_code = proc.wait()
return RunResult(exit_code, exit_code == 0, "".join(collected))
def run_plan(
plan: CommandPlan,
on_line: Callable[[str], None] | None = None,
cancel: threading.Event | None = None,
cwd: str | None = None,
) -> RunResult:
"""Plan icindeki komutlari sirayla calistirir; ilk hatada DURUR.
(Wave akisindaki "bir dalga basarisizsa sonraki dalgaya gecme" kuralinin
basit karsiligi: hata sonrasi komut calistirilmaz.)
"""
emit = on_line or (lambda _line: None)
if plan.is_empty:
return RunResult(2, False, "Çalıştırılacak komut yok.\n")
if plan.is_report_only:
for line in plan.report_lines:
emit(line)
return RunResult(0, True, "\n".join(plan.report_lines) + "\n")
if plan.dry_run:
for line in plan.lines:
emit(f"[deneme] {line}")
return RunResult(0, True, "[deneme] komut çalıştırılmadı.\n")
if plan.needs_root and not write_operations_enabled():
# Last line of defence: every execution path funnels through this
# function, so the root rule holds even if a future caller builds the
# plan itself instead of going through the GUI's run button.
emit("[ENGELLENDİ] " + NON_ROOT_DISABLED_HINT)
return RunResult(2, False, NON_ROOT_DISABLED_HINT + "\n")
chunks: list[str] = []
for index, argv in enumerate(plan.commands, start=1):
if not argv:
continue
if len(plan.commands) > 1:
emit(f"--- [{index}/{len(plan.commands)}] {shlex.join(argv)}")
result = run_command(argv, on_line=emit, cancel=cancel, cwd=cwd)
chunks.append(result.output)
if not result.ok:
emit(f"[HATA] Komut {result.exit_code} koduyla bitti: {shlex.join(argv)}")
return RunResult(result.exit_code, False, "".join(chunks))
return RunResult(0, True, "".join(chunks))
# ---------------------------------------------------------------------------
# Backend: yerlesik katalog (admin/bf-menu kurallari) + gercek modul adapteri
# ---------------------------------------------------------------------------
@dataclass
class GuiContext:
"""Arayuzun calisma baglami (yol ve kip bilgisi)."""
repo_root: Path = field(default_factory=repo_root_from_file)
inventory: Path | None = None
console_log: str = DEFAULT_CONSOLE_LOG
dry_run: bool = False
active_dealer: str | None = None
active_user: str | None = None
@classmethod
def detect(cls, repo: str | None = None, inventory: str | None = None,
dry_run: bool = False) -> "GuiContext":
root = Path(repo).expanduser().resolve() if repo else repo_root_from_file()
# GUI identity is loaded once into the session context. Individual
# surfaces may display it, but they do not invent a second identity.
try:
from bfos.preferences import load_preferences
preferences = load_preferences()
except (ImportError, OSError):
preferences = {}
return cls(
repo_root=root,
inventory=resolve_inventory(root, inventory),
console_log=os.environ.get("BF_CONSOLE_LOG", DEFAULT_CONSOLE_LOG),
dry_run=dry_run,
active_dealer=preferences.get("dealer_id"),
active_user=preferences.get("active_user"),
)
class CatalogueBackend:
"""Yerlesik katalog: admin/bf-menu'nun tablosu + komut uretimi.
Gercek moduller (bfos.operations / bfos.runner) yoksa arayuz bunu kullanir;
boylece GUI, backend yazilmadan once de dogru komutu gosterip calistirabilir.
"""
name = "yerleşik katalog (admin/bf-menu kuralları)"
def __init__(self, ctx: GuiContext):
self.ctx = ctx
self._groups: dict[str, list[str]] | None = None
# -- envanter (salt-okunur) -------------------------------------------
@property
def groups(self) -> dict[str, list[str]]:
if self._groups is None:
self._groups = (
parse_inventory(self.ctx.inventory) if self.ctx.inventory else {}
)
return self._groups
def list_operations(self) -> tuple[Operation, ...]:
return OPERATIONS
def list_waves(self) -> tuple[str, ...]:
present = waves_in_inventory(self.groups)
return present or WAVES
def inventory_note(self) -> str:
if self.ctx.inventory is None:
note = (
"Ansible envanteri bulunamadı (ansible/inventory/hosts.yml). "
"Filo işlemleri için envanteri oluşturun: "
"cp ansible/inventory/hosts.example.yml ansible/inventory/hosts.yml"
)
else:
note = f"Envanter: {self.ctx.inventory}"
if shutil.which("ansible-playbook") is None:
note += (
" | UYARI: ansible-playbook bulunamadı; filo işlemleri çalışmaz "
"(sudo apt install ansible)."
)
return note
# -- hedef -------------------------------------------------------------
def validate_target(self, operation: Operation, target: Target) -> str | None:
if target.kind == "dealer":
problem = dealer_error(target.value)
if problem:
return problem
if operation.scope in ("wave", "release"):
if target.kind == "dealer":
if self.ctx.inventory is None:
return (
"Bu işlem dalga kapılıdır ve cihazın dalgasını çözmek için "
"envanter gerekir; envanter bulunamadı."
)
wave = device_wave(dealer_to_host(target.value), self.groups)
if not wave:
return (
f"{dealer_to_host(target.value)} onaylı bir dalga grubunda değil "
f"({', '.join(WAVES)}). Önce cihazı dalgasına ekleyin."
)
elif target.kind == "all":
if operation.scope == "release":
return (
"Sürüm kapılı işlemler tüm filo ile çalıştırılamaz: "
"onaylı bir dalga grubu seçin (dalga dalga yayılım)."
)
elif target.kind == "group" and not is_wave(target.value):
return (
f"'{target.value}' onaylı bir dalga değil. Dalga kapılı işlemler "
f"şu dalgalardan birini ister: {', '.join(WAVES)}."
)
return None
def target_wave(self, operation: Operation, target: Target) -> str | None:
if target.kind == "group" and is_wave(target.value):
return target.value
if target.kind == "dealer":
return device_wave(dealer_to_host(target.value), self.groups)
return None
# -- kapilar (acik onay adimlari) ---------------------------------------
def gate_steps(
self, operation: Operation, target: Target | None
) -> tuple[GateStepInfo, ...]:
"""Calistirmadan once sorulacak ACIK onaylar (bf-menu ile ayni kurallar).
Hicbir kapi sessizce gecilmez: dalga adi yazdirilir, production icin
ayrica `production` yazilir, surum kapisi icin GATE yazilir, yikici
islemler icin ikinci onay istenir.
"""
steps: list[GateStepInfo] = []
wave = self.target_wave(operation, target) if target else None
book = (
playbook_path(self.ctx.repo_root, operation.playbook)
if operation.playbook
else None
)
gate_supported = bool(book and playbook_declares_var(book, "wave_gate_confirmed"))
if operation.scope in ("wave", "release") and wave:
if wave == PRODUCTION_WAVE:
steps.append(
GateStepInfo(
kind="token",
key="production_token",
prompt="PRODUCTION onayı — canlı filo",
message=(
"Seçilen dalga canlı filodur (production).\n"
"Onaylamak için kutuya 'production' yazın; başka bir şey iptal eder."
+ ("\nEk olarak dalga kapısı bayrağı onaylanacak." if gate_supported else "")
),
token=PRODUCTION_WAVE,
unlocks=("wave_gate_confirmed",) if gate_supported else (),
)
)
steps.append(
GateStepInfo(
kind="approval",
key="production_approval",
prompt="Production ek onayı",
message=(
"production_override / allow_production_reboot bayrakları "
"yalnız bu onayla eklenir (playbook ilgili değişkeni "
"kullanıyorsa).\nDevam edilsin mi?"
),
unlocks=("production_override", "allow_production_reboot"),
)
)
elif operation.scope == "release" and gate_supported:
steps.append(
GateStepInfo(
kind="token",
key="wave_gate_token",
prompt=f"Sürüm kapısı onayı (dalga: {wave})",
message=(
f"'{wave}' dalgası için sürüm kapısını onaylıyorsunuz.\n"
f"Önceki dalga sağlık kapısını geçtiyse '{GATE_TOKEN_RELEASE}' yazın."
),
token=GATE_TOKEN_RELEASE,
unlocks=("wave_gate_confirmed",),
)
)
if target is not None and target.kind == "all" and operation.scope in ("simple", "wave"):
steps.append(
GateStepInfo(
kind="token",
key="fleet_token",
prompt="FİLO GENELİ onayı",
message=(
"Komut envanterdeki HER cihaza uygulanacak.\n"
"Onaylamak için kutuya 'all' yazın."
),
token="all",
unlocks=("fleet_wide",),
)
)
if operation.destructive:
steps.append(
GateStepInfo(
kind="approval",
key="destructive_approval",
prompt="İKİNCİ ONAY — yıkıcı işlem",
message=(
f"'{operation.label}' seçili hedefte DEĞİŞİKLİK YAPAR.\n"
"Bu onay yalnız bu işlem için geçerlidir. Devam edilsin mi?"
),
)
)
return tuple(steps)
# -- komut uretimi -----------------------------------------------------
def build_command(
self,
operation: Operation,
target: Target | None = None,
options: Mapping[str, str] | None = None,
grants: Mapping[str, bool] | None = None,
check_mode: bool = False,
) -> CommandPlan:
opts = dict(options or {})
self._check_options(operation, opts)
if operation.scope in ("wave", "release") and target is not None:
problems = self.validate_target(operation, target)
if problems:
raise CommandError(problems)
if operation.is_fleet:
return self._build_fleet_plan(operation, target, opts, dict(grants or {}), check_mode)
return self._build_local_plan(operation, target, opts, check_mode)
def _check_options(self, operation: Operation, opts: Mapping[str, str]) -> None:
for spec in operation.options:
value = (opts.get(spec.name) or "").strip()
if spec.required and not value:
raise CommandError(f"'{spec.label}' zorunludur ({operation.label}).")
if value and spec.validator:
problem = spec.validator(value)
if problem:
raise CommandError(problem)
def _build_fleet_plan(
self,
operation: Operation,
target: Target | None,
opts: Mapping[str, str],
grants: Mapping[str, bool],
check_mode: bool = False,
) -> CommandPlan:
if target is None or not target.value:
raise CommandError(
"Hedef seçilmedi; komut üretilmedi. Bayi numarası veya dalga grubu seçin."
)
problem = self.validate_target(operation, target)
if problem:
raise CommandError(problem)
if self.ctx.inventory is None:
raise CommandError(
"Ansible envanteri bulunamadı; filo işlemleri komutu üretilemiyor. "
"Beklenen yol: ansible/inventory/hosts.yml (veya --inventory PATH)."
)
book = playbook_path(self.ctx.repo_root, operation.playbook)
if not book.is_file():
raise CommandError(
f"Playbook bulunamadı: {PLAYBOOK_DIR_REL}/{operation.playbook}\n"
"Doğru depo kopyasında çalıştırın (--repo PATH) veya checkout'u güncelleyin."
)
declared = playbook_scope(book)
if declared != operation.scope:
raise CommandError(
f"{operation.playbook} çalıştırılmıyor: konsol kapsamı '{operation.scope}' "
f"beklerken playbook '{declared}' beyan ediyor.\n"
"Playbook kapıların kaynağıdır; tablo ile playbook'u eşitleyin "
"(admin/bf-menu -> bf_playbook_check ile aynı kural)."
)
wave = self.target_wave(operation, target)
if operation.scope in ("wave", "release") and not wave:
raise CommandError(
"Bu işlem dalga kapsamlıdır ama hedefte onaylı bir dalga yok. "
"Bir dalga grubu seçin (" + ", ".join(WAVES) + ")."
)
argv: list[str] = [
"ansible-playbook",
"-i",
str(self.ctx.inventory),
str(book),
"--limit",
target.limit,
]
notes: list[str] = []
if operation.scope == "wave":
argv += ["-e", f"operation_wave={wave}", "-e", "operation_confirmed=true"]
if (
wave == PRODUCTION_WAVE
and grants.get("production_override")
and playbook_declares_var(book, "production_override")
):
argv += ["-e", "production_override=true"]
notes.append(
"Dalga kapısı: yalnız önceki dalga sağlık kapısını ve merkez onayını "
"geçtiyse çalıştırın (docs/10)."
)
elif operation.scope == "release":
argv += ["-e", f"release_wave={wave}"]
if grants.get("wave_gate_confirmed") and playbook_declares_var(
book, "wave_gate_confirmed"
):
argv += ["-e", "wave_gate_confirmed=true"]
if playbook_declares_var(book, "reboot_confirmed"):
argv += ["-e", "reboot_confirmed=true"]
if wave == PRODUCTION_WAVE:
if grants.get("production_override") and playbook_declares_var(
book, "production_override"
):
argv += ["-e", "production_override=true"]
if grants.get("allow_production_reboot") and playbook_declares_var(
book, "allow_production_reboot"
):
argv += ["-e", "allow_production_reboot=true"]
notes.append(
"Sürüm kapısı: dalga onayı ve sağlık kapısı olmadan çalıştırmayın (docs/10)."
)
if target.kind == "all" and operation.scope == "wave":
notes.append(
"Tüm filo: dalga kapılı playbook'lar dalga dalga ilerler; her dalgayı "
"ayrı ayrı onaylayın."
)
argv += self._extra_vars(operation.id, opts)
if check_mode:
argv.append("--check")
notes.append("--check: playbook hiçbir değişiklik uygulamaz (deneme koşusu).")
return CommandPlan(
commands=(tuple(argv),),
destructive=operation.destructive,
requires_root=operation.needs_root,
operation_id=operation.id,
target_label=target.label + (f" (dalga {wave})" if wave else ""),
notes=tuple(notes),
dry_run=self.ctx.dry_run,
check_mode=check_mode,
)
def _extra_vars(self, operation_id: str, opts: Mapping[str, str]) -> list[str]:
value = lambda name: (opts.get(name) or "").strip() # noqa: E731
extra: list[str] = []
if operation_id == "disk" and value("disk_warn_pct"):
extra += ["-e", f"disk_warn_pct={value('disk_warn_pct')}"]
elif operation_id == "fieldos-version" and value("fieldos_expected_version"):
extra += ["-e", f"fieldos_expected_version={value('fieldos_expected_version')}"]
elif operation_id == "collect-logs" and value("log_lines"):
extra += ["-e", f"log_lines={value('log_lines')}"]
elif operation_id == "reboot" and value("reboot_grace_seconds"):
extra += ["-e", f"reboot_grace_seconds={value('reboot_grace_seconds')}"]
elif operation_id == "service-restart" and value("service_name"):
extra += ["-e", f"service_name={value('service_name')}"]
elif operation_id == "run-script" and value("script_path"):
extra += ["-e", f"script_path={value('script_path')}"]
if value("script_args"):
extra += ["-e", f"script_args={value('script_args')}"]
elif operation_id == "deploy-update" and value("update_packages"):
pins = [
f'"{pin.strip()}"'
for pin in value("update_packages").split(",")
if pin.strip()
]
extra += ["-e", f"update_packages=[{','.join(pins)}]"]
return extra
def _installer_arg(self) -> str:
installer = self.ctx.repo_root / INSTALLER_REL
if not installer.is_file():
raise CommandError(
f"Kurulum betiği bulunamadı: {INSTALLER_REL}\n"
"Depo kopyasını doğru verin (--repo PATH) veya checkout'u güncelleyin."
)
return str(installer)
def _tool_argv(self, tool: str, *args: str, optional: bool = False) -> tuple[str, ...] | None:
path = resolve_tool(tool, self.ctx.repo_root)
if path is None:
if optional:
return None
raise CommandError(
f"Araç bulunamadı: {tool}\n"
"Bu araç depoda veya /usr/local/bin altında yok. Bağlamak için: "
"sudo admin/bf-bootstrap.sh"
)
return (path, *args)
def _build_local_plan(
self,
operation: Operation,
target: Target | None,
opts: Mapping[str, str],
check_mode: bool = False,
) -> CommandPlan:
dealer = ""
if operation.needs_dealer:
if target is None or target.kind != "dealer" or not target.value:
raise CommandError(
f"'{operation.label}' için bayi numarası gerekir "
"(hedef alanına 8 haneli bayi no yazın)."
)
problem = dealer_error(target.value)
if problem:
raise CommandError(problem)
dealer = target.value
commands: list[tuple[str, ...]] = []
notes: list[str] = []
op_id = operation.id
if op_id == "install-preflight":
commands.append(
("bash", self._installer_arg(), "--dealer-id", dealer, "--check")
)
notes.append("Ön kontrol hiçbir şey yazmaz (log/state/dizin oluşturmaz).")
elif op_id in ("install-local", "install-resume"):
argv = ["bash", self._installer_arg(), "--dealer-id", dealer]
if op_id == "install-resume" or opts.get("resume") == "1":
argv.append("--resume")
if opts.get("offline") == "1":
argv.append("--offline")
commands.append(tuple(argv))
notes.append(f"Hedef cihaz kimliği: {dealer_to_device_id(dealer)} ({dealer_to_host(dealer)}).")
elif op_id == "install-module":
commands.append(
(
"bash",
self._installer_arg(),
"--dealer-id",
dealer,
"--only",
(opts.get("only") or "").strip(),
)
)
elif op_id == "install-state":
commands.append(("cat", INSTALL_STATE_PATH))
elif op_id == "install-log":
commands.append(("tail", "-n", "40", INSTALL_LOG_PATH))
elif op_id == "central-targets":
commands.append(("bash", self._menu_arg(), "--list-targets"))
elif op_id == "central-placeholders":
commands.append(
(
"grep",
"-n",
"PLACEHOLDER",
str(self.ctx.repo_root / "ansible" / "inventory" / "group_vars" / "all.yml"),
)
)
notes.append("Eşleşme yoksa bu iyi haberdir (PLACEHOLDER kalmamış demektir).")
elif op_id == "central-endpoints":
commands.append(
(
"grep",
"-nE",
r"^[[:space:]]*(wg_subnet|wg_interface|wg_hub_endpoint|monitoring_endpoint|"
r"monitoring_job|update_repo_url|meg_image_tag|support_bundle_dir)[[:space:]]*:",
str(self.ctx.repo_root / "ansible" / "inventory" / "group_vars" / "all.yml"),
)
)
notes.append("Yalnız yapılandırma anahtarları okunur; secret hiçbir zaman gösterilmez.")
elif op_id == "central-checklist":
for line in CENTRAL_CHECKLIST:
notes.append(line)
elif op_id == "tools-catalogue":
commands.append(("bash", self._menu_arg(), "--list"))
elif op_id == "tools-git-pull":
commands.append(("git", "-C", str(self.ctx.repo_root), "pull", "--ff-only"))
notes.append("Çekme sonrası konsol değişmiş olabilir; arayüzü yeniden başlatın.")
elif op_id == "tools-gate-local":
commands.append(self._tool_argv("bf-check-local", "--check") or ())
elif op_id == "tools-gate-enrollment":
commands.append(self._tool_argv("bf-check-enrollment", "--check") or ())
elif op_id == "tools-gate-ready":
commands.append(self._tool_argv("bf-check-ready", "--check") or ())
elif op_id == "tools-diagnostics":
commands.append(self._tool_argv("bf-diagnostics") or ())
elif op_id == "tools-release":
commands.append(self._tool_argv("bf-release") or ())
elif op_id == "tools-support-bundle":
commands.append(
self._tool_argv("bf-support-bundle", "--out-dir", "/var/tmp") or ()
)
notes.append("Destek paketi parola, private key veya token içermez.")
elif op_id == "tools-console-log":
commands.append(("tail", "-n", "40", self.ctx.console_log))
elif op_id == "info-summary":
commands.extend(self._status_summary_commands())
elif op_id == "info-identity":
commands.append(("hostname",))
commands.append(
(
"grep",
"-nE",
r"^[[:space:]]*\"(device_id|phase)\"",
STATE_JSON_PATH,
)
)
elif op_id == "info-network":
commands.append(("ip", "-brief", "address"))
commands.append(("ip", "route", "show", "default"))
commands.append(("sh", "-c", "awk '/^nameserver/{print}' /etc/resolv.conf"))
commands.append(("wg", "show", "wg0", "latest-handshakes"))
notes.append("WireGuard private/preshared anahtarları istenmez ve gösterilmez.")
elif op_id == "info-resources":
commands.append(("df", "-h"))
commands.append(("free", "-h"))
commands.append(("uptime",))
elif op_id == "info-packages":
commands.append(
(
"dpkg-query",
"-W",
"-f=${Package} ${Version} ${Status}\n",
"docker-ce",
"wireguard",
"rustdesk",
"xrdp",
"openssh-server",
)
)
elif op_id == "info-services":
commands.append(
(
"systemctl",
"is-active",
"ssh",
"sshd",
"xrdp",
"rustdesk",
"docker",
"wg-quick@wg0",
"cron",
)
)
elif op_id == "info-journal":
commands.append(("journalctl", "-p", "err", "-n", "20", "--no-pager"))
elif op_id == "info-state":
commands.append(
(
"grep",
"-nE",
r"^[[:space:]]*\"(phase|enrollment_status|fleet_status|device_id|provisioning_id)\"",
STATE_JSON_PATH,
)
)
notes.append(
"Yalnız kimlik/durum alanları okunur; token ve anahtarlar gösterilmez."
)
else:
raise CommandError(f"Bilinmeyen operasyon: {operation.id}")
commands = [argv for argv in commands if argv]
if not commands:
raise CommandError(
f"'{operation.label}' için çalıştırılabilir komut üretilemedi. "
"Bu cihazda gerekli araç kurulu olmayabilir (sudo admin/bf-bootstrap.sh)."
)
# --check yalnizca destekleyen komutlara eklenir (kurulum betigi ve
# bf-check-* kapilari); tail/cat/grep/dpkg gibi komutlara eklenmez.
if check_mode and op_id in CHECK_CAPABLE_LOCAL_OPS:
commands = [tuple(argv) + ("--check",) for argv in commands]
notes.append("--check: yalnız denetim yapılır, değişiklik uygulanmaz.")
elif check_mode:
notes.append("--check bu işlem için geçerli değil; normal koşu hazırlanır.")
label = target.label if target is not None and target.value else "bu cihaz (yerel)"
return CommandPlan(
commands=tuple(commands),
destructive=operation.destructive,
requires_root=operation.needs_root,
operation_id=operation.id,
target_label=label,
notes=tuple(notes),
dry_run=self.ctx.dry_run,
check_mode=check_mode and op_id in CHECK_CAPABLE_LOCAL_OPS,
)
def _status_summary_commands(self) -> list[tuple[str, ...]]:
"""Durum ozeti: bakim sirasinda kaybolan cikti yerine guvenli, opsiyonel set."""
commands: list[tuple[str, ...]] = []
for tool, args in (("bf-status", ()), ("bf-enrollment-status", ("--check",))):
argv = self._tool_argv(tool, *args, optional=True)
if argv:
commands.append(argv)
commands.append(
(
"grep",
"-nE",
r"^[[:space:]]*\"(phase|enrollment_status|fleet_status|device_id|provisioning_id)\"",
STATE_JSON_PATH,
)
)
commands.append(("bash", self._menu_arg(), "--list-targets"))
return commands
# -- calistirma --------------------------------------------------------
def run(
self,
plan: CommandPlan,
on_line: Callable[[str], None] | None = None,
cancel: threading.Event | None = None,
) -> RunResult:
return run_plan(plan, on_line=on_line, cancel=cancel, cwd=str(self.ctx.repo_root))
CENTRAL_CHECKLIST: tuple[str, ...] = (
"1. WireGuard hub: uç nokta, anahtarlar ve her bf-<bayi-no> için eş ataması (docs/08).",
"2. Kayıt (enrollment) token servisi: tek kullanımlık, TTL sınırlı token (docs/29).",
"3. İzleme: Prometheus/VictoriaMetrics + Grafana, device_id etiketi = BF-<no> (docs/14).",
"4. Uzak erişim: RustDesk hbbs/hbbr veya MeshCentral, WireGuard üzerinden erişilebilir (docs/07).",
"5. Paket aynası + saha kurulumları için çevrimdışı anlık görüntü (docs/10, docs/27).",
"6. Gerçek ana makine adlarını içeren Ansible envanteri ve yönetici SSH anahtarı (docs/02, 06, 09).",
"Bu arayüz 6'yı denetler; 1-5'i hazırlamaya yardımcı olur, merkez sunucusu KURMAZ.",
)
class ModuleBackend:
"""Gercek bfos.operations / bfos.runner modullerini kullanan backend.
Paylasilan is mantigi TEK kaynaktan gelir (kopyalanmaz):
* katalog : bfos.operations (catalogue_rows / ALL_OPERATIONS)
* envanter/hedef: Inventory.load_repository + resolve_target
* komut uretimi : build_ansible_command / build_install_command /
build_tool_command / build_extra_vars
* kapilar : confirmation_plan + GatesGranted (dalga, production,
surum kapisi, yikici ikinci onay)
* denetim kaydi : bfos.runner.AuditLog (varsa) -- TUI ile ayni kayit.
Yerel (kurulum / arac / merkez / bilgi) islemler icin modulde karsiligi olan
kurucular kullanilir; karsiligi olmayan salt-okunur ekran ve komutlar
CatalogueBackend'e devredilir (bkz. _local).
"""
name = "bfos.operations + bfos.runner (gerçek backend)"
#: modulde komut kurucusu olan yerel arac islemleri: id -> (arac, sabit arg.)
_TOOL_OPS: Mapping[str, tuple[str, tuple[str, ...]]] = {
"tools-gate-local": ("bf-check-local", ()),
"tools-gate-enrollment": ("bf-check-enrollment", ()),
"tools-gate-ready": ("bf-check-ready", ()),
"tools-diagnostics": ("bf-diagnostics", ()),
"tools-release": ("bf-release", ()),
"tools-support-bundle": ("bf-support-bundle", ("--out-dir", "/var/tmp")),
}
def __init__(self, ops_module, runner_module, ctx: GuiContext):
self._ops = ops_module
self._runner = runner_module
self.ctx = ctx
self._local = CatalogueBackend(ctx)
self._repo = ops_module.Repository(str(ctx.repo_root))
self._inventory = ops_module.Inventory.load_repository(
self._repo, str(ctx.inventory) if ctx.inventory else ""
)
self._raw_ops: dict[str, Any] = {}
self._catalogue = self._load_catalogue()
self._by_id = {op.id: op for op in self._catalogue}
self._fleet_ids = frozenset(op.id for op in self._catalogue if op.is_fleet)
# -- katalog -----------------------------------------------------------
def _load_catalogue(self) -> tuple[Operation, ...]:
"""Modulun kataloğu (filo işlemleri) + yerel işlemler (kurulum/araç/...)."""
raw: Any = None
for name in ("catalogue_rows", "list_operations", "operations",
"ALL_OPERATIONS", "OPERATIONS", "OPS", "get_operations"):
candidate = getattr(self._ops, name, None)
if candidate is None:
continue
try:
value = candidate() if callable(candidate) else candidate
except Exception:
continue
if value:
raw = value
break
if not raw:
raise BackendUnavailable(
"bfos.operations içinde operasyon listesi bulunamadı.\n"
"Beklenen: catalogue_rows() / ALL_OPERATIONS "
"(bkz. bfos/ui_gui.py başlığındaki SÖZLEŞME)."
)
fleet: list[Operation] = []
for item in raw:
operation = self._normalize_operation(item)
fleet.append(operation)
self._raw_ops[operation.id] = item
if not fleet:
raise BackendUnavailable("bfos.operations boş bir operasyon listesi döndürdü.")
local = tuple(op for op in OPERATIONS if not op.is_fleet)
return tuple(fleet) + local
@staticmethod
def _normalize_operation(item) -> Operation:
"""Modul operasyonunu arayuzun Operation tipine cevirir (SÖZLEŞME)."""
if isinstance(item, Operation):
return item
get = (lambda key, default=None: item.get(key, default)) if isinstance(item, dict) \
else (lambda key, default=None: getattr(item, key, default))
op_id = get("id") or get("ident") or get("name") or get("operation_id")
if not op_id:
raise BackendUnavailable("Bir operasyonun id alanı yok (SÖZLEŞME: id zorunlu).")
label = get("label") or get("title") or get("description") or op_id
category = get("category") or "Cihaz işlemleri"
if category not in CATEGORIES:
category = "Cihaz işlemleri"
scope = str(get("scope") or get("declared_scope") or "simple")
prompt = get("prompt") or "none"
destructive = bool(get("destructive") or get("is_destructive") or False)
# requires_root is left None unless the source actually states it, so the
# Operation property derives it from `destructive`. Reading the field
# with a False default marked every real operation read-only and the
# root gate never fired; the catalogue's `destructive` flag is the
# honest signal. An operation that states the value keeps that answer.
stated_root = None
if isinstance(item, dict):
if "requires_root" in item:
stated_root = bool(item["requires_root"])
elif getattr(item, "requires_root", None) is not None:
stated_root = bool(item.requires_root)
return Operation(
id=str(op_id),
label=str(label),
category=str(category),
destructive=destructive,
scope=scope,
description=str(get("description") or ""),
playbook=str(get("playbook") or ""),
options=_prompt_options(str(prompt)),
requires_root=stated_root,
needs_dealer=bool(get("needs_dealer") or False),
)
# -- envanter ----------------------------------------------------------
def list_operations(self) -> tuple[Operation, ...]:
return self._catalogue
def list_waves(self) -> tuple[str, ...]:
try:
present = self._inventory.waves_present()
except Exception:
present = ()
return tuple(present) or WAVES
def inventory_note(self) -> str:
note = self._local.inventory_note()
if self._inventory.path:
note = f"Envanter: {self._inventory.path} ({self._inventory.count_all()} cihaz)"
if not self._have_ansible():
note += (
" | UYARI: ansible-playbook bulunamadı; filo işlemleri çalışmaz "
"(sudo apt install ansible)."
)
return note
def _have_ansible(self) -> bool:
checker = getattr(getattr(self._runner, "Runner", None), "have_ansible", None)
try:
if callable(checker):
return bool(checker())
except Exception:
pass
return shutil.which("ansible-playbook") is not None
def target_wave(self, operation: Operation, target: Target | None) -> str | None:
if target is None or not target.value:
return None
try:
resolved = self._resolve_target(target, operation)
except CommandError:
return None
return resolved.wave or None
def validate_target(self, operation: Operation, target: Target) -> str | None:
try:
self._resolve_target(target, operation)
except CommandError as exc:
return str(exc)
return None
def _resolve_target(self, target: Target, operation: Operation):
"""Modulun resolve_target'i ile hedefi cozer (Refused -> CommandError)."""
raw_op = self._raw_ops.get(operation.id)
mode = "any"
if raw_op is not None:
mode = getattr(raw_op, "target_mode", "any")
try:
return self._ops.resolve_target(target.value, mode, self._inventory)
except Exception as exc:
raise self._refusal(exc) from exc
@staticmethod
def _refusal(exc: Exception) -> CommandError:
"""Modulun Refused hatasini (mesaj + ayrinti) Turkce CommandError'a cevirir."""
message = getattr(exc, "message", None) or str(exc)
detail = getattr(exc, "detail", "") or ""
text = f"{message}\n{detail}".strip() if detail else str(message)
if isinstance(exc, CommandError):
return exc
return CommandError(text)
# -- kapilar -----------------------------------------------------------
def gate_steps(
self, operation: Operation, target: Target | None
) -> tuple[GateStepInfo, ...]:
"""Modulun confirmation_plan'i ile ayni kapi kurallari.
confirmation_plan yoksa/çağrılamazsa yerel ayna (CatalogueBackend)
kullanilir: kapi sormamak gibi sessiz bir davranis ASLA secilmez.
"""
if operation.id not in self._fleet_ids or target is None or not target.value:
return self._local.gate_steps(operation, target)
planner = getattr(self._ops, "confirmation_plan", None)
raw_op = self._raw_ops.get(operation.id)
if not callable(planner) or raw_op is None:
return self._local.gate_steps(operation, target)
try:
resolved = self._resolve_target(target, operation)
gate_supported = False
plays = getattr(self._repo, "playbook_uses", None)
if callable(plays):
gate_supported = bool(plays(raw_op.playbook, "wave_gate_confirmed"))
steps = planner(
raw_op, raw_op.declared_scope, resolved,
wave_gate_supported=gate_supported,
)
except Exception:
return self._local.gate_steps(operation, target)
infos: list[GateStepInfo] = []
for step in steps:
infos.append(
GateStepInfo(
kind=str(getattr(step, "kind", "approval")),
key=str(getattr(step, "key", "")),
prompt=str(getattr(step, "prompt", "")),
message=str(getattr(step, "message", "")),
token=str(getattr(step, "token", "")),
unlocks=tuple(getattr(step, "unlocks", ()) or ()),
)
)
return tuple(infos)
def _gates(self, grants: Mapping[str, bool]):
"""Onaylanan bayraklari modulun GatesGranted tipine cevirir."""
gates_cls = getattr(self._ops, "GatesGranted", None)
if gates_cls is None:
return None
fields = getattr(gates_cls, "__dataclass_fields__", {})
kwargs = {name: bool(value) for name, value in grants.items() if name in fields}
try:
return gates_cls(**kwargs)
except Exception:
return None
# -- komut uretimi -----------------------------------------------------
def build_command(
self,
operation: Operation,
target: Target | None = None,
options: Mapping[str, str] | None = None,
grants: Mapping[str, bool] | None = None,
check_mode: bool = False,
) -> CommandPlan:
opts = dict(options or {})
granted = dict(grants or {})
if operation.id in self._fleet_ids:
return self._build_fleet_plan(operation, target, opts, granted, check_mode)
for builder in (self._build_install_plan, self._build_tool_plan,
self._build_report_plan):
plan = builder(operation, target, opts, check_mode)
if plan is not None:
return plan
# Modulde karsiligi olmayan yerel komut/salt-okunur ekranlar.
return self._local.build_command(operation, target, opts, granted, check_mode)
def _build_fleet_plan(
self,
operation: Operation,
target: Target | None,
opts: Mapping[str, str],
grants: Mapping[str, bool],
check_mode: bool,
) -> CommandPlan:
raw_op = self._raw_ops.get(operation.id)
if raw_op is None:
raise CommandError(
f"'{operation.id}' operasyonunun tanımı backend'de bulunamadı."
)
if target is None or not target.value:
raise CommandError(
"Hedef seçilmedi; komut üretilmedi. Bayi numarası veya dalga grubu seçin."
)
if not self._inventory.path:
raise CommandError(
"Ansible envanteri bulunamadı; filo işlemleri komutu üretilemiyor. "
"Beklenen yol: ansible/inventory/hosts.yml (veya --inventory PATH)."
)
verify = getattr(self._repo, "verify_scope", None)
if callable(verify):
# Playbook kendi kapisinin kaynagidir; tablo ile uyusmazsa REDDET.
try:
verify(raw_op)
except Exception as exc:
raise self._refusal(exc) from exc
try:
resolved = self._resolve_target(target, operation)
extra = self._ops.build_extra_vars(raw_op.prompt, opts)
argv = self._ops.build_ansible_command(
self._repo, self._inventory.path, raw_op, raw_op.declared_scope,
resolved, extra_vars=extra, gates=self._gates(grants),
check_mode=check_mode,
)
except Exception as exc:
raise self._refusal(exc) from exc
notes: list[str] = []
if raw_op.declared_scope in ("wave", "release"):
notes.append(
"Dalga/sürüm kapısı: yalnız önceki dalga sağlık kapısını ve merkez "
"onayını geçtiyse çalıştırın (docs/10)."
)
if not grants.get("wave_gate_confirmed") and _declares(
self._repo, raw_op.playbook, "wave_gate_confirmed"
):
notes.append(
"Not: wave_gate_confirmed bayrağı yalnız sürüm kapısı onayı verilirse "
"komuta eklenir."
)
if check_mode:
notes.append("--check: playbook hiçbir değişiklik uygulamaz (deneme koşusu).")
return CommandPlan(
commands=(tuple(argv),),
destructive=bool(raw_op.destructive),
requires_root=bool(raw_op.destructive),
operation_id=operation.id,
target_label=str(getattr(resolved, "label", target.label)),
notes=tuple(notes),
dry_run=self.ctx.dry_run,
check_mode=check_mode,
)
def _build_install_plan(
self, operation: Operation, target: Target | None,
opts: Mapping[str, str], check_mode: bool,
) -> CommandPlan | None:
if not operation.id.startswith("install-"):
return None
builder = getattr(self._ops, "build_install_command", None)
if not callable(builder):
return None
if operation.id in ("install-state", "install-log"):
return None
dealer = ""
if operation.needs_dealer:
if target is None or target.kind != "dealer" or not target.value:
raise CommandError(
f"'{operation.label}' için bayi numarası gerekir "
"(hedef alanına 8 haneli bayi no yazın)."
)
problem = dealer_error(target.value)
if problem:
raise CommandError(problem)
dealer = target.value
try:
installer = self._repo.installer_path()
except Exception as exc:
raise self._refusal(exc) from exc
check = operation.id == "install-preflight" or check_mode
try:
argv = builder(
installer,
dealer,
offline=opts.get("offline") == "1",
resume=operation.id == "install-resume" or opts.get("resume") == "1",
only=(opts.get("only") or "").strip(),
check_mode=check,
)
except Exception as exc:
raise self._refusal(exc) from exc
notes = [f"Hedef cihaz kimliği: {dealer_to_device_id(dealer)} ({dealer_to_host(dealer)})."]
if operation.id == "install-preflight":
notes.append("Ön kontrol hiçbir şey yazmaz (log/state/dizin oluşturmaz).")
if check:
notes.append("--check: hiçbir değişiklik uygulanmaz.")
return CommandPlan(
commands=(tuple(argv),),
destructive=operation.destructive,
requires_root=operation.needs_root,
operation_id=operation.id,
target_label=f"cihaz {dealer_to_device_id(dealer)} ({dealer_to_host(dealer)})",
notes=tuple(notes),
dry_run=self.ctx.dry_run,
check_mode=check,
)
def _build_tool_plan(
self, operation: Operation, target: Target | None,
opts: Mapping[str, str], check_mode: bool,
) -> CommandPlan | None:
entry = self._TOOL_OPS.get(operation.id)
if entry is None:
return None
builder = getattr(self._ops, "build_tool_command", None)
tool, args = entry
try:
path = self._repo.tool_path(tool)
except Exception as exc:
raise self._refusal(exc) from exc
if not callable(builder):
return None
check = tool.startswith("bf-check-") or check_mode
argv = builder(path, list(args), check_mode=check)
notes: list[str] = []
if operation.id == "tools-support-bundle":
notes.append("Destek paketi parola, private key veya token içermez.")
return CommandPlan(
commands=(tuple(argv),),
destructive=operation.destructive,
requires_root=operation.needs_root,
operation_id=operation.id,
target_label="bu cihaz (yerel)",
notes=tuple(notes),
dry_run=self.ctx.dry_run,
check_mode=check,
)
def _build_report_plan(
self, operation: Operation, target: Target | None,
opts: Mapping[str, str], check_mode: bool,
) -> CommandPlan | None:
"""Modulun salt-okunur ekran ureticileri (komut CALISTIRILMAZ)."""
lines = self._module_report_lines(operation)
if lines is None:
return None
return CommandPlan(
report_lines=tuple(lines),
destructive=False,
operation_id=operation.id,
target_label="bu cihaz (yerel)",
notes=("Salt-okunur ekran: hiçbir komut çalıştırılmadı.",),
dry_run=self.ctx.dry_run,
)
def _module_report_lines(self, operation: Operation) -> tuple[str, ...] | None:
ops = self._ops
try:
if operation.id == "central-targets" and hasattr(ops, "targets_lines"):
return tuple(ops.targets_lines(self._inventory))
if operation.id == "central-placeholders" and hasattr(
ops, "central_readiness_lines"
):
return tuple(ops.central_readiness_lines(self._repo, self._inventory))
if operation.id == "central-endpoints" and hasattr(ops, "central_endpoint_lines"):
return tuple(ops.central_endpoint_lines(self._repo))
if operation.id == "central-checklist" and hasattr(ops, "central_checklist_lines"):
return tuple(ops.central_checklist_lines())
if operation.id == "info-identity" and hasattr(ops, "identity_lines"):
return tuple(ops.identity_lines())
if operation.id == "info-state" and hasattr(ops, "state_lines"):
return tuple(ops.state_lines())
if operation.id == "info-summary" and hasattr(ops, "environment_lines"):
lines = list(
ops.environment_lines(
self._repo, str(self.ctx.console_log), self._inventory,
self.ctx.dry_run, False,
)
)
if hasattr(ops, "state_lines"):
lines += [""]
lines += list(ops.state_lines())
return tuple(lines)
except Exception:
# Rapor uretilemedi: yerel (katalog) yoluna dusulur, hata gizlenmez.
return None
return None
# -- calistirma --------------------------------------------------------
def run(
self,
plan: CommandPlan,
on_line: Callable[[str], None] | None = None,
cancel: threading.Event | None = None,
) -> RunResult:
"""Komutlari canli ciktiyla calistirir; denetim kaydi modul uzerinden.
Not: bfos.runner.Runner komutu kendi stdio'suyla calistirir (TUI icin
dogru olan davranis budur) ve ciktiyi dondurmez; GUI metin kutusuna
satir satir akis icin buranin streaming calistiricisi kullanilir.
Denetim kaydi, modulun AuditLog'u varsa onun uzerinden yazilir, boylece
GUI ile TUI ayni bicimde kayit tutar.
"""
audit = self._audit_log()
redact = getattr(self._runner, "redact", None)
command_text = " && ".join(shlex.join(argv) for argv in plan.commands)
if callable(redact):
try:
command_text = str(redact(command_text))
except Exception:
pass
if audit is not None:
self._audit_write(audit, self._audit_level("AUDIT_RUN", "RUN"), command_text)
result = run_plan(plan, on_line=on_line, cancel=cancel, cwd=str(self.ctx.repo_root))
if audit is not None:
level = self._audit_level("AUDIT_OK", "OK") if result.ok \
else self._audit_level("AUDIT_FAIL", "FAIL")
self._audit_write(audit, level, f"exit={result.exit_code} :: {command_text}")
return result
def _audit_log(self):
"""bfos.runner.AuditLog (varsa) -- kurulamazsa None (kayit sessizce atlanir)."""
factory = getattr(self._runner, "AuditLog", None)
if factory is None:
return None
try:
return factory()
except Exception:
return None
@staticmethod
def _audit_write(audit, level: str, message: str) -> None:
try:
audit.write(level, message)
except Exception:
pass
def _audit_level(self, name: str, fallback: str) -> str:
config = getattr(self._ops, "config", None)
return str(getattr(config, name, fallback) or fallback)
def _declares(repo, playbook: str, var: str) -> bool:
"""Playbook VAR'i kullaniyor mu? (modulun okuma yolu varsa onu kullanir.)"""
checker = getattr(repo, "playbook_uses", None)
if not callable(checker):
return False
try:
return bool(checker(playbook, var))
except Exception:
return False
#: Modulun prompt turleri (bfos.operations PROMPT_*) icin arayuz secenekleri.
#: Anahtarlar modulun build_extra_vars() cevap anahtarlariyla AYNI olmalidir.
_PROMPT_OPTION_SPECS: Mapping[str, tuple[OptionSpec, ...]] = {
"none": (),
"packages": (
OptionSpec(
"packages",
"Paket pinleri (paket=sürüm, virgülle)",
required=True,
hint="Yalnız tam pin; 'latest' ve aralık kabul edilmez",
validator=validate_package_pins,
),
),
"service": (
OptionSpec(
"service",
"Servis adı",
kind="choice",
choices=SERVICE_ALLOWLIST,
required=True,
hint="Yalnız izinli servis adları",
validator=validate_service_name,
),
),
"script": (
OptionSpec(
"script",
"Script yolu",
required=True,
hint=f"İzinli dizinler: {', '.join(SCRIPT_DIRS)}",
validator=validate_script_path,
),
OptionSpec("script_args", "Script argümanları", default="", hint="İsteğe bağlı"),
),
"grace": (
OptionSpec(
"grace",
"Yeniden başlatma bekleme süresi (saniye)",
kind="int",
default="300",
validator=validate_grace,
),
),
"disk_pct": (
OptionSpec(
"disk_pct",
"Uyarı eşiği (%)",
kind="int",
default="",
hint="Boş = playbook varsayılanı (80)",
validator=validate_disk_pct,
),
),
"log_lines": (
OptionSpec(
"log_lines",
"Cihaz başına journal satırı",
kind="int",
default="500",
validator=validate_log_lines,
),
),
"expected_version": (
OptionSpec(
"expected_version",
"Beklenen Field OS sürümü",
default="",
hint="Boş = sapma (drift) denetimi atlanır",
validator=validate_expected_version,
),
),
}
def _prompt_options(prompt: str) -> tuple[OptionSpec, ...]:
return _PROMPT_OPTION_SPECS.get(prompt, ())
def load_backend(ctx: GuiContext):
"""Gercek modulleri dener; yoksa yerlesik katalog'a duser.
Donus: (backend, not) -- `not` Turkce bilgi satiridir (durum cubugunda
gosterilir), hata DEGILDIR.
"""
add_repo_to_path(ctx.repo_root)
try:
from bfos import operations as operations_module # type: ignore
except Exception as exc: # ImportError ve beklenmeyen hatalar
return (
CatalogueBackend(ctx),
f"bfos.operations bulunamadı ({type(exc).__name__}); yerleşik katalog "
"kullanılıyor (ortak Python kuralları).",
)
runner_module = None
try:
from bfos import runner as runner_module # type: ignore # noqa: F811
except Exception:
runner_module = None
if runner_module is None:
return (
CatalogueBackend(ctx),
"bfos.runner bulunamadı; komutlar yerleşik çalıştırıcıyla koşuluyor.",
)
try:
backend = ModuleBackend(operations_module, runner_module, ctx)
except BackendUnavailable as exc:
return CatalogueBackend(ctx), f"UYARI: {exc}"
except Exception as exc: # beklenmeyen modul hatasi arayuzu kilitlemesin
return (
CatalogueBackend(ctx),
f"UYARI: gerçek backend başlatılamadı ({type(exc).__name__}: {exc}); "
"yerleşik katalog kullanılıyor.",
)
catalogue = len(backend.list_operations())
fleet = len(backend._fleet_ids) # noqa: SLF001 - bilgi amacli sayim
return (
backend,
f"Gerçek backend: bfos.operations + bfos.runner "
f"({fleet} filo işlemi, {catalogue - fleet} yerel işlem).",
)
def add_repo_to_path(repo_root: Path) -> None:
root = str(repo_root)
if root not in sys.path:
sys.path.insert(0, root)
# ---------------------------------------------------------------------------
# Cikti paneli + canli akis
# ---------------------------------------------------------------------------
class OutputPane:
"""Komut ciktisi kuyusu: salt-okunur GORUNEN, icerlek bir metin alani.
Olculen sorun: alan duzenlenebilir gibi duruyordu ve ic bosluğu yoktu. Bu
yuzden alan bir kuyuya (``--bg-sunken``) oturur, 12px ic bosluk tasir,
``cursor="arrow"`` ve ``takefocus=0`` ile "buraya yazilmaz" der; yazi tipi
veri yuzudur (monospace) ki cikis kodlari ve kimlikler hizalansin.
Ciktinin ham hali KANITTIR, arayuz degil (docs/DESIGN.md kural 6): sonuc
once durum rozetinde gorunur, bu alan kaniti tutar ve gizlenebilir.
"""
def __init__(self, parent, styles: Mapping[str, str], height: int = 14,
log_path: str = DEFAULT_CONSOLE_LOG):
self.styles = styles
self.frame = ttk.Frame(parent, style=styles["panel"])
self.frame.pack(fill="both", expand=True)
head = ttk.Frame(self.frame, style=styles["panel"])
head.pack(fill="x", padx=theme.SP_3, pady=(theme.SP_3, theme.SP_2))
ttk.Label(head, text="Komut çıktısı", style=styles["panel_title"]).pack(side="left")
ttk.Label(
head,
text=f"salt-okunur · denetim kaydı: {log_path}",
style=styles["panel_faint"],
).pack(side="right")
self.well = ttk.Frame(self.frame, style=styles["sunken"])
self.well.pack(fill="both", expand=True,
padx=theme.SP_3, pady=(0, theme.SP_3))
self.text = tk.Text(
self.well,
height=height,
wrap="word",
font=theme.data_font(theme.SIZE_SM),
background=theme.BG_SUNKEN,
foreground=theme.TEXT,
relief="flat",
borderwidth=0,
highlightthickness=0,
padx=theme.SP_3,
pady=theme.SP_2,
cursor="arrow",
takefocus=0,
selectbackground=theme.ACCENT_SOFT,
selectforeground=theme.TEXT,
inactiveselectbackground=theme.ACCENT_SOFT,
)
self.text.pack(side="left", fill="both", expand=True)
bar = ttk.Scrollbar(self.well, orient="vertical", command=self.text.yview)
self.text.configure(yscrollcommand=bar.set)
bar.pack(side="right", fill="y")
self.text.configure(state="disabled")
self._queue: "queue.Queue[str]" = queue.Queue()
self._parent = parent
def append(self, line: str) -> None:
"""Herhangi bir is parcacigindan guvenle cagrilabilir (kuyruga yazar)."""
self._queue.put(line)
def _insert(self, line: str) -> None:
self.text.configure(state="normal")
self.text.insert("end", line + "\n")
self.text.configure(state="disabled")
def drain(self) -> None:
"""Ana is parcaciginda kuyrugu bosaltir (after ile periyodik cagrilir)."""
wrote = False
while True:
try:
line = self._queue.get_nowait()
except queue.Empty:
break
self._insert(line)
wrote = True
if wrote:
self.text.see("end")
self._parent.after(120, self.drain)
def clear(self) -> None:
self.text.configure(state="normal")
self.text.delete("1.0", "end")
self.text.configure(state="disabled")
def write_now(self, line: str) -> None:
self.append(line)
self.drain_once()
def drain_once(self) -> None:
while True:
try:
line = self._queue.get_nowait()
except queue.Empty:
break
self._insert(line)
self.text.see("end")
# -- gorunurluk (kanit erisilebilir kalir, ekrani kaplamaz) -------------
def set_visible(self, visible: bool) -> None:
"""Cikti kuyusunu gosterir/gizler; panel basligi yerinde kalir."""
if visible:
self.well.pack(fill="both", expand=True,
padx=theme.SP_3, pady=(0, theme.SP_3))
else:
self.well.pack_forget()
def is_visible(self) -> bool:
return bool(self.well.winfo_manager())
# ---------------------------------------------------------------------------
# Arayuz
# ---------------------------------------------------------------------------
class _LegacyBfGui: # pragma: no cover - compatibility-only, not an entry point
"""Blueforce Field OS Turkce masaustu arayuzu.
Yerlesim docs/DESIGN.md "Direction contract" yonergesine gore kurulur:
koyu bir baslik bandi (urun adi, hedef kimligi, cihaz durumu, yetki
seviyesi), altinda beyaz paneller halinde hedef cubugu, islem listesi ve
secenek paneli, bir kuyuda komut ciktisi ve tek bir eylem cubugu. Renk
yalnizca DURUM tasir; tum degerler bfos/ui_theme.py'den gelir (o da
web/static/tokens.css'i yansitir), bu dosyada sabit renk yoktur.
"""
#: Etiket sutunu genisligi: hedef satiri ve secenek satirlari AYNI hizada
#: baslar (olculen sorun: tutarsiz etiket genislikleri ve sikisik satir).
FIELD_LABEL_WIDTH = 22
def __init__(self, root, backend, ctx: GuiContext, *, full_mode: bool = False):
self.root = root
self.backend = backend
self.ctx = ctx
self.cancel_event = threading.Event()
self.running = False
self.option_widgets: dict[str, tuple[str, Any]] = {}
# Islem tanimlari BACKEND'den gelir (yerlesik katalog ya da gercek
# bfos.operations): secenek adlari backend ile ayni olmak zorundadir.
self._ops_by_id: dict[str, Operation] = {
op.id: op for op in backend.list_operations()
}
self._selected_operation: Operation | None = None
self._current_plan: CommandPlan | None = None
self._rows: dict[str, tuple[Operation, ...]] = {}
self._filters: dict[str, Any] = {}
self._output_visible = True
# Uygulama kabugu durumu: secili sayfa, secili satir, oturum
# etkinlik kaydi ve calisma kronometresi (sunum katmani; backend
# kurallarini degistirmez).
self._nav_buttons: dict[str, Any] = {}
self._pages: dict[str, Any] = {}
self._browser_inner: dict[str, Any] = {}
self._detail_body: dict[str, Any] = {}
self._activity: list[tuple[str, str, str, str, str]] = []
self._overview_card_frames: list[Any] = []
self._overview_quick_buttons: list[Any] = []
self._elapsed = widgets.Elapsed()
self._exec_title_text = "Etkin işlem yok."
self._current_page = CATEGORIES[0]
# Tema her widget'tan ONCE uygulanir: ttk stilleri dogduklari anda
# gecerli olmali, sonradan boyamak gri bir pencere ve titreme uretir.
self.styles = theme.apply_theme(root)
root.title(WINDOW_TITLE)
# Tk, geometry verilmezse en buyuk istegi olan cikti/alt cubuk
# kombinasyonuna gore pencereyi ekranin tamamindan da buyuk acabiliyor.
# Saha laptoplarinda ilk bakis tek ekrana sigmali; kullanici yine
# pencereyi buyutebilir. 860x620, detayli moddaki iki kolonun
# kullanilabilir en dar guvenli tabanidir.
root.geometry("1180x720")
root.minsize(860, 620)
root.columnconfigure(0, weight=1)
# Bantlar: baslik -> uyarilar -> hedef -> govde -> cikti -> eylem.
# minsize: minimum pencerede (860x620) islem listesi de cikti kuyusu da
# gorunur kalir; ikisi de agirlikla buyur, hicbiri digerini yutamaz.
root.rowconfigure(3, weight=3, minsize=200)
root.rowconfigure(4, weight=1, minsize=120)
self._build_header()
self._build_notices()
self._build_target_bar()
self.body = ttk.Frame(root, style=self.styles["frame"])
self.body.grid(row=3, column=0, sticky="nsew",
padx=theme.SP_4, pady=(theme.SP_3, 0))
self.simple_frame = ttk.Frame(self.body, style=self.styles["frame"])
self.detail_frame = ttk.Frame(self.body, style=self.styles["frame"])
self._build_simple_mode(self.simple_frame)
self._build_detail_mode(self.detail_frame)
output_area = ttk.Frame(root, style=self.styles["frame"])
output_area.grid(row=4, column=0, sticky="nsew",
padx=theme.SP_4, pady=(theme.SP_3, 0))
self._build_exec_header(output_area)
self.output = OutputPane(output_area, self.styles, height=5,
log_path=ctx.console_log)
self.output.drain()
self._build_footer()
self._apply_privilege_controls()
self.root.bind("<Configure>", self._on_window_resized)
self.show_mode(full=full_mode)
self._load_inventory()
self._tick_exec()
# -- kucuk yardimcilar -------------------------------------------------
def _badge(self, parent, word: str, style: str):
"""Sozcuklu rozet: renk TEK BASINA anlam tasimaz (kural 2)."""
return ttk.Label(parent, text=word, style=style)
def _bind_wrap(self, container, label, padding: int,
minimum: int = 220) -> None:
"""Etiketin sarma sinirini KAPSAYICININ genisliginden turetir.
Etiketi kendi genisligine baglamak, sarma genisligi etiketin istegiyle
birbirini besledigi icin hecelere bolunen metin uretiyordu (olculdu).
Kapsayicinin genisligi grid'den gelir ve kararli oldugu icin sarma
siniri oradan hesaplanir; alt sinir da metnin okunabilir kalmasi icin.
"""
def on_configure(event) -> None:
width = max(minimum, int(event.width) - padding)
if str(label.cget("wraplength")) != str(width):
label.configure(wraplength=width)
container.bind("<Configure>", on_configure)
def _set_status(self, word: str, severity: str = "idle") -> None:
"""Durum gostergesi: sozcuk + renk (duz gri bir etiket degil)."""
self.status.configure(text=word, style=theme.severity_style(severity))
# -- ust bilgi: koyu bant ---------------------------------------------
def _build_header(self) -> None:
"""Baslik bandi: urun adi, hedef kimligi, cihaz durumu, yetki.
Yetki seviyesi burada GORUNUR bir ogedir (rozet), gomulu bir cumle
degil: "bu pencereden ne yapabilirim?" sorusunun cevabi bir adim
geriden okunur (docs/DESIGN.md "Privilege").
"""
band = ttk.Frame(self.root, style=self.styles["header"])
band.grid(row=0, column=0, sticky="ew")
band.columnconfigure(2, weight=1)
identity = ttk.Frame(band, style=self.styles["header"])
identity.grid(row=0, column=0, sticky="w",
padx=theme.SP_4, pady=(theme.SP_3, theme.SP_1))
ttk.Label(identity, text=WINDOW_TITLE,
style=self.styles["header_title"]).pack(anchor="w")
ttk.Label(identity, text=APP_SUBTITLE,
style=self.styles["header_meta"]).pack(anchor="w")
target = ttk.Frame(band, style=self.styles["header"])
target.grid(row=0, column=1, sticky="w", padx=(theme.SP_5, 0),
pady=(theme.SP_3, theme.SP_1))
ttk.Label(target, text="HEDEF", style=self.styles["header_meta"]).pack(anchor="w")
self.header_target = ttk.Label(target, text="— seçilmedi",
style=self.styles["header_target"])
self.header_target.pack(anchor="w")
badges = ttk.Frame(band, style=self.styles["header"])
badges.grid(row=0, column=3, sticky="e",
padx=theme.SP_4, pady=(theme.SP_3, theme.SP_1))
# Iki rozet TEK satirda: bandin yuksekligi token'a (56px) yakin kalsin
# diye baskin satir urun adi + alt basligidir, rozetler degil.
self._state_word = device_state_word()
ttk.Label(badges, text="Cihaz durumu",
style=self.styles["header_meta"]).pack(side="left")
self.state_badge = self._badge(badges, self._state_word,
state_badge_style(self._state_word))
self.state_badge.pack(side="left", padx=(theme.SP_2, theme.SP_5))
self.privilege_level = privilege_level()
ttk.Label(badges, text="Yetki",
style=self.styles["header_meta"]).pack(side="left")
self.privilege_badge = self._badge(
badges, privilege_word(self.privilege_level),
theme.PRIVILEGE_BADGE_STYLE[self.privilege_level])
self.privilege_badge.pack(side="left", padx=(theme.SP_2, 0))
if self.ctx.dry_run:
self._badge(badges, "DENEME KİPİ",
self.styles["badge_info"]).pack(side="left",
padx=(theme.SP_3, 0))
def _build_notices(self) -> None:
"""Yetki ve kip uyarilari: basligin hemen altinda renkli bir serit.
Olculen sorun: kritik "Yonetici yetkisi yok" uyarisi monospace bir
paragrafta gomuluydu. Simdi yetki hem baslikta rozet, hem burada
sozcuk + renk; ustune duzene doken tam cumle.
"""
notices = ttk.Frame(self.root, style=self.styles["frame"])
shown = False
if not is_root():
# Soylenmezse operator kurulumu baslatir ve yari yolda yetki
# hatasiyla karsilasir; kapi bu yuzden acilista gorunur.
band = ttk.Frame(notices, style=self.styles["warn_band"])
band.pack(fill="x")
inner = ttk.Frame(band, style=self.styles["warn_band"])
inner.pack(fill="x", padx=theme.SP_3, pady=theme.SP_2)
ttk.Label(inner, text=privilege_word(PRIVILEGE_READ_ONLY),
style=self.styles["readonly"]).pack(side="left",
padx=(0, theme.SP_3))
ttk.Label(inner, text=ROOT_REQUIRED_NOTICE,
style=self.styles["warn_label"]).pack(side="left")
shown = True
if self.ctx.dry_run:
band = ttk.Frame(notices, style=self.styles["info_band"])
band.pack(fill="x", pady=(theme.SP_1, 0) if shown else (0, 0))
inner = ttk.Frame(band, style=self.styles["info_band"])
inner.pack(fill="x", padx=theme.SP_3, pady=theme.SP_2)
ttk.Label(inner, text="DENEME KİPİ",
style=self.styles["badge_info"]).pack(side="left",
padx=(0, theme.SP_3))
ttk.Label(inner, text="Komutlar gösterilir, çalıştırılmaz.",
style=self.styles["info_label"]).pack(side="left")
shown = True
if shown:
notices.grid(row=1, column=0, sticky="ew", padx=theme.SP_4,
pady=(theme.SP_3, 0))
# -- hedef cubugu ------------------------------------------------------
def _build_target_bar(self) -> None:
"""Hedef secer: parcali kapsam denetimi + baglama duyarli girdi.
Uc kapsam birbirini dislar (parcali denetimde tek secim gorunur);
basliktaki hedef kimligi her degisiklikte aynalanir, boylece kapsam
sayfalar arasi gezinirken de gorunur. Tum filo secimi yuksek etkili
bir uyari seridi acar. Alan adlari ve dogrulama akisi korunur
(dealer_entry / wave_combo / all_var / target_status).
"""
panel = ttk.Frame(self.root, style=self.styles["panel"])
panel.grid(row=2, column=0, sticky="ew", padx=theme.SP_4,
pady=(theme.SP_3, 0))
panel.columnconfigure(1, weight=1)
ttk.Label(panel, text="HEDEF", style=self.styles["panel_title"]).grid(
row=0, column=0, sticky="w", padx=theme.SP_3,
pady=(theme.SP_2, 0))
self.mode_button = ttk.Button(panel, text=SIMPLE_MODE_LABEL,
style=self.styles["button"],
command=self.toggle_mode)
self.mode_button.grid(row=0, column=2, sticky="e", padx=theme.SP_3,
pady=(theme.SP_2, 0))
self.dealer_var = tk.StringVar()
self.wave_var = tk.StringVar()
self.all_var = tk.IntVar(value=0)
self.scope_control = widgets.SegmentedControl(
panel, self.styles, SCOPE_OPTIONS, "device",
command=self._on_scope_selected)
self.scope_control.grid(row=1, column=0, sticky="w",
padx=theme.SP_3, pady=(theme.SP_2, 0))
inputs = ttk.Frame(panel, style=self.styles["panel"])
inputs.grid(row=2, column=0, columnspan=3, sticky="ew",
padx=theme.SP_3, pady=(theme.SP_1, 0))
inputs.columnconfigure(1, weight=1)
self._dealer_row = ttk.Frame(inputs, style=self.styles["panel"])
ttk.Label(self._dealer_row, text="Bayi no (8 hane)",
style=self.styles["field_label"],
width=16, anchor="e").pack(side="left",
padx=(0, theme.SP_2))
self.dealer_entry = ttk.Entry(self._dealer_row,
textvariable=self.dealer_var,
width=14, style=self.styles["entry"])
self.dealer_entry.pack(side="left")
self.dealer_entry.bind("<KeyRelease>",
lambda _e: self._on_dealer_change())
self._wave_row = ttk.Frame(inputs, style=self.styles["panel"])
ttk.Label(self._wave_row, text="Dalga",
style=self.styles["field_label"],
width=16, anchor="e").pack(side="left",
padx=(0, theme.SP_2))
self.wave_combo = ttk.Combobox(self._wave_row,
textvariable=self.wave_var,
values=list(WAVES), width=14,
state="readonly")
self.wave_combo.pack(side="left")
self.wave_combo.bind("<<ComboboxSelected>>",
lambda _e: self._on_target_change())
self._fleet_banner = ttk.Frame(inputs, style=self.styles["warn_band"])
fleet_inner = ttk.Frame(self._fleet_banner,
style=self.styles["warn_band"])
fleet_inner.pack(fill="x", padx=theme.SP_3, pady=theme.SP_2)
ttk.Label(
fleet_inner,
text=("Tüm filo seçili: işlem filonun TAMAMINI etkiler. "
"Dalga kapılı işlemler dalga dalga ilerler; yetki ve "
"kapı onayları her dalga için yeniden sorulur."),
style=self.styles["warn_label"],
justify="left",
).pack(anchor="w")
self._bind_wrap(fleet_inner, fleet_inner.winfo_children()[0],
theme.SP_6)
self._prod_note = ttk.Label(
inputs,
text=("production dalgası: çalıştırma sırasında “production” "
"sözcüğünü yazmanız istenir."),
style=self.styles["app_warn"],
justify="left",
)
self.target_status = ttk.Label(panel, text="",
style=self.styles["panel_muted"])
self.target_status.grid(row=3, column=0, columnspan=3, sticky="w",
padx=theme.SP_3, pady=(theme.SP_1, theme.SP_2))
self._sync_target_inputs()
def _on_scope_selected(self, key: str) -> None:
"""Parcali denetim degisimi: ilgili girdi gosterilir, kapsam aynalanir."""
self.all_var.set(1 if key == "all" else 0)
if key == "device":
self.dealer_entry.focus_set()
self._sync_target_inputs()
self._on_target_change()
def _sync_target_inputs(self) -> None:
"""Yalnizca secili kapsamin girdisini gosterir (digerleri gizlenir)."""
scope = (self.scope_control.get() if hasattr(self, "scope_control")
else "device")
if scope == "device":
self._wave_row.pack_forget()
self._fleet_banner.pack_forget()
self._prod_note.pack_forget()
self._dealer_row.pack(fill="x")
elif scope == "wave":
self._dealer_row.pack_forget()
self._fleet_banner.pack_forget()
self._wave_row.pack(fill="x")
if self.wave_var.get().strip() == PRODUCTION_WAVE:
self._prod_note.pack(fill="x", pady=(theme.SP_1, 0))
else:
self._prod_note.pack_forget()
else:
self._dealer_row.pack_forget()
self._wave_row.pack_forget()
self._prod_note.pack_forget()
self._fleet_banner.pack(fill="x")
# -- yurutm cekmecesi basligi ------------------------------------------
def _build_exec_header(self, parent) -> None:
"""Yurutme cekmecesi basligi: durum rozeti + islem/hedef/sure.
Bosken tek sessiz satirdir ("Etkin işlem yok."); calisirken islem
adi, hedef ve gecen sureyi gosterir; bitince sonucu ve toplam sureyi
yazar. Kanit (log kuyusu) altinda aynen durur.
"""
strip = ttk.Frame(parent, style=self.styles["panel"])
strip.pack(fill="x", padx=theme.SP_3, pady=(theme.SP_3, 0))
self.exec_pill = ttk.Label(strip, text="BOŞTA",
style=theme.severity_style("idle"))
self.exec_pill.pack(side="left")
self.exec_title = ttk.Label(strip, text=self._exec_title_text,
style=self.styles["panel_muted"])
self.exec_title.pack(side="left", padx=(theme.SP_3, 0))
self.exec_elapsed = ttk.Label(strip, text="",
style=self.styles["panel_faint"])
self.exec_elapsed.pack(side="right")
def _tick_exec(self) -> None:
"""Kronometre guncelleme: calisirken sure, degilse sessizlik."""
if self.running:
self.exec_elapsed.configure(text=self._elapsed.text())
elif not self._elapsed_text_hold():
self.exec_elapsed.configure(text="")
self.root.after(500, self._tick_exec)
def _elapsed_text_hold(self) -> bool:
"""Bitis suresi gosterimde kalsin mi (sonucu gorunur tutar)."""
return bool(self.exec_elapsed.cget("text"))
def _set_exec_running(self, operation_label: str,
target_label: str) -> None:
self._elapsed.start()
self._exec_title_text = f"{operation_label} · {target_label}"
self.exec_title.configure(text=self._exec_title_text)
self.exec_pill.configure(text="ÇALIŞIYOR",
style=theme.severity_style("info"))
def _set_exec_finished(self, ok: bool, exit_code: int) -> None:
self._elapsed.stop()
total = self._elapsed.text()
if ok:
self.exec_pill.configure(text="TAMAM",
style=theme.severity_style("ok"))
self.exec_title.configure(
text=f"{self._exec_title_text} · {total} sürdü")
else:
self.exec_pill.configure(text="HATA",
style=theme.severity_style("danger"))
self.exec_title.configure(
text=(f"{self._exec_title_text} · çıkış kodu {exit_code} · "
f"{total} sürdü"))
def _set_exec_idle(self, note: str = "Etkin işlem yok.") -> None:
self._exec_title_text = note
self.exec_title.configure(text=note)
self.exec_pill.configure(text="BOŞTA",
style=theme.severity_style("idle"))
self.exec_elapsed.configure(text="")
# -- eylem cubugu ------------------------------------------------------
def _build_footer(self) -> None:
"""Eylem cubugu, IKI satir: eylem + durum, altinda sessiz araclar.
Tek satira sigdirilan cubukta dugmeler birbirini sikistiriyor ve
sonuncusunun yazisi kirpiliyordu (olculdu: "Cikis" dugmesi "Ck"
gorunuyordu). Birincil eylem kendi satirini alir; yardimcilar ikinci
satirda durur, boylece hicbir kontrol digerinin yerini yemez.
"""
footer = ttk.Frame(self.root, style=self.styles["frame"])
footer.grid(row=5, column=0, sticky="ew", padx=theme.SP_4,
pady=(theme.SP_3, theme.SP_4))
footer.columnconfigure(0, weight=1)
action = ttk.Frame(footer, style=self.styles["frame"])
action.grid(row=0, column=0, sticky="ew")
self.run_button = ttk.Button(action, text=RUN_BUTTON_LABEL,
style=self.styles["accent_button"],
command=self._on_run_clicked)
self.run_button.pack(side="left")
# Kapali bir yazma isleminin NEDENI, dugmenin yanindaki ayni satirda
# durur (docs/DESIGN.md "Privilege": gizleme yok, nedeni goster).
self.run_reason = ttk.Label(action, text="", style=self.styles["app_warn"])
self.run_reason.pack(side="left", padx=(theme.SP_3, 0))
self.status = ttk.Label(action, text="HAZIR", style=self.styles["status"])
self.status.pack(side="right")
tools = ttk.Frame(footer, style=self.styles["frame"])
tools.grid(row=1, column=0, sticky="ew", pady=(theme.SP_2, 0))
ttk.Button(tools, text="Çıktıyı temizle", style=self.styles["button"],
command=self._clear_output).pack(side="left")
self.cancel_button = ttk.Button(tools, text="Çalışanı iptal et",
style=self.styles["button"],
command=self._cancel, state="disabled")
self.cancel_button.pack(side="left", padx=(theme.SP_2, 0))
self.check_var = tk.IntVar(value=0)
ttk.Checkbutton(tools, text="Yalnız denetim (--check)",
variable=self.check_var,
style=self.styles["check"]).pack(side="left",
padx=(theme.SP_3, 0))
self.output_button = ttk.Button(tools, text=OUTPUT_HIDE_LABEL,
style=self.styles["quiet_button"],
command=self._toggle_output)
self.output_button.pack(side="left", padx=(theme.SP_2, 0))
# Tek cikis kontrolu: burada (detayli mod) ya da basit modda ucuncu
# buyuk dugme olarak. Ikisi AYNI ANDA gorunmez (olculen sorun: "Cik"
# iki kez duruyordu).
self.exit_button = ttk.Button(tools, text="Çıkış",
style=self.styles["button"],
command=self.root.destroy)
# Depo ve katalog kaynagi: sessiz bir satir. Baslik bandini token
# yuksekliginde (56px) tutmak icin banddan asagi alindi ve hicbir
# dugmeyle yer yarisi yapmamasi icin kendi satirina konuldu: depo yolu
# okunmasi gereken bir bilgi degil, gorunur kalmasi yeter.
ttk.Label(
footer,
text=f"Depo: {self.ctx.repo_root} · {self.backend.name}",
style=self.styles["faint"],
).grid(row=2, column=0, sticky="w", pady=(theme.SP_2, 0))
# -- basit mod: genel bakis ------------------------------------------------
def _build_simple_mode(self, parent) -> None:
"""Genel Bakis: durum kartlari + hizli islemler (form degil).
Operator 3 saniyede anlar: hedef ne, cihaz ne durumda, yetki ne,
ne yapilabilir. Uc buyuk karar dugmesi korunur (Kur / Durum /
Cikis; aksan yalnizca Kur'da). Yetkisiz oturumda Kur kapali ve
nedeni dugmenin uzerinde yazar.
"""
page = ttk.Frame(parent, style=self.styles["frame"])
page.pack(fill="both", expand=True)
page.columnconfigure(0, weight=1)
page.columnconfigure(1, weight=1)
ttk.Label(page, text="Genel Bakış",
style=self.styles["section"]).grid(
row=0, column=0, columnspan=2, sticky="w",
pady=(0, theme.SP_3))
self._overview_cards = ttk.Frame(page, style=self.styles["frame"])
self._overview_cards.grid(row=1, column=0, columnspan=2,
sticky="ew", pady=(0, theme.SP_3))
for column in range(4):
self._overview_cards.columnconfigure(column, weight=1)
self._overview_values: dict[str, Any] = {}
for column, (key, title) in enumerate((
("target", "HEDEF"),
("state", "CİHAZ DURUMU"),
("privilege", "YETKİ"),
("inventory", "ENVANTER"),
)):
card = ttk.Frame(self._overview_cards, style=self.styles["panel"])
card.grid(row=0, column=column, sticky="nsew",
padx=(0, theme.SP_3) if column < 3 else (0, 0))
ttk.Label(card, text=title,
style=self.styles["panel_title"]).pack(
anchor="w", padx=theme.SP_3, pady=(theme.SP_3, theme.SP_1))
value = ttk.Label(card, text="—",
style=self.styles["panel_label"],
justify="left")
value.pack(anchor="w", padx=theme.SP_3)
sub = ttk.Label(card, text="", style=self.styles["panel_muted"],
justify="left")
sub.pack(anchor="w", padx=theme.SP_3,
pady=(theme.SP_1, theme.SP_3))
self._bind_wrap(card, sub, theme.SP_7)
self._overview_values[key] = (value, sub)
self._overview_card_frames.append(card)
self._refresh_overview()
box = ttk.Frame(page, style=self.styles["panel"])
box.grid(row=2, column=0, columnspan=2, sticky="ew")
inner = ttk.Frame(box, style=self.styles["panel"])
inner.pack(fill="x", padx=theme.SP_6, pady=theme.SP_5)
ttk.Label(inner, text="Hızlı işlemler",
style=self.styles["panel_title"]).pack(
anchor="w", pady=(0, theme.SP_3))
for label in SIMPLE_MODE_BUTTONS:
command = {
"Kur": self._simple_install,
"Durum": self._simple_status,
"Çıkış": self.root.destroy,
}[label]
# Aksan yalnizca "operatorun geldigi eylem" icin: Kur.
style_name = (self.styles["big_accent_button"] if label == "Kur"
else self.styles["big_button"])
button = ttk.Button(inner, text=label, command=command,
style=style_name, width=18)
button.pack(side="left", fill="x", expand=True,
padx=(0, theme.SP_2), pady=theme.SP_2)
self._overview_quick_buttons.append(button)
if label == "Kur":
self.install_button = button
ttk.Label(
inner,
text=(
"Kur: bu cihazı kurar (bayi no gerekir). Durum: salt-okunur özet.\n"
"Daha fazlası için üstteki '" + SIMPLE_MODE_LABEL + "' düğmesini kullanın."
),
style=self.styles["panel_muted"],
justify="left",
).pack(anchor="w", pady=(theme.SP_4, 0))
self._apply_privilege_controls()
def _overview_card(self, key: str, value: str, sub: str = "",
value_style: str | None = None) -> None:
"""Genel Bakis kartini gunceller (hedef degisince aynalanir)."""
if not hasattr(self, "_overview_values"):
return
value_label, sub_label = self._overview_values[key]
value_label.configure(text=value)
if value_style is not None:
value_label.configure(style=value_style)
sub_label.configure(text=sub)
def _refresh_overview(self) -> None:
"""Kartlar: hedef kimligi, cihaz durumu, yetki, envanter ozeti."""
if not hasattr(self, "_overview_values"):
return
target = self.current_target() if hasattr(self, "dealer_var") else None
if target is None or (target.kind == "dealer"
and not validate_dealer(target.value)):
scope_word = {"device": "cihaz", "wave": "dalga",
"all": "tüm filo"}.get(
self.scope_control.get()
if hasattr(self, "scope_control") else "device", "cihaz")
self._overview_card("target", "— seçilmedi",
f"Hedef türü: {scope_word}. "
"Yukarıdaki Hedef panelinden seçin.")
else:
kind_word = {"dealer": "Tek cihaz", "group": "Dalga grubu",
"all": "Tüm filo"}.get(target.kind, target.kind)
self._overview_card("target", target.label, kind_word)
word = device_state_word()
self._overview_card("state", word,
"state.json cihaz kaydı",
state_badge_style(word))
priv_level = (self.privilege_level
if hasattr(self, "privilege_level")
else privilege_level())
if priv_level == PRIVILEGE_FULL:
self._overview_card("privilege", privilege_word(priv_level),
"Kurulum ve değişiklik açık.",
self.styles["panel_ok"])
else:
self._overview_card("privilege", privilege_word(priv_level),
"Salt-okunur raporlar açık. "
"Değişiklik için: sudo bf",
self.styles["panel_danger"])
waves = self.backend.list_waves() if hasattr(self, "backend") else ()
note = self.backend.inventory_note() if hasattr(self, "backend") else ""
self._overview_card(
"inventory",
f"{len(waves)} dalga" if waves else "envanter yok",
", ".join(waves[:3]) + ("…" if len(waves) > 3 else "")
if waves else note[:80])
def _simple_install(self) -> None:
choice = self._choice_dialog(
"Kurulum",
"Bu cihazda ne yapılacak?\n\n"
"Ön kontrol (--check) hiçbir şey yazmaz; kurulum cihazı değiştirir.",
(("Ön kontrol (--check)", "install-preflight"),
("Kurulumu başlat", "install-local"),
("Vazgeç", None)),
)
if not choice:
return
operation = self._ops_by_id.get(choice)
if operation is not None:
self._execute(operation)
def _simple_status(self) -> None:
operation = self._ops_by_id.get("info-summary")
if operation is None:
return
self._execute(operation)
# -- detayli mod: uygulama kabugu ----------------------------------------
def _build_detail_mode(self, parent) -> None:
"""Detayli mod: kenar cubugu + sayfa alani (uygulama kabugu).
Kenar cubugunda bes katalog kategorisi (adetleriyle) ve Etkinlik
sayfasi vardir. Her kategori sayfasi solda islem tarayici, sagda
islem ayrintisidir. Ozet seritte TEK buyuk rakam (islem sayisi)
durur; yikici/salt-okunur dagilimi sessiz metindir.
"""
strip = ttk.Frame(parent, style=self.styles["frame"])
strip.pack(fill="x", pady=(0, theme.SP_3))
operations = tuple(self._ops_by_id.values())
writes = sum(1 for op in operations if operation_is_write(op))
ttk.Label(strip, text=str(len(operations)),
style=self.styles["figure"]).pack(side="left")
ttk.Label(strip, text="işlem", style=self.styles["muted"]).pack(
side="left", padx=(theme.SP_1, theme.SP_4))
ttk.Label(strip, text=f"{writes} yıkıcı", style=self.styles["muted"]).pack(
side="left", padx=(0, theme.SP_3))
ttk.Label(strip, text=f"{len(operations) - writes} salt-okunur",
style=self.styles["muted"]).pack(side="left")
if self.privilege_level == PRIVILEGE_READ_ONLY:
ttk.Label(strip, text=f"yazma işlemleri kapalı · {NON_ROOT_DISABLED_HINT}",
style=self.styles["app_warn"]).pack(side="right")
for category in CATEGORIES:
# Satirlar backend'den gelir (katalog sirasi korunur); backend bir
# kategoriyi bos birakirsa yerlesik tabloya dusulur.
self._rows[category] = tuple(
op for op in self._ops_by_id.values() if op.category == category
) or operations_by_category(category)
shell = ttk.Frame(parent, style=self.styles["frame"])
shell.pack(fill="both", expand=True)
shell.columnconfigure(0, minsize=theme.SIDEBAR_W)
shell.columnconfigure(1, weight=1)
shell.rowconfigure(0, weight=1)
self._sidebar = ttk.Frame(shell, style=self.styles["panel"])
self._sidebar.grid(row=0, column=0, sticky="nsw",
padx=(0, theme.SP_3))
ttk.Label(self._sidebar, text="BÖLÜMLER",
style=self.styles["panel_title"]).pack(
anchor="w", padx=theme.SP_3, pady=(theme.SP_3, theme.SP_2))
for page in SIDEBAR_PAGES:
self._build_nav_item(self._sidebar, page)
content = ttk.Frame(shell, style=self.styles["frame"])
content.grid(row=0, column=1, sticky="nsew")
content.columnconfigure(0, weight=1)
content.rowconfigure(0, weight=1)
for page in SIDEBAR_PAGES:
if page == ACTIVITY_PAGE:
frame = self._build_activity_page(content)
else:
frame = self._build_category_page(content, page)
self._pages[page] = frame
self._select_page(self._current_page)
self.root.bind("<Configure>", self._on_window_resized)
def _nav_count(self, page: str) -> int:
if page == ACTIVITY_PAGE:
return len(self._activity)
return len(self._rows.get(page, ()))
def _build_nav_item(self, parent, page: str) -> None:
"""Kenar cubugu satiri: ikon + etiket + adet (klavye ile erisilir)."""
row = tk.Frame(parent, background=theme.BG_PANEL,
highlightthickness=1,
highlightbackground=theme.BG_PANEL,
takefocus=1)
row.pack(fill="x", padx=theme.SP_2, pady=1)
icon = widgets.line_icon(row, PAGE_ICONS.get(page, "info"),
size=20, colour=theme.TEXT_MUTED,
bg=theme.BG_PANEL)
icon.pack(side="left", padx=(theme.SP_2, 0), pady=theme.SP_2)
label = tk.Label(row, text=page, background=theme.BG_PANEL,
foreground=theme.TEXT,
font=theme.ui_font(theme.SIZE_SM))
label.pack(side="left", padx=(theme.SP_2, 0))
count = tk.Label(row, text=str(self._nav_count(page)),
background=theme.BG_SUNKEN, foreground=theme.IDLE,
font=theme.ui_font(theme.SIZE_XS))
count.pack(side="right", padx=theme.SP_2)
self._nav_buttons[page] = (row, icon, label, count)
def activate(_event=None, page=page) -> None:
self._select_page(page)
for widget in (row, icon, label, count):
widget.bind("<Button-1>", activate)
widget.bind("<Return>", activate)
widget.bind("<space>", activate)
row.bind("<FocusIn>", lambda _e, r=row: r.configure(
highlightbackground=theme.FOCUS_RING))
row.bind("<FocusOut>", lambda _e, r=row: r.configure(
highlightbackground=r.cget("background")))
def _paint_nav(self) -> None:
"""Secili sayfa vurgulu; gerisi sessiz (vurgu = aksan, tek yerde)."""
for page, (row, icon, label, count) in self._nav_buttons.items():
selected = page == self._current_page
bg = theme.ACCENT_SOFT if selected else theme.BG_PANEL
row.configure(background=bg, highlightbackground=bg)
icon.configure(background=bg)
label.configure(background=bg)
count.configure(
background=theme.ACCENT_SOFT if selected else theme.BG_SUNKEN)
def _select_page(self, page: str) -> None:
"""Sayfa degisimi: hedef ve yurutme cubugu sabit kalir."""
if page not in self._pages:
return
self._current_page = page
for name, frame in self._pages.items():
if name == page:
frame.grid(row=0, column=0, sticky="nsew")
else:
frame.grid_remove()
if page == ACTIVITY_PAGE:
self._render_activity()
self._paint_nav()
def _on_window_resized(self, event) -> None:
"""Dar pencerede kabugu ve Genel Bakis kartlarini toplar."""
if event.widget is not self.root:
return
if event.width < 900:
if hasattr(self, "_sidebar"):
self._sidebar.grid_remove()
elif hasattr(self, "_sidebar"):
self._sidebar.grid()
cards = getattr(self, "_overview_card_frames", ())
if cards:
columns = 2 if event.width < 1120 else 4
for index, card in enumerate(cards):
card.grid_configure(
row=index // columns,
column=index % columns,
padx=(0, theme.SP_3) if index % columns < columns - 1 else (0, 0),
pady=(0, theme.SP_2) if columns == 2 and index // columns == 0 else (0, 0),
)
for column in range(4):
self._overview_cards.columnconfigure(column, weight=1 if column < columns else 0)
def _build_category_page(self, parent, category: str):
"""Kategori sayfasi: solda islem tarayici, sagda ayrinti paneli."""
page = ttk.Frame(parent, style=self.styles["frame"])
page.columnconfigure(0, weight=3)
page.columnconfigure(1, weight=2)
page.rowconfigure(0, weight=1)
browser = ttk.Frame(page, style=self.styles["panel"])
browser.grid(row=0, column=0, sticky="nsew", padx=(0, theme.SP_3))
browser.columnconfigure(0, weight=1)
browser.rowconfigure(2, weight=1)
head = ttk.Frame(browser, style=self.styles["panel"])
head.grid(row=0, column=0, sticky="ew", padx=theme.SP_3,
pady=(theme.SP_3, theme.SP_2))
ttk.Label(head, text="İŞLEMLER",
style=self.styles["panel_title"]).pack(side="left")
ttk.Label(head, text=f"{len(self._rows[category])} işlem",
style=self.styles["panel_faint"]).pack(side="right")
filt = ttk.Frame(browser, style=self.styles["panel"])
filt.grid(row=1, column=0, sticky="ew", padx=theme.SP_3,
pady=(0, theme.SP_2))
ttk.Label(filt, text="Süz:", style=self.styles["field_label"],
width=5, anchor="e").pack(side="left", padx=(0, theme.SP_2))
query = tk.StringVar()
self._filters[category] = query
entry = ttk.Entry(filt, textvariable=query, style=self.styles["entry"])
entry.pack(side="left", fill="x", expand=True)
entry.bind("<KeyRelease>",
lambda _e, c=category: self._render_browser(c))
holder, inner = widgets.scrollable_rows(browser, self.styles)
holder.grid(row=2, column=0, sticky="nsew",
padx=(theme.SP_2, 0), pady=(0, theme.SP_2))
self._browser_inner[category] = inner
self._render_browser(category)
detail = ttk.Frame(page, style=self.styles["panel"])
detail.grid(row=0, column=1, sticky="nsew")
detail.columnconfigure(0, weight=1)
detail.rowconfigure(1, weight=1)
ttk.Label(detail, text="İŞLEM AYRINTISI",
style=self.styles["panel_title"]).grid(
row=0, column=0, sticky="w", padx=theme.SP_3,
pady=(theme.SP_3, theme.SP_2))
body_holder, body = widgets.scrollable_rows(detail, self.styles)
body_holder.grid(row=1, column=0, sticky="nsew",
padx=(theme.SP_2, 0), pady=(0, theme.SP_2))
self._detail_body[category] = body
hint = ttk.Label(
detail,
text="Komut her zaman önce gösterilir ve onaysız çalıştırılmaz.",
style=self.styles["panel_muted"],
justify="left",
)
hint.grid(row=2, column=0, sticky="w", padx=theme.SP_3,
pady=(0, theme.SP_3))
self._bind_wrap(detail, hint, theme.SP_6)
placeholder = ttk.Label(
body,
text="Soldan bir işlem seçin; açıklaması, kapsamı, yetkisi ve "
"komut önizlemesi burada görünür.",
style=self.styles["panel_muted"],
justify="left",
)
placeholder.pack(anchor="w", padx=theme.SP_3, pady=theme.SP_3)
self._bind_wrap(body, placeholder, theme.SP_7)
return page
def _render_browser(self, category: str) -> None:
"""Tarayici satirlari: ikon + ad + aciklama + rozetler + ok."""
inner = self._browser_inner[category]
for child in inner.winfo_children():
child.destroy()
query = self._filters[category].get().strip().casefold()
shown = 0
for op in self._rows[category]:
if query and query not in f"{op.label} {op.id}".casefold():
continue
self._build_operation_row(inner, category, op)
shown += 1
if not shown:
ttk.Label(inner, text="Aramaya uyan işlem yok.",
style=self.styles["panel_muted"]).pack(
anchor="w", padx=theme.SP_3, pady=theme.SP_3)
def _build_operation_row(self, parent, category: str,
operation: Operation) -> None:
"""Tek tarayici satiri: tiklanabilir, klavye ile secilebilir."""
blocked = write_blocked_reason(operation) is not None
selected = (self._selected_operation is not None
and self._selected_operation.id == operation.id)
bg = theme.ACCENT_SOFT if selected else theme.BG_PANEL
row = tk.Frame(parent, background=bg, takefocus=1,
highlightthickness=1,
highlightbackground=(theme.FOCUS_RING if selected
else bg))
row.pack(fill="x", padx=theme.SP_2, pady=1)
bar = tk.Frame(row, background=(theme.ACCENT if selected
else bg),
width=3)
bar.pack(side="left", fill="y")
icon_colour = theme.DANGER if operation.destructive else theme.IDLE
icon = widgets.line_icon(
row, PAGE_ICONS.get(category, "info"), size=22,
colour=icon_colour, bg=bg)
icon.pack(side="left", padx=(theme.SP_2, 0), pady=theme.SP_2)
texts = tk.Frame(row, background=bg)
texts.pack(side="left", fill="x", expand=True,
padx=(theme.SP_2, 0), pady=theme.SP_1)
title = tk.Label(texts, text=operation.label, anchor="w",
background=bg,
foreground=(theme.IDLE if blocked else theme.TEXT),
font=theme.ui_font(theme.SIZE_SM, "bold"))
title.pack(fill="x")
short_desc = operation.description or "Açıklama yok."
if len(short_desc) > 96:
short_desc = short_desc[:95] + "…"
desc = tk.Label(texts, anchor="w", justify="left",
text=short_desc,
background=bg, foreground=theme.TEXT_MUTED,
font=theme.ui_font(theme.SIZE_XS))
desc.pack(fill="x")
badges = tk.Frame(row, background=bg)
badges.pack(side="left", padx=theme.SP_2)
type_style = (self.styles["badge_destructive"] if operation.destructive
else self.styles["badge_readonly"])
ttk.Label(badges,
text=(TYPE_WRITE_LABEL if operation.destructive
else TYPE_READ_LABEL),
style=type_style).pack(anchor="e", pady=(0, 2))
ttk.Label(badges, text=BLOCKED_LABEL if blocked else AVAILABLE_LABEL,
style=(self.styles["badge_closed"] if blocked
else self.styles["badge_open"])).pack(anchor="e")
chev = widgets.line_icon(row, "chevron", size=16,
colour=theme.TEXT_FAINT, bg=bg)
chev.pack(side="right", padx=(0, theme.SP_2))
def activate(_event=None, op=operation) -> None:
self._select_operation(op, category)
for widget in (row, icon, texts, title, desc, badges, chev, bar):
widget.bind("<Button-1>", activate)
row.bind("<Return>", activate)
row.bind("<space>", activate)
row.bind("<FocusIn>", lambda _e, r=row: r.configure(
highlightbackground=theme.FOCUS_RING))
row.bind("<FocusOut>", lambda _e, r=row, s=selected: r.configure(
highlightbackground=theme.FOCUS_RING if s else r.cget("background")))
def _select_operation(self, operation: Operation, category: str) -> None:
self._selected_operation = operation
self._selected_category = category
self._render_browser(category)
self._render_details(operation, category)
self._refresh_run_control()
def _build_activity_page(self, parent):
"""Etkinlik sayfasi: bu oturumda biten calismalar (zaman/islem/hedef/sonuc)."""
page = ttk.Frame(parent, style=self.styles["frame"])
page.columnconfigure(0, weight=1)
page.rowconfigure(1, weight=1)
ttk.Label(page, text="Etkinlik",
style=self.styles["section"]).grid(
row=0, column=0, sticky="w", pady=(0, theme.SP_1))
ttk.Label(
page,
text=("Bu oturumda çalıştırılan işlemler. Değişikliklerin tam "
"denetim kaydı sunucuda tutulur; burada sır (gizli bilgi) "
"gösterilmez."),
style=self.styles["muted"],
justify="left",
).grid(row=1, column=0, sticky="w", pady=(0, theme.SP_2))
holder, inner = widgets.scrollable_rows(page, self.styles)
holder.grid(row=2, column=0, sticky="nsew")
self._browser_inner[ACTIVITY_PAGE] = inner
self._render_activity()
return page
def _record_activity(self, operation_label: str, target_label: str,
ok: bool, detail: str) -> None:
"""Oturum kaydina bir satir ekler (sunu sozlugu degistirmez)."""
import datetime
stamp = datetime.datetime.now().strftime("%H:%M:%S")
word = "TAMAM" if ok else "HATA"
self._activity.append((stamp, operation_label, target_label, word,
detail))
count_widget = self._nav_buttons.get(ACTIVITY_PAGE)
if count_widget:
count_widget[3].configure(text=str(len(self._activity)))
if self._current_page == ACTIVITY_PAGE:
self._render_activity()
def _render_activity(self) -> None:
inner = self._browser_inner.get(ACTIVITY_PAGE)
if inner is None:
return
for child in inner.winfo_children():
child.destroy()
if not self._activity:
ttk.Label(inner, text="Henüz işlem çalıştırılmadı.",
style=self.styles["panel_label"]).pack(
anchor="w", padx=theme.SP_3, pady=(theme.SP_3, 0))
ttk.Label(
inner,
text=("Bir işlem çalıştırdığınızda zamanı, adı, hedefi ve "
"sonucu burada listelenir."),
style=self.styles["panel_muted"],
justify="left",
).pack(anchor="w", padx=theme.SP_3, pady=(0, theme.SP_3))
return
for stamp, label, target, word, detail in reversed(self._activity):
row = ttk.Frame(inner, style=self.styles["panel"])
row.pack(fill="x", padx=theme.SP_3, pady=theme.SP_1)
self._badge(row, word,
self.styles["badge_open"] if word == "TAMAM"
else self.styles["badge_destructive"]).pack(side="left")
ttk.Label(row, text=f"{stamp} · {label}",
style=self.styles["panel_label"]).pack(
side="left", padx=(theme.SP_2, 0))
ttk.Label(row, text=f"{target} · {detail}",
style=self.styles["panel_muted"]).pack(
side="left", padx=(theme.SP_2, 0))
def _render_details(self, operation: Operation, category: str) -> None:
"""Ayrinti paneli: baslik + aciklama + meta + uyari + secenek + onizleme.
Birincil eylem buradaki "Komutu göster ve çalıştır" düğmesidir
(alt çubuktakiyle aynı akışa girer); tehlike stili YALNIZCA bu
düğmededir, panelin tamamı kırmızıya boyanmaz.
"""
body = self._detail_body[category]
for child in body.winfo_children():
child.destroy()
self.option_widgets = {}
ttk.Label(body, text=operation.label,
style=self.styles["section"]).pack(
anchor="w", padx=theme.SP_3, pady=(theme.SP_3, theme.SP_1))
desc = ttk.Label(body, text=operation.description or "Açıklama yok.",
style=self.styles["panel_label"], justify="left")
desc.pack(anchor="w", padx=theme.SP_3)
self._bind_wrap(body, desc, theme.SP_7)
if operation.playbook:
ttk.Label(
body,
text=f"Ansible playbook: {PLAYBOOK_DIR_REL}/{operation.playbook}",
style=self.styles["panel_faint"],
justify="left",
).pack(anchor="w", padx=theme.SP_3, pady=(theme.SP_1, 0))
target = self.current_target()
target_word = (target.label if target is not None else "— seçilmedi")
widgets.meta_grid(body, self.styles, (
("HEDEF", target_word),
("KAPSAM", SCOPE_LABELS.get(operation.scope, operation.scope)),
("YETKİ", ("root gerekir (sudo bf)" if operation.needs_root
else "gerekmez (salt-okunur)")),
("TÜR", ("YIKICI — iki kez onay" if operation.destructive
else "salt-okunur")),
)).pack(fill="x", padx=theme.SP_3, pady=(theme.SP_3, 0))
reason = write_blocked_reason(operation)
if operation.destructive:
self._detail_banner(
body,
"BU İŞLEM YIKICIDIR: komut önce gösterilir, sonra iki kez "
"onay istenir.", danger=True)
if operation.scope in ("wave", "release") or (
target is not None and target.kind == "all"):
self._detail_banner(
body,
"Filo kapsamı: etki alanı geniştir. Onay penceresinde hedefi "
"yeniden denetleyin.", danger=False)
if reason:
self._detail_banner(body, reason, danger=False)
if operation.options:
form = ttk.Frame(body, style=self.styles["panel"])
form.pack(fill="x", padx=theme.SP_3, pady=(theme.SP_3, 0))
form.columnconfigure(1, weight=1)
row = 0
for spec in operation.options:
ttk.Label(form, text=spec.label + ":",
style=self.styles["field_label"],
width=self.FIELD_LABEL_WIDTH, anchor="e").grid(
row=row, column=0, sticky="e", padx=(0, theme.SP_2),
pady=theme.SP_1)
if spec.kind == "bool":
var = tk.BooleanVar(value=spec.default == "1")
widget = ttk.Checkbutton(form, variable=var,
style=self.styles["panel_check"])
self.option_widgets[spec.name] = ("bool", var)
elif spec.kind == "choice":
var = tk.StringVar(
value=spec.default or (spec.choices[0]
if spec.choices else ""))
widget = ttk.Combobox(form, textvariable=var,
values=list(spec.choices), width=22,
state="readonly")
self.option_widgets[spec.name] = ("str", var)
else:
var = tk.StringVar(value=spec.default)
widget = ttk.Entry(form, textvariable=var, width=22)
self.option_widgets[spec.name] = ("str", var)
widget.grid(row=row, column=1, sticky="w", pady=theme.SP_1)
if spec.hint:
ttk.Label(form, text=spec.hint,
style=self.styles["panel_faint"],
wraplength=theme.SP_7 * 3,
justify="left").grid(
row=row, column=2, sticky="w", padx=(theme.SP_2, 0))
row += 1
preview = widgets.CollapsiblePanel(body, self.styles,
"Komut önizlemesi")
preview.pack(fill="x", padx=theme.SP_3, pady=(theme.SP_3, 0))
preview_text_box = tk.Text(
preview.body, height=6, wrap="word",
font=theme.data_font(theme.SIZE_SM),
background=theme.BG_SUNKEN, foreground=theme.TEXT,
relief="flat", borderwidth=0, highlightthickness=0,
padx=theme.SP_3, pady=theme.SP_2,
cursor="arrow", takefocus=0)
preview_text_box.pack(fill="x", expand=True)
preview_text_box.insert(
"end", "Henüz oluşturulmadı. Aşağıdaki düğme komutu GERÇEK "
"backend ile üretir; hiçbir şey çalıştırmaz.")
preview_text_box.configure(state="disabled")
ttk.Button(preview.body, text="Önizlemeyi oluştur",
style=self.styles["quiet_button"],
command=lambda: self._refresh_preview(
operation, preview, preview_text_box)).pack(
anchor="w", pady=(theme.SP_2, 0))
action_row = ttk.Frame(body, style=self.styles["panel"])
action_row.pack(fill="x", padx=theme.SP_3, pady=theme.SP_3)
self._detail_run_button = ttk.Button(
action_row, text=RUN_BUTTON_LABEL,
style=(self.styles["danger_button"] if operation.destructive
else self.styles["accent_button"]),
command=lambda: self._execute(operation))
self._detail_run_button.pack(side="left")
if reason:
self._detail_run_button.configure(text=RUN_BUTTON_LABEL + " — kapalı",
style=self.styles["button"],
state="disabled")
ttk.Label(action_row, text=reason,
style=self.styles["app_warn"],
justify="left").pack(side="left",
padx=(theme.SP_3, 0))
elif operation.destructive:
ttk.Label(action_row, text="Yıkıcı işlem: iki kez onay istenir.",
style=self.styles["app_warn"],
justify="left").pack(side="left",
padx=(theme.SP_3, 0))
def _detail_banner(self, parent, text: str, danger: bool) -> None:
band = ttk.Frame(
parent,
style=(self.styles["warn_band"] if not danger
else self.styles["panel"]))
band.pack(fill="x", padx=theme.SP_3, pady=(theme.SP_2, 0))
label = ttk.Label(band, text=text,
style=(self.styles["warn_label"] if not danger
else self.styles["panel_danger"]),
justify="left")
label.pack(anchor="w", padx=theme.SP_3, pady=theme.SP_2)
self._bind_wrap(band, label, theme.SP_7)
def _refresh_preview(self, operation: Operation, panel,
box) -> None:
"""Komut onizlemesi: gercek backend uretir, calistirma YOK.
Kapilarin grant bayraklari henuz verilmedigi icin gosterilen komut
"temel" komuttur; kesin komut onaylardan sonra yeniden gosterilir
(akisi `_execute` korur). Backend reddi satir icinde Turkce yazi
olarak gorunur.
"""
target = (self.current_target()
if operation.is_fleet or operation.needs_dealer else None)
options = self._collect_options(operation)
try:
plan = self.backend.build_command(
operation, target, options,
check_mode=bool(self.check_var.get()))
lines = list(preview_text(plan, operation=operation,
target=target).splitlines())
except (CommandError, BackendUnavailable) as exc:
lines = [f"Önizleme üretilemedi: {exc}"]
box.configure(state="normal")
box.delete("1.0", "end")
for line in lines:
box.insert("end", line + "\n")
box.configure(state="disabled")
panel.set_expanded(True)
# -- hedef -------------------------------------------------------------
def _load_inventory(self) -> None:
waves = self.backend.list_waves()
self.wave_combo.configure(values=list(waves))
if waves:
self.wave_var.set(waves[0])
note = self.backend.inventory_note()
self.output.write_now(f"[i] {note}")
self._set_status("HAZIR", "idle")
self._on_target_change()
def _on_dealer_change(self) -> None:
value = self.dealer_var.get().strip()
problem = dealer_error(value) if value else None
# Alanin kendisi de durumu gosterir: gecersiz giris cercevede.
self.dealer_entry.configure(
style=self.styles["invalid_entry"] if problem else self.styles["entry"]
)
if problem:
self.target_status.configure(text=problem,
style=self.styles["panel_danger"])
elif value:
self.target_status.configure(
text=f"Bayi no geçerli: {dealer_to_device_id(value)}",
style=self.styles["panel_ok"])
self._set_status("HAZIR", "idle")
else:
self.target_status.configure(text="", style=self.styles["panel_muted"])
self._update_header_target()
def _on_target_change(self) -> None:
if self.all_var.get():
self.target_status.configure(
text="Tüm filo seçildi: dalga kapılı işlemler dalga dalga ilerler.",
style=self.styles["panel_muted"],
)
self.dealer_entry.configure(style=self.styles["entry"])
else:
self._on_dealer_change()
self._update_header_target()
def _update_header_target(self) -> None:
"""Basliktaki hedef kimligi her secim degisikliginde guncellenir.
Ayni anda Genel Bakis karti tazelenir ve uretim/tum-filo notlari
kapsama gore esitlenir. Secili islemin secenegi yoksa (kayip veri
riski yok) ayrintidaki hedef satiri da tazelenir; secenekli
islemlerde operatorun girdisi korunur (calisma ani penceresi
zaten guncel hedefi gosterir).
"""
target = self.current_target()
if target is None or (target.kind == "dealer"
and not validate_dealer(target.value)):
self.header_target.configure(text="— seçilmedi")
else:
self.header_target.configure(text=target.label)
self._sync_target_inputs()
self._refresh_overview()
operation = self._selected_operation
category = getattr(self, "_selected_category", None)
if (operation is not None and not operation.options
and category in self._detail_body):
self._render_details(operation, category)
def current_target(self) -> Target | None:
# Parcali denetim tek dogruluk kaynagidir: gorunur kapsam neyse hedef
# odur. (Eskiden bayi metni dalgayi ezerdi; tum-filo uyarisi ekranda
# dururken komut tek cihaza gitmemelidir.)
scope = (self.scope_control.get() if hasattr(self, "scope_control")
else "device")
if scope == "all" or self.all_var.get():
return Target("all", "all")
if scope == "wave":
wave = self.wave_var.get().strip()
return Target("group", wave) if wave else None
dealer = self.dealer_var.get().strip()
if dealer:
return Target("dealer", dealer)
return None
# -- yetki ve birincil eylem ------------------------------------------
def _apply_privilege_controls(self) -> None:
"""Yetkisiz oturum: yazma dugmeleri KAPALI, nedeni dugmenin uzerinde.
Gizleme yok: yetkisiz operatorde "bu islem var ama kapali" bilgisi
gorunur kalmali, yoksa araci eksik sanir (docs/DESIGN.md "Privilege").
"""
if hasattr(self, "install_button"):
if self.privilege_level == PRIVILEGE_READ_ONLY:
self.install_button.configure(
text="Kur — kapalı (yönetici yetkisi gerekir: sudo bf)",
state="disabled",
)
else:
self.install_button.configure(text=SIMPLE_MODE_BUTTONS[0],
state="normal")
if hasattr(self, "run_button"):
self._refresh_run_control()
def _refresh_run_control(self) -> None:
"""Birincil eylem: aksan, yikici icin kirmizi, yetkisizde kapali."""
operation = self._selected_operation
self.run_reason.configure(text="")
if operation is None:
self.run_button.configure(text=RUN_BUTTON_LABEL,
style=self.styles["accent_button"],
state="normal")
return
reason = write_blocked_reason(operation)
if reason:
self.run_button.configure(text=RUN_BUTTON_LABEL + " — kapalı",
style=self.styles["button"],
state="disabled")
self.run_reason.configure(text=reason, style=self.styles["app_warn"])
return
self.run_button.configure(
text=RUN_BUTTON_LABEL,
style=(self.styles["danger_button"] if operation.destructive
else self.styles["accent_button"]),
state="normal",
)
if operation.destructive:
self.run_reason.configure(text="Yıkıcı işlem: iki kez onay istenir.",
style=self.styles["app_warn"])
def _toggle_output(self) -> None:
"""Kanit panelini gizler/gosterir; sonuc durum rozetinde kalir."""
self._output_visible = not self._output_visible
self.output.set_visible(self._output_visible)
self.output_button.configure(
text=OUTPUT_HIDE_LABEL if self._output_visible else OUTPUT_SHOW_LABEL
)
# -- mod degistirme ----------------------------------------------------
def show_mode(self, *, full: bool) -> None:
self.simple_frame.pack_forget()
self.detail_frame.pack_forget()
if full:
self.detail_frame.pack(fill="both", expand=True)
self.mode_button.configure(text=SIMPLE_MODE_BACK_LABEL)
# Tek cikis kontrolu: detayli modda alt cubukta, basit modda ucuncu
# buyuk dugme olarak durur; ikisi ayni anda gorunmez.
self.exit_button.pack(side="right")
else:
self.simple_frame.pack(fill="both", expand=True)
self.mode_button.configure(text=SIMPLE_MODE_LABEL)
self.exit_button.pack_forget()
self.full_mode = full
def toggle_mode(self) -> None:
self.show_mode(full=not self.full_mode)
# -- calistirma akisi --------------------------------------------------
def _on_run_clicked(self) -> None:
if self._selected_operation is None:
messagebox.showinfo(
"Seçim gerekli",
"Önce listeden bir işlem seçin, sonra 'Komutu göster ve çalıştır' düğmesine basın.",
)
return
self._execute(self._selected_operation)
def _collect_options(self, operation: Operation) -> dict[str, str]:
options: dict[str, str] = {}
for spec in operation.options:
entry = self.option_widgets.get(spec.name)
if not entry:
continue
kind, variable = entry
if kind == "bool":
options[spec.name] = "1" if variable.get() else "0"
else:
options[spec.name] = str(variable.get()).strip()
return options
def _execute(self, operation: Operation) -> None:
if self.running:
messagebox.showinfo("Bekleyin", "Zaten çalışan bir işlem var.")
return
target = self.current_target() if operation.is_fleet or operation.needs_dealer else None
if operation.needs_dealer and (target is None or target.kind != "dealer"):
messagebox.showerror(
"Bayi numarası gerekli",
"Bu işlem için bayi numarası girin (8 hane, örn. 12010193).",
)
return
if target is not None and target.kind == "dealer":
problem = dealer_error(target.value)
if problem:
messagebox.showerror("Geçersiz bayi numarası", problem)
return
options = self._collect_options(operation)
check_mode = bool(self.check_var.get())
try:
draft = self.backend.build_command(
operation, target, options, check_mode=check_mode
)
except CommandError as exc:
messagebox.showerror("Komut üretilemedi", str(exc))
return
except BackendUnavailable as exc:
messagebox.showerror("Backend hatası", str(exc))
return
# Salt-okunur rapor (komut yok): dogrudan cikti alanina yazilir.
if draft.is_report_only:
self._current_plan = draft
self._show_plan(draft, title=f"Salt-okunur ekran: {operation.label}")
self._start_run(draft)
return
# 1) Taslak komut GORSTERILIR ve onaylanir: onaysiz hicbir sey calismaz.
if not self._confirm_plan(draft, operation, target, title="Komutu onaylayın"):
self.output.write_now("[i] Kullanıcı vazgeçti; komut çalıştırılmadı.")
return
# 2) Acik kapilar: dalga adi, production, filo geneli, yikici ikinci onay.
grants = self._collect_grants(operation, target)
if grants is None:
self.output.write_now("[i] Kapı onayı verilmedi; işlem iptal edildi.")
return
# 3) Kapilar eklendikten sonraki KESIN komut yeniden gosterilir.
try:
plan = self.backend.build_command(
operation, target, options, grants=grants, check_mode=check_mode
)
except CommandError as exc:
messagebox.showerror("Komut üretilemedi", str(exc))
return
if plan.lines != draft.lines:
if not self._confirm_plan(
plan, operation, target,
title="KESİN KOMUT (onayların eklendiği son hâl)",
):
self.output.write_now("[i] Kesin komut onaylanmadı; işlem iptal edildi.")
return
self._current_plan = plan
self._show_plan(plan, title=f"Çalıştırılıyor: {operation.label}")
self._start_run(plan)
def _confirm_plan(
self, plan: CommandPlan, operation: Operation,
target: Target | None, title: str,
) -> bool:
"""Komutu gosteren onay penceresi: ISLEM / HEDEF / KAPSAM / ETKI.
Yikici planda odak ve varsayilan Vazgec'tedir (Enter ile yikim
onaylanamaz); filo/uretim kapsami ayrica vurgulanir. Onaysiz hicbir
sey calismaz (akisi `_execute` korur).
"""
target_label = target.label if target is not None else "—"
scope_label = SCOPE_LABELS.get(operation.scope, operation.scope)
if plan.destructive:
effect = ("Seçili hedefte kalıcı değişiklik yapar; hizmetler "
"kesintiye uğrayabilir.")
elif plan.is_report_only:
effect = "Salt-okunur rapor: hiçbir şey çalıştırılmaz."
else:
effect = "Salt-okunur denetim: sistemde değişiklik yapmaz."
if plan.notes:
effect += " " + " ".join(plan.notes)
scope_warning = None
if target is not None and target.kind == "all":
scope_warning = ("KAPSAM — TÜM FİLO: bu komut filodaki TÜM "
"cihazları etkiler.")
elif ((target is not None and target.kind == "group"
and target.value == PRODUCTION_WAVE)
or operation.scope == "release"):
scope_warning = ("KAPSAM — ÜRETİM: etkisi en geniş dalga. "
"Hedefi ve komutu satır satır denetleyin.")
preview_lines = (list(plan.lines) if plan.lines
else list(plan.report_lines or ()))
return widgets.ask_confirmation(
self.root, self.styles,
title=title,
operation=operation.label,
target_label=target_label,
scope_label=scope_label,
effect=effect,
preview_lines=preview_lines,
confirm_label=operation.label,
scope_warning=scope_warning,
danger=bool(plan.destructive),
)
def _collect_grants(
self, operation: Operation, target: Target | None
) -> dict[str, bool] | None:
"""Backend'in istedigi acik onaylari sorar; iptalde None doner."""
try:
steps = self.backend.gate_steps(operation, target)
except CommandError as exc:
messagebox.showerror("Kapı onayları alınamadı", str(exc))
return None
grants: dict[str, bool] = {}
for step in steps:
if step.kind == "token":
answer = self._token_dialog(step.prompt or "Onay", step.message, step.token)
if answer is None or answer.strip() != step.token:
return None
else:
if not messagebox.askyesno(
step.prompt or "Onay",
step.message or "Devam edilsin mi?",
icon="warning",
default="no",
):
return None
for name in step.unlocks:
grants[name] = True
return grants
def _show_plan(self, plan: CommandPlan, title: str) -> None:
"""Komutu/raporu cikti alanina yazar (kayit ve gozden gecirme icin)."""
self.output.write_now("")
self.output.write_now("=" * 72)
self.output.write_now(f" {title}")
if plan.report_lines and not plan.lines:
for line in plan.report_lines:
self.output.write_now(" " + line)
else:
for line in plan.lines:
self.output.write_now(" " + line)
self.output.write_now("=" * 72)
def _start_run(self, plan: CommandPlan) -> None:
if plan.needs_root and not write_operations_enabled():
# The button is disabled for this case, so reaching here means the
# gate was bypassed (stale widget state, a future caller). Refusing
# in the one place every run goes through keeps the rule true for
# the whole surface instead of one control.
self.output.write_now("[ENGELLENDİ] " + NON_ROOT_DISABLED_HINT)
self._set_status("YETKİ YOK", "danger")
self._set_exec_idle("Engellendi: yetki yok.")
return
self.running = True
self.cancel_event.clear()
self.run_button.configure(state="disabled")
self.cancel_button.configure(state="normal")
self._set_status("ÇALIŞIYOR", "info")
operation = self._ops_by_id.get(plan.operation_id)
self._run_label = (operation.label if operation is not None
else plan.operation_id)
self._run_target = plan.target_label or "—"
self._set_exec_running(self._run_label, self._run_target)
def worker() -> None:
try:
result = self.backend.run(
plan, on_line=self.output.append, cancel=self.cancel_event
)
except Exception as exc: # backend hatasi arayuzu kilitlemesin
self.output.append(f"[HATA] Çalıştırma başarısız: {exc}")
result = RunResult(1, False, "")
self.root.after(0, lambda: self._finish_run(result))
threading.Thread(target=worker, daemon=True).start()
def _finish_run(self, result: RunResult) -> None:
self.running = False
self._refresh_run_control()
self.cancel_button.configure(state="disabled")
self._set_exec_finished(bool(result.ok), result.exit_code)
label = getattr(self, "_run_label", "işlem")
target = getattr(self, "_run_target", "—")
if result.ok:
self.output.append("[OK] Komut başarıyla tamamlandı (çıkış kodu 0).")
self._set_status("TAMAM (çıkış kodu 0)", "ok")
self._record_activity(label, target, True, "çıkış kodu 0")
else:
self.output.append(f"[HATA] Komut {result.exit_code} koduyla bitti.")
self._set_status(f"HATA (çıkış kodu {result.exit_code})", "danger")
self._record_activity(label, target, False,
f"çıkış kodu {result.exit_code}")
def _cancel(self) -> None:
self.cancel_event.set()
self._set_status("İPTAL İSTENDİ", "warn")
def _clear_output(self) -> None:
self.output.clear()
# -- kucuk diyaloglar --------------------------------------------------
def _choice_dialog(self, title: str, message: str,
choices: Sequence[tuple[str, str | None]]) -> str | None:
dialog = tk.Toplevel(self.root)
dialog.title(title)
dialog.configure(bg=theme.BG_APP)
dialog.transient(self.root)
dialog.grab_set()
result: dict[str, str | None] = {"value": None}
ttk.Label(dialog, text=message, style=self.styles["dialog"],
wraplength=460, justify="left").pack(
padx=theme.SP_4, pady=theme.SP_4, anchor="w"
)
def choose(value: str | None) -> None:
result["value"] = value
dialog.destroy()
row = ttk.Frame(dialog, style=self.styles["frame"])
row.pack(fill="x", padx=theme.SP_4, pady=(0, theme.SP_4))
for index, (label, value) in enumerate(choices):
# Ilk secenek aksandir (operatorun geldigi eylem), "Vazgec" sessiz.
style_name = (self.styles["quiet_button"] if value is None
else self.styles["accent_button"] if index == 0
else self.styles["button"])
ttk.Button(row, text=label, style=style_name,
command=lambda v=value: choose(v)).pack(
side="left", padx=(0, theme.SP_2)
)
dialog.wait_window()
return result["value"]
def _token_dialog(self, title: str, message: str, token: str) -> str | None:
"""Yazili kapi onayi: operator jetonu birebir yazar (karsilastirma ayni).
Pencere yalnizca metni toplar; eslesme kurali `_collect_grants`
icinde degismeden durur. Filo/uretim kapsami pencerede ayrica
vurgulanir.
"""
target = self.current_target()
scope_warning = None
if target is not None and target.kind == "all":
scope_warning = ("KAPSAM — TÜM FİLO: yazdığınız sözcük filonun "
"TAMAMI için kapıyı açar.")
elif target is not None and target.kind == "group" and target.value == PRODUCTION_WAVE:
scope_warning = ("KAPSAM — ÜRETİM: yazdığınız sözcük üretim "
"dalgası için kapıyı açar.")
return widgets.ask_token(
self.root, self.styles,
title=title,
message=message,
scope_warning=scope_warning,
)
# ---------------------------------------------------------------------------
# Katalog yazdirma (pencere ACMADAN dogrulama icin)
# ---------------------------------------------------------------------------
def print_catalogue(backend, stream=None) -> None:
out = stream or sys.stdout
operations = backend.list_operations()
print(f"Blueforce Field OS — işlem kataloğu ({len(operations)} işlem)", file=out)
for category in CATEGORIES:
items = [op for op in operations if op.category == category]
print(f"\n== {category} ({len(items)} işlem) ==", file=out)
for op in items:
kind = "YIKICI" if op.destructive else "salt-okunur"
playbook = op.playbook or "-"
print(
f" {op.id:<24} {op.label:<46} kapsam={op.scope:<8} {kind:<11} playbook={playbook}",
file=out,
)
# ---------------------------------------------------------------------------
# Yetki kapisi
# ---------------------------------------------------------------------------
def is_root() -> bool:
"""True when the process can install, write state and drive Ansible.
The installer writes to the filesystem, manages services and applies the
firewall baseline, so the GUI is expected to run under sudo. Reports read
the repository and the inventory and work without it, which is why the
console still starts as a normal user and explains the difference instead
of refusing to launch.
"""
geteuid = getattr(os, "geteuid", None)
return False if geteuid is None else geteuid() == 0
ROOT_REQUIRED_NOTICE = (
"Yönetici yetkisi yok: salt-okunur raporlar çalışır, kurulum ve değişiklik "
"gerektiren işlemler kapalıdır. Tam yetki için: sudo bf"
)
NON_ROOT_DISABLED_HINT = "Bu işlem yönetici yetkisi gerektirir (sudo bf)."
def write_operations_enabled() -> bool:
"""Whether operations that change the device may be offered."""
return is_root()
# ---------------------------------------------------------------------------
# Baslatma
# ---------------------------------------------------------------------------
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="python3 -m bfos.ui_gui",
description="Blueforce Field OS masaüstü arayüzü (Türkçe, tkinter).",
)
parser.add_argument("--full", action="store_true",
help="Doğrudan detaylı modda başlat (kategori sekmeleri).")
parser.add_argument("--install", action="store_true",
help="Yalnızca bağımsız kurulum konsolunu aç.")
parser.add_argument("--portal", action="store_true",
help="Yalnızca bağımsız uzaktan yönetim portalını aç.")
parser.add_argument("--repo", default=None, help="Depo kökü (varsayılan: bfos'un üst dizini).")
parser.add_argument("--inventory", default=None, help="Ansible envanteri (varsayılan: otomatik).")
parser.add_argument("--dry-run", action="store_true",
help="Komutları göster, hiçbir şey çalıştırma.")
parser.add_argument("--list", action="store_true",
help="Pencere açmadan işlem kataloğunu yazdır ve çık.")
return parser.parse_args(argv)
def main(argv: Sequence[str] | None = None) -> int:
args = parse_args(argv)
if args.install and args.portal:
print("--install ve --portal aynı anda kullanılamaz.", file=sys.stderr)
return 2
ctx = GuiContext.detect(repo=args.repo, inventory=args.inventory, dry_run=args.dry_run)
backend, note = load_backend(ctx)
if args.list:
print_catalogue(backend)
print(f"\n[i] {note}")
return 0
if not HAVE_TKINTER:
print(tkinter_missing_message(TK_IMPORT_ERROR), file=sys.stderr)
return 3
problem = headless_error()
if problem:
print(problem, file=sys.stderr)
return 3
root = tk.Tk()
try:
from bfos.ui_surfaces import launch_surface
launch_surface(
root, backend, ctx,
"install" if args.install else "portal" if args.portal or args.full else None,
)
except tk.TclError as exc: # DISPLAY var ama erisilemiyor
print(window_start_error(exc), file=sys.stderr)
return 3
root.mainloop()
return 0
if __name__ == "__main__": # pragma: no cover
sys.exit(main())
Reference in a new issue View git blame Copy permalink