Add virtualization inventory report

This commit is contained in:
dido committed 2026-09-29 12:37:57 +03:00
1 parent 43518d51cc
commit ace93c721f
3 files changed
+713

No files matched your search

+7
View File
@@ -23,6 +23,8 @@ strixscripts/
├── System/
│ ├── network-diagnostic.sh подробна мрежова диагностика, топ 10 Wi-Fi мрежи и export
│ └── system-report.sh моментен системен отчет с TXT/JSON/HTML export
├── Virtualisation/
│ └── virtualisation-check.sh проверка на инсталирани виртуализационни решения и инстанции
└── <категория>/ функционална група скриптове (предстои)
```
@@ -60,3 +62,8 @@ strixscripts/
- [`system-report.sh`](./System/system-report.md) — събира моментно състояние
на системата, като групира Ethernet, Wi-Fi и виртуалните интерфейси, и
предлага отчет във формат TXT, JSON или HTML.
### Virtualisation/
- [`virtualisation-check.sh`](./Virtualisation/virtualisation-check.md) — открива
VM/container решения и отчита инсталираните и работещите им инстанции.
+73
View File
@@ -0,0 +1,73 @@
# virtualisation-check.sh
Проверява кои виртуализационни и container решения са инсталирани и отчита
инстанциите им, състоянието им и наличните локални container images. Скриптът
е само за проверка: не стартира, спира или променя виртуални машини и контейнери.
## Какво проверява
За виртуални машини се откриват Libvirt (`virsh`), Oracle VirtualBox
(`VBoxManage`), VMware Workstation (`vmrun`), Multipass, Podman machines,
Proxmox VE (`qm`) и активни QEMU/KVM процеси. За контейнерни
решения се проверяват Docker, Podman, класически LXC (`lxc-ls`), LXD (`lxc`),
Incus, containerd (`ctr`), nerdctl и CRI инструменти (`crictl`). Ако е наличен
`kubectl`, се отчита и състоянието на pod-овете в текущия Kubernetes context;
той може да сочи към отдалечен cluster.
За всеки инструмент първо се проверява дали програмата е инсталирана. При
налична програма скриптът заявява списъка и състоянието на инстанциите.
Инсталиран инструмент, чийто daemon/socket/cluster не може да бъде достигнат,
се отчита като „Installed; query failed“ с причината, вместо да се приема,
че няма инстанции. Docker/Podman images се показват отделно от контейнерите.
## Изисквания
- Linux, Bash и Python 3
- `cat` и `python3` са задължителни за изпълнението на скрипта
- Всички VM/container CLI инструменти са незадължителни; липсващите се
отбелязват като „Not installed“
- За някои system-level инстанции може да са нужни допълнителни права; скриптът
не изисква root и показва грешка за съответния provider, ако няма достъп
## Употреба
```bash
./Virtualisation/virtualisation-check.sh
./Virtualisation/virtualisation-check.sh --help
```
| Опция | Описание |
|-------|----------|
| `-o`, `--output PATH` | Записва директно във файл; без тази опция предлага интерактивен запис |
| `-f`, `--format FORMAT` | Формат `txt`, `json` или `html` |
| `--force` | Позволява презапис на съществуващ файл без потвърждение |
| `-h`, `--help` | Показва помощта и приключва |
Без зададен `--format` форматът се разпознава от разширението на `--output`;
ако няма поддържано разширение, се използва TXT. При интерактивен запис
началният път е `~/virtualisation-report.<format>`, като TXT е форматът по
подразбиране. В non-interactive режим без `--output` отчетът се извежда към
stdout; `--format` определя сериализацията.
## Примери
```bash
# Интерактивна проверка и предложение за TXT отчет
./Virtualisation/virtualisation-check.sh
# JSON отчет във файл
./Virtualisation/virtualisation-check.sh --output "$HOME/virtualisation.json" --format json
# HTML отчет, разрешаващ презапис
./Virtualisation/virtualisation-check.sh --output /tmp/virtualisation.html --force
```
## Ограничения
Наличието на CLI binary означава, че решението е инсталирано, но не гарантира,
че daemon-ът му работи или че текущият потребител има права за достъп. Docker,
Podman, LXC/LXD/Incus и containerd показват само тези инстанции, които техният
локален CLI може да прочете. `kubectl` отчита cluster-ът от текущия context,
който може да е отдалечен. QEMU/KVM процесите се откриват през `/proc`; спрени
QEMU машини, които не се управляват от Libvirt/Proxmox/друг frontend, не могат
да бъдат изброени само по наличието на QEMU binary.
+633
View File
@@ -0,0 +1,633 @@
#!/usr/bin/env bash
#
# virtualisation-check.sh — discover virtualization tools and inspect their instances.
# Full documentation: see virtualisation-check.md in this same folder.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=lib/common.sh
source "${SCRIPT_DIR}/../lib/common.sh"
SCRIPT_HELP=$(cat <<'HELP'
Usage: ./Virtualisation/virtualisation-check.sh [OPTIONS]
Detects installed virtual-machine and container solutions, then reports their
configured/running instances without starting or changing any of them. Reports
can be displayed or saved as TXT, JSON, or HTML.
Options:
-o, --output PATH Save directly to PATH; otherwise ask interactively
-f, --format FORMAT Report format: txt, json, or html
--force Overwrite an existing output file without prompting
-h, --help Show this help message and exit
Examples:
./Virtualisation/virtualisation-check.sh
./Virtualisation/virtualisation-check.sh --output "$HOME/virtualisation.json" --format json
./Virtualisation/virtualisation-check.sh --output /tmp/virtualisation.html --force
HELP
)
common::show_help_if_requested "$@"
OUTPUT_PATH=""
OUTPUT_FORMAT=""
FORCE=0
while [[ $# -gt 0 ]]; do
case "$1" in
-o|--output)
[[ $# -ge 2 && -n "$2" ]] || common::die "$1 requires a path"
OUTPUT_PATH="$2"
shift 2
;;
-f|--format)
[[ $# -ge 2 ]] || common::die "$1 requires txt, json, or html"
OUTPUT_FORMAT="${2,,}"
case "$OUTPUT_FORMAT" in
txt|json|html) ;;
*) common::die "unsupported report format: $2 (use txt, json, or html)" ;;
esac
shift 2
;;
--force)
FORCE=1
shift
;;
*)
common::die "unknown argument: $1"
;;
esac
done
common::require_cmd python3
common::require_cmd cat
python3 -c "$(cat <<'PYCODE'
import datetime
import html
import json
import os
import pathlib
import re
import shutil
import subprocess
import sys
import tempfile
def execute(executable, arguments, timeout=20):
path = shutil.which(executable)
if path is None:
return None, f"{executable} is not installed"
try:
result = subprocess.run(
[path, *arguments],
check=False,
capture_output=True,
text=True,
errors="replace",
timeout=timeout,
)
except subprocess.TimeoutExpired:
return None, f"{executable} timed out after {timeout} seconds"
except OSError as error:
return None, str(error)
if result.returncode != 0:
detail = result.stderr.strip() or result.stdout.strip() or f"exit status {result.returncode}"
return None, detail
return result.stdout.strip(), None
def parse_json_lines(output):
values = []
for line in (output or "").splitlines():
try:
values.append(json.loads(line))
except json.JSONDecodeError:
continue
return values
def solution_record(name, category, executable, collector, note=None):
path = shutil.which(executable)
record = {
"name": name,
"category": category,
"executable": executable,
"installed": path is not None,
"path": path,
"status": "Not installed" if path is None else "Installed",
"note": note,
"instances": [],
"images": [],
}
if path is None:
return record
try:
instances, images = collector(executable)
record["instances"] = instances
record["images"] = images or []
record["status"] = "Available"
except RuntimeError as error:
record["status"] = "Installed; query failed"
record["note"] = str(error)
return record
def collect_virsh(executable):
output, error = execute(executable, ["list", "--all", "--name"])
if error:
raise RuntimeError(error)
instances = []
for name in (line.strip() for line in output.splitlines()):
if not name:
continue
state, state_error = execute(executable, ["domstate", name])
instances.append({
"name": name,
"state": state or "Unknown",
"running": bool(state and state.strip().lower() == "running"),
"details": state_error,
})
return instances, []
def collect_virtualbox(executable):
output, error = execute(executable, ["list", "vms"])
if error:
raise RuntimeError(error)
running_output, running_error = execute(executable, ["list", "runningvms"])
running_ids = set()
if running_output:
running_ids = set(re.findall(r"\{([0-9a-fA-F-]+)\}", running_output))
instances = []
for line in output.splitlines():
match = re.match(r'^"(.*)"\s+\{([0-9a-fA-F-]+)\}', line)
if match:
instances.append({
"name": match.group(1),
"id": match.group(2),
"state": "Running" if match.group(2) in running_ids else "Powered off",
"running": match.group(2) in running_ids,
})
if running_error:
raise RuntimeError(running_error)
return instances, []
def collect_vmrun(executable):
output, error = execute(executable, ["list"])
if error:
raise RuntimeError(error)
instances = []
for path in output.splitlines()[1:]:
path = path.strip()
if path:
instances.append({
"name": pathlib.Path(path).parent.name or pathlib.Path(path).stem,
"state": "Running",
"running": True,
"details": path,
})
return instances, []
def collect_multipass(executable):
output, error = execute(executable, ["list", "--format", "json"])
if error:
raise RuntimeError(error)
try:
items = json.loads(output or "{}").get("list", [])
except json.JSONDecodeError as error:
raise RuntimeError(f"invalid JSON from multipass: {error}") from error
instances = []
for item in items:
state = item.get("state") or "Unknown"
instances.append({
"name": item.get("name") or "Unknown",
"state": state,
"running": state.lower() == "running",
"details": ", ".join(item.get("ipv4") or []),
})
return instances, []
def collect_podman_machines(executable):
output, error = execute(executable, ["machine", "list", "--format", "json"])
if error:
raise RuntimeError(error)
try:
items = json.loads(output or "[]")
except json.JSONDecodeError as error:
raise RuntimeError(f"invalid JSON from podman machine: {error}") from error
instances = []
for item in items:
state = str(item.get("Running", item.get("State", "Unknown")))
running = state.lower() in ("true", "running")
instances.append({
"name": item.get("Name") or item.get("name") or "Unknown",
"state": "Running" if running else ("Stopped" if state.lower() in ("false", "stopped") else state),
"running": running,
"details": item.get("VMType") or item.get("vmtype"),
})
return instances, []
def collect_proxmox(executable):
output, error = execute(executable, ["list"])
if error:
raise RuntimeError(error)
instances = []
for line in output.splitlines():
match = re.match(r"^\s*(\d+)\s+(.+?)\s+(running|stopped)\s+", line, re.IGNORECASE)
if match:
state = match.group(3).lower()
instances.append({"id": match.group(1), "name": match.group(2), "state": state, "running": state == "running"})
return instances, []
def collect_qemu_processes():
executables = sorted(set(
[pathlib.Path(path).name for pattern in ("/usr/bin/qemu-system-*", "/usr/local/bin/qemu-system-*") for path in pathlib.Path("/").glob(pattern.lstrip("/"))]
+ [name for name in ("qemu-system-x86_64", "qemu-system-aarch64", "qemu-kvm") if shutil.which(name)]
))
processes = []
proc = pathlib.Path("/proc")
for entry in proc.iterdir():
if not entry.name.isdigit():
continue
try:
arguments = (entry / "cmdline").read_bytes().decode(errors="replace").split("\0")
except OSError:
continue
command = pathlib.Path(arguments[0]).name if arguments and arguments[0] else ""
if not command.startswith(("qemu-system-", "qemu-kvm")):
continue
name = command
if "-name" in arguments:
position = arguments.index("-name") + 1
if position < len(arguments):
name = arguments[position].split(",", 1)[0].removeprefix("guest=") or command
processes.append({"name": name, "id": entry.name, "state": "Running", "running": True, "executable": command})
return executables, processes
def collect_container_cli(executable):
output, error = execute(executable, ["ps", "--all", "--format", "{{json .}}"])
if error:
raise RuntimeError(error)
instances = []
for item in parse_json_lines(output):
state = str(item.get("State") or item.get("Status") or "Unknown")
instances.append({
"id": item.get("ID") or item.get("Id"),
"name": item.get("Names") or item.get("Name") or "Unknown",
"image": item.get("Image") or "Unknown",
"state": state,
"running": state.lower().startswith("running"),
"details": item.get("Status"),
})
image_output, _ = execute(executable, ["image", "ls", "--format", "{{json .}}"])
images = []
for item in parse_json_lines(image_output):
images.append({
"name": item.get("Repository", "<none>"),
"tag": item.get("Tag", "<none>"),
"id": item.get("ID"),
"size": item.get("Size"),
})
return instances, images
def collect_lxc_classic(executable):
output, error = execute(executable, ["--fancy"])
if error:
output, error = execute(executable, [])
if error:
raise RuntimeError(error)
lines = output.splitlines()
instances = []
header = lines[0].split() if lines and "NAME" in lines[0].upper() else []
if header:
for line in lines[1:]:
fields = line.split()
if not fields:
continue
values = dict(zip(header, fields))
name = values.get("NAME") or fields[0]
state = values.get("STATE") or "Unknown"
instances.append({"name": name, "state": state, "running": state.lower() == "running"})
else:
instances = [{"name": name, "state": "Unknown", "running": False} for name in output.split()]
return instances, []
def collect_lxd_style(executable):
output, error = execute(executable, ["list", "--format=json"])
if error:
raise RuntimeError(error)
try:
items = json.loads(output or "[]")
except json.JSONDecodeError as error:
raise RuntimeError(f"invalid JSON from {executable}: {error}") from error
instances = []
for item in items:
state = item.get("status") or "Unknown"
instances.append({
"name": item.get("name") or "Unknown",
"id": item.get("location"),
"state": state,
"running": state.lower() == "running",
"details": item.get("type"),
})
return instances, []
def collect_ctr(executable):
output, error = execute(executable, ["namespaces", "list"])
if error:
raise RuntimeError(error)
namespaces = []
for line in output.splitlines()[1:]:
fields = line.split()
if fields and fields[0] not in ("NAME", "NAME:"):
namespaces.append(fields[0])
instances = []
for namespace in namespaces:
containers, container_error = execute(executable, ["--namespace", namespace, "containers", "list"])
tasks, _ = execute(executable, ["--namespace", namespace, "tasks", "list"])
active = {}
task_lines = (tasks or "").splitlines()
if task_lines:
headers = task_lines[0].split()
for line in task_lines[1:]:
values = line.split()
row = dict(zip(headers, values))
if row.get("TASK"):
active[row["TASK"]] = row.get("STATUS", "Unknown")
if container_error:
continue
for line in (containers or "").splitlines()[1:]:
fields = line.split()
if fields:
state = active.get(fields[0], "Created")
instances.append({"name": fields[0], "namespace": namespace, "image": fields[1] if len(fields) > 1 else None, "state": state, "running": state.upper() == "RUNNING"})
return instances, []
def collect_crictl(executable):
output, error = execute(executable, ["ps", "--all", "--output", "json"])
if error:
raise RuntimeError(error)
try:
items = json.loads(output or "{}").get("containers", [])
except json.JSONDecodeError as error:
raise RuntimeError(f"invalid JSON from crictl: {error}") from error
instances = []
for item in items:
state = str(item.get("state", "Unknown"))
instances.append({
"id": item.get("id"),
"name": (item.get("metadata") or {}).get("name") or "Unknown",
"image": (item.get("image") or {}).get("image"),
"state": state,
"running": state.upper() == "CONTAINER_RUNNING",
"details": item.get("podSandboxId"),
})
return instances, []
def collect_kubernetes(executable):
output, error = execute(executable, ["get", "pods", "--all-namespaces", "--output", "json"])
if error:
raise RuntimeError(error)
try:
pods = json.loads(output or "{}").get("items", [])
except json.JSONDecodeError as error:
raise RuntimeError(f"invalid JSON from kubectl: {error}") from error
instances = []
for pod in pods:
metadata = pod.get("metadata") or {}
state = (pod.get("status") or {}).get("phase") or "Unknown"
instances.append({
"name": metadata.get("name") or "Unknown",
"namespace": metadata.get("namespace") or "default",
"state": state,
"running": state.lower() == "running",
"details": ", ".join(container.get("image", "") for container in (pod.get("spec") or {}).get("containers", [])),
})
return instances, []
def collect_report():
providers = [
solution_record("Libvirt", "Virtual machines", "virsh", collect_virsh),
solution_record("Oracle VirtualBox", "Virtual machines", "VBoxManage", collect_virtualbox),
solution_record("VMware Workstation", "Virtual machines", "vmrun", collect_vmrun,
"vmrun lists running VMs only; stopped VMs are not enumerated."),
solution_record("Multipass", "Virtual machines", "multipass", collect_multipass),
solution_record("Podman machines", "Virtual machines", "podman", collect_podman_machines,
"Queries Podman-managed VMs separately from Podman containers."),
solution_record("Proxmox VE", "Virtual machines", "qm", collect_proxmox),
solution_record("Docker", "Containers", "docker", collect_container_cli),
solution_record("Podman", "Containers", "podman", collect_container_cli),
solution_record("LXC", "Containers", "lxc-ls", collect_lxc_classic),
solution_record("LXD", "Containers", "lxc", collect_lxd_style),
solution_record("Incus", "Containers", "incus", collect_lxd_style),
solution_record("containerd (ctr)", "Containers", "ctr", collect_ctr),
solution_record("nerdctl", "Containers", "nerdctl", collect_container_cli),
solution_record("CRI-O / CRI runtime (crictl)", "Containers", "crictl", collect_crictl),
solution_record("Kubernetes workloads", "Orchestration", "kubectl", collect_kubernetes,
"Queries the current kubectl context; this may be a remote cluster."),
]
qemu_binaries, qemu_instances = collect_qemu_processes()
providers.append({
"name": "QEMU/KVM processes",
"category": "Virtual machines",
"executable": "qemu-system-*",
"installed": bool(qemu_binaries),
"path": ", ".join(qemu_binaries) if qemu_binaries else None,
"status": "Available" if qemu_binaries else "Not installed",
"note": "Detected running QEMU processes from /proc; stopped VMs are managed by their frontend.",
"instances": qemu_instances,
"images": [],
})
return {
"generated_at": datetime.datetime.now().astimezone().isoformat(timespec="seconds"),
"providers": providers,
"summary": {
"installed_solutions": sum(1 for item in providers if item["installed"]),
"running_instances": sum(1 for item in providers for instance in item["instances"] if instance.get("running")),
"known_instances": sum(len(item["instances"]) for item in providers),
},
}
def render_text(report):
lines = [
"VIRTUALISATION STATUS REPORT",
"============================",
f"Generated: {report['generated_at']}",
f"Installed solutions: {report['summary']['installed_solutions']}",
f"Known instances: {report['summary']['known_instances']}",
f"Running instances: {report['summary']['running_instances']}",
]
for category in ("Virtual machines", "Containers", "Orchestration"):
providers = [item for item in report["providers"] if item["category"] == category]
lines.extend(["", category.upper(), "-" * len(category)])
for provider in providers:
state = "installed" if provider["installed"] else "not installed"
lines.append(f"{provider['name']}: {state} [{provider['status']}]")
if provider["path"]:
lines.append(f" Executable: {provider['path']}")
if provider["note"]:
lines.append(f" Note: {provider['note']}")
if provider["instances"]:
for instance in provider["instances"]:
label = instance.get("name") or instance.get("id") or "Unknown instance"
details = [value for value in (instance.get("namespace"), instance.get("image"), instance.get("details")) if value]
suffix = f" ({'; '.join(str(value) for value in details)})" if details else ""
lines.append(f" - {label}: {instance['state']}{suffix}")
elif provider["installed"] and provider["status"] == "Available":
lines.append(" No instances found.")
elif not provider["installed"]:
lines.append(" Skipped because the tool is not installed.")
if provider["images"]:
lines.append(f" Local images: {len(provider['images'])}")
for image in provider["images"]:
lines.append(f" - {image.get('name')}:{image.get('tag')} [{image.get('id')}; {image.get('size')}]")
return "\n".join(lines) + "\n"
def render_html(report):
def esc(value):
return html.escape("Unavailable" if value is None or value == "" else str(value))
sections = []
for category in ("Virtual machines", "Containers", "Orchestration"):
providers = [item for item in report["providers"] if item["category"] == category]
cards = []
for provider in providers:
status_class = "available" if provider["installed"] and provider["status"] == "Available" else "missing" if not provider["installed"] else "warning"
rows = []
for instance in provider["instances"]:
state_class = "running" if instance.get("running") else "stopped"
details = " · ".join(str(value) for value in (instance.get("namespace"), instance.get("image"), instance.get("details")) if value)
rows.append(f"<tr><td>{esc(instance.get('name') or instance.get('id'))}</td><td><span class=\"{state_class}\">{esc(instance['state'])}</span></td><td>{esc(details)}</td></tr>")
if rows:
table = "<table><thead><tr><th>Instance</th><th>State</th><th>Details</th></tr></thead><tbody>" + "".join(rows) + "</tbody></table>"
elif provider["installed"] and provider["status"] == "Available":
table = "<p class=\"muted\">No instances found.</p>"
elif not provider["installed"]:
table = "<p class=\"muted\">Not installed.</p>"
else:
table = f"<p class=\"warning-text\">{esc(provider['note'] or provider['status'])}</p>"
note = f"<p class=\"muted\">{esc(provider['note'])}</p>" if provider["note"] and provider["status"] == "Available" else ""
image_detail = f"<p class=\"muted\">Local images: {len(provider['images'])}</p>" if provider["images"] else ""
cards.append(
f"<article class=\"provider\"><header><h3>{esc(provider['name'])}</h3><span class=\"badge {status_class}\">{esc(provider['status'])}</span></header>"
f"<p class=\"path\">{esc(provider['path'])}</p>{note}{table}{image_detail}</article>"
)
sections.append(f"<section><h2>{html.escape(category)}</h2><div class=\"providers\">{''.join(cards)}</div></section>")
summary = report["summary"]
return (
"<!doctype html><html lang=\"en\"><head><meta charset=\"utf-8\"><meta name=\"viewport\" content=\"width=device-width, initial-scale=1\"><title>Virtualisation Status</title>"
"<style>:root{--ink:#1b282d;--muted:#62747b;--line:#d8e1e3;--paper:#f2f6f5;--panel:#fff;--teal:#087e80;--green:#237544;--red:#9c3934;--amber:#8a5c18}*{box-sizing:border-box}body{margin:0;background:var(--paper);color:var(--ink);font:15px/1.5 system-ui,-apple-system,\"Segoe UI\",sans-serif}main{max-width:1180px;margin:auto;padding:32px 22px 48px}.masthead{border-bottom:2px solid var(--teal);padding-bottom:16px}h1{margin:0;font-size:28px}.muted,.path{color:var(--muted);font-size:13px}.summary{display:grid;grid-template-columns:repeat(3,minmax(0,1fr));gap:12px;margin:18px 0}.metric,section{background:var(--panel);border:1px solid var(--line);padding:17px}.metric span,.metric strong{display:block}.metric span{color:var(--muted);font-size:13px}.metric strong{font-size:23px;margin-top:4px}section{margin-top:14px}h2{font-size:19px;margin:0 0 14px}.providers{display:grid;grid-template-columns:repeat(2,minmax(0,1fr));gap:10px}.provider{border:1px solid var(--line);padding:13px;min-width:0}.provider header{display:flex;justify-content:space-between;gap:12px;align-items:center}.provider h3{margin:0;font-size:16px}.badge{font-size:12px;font-weight:650}.badge.available,.running{color:var(--green)}.badge.missing,.stopped{color:var(--muted)}.badge.warning,.warning-text{color:var(--amber)}.path{overflow-wrap:anywhere}table{width:100%;border-collapse:collapse;text-align:left;font-size:13px}th,td{padding:7px 8px;border-bottom:1px solid var(--line);vertical-align:top}th{color:var(--muted);font-size:11px;text-transform:uppercase}@media(max-width:700px){main{padding:20px 12px}.summary{grid-template-columns:1fr 1fr}.providers{grid-template-columns:1fr}}</style></head><body><main>"
f"<header class=\"masthead\"><h1>Virtualisation Status</h1><p>Generated {esc(report['generated_at'])}</p></header>"
f"<div class=\"summary\"><article class=\"metric\"><span>Installed solutions</span><strong>{summary['installed_solutions']}</strong></article><article class=\"metric\"><span>Known instances</span><strong>{summary['known_instances']}</strong></article><article class=\"metric\"><span>Running instances</span><strong>{summary['running_instances']}</strong></article></div>"
+ "".join(sections)
+ "</main></body></html>\n"
)
def ask(prompt, default):
try:
answer = input(f"{prompt} [{default}]: ").strip()
except EOFError:
return default
return answer or default
def ask_yes_no(prompt, default):
marker = "Y/n" if default else "y/N"
try:
answer = input(f"{prompt} [{marker}]: ").strip().lower()
except EOFError:
return default
if not answer:
return default
return answer in ("y", "yes")
def infer_format(path, requested):
if requested:
return requested
suffix = pathlib.Path(path).suffix.lower().lstrip(".")
return suffix if suffix in ("txt", "json", "html") else "txt"
def render_report(report, format_name):
if format_name == "json":
return json.dumps(report, ensure_ascii=False, indent=2) + "\n"
if format_name == "html":
return render_html(report)
return render_text(report)
def save_report(path_text, format_name, force, interactive, report):
path = pathlib.Path(path_text).expanduser()
if not path.parent.exists() or not path.parent.is_dir():
raise OSError(f"output directory does not exist: {path.parent}")
if path.exists() and not force:
if not interactive or not ask_yes_no(f"{path} already exists. Overwrite it?", False):
return False
rendered = render_report(report, format_name)
temporary = None
try:
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as output:
temporary = pathlib.Path(output.name)
output.write(rendered)
os.replace(temporary, path)
finally:
if temporary is not None and temporary.exists():
temporary.unlink()
print(f"Report saved: {path}", file=sys.stderr)
return True
def main():
output_path, requested_format, force_value = sys.argv[1:4]
force = force_value == "1"
interactive = sys.stdin.isatty() and sys.stdout.isatty()
report = collect_report()
text_report = render_text(report)
if output_path:
format_name = infer_format(output_path, requested_format)
if not save_report(output_path, format_name, force, interactive, report):
print("Report not written; existing file was not overwritten.", file=sys.stderr)
return 1 if not interactive else 0
return 0
if not interactive:
print(render_report(report, requested_format or "txt"), end="")
return 0
print(text_report, end="")
if not ask_yes_no("Save this report to a file?", True):
return 0
format_name = requested_format or ask("Report format (txt/json/html)", "txt").lower()
if format_name not in ("txt", "json", "html"):
print("Unsupported format; using txt.", file=sys.stderr)
format_name = "txt"
path = ask("Output path", str(pathlib.Path.home() / f"virtualisation-report.{format_name}"))
if not save_report(path, format_name, force, interactive, report):
print("Report not written; existing file was not overwritten.", file=sys.stderr)
return 0
try:
raise SystemExit(main())
except (OSError, RuntimeError, json.JSONDecodeError) as error:
print(f"Error: {error}", file=sys.stderr)
raise SystemExit(1)
PYCODE
)" "$OUTPUT_PATH" "$OUTPUT_FORMAT" "$FORCE"