""" Rich presenter implementation for enhanced console output. Provides rich terminal UI capabilities using the Rich library for colors, formatting, syntax highlighting, and interactive elements while maintaining backward compatibility with console mode. """ import sys import time from typing import Any, Dict, List, Optional, Union, Callable from pathlib import Path try: from rich.console import Console from rich.panel import Panel from rich.tree import Tree from rich.progress import Progress, TaskID, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn from rich.syntax import Syntax from rich.prompt import Confirm, Prompt from rich.text import Text from rich.rule import Rule from rich.live import Live from rich.markdown import Markdown from rich.table import Table from rich.align import Align from rich.padding import Padding RICH_AVAILABLE = True except ImportError: RICH_AVAILABLE = False # Fallback classes for type hints Console = None Panel = None Tree = None Progress = None TaskID = None from .base import BasePresenter, ProgressContext, TreeNode, TreeData class RichProgressContext(ProgressContext): """Rich-specific progress context using Rich's Progress display.""" def __init__(self, presenter: 'RichPresenter', task_name: str, total: Optional[int] = None): super().__init__(presenter, task_name, total) self.progress = None self.task_id = None self.live = None def start(self) -> None: """Start the Rich progress indicator.""" if not RICH_AVAILABLE: return # Create progress with appropriate columns columns = [ SpinnerColumn(), TextColumn("[progress.description]{task.description}"), ] if self.total is not None: columns.extend([ BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), TimeElapsedColumn(), ]) self.progress = Progress(*columns) self.task_id = self.progress.add_task( self.task_name, total=self.total ) # Start live display self.live = Live(self.progress, refresh_per_second=10) self.live.start() def update(self, advance: int = 1, description: Optional[str] = None) -> None: """Update the progress indicator.""" if not RICH_AVAILABLE or not self.progress or self.task_id is None: return self.current += advance update_kwargs = {"advance": advance} if description: update_kwargs["description"] = description self.task_name = description self.progress.update(self.task_id, **update_kwargs) def finish(self) -> None: """Finish and hide the progress indicator.""" if not RICH_AVAILABLE: return if self.live: self.live.stop() self.completed = True # Show completion message if hasattr(self.presenter, 'console'): self.presenter.console.print( f"✅ {self.task_name} completed", style="green" ) class RichPresenter(BasePresenter): """Rich presenter implementation using the Rich library for enhanced output.""" def __init__(self, **kwargs): super().__init__(**kwargs) if not RICH_AVAILABLE: self._show_rich_not_available() # Fall back to basic print statements self.console = None else: # Initialize Rich console with configuration self.console = Console( force_terminal=kwargs.get('force_terminal', None), width=kwargs.get('width', None), height=kwargs.get('height', None), color_system=kwargs.get('color_system', 'auto'), legacy_windows=kwargs.get('legacy_windows', True) ) def _show_rich_not_available(self) -> None: """Show message when Rich is not available.""" print("⚠️ Rich library not available. Install with: pip install rich") print(" Falling back to basic console output.") print() def _fallback_print(self, message: str, prefix: str = "") -> None: """Fallback to basic print when Rich is not available.""" if prefix: print(f"{prefix} {message}") else: print(message) # Status message methods def show_info(self, message: str, **kwargs) -> None: """Display an informational message with Rich formatting.""" if not RICH_AVAILABLE or not self.console: self._fallback_print(message, "ℹ️") return title = kwargs.get('title') if title: panel = Panel( message, title=title, border_style="blue", padding=(0, 1) ) self.console.print(panel) else: self.console.print(f"ℹ️ {message}", style="blue") def show_error(self, message: str, **kwargs) -> None: """Display an error message with Rich formatting.""" if not RICH_AVAILABLE or not self.console: self._fallback_print(message, "❌") return title = kwargs.get('title', "Error") panel = Panel( message, title=title, border_style="red", padding=(0, 1) ) self.console.print(panel) def show_success(self, message: str, **kwargs) -> None: """Display a success message with Rich formatting.""" if not RICH_AVAILABLE or not self.console: self._fallback_print(message, "✅") return title = kwargs.get('title') if title: panel = Panel( message, title=title, border_style="green", padding=(0, 1) ) self.console.print(panel) else: self.console.print(f"✅ {message}", style="green") def show_warning(self, message: str, **kwargs) -> None: """Display a warning message with Rich formatting.""" if not RICH_AVAILABLE or not self.console: self._fallback_print(message, "⚠️") return title = kwargs.get('title') if title: panel = Panel( message, title=title, border_style="yellow", padding=(0, 1) ) self.console.print(panel) else: self.console.print(f"⚠️ {message}", style="yellow") # Formatting methods def show_header(self, title: str, **kwargs) -> None: """Display a section header with Rich formatting.""" if not RICH_AVAILABLE or not self.console: self._fallback_print(f"\n{'=' * 50}") self._fallback_print(title.upper()) self._fallback_print('=' * 50) return subtitle = kwargs.get('subtitle') style = kwargs.get('style', 'bold blue') # Create header panel header_content = Text(title, style=style) if subtitle: header_content.append(f"\n{subtitle}", style="dim") panel = Panel( Align.center(header_content), border_style="blue", padding=(1, 2) ) self.console.print() self.console.print(panel) self.console.print() def show_separator(self, **kwargs) -> None: """Display a visual separator with Rich formatting.""" if not RICH_AVAILABLE or not self.console: self._fallback_print("-" * 50) return style = kwargs.get('style', 'dim') character = kwargs.get('character') title = kwargs.get('title') if character: rule = Rule(title=title, style=style, characters=character) else: rule = Rule(title=title, style=style) self.console.print(rule) # Content display methods def show_text(self, content: str, **kwargs) -> None: """Display plain text content with Rich formatting.""" if not RICH_AVAILABLE or not self.console: self._fallback_print(content) return wrap = kwargs.get('wrap', True) indent = kwargs.get('indent', 0) style = kwargs.get('style') text = Text(content, style=style) if indent > 0: text = Padding(text, (0, 0, 0, indent)) self.console.print(text, soft_wrap=wrap) def show_code(self, code: str, language: Optional[str] = None, **kwargs) -> None: """Display code with syntax highlighting using Rich.""" if not RICH_AVAILABLE or not self.console: self._fallback_print(f"```{language or ''}") self._fallback_print(code) self._fallback_print("```") return line_numbers = kwargs.get('line_numbers', True) theme = kwargs.get('theme', 'monokai') title = kwargs.get('title') # Create syntax object syntax = Syntax( code, language or 'text', theme=theme, line_numbers=line_numbers, word_wrap=True ) if title: panel = Panel( syntax, title=title, border_style="dim", padding=(0, 1) ) self.console.print(panel) else: self.console.print(syntax) def show_tree(self, tree_data: TreeData, **kwargs) -> None: """Display hierarchical tree structure using Rich Tree.""" if not RICH_AVAILABLE or not self.console: self._fallback_tree_display(tree_data, **kwargs) return expand_all = kwargs.get('expand_all', True) show_icons = kwargs.get('show_icons', True) title = kwargs.get('title', "Tree Structure") # Create Rich tree tree = Tree(title, expanded=expand_all) # Build tree structure if isinstance(tree_data, dict): self._build_tree_from_dict(tree, tree_data, show_icons) elif isinstance(tree_data, TreeNode): self._build_tree_from_node(tree, tree_data, show_icons) elif isinstance(tree_data, str): # Parse string representation tree.add(tree_data) else: tree.add(str(tree_data)) self.console.print(tree) def _fallback_tree_display(self, tree_data: TreeData, **kwargs) -> None: """Fallback tree display when Rich is not available.""" if isinstance(tree_data, dict): self._print_dict_tree(tree_data) elif isinstance(tree_data, TreeNode): self._print_node_tree(tree_data) else: self._fallback_print(str(tree_data)) def _print_dict_tree(self, data: dict, prefix: str = "", is_last: bool = True) -> None: """Print dictionary as tree structure (fallback).""" items = list(data.items()) for i, (key, value) in enumerate(items): is_last_item = i == len(items) - 1 current_prefix = "└── " if is_last_item else "├── " next_prefix = prefix + (" " if is_last_item else "│ ") print(f"{prefix}{current_prefix}{key}") if isinstance(value, dict): self._print_dict_tree(value, next_prefix, is_last_item) elif isinstance(value, list): for j, item in enumerate(value): item_is_last = j == len(value) - 1 item_prefix = "└── " if item_is_last else "├── " print(f"{next_prefix}{item_prefix}{item}") def _print_node_tree(self, node: TreeNode, prefix: str = "", is_last: bool = True) -> None: """Print TreeNode as tree structure (fallback).""" current_prefix = "└── " if is_last else "├── " next_prefix = prefix + (" " if is_last else "│ ") print(f"{prefix}{current_prefix}{node.name}") for i, child in enumerate(node.children): child_is_last = i == len(node.children) - 1 self._print_node_tree(child, next_prefix, child_is_last) def _build_tree_from_dict(self, tree: Tree, data: dict, show_icons: bool = True) -> None: """Build Rich tree from dictionary data.""" for key, value in data.items(): if isinstance(value, dict): if 'type' in value and value['type'] == 'directory': icon = "📁 " if show_icons else "" branch = tree.add(f"{icon}{key}/") if 'children' in value: self._build_tree_from_dict(branch, value['children'], show_icons) else: branch = tree.add(key) self._build_tree_from_dict(branch, value, show_icons) else: icon = self._get_file_icon(key) if show_icons else "" tree.add(f"{icon}{key}") def _build_tree_from_node(self, tree: Tree, node: TreeNode, show_icons: bool = True) -> None: """Build Rich tree from TreeNode.""" icon = "" if show_icons: if node.metadata.get('type') == 'directory': icon = "📁 " elif node.metadata.get('type') == 'file': icon = self._get_file_icon(node.name) elif node.metadata.get('type') == 'error': icon = "⚠️ " display_name = f"{icon}{node.name}" if node.metadata.get('type') == 'directory': display_name += "/" branch = tree.add(display_name) for child in node.children: self._build_tree_from_node(branch, child, show_icons) def _get_file_icon(self, filename: str) -> str: """Get appropriate icon for file based on extension.""" name_lower = filename.lower() # Special files special_files = { 'readme.md': '📖 ', 'readme.txt': '📖 ', 'readme.rst': '📖 ', 'license': '📄 ', 'license.txt': '📄 ', 'license.md': '📄 ', 'dockerfile': '🐳 ', 'docker-compose.yml': '🐳 ', 'docker-compose.yaml': '🐳 ', '.gitignore': '🔧 ', '.gitattributes': '🔧 ', 'package.json': '📦 ', 'package-lock.json': '📦 ', 'requirements.txt': '🐍 ', 'pyproject.toml': '🐍 ', 'setup.py': '🐍 ', 'cargo.toml': '🦀 ', 'cargo.lock': '🦀 ', 'pom.xml': '☕ ', 'build.gradle': '☕ ', 'go.mod': '🐹 ', 'go.sum': '🐹 ' } if name_lower in special_files: return special_files[name_lower] # File extensions ext = Path(filename).suffix.lower() ext_icons = { '.py': '🐍 ', '.js': '🟨 ', '.ts': '🔷 ', '.html': '🌐 ', '.css': '🎨 ', '.json': '📋 ', '.xml': '📄 ', '.yaml': '⚙️ ', '.yml': '⚙️ ', '.md': '📝 ', '.txt': '📄 ', '.log': '📜 ', '.sql': '🗃️ ', '.sh': '🐚 ', '.bat': '⚡ ', '.exe': '⚙️ ', '.dll': '🔧 ', '.jar': '☕ ', '.war': '☕ ', '.zip': '📦 ', '.tar': '📦 ', '.gz': '📦 ', '.pdf': '📕 ', '.doc': '📘 ', '.docx': '📘 ', '.xls': '📗 ', '.xlsx': '📗 ', '.ppt': '📙 ', '.pptx': '📙 ', '.jpg': '🖼️ ', '.jpeg': '🖼️ ', '.png': '🖼️ ', '.gif': '🖼️ ', '.svg': '🖼️ ', '.mp3': '🎵 ', '.mp4': '🎬 ', '.avi': '🎬 ', '.mov': '🎬 ' } return ext_icons.get(ext, '📄 ') # Interactive methods def stream_content(self, content_generator: Union[Callable[[], str], Any], **kwargs) -> None: """Display streaming content using Rich Live display.""" if not RICH_AVAILABLE or not self.console: # Fallback to simple streaming try: content = content_generator() self._fallback_print(content) except Exception as e: self._fallback_print(f"Error streaming content: {e}") return prefix = kwargs.get('prefix', '') suffix = kwargs.get('suffix', '') show_cursor = kwargs.get('show_cursor', True) # Create live display for streaming content_text = Text() if prefix: content_text.append(prefix, style="dim") with Live(content_text, refresh_per_second=10, console=self.console) as live: try: # Stream content full_content = "" for chunk in self._chunk_generator(content_generator): if chunk: full_content += chunk # Update display display_text = Text() if prefix: display_text.append(prefix, style="dim") display_text.append(full_content) if show_cursor: display_text.append("▋", style="blink") if suffix: display_text.append(suffix, style="dim") live.update(display_text) time.sleep(0.05) # Small delay for smooth streaming # Final update without cursor final_text = Text() if prefix: final_text.append(prefix, style="dim") final_text.append(full_content) if suffix: final_text.append(suffix, style="dim") live.update(final_text) except Exception as e: self.show_error(f"Error during content streaming: {e}") def _chunk_generator(self, content_generator: Callable[[], str]): """Convert content generator to chunks for streaming display.""" try: content = content_generator() if isinstance(content, str): # Split into chunks for streaming effect words = content.split() current_chunk = "" for word in words: current_chunk += word + " " if len(current_chunk) > 50: # Chunk size yield current_chunk current_chunk = "" if current_chunk: yield current_chunk else: yield str(content) except Exception as e: yield f"Error: {e}" def confirm_action(self, message: str, **kwargs) -> bool: """Prompt user for confirmation using Rich Prompt.""" if not RICH_AVAILABLE or not self.console: # Fallback to input() response = input(f"{message} (y/N): ").strip().lower() return response in ['y', 'yes'] default = kwargs.get('default', False) choices = kwargs.get('choices', ['y', 'n']) try: return Confirm.ask( message, default=default, console=self.console ) except (KeyboardInterrupt, EOFError): return False def show_progress(self, task_name: str, total: Optional[int] = None, **kwargs) -> ProgressContext: """Create and return a Rich progress context.""" return RichProgressContext(self, task_name, total) # Utility methods def clear_screen(self) -> None: """Clear the screen using Rich console.""" if RICH_AVAILABLE and self.console: self.console.clear() else: # Fallback import os os.system('cls' if os.name == 'nt' else 'clear') def set_title(self, title: str) -> None: """Set the console title using Rich.""" if RICH_AVAILABLE and self.console: self.console.set_window_title(title) def flush(self) -> None: """Flush console output.""" if RICH_AVAILABLE and self.console: # Rich console handles flushing automatically pass else: sys.stdout.flush() # Additional Rich-specific methods def show_table(self, data: List[Dict[str, Any]], **kwargs) -> None: """Display data in a Rich table format.""" if not RICH_AVAILABLE or not self.console: # Fallback to simple text table if data: headers = list(data[0].keys()) self._fallback_print(" | ".join(headers)) self._fallback_print("-" * (len(" | ".join(headers)))) for row in data: self._fallback_print(" | ".join(str(row.get(h, "")) for h in headers)) return if not data: return title = kwargs.get('title', 'Data Table') show_header = kwargs.get('show_header', True) table = Table(title=title, show_header=show_header) # Add columns headers = list(data[0].keys()) for header in headers: table.add_column(header, style="cyan") # Add rows for row in data: table.add_row(*[str(row.get(h, "")) for h in headers]) self.console.print(table) def show_markdown(self, markdown_content: str, **kwargs) -> None: """Display markdown content using Rich.""" if not RICH_AVAILABLE or not self.console: self._fallback_print(markdown_content) return md = Markdown(markdown_content) self.console.print(md)