scott
c157e14fa9
Split single-file script into focused modules: colors.py – ANSI helpers and shared logger summary.py – Summary dataclass and report renderer archive.py – Debug archive parser (SCALE + CORE layouts) client.py – WebSocket engine, TrueNASClient, dataset utilities migrate.py – Payload builders, migrate_smb_shares, migrate_nfs_shares cli.py – Interactive wizard, argparse, run(), main() __main__.py – python -m truenas_migrate entry point truenas_migrate.py retained as a one-line compatibility shim. Both 'python truenas_migrate.py' and 'python -m truenas_migrate' work. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
"""Migration summary dataclass and report renderer."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from .colors import (
|
|
_dim, _bold, _red, _yellow, _cyan,
|
|
_bold_red, _bold_green, _bold_yellow, _vis_len,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Summary:
|
|
smb_found: int = 0
|
|
smb_created: int = 0
|
|
smb_skipped: int = 0
|
|
smb_failed: int = 0
|
|
|
|
nfs_found: int = 0
|
|
nfs_created: int = 0
|
|
nfs_skipped: int = 0
|
|
nfs_failed: int = 0
|
|
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
# Populated during dry-run dataset safety checks
|
|
paths_to_create: list[str] = field(default_factory=list)
|
|
missing_datasets: list[str] = field(default_factory=list)
|
|
|
|
def report(self) -> str:
|
|
w = 60
|
|
|
|
def _stat(label: str, n: int, color_fn) -> str:
|
|
s = f"{label}={n}"
|
|
return color_fn(s) if n > 0 else _dim(s)
|
|
|
|
smb_val = (
|
|
f"{_dim('found=' + str(self.smb_found))} "
|
|
f"{_stat('created', self.smb_created, _bold_green)} "
|
|
f"{_stat('skipped', self.smb_skipped, _yellow)} "
|
|
f"{_stat('failed', self.smb_failed, _bold_red)}"
|
|
)
|
|
nfs_val = (
|
|
f"{_dim('found=' + str(self.nfs_found))} "
|
|
f"{_stat('created', self.nfs_created, _bold_green)} "
|
|
f"{_stat('skipped', self.nfs_skipped, _yellow)} "
|
|
f"{_stat('failed', self.nfs_failed, _bold_red)}"
|
|
)
|
|
|
|
hr = _cyan("─" * w)
|
|
tl = _cyan("┌"); tr = _cyan("┐")
|
|
ml = _cyan("├"); mr = _cyan("┤")
|
|
bl = _cyan("└"); br = _cyan("┘")
|
|
side = _cyan("│")
|
|
|
|
title_text = "MIGRATION SUMMARY"
|
|
lpad = (w - len(title_text)) // 2
|
|
rpad = w - len(title_text) - lpad
|
|
title_row = f"{side}{' ' * lpad}{_bold(title_text)}{' ' * rpad}{side}"
|
|
|
|
def row(label: str, val: str) -> str:
|
|
right = max(0, w - 2 - len(label) - _vis_len(val))
|
|
return f"{side} {_dim(label)}{val}{' ' * right} {side}"
|
|
|
|
lines = [
|
|
"",
|
|
f"{tl}{hr}{tr}",
|
|
title_row,
|
|
f"{ml}{hr}{mr}",
|
|
row("SMB shares : ", smb_val),
|
|
row("NFS shares : ", nfs_val),
|
|
f"{bl}{hr}{br}",
|
|
]
|
|
|
|
if self.errors:
|
|
lines.append(f"\n {_bold_red(str(len(self.errors)) + ' error(s):')} ")
|
|
for e in self.errors:
|
|
lines.append(f" {_red('•')} {e}")
|
|
|
|
if self.missing_datasets:
|
|
lines.append(
|
|
f"\n {_bold_yellow('WARNING:')} "
|
|
f"{len(self.missing_datasets)} share path(s) have no "
|
|
"matching dataset on the destination:"
|
|
)
|
|
for p in self.missing_datasets:
|
|
lines.append(f" {_yellow('•')} {p}")
|
|
lines.append(
|
|
" These paths must exist before shares can be created.\n"
|
|
" Use interactive mode or answer 'y' at the dataset prompt to create them."
|
|
)
|
|
lines.append("")
|
|
return "\n".join(lines)
|