""" Base command class for MiniMax Code CLI commands. Provides common functionality and utilities for all command implementations. Enhanced with security improvements and caching. """ import os import logging import time from typing import Any, Dict, List, Optional, Union, Tuple from abc import ABC, abstractmethod from pathlib import Path from huggingface_hub import InferenceClient from ..config import Configuration from ..chat import create_chat_completion from ..exceptions import FileOperationError, ValidationError from ..security import sanitize_file_path, validate_user_input, secure_backup_filename, SecurityError from ..cache import get_cache_manager from ..presentation.base import BasePresenter class BaseCommand(ABC): """Base class for all CLI commands.""" def __init__(self, client: InferenceClient, config: Configuration, presenter: BasePresenter) -> None: """Initialize the base command with client, configuration, and presenter. Args: client: HuggingFace inference client config: Application configuration presenter: Presenter instance for output handling """ self.client = client self.config = config self.presenter = presenter self.logger = logging.getLogger(self.__class__.__name__) self.cache_manager = get_cache_manager() self.operation_history = [] # Track operations for rollback @abstractmethod def execute(self, args: Any) -> bool: """Execute the command with given arguments. Args: args: Parsed command line arguments Returns: True if command executed successfully, False otherwise """ pass def read_file(self, file_path: str, base_dir: Optional[str] = None) -> Optional[str]: """Safely read a file and return its contents with security validation. Args: file_path: Path to the file to read base_dir: Base directory to restrict access to Returns: File contents as string, or None if error occurred Raises: FileOperationError: If file operation fails SecurityError: If security validation fails """ try: # Security validation first safe_path = sanitize_file_path(file_path, base_dir, allow_absolute=True) if not safe_path.exists(): self.logger.error(f"File not found: {safe_path}") raise FileOperationError( f"File not found: {safe_path}", error_code="FILE_NOT_FOUND", details={"file_path": str(safe_path)} ) if not safe_path.is_file(): self.logger.error(f"Path is not a file: {safe_path}") raise FileOperationError( f"Path is not a file: {safe_path}", error_code="NOT_A_FILE", details={"file_path": str(safe_path)} ) # Check file size for security file_size = safe_path.stat().st_size max_size = 10 * 1024 * 1024 # 10MB limit if file_size > max_size: raise FileOperationError( f"File too large: {file_size} bytes > {max_size}", error_code="FILE_TOO_LARGE", details={"file_path": str(safe_path), "size": file_size} ) with open(safe_path, 'r', encoding='utf-8') as f: content = f.read() self.logger.debug(f"Successfully read {len(content)} characters from {safe_path}") return content except FileOperationError: # Re-raise our custom exceptions raise except PermissionError as e: self.logger.error(f"Permission denied reading file {file_path}: {e}") raise FileOperationError( f"Permission denied: {file_path}", error_code="PERMISSION_DENIED", details={"file_path": file_path} ) except UnicodeDecodeError as e: self.logger.error(f"Unicode decode error reading file {file_path}: {e}") raise FileOperationError( f"File encoding error: {file_path}", error_code="ENCODING_ERROR", details={"file_path": file_path, "encoding": "utf-8"} ) except Exception as e: self.logger.error(f"Unexpected error reading file {file_path}: {e}") raise FileOperationError( f"Unexpected error reading file: {str(e)}", error_code="READ_ERROR", details={"file_path": file_path, "error_type": type(e).__name__} ) def write_file(self, file_path: str, content: str, backup: bool = False) -> bool: """Safely write content to a file with optional backup.""" try: # Create backup if requested and file exists if backup and os.path.exists(file_path): backup_path = f"{file_path}.backup" with open(file_path, 'r', encoding='utf-8') as src: with open(backup_path, 'w', encoding='utf-8') as dst: dst.write(src.read()) self.logger.info(f"Backup created: {backup_path}") # Write the new content with open(file_path, 'w', encoding='utf-8') as f: f.write(content) self.logger.info(f"File written: {file_path}") return True except Exception as e: self.logger.error(f"Error writing file {file_path}: {e}") return False def get_file_extension(self, file_path: str) -> str: """Get the file extension.""" return os.path.splitext(file_path)[1].lower() def detect_language(self, file_path: str) -> str: """Detect programming language from file extension.""" ext_map = { '.py': 'python', '.js': 'javascript', '.ts': 'typescript', '.java': 'java', '.cpp': 'cpp', '.c': 'c', '.cs': 'csharp', '.go': 'go', '.rs': 'rust', '.php': 'php', '.rb': 'ruby', '.swift': 'swift', '.kt': 'kotlin', '.scala': 'scala', '.sh': 'bash', '.sql': 'sql', '.html': 'html', '.css': 'css', '.json': 'json', '.yaml': 'yaml', '.yml': 'yaml', '.xml': 'xml', '.md': 'markdown' } ext = self.get_file_extension(file_path) return ext_map.get(ext, 'text') def ask_ai(self, prompt: str, context: Optional[str] = None, use_cache: bool = True, stream: bool = False) -> Optional[str]: """Send a prompt to the AI and get response with caching support. Args: prompt: The prompt to send to the AI context: Optional context to include with the prompt use_cache: Whether to use cached responses stream: Whether to stream the response through the presenter Returns: AI response as string, or None if error occurred """ try: # Validate input prompt = validate_user_input(prompt, max_length=50000, allow_code=True) if context: context = validate_user_input(context, max_length=100000, allow_code=True) # Combine context and prompt if context provided full_prompt = prompt if context: full_prompt = f"Context:\n{context}\n\nRequest:\n{prompt}" # Check cache first if enabled if use_cache: model_params = { 'model_name': self.config.model_name, 'temperature': getattr(self.config, 'temperature', 0.7), 'max_tokens': getattr(self.config, 'max_tokens', 4000) } cached_response = self.cache_manager.get_cached_ai_response(full_prompt, model_params) if cached_response: self.logger.debug("Using cached AI response") if stream: # For cached responses, we can simulate streaming by yielding chunks def content_generator(): words = cached_response.split() for i, word in enumerate(words): if i == 0: yield word else: yield f" {word}" time.sleep(0.01) # Small delay to simulate streaming self.presenter.stream_content(content_generator) return cached_response # Make AI request if stream: # For streaming, we need to handle the response differently response_chunks = [] def content_generator(): nonlocal response_chunks try: for chunk in create_chat_completion( client=self.client, model_name=self.config.model_name, user_message=full_prompt, stream=True ): if chunk: response_chunks.append(chunk) yield chunk except Exception as e: self.logger.error(f"Error during streaming: {e}") yield f"\n[Error: {str(e)}]" # Stream the content through the presenter self.presenter.stream_content(content_generator) # Combine chunks for caching and return full_response = ''.join(response_chunks) # Cache the response if successful and caching enabled if full_response and use_cache: self.cache_manager.cache_ai_response(full_prompt, model_params, full_response) return full_response else: # Non-streaming request response = create_chat_completion( client=self.client, model_name=self.config.model_name, user_message=full_prompt, stream=False ) # Cache the response if successful and caching enabled if response and use_cache: model_params = { 'model_name': self.config.model_name, 'temperature': getattr(self.config, 'temperature', 0.7), 'max_tokens': getattr(self.config, 'max_tokens', 4000) } self.cache_manager.cache_ai_response(full_prompt, model_params, response) return response except SecurityError as e: self.logger.error(f"Security validation failed: {e}") self.presenter.show_error(f"Security validation failed: {str(e)}") return None except Exception as e: self.logger.error(f"Error communicating with AI: {e}") self.presenter.show_error(f"Error communicating with AI: {str(e)}") return None def format_code_block(self, code: str, language: str = "") -> str: """Format code in a markdown code block.""" return f"```{language}\n{code}\n```" def print_success(self, message: str): """Print a success message.""" self.presenter.show_success(message) def print_error(self, message: str): """Print an error message.""" self.presenter.show_error(message) def print_info(self, message: str): """Print an info message.""" self.presenter.show_info(message) def print_warning(self, message: str): """Print a warning message.""" self.presenter.show_warning(message) def confirm_action(self, message: str) -> bool: """Ask user for confirmation.""" try: return self.presenter.confirm_action(message) except KeyboardInterrupt: return False def show_table(self, data: List[Dict[str, Any]], **kwargs) -> None: """Display tabular data using the presenter. Args: data: List of dictionaries representing table rows **kwargs: Additional formatting options """ self.presenter.show_table(data, **kwargs) def show_code(self, code: str, language: Optional[str] = None, **kwargs) -> None: """Display code with syntax highlighting using the presenter. Args: code: The code content to display language: Programming language for syntax highlighting **kwargs: Additional formatting options """ self.presenter.show_code(code, language, **kwargs) def show_tree(self, tree_data: Union[Dict, List, str], **kwargs) -> None: """Display hierarchical tree structure using the presenter. Args: tree_data: Tree structure data **kwargs: Additional formatting options """ self.presenter.show_tree(tree_data, **kwargs) def show_markdown(self, content: str, **kwargs) -> None: """Display markdown content using the presenter. Args: content: Markdown content to display **kwargs: Additional formatting options """ self.presenter.show_markdown(content, **kwargs) def show_notification(self, message: str, level: str = 'info', **kwargs) -> None: """Display a notification message using the presenter. Args: message: The notification message level: Notification level ('info', 'warning', 'error', 'success') **kwargs: Additional options """ self.presenter.show_notification(message, level, **kwargs) def get_lines_from_file(self, file_path: str, line_range: str) -> Optional[str]: """Extract specific lines from a file.""" try: content = self.read_file(file_path) if not content: return None lines = content.split('\n') # Parse line range (e.g., "10-20" or "15") if '-' in line_range: start, end = map(int, line_range.split('-')) start = max(1, start) - 1 # Convert to 0-based index end = min(len(lines), end) selected_lines = lines[start:end] else: line_num = int(line_range) - 1 # Convert to 0-based index if 0 <= line_num < len(lines): selected_lines = [lines[line_num]] else: return None return '\n'.join(selected_lines) except Exception as e: self.logger.error(f"Error extracting lines from {file_path}: {e}") return None def create_prompt_template(self, template_type: str, **kwargs) -> str: """Create standardized prompts for different operations.""" templates = { 'code_review': """ Please review the following {language} code and provide feedback on: - Code quality and best practices - Potential bugs or issues - Performance considerations - Security concerns - Readability and maintainability {focus_instruction} Code: {code} Please provide specific, actionable feedback. """, 'code_explanation': """ Please explain the following {language} code in detail: - What does this code do? - How does it work? - What are the key components? - Are there any important patterns or techniques used? Code: {code} Please provide a clear, educational explanation. """, 'code_generation': """ Please generate {language} code for the following requirement: {description} Requirements: - Write clean, well-documented code - Follow best practices for {language} - Include appropriate error handling - Add helpful comments {additional_requirements} """, 'code_debugging': """ Please help debug the following {language} code: Code: {code} {error_info} Please: 1. Identify the likely cause of the issue 2. Suggest specific fixes 3. Explain why the issue occurred 4. Provide the corrected code if possible """, 'code_optimization': """ Please optimize the following {language} code for {target}: Code: {code} Please: 1. Identify optimization opportunities 2. Provide optimized version 3. Explain the improvements made 4. Mention any trade-offs """ } template = templates.get(template_type, "") return template.format(**kwargs)