TrueMigration/truenas_migrate/summary.py

"""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

    iscsi_extents_found:       int = 0
    iscsi_extents_created:     int = 0
    iscsi_extents_skipped:     int = 0
    iscsi_extents_failed:      int = 0

    iscsi_initiators_found:    int = 0
    iscsi_initiators_created:  int = 0
    iscsi_initiators_skipped:  int = 0
    iscsi_initiators_failed:   int = 0

    iscsi_portals_found:       int = 0
    iscsi_portals_created:     int = 0
    iscsi_portals_skipped:     int = 0
    iscsi_portals_failed:      int = 0

    iscsi_targets_found:       int = 0
    iscsi_targets_created:     int = 0
    iscsi_targets_skipped:     int = 0
    iscsi_targets_failed:      int = 0

    iscsi_targetextents_found:   int = 0
    iscsi_targetextents_created: int = 0
    iscsi_targetextents_skipped: int = 0
    iscsi_targetextents_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)

    @property
    def _has_iscsi(self) -> bool:
        return (self.iscsi_extents_found + self.iscsi_initiators_found +
                self.iscsi_portals_found + self.iscsi_targets_found +
                self.iscsi_targetextents_found) > 0

    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)

        def _iscsi_val(found, created, skipped, failed) -> str:
            return (
                f"{_dim('found=' + str(found))}  "
                f"{_stat('created', created, _bold_green)}  "
                f"{_stat('skipped', skipped, _yellow)}  "
                f"{_stat('failed',  failed,  _bold_red)}"
            )

        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),
        ]

        if self._has_iscsi:
            lines.append(f"{ml}{hr}{mr}")
            lines.append(row("iSCSI extents :  ", _iscsi_val(
                self.iscsi_extents_found, self.iscsi_extents_created,
                self.iscsi_extents_skipped, self.iscsi_extents_failed)))
            lines.append(row("iSCSI initiators: ", _iscsi_val(
                self.iscsi_initiators_found, self.iscsi_initiators_created,
                self.iscsi_initiators_skipped, self.iscsi_initiators_failed)))
            lines.append(row("iSCSI portals :  ", _iscsi_val(
                self.iscsi_portals_found, self.iscsi_portals_created,
                self.iscsi_portals_skipped, self.iscsi_portals_failed)))
            lines.append(row("iSCSI targets :  ", _iscsi_val(
                self.iscsi_targets_found, self.iscsi_targets_created,
                self.iscsi_targets_skipped, self.iscsi_targets_failed)))
            lines.append(row("iSCSI tgt↔ext :  ", _iscsi_val(
                self.iscsi_targetextents_found, self.iscsi_targetextents_created,
                self.iscsi_targetextents_skipped, self.iscsi_targetextents_failed)))

        lines.append(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)