Terminal: Das prompt, dprompt and dbanner (0.9.1)

- das-shell 0.1.0: the Das two-line bash prompt by default (replaces the PS1 of Void's skel .bashrc once), dprompt (Textual prompt builder with presets and live preview) and dbanner (SSH login banner editor: canvas tools, fields with values, widgets, das-banner service for cached ones, Das OS logo and tryzub pictures). Python tests in make test. - das-server 0.3.0 depends on das-shell; the server installer mentions dprompt and dbanner; the desktop image includes it. - run-qemu.sh forwards the guest's SSH to localhost:2222. - Hardware fixes: Ventoy menu icon, boot menu titles "Das OS <version>". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 09:17:10 +03:00
parent 66ea614e0d
commit a2bffc7fe9
72 changed files with 5506 additions and 23 deletions
@@ -0,0 +1,415 @@
+"""Values for banner fields ({hostname}, {ip}, ...) and widgets: blocks of lines
+from built-in functions, scripts, commands or files.
+
+"live" widgets run at login with a short timeout; "cached" ones are run by the
+das-banner service, and the login only reads their last output from CACHE_DIR.
+"""
+
+import concurrent.futures
+import datetime
+import hashlib
+import json
+import os
+import socket
+import subprocess
+import time
+
+from . import CACHE_DIR, DATA_DIR, colors
+from .i18n import t
+
+LIVE_TIMEOUT = 1.5
+CACHED_TIMEOUT = 60
+
+
+# --- field values -----------------------------------------------------------------
+
+def _read(path):
+    try:
+        with open(path, encoding="utf-8", errors="replace") as f:
+            return f.read()
+    except OSError:
+        return ""
+
+
+def _ips():
+    try:
+        out = subprocess.run(["ip", "-4", "-o", "addr", "show", "scope", "global"],
+                             capture_output=True, text=True, timeout=1).stdout
+        ips = [line.split()[3].split("/")[0] for line in out.splitlines() if len(line.split()) > 3]
+        if ips:
+            return ips
+    except (OSError, subprocess.SubprocessError):
+        pass
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
+            s.connect(("192.0.2.1", 9))  # no packet is sent
+            return [s.getsockname()[0]]
+    except OSError:
+        return []
+
+
+def _os_name():
+    for line in _read("/etc/os-release").splitlines():
+        if line.startswith("PRETTY_NAME="):
+            return line.split("=", 1)[1].strip().strip('"')
+    return "Linux"
+
+
+def _duration(seconds):
+    days, rest = divmod(int(seconds), 86400)
+    hours, rest = divmod(rest, 3600)
+    minutes = rest // 60
+    parts = []
+    if days:
+        parts.append(t(f"{days} дн", f"{days} d"))
+    if hours or days:
+        parts.append(t(f"{hours} год", f"{hours} h"))
+    if not days:
+        parts.append(t(f"{minutes} хв", f"{minutes} min"))
+    return " ".join(parts)
+
+
+def _uptime():
+    try:
+        return _duration(float(_read("/proc/uptime").split()[0]))
+    except (IndexError, ValueError):
+        return "?"
+
+
+def _meminfo():
+    info = {}
+    for line in _read("/proc/meminfo").splitlines():
+        key, _, value = line.partition(":")
+        try:
+            info[key] = int(value.split()[0]) * 1024
+        except (IndexError, ValueError):
+            pass
+    return info
+
+
+def human(n):
+    for unit in ("B", "K", "M", "G", "T", "P"):
+        if abs(n) < 1024 or unit == "P":
+            return f"{n:.0f}{unit}" if unit in ("B", "K") or n >= 100 else f"{n:.1f}{unit}"
+        n /= 1024
+    return str(n)
+
+
+def _mem():
+    m = _meminfo()
+    total = m.get("MemTotal", 0)
+    used = total - m.get("MemAvailable", 0)
+    return f"{human(used)}/{human(total)}"
+
+
+def _cpu():
+    for line in _read("/proc/cpuinfo").splitlines():
+        if line.startswith("model name"):
+            return " ".join(line.split(":", 1)[1].split())
+    return os.uname().machine
+
+
+def _users():
+    try:
+        out = subprocess.run(["who"], capture_output=True, text=True, timeout=1).stdout
+        return str(len({line.split()[0] for line in out.splitlines() if line.strip()}))
+    except (OSError, subprocess.SubprocessError):
+        return "?"
+
+
+PROVIDERS = {
+    "hostname": lambda: socket.gethostname().split(".")[0],
+    "fqdn": socket.getfqdn,
+    "ip": lambda: (_ips() or ["?"])[0],
+    "ips": lambda: " ".join(_ips()),
+    "os": _os_name,
+    "kernel": lambda: os.uname().release,
+    "arch": lambda: os.uname().machine,
+    "uptime": _uptime,
+    "load": lambda: " ".join(_read("/proc/loadavg").split()[:3]),
+    "mem": _mem,
+    "cpu": _cpu,
+    "cores": lambda: str(os.cpu_count() or 1),
+    "date": lambda: datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
+    "users": _users,
+    "user": lambda: os.environ.get("USER") or os.environ.get("LOGNAME") or "?",
+}
+
+VAR_HELP = {
+    "hostname": ("ім'я комп'ютера", "host name"),
+    "fqdn": ("повне ім'я в мережі", "fully qualified name"),
+    "ip": ("перша IPv4-адреса", "first IPv4 address"),
+    "ips": ("усі IPv4-адреси", "all IPv4 addresses"),
+    "os": ("назва системи", "system name"),
+    "kernel": ("версія ядра", "kernel version"),
+    "arch": ("архітектура", "architecture"),
+    "uptime": ("час роботи", "uptime"),
+    "load": ("навантаження", "load average"),
+    "mem": ("пам'ять зайнято/всього", "memory used/total"),
+    "cpu": ("процесор", "processor"),
+    "cores": ("кількість ядер", "number of cores"),
+    "date": ("дата й час", "date and time"),
+    "users": ("користувачів у системі", "users logged in"),
+    "user": ("хто входить", "who logs in"),
+}
+
+
+class Values(dict):
+    """Field values, computed only when a field asks for them."""
+
+    def __init__(self, user_vars=None):
+        super().__init__()
+        self.user_vars = dict(user_vars or {})
+
+    def get(self, key, default=None):
+        if key in self.user_vars:
+            return self.user_vars[key]
+        if key not in self and key in PROVIDERS:
+            try:
+                self[key] = PROVIDERS[key]()
+            except Exception:  # a broken value must not break the login
+                self[key] = "?"
+        return super().get(key, default)
+
+
+# --- built-in widgets -------------------------------------------------------------
+
+def _bar(percent, width=10):
+    full = round(percent / 100 * width)
+    color = "green" if percent < 70 else ("yellow" if percent < 90 else "red")
+    return (colors.sgr(color, None, True) + "█" * full + colors.sgr("bright_black") + "░" * (width - full)
+            + colors.RESET)
+
+
+PSEUDO_FS = {"proc", "sysfs", "devtmpfs", "devpts", "tmpfs", "cgroup", "cgroup2", "securityfs",
+             "pstore", "efivarfs", "bpf", "debugfs", "tracefs", "configfs", "fusectl", "mqueue",
+             "hugetlbfs", "autofs", "binfmt_misc", "overlay", "squashfs", "nsfs", "ramfs", "rpc_pipefs",
+             "fuse.portal", "fuse.gvfsd-fuse", "iso9660", "zram"}
+
+
+def _mounts():
+    seen, out = set(), []
+    for line in _read("/proc/mounts").splitlines():
+        parts = line.split()
+        if len(parts) < 3 or parts[2] in PSEUDO_FS or not parts[0].startswith("/"):
+            continue
+        mount = parts[1].replace("\\040", " ")
+        if parts[0] in seen or mount.startswith(("/snap", "/var/lib/docker", "/.snapshots", "/run")):
+            continue
+        seen.add(parts[0])
+        out.append(mount)
+    return out
+
+
+def w_disks(args, values):
+    lines = [colors.sgr("white", None, True) + t("Диски:", "Disks:") + colors.RESET]
+    rows = []
+    for mount in args.get("mounts") or _mounts():
+        try:
+            st = os.statvfs(mount)
+        except OSError:
+            continue
+        total = st.f_blocks * st.f_frsize
+        if not total:
+            continue
+        used = total - st.f_bfree * st.f_frsize
+        rows.append((mount, total, used))
+    width = max([len(r[0]) for r in rows] + [1])
+    for mount, total, used in rows:
+        pct = round(used * 100 / total)
+        lines.append(f" {mount.ljust(width)} {human(total):>6} {_bar(pct)} {pct:3d}%")
+    return lines
+
+
+def w_memory(args, values):
+    m = _meminfo()
+    lines = []
+    for label, total, free in ((t("Пам'ять", "Memory"), m.get("MemTotal", 0), m.get("MemAvailable", 0)),
+                               ("Swap", m.get("SwapTotal", 0), m.get("SwapFree", 0))):
+        if not total:
+            continue
+        pct = round((total - free) * 100 / total)
+        lines.append(f"{colors.sgr('white', None, True)}{label + ':':<9}{colors.RESET} "
+                     f"{human(total - free):>6}/{human(total):<6} {_bar(pct)} {pct:3d}%")
+    return lines
+
+
+def w_uptime(args, values):
+    return [f"{colors.sgr('white', None, True)}{t('Працює:', 'Uptime:')}{colors.RESET} "
+            f"{colors.sgr('red', None, True)}{values.get('uptime')}{colors.RESET}"]
+
+
+def w_load(args, values):
+    return [f"{colors.sgr('white', None, True)}{t('Навантаження:', 'Load:')}{colors.RESET} "
+            f"{values.get('load')} ({t('ядер', 'cores')}: {values.get('cores')})"]
+
+
+def w_network(args, values):
+    lines = [colors.sgr("white", None, True) + t("Мережа:", "Network:") + colors.RESET]
+    try:
+        out = subprocess.run(["ip", "-o", "addr", "show", "scope", "global"],
+                             capture_output=True, text=True, timeout=1).stdout
+    except (OSError, subprocess.SubprocessError):
+        return []
+    for line in out.splitlines():
+        p = line.split()
+        if len(p) > 3:
+            lines.append(f" {p[1]:<10} {colors.sgr('cyan', None, True)}{p[3]}{colors.RESET}")
+    return lines if len(lines) > 1 else []
+
+
+BUILTINS = {
+    "uptime": (w_uptime, ("Час роботи", "Uptime")),
+    "load": (w_load, ("Навантаження", "Load average")),
+    "memory": (w_memory, ("Пам'ять і swap", "Memory and swap")),
+    "disks": (w_disks, ("Диски", "Disks")),
+    "network": (w_network, ("Мережа", "Network")),
+}
+
+
+def widget_dir():
+    return os.path.join(DATA_DIR, "banner", "widgets")
+
+
+def script_meta(path):
+    """# dbanner-<key>: value lines of a widget script."""
+    meta = {}
+    for line in _read(path).splitlines()[:20]:
+        if line.startswith("# dbanner-"):
+            key, _, value = line[10:].partition(":")
+            meta[key.strip()] = value.strip()
+    return meta
+
+
+def available():
+    """(name, title, mode) of every built-in widget and widget script."""
+    out = [(name, t(*names), "live") for name, (_, names) in BUILTINS.items()]
+    try:
+        scripts = sorted(f for f in os.listdir(widget_dir()) if f.endswith(".sh") and f != "lib.sh")
+    except OSError:
+        scripts = []
+    for f in scripts:
+        meta = script_meta(os.path.join(widget_dir(), f))
+        out.append((f[:-3], t(meta.get("title-uk", f[:-3]), meta.get("title-en", f[:-3])),
+                    meta.get("mode", "live")))
+    return out
+
+
+def describe(w):
+    """A short line about a widget for lists."""
+    kind = w.get("type", "builtin")
+    if kind == "builtin":
+        names = {n: title_ for n, title_, _ in available()}
+        return names.get(w.get("name"), w.get("name", "?"))
+    if kind == "command":
+        return "$ " + w.get("cmd", "")
+    return t("файл ", "file ") + w.get("path", "")
+
+
+def mode_of(w):
+    if w.get("mode"):
+        return w["mode"]
+    if w.get("type", "builtin") == "builtin" and w.get("name") not in BUILTINS:
+        return script_meta(os.path.join(widget_dir(), w.get("name", "") + ".sh")).get("mode", "live")
+    return "live"
+
+
+def interval_of(w):
+    if w.get("interval"):
+        return int(w["interval"])
+    if w.get("type", "builtin") == "builtin":
+        meta = script_meta(os.path.join(widget_dir(), w.get("name", "") + ".sh"))
+        if meta.get("interval", "").isdigit():
+            return int(meta["interval"])
+    return 600
+
+
+def cache_path(w):
+    key = hashlib.sha1(json.dumps(w, sort_keys=True).encode()).hexdigest()[:16]
+    return os.path.join(CACHE_DIR, key + ".txt")
+
+
+def _env(design_vars):
+    env = dict(os.environ)
+    for key, value in (design_vars or {}).items():
+        env["DBANNER_" + key.upper()] = str(value)
+    env["DBANNER_LANG"] = "uk" if t("uk", "en") == "uk" else "en"
+    return env
+
+
+def run(w, values, design_vars=None, timeout=None):
+    """Runs a widget now and returns its lines (empty when it fails)."""
+    kind = w.get("type", "builtin")
+    try:
+        if kind == "file":
+            text = _read(os.path.expanduser(w.get("path", "")))
+            lines = text.rstrip("\n").split("\n") if text.strip() else []
+        elif kind == "builtin" and w.get("name") in BUILTINS:
+            lines = BUILTINS[w["name"]][0](w.get("args") or {}, values)
+        else:
+            if kind == "builtin":
+                cmd = [os.path.join(widget_dir(), w.get("name", "") + ".sh")]
+            else:
+                cmd = ["sh", "-c", w.get("cmd", "")]
+            limit = timeout or float(w.get("timeout") or (CACHED_TIMEOUT if mode_of(w) == "cached" else LIVE_TIMEOUT))
+            proc = subprocess.run(cmd, capture_output=True, text=True, errors="replace",
+                                  timeout=limit, env=_env(design_vars), stdin=subprocess.DEVNULL)
+            out = proc.stdout.rstrip("\n")
+            lines = out.split("\n") if out.strip() else []
+    except Exception:  # a broken widget must not break the login
+        return []
+    if lines and w.get("title"):
+        lines = [colors.sgr("white", None, True) + w["title"] + colors.RESET] + lines
+    return lines
+
+
+def read_cache(w):
+    text = _read(cache_path(w))
+    return text.rstrip("\n").split("\n") if text.strip() else []
+
+
+def blocks(design, values, use_cache=True):
+    """Lines of every widget of a design, in order; live ones run in parallel."""
+    widgets = design.get("widgets", [])
+    result = [[] for _ in widgets]
+    jobs = {}
+    # No "with": leaving it would wait for a widget that hangs
+    pool = concurrent.futures.ThreadPoolExecutor(max_workers=8)
+    for i, w in enumerate(widgets):
+        if w.get("enabled") is False:
+            continue
+        if use_cache and mode_of(w) == "cached":
+            result[i] = read_cache(w)
+        else:
+            jobs[pool.submit(run, w, values, design.get("vars"))] = i
+    try:
+        for fut in concurrent.futures.as_completed(jobs, timeout=LIVE_TIMEOUT + 1 if use_cache else None):
+            result[jobs[fut]] = fut.result()
+    except concurrent.futures.TimeoutError:
+        pass
+    pool.shutdown(wait=False, cancel_futures=True)
+    return result
+
+
+def refresh(design, force=False):
+    """Runs the cached widgets whose output is older than their interval."""
+    values = None
+    os.makedirs(CACHE_DIR, mode=0o755, exist_ok=True)
+    for w in design.get("widgets", []):
+        if w.get("enabled") is False or mode_of(w) != "cached":
+            continue
+        path = cache_path(w)
+        try:
+            age = time.time() - os.stat(path).st_mtime
+        except OSError:
+            age = None
+        if not force and age is not None and age < interval_of(w):
+            continue
+        if values is None:
+            values = Values(design.get("vars"))
+        lines = run(w, values, design.get("vars"))
+        tmp = path + ".tmp"
+        with open(tmp, "w", encoding="utf-8") as f:
+            f.write("\n".join(lines) + ("\n" if lines else ""))
+        os.chmod(tmp, 0o644)
+        os.replace(tmp, path)