mirror of
https://github.com/SuperClaude-Org/SuperClaude_Framework.git
synced 2025-12-17 17:56:46 +00:00
* Fix: Install only selected MCP servers and ensure valid empty backups This commit addresses two separate issues: 1. **MCP Installation:** The `install` command was installing all MCP servers instead of only the ones selected by the user. The `_install` method in `setup/components/mcp.py` was iterating through all available servers, not the user's selection. This has been fixed to respect the `selected_mcp_servers` configuration. A new test has been added to verify this fix. 2. **Backup Creation:** The `create_backup` method in `setup/core/installer.py` created an invalid `.tar.gz` file when the backup source was empty. This has been fixed to ensure that a valid, empty tar archive is always created. A test was added for this as well. Co-authored-by: Mithun Gowda B <mithungowda.b7411@gmail.com> Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com> * Fix: Correct installer validation for MCP and MCP Docs components This commit fixes a validation issue in the installer where it would incorrectly fail after a partial installation of MCP servers. The `MCPComponent` validation logic was checking for all "required" servers, regardless of whether they were selected by the user. This has been corrected to only validate the servers that were actually installed, by checking against the list of installed servers stored in the metadata. The metadata storage has also been fixed to only record the installed servers. The `MCPDocsComponent` was failing validation because it was not being registered in the metadata if no documentation files were installed. This has been fixed by ensuring the post-installation hook runs even when no files are copied. New tests have been added for both components to verify the corrected logic. Co-authored-by: Mithun Gowda B <mithungowda.b7411@gmail.com> Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com> * Fix: Allow re-installation of components and correct validation logic This commit fixes a bug that prevented new MCP servers from being installed on subsequent runs of the installer. It also fixes the validation logic that was causing failures after a partial installation. The key changes are: 1. A new `is_reinstallable` method has been added to the base `Component` class. This allows certain components (like the `mcp` component) to be re-run even if they are already marked as installed. 2. The installer logic has been updated to respect this new method. 3. The `MCPComponent` now correctly stores only the installed servers in the metadata. 4. The validation logic for `MCPComponent` and `MCPDocsComponent` has been corrected to prevent incorrect failures. New tests have been added to verify all aspects of the new logic. Co-authored-by: Mithun Gowda B <mithungowda.b7411@gmail.com> Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com> * feat: Display authors in UI header and update author info This commit implements the user's request to display author names and emails in the UI header of the installer. The key changes are: 1. The `__email__` field in `SuperClaude/__init__.py` has been updated to include both authors' emails. 2. The `display_header` function in `setup/utils/ui.py` has been modified to read the author and email information and display it. 3. A new test has been added to `tests/test_ui.py` to verify the new UI output. Co-authored-by: Mithun Gowda B <mithungowda.b7411@gmail.com> Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com> * feat: Version bump to 4.1.0 and various fixes This commit prepares the project for the v4.1.0 release. It includes a version bump across all relevant files and incorporates several bug fixes and feature enhancements from recent tasks. Key changes in this release: - **Version Bump**: The project version has been updated from 4.0.9 to 4.1.0 in all configuration files, documentation, and source code. - **Installer Fixes**: - Components can now be marked as `reinstallable`, allowing them to be re-run on subsequent installations. This fixes a bug where new MCP servers could not be added. - The validation logic for `mcp` and `mcp_docs` components has been corrected to avoid incorrect failures. - A bug in the backup creation process that created invalid empty archives has been fixed. - **UI Enhancements**: - Author names and emails are now displayed in the installer UI header. - **Metadata Updates**: - Mithun Gowda B has been added as an author. - **New Tests**: - Comprehensive tests have been added for the installer logic, MCP components, and UI changes to ensure correctness and prevent regressions. Co-authored-by: Mithun Gowda B <mithungowda.b7411@gmail.com> Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com> * fix: Resolve dependencies for partial installs and other fixes This commit addresses several issues, the main one being a dependency resolution failure during partial installations. Key changes: - **Dependency Resolution**: The installer now correctly resolves the full dependency tree when a user requests to install a subset of components. This fixes the "Unknown component: core" error. - **Component Re-installation**: A new `is_reinstallable` flag allows components like `mcp` to be re-run on subsequent installs, enabling the addition of new servers. - **Validation Logic**: The validation for `mcp` and `mcp_docs` has been corrected to avoid spurious failures. - **UI and Metadata**: Author information has been added to the UI header and source files. - **Version Bump**: The project version has been updated to 4.1.0. - **Tests**: New tests have been added to cover all the above changes. Co-authored-by: Mithun Gowda B <mithungowda.b7411@gmail.com> Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com> * fix: Installer fixes and version bump to 4.1.0 This commit includes a collection of fixes for the installer logic, UI enhancements, and a version bump to 4.1.0. Key changes: - **Dependency Resolution**: The installer now correctly resolves the full dependency tree for partial installations, fixing the "Unknown component: core" error. - **Component Re-installation**: A new `is_reinstallable` flag allows components like `mcp` to be re-run to add new servers. - **MCP Installation**: The non-interactive installation of the `mcp` component now correctly prompts the user to select servers. - **Validation Logic**: The post-installation validation logic has been corrected to only validate components from the current session and to use the correct list of installed servers. - **UI & Metadata**: Author information has been added to the UI and source files. - **Version Bump**: The project version has been updated from 4.0.9 to 4.1.0 across all files. - **Tests**: New tests have been added to cover all the bug fixes. Co-authored-by: Mithun Gowda B <mithungowda.b7411@gmail.com> Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com> * feat: Add --authors flag and multiple installer fixes This commit introduces the `--authors` flag to display author information and includes a collection of fixes for the installer logic. Key changes: - **New Feature**: Added an `--authors` flag that displays the names, emails, and GitHub usernames of the project authors. - **Dependency Resolution**: Fixed a critical bug where partial installations would fail due to unresolved dependencies. - **Component Re-installation**: Added a mechanism to allow components to be "reinstallable", fixing an issue that prevented adding new MCP servers on subsequent runs. - **MCP Installation**: The non-interactive installation of the `mcp` component now correctly prompts for server selection. - **Validation Logic**: Corrected the post-installation validation to prevent spurious errors. - **Version Bump**: The project version has been updated to 4.1.0. - **Metadata**: Author and GitHub information has been added to the source files. - **UI**: The installer header now displays author information. - **Tests**: Added new tests for all new features and bug fixes. Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com> --------- Co-authored-by: Mithun Gowda B <mithungowda.b7411@gmail.com> Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> Co-authored-by: Jules <jules-ai-assistant@users.noreply.github.com>
514 lines
17 KiB
Python
514 lines
17 KiB
Python
"""
|
|
User interface utilities for SuperClaude installation system
|
|
Cross-platform console UI with colors and progress indication
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
import shutil
|
|
import getpass
|
|
from typing import List, Optional, Any, Dict, Union
|
|
from enum import Enum
|
|
|
|
# Try to import colorama for cross-platform color support
|
|
try:
|
|
import colorama
|
|
from colorama import Fore, Back, Style
|
|
colorama.init(autoreset=True)
|
|
COLORAMA_AVAILABLE = True
|
|
except ImportError:
|
|
COLORAMA_AVAILABLE = False
|
|
# Fallback color codes for Unix-like systems
|
|
class MockFore:
|
|
RED = '\033[91m' if sys.platform != 'win32' else ''
|
|
GREEN = '\033[92m' if sys.platform != 'win32' else ''
|
|
YELLOW = '\033[93m' if sys.platform != 'win32' else ''
|
|
BLUE = '\033[94m' if sys.platform != 'win32' else ''
|
|
MAGENTA = '\033[95m' if sys.platform != 'win32' else ''
|
|
CYAN = '\033[96m' if sys.platform != 'win32' else ''
|
|
WHITE = '\033[97m' if sys.platform != 'win32' else ''
|
|
|
|
class MockStyle:
|
|
RESET_ALL = '\033[0m' if sys.platform != 'win32' else ''
|
|
BRIGHT = '\033[1m' if sys.platform != 'win32' else ''
|
|
|
|
Fore = MockFore()
|
|
Style = MockStyle()
|
|
|
|
|
|
class Colors:
|
|
"""Color constants for console output"""
|
|
RED = Fore.RED
|
|
GREEN = Fore.GREEN
|
|
YELLOW = Fore.YELLOW
|
|
BLUE = Fore.BLUE
|
|
MAGENTA = Fore.MAGENTA
|
|
CYAN = Fore.CYAN
|
|
WHITE = Fore.WHITE
|
|
RESET = Style.RESET_ALL
|
|
BRIGHT = Style.BRIGHT
|
|
|
|
|
|
class ProgressBar:
|
|
"""Cross-platform progress bar with customizable display"""
|
|
|
|
def __init__(self, total: int, width: int = 50, prefix: str = '', suffix: str = ''):
|
|
"""
|
|
Initialize progress bar
|
|
|
|
Args:
|
|
total: Total number of items to process
|
|
width: Width of progress bar in characters
|
|
prefix: Text to display before progress bar
|
|
suffix: Text to display after progress bar
|
|
"""
|
|
self.total = total
|
|
self.width = width
|
|
self.prefix = prefix
|
|
self.suffix = suffix
|
|
self.current = 0
|
|
self.start_time = time.time()
|
|
|
|
# Get terminal width for responsive display
|
|
try:
|
|
self.terminal_width = shutil.get_terminal_size().columns
|
|
except OSError:
|
|
self.terminal_width = 80
|
|
|
|
def update(self, current: int, message: str = '') -> None:
|
|
"""
|
|
Update progress bar
|
|
|
|
Args:
|
|
current: Current progress value
|
|
message: Optional message to display
|
|
"""
|
|
self.current = current
|
|
percent = min(100, (current / self.total) * 100) if self.total > 0 else 100
|
|
|
|
# Calculate filled and empty portions
|
|
filled_width = int(self.width * current / self.total) if self.total > 0 else self.width
|
|
filled = '█' * filled_width
|
|
empty = '░' * (self.width - filled_width)
|
|
|
|
# Calculate elapsed time and ETA
|
|
elapsed = time.time() - self.start_time
|
|
if current > 0:
|
|
eta = (elapsed / current) * (self.total - current)
|
|
eta_str = f" ETA: {self._format_time(eta)}"
|
|
else:
|
|
eta_str = ""
|
|
|
|
# Format progress line
|
|
if message:
|
|
status = f" {message}"
|
|
else:
|
|
status = ""
|
|
|
|
progress_line = (
|
|
f"\r{self.prefix}[{Colors.GREEN}{filled}{Colors.WHITE}{empty}{Colors.RESET}] "
|
|
f"{percent:5.1f}%{status}{eta_str}"
|
|
)
|
|
|
|
# Truncate if too long for terminal
|
|
max_length = self.terminal_width - 5
|
|
if len(progress_line) > max_length:
|
|
# Remove color codes for length calculation
|
|
plain_line = progress_line.replace(Colors.GREEN, '').replace(Colors.WHITE, '').replace(Colors.RESET, '')
|
|
if len(plain_line) > max_length:
|
|
progress_line = progress_line[:max_length] + "..."
|
|
|
|
print(progress_line, end='', flush=True)
|
|
|
|
def increment(self, message: str = '') -> None:
|
|
"""
|
|
Increment progress by 1
|
|
|
|
Args:
|
|
message: Optional message to display
|
|
"""
|
|
self.update(self.current + 1, message)
|
|
|
|
def finish(self, message: str = 'Complete') -> None:
|
|
"""
|
|
Complete progress bar
|
|
|
|
Args:
|
|
message: Completion message
|
|
"""
|
|
self.update(self.total, message)
|
|
print() # New line after completion
|
|
|
|
def _format_time(self, seconds: float) -> str:
|
|
"""Format time duration as human-readable string"""
|
|
if seconds < 60:
|
|
return f"{seconds:.0f}s"
|
|
elif seconds < 3600:
|
|
return f"{seconds/60:.0f}m {seconds%60:.0f}s"
|
|
else:
|
|
hours = seconds // 3600
|
|
minutes = (seconds % 3600) // 60
|
|
return f"{hours:.0f}h {minutes:.0f}m"
|
|
|
|
|
|
class Menu:
|
|
"""Interactive menu system with keyboard navigation"""
|
|
|
|
def __init__(self, title: str, options: List[str], multi_select: bool = False):
|
|
"""
|
|
Initialize menu
|
|
|
|
Args:
|
|
title: Menu title
|
|
options: List of menu options
|
|
multi_select: Allow multiple selections
|
|
"""
|
|
self.title = title
|
|
self.options = options
|
|
self.multi_select = multi_select
|
|
self.selected = set() if multi_select else None
|
|
|
|
def display(self) -> Union[int, List[int]]:
|
|
"""
|
|
Display menu and get user selection
|
|
|
|
Returns:
|
|
Selected option index (single) or list of indices (multi-select)
|
|
"""
|
|
print(f"\n{Colors.CYAN}{Colors.BRIGHT}{self.title}{Colors.RESET}")
|
|
print("=" * len(self.title))
|
|
|
|
for i, option in enumerate(self.options, 1):
|
|
if self.multi_select:
|
|
marker = "[x]" if i-1 in (self.selected or set()) else "[ ]"
|
|
print(f"{Colors.YELLOW}{i:2d}.{Colors.RESET} {marker} {option}")
|
|
else:
|
|
print(f"{Colors.YELLOW}{i:2d}.{Colors.RESET} {option}")
|
|
|
|
if self.multi_select:
|
|
print(f"\n{Colors.BLUE}Enter numbers separated by commas (e.g., 1,3,5) or 'all' for all options:{Colors.RESET}")
|
|
else:
|
|
print(f"\n{Colors.BLUE}Enter your choice (1-{len(self.options)}):{Colors.RESET}")
|
|
|
|
while True:
|
|
try:
|
|
user_input = input("> ").strip().lower()
|
|
|
|
if self.multi_select:
|
|
if user_input == 'all':
|
|
return list(range(len(self.options)))
|
|
elif user_input == '':
|
|
return []
|
|
else:
|
|
# Parse comma-separated numbers
|
|
selections = []
|
|
for part in user_input.split(','):
|
|
part = part.strip()
|
|
if part.isdigit():
|
|
idx = int(part) - 1
|
|
if 0 <= idx < len(self.options):
|
|
selections.append(idx)
|
|
else:
|
|
raise ValueError(f"Invalid option: {part}")
|
|
else:
|
|
raise ValueError(f"Invalid input: {part}")
|
|
return list(set(selections)) # Remove duplicates
|
|
else:
|
|
if user_input.isdigit():
|
|
choice = int(user_input) - 1
|
|
if 0 <= choice < len(self.options):
|
|
return choice
|
|
else:
|
|
print(f"{Colors.RED}Invalid choice. Please enter a number between 1 and {len(self.options)}.{Colors.RESET}")
|
|
else:
|
|
print(f"{Colors.RED}Please enter a valid number.{Colors.RESET}")
|
|
|
|
except (ValueError, KeyboardInterrupt) as e:
|
|
if isinstance(e, KeyboardInterrupt):
|
|
print(f"\n{Colors.YELLOW}Operation cancelled.{Colors.RESET}")
|
|
return [] if self.multi_select else -1
|
|
else:
|
|
print(f"{Colors.RED}Invalid input: {e}{Colors.RESET}")
|
|
|
|
|
|
def confirm(message: str, default: bool = True) -> bool:
|
|
"""
|
|
Ask for user confirmation
|
|
|
|
Args:
|
|
message: Confirmation message
|
|
default: Default response if user just presses Enter
|
|
|
|
Returns:
|
|
True if confirmed, False otherwise
|
|
"""
|
|
suffix = "[Y/n]" if default else "[y/N]"
|
|
print(f"{Colors.BLUE}{message} {suffix}{Colors.RESET}")
|
|
|
|
while True:
|
|
try:
|
|
response = input("> ").strip().lower()
|
|
|
|
if response == '':
|
|
return default
|
|
elif response in ['y', 'yes', 'true', '1']:
|
|
return True
|
|
elif response in ['n', 'no', 'false', '0']:
|
|
return False
|
|
else:
|
|
print(f"{Colors.RED}Please enter 'y' or 'n' (or press Enter for default).{Colors.RESET}")
|
|
|
|
except KeyboardInterrupt:
|
|
print(f"\n{Colors.YELLOW}Operation cancelled.{Colors.RESET}")
|
|
return False
|
|
|
|
|
|
def display_header(title: str, subtitle: str = '') -> None:
|
|
"""
|
|
Display formatted header
|
|
|
|
Args:
|
|
title: Main title
|
|
subtitle: Optional subtitle
|
|
"""
|
|
from SuperClaude import __author__, __email__
|
|
|
|
print(f"\n{Colors.CYAN}{Colors.BRIGHT}{'='*60}{Colors.RESET}")
|
|
print(f"{Colors.CYAN}{Colors.BRIGHT}{title:^60}{Colors.RESET}")
|
|
if subtitle:
|
|
print(f"{Colors.WHITE}{subtitle:^60}{Colors.RESET}")
|
|
|
|
# Display authors
|
|
authors = [a.strip() for a in __author__.split(',')]
|
|
emails = [e.strip() for e in __email__.split(',')]
|
|
|
|
author_lines = []
|
|
for i in range(len(authors)):
|
|
name = authors[i]
|
|
email = emails[i] if i < len(emails) else ''
|
|
author_lines.append(f"{name} <{email}>")
|
|
|
|
authors_str = " | ".join(author_lines)
|
|
print(f"{Colors.BLUE}{authors_str:^60}{Colors.RESET}")
|
|
|
|
print(f"{Colors.CYAN}{Colors.BRIGHT}{'='*60}{Colors.RESET}\n")
|
|
|
|
|
|
def display_authors() -> None:
|
|
"""Display author information"""
|
|
from SuperClaude import __author__, __email__, __github__
|
|
|
|
print(f"\n{Colors.CYAN}{Colors.BRIGHT}{'='*60}{Colors.RESET}")
|
|
print(f"{Colors.CYAN}{Colors.BRIGHT}{'SuperClaude Authors':^60}{Colors.RESET}")
|
|
print(f"{Colors.CYAN}{Colors.BRIGHT}{'='*60}{Colors.RESET}\n")
|
|
|
|
authors = [a.strip() for a in __author__.split(',')]
|
|
emails = [e.strip() for e in __email__.split(',')]
|
|
github_users = [g.strip() for g in __github__.split(',')]
|
|
|
|
for i in range(len(authors)):
|
|
name = authors[i]
|
|
email = emails[i] if i < len(emails) else 'N/A'
|
|
github = github_users[i] if i < len(github_users) else 'N/A'
|
|
|
|
print(f" {Colors.BRIGHT}{name}{Colors.RESET}")
|
|
print(f" Email: {Colors.YELLOW}{email}{Colors.RESET}")
|
|
print(f" GitHub: {Colors.YELLOW}https://github.com/{github}{Colors.RESET}")
|
|
print()
|
|
|
|
print(f"{Colors.CYAN}{'='*60}{Colors.RESET}\n")
|
|
|
|
|
|
def display_info(message: str) -> None:
|
|
"""Display info message"""
|
|
print(f"{Colors.BLUE}[INFO] {message}{Colors.RESET}")
|
|
|
|
|
|
def display_success(message: str) -> None:
|
|
"""Display success message"""
|
|
print(f"{Colors.GREEN}[✓] {message}{Colors.RESET}")
|
|
|
|
|
|
def display_warning(message: str) -> None:
|
|
"""Display warning message"""
|
|
print(f"{Colors.YELLOW}[!] {message}{Colors.RESET}")
|
|
|
|
|
|
def display_error(message: str) -> None:
|
|
"""Display error message"""
|
|
print(f"{Colors.RED}[✗] {message}{Colors.RESET}")
|
|
|
|
|
|
def display_step(step: int, total: int, message: str) -> None:
|
|
"""Display step progress"""
|
|
print(f"{Colors.CYAN}[{step}/{total}] {message}{Colors.RESET}")
|
|
|
|
|
|
def display_table(headers: List[str], rows: List[List[str]], title: str = '') -> None:
|
|
"""
|
|
Display data in table format
|
|
|
|
Args:
|
|
headers: Column headers
|
|
rows: Data rows
|
|
title: Optional table title
|
|
"""
|
|
if not rows:
|
|
return
|
|
|
|
# Calculate column widths
|
|
col_widths = [len(header) for header in headers]
|
|
for row in rows:
|
|
for i, cell in enumerate(row):
|
|
if i < len(col_widths):
|
|
col_widths[i] = max(col_widths[i], len(str(cell)))
|
|
|
|
# Display title
|
|
if title:
|
|
print(f"\n{Colors.CYAN}{Colors.BRIGHT}{title}{Colors.RESET}")
|
|
print()
|
|
|
|
# Display headers
|
|
header_line = " | ".join(f"{header:<{col_widths[i]}}" for i, header in enumerate(headers))
|
|
print(f"{Colors.YELLOW}{header_line}{Colors.RESET}")
|
|
print("-" * len(header_line))
|
|
|
|
# Display rows
|
|
for row in rows:
|
|
row_line = " | ".join(f"{str(cell):<{col_widths[i]}}" for i, cell in enumerate(row))
|
|
print(row_line)
|
|
|
|
print()
|
|
|
|
|
|
def prompt_api_key(service_name: str, env_var_name: str) -> Optional[str]:
|
|
"""
|
|
Prompt for API key with security and UX best practices
|
|
|
|
Args:
|
|
service_name: Human-readable service name (e.g., "Magic", "Morphllm")
|
|
env_var_name: Environment variable name (e.g., "TWENTYFIRST_API_KEY")
|
|
|
|
Returns:
|
|
API key string if provided, None if skipped
|
|
"""
|
|
print(f"{Colors.BLUE}[API KEY] {service_name} requires: {Colors.BRIGHT}{env_var_name}{Colors.RESET}")
|
|
print(f"{Colors.WHITE}Visit the service documentation to obtain your API key{Colors.RESET}")
|
|
print(f"{Colors.YELLOW}Press Enter to skip (you can set this manually later){Colors.RESET}")
|
|
|
|
try:
|
|
# Use getpass for hidden input
|
|
api_key = getpass.getpass(f"Enter {env_var_name}: ").strip()
|
|
|
|
if not api_key:
|
|
print(f"{Colors.YELLOW}[SKIPPED] {env_var_name} - set manually later{Colors.RESET}")
|
|
return None
|
|
|
|
# Basic validation (non-empty, reasonable length)
|
|
if len(api_key) < 10:
|
|
print(f"{Colors.RED}[WARNING] API key seems too short. Continue anyway? (y/N){Colors.RESET}")
|
|
if not confirm("", default=False):
|
|
return None
|
|
|
|
print(f"{Colors.GREEN}[✓] {env_var_name} configured{Colors.RESET}")
|
|
return api_key
|
|
|
|
except KeyboardInterrupt:
|
|
print(f"\n{Colors.YELLOW}[SKIPPED] {env_var_name}{Colors.RESET}")
|
|
return None
|
|
|
|
|
|
def wait_for_key(message: str = "Press Enter to continue...") -> None:
|
|
"""Wait for user to press a key"""
|
|
try:
|
|
input(f"{Colors.BLUE}{message}{Colors.RESET}")
|
|
except KeyboardInterrupt:
|
|
print(f"\n{Colors.YELLOW}Operation cancelled.{Colors.RESET}")
|
|
|
|
|
|
def clear_screen() -> None:
|
|
"""Clear terminal screen"""
|
|
import os
|
|
os.system('cls' if os.name == 'nt' else 'clear')
|
|
|
|
|
|
class StatusSpinner:
|
|
"""Simple status spinner for long operations"""
|
|
|
|
def __init__(self, message: str = "Working..."):
|
|
"""
|
|
Initialize spinner
|
|
|
|
Args:
|
|
message: Message to display with spinner
|
|
"""
|
|
self.message = message
|
|
self.spinning = False
|
|
self.chars = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
|
|
self.current = 0
|
|
|
|
def start(self) -> None:
|
|
"""Start spinner in background thread"""
|
|
import threading
|
|
|
|
def spin():
|
|
while self.spinning:
|
|
char = self.chars[self.current % len(self.chars)]
|
|
print(f"\r{Colors.BLUE}{char} {self.message}{Colors.RESET}", end='', flush=True)
|
|
self.current += 1
|
|
time.sleep(0.1)
|
|
|
|
self.spinning = True
|
|
self.thread = threading.Thread(target=spin, daemon=True)
|
|
self.thread.start()
|
|
|
|
def stop(self, final_message: str = '') -> None:
|
|
"""
|
|
Stop spinner
|
|
|
|
Args:
|
|
final_message: Final message to display
|
|
"""
|
|
self.spinning = False
|
|
if hasattr(self, 'thread'):
|
|
self.thread.join(timeout=0.2)
|
|
|
|
# Clear spinner line
|
|
print(f"\r{' ' * (len(self.message) + 5)}\r", end='')
|
|
|
|
if final_message:
|
|
print(final_message)
|
|
|
|
|
|
def format_size(size_bytes: int) -> str:
|
|
"""Format file size in human-readable format"""
|
|
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
|
|
if size_bytes < 1024.0:
|
|
return f"{size_bytes:.1f} {unit}"
|
|
size_bytes /= 1024.0
|
|
return f"{size_bytes:.1f} PB"
|
|
|
|
|
|
def format_duration(seconds: float) -> str:
|
|
"""Format duration in human-readable format"""
|
|
if seconds < 1:
|
|
return f"{seconds*1000:.0f}ms"
|
|
elif seconds < 60:
|
|
return f"{seconds:.1f}s"
|
|
elif seconds < 3600:
|
|
minutes = seconds // 60
|
|
secs = seconds % 60
|
|
return f"{minutes:.0f}m {secs:.0f}s"
|
|
else:
|
|
hours = seconds // 3600
|
|
minutes = (seconds % 3600) // 60
|
|
return f"{hours:.0f}h {minutes:.0f}m"
|
|
|
|
|
|
def truncate_text(text: str, max_length: int, suffix: str = "...") -> str:
|
|
"""Truncate text to maximum length with optional suffix"""
|
|
if len(text) <= max_length:
|
|
return text
|
|
|
|
return text[:max_length - len(suffix)] + suffix
|