Add interactive wizard mode

When run with no arguments the script now guides the user through the
full migration interactively:
  1. Lists debug archives found in the current directory and prompts
     for selection (auto-selects when only one is present)
  2. Prompts for destination host, port, and API key (key input hidden)
  3. Prompts for migration scope (SMB shares / NFS shares / SMB config)
  4. Runs a dry run and displays the summary
  5. Asks for confirmation before applying changes live

The archive is parsed once and reused for both the dry and live runs.
The existing CLI flag interface is unchanged.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-03 21:37:11 -05:00
parent 1e2a972d33
commit ab1781dbe2

View File

@@ -51,6 +51,7 @@ import argparse
import asyncio
import base64
import contextlib
import getpass
import hashlib
import json
import logging
@@ -920,8 +921,12 @@ async def migrate_smb_config(
# CLI
# ─────────────────────────────────────────────────────────────────────────────
async def run(args: argparse.Namespace) -> None:
archive = parse_archive(args.debug_tar)
async def run(
args: argparse.Namespace,
archive: Optional[dict] = None,
) -> Summary:
if archive is None:
archive = parse_archive(args.debug_tar)
migrate_set = set(args.migrate)
if args.dry_run:
@@ -950,12 +955,144 @@ async def run(args: argparse.Namespace) -> None:
await migrate_smb_config(
client, archive["smb_config"], args.dry_run, summary)
print(summary.report())
if summary.errors:
return summary
# ─────────────────────────────────────────────────────────────────────────────
# Interactive wizard
# ─────────────────────────────────────────────────────────────────────────────
def _find_debug_archives(directory: str = ".") -> list[Path]:
"""Return sorted list of TrueNAS debug archives found in *directory*."""
patterns = ("*.tgz", "*.tar.gz", "*.tar", "*.txz", "*.tar.xz")
found: set[Path] = set()
for pat in patterns:
found.update(Path(directory).glob(pat))
return sorted(found)
def _prompt(label: str, default: str = "") -> str:
suffix = f" [{default}]" if default else ""
try:
val = input(f"{label}{suffix}: ").strip()
return val if val else default
except (EOFError, KeyboardInterrupt):
print()
sys.exit(0)
def _confirm(label: str) -> bool:
try:
return input(f"{label} [y/N]: ").strip().lower() in ("y", "yes")
except (EOFError, KeyboardInterrupt):
print()
return False
def interactive_mode() -> None:
"""Interactive wizard: pick archive → configure → dry run → confirm → apply."""
print("\n=== TrueNAS Share Migration Tool ===\n")
# 1 ── Locate debug archive ────────────────────────────────────────────────
archives = _find_debug_archives()
if not archives:
sys.exit(
"No debug archives (.tgz / .tar.gz / .tar / .txz) found in the "
"current directory.\n"
"Copy your TrueNAS debug file here, or use --debug-tar to specify a path."
)
if len(archives) == 1:
chosen = archives[0]
print(f"Archive: {chosen.name} ({chosen.stat().st_size / 1_048_576:.1f} MB)\n")
else:
print("Debug archives found:\n")
for i, p in enumerate(archives, 1):
print(f" {i}. {p.name} ({p.stat().st_size / 1_048_576:.1f} MB)")
print()
while True:
raw = _prompt(f"Select archive [1-{len(archives)}]")
if raw.isdigit() and 1 <= int(raw) <= len(archives):
chosen = archives[int(raw) - 1]
break
print(f" Enter a number from 1 to {len(archives)}.")
# 2 ── Destination ─────────────────────────────────────────────────────────
print()
host = ""
while not host:
host = _prompt("Destination TrueNAS host or IP")
if not host:
print(" Host is required.")
port_raw = _prompt("WebSocket port", default="443")
port = int(port_raw) if port_raw.isdigit() else 443
# 3 ── API key ─────────────────────────────────────────────────────────────
api_key = ""
while not api_key:
try:
api_key = getpass.getpass("API key (input hidden): ").strip()
except (EOFError, KeyboardInterrupt):
print()
sys.exit(0)
if not api_key:
print(" API key is required.")
# 4 ── Migration scope ─────────────────────────────────────────────────────
print("\nWhat to migrate?")
print(" 1. SMB shares")
print(" 2. NFS shares")
print(" 3. SMB global config")
sel_raw = _prompt(
"Selection (space-separated numbers, Enter for all)", default="1 2 3"
)
_sel_map = {"1": "smb", "2": "nfs", "3": "smb-config"}
migrate: list[str] = []
for tok in sel_raw.split():
if tok in _sel_map and _sel_map[tok] not in migrate:
migrate.append(_sel_map[tok])
if not migrate:
migrate = ["smb", "nfs", "smb-config"]
# 5 ── Parse archive once (reused for dry + live runs) ────────────────────
print()
archive_data = parse_archive(str(chosen))
base_ns = dict(
debug_tar=str(chosen),
dest=host,
port=port,
api_key=api_key,
verify_ssl=False,
migrate=migrate,
)
# 6 ── Dry run ─────────────────────────────────────────────────────────────
dry_summary = asyncio.run(
run(argparse.Namespace(**base_ns, dry_run=True), archive_data)
)
print(dry_summary.report())
if not _confirm(f"Apply these changes to {host}?"):
print("Aborted no changes made.")
sys.exit(0)
# 7 ── Live run ────────────────────────────────────────────────────────────
print()
live_summary = asyncio.run(
run(argparse.Namespace(**base_ns, dry_run=False), archive_data)
)
print(live_summary.report())
if live_summary.errors:
sys.exit(2)
def main() -> None:
if len(sys.argv) == 1:
interactive_mode()
return
p = argparse.ArgumentParser(
prog="truenas_migrate.py",
description=(
@@ -1042,7 +1179,10 @@ def main() -> None:
if not args.api_key:
p.error("--api-key is required.")
asyncio.run(run(args))
summary = asyncio.run(run(args))
print(summary.report())
if summary.errors:
sys.exit(2)
if __name__ == "__main__":