menu formatting update
This commit is contained in:
David Schroeder
+64
-13
@@ -5,6 +5,9 @@ import csv
|
||||
from datetime import datetime
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import shutil
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
@@ -119,7 +122,7 @@ def scan(root=Path('/etc/pve/nodes'), get=api):
|
||||
if node.get('status') != 'online':
|
||||
warnings.append(f'{name}: offline; storage not scanned')
|
||||
continue
|
||||
print(f'Scanning {name} storage...', file=sys.stderr, flush=True)
|
||||
ReportDisplay(sys.stderr).line(f'Scanning {name} storage...', 'muted')
|
||||
try:
|
||||
storages = get(f'/nodes/{name}/storage')
|
||||
except (RuntimeError, ValueError, OSError, subprocess.TimeoutExpired) as exc:
|
||||
@@ -176,26 +179,68 @@ def format_size(size_bytes, approximate=False):
|
||||
size /= 1024
|
||||
|
||||
|
||||
class ReportDisplay:
|
||||
"""Keep redirected reports plain and wrap text before applying ANSI styling."""
|
||||
def __init__(self, stream=None):
|
||||
self.stream = stream if stream is not None else sys.stdout
|
||||
self.color = (self.stream.isatty() and os.environ.get('TERM', 'dumb') != 'dumb'
|
||||
and 'NO_COLOR' not in os.environ)
|
||||
self.width = max(20, min(100, shutil.get_terminal_size((80, 24)).columns))
|
||||
|
||||
def line(self, text='', style='', indent=0):
|
||||
codes = {'bold': '1', 'muted': '90', 'accent': '96',
|
||||
'success': '32', 'warning': '33', 'error': '31'}
|
||||
prefix = '\033[' + codes[style] + 'm' if self.color and style else ''
|
||||
reset = '\033[0m' if prefix else ''
|
||||
for line in textwrap.wrap(str(text), width=self.width - indent,
|
||||
replace_whitespace=False) or ['']:
|
||||
print(' ' * indent + prefix + line + reset, file=self.stream, flush=True)
|
||||
|
||||
|
||||
def render(rows, warnings, coverage):
|
||||
print('\nCluster orphaned / unused disk report (read-only)')
|
||||
print('Coverage: ' + ('INCOMPLETE / NEEDS REVIEW' if warnings else 'all discovered disk storage scanned'))
|
||||
print('Candidates are not proof of safe deletion. Storage-managed volumes only; excludes backups, ISOs and passthrough disks.')
|
||||
for entry in coverage:
|
||||
print(' ' + entry)
|
||||
display = ReportDisplay()
|
||||
display.line()
|
||||
display.line('Scan finished — incomplete coverage' if warnings else 'Scan finished',
|
||||
'warning' if warnings else 'success')
|
||||
display.line(f'{len(coverage)} storage location(s) scanned.', 'muted')
|
||||
for warning in warnings:
|
||||
if warning.endswith(': inactive or disabled; not scanned'):
|
||||
continue
|
||||
print('WARNING: ' + warning)
|
||||
display.line('WARNING: ' + warning, 'warning')
|
||||
display.line()
|
||||
display.line('Results', 'bold')
|
||||
counts = {status: sum(row['status'] == status for row in rows)
|
||||
for status in ('Unused disk', 'Orphan candidate', 'Needs review')}
|
||||
display.line(f"{counts['Unused disk']} unused disk(s) "
|
||||
f"{counts['Orphan candidate']} orphan candidate(s) "
|
||||
f"{counts['Needs review']} needing review", indent=2)
|
||||
group = None
|
||||
for row in sorted(rows, key=lambda r: (r['node'], r['storage'], r['volume'])):
|
||||
if group != (row['node'], row['storage']):
|
||||
group = (row['node'], row['storage'])
|
||||
print(f'\n{group[0]} / {group[1]}')
|
||||
display.line()
|
||||
display.line(row['storage'], 'bold')
|
||||
node = row['node']
|
||||
context = ('Shared storage · scanned via ' + node[len('shared (via '):-1]
|
||||
if node.startswith('shared (via ') else 'Node: ' + node)
|
||||
display.line(context, 'muted')
|
||||
display.line()
|
||||
if row['status'] == 'Unused disk':
|
||||
display.line(row['detail'] + ' [Unused disk]', 'bold', indent=2)
|
||||
else:
|
||||
display.line('[' + row['status'] + ']', 'warning', indent=2)
|
||||
display.line(row['detail'], 'muted', indent=2)
|
||||
display.line('Volume: ' + row['volume'], indent=2)
|
||||
size = format_size(row['size_bytes'], row.get('size_is_approximate', False))
|
||||
print(f" {row['volume']} Disk size: {size} {row['status']}\n {row['detail']}")
|
||||
print(f'\n{len(rows)} volume(s) reported.')
|
||||
display.line('Disk size: ' + size, indent=2)
|
||||
display.line()
|
||||
display.line(f'{len(rows)} volume(s) reported.', 'bold')
|
||||
if not rows:
|
||||
print('No unreferenced or unused volumes found in the successfully scanned storage.')
|
||||
display.line('No unreferenced or unused volumes found in the successfully scanned storage.')
|
||||
if any(row.get('size_is_approximate') for row in rows):
|
||||
display.line('~ indicates an approximate size.', 'muted')
|
||||
display.line('Review candidates before deletion. No disks were changed.', 'muted')
|
||||
display.line('Storage-managed volumes only; excludes backups, ISOs and passthrough disks.', 'muted')
|
||||
|
||||
|
||||
def export_csv(path, rows):
|
||||
@@ -240,10 +285,16 @@ def main():
|
||||
render(rows, warnings, coverage)
|
||||
destination = args.csv
|
||||
if args.interactive and sys.stdin.isatty() and not destination:
|
||||
destination = input('\nCSV export file or folder (Enter to skip): ').strip()
|
||||
display = ReportDisplay()
|
||||
display.line()
|
||||
display.line('CSV export file or folder:', 'accent')
|
||||
display.line('Enter a path, or press Enter to skip.', 'muted')
|
||||
destination = input('> ').strip()
|
||||
if destination:
|
||||
destination = export_csv(destination, rows)
|
||||
print(f'CSV saved: {destination}')
|
||||
display = ReportDisplay()
|
||||
display.line('Report saved', 'success')
|
||||
display.line(str(destination))
|
||||
return 2 if warnings else 0
|
||||
except (OSError, RuntimeError, ValueError, subprocess.TimeoutExpired) as exc:
|
||||
print(f'Disk report failed: {exc}', file=sys.stderr)
|
||||
|
||||
Reference in New Issue
Block a user