1 Commits

Author SHA1 Message Date
engel 4322393de7 Release 1.2.4: export lease table for sibling tools; auto-select single NIC
Release / build (push) Has been cancelled
- Write a lease-table snapshot to %PROGRAMDATA%\vegman\leases.json (contract consumed by vrcx for BMC/SDS discovery)
- Auto-select the network adapter when only one is found (configurable via auto_select_single; asked on first run)
2026-06-03 17:55:51 +03:00
6 changed files with 117 additions and 13 deletions
+1 -1
View File
@@ -5,5 +5,5 @@ The single source of truth for the project version. Bump this before tagging
a release; CI reads the tag, the code reads this constant.
"""
__version__ = "1.2.3"
__version__ = "1.2.4"
REPO = "engel/dhcpsrv" # self-hosted Gitea (git.engelgardt23.ru)
+6 -2
View File
@@ -21,7 +21,7 @@ from .config import load_config
from .i18n import set_language, t
def _select_nic(console: Console) -> dict | None:
def _select_nic(console: Console, auto_single: bool = True) -> dict | None:
console.rule(f"[bold cyan]{t('available_adapters')}")
adapters = list_adapters()
if not adapters:
@@ -30,6 +30,10 @@ def _select_nic(console: Console) -> dict | None:
for i, a in enumerate(adapters, 1):
ip = a.get("IPv4") or ""
console.print(f" {i}) [{a['Status']}] {a['Name']} ({a['Description']}) {ip}")
if auto_single and len(adapters) == 1:
only = adapters[0]
console.print(f"[green]{t('only_adapter', name=only['Name'])}[/]")
return only
while True:
s = Prompt.ask(t("select_adapter")).strip()
if s.isdigit() and 1 <= int(s) <= len(adapters):
@@ -64,7 +68,7 @@ def main() -> None:
console.print(title)
console.print()
nic = _select_nic(console)
nic = _select_nic(console, cfg_data.get("auto_select_single", True))
if not nic:
input(t("press_enter")); return
+37 -10
View File
@@ -20,16 +20,25 @@ CONFIG_HEADER = """\
# ---------------------------------------------------------------------------
# dhcpsrv configuration
#
# To change the interface language, edit the 'language' value below.
# Valid values: en, ru
# language : interface language. Valid: en, ru
# auto_select_single : auto-pick the network adapter when only one is found,
# without asking (yes/no).
#
# Чтобы сменить язык интерфейса, измените значение 'language' ниже.
# Допустимые значения: en, ru
# language : язык интерфейса. Допустимо: en, ru
# auto_select_single : автоматически выбирать сетевой адаптер, если он
# единственный, не спрашивая (yes/no).
# ---------------------------------------------------------------------------
"""
def _to_bool(s: str, default: bool) -> bool:
s = (s or "").strip().lower()
if s in ("yes", "true", "y", "1", "on"): return True
if s in ("no", "false", "n", "0", "off"): return False
return default
def app_dir() -> Path:
"""Directory holding the running executable (or source folder when run via python)."""
if getattr(sys, "frozen", False):
@@ -60,10 +69,26 @@ def _ask_language() -> str:
print("Please enter 1 or 2 / Введите 1 или 2")
def _write_config(lang: str) -> None:
def _ask_auto_select() -> bool:
"""First-run prompt — auto-select a lone adapter? Bilingual stdin."""
print()
print("Auto-select the network adapter when only one is found? (y/n)")
print("Автовыбор сетевого адаптера, если он единственный? (y/n)")
while True:
try:
c = input("> ").strip().lower()
except (EOFError, KeyboardInterrupt):
return True
if c in ("y", "yes", "д", "да", "1"): return True
if c in ("n", "no", "н", "нет", "0"): return False
print("y / n")
def _write_config(lang: str, auto_select: bool) -> None:
path = config_path()
cp = configparser.ConfigParser()
cp["General"] = {"language": lang}
cp["General"] = {"language": lang,
"auto_select_single": "yes" if auto_select else "no"}
with path.open("w", encoding="utf-8") as f:
f.write(CONFIG_HEADER)
cp.write(f)
@@ -75,20 +100,22 @@ def load_config() -> dict:
path = config_path()
if not path.exists():
lang = _ask_language()
auto = _ask_auto_select()
try:
_write_config(lang)
_write_config(lang, auto)
except OSError:
# read-only location — fall back to in-memory default
pass
return {"language": lang}
return {"language": lang, "auto_select_single": auto}
cp = configparser.ConfigParser()
try:
cp.read(path, encoding="utf-8")
except (configparser.Error, OSError):
return {"language": DEFAULT_LANG}
return {"language": DEFAULT_LANG, "auto_select_single": True}
lang = (cp.get("General", "language", fallback=DEFAULT_LANG) or DEFAULT_LANG).strip().lower()
if lang not in SUPPORTED_LANGS:
lang = DEFAULT_LANG
return {"language": lang}
auto = _to_bool(cp.get("General", "auto_select_single", fallback="yes"), True)
return {"language": lang, "auto_select_single": auto}
+20
View File
@@ -16,6 +16,7 @@ from typing import Callable, Optional
from .network import ping_one
from .i18n import t as _t
from .leases import write_leases
# ---------- helpers ----------
@@ -147,6 +148,23 @@ class DhcpServer:
o += bytes([255])
return bytes(pkt) + bytes(o)
# --- lease-table export (contract for sibling tools, e.g. vrcx) ---
def export_leases(self) -> None:
"""Snapshot the lease table to the shared JSON file. Best-effort."""
with self.lock:
leases = [
{
"ip": int2ip(c["ip_int"]),
"mac": mac,
"host": c.get("host", ""),
"ping_ok": bool(c.get("ping_ok")),
"last": c.get("last", ""),
}
for mac, c in self.clients.items()
if c.get("ip_int")
]
write_leases(self.cfg.server_ip, leases, now_s())
# --- ping loop, runs in its own thread ---
def ping_loop(self, stop: threading.Event) -> None:
from concurrent.futures import ThreadPoolExecutor
@@ -166,6 +184,8 @@ class DhcpServer:
changed = True
if changed:
self.on_change()
# Refresh the export every tick so its mtime reflects liveness.
self.export_leases()
# --- main server loop ---
def run(self, stop: threading.Event) -> None:
+2
View File
@@ -23,6 +23,7 @@ STRINGS: dict[str, dict[str, str]] = {
"update_available": "Update available ({tag})",
"available_adapters": "Available adapters",
"no_adapters": "No suitable wired adapters found.",
"only_adapter": "Only one adapter, auto-selected: {name}",
"select_adapter": "Select adapter number",
"invalid_selection": "Invalid selection.",
"press_enter": "Press Enter to exit",
@@ -60,6 +61,7 @@ STRINGS: dict[str, dict[str, str]] = {
"update_available": "Доступно обновление ({tag})",
"available_adapters": "Доступные адаптеры",
"no_adapters": "Подходящие проводные адаптеры не найдены.",
"only_adapter": "Адаптер один, выбран автоматически: {name}",
"select_adapter": "Введите номер адаптера",
"invalid_selection": "Неверный выбор.",
"press_enter": "Нажмите Enter для выхода",
+51
View File
@@ -0,0 +1,51 @@
"""
Lease-table export — the shared contract with sibling tools (e.g. vrcx).
dhcpsrv owns the DHCP state; other tools must NOT reach into it. Instead we
periodically dump a small, self-describing JSON snapshot to a fixed,
tool-neutral location. Consumers read that file (and judge its freshness by
the file mtime). No shared code, no network call between the tools.
Default location: %PROGRAMDATA%\\vegman\\leases.json (override via the
VEGMAN_LEASES env var). The directory is created on first write.
"""
from __future__ import annotations
import json
import os
import tempfile
from pathlib import Path
SCHEMA = 1
def leases_path() -> Path:
override = os.environ.get("VEGMAN_LEASES")
if override:
return Path(override)
base = os.environ.get("PROGRAMDATA") or os.environ.get("LOCALAPPDATA") or "."
return Path(base) / "vegman" / "leases.json"
def write_leases(server_ip: str, leases: list[dict], updated: str) -> None:
"""Atomically write the snapshot. Best-effort: swallow I/O errors so a
read-only/locked target can never disturb the DHCP service."""
path = leases_path()
payload = {
"schema": SCHEMA,
"server_ip": server_ip,
"updated": updated,
"leases": leases,
}
try:
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".leases-", suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False)
os.replace(tmp, path)
finally:
if os.path.exists(tmp):
os.remove(tmp)
except OSError:
pass