Skip to content

Instantly share code, notes, and snippets.

@r0x0d
Last active June 5, 2025 17:15
Show Gist options
  • Select an option

  • Save r0x0d/4c49eb85bad53df2c9ac09803541c5e3 to your computer and use it in GitHub Desktop.

Select an option

Save r0x0d/4c49eb85bad53df2c9ac09803541c5e3 to your computer and use it in GitHub Desktop.
claude 4 re-thought how our commands module should look like

Simplified Command Structure

This directory contains a simplified, elegant command structure inspired by typer/click but using argparse. The new structure eliminates complex class hierarchies, factories, and operations in favor of a clean decorator-based approach with two powerful patterns: action handlers and subcommands.

Key Benefits

  1. Simplicity: Functions instead of complex class hierarchies
  2. No branching logic: Eliminate if/else chains in command functions
  3. Decorator-based: Easy-to-use decorators for command and argument registration
  4. Action handlers: Tie specific functions directly to command-line flags
  5. Subcommands: Clean separation of complex operations
  6. Less boilerplate: Minimal code required to create new commands
  7. Easy to understand: Clear, linear code flow
  8. Maintainable: Reduced complexity makes the code easier to maintain and extend

Architecture Overview

Core Components

  • registry.py: Command registration system with decorators
  • utils.py: Common utilities and helpers for commands
  • base.py: Legacy base classes (deprecated, to be removed)
  • Individual command files: chat.py, history.py, feedback.py, shell.py
  • chat_new.py: Example showing new action handlers and subcommands
  • example.py: Complete demonstration of both patterns

Two Approaches for Handling Command Operations

1. Action Handlers (For Simple Flags)

Perfect for simple flags that trigger specific actions. No more branching logic!

def _handle_list(args: Namespace, context: CommandContext) -> int:
    """Handler for --list flag."""
    utils = create_utils(context)
    # Handle list operation
    return 0

@command("mycommand", help="My command")
@argument("--list", action="store_true", help="List items", handler=_handle_list)
def my_command(args: Namespace, context: CommandContext) -> int:
    """Main command - only handles default behavior."""
    # This only runs when --list is NOT used
    return 0

2. Subcommands (For Complex Operations)

Perfect for complex operations with their own arguments:

@subcommand("mycommand", "create", help="Create a new item")
@argument("name", help="Name of the item")
@argument("--type", help="Type of item")
def create_item(args: Namespace, context: CommandContext) -> int:
    """Completely separate function for create operation."""
    utils = create_utils(context)
    # Handle create operation
    return 0

# Usage: mycommand create "item-name" --type "special"

Creating a New Command

You have three approaches depending on your needs:

Approach 1: Simple Command (Traditional)

@command("newcmd", help="My new command")
@argument("message", help="Message to display")
def new_command(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    utils.render_success(args.message)
    return 0

Approach 2: With Action Handlers (Eliminates Branching)

def _handle_uppercase(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    utils.render_success(args.message.upper())
    return 0

@command("newcmd", help="My new command")
@argument("message", help="Message to display")
@argument("-u", "--uppercase", action="store_true", help="Uppercase", handler=_handle_uppercase)
def new_command(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    # This only runs when --uppercase is NOT used
    utils.render_success(args.message)
    return 0

Approach 3: With Subcommands (Complex Operations)

@subcommand("newcmd", "process", help="Process the message")
@argument("message", help="Message to process")
@argument("--method", choices=["upper", "lower"], help="Processing method")
def process_message(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    if args.method == "upper":
        result = args.message.upper()
    else:
        result = args.message.lower()
    utils.render_success(result)
    return 0

@command("newcmd", help="My new command with subcommands")
def new_command(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    utils.render_success("Use 'newcmd process <message>' to process messages")
    return 0

# Usage: newcmd process "Hello" --method upper

Step 2: Import in __init__.py

Add your command module to the imports in __init__.py:

from command_line_assistant.commands import chat, feedback, history, shell, newcmd

That's it! No factories, no complex registration, no multiple inheritance.

Available Utilities

The CommandUtils class provides common functionality:

Renderers

  • utils.render_success(message) - Success messages
  • utils.render_warning(message) - Warning messages
  • utils.render_error(message) - Error messages
  • utils.text_renderer - Direct access to text renderer
  • utils.warning_renderer - Direct access to warning renderer
  • utils.error_renderer - Direct access to error renderer

D-Bus Proxies

  • utils.chat_proxy - Chat interface proxy
  • utils.history_proxy - History interface proxy
  • utils.user_proxy - User interface proxy

Helper Methods

  • utils.get_user_id() - Get current user ID
  • utils.context - Access to command context

Migration from Old Structure

Old Structure (Complex with Branching)

class MyOperationType(CommandOperationType):
    LIST = auto()
    DELETE = auto()
    CREATE = auto()

class MyOperationFactory(CommandOperationFactory):
    _arg_to_operation = {
        "list": MyOperationType.LIST,
        "delete": MyOperationType.DELETE,
        "create": MyOperationType.CREATE,
    }

@MyOperationFactory.register(MyOperationType.LIST)
class ListOperation(BaseOperation):
    def execute(self) -> None:
        # List logic
        pass

# ... more operation classes ...

class MyCommand(BaseCLICommand):
    def run(self) -> int:
        factory = MyOperationFactory()
        operation = factory.create_operation(self._args, self._context)
        if operation:
            operation.execute()
        return 0

def register_subcommand(parser):
    # Complex argument setup...
    pass

New Structure Option 1: Action Handlers

def _handle_list(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    # List logic here
    return 0

def _handle_delete(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    # Delete logic here  
    return 0

@command("mycommand", help="My command")
@argument("--list", action="store_true", help="List items", handler=_handle_list)
@argument("--delete", help="Delete item", handler=_handle_delete)
def my_command(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    # Default behavior only - no branching!
    return 0

New Structure Option 2: Subcommands

@subcommand("mycommand", "list", help="List items")
def list_items(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    # List logic here
    return 0

@subcommand("mycommand", "delete", help="Delete an item")
@argument("item_name", help="Name of item to delete")
def delete_item(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    # Delete logic here
    return 0

@command("mycommand", help="My command")
def my_command(args: Namespace, context: CommandContext) -> int:
    utils = create_utils(context)
    # Default behavior when no subcommand used
    return 0

# Usage: mycommand list, mycommand delete item-name

Decorator Details

@command(name, help=None, description=None)

Registers a function as a CLI command.

  • name: Command name (required)
  • help: Short help text (optional)
  • description: Long description (optional)

@argument(*args, handler=None, **kwargs)

Adds an argument to the command. Uses the same signature as argparse.add_argument().

  • handler: Optional function to handle this specific argument (eliminates branching)
  • All other kwargs are passed to argparse.add_argument()

Note: Arguments are applied in reverse order (decorators are bottom-up), so the last @argument decorator will be the first argument added to the parser.

@subcommand(parent_command, name, help=None, description=None)

Registers a function as a subcommand of an existing command.

  • parent_command: Name of parent command (required)
  • name: Subcommand name (required)
  • help: Short help text (optional)
  • description: Long description (optional)

Subcommands can have their own @argument decorators for subcommand-specific arguments.

Real-World Examples

Example 1: Chat Command (chat_new.py)

Shows both approaches in one command:

# Action handler for simple flags
def _handle_list(args: Namespace, context: CommandContext) -> int:
    return list_chats(args, context)

# Subcommands for complex operations  
@subcommand("chat", "delete", help="Delete a specific chat")
@argument("chat_name", help="Name of the chat to delete")
def delete_chat(args: Namespace, context: CommandContext) -> int:
    # Implementation here
    pass

@command("chat", help="Command to ask a question to the LLM")
@argument("query_string", nargs="?", help="The question", default="")
@argument("-l", "--list", action="store_true", help="List chats", handler=_handle_list)
def chat_command(args: Namespace, context: CommandContext) -> int:
    # Only handles direct questions - no branching!
    pass

Example 2: Complete Demo (example.py)

See example.py for a complete example showing:

  • Action handlers for simple flags
  • Subcommands for complex operations
  • Positional and optional arguments
  • Different argument types
  • Error handling
  • Using various utilities

Usage examples:

  • example "Hello World" - Main command
  • example --uppercase "hello" - Action handler
  • example greet Alice --formal - Subcommand
  • example count 10 --step 2 - Subcommand with args

Error Handling

Commands should:

  1. Use try/except blocks for error handling
  2. Use appropriate exception types (e.g., ChatCommandException)
  3. Render errors using utils.render_error()
  4. Return appropriate exit codes (0 for success, non-zero for failure)

Best Practices

General

  1. Choose the right pattern:
    • Action handlers for simple flags that trigger one action
    • Subcommands for complex operations with their own arguments
    • Traditional approach for simple commands without branching
  2. Keep functions focused: Each function should do one thing well
  3. Use utilities: Leverage CommandUtils for common operations
  4. Handle errors gracefully: Always catch and handle exceptions properly
  5. Return appropriate exit codes: 0 for success, non-zero for errors
  6. Use descriptive help text: Make commands self-documenting

Action Handlers

  1. Use for simple flags: Perfect for --list, --delete, --help style flags
  2. Keep handlers simple: Each handler should be a focused function
  3. Name handlers clearly: Use _handle_<action> naming convention
  4. Avoid complex logic: If it needs lots of arguments, consider subcommands

Subcommands

  1. Use for complex operations: Perfect for operations that need their own arguments
  2. Group related functionality: Keep related subcommands in the same file
  3. Make subcommands independent: Each should be a complete, standalone function
  4. Use clear names: Subcommand names should be action verbs

Migration Strategy

  1. Start with action handlers: Convert simple if/else branching first
  2. Move to subcommands: For complex operations with many arguments
  3. Keep main command simple: Should only handle default behavior
  4. Test incrementally: Migrate one operation at a time

Legacy Support

The old base classes and factory pattern are still available but deprecated. New commands should use the simplified structure. The old system will be removed in a future version.

diff --git a/command_line_assistant/commands/README.md b/command_line_assistant/commands/README.md
new file mode 100644
index 0000000..240428f
--- /dev/null
+++ b/command_line_assistant/commands/README.md
@@ -0,0 +1,345 @@
+# Simplified Command Structure
+
+This directory contains a simplified, elegant command structure inspired by typer/click but using argparse. The new structure eliminates complex class hierarchies, factories, and operations in favor of a clean decorator-based approach with two powerful patterns: **action handlers** and **subcommands**.
+
+## Key Benefits
+
+1. **Simplicity**: Functions instead of complex class hierarchies
+2. **No branching logic**: Eliminate if/else chains in command functions
+3. **Decorator-based**: Easy-to-use decorators for command and argument registration
+4. **Action handlers**: Tie specific functions directly to command-line flags
+5. **Subcommands**: Clean separation of complex operations
+6. **Less boilerplate**: Minimal code required to create new commands
+7. **Easy to understand**: Clear, linear code flow
+8. **Maintainable**: Reduced complexity makes the code easier to maintain and extend
+
+## Architecture Overview
+
+### Core Components
+
+- **`registry.py`**: Command registration system with decorators
+- **`utils.py`**: Common utilities and helpers for commands
+- **`base.py`**: Legacy base classes (deprecated, to be removed)
+- Individual command files: `chat.py`, `history.py`, `feedback.py`, `shell.py`
+- **`chat_new.py`**: Example showing new action handlers and subcommands
+- **`example.py`**: Complete demonstration of both patterns
+
+### Two Approaches for Handling Command Operations
+
+#### 1. Action Handlers (For Simple Flags)
+
+Perfect for simple flags that trigger specific actions. No more branching logic!
+
+```python
+def _handle_list(args: Namespace, context: CommandContext) -> int:
+ """Handler for --list flag."""
+ utils = create_utils(context)
+ # Handle list operation
+ return 0
+
+@command("mycommand", help="My command")
+@argument("--list", action="store_true", help="List items", handler=_handle_list)
+def my_command(args: Namespace, context: CommandContext) -> int:
+ """Main command - only handles default behavior."""
+ # This only runs when --list is NOT used
+ return 0
+```
+
+#### 2. Subcommands (For Complex Operations)
+
+Perfect for complex operations with their own arguments:
+
+```python
+@subcommand("mycommand", "create", help="Create a new item")
+@argument("name", help="Name of the item")
+@argument("--type", help="Type of item")
+def create_item(args: Namespace, context: CommandContext) -> int:
+ """Completely separate function for create operation."""
+ utils = create_utils(context)
+ # Handle create operation
+ return 0
+
+# Usage: mycommand create "item-name" --type "special"
+```
+
+## Creating a New Command
+
+You have three approaches depending on your needs:
+
+### Approach 1: Simple Command (Traditional)
+
+```python
+@command("newcmd", help="My new command")
+@argument("message", help="Message to display")
+def new_command(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ utils.render_success(args.message)
+ return 0
+```
+
+### Approach 2: With Action Handlers (Eliminates Branching)
+
+```python
+def _handle_uppercase(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ utils.render_success(args.message.upper())
+ return 0
+
+@command("newcmd", help="My new command")
+@argument("message", help="Message to display")
+@argument("-u", "--uppercase", action="store_true", help="Uppercase", handler=_handle_uppercase)
+def new_command(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ # This only runs when --uppercase is NOT used
+ utils.render_success(args.message)
+ return 0
+```
+
+### Approach 3: With Subcommands (Complex Operations)
+
+```python
+@subcommand("newcmd", "process", help="Process the message")
+@argument("message", help="Message to process")
+@argument("--method", choices=["upper", "lower"], help="Processing method")
+def process_message(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ if args.method == "upper":
+ result = args.message.upper()
+ else:
+ result = args.message.lower()
+ utils.render_success(result)
+ return 0
+
+@command("newcmd", help="My new command with subcommands")
+def new_command(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ utils.render_success("Use 'newcmd process <message>' to process messages")
+ return 0
+
+# Usage: newcmd process "Hello" --method upper
+```
+
+### Step 2: Import in `__init__.py`
+
+Add your command module to the imports in `__init__.py`:
+
+```python
+from command_line_assistant.commands import chat, feedback, history, shell, newcmd
+```
+
+That's it! No factories, no complex registration, no multiple inheritance.
+
+## Available Utilities
+
+The `CommandUtils` class provides common functionality:
+
+### Renderers
+- `utils.render_success(message)` - Success messages
+- `utils.render_warning(message)` - Warning messages
+- `utils.render_error(message)` - Error messages
+- `utils.text_renderer` - Direct access to text renderer
+- `utils.warning_renderer` - Direct access to warning renderer
+- `utils.error_renderer` - Direct access to error renderer
+
+### D-Bus Proxies
+- `utils.chat_proxy` - Chat interface proxy
+- `utils.history_proxy` - History interface proxy
+- `utils.user_proxy` - User interface proxy
+
+### Helper Methods
+- `utils.get_user_id()` - Get current user ID
+- `utils.context` - Access to command context
+
+## Migration from Old Structure
+
+### Old Structure (Complex with Branching)
+```python
+class MyOperationType(CommandOperationType):
+ LIST = auto()
+ DELETE = auto()
+ CREATE = auto()
+
+class MyOperationFactory(CommandOperationFactory):
+ _arg_to_operation = {
+ "list": MyOperationType.LIST,
+ "delete": MyOperationType.DELETE,
+ "create": MyOperationType.CREATE,
+ }
+
+@MyOperationFactory.register(MyOperationType.LIST)
+class ListOperation(BaseOperation):
+ def execute(self) -> None:
+ # List logic
+ pass
+
+# ... more operation classes ...
+
+class MyCommand(BaseCLICommand):
+ def run(self) -> int:
+ factory = MyOperationFactory()
+ operation = factory.create_operation(self._args, self._context)
+ if operation:
+ operation.execute()
+ return 0
+
+def register_subcommand(parser):
+ # Complex argument setup...
+ pass
+```
+
+### New Structure Option 1: Action Handlers
+```python
+def _handle_list(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ # List logic here
+ return 0
+
+def _handle_delete(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ # Delete logic here
+ return 0
+
+@command("mycommand", help="My command")
+@argument("--list", action="store_true", help="List items", handler=_handle_list)
+@argument("--delete", help="Delete item", handler=_handle_delete)
+def my_command(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ # Default behavior only - no branching!
+ return 0
+```
+
+### New Structure Option 2: Subcommands
+```python
+@subcommand("mycommand", "list", help="List items")
+def list_items(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ # List logic here
+ return 0
+
+@subcommand("mycommand", "delete", help="Delete an item")
+@argument("item_name", help="Name of item to delete")
+def delete_item(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ # Delete logic here
+ return 0
+
+@command("mycommand", help="My command")
+def my_command(args: Namespace, context: CommandContext) -> int:
+ utils = create_utils(context)
+ # Default behavior when no subcommand used
+ return 0
+
+# Usage: mycommand list, mycommand delete item-name
+```
+
+## Decorator Details
+
+### @command(name, help=None, description=None)
+Registers a function as a CLI command.
+
+- `name`: Command name (required)
+- `help`: Short help text (optional)
+- `description`: Long description (optional)
+
+### @argument(*args, handler=None, **kwargs)
+Adds an argument to the command. Uses the same signature as `argparse.add_argument()`.
+
+- `handler`: Optional function to handle this specific argument (eliminates branching)
+- All other kwargs are passed to `argparse.add_argument()`
+
+**Note**: Arguments are applied in reverse order (decorators are bottom-up), so the last `@argument` decorator will be the first argument added to the parser.
+
+### @subcommand(parent_command, name, help=None, description=None)
+Registers a function as a subcommand of an existing command.
+
+- `parent_command`: Name of parent command (required)
+- `name`: Subcommand name (required)
+- `help`: Short help text (optional)
+- `description`: Long description (optional)
+
+Subcommands can have their own `@argument` decorators for subcommand-specific arguments.
+
+## Real-World Examples
+
+### Example 1: Chat Command (chat_new.py)
+
+Shows both approaches in one command:
+
+```python
+# Action handler for simple flags
+def _handle_list(args: Namespace, context: CommandContext) -> int:
+ return list_chats(args, context)
+
+# Subcommands for complex operations
+@subcommand("chat", "delete", help="Delete a specific chat")
+@argument("chat_name", help="Name of the chat to delete")
+def delete_chat(args: Namespace, context: CommandContext) -> int:
+ # Implementation here
+ pass
+
+@command("chat", help="Command to ask a question to the LLM")
+@argument("query_string", nargs="?", help="The question", default="")
+@argument("-l", "--list", action="store_true", help="List chats", handler=_handle_list)
+def chat_command(args: Namespace, context: CommandContext) -> int:
+ # Only handles direct questions - no branching!
+ pass
+```
+
+### Example 2: Complete Demo (example.py)
+
+See `example.py` for a complete example showing:
+- Action handlers for simple flags
+- Subcommands for complex operations
+- Positional and optional arguments
+- Different argument types
+- Error handling
+- Using various utilities
+
+Usage examples:
+- `example "Hello World"` - Main command
+- `example --uppercase "hello"` - Action handler
+- `example greet Alice --formal` - Subcommand
+- `example count 10 --step 2` - Subcommand with args
+
+## Error Handling
+
+Commands should:
+1. Use try/except blocks for error handling
+2. Use appropriate exception types (e.g., `ChatCommandException`)
+3. Render errors using `utils.render_error()`
+4. Return appropriate exit codes (0 for success, non-zero for failure)
+
+## Best Practices
+
+### General
+1. **Choose the right pattern**:
+ - Action handlers for simple flags that trigger one action
+ - Subcommands for complex operations with their own arguments
+ - Traditional approach for simple commands without branching
+2. **Keep functions focused**: Each function should do one thing well
+3. **Use utilities**: Leverage `CommandUtils` for common operations
+4. **Handle errors gracefully**: Always catch and handle exceptions properly
+5. **Return appropriate exit codes**: 0 for success, non-zero for errors
+6. **Use descriptive help text**: Make commands self-documenting
+
+### Action Handlers
+1. **Use for simple flags**: Perfect for `--list`, `--delete`, `--help` style flags
+2. **Keep handlers simple**: Each handler should be a focused function
+3. **Name handlers clearly**: Use `_handle_<action>` naming convention
+4. **Avoid complex logic**: If it needs lots of arguments, consider subcommands
+
+### Subcommands
+1. **Use for complex operations**: Perfect for operations that need their own arguments
+2. **Group related functionality**: Keep related subcommands in the same file
+3. **Make subcommands independent**: Each should be a complete, standalone function
+4. **Use clear names**: Subcommand names should be action verbs
+
+### Migration Strategy
+1. **Start with action handlers**: Convert simple if/else branching first
+2. **Move to subcommands**: For complex operations with many arguments
+3. **Keep main command simple**: Should only handle default behavior
+4. **Test incrementally**: Migrate one operation at a time
+
+## Legacy Support
+
+The old base classes and factory pattern are still available but deprecated. New commands should use the simplified structure. The old system will be removed in a future version.
\ No newline at end of file
diff --git a/command_line_assistant/commands/__init__.py b/command_line_assistant/commands/__init__.py
index 6c0d154..073f55c 100644
--- a/command_line_assistant/commands/__init__.py
+++ b/command_line_assistant/commands/__init__.py
@@ -1 +1,7 @@
-"""Module to track all the CLI commands."""
+"""Simplified command registration module."""
+
+# Import all command modules to register them
+from command_line_assistant.commands import chat, feedback, history, shell
+from command_line_assistant.commands.registry import register_all_commands
+
+__all__ = ["register_all_commands"]
diff --git a/command_line_assistant/commands/chat.py b/command_line_assistant/commands/chat.py
index 71fe9d8..a9c7368 100644
--- a/command_line_assistant/commands/chat.py
+++ b/command_line_assistant/commands/chat.py
@@ -1,26 +1,18 @@
-"""Module to handle the chat command."""
+"""Simplified chat command implementation."""
import argparse
import logging
import platform
from argparse import Namespace
-from enum import auto
from io import TextIOWrapper
-from typing import ClassVar, Optional
+from typing import Optional
-from command_line_assistant.commands.base import (
- BaseCLICommand,
- BaseOperation,
- CommandOperationFactory,
- CommandOperationType,
-)
+from command_line_assistant.commands.registry import argument, command
+from command_line_assistant.commands.utils import create_utils
from command_line_assistant.dbus.exceptions import (
ChatNotFoundError,
HistoryNotEnabledError,
)
-from command_line_assistant.dbus.interfaces.chat import ChatInterface
-from command_line_assistant.dbus.interfaces.history import HistoryInterface
-from command_line_assistant.dbus.interfaces.user import UserInterface
from command_line_assistant.dbus.structures.chat import (
AttachmentInput,
ChatList,
@@ -32,31 +24,20 @@ from command_line_assistant.dbus.structures.chat import (
)
from command_line_assistant.exceptions import ChatCommandException, StopInteractiveMode
from command_line_assistant.rendering.decorators.colors import ColorDecorator
-from command_line_assistant.rendering.decorators.text import (
- WriteOncePerSessionDecorator,
-)
-from command_line_assistant.rendering.renders.interactive import InteractiveRenderer
-from command_line_assistant.rendering.renders.markdown import MarkdownRenderer
-from command_line_assistant.rendering.renders.spinner import SpinnerRenderer
-from command_line_assistant.rendering.renders.text import TextRenderer
+from command_line_assistant.rendering.decorators.text import WriteOncePerSessionDecorator
from command_line_assistant.terminal.parser import (
find_output_by_index,
parse_terminal_output,
)
from command_line_assistant.terminal.reader import TERMINAL_CAPTURE_FILE
from command_line_assistant.utils.benchmark import TimingLogger
-from command_line_assistant.utils.cli import (
- CommandContext,
- SubParsersAction,
-)
+from command_line_assistant.utils.cli import CommandContext
from command_line_assistant.utils.files import NamedFileLock, guess_mimetype
from command_line_assistant.utils.renderers import (
- create_error_renderer,
create_interactive_renderer,
create_markdown_renderer,
create_spinner_renderer,
create_text_renderer,
- create_warning_renderer,
format_datetime,
human_readable_size,
)
@@ -67,64 +48,41 @@ timing = TimingLogger(
filtered_params=[
"question",
"stdin",
- "attachment",
+ "attachment",
"attachment_mimetype",
"last_output",
]
)
-#: Max input size we want to allow to be submitted to the backend. This
-#: corresponds to 2KB (2048 bytes)
+# Constants
MAX_QUESTION_SIZE: int = 2048
-
-#: Legal notice that we need to output once per user
LEGAL_NOTICE = (
"This feature uses AI technology. Do not include any personal information or "
"other sensitive information in your input. Interactions may be used to "
"improve Red Hat's products or services."
)
-#: Always good to have legal message.
ALWAYS_LEGAL_MESSAGE = "Always review AI-generated content prior to use."
-
-#: Default chat description when none is given
DEFAULT_CHAT_DESCRIPTION = "Default Command Line Assistant Chat."
-
-#: Default chat name when none is given
DEFAULT_CHAT_NAME = "default"
def _read_last_terminal_output(index: int) -> str:
- """Internal function to handle reading the last terminal output.
-
- Arguments:
- index (int): The index to grab the output from the list
-
- Returns:
- str: The data read or an empty string
- """
+ """Read the last terminal output by index."""
logger.info("Reading terminal output.")
contents = parse_terminal_output()
-
+
if not contents:
logger.info("No contents found during reading the terminal output.")
return ""
-
- last_output = find_output_by_index(index=index, output=contents)
- return last_output
+
+ return find_output_by_index(index=index, output=contents)
def _parse_attachment_file(attachment: Optional[TextIOWrapper] = None) -> str:
- """Parse the attachment file and read its contents.
-
- Arguments:
- attachment (Optional[TextIOWrapper], optional): The attachment that will be parsed
-
- Returns:
- str: Either the str read or None.
- """
+ """Parse attachment file and read its contents."""
if not attachment:
return ""
-
+
try:
return attachment.read().strip()
except UnicodeDecodeError as e:
@@ -134,669 +92,387 @@ def _parse_attachment_file(attachment: Optional[TextIOWrapper] = None) -> str:
def _get_input_source(query: str, stdin: str, attachment: str, last_output: str) -> str:
- """
- Determine and return the appropriate input source based on combination rules.
-
- Arguments:
- query (str): The question to be asked
- stdin (str): The input redirect via stdin
- attachment (str): The attachment contents
- attachment_mimetype (str): The mimetype of the attachment
- last_output (str): The last out read from the terminal
-
- Warning:
- This is set to be deprecated in the future when we normalize the API
- backend to accept the context and works with it.
-
- Rules:
- 1. Positional query only -> use positional query
- 2. Stdin query only -> use stdin query
- 3. File query only -> use file query
- 4. Stdin + positional query -> combine as "{positional_query} {stdin}"
- 5. Stdin + file query -> combine as "{stdin} {file_query}"
- 6. Positional + file query -> combine as "{positional_query} {file_query}"
- 7. Positional + last output -> combine as "{positional_query} {last_output}"
- 8. Positional + attachment + last output -> combine as "{positional_query} {attachment} {last_output}"
- 99. All three sources -> use only positional and file as "{positional_query} {file_query}"
-
- Raises:
- ValueError: If no input source is provided
-
- Returns:
- str: The query string from the selected input source(s)
- """
- # Rule 99: All present - positional and file take precedence
+ """Determine and return the appropriate input source based on combination rules."""
+ # All present - positional and file take precedence
if all([query, stdin, attachment, last_output]):
logger.debug("Using positional query and file input. Stdin will be ignored.")
return f"{query} {attachment}"
-
- # Rule 8: positional + attachment + last output
+
+ # Positional + attachment + last output
if query and attachment and last_output:
- logger.info(
- "Positional query, attachment and last output found. Using all of them at once."
- )
+ logger.info("Positional query, attachment and last output found. Using all of them at once.")
return f"{query} {attachment} {last_output}"
-
- # Rule 7: positional + last_output
+
+ # Positional + last_output
if query and last_output:
logger.info("Positional query and last output found. Using them.")
return f"{query} {last_output}"
-
- # Rule 6: Positional + file
+
+ # Positional + file
if query and attachment:
logger.info("Positional query and attachment found. Using them.")
return f"{query} {attachment}"
-
- # Rule 5: Stdin + file
+
+ # Stdin + file
if stdin and attachment:
logger.info("stdin and attachment found. Using them.")
return f"{stdin} {attachment}"
-
- # Rule 4: Stdin + positional
+
+ # Stdin + positional
if stdin and query:
- logger.info("Positional query and stidn found. Using them.")
+ logger.info("Positional query and stdin found. Using them.")
return f"{query} {stdin}"
-
- # Rules 1-3: Single source - return first non-empty source
- logger.info(
- "Defaulting to use any of positional query, stdin, attachment or last output since no combinations where provided."
- )
+
+ # Single source - return first non-empty source
+ logger.info("Defaulting to use any of positional query, stdin, attachment or last output since no combinations where provided.")
source = next(
(src for src in [query, stdin, attachment, last_output] if src),
None,
)
-
+
if source:
return source
-
+
logger.error("Couldn't find a match.")
raise ValueError(
"No input provided. Please provide input via file, stdin, or direct query."
)
-class ChatOperationType(CommandOperationType):
- """Enum to control the operations for the command"""
-
- LIST_CHATS = auto()
- DELETE_CHAT = auto()
- DELETE_ALL_CHATS = auto()
- INTERACTIVE_CHAT = auto()
- SINGLE_QUESTION = auto()
+def _create_chat_session(utils, user_id: str, name: str, description: str) -> str:
+ """Create a new chat session for a given conversation."""
+ has_chat_id = None
+ try:
+ has_chat_id = utils.chat_proxy.GetChatId(user_id, name)
+ except ChatNotFoundError:
+ # It's okay to swallow this exception as if there is no chat for
+ # this user, we will create one.
+ pass
+
+ # To avoid doing this check inside the CreateChat method, let's do it
+ # in here.
+ if has_chat_id:
+ return has_chat_id
+
+ return utils.chat_proxy.CreateChat(user_id, name, description)
-class ChatOperationFactory(CommandOperationFactory):
- """Factory for creating shell operations with decorator-based registration"""
-
- # Mapping of CLI arguments to operation types
- _arg_to_operation: ClassVar[dict[str, CommandOperationType]] = {
- "list": ChatOperationType.LIST_CHATS,
- "delete": ChatOperationType.DELETE_CHAT,
- "delete_all": ChatOperationType.DELETE_ALL_CHATS,
- "interactive": ChatOperationType.INTERACTIVE_CHAT,
- "query_string": ChatOperationType.SINGLE_QUESTION,
- "stdin": ChatOperationType.SINGLE_QUESTION,
- "attachment": ChatOperationType.SINGLE_QUESTION,
- "with_output": ChatOperationType.SINGLE_QUESTION,
- }
+def _display_response(response: str) -> None:
+ """Display message to the terminal."""
+ legal_renderer = create_text_renderer(
+ decorators=[
+ ColorDecorator(foreground="lightyellow"),
+ WriteOncePerSessionDecorator(state_filename="legal"),
+ ]
+ )
+ notice_renderer = create_text_renderer(
+ decorators=[ColorDecorator(foreground="lightyellow")]
+ )
+ markdown_renderer = create_markdown_renderer()
+
+ legal_renderer.render(LEGAL_NOTICE)
+ markdown_renderer.render(response)
+ notice_renderer.render(ALWAYS_LEGAL_MESSAGE)
-class BaseChatOperation(BaseOperation):
- """Base class for handling chat operations
-
- Attributes:
- spinner_renderer (SpinnerRenderer): The instance of a spinner renderer
- legal_renderer (TextRenderer): Instance of text renderer to show legal message
- notice_renderer (TextRenderer): Instance of text renderer to show notice message
- interactive_renderer (InteractiveRenderer): Instance of interactive renderer to handle interactive mode
- """
-
- def __init__(
- self,
- text_renderer: TextRenderer,
- warning_renderer: TextRenderer,
- error_renderer: TextRenderer,
- args: Namespace,
- context: CommandContext,
- chat_proxy: ChatInterface,
- history_proxy: HistoryInterface,
- user_proxy: UserInterface,
- ) -> None:
- """Constructor of the class.
-
- Arguments:
- text_renderer (TextRenderer): Instance of text renderer class
- warning_renderer (TextRenderer): Instance of text renderer class
- error_renderer (TextRenderer): Instance of text renderer class
- args (Namespace): The arguments from CLI
- context (CommandContext): Context for the commands
- chat_proxy (ChatInterface): The proxy object for dbus chat
- history_proxy (HistoryInterface): The proxy object for dbus history
- user_proxy (HistoryInterface): The proxy object for dbus user
- """
- super().__init__(
- text_renderer,
- warning_renderer,
- error_renderer,
- args,
- context,
- chat_proxy,
- history_proxy,
- user_proxy,
+@timing.timeit
+def _submit_question(
+ utils,
+ user_id: str,
+ chat_id: str,
+ question: str,
+ stdin: str,
+ attachment: str,
+ attachment_mimetype: str,
+ last_output: str,
+ skip_spinner: Optional[bool] = False,
+) -> str:
+ """Submit the question over dbus."""
+ final_question = _get_input_source(question, stdin, attachment, last_output)
+
+ if len(final_question) > MAX_QUESTION_SIZE:
+ final_question_size = human_readable_size(len(final_question))
+ max_question_size = human_readable_size(MAX_QUESTION_SIZE)
+ utils.render_warning(
+ f"The total size of your question and context ({final_question_size}) exceeds the limit of {max_question_size}. Trimming it down to fit in the expected size, you may lose some context."
)
- self.spinner_renderer: SpinnerRenderer = create_spinner_renderer(
- message="Asking RHEL Lightspeed",
+ logger.debug(
+ "Total size of question (%s) exceeds defined limit of %s.",
+ len(final_question),
+ MAX_QUESTION_SIZE,
)
- self.legal_renderer: TextRenderer = create_text_renderer(
- decorators=[
- ColorDecorator(foreground="lightyellow"),
- WriteOncePerSessionDecorator(state_filename="legal"),
- ]
+ final_question = final_question[:MAX_QUESTION_SIZE]
+ logger.debug(
+ "Final size of question after the limit %s.", len(final_question)
)
- self.notice_renderer: TextRenderer = create_text_renderer(
- decorators=[ColorDecorator(foreground="lightyellow")]
+
+ spinner_renderer = create_spinner_renderer(message="Asking RHEL Lightspeed")
+
+ if skip_spinner:
+ response = _get_response(utils, user_id, final_question, stdin, attachment, attachment_mimetype, last_output)
+ else:
+ with spinner_renderer:
+ response = _get_response(utils, user_id, final_question, stdin, attachment, attachment_mimetype, last_output)
+
+ try:
+ utils.history_proxy.WriteHistory(chat_id, user_id, final_question, response)
+ except HistoryNotEnabledError:
+ logger.warning(
+ "The history is disabled in the configuration file. Skipping the write to the history."
)
- self.interactive_renderer: InteractiveRenderer = create_interactive_renderer()
- self.markdown_renderer: MarkdownRenderer = create_markdown_renderer()
+
+ return response
- def _display_response(self, response: str) -> None:
- """Internal method to display message to the terminal
- Arguments:
- response(str): The message to be displayed
- """
- self.legal_renderer.render(LEGAL_NOTICE)
- self.markdown_renderer.render(response)
- self.notice_renderer.render(ALWAYS_LEGAL_MESSAGE)
+@timing.timeit
+def _get_response(
+ utils,
+ user_id: str,
+ question: str,
+ stdin: str,
+ attachment: str,
+ attachment_mimetype: str,
+ last_output: str,
+) -> str:
+ """Get the response from the chat session."""
+ message_input = Question(
+ message=question,
+ stdin=StdinInput(stdin=stdin),
+ attachment=AttachmentInput(
+ contents=attachment, mimetype=attachment_mimetype
+ ),
+ terminal=TerminalInput(output=last_output),
+ systeminfo=SystemInfo(
+ os=utils.context.os_release["name"],
+ version=utils.context.os_release["version_id"],
+ arch=platform.machine(),
+ id=utils.context.os_release["id"],
+ ),
+ )
+ response = utils.chat_proxy.AskQuestion(user_id, message_input.structure())
+
+ return Response.from_structure(response).message
- @timing.timeit
- def _submit_question(
- self,
- user_id: str,
- chat_id: str,
- question: str,
- stdin: str,
- attachment: str,
- attachment_mimetype: str,
- last_output: str,
- skip_spinner: Optional[bool] = False,
- ) -> str:
- """Submit the question over dbus.
- Arguments:
- user_id (str): The unique identifier for the user
- chat_id (str): The unique identifier for the chat
- question (str): The question to be asked
- stdin (str): The input redirect via stdin
- attachment (str): The attachment contents
- attachment_mimetype (str): The mimetype of the attachment
- last_output (str): The last out read from the terminal
- skip_spinner (Optional[bool], optional): Skip the spinner renderer. Defaults to False.
-
- Returns:
- str: The response from the backend server
- """
- final_question = _get_input_source(question, stdin, attachment, last_output)
- response = None
-
- if len(final_question) > MAX_QUESTION_SIZE:
- final_question_size = human_readable_size(len(final_question))
- max_question_size = human_readable_size(MAX_QUESTION_SIZE)
- self.warning_renderer.render(
- f"The total size of your question and context ({final_question_size}) exceeds the limit of {max_question_size}. Trimming it down to fit in the expected size, you may lose some context."
+# Command implementations
+@command("chat", help="Command to ask a question to the LLM")
+@argument("query_string", nargs="?", help="The question that will be sent to the LLM", default="")
+@argument("-a", "--attachment", nargs="?", type=argparse.FileType("r"), help="File attachment to be read and sent alongside the query")
+@argument("-i", "--interactive", action="store_true", help="Start an interactive chat session")
+@argument("-w", "--with-output", nargs="?", type=int, help="Add output from terminal as context for the query. Use 1 to retrieve the latest output, 2 to and so on. First, enable the terminal capture with 'c shell --enable-capture' for this option to work.")
+@argument("-r", "--raw", action="store_true", help="Interact with the backend in raw mode. No spinners, emojis or anything.")
+@argument("-l", "--list", action="store_true", help="List all chats")
+@argument("-d", "--delete", nargs="?", default="", help="Delete a chat session. Specify the chat session by its name.")
+@argument("--delete-all", action="store_true", help="Delete all chats")
+@argument("-n", "--name", nargs="?", help="Give a name to the chat session. Parameter has to be used together with sending a query. Otherwise has no effect.")
+@argument("--description", nargs="?", help="Give a description to the chat session. Parameter has to be used together with sending a query. Otherwise has no effect.")
+def chat_command(args: Namespace, context: CommandContext) -> int:
+ """Main chat command implementation."""
+ utils = create_utils(context)
+
+ try:
+ # Handle special arguments preprocessing
+ if args.with_output:
+ logger.debug("Converting the index to a negative number in order to reverse search the output list.")
+ logger.debug("Original index is %s", args.with_output)
+ args.with_output = -abs(args.with_output)
+
+ # Set default name and description
+ if not args.description and not args.name:
+ args.description = DEFAULT_CHAT_DESCRIPTION
+ args.name = DEFAULT_CHAT_NAME
+ logger.debug("No name or description provided. Using default values.")
+
+ if not args.description and args.name:
+ args.description = DEFAULT_CHAT_DESCRIPTION
+ utils.render_warning(
+ "Chat description not provided. Using the default description: "
+ f"'{DEFAULT_CHAT_DESCRIPTION}'. You can specify a custom description using the '--description' option."
)
- logger.debug(
- "Total size of question (%s) exceeds defined limit of %s.",
- len(final_question),
- MAX_QUESTION_SIZE,
- )
- final_question = final_question[:MAX_QUESTION_SIZE]
- logger.debug(
- "Final size of question after the limit %s.", len(final_question)
- )
-
- if skip_spinner:
- response = self._get_response(
- user_id=user_id,
- question=final_question,
- stdin=stdin,
- attachment=attachment,
- attachment_mimetype=attachment_mimetype,
- last_output=last_output,
+
+ if not args.name and args.description:
+ args.name = DEFAULT_CHAT_NAME
+ utils.render_warning(
+ "Chat name not provided. Using the default name: "
+ f"'{DEFAULT_CHAT_NAME}'. You can specify a custom name using the '--name' option."
)
+
+ # Handle different operations
+ if args.list:
+ return _list_chats(utils)
+ elif args.delete:
+ return _delete_chat(utils, args.delete)
+ elif args.delete_all:
+ return _delete_all_chats(utils)
+ elif args.interactive:
+ return _interactive_chat(utils, args)
else:
- with self.spinner_renderer:
- response = self._get_response(
- user_id=user_id,
- question=final_question,
- stdin=stdin,
- attachment=attachment,
- attachment_mimetype=attachment_mimetype,
- last_output=last_output,
- )
+ return _single_question(utils, args)
+
+ except ChatCommandException as e:
+ logger.info("Failed to execute chat command: %s", str(e))
+ utils.render_error(str(e))
+ return e.code
- try:
- self.history_proxy.WriteHistory(chat_id, user_id, final_question, response)
- except HistoryNotEnabledError:
- logger.warning(
- "The history is disabled in the configuration file. Skipping the write to the history."
- )
- return response
-
- @timing.timeit
- def _get_response(
- self,
- user_id: str,
- question: str,
- stdin: str,
- attachment: str,
- attachment_mimetype: str,
- last_output: str,
- ) -> str:
- """Get the response from the chat session.
-
- Arguments:
- user_id (str): The user identifier
- chat_id (str): The chat session identifier
- question (str): The question to be asked
- stdin (str): The input redirect via stdin
- attachment (str): The attachment contents
- attachment_mimetype (str): The mimetype of the attachment
- last_output (str): The last out read from the terminal
-
- Returns:
- str: The response from the chat session
- """
- message_input = Question(
- message=question,
- stdin=StdinInput(stdin=stdin),
- attachment=AttachmentInput(
- contents=attachment, mimetype=attachment_mimetype
- ),
- terminal=TerminalInput(output=last_output),
- systeminfo=SystemInfo(
- os=self.context.os_release["name"],
- version=self.context.os_release["version_id"],
- arch=platform.machine(),
- id=self.context.os_release["id"],
- ),
+def _list_chats(utils) -> int:
+ """List all chats operation."""
+ user_id = utils.get_user_id()
+ all_chats = ChatList.from_structure(utils.chat_proxy.GetAllChatFromUser(user_id))
+
+ if not all_chats.chats:
+ utils.render_success("No chats available.")
+ return 0
+
+ utils.render_success(f"Found a total of {len(all_chats.chats)} chats:")
+ for index, chat in enumerate(all_chats.chats):
+ created_at = format_datetime(chat.created_at)
+ utils.render_success(
+ f"{index}. Chat: {chat.name} - {chat.description} (created at: {created_at})"
)
- response = self.chat_proxy.AskQuestion(
- user_id,
- message_input.structure(),
+ return 0
+
+
+def _delete_chat(utils, chat_name: str) -> int:
+ """Delete a specific chat operation."""
+ try:
+ user_id = utils.get_user_id()
+ utils.chat_proxy.DeleteChatForUser(user_id, chat_name)
+ utils.render_success(f"Chat {chat_name} deleted successfully.")
+ return 0
+ except ChatNotFoundError as e:
+ raise ChatCommandException(
+ f"Failed to delete requested chat {str(e)}"
+ ) from e
+
+
+def _delete_all_chats(utils) -> int:
+ """Delete all chats operation."""
+ try:
+ user_id = utils.get_user_id()
+ utils.chat_proxy.DeleteAllChatForUser(user_id)
+ utils.render_success("Deleted all chats successfully.")
+ return 0
+ except ChatNotFoundError as e:
+ raise ChatCommandException(
+ f"Failed to delete all requested chats {str(e)}"
+ ) from e
+
+
+def _interactive_chat(utils, args) -> int:
+ """Interactive chat operation."""
+ terminal_file_lock = NamedFileLock(name="terminal")
+
+ if terminal_file_lock.is_locked:
+ raise ChatCommandException(
+ f"Detected a terminal capture session running with pid '{terminal_file_lock.pid}'."
+ " Interactive chat mode is not available while terminal capture is active, you must stop the previous one."
)
-
- return Response.from_structure(response).message
-
- def _create_chat_session(self, user_id: str, name: str, description: str) -> str:
- """Create a new chat session for a given conversation.
-
- Arguments:
- user_id (str): The user identifier
- name (str): The name of the chat
- description (str): The description of the chat
-
- Returns:
- str: The identifier of the chat session.
- """
- has_chat_id = None
- try:
- has_chat_id = self.chat_proxy.GetChatId(user_id, name)
- except ChatNotFoundError:
- # It's okay to swallow this exception as if there is no chat for
- # this user, we will create one.
- pass
-
- # To avoid doing this check inside the CreateChat method, let's do it
- # in here.
- if has_chat_id:
- return has_chat_id
-
- return self.chat_proxy.CreateChat(
- user_id,
- name,
- description,
- )
-
-
-@ChatOperationFactory.register(ChatOperationType.LIST_CHATS)
-class ListChatsOperation(BaseChatOperation):
- """Class that holds the list operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- all_chats = ChatList.from_structure(self.chat_proxy.GetAllChatFromUser(user_id))
- if not all_chats.chats:
- self.text_renderer.render("No chats available.")
-
- self.text_renderer.render(f"Found a total of {len(all_chats.chats)} chats:")
- for index, chat in enumerate(all_chats.chats):
- created_at = format_datetime(chat.created_at)
- self.text_renderer.render(
- f"{index}. Chat: {chat.name} - {chat.description} (created at: {created_at})"
- )
-
-
-@ChatOperationFactory.register(ChatOperationType.DELETE_CHAT)
-class DeleteChatOperation(BaseChatOperation):
- """Class that holds the delete operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- try:
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- self.chat_proxy.DeleteChatForUser(user_id, self.args.delete)
- self.text_renderer.render(f"Chat {self.args.delete} deleted successfully.")
- except ChatNotFoundError as e:
- raise ChatCommandException(
- f"Failed to delete requested chat {str(e)}"
- ) from e
-
-
-@ChatOperationFactory.register(ChatOperationType.DELETE_ALL_CHATS)
-class DeleteAllChatsOperation(BaseChatOperation):
- """Class that holds the delete all operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- try:
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- self.chat_proxy.DeleteAllChatForUser(user_id)
- self.text_renderer.render("Deleted all chats successfully.")
- except ChatNotFoundError as e:
- raise ChatCommandException(
- f"Failed to delete all requested chats {str(e)}"
- ) from e
-
-
-@ChatOperationFactory.register(ChatOperationType.INTERACTIVE_CHAT)
-class InteractiveChatOperation(BaseChatOperation):
- """Class that initiates the interactive mode"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
-
- terminal_file_lock = NamedFileLock(name="terminal")
-
- if terminal_file_lock.is_locked:
- raise ChatCommandException(
- f"Detected a terminal capture session running with pid '{terminal_file_lock.pid}'."
- " Interactive chat mode is not available while terminal capture is active, you must stop the previous one."
- )
-
- try:
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- chat_id = self._create_chat_session(
- user_id, self.args.name, self.args.description
- )
- attachment = _parse_attachment_file(self.args.attachment)
- attachment_mimetype = guess_mimetype(self.args.attachment)
- stdin = self.args.stdin
-
- while True:
- self.interactive_renderer.render(">>> ")
- question = self.interactive_renderer.output
- if not question:
- self.error_renderer.render(
- "Your question can't be empty. Please, try again."
- )
- continue
- response = self._submit_question(
- user_id=user_id,
- chat_id=chat_id,
- question=question,
- stdin=stdin,
- attachment=attachment,
- attachment_mimetype=attachment_mimetype,
- # For now, we won't deal with last output in interactive mode.
- last_output="",
- skip_spinner=self.args.raw,
- )
- self._display_response(response)
- except (KeyboardInterrupt, EOFError) as e:
- raise ChatCommandException(
- "Detected keyboard interrupt. Stopping interactive mode."
- ) from e
- except StopInteractiveMode:
- return
-
-
-@ChatOperationFactory.register(ChatOperationType.SINGLE_QUESTION)
-class SingleQuestionOperation(BaseChatOperation):
- """Class that holds the single question ask operation"""
-
- def _validate_query(self):
- """Helper function to validate query.
-
- Raises:
- ChatCommandException: In case the query has invalid sizing (less than 1 character).
- """
- # If both are empty, raise exception
- if not self.args.query_string and not self.args.stdin:
- logger.debug("Both query_string and stdin are empty.")
- raise ChatCommandException(
- "Your query needs to have at least 2 characters. Either query or stdin are empty."
- )
-
- # If query_string has content but is too short
- if self.args.query_string and len(self.args.query_string.strip()) <= 1:
- logger.debug(
- "Query string has only 1 or 0 characters after stripping: '%s'",
- self.args.query_string,
- )
- raise ChatCommandException(
- "Your query needs to have at least 2 characters."
- )
-
- # If stdin has content but is too short
- if self.args.stdin and len(self.args.stdin.strip()) <= 1:
- logger.debug(
- "Stdin has only 1 or 0 characters after stripping: '%s'",
- self.args.stdin,
- )
- raise ChatCommandException(
- "Your stdin input needs to have at least 2 characters."
- )
-
- # If the user tries to do "c -w 1" or "c -w 1 "help me figure this out"
- # and there is no terminal capture log file, we just error out.
- if self.args.with_output and not TERMINAL_CAPTURE_FILE.exists():
- raise ChatCommandException(
- "Adding context from terminal output is only allowed if terminal capture is active."
- )
-
- @timing.timeit
- def execute(self) -> None:
- """Default method to execute the operation"""
-
- self._validate_query()
-
- try:
- last_terminal_output = ""
- if self.args.with_output:
- logger.debug(
- "Retrieving context from with-output parameter (index %s)",
- self.args.with_output,
- )
- last_terminal_output = _read_last_terminal_output(self.args.with_output)
-
- attachment = _parse_attachment_file(self.args.attachment)
- attachment_mimetype = guess_mimetype(self.args.attachment)
- stdin = self.args.stdin.strip() if self.args.stdin else ""
- question = self.args.query_string.strip() if self.args.query_string else ""
-
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- chat_id = self._create_chat_session(
- user_id, self.args.name, self.args.description
- )
- response = self._submit_question(
+
+ try:
+ user_id = utils.get_user_id()
+ chat_id = _create_chat_session(utils, user_id, args.name, args.description)
+ attachment = _parse_attachment_file(args.attachment)
+ attachment_mimetype = guess_mimetype(args.attachment)
+ stdin = args.stdin if hasattr(args, 'stdin') else ""
+
+ interactive_renderer = create_interactive_renderer()
+
+ while True:
+ interactive_renderer.render(">>> ")
+ question = interactive_renderer.output
+ if not question:
+ utils.render_error("Your question can't be empty. Please, try again.")
+ continue
+
+ response = _submit_question(
+ utils=utils,
user_id=user_id,
chat_id=chat_id,
question=question,
stdin=stdin,
attachment=attachment,
attachment_mimetype=attachment_mimetype,
- last_output=last_terminal_output,
- skip_spinner=self.args.raw,
+ last_output="", # For now, we won't deal with last output in interactive mode
+ skip_spinner=args.raw,
)
-
- self._display_response(response)
- except ValueError as e:
- raise ChatCommandException(
- f"Failed to get a response from LLM. {str(e)}"
- ) from e
+ _display_response(response)
+
+ except (KeyboardInterrupt, EOFError) as e:
+ raise ChatCommandException(
+ "Detected keyboard interrupt. Stopping interactive mode."
+ ) from e
+ except StopInteractiveMode:
+ return 0
+
+ return 0
-class ChatCommand(BaseCLICommand):
- """Class that represents the chat command."""
-
- def run(self) -> int:
- """Main entrypoint for the command to run.
-
- Returns:
- int: Status code of the execution
- """
- error_renderer = create_error_renderer()
- operation_factory = ChatOperationFactory()
- try:
- operation = operation_factory.create_operation(
- self._args,
- self._context,
- text_renderer=create_text_renderer(
- decorators=[ColorDecorator()],
- ),
- error_renderer=error_renderer,
- )
-
- if operation:
- operation.execute()
- return 0
- except ChatCommandException as e:
- logger.info("Failed to execute chat command: %s", str(e))
- error_renderer.render(str(e))
- return e.code
-
-
-def register_subcommand(parser: SubParsersAction) -> None:
- """
- Register this command to argparse so it's available for the root parser.
-
- Arguments:
- parser (SubParsersAction): Root parser to register command-specific arguments
- """
- chat_parser = parser.add_parser(
- "chat",
- help="Command to ask a question to the LLM.",
- )
-
- question_group = chat_parser.add_argument_group("Question Options")
- # Positional argument, required only if no optional arguments are provided
- question_group.add_argument(
- "query_string",
- nargs="?",
- help="The question that will be sent to the LLM",
- default="",
- )
- question_group.add_argument(
- "-a",
- "--attachment",
- nargs="?",
- type=argparse.FileType("r"),
- help="File attachment to be read and sent alongside the query",
- )
- question_group.add_argument(
- "-i",
- "--interactive",
- action="store_true",
- help="Start an interactive chat session",
- )
- question_group.add_argument(
- "-w",
- "--with-output",
- nargs="?",
- type=int,
- help=(
- "Add output from terminal as context for the query. Use 1 to retrieve "
- "the latest output, 2 to and so on. First, enable the terminal "
- "capture with 'c shell --enable-capture' for this option to work."
- ),
- )
- question_group.add_argument(
- "-r",
- "--raw",
- action="store_true",
- help="Interact with the backend in raw mode. No spinners, emojis or anything.",
- )
-
- chat_arguments = chat_parser.add_argument_group("Chat Options")
- chat_arguments.add_argument(
- "-l", "--list", action="store_true", help="List all chats"
- )
- chat_arguments.add_argument(
- "-d",
- "--delete",
- nargs="?",
- default="",
- help="Delete a chat session. Specify the chat session by its name.",
- )
- chat_arguments.add_argument(
- "--delete-all", action="store_true", help="Delete all chats"
- )
- chat_arguments.add_argument(
- "-n",
- "--name",
- nargs="?",
- help="Give a name to the chat session. Parameter has to be used together with sending a query. Otherwise has no effect.",
- )
- chat_arguments.add_argument(
- "--description",
- nargs="?",
- help="Give a description to the chat session. Parameter has to be used together with sending a query. Otherwise has no effect.",
- )
-
- chat_parser.set_defaults(func=_command_factory)
-
-
-def _command_factory(args: Namespace) -> ChatCommand:
- """Internal command factory to create the command class
-
- Arguments:
- args (Namespace): The arguments processed with argparse.
-
- Returns:
- ChatCommand: Return an instance of class
- """
- if args.with_output:
+def _single_question(utils, args) -> int:
+ """Single question operation."""
+ # Validate query
+ if not args.query_string and not getattr(args, 'stdin', None):
+ logger.debug("Both query_string and stdin are empty.")
+ raise ChatCommandException(
+ "Your query needs to have at least 2 characters. Either query or stdin are empty."
+ )
+
+ if args.query_string and len(args.query_string.strip()) <= 1:
logger.debug(
- "Converting the index to a negative number in order to reverse search the output list."
+ "Query string has only 1 or 0 characters after stripping: '%s'",
+ args.query_string,
)
- logger.debug("Original index is %s", args.with_output)
- args.with_output = -abs(args.with_output)
-
- warning_renderer = create_warning_renderer()
-
- # Overriding the default description in case the user has not given us any.
- # We don't log this as warning to avoid spamming the user terminal with
- # this message.
- if not args.description and not args.name:
- args.description = DEFAULT_CHAT_DESCRIPTION
- args.name = DEFAULT_CHAT_NAME
- logger.debug("No name or description provided. Using default values.")
-
- if not args.description and args.name:
- args.description = DEFAULT_CHAT_DESCRIPTION
- warning_renderer.render(
- "Chat description not provided. Using the default description: "
- f"'{DEFAULT_CHAT_DESCRIPTION}'. You can specify a custom description using the '--description' option."
+ raise ChatCommandException(
+ "Your query needs to have at least 2 characters."
)
-
- if not args.name and args.description:
- args.name = DEFAULT_CHAT_NAME
- warning_renderer.render(
- "Chat name not provided. Using the default name: "
- f"'{DEFAULT_CHAT_NAME}'. You can specify a custom name using the '--name' option."
+
+ stdin = getattr(args, 'stdin', '') or ''
+ if stdin and len(stdin.strip()) <= 1:
+ logger.debug(
+ "Stdin has only 1 or 0 characters after stripping: '%s'",
+ stdin,
)
-
- return ChatCommand(args)
+ raise ChatCommandException(
+ "Your stdin input needs to have at least 2 characters."
+ )
+
+ if args.with_output and not TERMINAL_CAPTURE_FILE.exists():
+ raise ChatCommandException(
+ "Adding context from terminal output is only allowed if terminal capture is active."
+ )
+
+ try:
+ last_terminal_output = ""
+ if args.with_output:
+ logger.debug(
+ "Retrieving context from with-output parameter (index %s)",
+ args.with_output,
+ )
+ last_terminal_output = _read_last_terminal_output(args.with_output)
+
+ attachment = _parse_attachment_file(args.attachment)
+ attachment_mimetype = guess_mimetype(args.attachment)
+ stdin = stdin.strip()
+ question = args.query_string.strip() if args.query_string else ""
+
+ user_id = utils.get_user_id()
+ chat_id = _create_chat_session(utils, user_id, args.name, args.description)
+ response = _submit_question(
+ utils=utils,
+ user_id=user_id,
+ chat_id=chat_id,
+ question=question,
+ stdin=stdin,
+ attachment=attachment,
+ attachment_mimetype=attachment_mimetype,
+ last_output=last_terminal_output,
+ skip_spinner=args.raw,
+ )
+
+ _display_response(response)
+ return 0
+
+ except ValueError as e:
+ raise ChatCommandException(
+ f"Failed to get a response from LLM. {str(e)}"
+ ) from e
\ No newline at end of file
diff --git a/command_line_assistant/commands/chat_new.py b/command_line_assistant/commands/chat_new.py
new file mode 100644
index 0000000..d66c187
--- /dev/null
+++ b/command_line_assistant/commands/chat_new.py
@@ -0,0 +1,513 @@
+"""New chat command demonstrating action handlers and subcommands."""
+
+import argparse
+import logging
+import platform
+from argparse import Namespace
+from io import TextIOWrapper
+from typing import Optional
+
+from command_line_assistant.commands.registry import argument, command, subcommand
+from command_line_assistant.commands.utils import create_utils
+from command_line_assistant.dbus.exceptions import (
+ ChatNotFoundError,
+ HistoryNotEnabledError,
+)
+from command_line_assistant.dbus.structures.chat import (
+ AttachmentInput,
+ ChatList,
+ Question,
+ Response,
+ StdinInput,
+ SystemInfo,
+ TerminalInput,
+)
+from command_line_assistant.exceptions import ChatCommandException, StopInteractiveMode
+from command_line_assistant.rendering.decorators.colors import ColorDecorator
+from command_line_assistant.rendering.decorators.text import WriteOncePerSessionDecorator
+from command_line_assistant.terminal.parser import (
+ find_output_by_index,
+ parse_terminal_output,
+)
+from command_line_assistant.terminal.reader import TERMINAL_CAPTURE_FILE
+from command_line_assistant.utils.benchmark import TimingLogger
+from command_line_assistant.utils.cli import CommandContext
+from command_line_assistant.utils.files import NamedFileLock, guess_mimetype
+from command_line_assistant.utils.renderers import (
+ create_interactive_renderer,
+ create_markdown_renderer,
+ create_spinner_renderer,
+ create_text_renderer,
+ format_datetime,
+ human_readable_size,
+)
+
+logger = logging.getLogger(__name__)
+
+timing = TimingLogger(
+ filtered_params=[
+ "question",
+ "stdin",
+ "attachment",
+ "attachment_mimetype",
+ "last_output",
+ ]
+)
+
+# Constants
+MAX_QUESTION_SIZE: int = 2048
+LEGAL_NOTICE = (
+ "This feature uses AI technology. Do not include any personal information or "
+ "other sensitive information in your input. Interactions may be used to "
+ "improve Red Hat's products or services."
+)
+ALWAYS_LEGAL_MESSAGE = "Always review AI-generated content prior to use."
+DEFAULT_CHAT_DESCRIPTION = "Default Command Line Assistant Chat."
+DEFAULT_CHAT_NAME = "default"
+
+
+def _read_last_terminal_output(index: int) -> str:
+ """Read the last terminal output by index."""
+ logger.info("Reading terminal output.")
+ contents = parse_terminal_output()
+
+ if not contents:
+ logger.info("No contents found during reading the terminal output.")
+ return ""
+
+ return find_output_by_index(index=index, output=contents)
+
+
+def _parse_attachment_file(attachment: Optional[TextIOWrapper] = None) -> str:
+ """Parse attachment file and read its contents."""
+ if not attachment:
+ return ""
+
+ try:
+ return attachment.read().strip()
+ except UnicodeDecodeError as e:
+ raise ValueError(
+ "File appears to be binary or contains invalid text encoding"
+ ) from e
+
+
+def _get_input_source(query: str, stdin: str, attachment: str, last_output: str) -> str:
+ """Determine and return the appropriate input source based on combination rules."""
+ # All present - positional and file take precedence
+ if all([query, stdin, attachment, last_output]):
+ logger.debug("Using positional query and file input. Stdin will be ignored.")
+ return f"{query} {attachment}"
+
+ # Positional + attachment + last output
+ if query and attachment and last_output:
+ logger.info("Positional query, attachment and last output found. Using all of them at once.")
+ return f"{query} {attachment} {last_output}"
+
+ # Positional + last_output
+ if query and last_output:
+ logger.info("Positional query and last output found. Using them.")
+ return f"{query} {last_output}"
+
+ # Positional + file
+ if query and attachment:
+ logger.info("Positional query and attachment found. Using them.")
+ return f"{query} {attachment}"
+
+ # Stdin + file
+ if stdin and attachment:
+ logger.info("stdin and attachment found. Using them.")
+ return f"{stdin} {attachment}"
+
+ # Stdin + positional
+ if stdin and query:
+ logger.info("Positional query and stdin found. Using them.")
+ return f"{query} {stdin}"
+
+ # Single source - return first non-empty source
+ logger.info("Defaulting to use any of positional query, stdin, attachment or last output since no combinations where provided.")
+ source = next(
+ (src for src in [query, stdin, attachment, last_output] if src),
+ None,
+ )
+
+ if source:
+ return source
+
+ logger.error("Couldn't find a match.")
+ raise ValueError(
+ "No input provided. Please provide input via file, stdin, or direct query."
+ )
+
+
+def _create_chat_session(utils, user_id: str, name: str, description: str) -> str:
+ """Create a new chat session for a given conversation."""
+ has_chat_id = None
+ try:
+ has_chat_id = utils.chat_proxy.GetChatId(user_id, name)
+ except ChatNotFoundError:
+ # It's okay to swallow this exception as if there is no chat for
+ # this user, we will create one.
+ pass
+
+ # To avoid doing this check inside the CreateChat method, let's do it
+ # in here.
+ if has_chat_id:
+ return has_chat_id
+
+ return utils.chat_proxy.CreateChat(user_id, name, description)
+
+
+def _display_response(response: str) -> None:
+ """Display message to the terminal."""
+ legal_renderer = create_text_renderer(
+ decorators=[
+ ColorDecorator(foreground="lightyellow"),
+ WriteOncePerSessionDecorator(state_filename="legal"),
+ ]
+ )
+ notice_renderer = create_text_renderer(
+ decorators=[ColorDecorator(foreground="lightyellow")]
+ )
+ markdown_renderer = create_markdown_renderer()
+
+ legal_renderer.render(LEGAL_NOTICE)
+ markdown_renderer.render(response)
+ notice_renderer.render(ALWAYS_LEGAL_MESSAGE)
+
+
+@timing.timeit
+def _submit_question(
+ utils,
+ user_id: str,
+ chat_id: str,
+ question: str,
+ stdin: str,
+ attachment: str,
+ attachment_mimetype: str,
+ last_output: str,
+ skip_spinner: Optional[bool] = False,
+) -> str:
+ """Submit the question over dbus."""
+ final_question = _get_input_source(question, stdin, attachment, last_output)
+
+ if len(final_question) > MAX_QUESTION_SIZE:
+ final_question_size = human_readable_size(len(final_question))
+ max_question_size = human_readable_size(MAX_QUESTION_SIZE)
+ utils.render_warning(
+ f"The total size of your question and context ({final_question_size}) exceeds the limit of {max_question_size}. Trimming it down to fit in the expected size, you may lose some context."
+ )
+ logger.debug(
+ "Total size of question (%s) exceeds defined limit of %s.",
+ len(final_question),
+ MAX_QUESTION_SIZE,
+ )
+ final_question = final_question[:MAX_QUESTION_SIZE]
+ logger.debug(
+ "Final size of question after the limit %s.", len(final_question)
+ )
+
+ spinner_renderer = create_spinner_renderer(message="Asking RHEL Lightspeed")
+
+ if skip_spinner:
+ response = _get_response(utils, user_id, final_question, stdin, attachment, attachment_mimetype, last_output)
+ else:
+ with spinner_renderer:
+ response = _get_response(utils, user_id, final_question, stdin, attachment, attachment_mimetype, last_output)
+
+ try:
+ utils.history_proxy.WriteHistory(chat_id, user_id, final_question, response)
+ except HistoryNotEnabledError:
+ logger.warning(
+ "The history is disabled in the configuration file. Skipping the write to the history."
+ )
+
+ return response
+
+
+@timing.timeit
+def _get_response(
+ utils,
+ user_id: str,
+ question: str,
+ stdin: str,
+ attachment: str,
+ attachment_mimetype: str,
+ last_output: str,
+) -> str:
+ """Get the response from the chat session."""
+ message_input = Question(
+ message=question,
+ stdin=StdinInput(stdin=stdin),
+ attachment=AttachmentInput(
+ contents=attachment, mimetype=attachment_mimetype
+ ),
+ terminal=TerminalInput(output=last_output),
+ systeminfo=SystemInfo(
+ os=utils.context.os_release["name"],
+ version=utils.context.os_release["version_id"],
+ arch=platform.machine(),
+ id=utils.context.os_release["id"],
+ ),
+ )
+ response = utils.chat_proxy.AskQuestion(user_id, message_input.structure())
+
+ return Response.from_structure(response).message
+
+
+# ============================================================================
+# SUBCOMMANDS - Each operation gets its own clean function
+# ============================================================================
+
+@subcommand("chat", "list", help="List all chats")
+def list_chats(args: Namespace, context: CommandContext) -> int:
+ """List all chats operation."""
+ utils = create_utils(context)
+
+ try:
+ user_id = utils.get_user_id()
+ all_chats = ChatList.from_structure(utils.chat_proxy.GetAllChatFromUser(user_id))
+
+ if not all_chats.chats:
+ utils.render_success("No chats available.")
+ return 0
+
+ utils.render_success(f"Found a total of {len(all_chats.chats)} chats:")
+ for index, chat in enumerate(all_chats.chats):
+ created_at = format_datetime(chat.created_at)
+ utils.render_success(
+ f"{index}. Chat: {chat.name} - {chat.description} (created at: {created_at})"
+ )
+ return 0
+
+ except ChatCommandException as e:
+ utils.render_error(str(e))
+ return e.code
+
+
+@subcommand("chat", "delete", help="Delete a specific chat")
+@argument("chat_name", help="Name of the chat to delete")
+def delete_chat(args: Namespace, context: CommandContext) -> int:
+ """Delete a specific chat operation."""
+ utils = create_utils(context)
+
+ try:
+ user_id = utils.get_user_id()
+ utils.chat_proxy.DeleteChatForUser(user_id, args.chat_name)
+ utils.render_success(f"Chat {args.chat_name} deleted successfully.")
+ return 0
+ except ChatNotFoundError as e:
+ error_msg = f"Failed to delete requested chat {str(e)}"
+ utils.render_error(error_msg)
+ return 1
+
+
+@subcommand("chat", "delete-all", help="Delete all chats")
+def delete_all_chats(args: Namespace, context: CommandContext) -> int:
+ """Delete all chats operation."""
+ utils = create_utils(context)
+
+ try:
+ user_id = utils.get_user_id()
+ utils.chat_proxy.DeleteAllChatForUser(user_id)
+ utils.render_success("Deleted all chats successfully.")
+ return 0
+ except ChatNotFoundError as e:
+ error_msg = f"Failed to delete all requested chats {str(e)}"
+ utils.render_error(error_msg)
+ return 1
+
+
+@subcommand("chat", "interactive", help="Start an interactive chat session")
+@argument("-a", "--attachment", nargs="?", type=argparse.FileType("r"), help="File attachment to be read and sent alongside the query")
+@argument("-r", "--raw", action="store_true", help="Interact with the backend in raw mode. No spinners, emojis or anything.")
+@argument("-n", "--name", nargs="?", help="Give a name to the chat session.")
+@argument("--description", nargs="?", help="Give a description to the chat session.")
+def interactive_chat(args: Namespace, context: CommandContext) -> int:
+ """Interactive chat operation."""
+ utils = create_utils(context)
+
+ try:
+ # Set defaults
+ name = args.name or DEFAULT_CHAT_NAME
+ description = args.description or DEFAULT_CHAT_DESCRIPTION
+
+ terminal_file_lock = NamedFileLock(name="terminal")
+
+ if terminal_file_lock.is_locked:
+ raise ChatCommandException(
+ f"Detected a terminal capture session running with pid '{terminal_file_lock.pid}'."
+ " Interactive chat mode is not available while terminal capture is active, you must stop the previous one."
+ )
+
+ user_id = utils.get_user_id()
+ chat_id = _create_chat_session(utils, user_id, name, description)
+ attachment = _parse_attachment_file(args.attachment)
+ attachment_mimetype = guess_mimetype(args.attachment)
+
+ interactive_renderer = create_interactive_renderer()
+
+ while True:
+ interactive_renderer.render(">>> ")
+ question = interactive_renderer.output
+ if not question:
+ utils.render_error("Your question can't be empty. Please, try again.")
+ continue
+
+ response = _submit_question(
+ utils=utils,
+ user_id=user_id,
+ chat_id=chat_id,
+ question=question,
+ stdin="",
+ attachment=attachment,
+ attachment_mimetype=attachment_mimetype,
+ last_output="", # For now, we won't deal with last output in interactive mode
+ skip_spinner=args.raw,
+ )
+ _display_response(response)
+
+ except (KeyboardInterrupt, EOFError) as e:
+ raise ChatCommandException(
+ "Detected keyboard interrupt. Stopping interactive mode."
+ ) from e
+ except StopInteractiveMode:
+ return 0
+ except ChatCommandException as e:
+ utils.render_error(str(e))
+ return e.code
+
+ return 0
+
+
+# ============================================================================
+# MAIN COMMAND - Handles direct questions and uses action handlers
+# ============================================================================
+
+def _handle_interactive_mode(args: Namespace, context: CommandContext) -> int:
+ """Handler for --interactive flag."""
+ # Copy relevant arguments to match interactive subcommand
+ args.name = getattr(args, 'name', None)
+ args.description = getattr(args, 'description', None)
+ args.attachment = getattr(args, 'attachment', None)
+ args.raw = getattr(args, 'raw', False)
+ return interactive_chat(args, context)
+
+
+def _handle_list(args: Namespace, context: CommandContext) -> int:
+ """Handler for --list flag."""
+ return list_chats(args, context)
+
+
+@command("chat", help="Command to ask a question to the LLM")
+@argument("query_string", nargs="?", help="The question that will be sent to the LLM", default="")
+@argument("-a", "--attachment", nargs="?", type=argparse.FileType("r"), help="File attachment to be read and sent alongside the query")
+@argument("-w", "--with-output", nargs="?", type=int, help="Add output from terminal as context for the query. Use 1 to retrieve the latest output, 2 to and so on. First, enable the terminal capture with 'c shell --enable-capture' for this option to work.")
+@argument("-r", "--raw", action="store_true", help="Interact with the backend in raw mode. No spinners, emojis or anything.")
+@argument("-n", "--name", nargs="?", help="Give a name to the chat session.")
+@argument("--description", nargs="?", help="Give a description to the chat session.")
+@argument("-i", "--interactive", action="store_true", help="Start an interactive chat session", handler=_handle_interactive_mode)
+@argument("-l", "--list", action="store_true", help="List all chats", handler=_handle_list)
+def chat_command(args: Namespace, context: CommandContext) -> int:
+ """Main chat command - handles direct questions."""
+ utils = create_utils(context)
+
+ try:
+ # Handle special arguments preprocessing
+ if args.with_output:
+ logger.debug("Converting the index to a negative number in order to reverse search the output list.")
+ logger.debug("Original index is %s", args.with_output)
+ args.with_output = -abs(args.with_output)
+
+ # Set default name and description
+ name = args.name or DEFAULT_CHAT_NAME
+ description = args.description or DEFAULT_CHAT_DESCRIPTION
+
+ if not args.description and args.name:
+ utils.render_warning(
+ "Chat description not provided. Using the default description: "
+ f"'{DEFAULT_CHAT_DESCRIPTION}'. You can specify a custom description using the '--description' option."
+ )
+
+ if not args.name and args.description:
+ utils.render_warning(
+ "Chat name not provided. Using the default name: "
+ f"'{DEFAULT_CHAT_NAME}'. You can specify a custom name using the '--name' option."
+ )
+
+ # This is the default behavior - single question
+ return _single_question(utils, args, name, description)
+
+ except ChatCommandException as e:
+ logger.info("Failed to execute chat command: %s", str(e))
+ utils.render_error(str(e))
+ return e.code
+
+
+def _single_question(utils, args, name: str, description: str) -> int:
+ """Single question operation."""
+ # Validate query
+ if not args.query_string and not getattr(args, 'stdin', None):
+ logger.debug("Both query_string and stdin are empty.")
+ raise ChatCommandException(
+ "Your query needs to have at least 2 characters. Either query or stdin are empty."
+ )
+
+ if args.query_string and len(args.query_string.strip()) <= 1:
+ logger.debug(
+ "Query string has only 1 or 0 characters after stripping: '%s'",
+ args.query_string,
+ )
+ raise ChatCommandException(
+ "Your query needs to have at least 2 characters."
+ )
+
+ stdin = getattr(args, 'stdin', '') or ''
+ if stdin and len(stdin.strip()) <= 1:
+ logger.debug(
+ "Stdin has only 1 or 0 characters after stripping: '%s'",
+ stdin,
+ )
+ raise ChatCommandException(
+ "Your stdin input needs to have at least 2 characters."
+ )
+
+ if args.with_output and not TERMINAL_CAPTURE_FILE.exists():
+ raise ChatCommandException(
+ "Adding context from terminal output is only allowed if terminal capture is active."
+ )
+
+ try:
+ last_terminal_output = ""
+ if args.with_output:
+ logger.debug(
+ "Retrieving context from with-output parameter (index %s)",
+ args.with_output,
+ )
+ last_terminal_output = _read_last_terminal_output(args.with_output)
+
+ attachment = _parse_attachment_file(args.attachment)
+ attachment_mimetype = guess_mimetype(args.attachment)
+ stdin = stdin.strip()
+ question = args.query_string.strip() if args.query_string else ""
+
+ user_id = utils.get_user_id()
+ chat_id = _create_chat_session(utils, user_id, name, description)
+ response = _submit_question(
+ utils=utils,
+ user_id=user_id,
+ chat_id=chat_id,
+ question=question,
+ stdin=stdin,
+ attachment=attachment,
+ attachment_mimetype=attachment_mimetype,
+ last_output=last_terminal_output,
+ skip_spinner=args.raw,
+ )
+
+ _display_response(response)
+ return 0
+
+ except ValueError as e:
+ raise ChatCommandException(
+ f"Failed to get a response from LLM. {str(e)}"
+ ) from e
\ No newline at end of file
diff --git a/command_line_assistant/commands/example.py b/command_line_assistant/commands/example.py
new file mode 100644
index 0000000..e1f1fa5
--- /dev/null
+++ b/command_line_assistant/commands/example.py
@@ -0,0 +1,133 @@
+"""Example command showing both action handlers and subcommands."""
+
+import logging
+from argparse import Namespace
+
+from command_line_assistant.commands.registry import argument, command, subcommand
+from command_line_assistant.commands.utils import create_utils
+from command_line_assistant.exceptions import ChatCommandException
+from command_line_assistant.utils.cli import CommandContext
+
+logger = logging.getLogger(__name__)
+
+
+# ============================================================================
+# SUBCOMMANDS - Each operation gets its own clean function
+# ============================================================================
+
+@subcommand("example", "greet", help="Greet someone")
+@argument("name", help="Name of person to greet")
+@argument("-f", "--formal", action="store_true", help="Use formal greeting")
+def greet_subcommand(args: Namespace, context: CommandContext) -> int:
+ """Greet subcommand - completely separate from main command."""
+ utils = create_utils(context)
+
+ try:
+ if args.formal:
+ greeting = f"Good day, {args.name}!"
+ else:
+ greeting = f"Hey {args.name}!"
+
+ utils.render_success(greeting)
+ return 0
+
+ except Exception as e:
+ utils.render_error(f"Greeting failed: {e}")
+ return 1
+
+
+@subcommand("example", "count", help="Count numbers")
+@argument("max_num", type=int, help="Maximum number to count to")
+@argument("--step", type=int, default=1, help="Step size for counting")
+def count_subcommand(args: Namespace, context: CommandContext) -> int:
+ """Count subcommand."""
+ utils = create_utils(context)
+
+ try:
+ for i in range(1, args.max_num + 1, args.step):
+ utils.render_success(f"Count: {i}")
+
+ utils.render_warning(f"Finished counting to {args.max_num}")
+ return 0
+
+ except Exception as e:
+ utils.render_error(f"Counting failed: {e}")
+ return 1
+
+
+# ============================================================================
+# ACTION HANDLERS - Functions tied to specific arguments
+# ============================================================================
+
+def _handle_uppercase(args: Namespace, context: CommandContext) -> int:
+ """Handler for --uppercase flag."""
+ utils = create_utils(context)
+ message = getattr(args, 'message', 'Hello World!').upper()
+ utils.render_success(f"UPPERCASE: {message}")
+ return 0
+
+
+def _handle_error(args: Namespace, context: CommandContext) -> int:
+ """Handler for --error flag."""
+ utils = create_utils(context)
+ utils.render_error("This is a demonstration error via action handler!")
+ return 1
+
+
+# ============================================================================
+# MAIN COMMAND - Uses action handlers for some flags
+# ============================================================================
+
+@command("example", help="Example command showing action handlers and subcommands")
+@argument("message", nargs="?", help="Message to display", default="Hello World!")
+@argument("-c", "--count", type=int, default=1, help="Number of times to repeat the message")
+@argument("-u", "--uppercase", action="store_true", help="Display message in uppercase", handler=_handle_uppercase)
+@argument("--error", action="store_true", help="Demonstrate error handling", handler=_handle_error)
+def example_command(args: Namespace, context: CommandContext) -> int:
+ """Example command implementation demonstrating both approaches.
+
+ This shows two approaches:
+
+ 1. ACTION HANDLERS: Use handler= parameter in @argument
+ - Perfect for simple flags that trigger specific actions
+ - Eliminates if/else branching in main function
+ - Example: --uppercase, --error flags
+
+ 2. SUBCOMMANDS: Use @subcommand decorator
+ - Perfect for complex operations with their own arguments
+ - Completely separate functions
+ - Example: 'example greet John', 'example count 10'
+
+ This main function only handles the default behavior when no
+ action handlers or subcommands are triggered.
+ """
+ utils = create_utils(context)
+
+ try:
+ message = args.message
+
+ for i in range(args.count):
+ if args.count > 1:
+ utils.render_success(f"{i+1}. {message}")
+ else:
+ utils.render_success(message)
+
+ # Demonstrate different renderer types
+ utils.render_warning("This is a warning message")
+
+ # Show user info
+ user_id = utils.get_user_id()
+ utils.render_success(f"Current user ID: {user_id}")
+
+ utils.render_success("\nTry these commands:")
+ utils.render_success(" example greet Alice")
+ utils.render_success(" example count 5")
+ utils.render_success(" example --uppercase 'hello world'")
+ utils.render_success(" example --error")
+
+ return 0
+
+ except ChatCommandException as e:
+ logger.info("Example command failed: %s", str(e))
+ utils.render_error(str(e))
+ return 1
\ No newline at end of file
diff --git a/command_line_assistant/commands/feedback.py b/command_line_assistant/commands/feedback.py
index 3f51c84..22d8563 100644
--- a/command_line_assistant/commands/feedback.py
+++ b/command_line_assistant/commands/feedback.py
@@ -1,121 +1,33 @@
-"""Module to handle the feedback command."""
+"""Simplified feedback command implementation."""
-import argparse
import logging
from argparse import Namespace
-from enum import auto
-from typing import ClassVar
-from command_line_assistant.commands.base import (
- BaseCLICommand,
- BaseOperation,
- CommandOperationFactory,
- CommandOperationType,
-)
-from command_line_assistant.exceptions import (
- FeedbackCommandException,
-)
-from command_line_assistant.rendering.renders.text import TextRenderer
-from command_line_assistant.utils.cli import (
- SubParsersAction,
- create_subparser,
-)
-from command_line_assistant.utils.renderers import (
- create_error_renderer,
-)
-
-WARNING_MESSAGE = "Do not include any personal information or other sensitive information in your feedback. Feedback may be used to improve Red Hat's products or services."
+from command_line_assistant.commands.registry import argument, command
+from command_line_assistant.commands.utils import create_utils
+from command_line_assistant.exceptions import FeedbackCommandException
+from command_line_assistant.utils.cli import CommandContext
logger = logging.getLogger(__name__)
-
-class FeedbackOperationType(CommandOperationType):
- """Enum to control the operations for the command"""
-
- DEFAULT = auto()
+WARNING_MESSAGE = "Do not include any personal information or other sensitive information in your feedback. Feedback may be used to improve Red Hat's products or services."
-class FeedbackOperationFactory(CommandOperationFactory):
- """Factory for creating feedback operations with decorator-based registration"""
-
- # Mapping of CLI arguments to operation types
- _arg_to_operation: ClassVar[dict[str, CommandOperationType]] = {
- "submit": FeedbackOperationType.DEFAULT,
- }
-
-
-class BaseFeedbackOperation(BaseOperation):
- """Base feedback operation common to all operations."""
-
-
-@FeedbackOperationFactory.register(FeedbackOperationType.DEFAULT)
-class DefaultFeedbackOperation(BaseFeedbackOperation):
- """Class to hold the default feedback operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- self.warning_renderer.render(WARNING_MESSAGE)
-
+@command("feedback", help="Submit feedback about the Command Line Assistant responses and interactions.")
+@argument("--submit", action="store_true", default=True, help="Submit feedback (default action)")
+def feedback_command(args: Namespace, context: CommandContext) -> int:
+ """Feedback command implementation."""
+ utils = create_utils(context)
+
+ try:
+ utils.render_warning(WARNING_MESSAGE)
+
feedback_message = "To submit feedback, use the following email address: <cla-feedback@redhat.com>."
- self.text_renderer.render(feedback_message)
-
-
-class FeedbackCommand(BaseCLICommand):
- """Class that represents the feedback command."""
-
- def run(self) -> int:
- """Main entrypoint for the command to run.
-
- Returns:
- int: Return the status code for the operation
- """
- error_renderer: TextRenderer = create_error_renderer()
- operation_factory = FeedbackOperationFactory()
- try:
- # Get and execute the appropriate operation
- operation = operation_factory.create_operation(
- self._args, self._context, error_renderer=error_renderer
- )
- if operation:
- operation.execute()
-
- return 0
- except FeedbackCommandException as e:
- logger.info("Failed to execute feedback command: %s", str(e))
- error_renderer.render(str(e))
- return e.code
-
-
-def register_subcommand(parser: SubParsersAction):
- """
- Register this command to argparse so it's available for the root parser.
-
- Arguments:
- parser (SubParsersAction): Root parser to register command-specific arguments
- """
- feedback_parser = create_subparser(
- parser,
- "feedback",
- "Submit feedback about the Command Line Assistant responses and interactions.",
- )
-
- feedback_parser.add_argument(
- "--submit",
- action="store_true",
- default=True,
- help=argparse.SUPPRESS,
- )
-
- feedback_parser.set_defaults(func=_command_factory)
-
-
-def _command_factory(args: Namespace) -> FeedbackCommand:
- """Internal command factory to create the command class
-
- Arguments:
- args (Namespace): The arguments processed with argparse.
-
- Returns:
- FeedbackCommand: Return an instance of class
- """
- return FeedbackCommand(args)
+ utils.render_success(feedback_message)
+
+ return 0
+
+ except FeedbackCommandException as e:
+ logger.info("Failed to execute feedback command: %s", str(e))
+ utils.render_error(str(e))
+ return e.code
\ No newline at end of file
diff --git a/command_line_assistant/commands/history.py b/command_line_assistant/commands/history.py
index 40bcc1c..b4bafa1 100644
--- a/command_line_assistant/commands/history.py
+++ b/command_line_assistant/commands/history.py
@@ -1,35 +1,20 @@
-"""Module to handle the history command."""
+"""Simplified history command implementation."""
import logging
from argparse import Namespace
-from enum import auto
-from typing import ClassVar
-from command_line_assistant.commands.base import (
- BaseCLICommand,
- BaseOperation,
- CommandOperationFactory,
- CommandOperationType,
-)
+from command_line_assistant.commands.registry import argument, command
+from command_line_assistant.commands.utils import create_utils
from command_line_assistant.dbus.exceptions import (
HistoryNotAvailableError,
HistoryNotEnabledError,
)
-from command_line_assistant.dbus.interfaces.chat import ChatInterface
-from command_line_assistant.dbus.interfaces.history import HistoryInterface
-from command_line_assistant.dbus.interfaces.user import UserInterface
from command_line_assistant.dbus.structures.chat import ChatList
from command_line_assistant.dbus.structures.history import HistoryList
from command_line_assistant.exceptions import HistoryCommandException
from command_line_assistant.rendering.decorators.colors import ColorDecorator
-from command_line_assistant.rendering.renders.text import TextRenderer
-from command_line_assistant.utils.cli import (
- CommandContext,
- SubParsersAction,
- create_subparser,
-)
+from command_line_assistant.utils.cli import CommandContext
from command_line_assistant.utils.renderers import (
- create_error_renderer,
create_markdown_renderer,
create_text_renderer,
format_datetime,
@@ -37,389 +22,218 @@ from command_line_assistant.utils.renderers import (
logger = logging.getLogger(__name__)
-
-#: Message for when the history is not available yet.
+# Constants
HISTORY_NOT_AVAILABLE_MESSAGE = (
"Looks like no history was found. Try asking something first!"
)
-
-#: Message for when the history is not enabled yet.
HISTORY_NOT_ENABLED_MESSAGE = "Looks like history is not enabled yet. Enable it in the configuration file before trying to access history."
-
-#: History not enabled debug message.
HISTORY_NOT_ENABLED_DEBUG = "History is not enabled. Nothing to do."
-class HistoryOperationType(CommandOperationType):
- """Enum to control the operations for the command"""
+def _show_history(utils, entries: HistoryList) -> None:
+ """Internal method to show the history in a standardized way."""
+ if not entries.histories:
+ utils.render_success("No history entries found")
+ return
- CLEAR = auto()
- CLEAR_ALL = auto()
- FIRST = auto()
- LAST = auto()
- FILTER = auto()
- ALL = auto()
+ # Create specialized renderers for different parts
+ question_renderer = create_markdown_renderer(
+ decorators=[ColorDecorator(foreground="cyan")]
+ )
+ answer_renderer = create_markdown_renderer(
+ decorators=[ColorDecorator(foreground="green")]
+ )
+ metadata_renderer = create_text_renderer(
+ decorators=[ColorDecorator(foreground="yellow")]
+ )
+
+ for entry in entries.histories:
+ # Render question block
+ question_text = f"## 🤔 Question\n{entry.question}"
+ question_renderer.render(question_text)
+
+ # Add a small spacing
+ utils.text_renderer.render("")
+
+ # Render answer block
+ answer_text = f"## 🤖 Answer\n{entry.response}"
+ answer_renderer.render(answer_text)
+
+ from_chat_message = f"\n*From chat: {entry.chat_name}*"
+ metadata_renderer.render(from_chat_message)
+
+ created_at_message = f"*Created at: {format_datetime(entry.created_at)}*"
+ metadata_renderer.render(created_at_message)
+
+ # Add separator between entries if needed
+ if len(entries.histories) > 1:
+ utils.text_renderer.render(
+ "\n" + "═" * (len(created_at_message) - 1) + "\n"
+ )
-class HistoryOperationFactory(CommandOperationFactory):
- """Factory for creating shell operations with decorator-based registration"""
-
- # Mapping of CLI arguments to operation types
- _arg_to_operation: ClassVar[dict[str, CommandOperationType]] = {
- "clear": HistoryOperationType.CLEAR,
- "clear_all": HistoryOperationType.CLEAR_ALL,
- "first": HistoryOperationType.FIRST,
- "last": HistoryOperationType.LAST,
- "filter": HistoryOperationType.FILTER,
- "all": HistoryOperationType.ALL,
- }
+@command("history", help="Manage Conversation History")
+@argument("-f", "--first", action="store_true", help="Get the first conversation from history.")
+@argument("-l", "--last", action="store_true", help="Get the last conversation from history.")
+@argument("--filter", help="Search for a specific keyword of text in the history.")
+@argument("-a", "--all", action="store_true", help="Get all the conversation history.")
+@argument("--from-chat", help="Specify from which chat we should retrieve the history. Default chat is 'default'", default="default")
+@argument("-c", "--clear", action="store_true", help="Clear the entire history for a given chat. Use --from-chat with its given name to clear that particular history.")
+@argument("--clear-all", action="store_true", help="Clear the entire history.")
+def history_command(args: Namespace, context: CommandContext) -> int:
+ """History command implementation."""
+ utils = create_utils(context)
+
+ try:
+ # Set default operation if none specified
+ if not (args.last or args.first or args.filter or args.clear or args.clear_all):
+ args.all = True
+
+ # Handle different operations
+ if args.clear:
+ return _clear_history(utils, args.from_chat)
+ elif args.clear_all:
+ return _clear_all_history(utils)
+ elif args.first:
+ return _first_history(utils, args.from_chat)
+ elif args.last:
+ return _last_history(utils, args.from_chat)
+ elif args.filter:
+ return _filter_history(utils, args.filter, args.from_chat)
+ else: # args.all
+ return _all_history(utils)
+
+ except HistoryCommandException as e:
+ logger.info("Failed to execute history command: %s", str(e))
+ utils.render_error(str(e))
+ return e.code
-class BaseHistoryOperation(BaseOperation):
- """Base history operation common to all operations
+def _clear_history(utils, from_chat: str) -> int:
+ """Clear history operation."""
+ try:
+ user_id = utils.get_user_id()
+ is_chat_available = utils.chat_proxy.IsChatAvailable(user_id, from_chat)
- Warning:
- The proxy attributes in this class are not really mapping to interface.
- It maps to internal dasbus ObjectProxy, but to avoid pyright syntax
- errors, we type then as their respective interfaces. The objective of
- the `ObjectProxy` is to serve as a proxy for the real interfaces.
+ if not is_chat_available:
+ raise HistoryCommandException(
+ f"Nothing to clean as {from_chat} chat is not available."
+ )
- Attributes:
- q_renderer (TextRenderer): Instance of a text renderer to render questions
- a_renderer (TextRenderer): Instance of a text renderer to render answers
- """
+ utils.history_proxy.ClearHistory(user_id, from_chat)
+ utils.render_success("Cleaning the history.")
+ return 0
+ except HistoryNotAvailableError as e:
+ logger.debug("Failed to clear the history: %s", str(e))
+ raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
+ except HistoryNotEnabledError as e:
+ logger.debug(HISTORY_NOT_ENABLED_DEBUG)
+ raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
- def __init__(
- self,
- text_renderer: TextRenderer,
- warning_renderer: TextRenderer,
- error_renderer: TextRenderer,
- args: Namespace,
- context: CommandContext,
- chat_proxy: ChatInterface,
- history_proxy: HistoryInterface,
- user_proxy: UserInterface,
- ):
- """Constructor of the class.
- Arguments:
- text_renderer (TextRenderer): Instance of text renderer class
- warning_renderer (TextRenderer): Instance of text renderer class
- error_renderer (TextRenderer): Instance of text renderer class
- args (Namespace): The arguments from CLI
- context (CommandContext): Context for the commands
- chat_proxy (ChatInterface): The proxy object for dbus chat
- history_proxy (HistoryInterface): The proxy object for dbus history
- user_proxy (HistoryInterface): The proxy object for dbus user
- """
- super().__init__(
- text_renderer,
- warning_renderer,
- error_renderer,
- args,
- context,
- chat_proxy,
- history_proxy,
- user_proxy,
- )
- # Add markdown renderer as a standard renderer
- self.markdown_renderer = create_markdown_renderer()
-
- def _show_history(self, entries: HistoryList) -> None:
- """Internal method to show the history in a standardized way
-
- Arguments:
- entries (HistoryItem): The list of entries in the history
- """
- if not entries.histories:
- self.text_renderer.render("No history entries found")
- return
-
- # Create specialized renderers for different parts
- question_renderer = create_markdown_renderer(
- decorators=[
- ColorDecorator(foreground="cyan"),
- ]
+def _clear_all_history(utils) -> int:
+ """Clear all history operation."""
+ try:
+ user_id = utils.get_user_id()
+ all_user_chats = ChatList.from_structure(
+ utils.chat_proxy.GetAllChatFromUser(user_id)
)
- answer_renderer = create_markdown_renderer(
- decorators=[
- ColorDecorator(foreground="green"),
- ]
+ if not all_user_chats.chats:
+ raise HistoryCommandException(
+ "Nothing to clean as there is no chat session in place."
+ )
+
+ utils.render_success("Cleaning the history.")
+ utils.history_proxy.ClearAllHistory(user_id)
+ return 0
+ except HistoryNotAvailableError as e:
+ logger.debug("Failed to clear the history: %s", str(e))
+ raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
+ except HistoryNotEnabledError as e:
+ logger.debug(HISTORY_NOT_ENABLED_DEBUG)
+ raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
+
+
+def _first_history(utils, from_chat: str) -> int:
+ """First history operation."""
+ try:
+ user_id = utils.get_user_id()
+ utils.render_success("Getting first conversation from history.")
+ response = utils.history_proxy.GetFirstConversation(user_id, from_chat)
+ history = HistoryList.from_structure(response)
+
+ # Display the conversation
+ _show_history(utils, history)
+ return 0
+ except HistoryNotAvailableError as e:
+ logger.debug("Failed to retrieve the first history entry: %s", str(e))
+ raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
+ except HistoryNotEnabledError as e:
+ logger.debug(HISTORY_NOT_ENABLED_DEBUG)
+ raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
+
+
+def _last_history(utils, from_chat: str) -> int:
+ """Last history operation."""
+ try:
+ utils.render_success("Getting last conversation from history.")
+ user_id = utils.get_user_id()
+ response = utils.history_proxy.GetLastConversation(user_id, from_chat)
+
+ history = HistoryList.from_structure(response)
+ # Display the conversation
+ _show_history(utils, history)
+ return 0
+ except HistoryNotAvailableError as e:
+ logger.debug("Failed to retrieve the last history entry: %s", str(e))
+ raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
+ except HistoryNotEnabledError as e:
+ logger.debug(HISTORY_NOT_ENABLED_DEBUG)
+ raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
+
+
+def _filter_history(utils, filter_text: str, from_chat: str) -> int:
+ """Filter history operation."""
+ try:
+ utils.render_success("Filtering conversation history.")
+ user_id = utils.get_user_id()
+ response = utils.history_proxy.GetFilteredConversation(
+ user_id, filter_text, from_chat
)
- metadata_renderer = create_text_renderer(
- decorators=[
- ColorDecorator(foreground="yellow"),
- ]
+ # Handle and display the response
+ history = HistoryList.from_structure(response)
+
+ # Display the conversation
+ _show_history(utils, history)
+ return 0
+ except HistoryNotAvailableError as e:
+ logger.debug(
+ "Failed to retrieve entries with filter '%s': %s",
+ filter_text,
+ str(e),
)
-
- for entry in entries.histories:
- # Render question block
- question_text = f"## 🤔 Question\n{entry.question}"
- question_renderer.render(question_text)
-
- # Add a small spacing
- self.text_renderer.render("")
-
- # Render answer block
- answer_text = f"## 🤖 Answer\n{entry.response}"
- answer_renderer.render(answer_text)
-
- from_chat_message = f"\n*From chat: {entry.chat_name}*"
- metadata_renderer.render(from_chat_message)
-
- created_at_message = f"*Created at: {format_datetime(entry.created_at)}*"
- metadata_renderer.render(created_at_message)
- # Add separator between entries if needed
- if len(entries.histories) > 1:
- self.text_renderer.render(
- "\n" + "═" * (len(created_at_message) - 1) + "\n"
- )
+ raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
+ except HistoryNotEnabledError as e:
+ logger.debug(HISTORY_NOT_ENABLED_DEBUG)
+ raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
-@HistoryOperationFactory.register(HistoryOperationType.CLEAR)
-class ClearHistoryOperation(BaseHistoryOperation):
- """Class to hold the clean operation"""
+def _all_history(utils) -> int:
+ """All history operation."""
+ try:
+ utils.render_success("Getting all conversations from history.")
+ user_id = utils.get_user_id()
+ response = utils.history_proxy.GetHistory(user_id)
+ history = HistoryList.from_structure(response)
- def execute(self) -> None:
- """Default method to execute the operation"""
- try:
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- is_chat_available = self.chat_proxy.IsChatAvailable(
- user_id, self.args.from_chat
- )
-
- if not is_chat_available:
- raise HistoryCommandException(
- "Nothing to clean as %s chat is not available."
- % self.args.from_chat
- )
-
- self.history_proxy.ClearHistory(user_id, self.args.from_chat)
- self.text_renderer.render("Cleaning the history.")
- except HistoryNotAvailableError as e:
- logger.debug("Failed to clear the history: %s", str(e))
- raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
- except HistoryNotEnabledError as e:
- logger.debug(HISTORY_NOT_ENABLED_DEBUG)
- raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
-
-
-@HistoryOperationFactory.register(HistoryOperationType.CLEAR_ALL)
-class ClearAllHistoryOperation(BaseHistoryOperation):
- """Class to hold the clean operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- try:
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- all_user_chats = ChatList.from_structure(
- self.chat_proxy.GetAllChatFromUser(user_id)
- )
-
- if not all_user_chats.chats:
- raise HistoryCommandException(
- "Nothing to clean as there is no chat session in place."
- )
-
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- self.text_renderer.render("Cleaning the history.")
- self.history_proxy.ClearAllHistory(user_id)
- except HistoryNotAvailableError as e:
- logger.debug("Failed to clear the history: %s", str(e))
- raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
- except HistoryNotEnabledError as e:
- logger.debug(HISTORY_NOT_ENABLED_DEBUG)
- raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
-
-
-@HistoryOperationFactory.register(HistoryOperationType.FIRST)
-class FirstHistoryOperation(BaseHistoryOperation):
- """Class to hold the first history operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- try:
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- self.text_renderer.render("Getting first conversation from history.")
- response = self.history_proxy.GetFirstConversation(
- user_id, self.args.from_chat
- )
- history = HistoryList.from_structure(response)
-
- # Display the conversation
- self._show_history(history)
- except HistoryNotAvailableError as e:
- logger.debug("Failed to retrieve the first history entry: %s", str(e))
- raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
- except HistoryNotEnabledError as e:
- logger.debug(HISTORY_NOT_ENABLED_DEBUG)
- raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
-
-
-@HistoryOperationFactory.register(HistoryOperationType.LAST)
-class LastHistoryOperation(BaseHistoryOperation):
- """Class to hold the last history operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- try:
- self.text_renderer.render("Getting last conversation from history.")
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- response = self.history_proxy.GetLastConversation(
- user_id, self.args.from_chat
- )
-
- history = HistoryList.from_structure(response)
- # Display the conversation
- self._show_history(history)
- except HistoryNotAvailableError as e:
- logger.debug("Failed to retrieve the last history entry: %s", str(e))
- raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
- except HistoryNotEnabledError as e:
- logger.debug(HISTORY_NOT_ENABLED_DEBUG)
- raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
-
-
-@HistoryOperationFactory.register(HistoryOperationType.FILTER)
-class FilteredHistoryOperation(BaseHistoryOperation):
- """Class to hold the filtering history operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- try:
- self.text_renderer.render("Filtering conversation history.")
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- response = self.history_proxy.GetFilteredConversation(
- user_id, self.args.filter, self.args.from_chat
- )
-
- # Handle and display the response
- history = HistoryList.from_structure(response)
-
- # Display the conversation
- self._show_history(history)
- except HistoryNotAvailableError as e:
- logger.debug(
- "Failed to retrieve entries with filter '%s': %s",
- self.args.filter,
- str(e),
- )
- raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
- except HistoryNotEnabledError as e:
- logger.debug(HISTORY_NOT_ENABLED_DEBUG)
- raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
-
-
-@HistoryOperationFactory.register(HistoryOperationType.ALL)
-class AllHistoryOperation(BaseHistoryOperation):
- """Class to hold the reading of all history operation."""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- try:
- self.text_renderer.render("Getting all conversations from history.")
- user_id = self.user_proxy.GetUserId(self.context.effective_user_id)
- response = self.history_proxy.GetHistory(user_id)
- history = HistoryList.from_structure(response)
-
- # Display the conversation
- self._show_history(history)
- except HistoryNotAvailableError as e:
- logger.debug("Failed to retrieve the all history entries: %s", str(e))
- raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
- except HistoryNotEnabledError as e:
- logger.debug(HISTORY_NOT_ENABLED_DEBUG)
- raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
-
-
-class HistoryCommand(BaseCLICommand):
- """Class that represents the history command."""
-
- def run(self) -> int:
- """Main entrypoint for the command to run.
-
- Returns:
- int: Status code of the execution.
- """
- error_renderer: TextRenderer = create_error_renderer()
- operation_factory = HistoryOperationFactory()
- try:
- operation = operation_factory.create_operation(
- self._args, self._context, error_renderer=error_renderer
- )
-
- if operation:
- operation.execute()
- return 0
- except HistoryCommandException as e:
- logger.info("Failed to execute history command: %s", str(e))
- error_renderer.render(str(e))
- return e.code
-
-
-def register_subcommand(parser: SubParsersAction):
- """
- Register this command to argparse so it's available for the root parser.
-
- Arguments:
- parser (SubParsersAction): Root parser to register command-specific arguments
- """
- history_parser = create_subparser(parser, "history", "Manage Conversation History")
-
- filtering_options = history_parser.add_argument_group("Filtering Options")
- filtering_options.add_argument(
- "-f",
- "--first",
- action="store_true",
- help="Get the first conversation from history.",
- )
- filtering_options.add_argument(
- "-l",
- "--last",
- action="store_true",
- help="Get the last conversation from history.",
- )
- filtering_options.add_argument(
- "--filter",
- help="Search for a specific keyword of text in the history.",
- )
- filtering_options.add_argument(
- "-a", "--all", action="store_true", help="Get all the conversation history."
- )
- filtering_options.add_argument(
- "--from-chat",
- help="Specify from which chat we should retrieve the history. Default chat is 'default'",
- default="default",
- )
-
- management_options = history_parser.add_argument_group("Management Options")
- management_options.add_argument(
- "-c",
- "--clear",
- action="store_true",
- help="Clear the entire history for a given chat. Use --from-chat with its given name to clear that particular history.",
- )
- management_options.add_argument(
- "--clear-all",
- action="store_true",
- help="Clear the entire history.",
- )
-
- history_parser.set_defaults(func=_command_factory)
-
-
-def _command_factory(args: Namespace) -> HistoryCommand:
- """Internal command factory to create the command class
-
- Arguments:
- args (Namespace): The arguments processed with argparse.
-
- Returns:
- HistoryCommand: Return an instance of class
- """
- if not (args.last or args.first or args.filter):
- args.all = True
-
- return HistoryCommand(args)
+ # Display the conversation
+ _show_history(utils, history)
+ return 0
+ except HistoryNotAvailableError as e:
+ logger.debug("Failed to retrieve the all history entries: %s", str(e))
+ raise HistoryCommandException(HISTORY_NOT_AVAILABLE_MESSAGE) from e
+ except HistoryNotEnabledError as e:
+ logger.debug(HISTORY_NOT_ENABLED_DEBUG)
+ raise HistoryCommandException(HISTORY_NOT_ENABLED_MESSAGE) from e
\ No newline at end of file
diff --git a/command_line_assistant/commands/registry.py b/command_line_assistant/commands/registry.py
new file mode 100644
index 0000000..c0815fb
--- /dev/null
+++ b/command_line_assistant/commands/registry.py
@@ -0,0 +1,292 @@
+"""Simplified command registry system for CLI commands."""
+
+import logging
+from argparse import ArgumentParser, Namespace, _SubParsersAction
+from functools import wraps
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+
+from command_line_assistant.utils.cli import CommandContext
+
+logger = logging.getLogger(__name__)
+
+# Type definitions
+CommandFunc = Callable[[Namespace, CommandContext], int]
+ArgumentSpec = Tuple[Tuple[str, ...], Dict[str, Any]]
+ActionHandler = Callable[[Namespace, CommandContext], int]
+
+class CommandRegistry:
+ """Registry for CLI commands with decorator-based registration."""
+
+ def __init__(self):
+ self._commands: Dict[str, 'Command'] = {}
+ self._subcommands: Dict[str, Dict[str, 'SubCommand']] = {}
+
+ def command(
+ self,
+ name: str,
+ help: Optional[str] = None,
+ description: Optional[str] = None
+ ) -> Callable[[CommandFunc], CommandFunc]:
+ """Decorator to register a command function.
+
+ Args:
+ name: Command name
+ help: Short help text
+ description: Longer description
+
+ Returns:
+ Decorator function
+ """
+ def decorator(func: CommandFunc) -> CommandFunc:
+ cmd = Command(name, func, help, description)
+ self._commands[name] = cmd
+
+ logger.debug(f"Registered command: {name}")
+ return func
+
+ return decorator
+
+ def argument(
+ self,
+ *args: str,
+ handler: Optional[ActionHandler] = None,
+ **kwargs: Any
+ ) -> Callable[[CommandFunc], CommandFunc]:
+ """Decorator to add arguments to a command.
+
+ Args:
+ *args: Positional argument names
+ handler: Optional function to handle this specific argument
+ **kwargs: Argument options
+
+ Returns:
+ Decorator function
+ """
+ def decorator(func: CommandFunc) -> CommandFunc:
+ if not hasattr(func, '_cmd_arguments'):
+ setattr(func, '_cmd_arguments', [])
+
+ # Store handler info in kwargs if provided
+ if handler:
+ kwargs['_handler'] = handler
+
+ getattr(func, '_cmd_arguments').append((args, kwargs))
+ return func
+
+ return decorator
+
+ def subcommand(
+ self,
+ parent_command: str,
+ name: str,
+ help: Optional[str] = None,
+ description: Optional[str] = None
+ ) -> Callable[[CommandFunc], CommandFunc]:
+ """Decorator to register a subcommand function.
+
+ Args:
+ parent_command: Name of parent command
+ name: Subcommand name
+ help: Short help text
+ description: Longer description
+
+ Returns:
+ Decorator function
+ """
+ def decorator(func: CommandFunc) -> CommandFunc:
+ if parent_command not in self._subcommands:
+ self._subcommands[parent_command] = {}
+
+ subcmd = SubCommand(name, func, help, description)
+ self._subcommands[parent_command][name] = subcmd
+
+ logger.debug(f"Registered subcommand: {parent_command} {name}")
+ return func
+
+ return decorator
+
+ def register_commands(self, parser: _SubParsersAction) -> None:
+ """Register all commands with the argument parser.
+
+ Args:
+ parser: Subparser to register commands with
+ """
+ for cmd in self._commands.values():
+ subcommands = self._subcommands.get(cmd.name, {})
+ cmd.register(parser, subcommands)
+
+ def get_command(self, name: str) -> Optional['Command']:
+ """Get a command by name.
+
+ Args:
+ name: Command name
+
+ Returns:
+ Command instance or None
+ """
+ return self._commands.get(name)
+
+
+class Command:
+ """Represents a single CLI command."""
+
+ def __init__(
+ self,
+ name: str,
+ func: CommandFunc,
+ help: Optional[str] = None,
+ description: Optional[str] = None
+ ):
+ self.name = name
+ self.func = func
+ self.help = help
+ self.description = description
+ self.arguments: List[ArgumentSpec] = getattr(func, '_cmd_arguments', [])
+ self.action_handlers: Dict[str, ActionHandler] = {}
+
+ def register(self, parser: _SubParsersAction, subcommands: Dict[str, 'SubCommand'] = None) -> None:
+ """Register this command with the argument parser.
+
+ Args:
+ parser: Subparser to register with
+ subcommands: Optional subcommands for this command
+ """
+ if subcommands:
+ # Create command with subcommands
+ subparser = parser.add_parser(
+ self.name,
+ help=self.help,
+ description=self.description
+ )
+
+ # Create subparsers for this command
+ subparsers = subparser.add_subparsers(
+ dest=f'{self.name}_subcommand',
+ help=f'{self.name} subcommands'
+ )
+
+ # Register each subcommand
+ for subcmd in subcommands.values():
+ subcmd_parser = subparsers.add_parser(
+ subcmd.name,
+ help=subcmd.help,
+ description=subcmd.description
+ )
+
+ # Add subcommand arguments
+ for args, kwargs in reversed(subcmd.arguments):
+ subcmd_parser.add_argument(*args, **kwargs)
+
+ subcmd_parser.set_defaults(func=self._create_subcommand_wrapper(subcmd))
+
+ # Set default for main command (when no subcommand is used)
+ subparser.set_defaults(func=self._create_command_wrapper())
+ else:
+ # Regular command without subcommands
+ subparser = parser.add_parser(
+ self.name,
+ help=self.help,
+ description=self.description
+ )
+
+ # Add arguments in reverse order (decorators are applied bottom-up)
+ for args, kwargs in reversed(self.arguments):
+ # Extract handler if present
+ handler = kwargs.pop('_handler', None)
+ if handler:
+ # Store handler mapped to argument name
+ arg_name = self._get_arg_name(args, kwargs)
+ if arg_name:
+ self.action_handlers[arg_name] = handler
+
+ subparser.add_argument(*args, **kwargs)
+
+ # Set the command function as default
+ subparser.set_defaults(func=self._create_command_wrapper())
+
+ def _get_arg_name(self, args: Tuple[str, ...], kwargs: Dict[str, Any]) -> Optional[str]:
+ """Extract the argument name from argparse arguments."""
+ dest = kwargs.get('dest')
+ if dest:
+ return dest
+
+ # Find the longest argument name (usually the --long-form)
+ for arg in args:
+ if arg.startswith('--'):
+ return arg[2:].replace('-', '_')
+
+ # Fallback to positional argument
+ for arg in args:
+ if not arg.startswith('-'):
+ return arg
+
+ return None
+
+ def _create_subcommand_wrapper(self, subcmd: 'SubCommand') -> Callable[[Namespace], Any]:
+ """Create wrapper for subcommand."""
+ def wrapper(args: Namespace) -> Any:
+ context = CommandContext()
+ try:
+ return subcmd.func(args, context)
+ except Exception as e:
+ logger.error(f"Subcommand {self.name} {subcmd.name} failed: {e}")
+ raise
+
+ return wrapper
+
+ def _create_command_wrapper(self) -> Callable[[Namespace], Any]:
+ """Create a wrapper function for the command.
+
+ Returns:
+ Wrapper function
+ """
+ def wrapper(args: Namespace) -> Any:
+ context = CommandContext()
+ try:
+ # Check if any action handlers should be called
+ for arg_name, handler in self.action_handlers.items():
+ if hasattr(args, arg_name) and getattr(args, arg_name):
+ return handler(args, context)
+
+ # No specific handler found, call main function
+ return self.func(args, context)
+ except Exception as e:
+ logger.error(f"Command {self.name} failed: {e}")
+ raise
+
+ return wrapper
+
+
+class SubCommand:
+ """Represents a subcommand within a main command."""
+
+ def __init__(
+ self,
+ name: str,
+ func: CommandFunc,
+ help: Optional[str] = None,
+ description: Optional[str] = None
+ ):
+ self.name = name
+ self.func = func
+ self.help = help
+ self.description = description
+ self.arguments: List[ArgumentSpec] = getattr(func, '_cmd_arguments', [])
+
+
+# Global registry instance
+registry = CommandRegistry()
+
+# Export decorators for convenience
+command = registry.command
+argument = registry.argument
+subcommand = registry.subcommand
+
+# Helper function to register all commands
+def register_all_commands(parser: _SubParsersAction) -> None:
+ """Register all commands with the parser.
+
+ Args:
+ parser: Subparser to register commands with
+ """
+ registry.register_commands(parser)
\ No newline at end of file
diff --git a/command_line_assistant/commands/shell.py b/command_line_assistant/commands/shell.py
index 1fd26a8..5964f7a 100644
--- a/command_line_assistant/commands/shell.py
+++ b/command_line_assistant/commands/shell.py
@@ -1,237 +1,127 @@
-"""Module to handle the shell command."""
+"""Simplified shell command implementation."""
import logging
from argparse import Namespace
-from enum import auto
from pathlib import Path
-from typing import ClassVar, Union
+from typing import Union
-from command_line_assistant.commands.base import (
- BaseCLICommand,
- BaseOperation,
- CommandOperationFactory,
- CommandOperationType,
-)
+from command_line_assistant.commands.registry import argument, command
+from command_line_assistant.commands.utils import create_utils
from command_line_assistant.exceptions import ShellCommandException
-from command_line_assistant.integrations import (
- BASH_INTERACTIVE,
-)
-from command_line_assistant.rendering.renders.text import TextRenderer
+from command_line_assistant.integrations import BASH_INTERACTIVE
from command_line_assistant.terminal.reader import (
TERMINAL_CAPTURE_FILE,
start_capturing,
)
-from command_line_assistant.utils.cli import (
- SubParsersAction,
- create_subparser,
-)
+from command_line_assistant.utils.cli import CommandContext
from command_line_assistant.utils.files import NamedFileLock, create_folder, write_file
-from command_line_assistant.utils.renderers import (
- create_error_renderer,
-)
-
-#: The path to bashrc.d folder
-BASH_RC_D_PATH: Path = Path("~/.bashrc.d").expanduser()
-
-#: The complete path to the integration file.
-INTERACTIVE_MODE_INTEGRATION_FILE: Path = Path(BASH_RC_D_PATH, "cla-interactive.bashrc")
-
-#: The complete path to the persistent terminal capture mode.
-PERSISTENT_TERMINAL_CAPTURE_FILE: Path = Path(
- BASH_RC_D_PATH, "cla-persistent-capture.bashrc"
-)
-
-#: File to track all the CLA environment variable exports.
-ESSENTIAL_EXPORTS_FILE: Path = Path(BASH_RC_D_PATH, "cla-exports.bashrc")
logger = logging.getLogger(__name__)
-
-class ShellOperationType(CommandOperationType):
- """Enum to control the operations for the command"""
-
- ENABLE_INTERACTIVE = auto()
- DISABLE_INTERACTIVE = auto()
- ENABLE_CAPTURE = auto()
+# Constants
+BASH_RC_D_PATH: Path = Path("~/.bashrc.d").expanduser()
+INTERACTIVE_MODE_INTEGRATION_FILE: Path = Path(BASH_RC_D_PATH, "cla-interactive.bashrc")
+PERSISTENT_TERMINAL_CAPTURE_FILE: Path = Path(BASH_RC_D_PATH, "cla-persistent-capture.bashrc")
+ESSENTIAL_EXPORTS_FILE: Path = Path(BASH_RC_D_PATH, "cla-exports.bashrc")
-class ShellOperationFactory(CommandOperationFactory):
- """Factory for creating shell operations with decorator-based registration"""
-
- # Mapping of CLI arguments to operation types
- _arg_to_operation: ClassVar[dict[str, CommandOperationType]] = {
- "enable_interactive": ShellOperationType.ENABLE_INTERACTIVE,
- "disable_interactive": ShellOperationType.DISABLE_INTERACTIVE,
- "enable_capture": ShellOperationType.ENABLE_CAPTURE,
- }
+def _initialize_bash_folder() -> None:
+ """Initialize the bash folder."""
+ create_folder(BASH_RC_D_PATH)
-# Base class for shell operations with common functionality
-class BaseShellOperation(BaseOperation):
- """Base shell operation common to all operations."""
+def _write_bash_functions(utils, file: Path, contents: Union[bytes, str]) -> None:
+ """Write bash function to the desired location."""
+ _initialize_bash_folder()
+ if file.exists():
+ logger.info("File already exists at %s.", file)
+ utils.render_warning(
+ f"The integration is already present and enabled at {file}! "
+ "Restart your terminal or source ~/.bashrc in case it's not working."
+ )
+ return
- def _initialize_bash_folder(self) -> None:
- """Internal function to initialize the bash folder"""
- # Always ensure essential exports are in place
- create_folder(BASH_RC_D_PATH)
+ write_file(contents, file)
+ utils.render_success(
+ f"Integration successfully added at {file}. "
+ "In order to use it, please restart your terminal or source ~/.bashrc"
+ )
- def _write_bash_functions(self, file: Path, contents: Union[bytes, str]) -> None:
- """Internal funtion to write the bash function to the desired location
- Arguments:
- file (Path): The path object with the correct location
- contents (Union[bytes, str]): The contents to be written in the file.
- """
- self._initialize_bash_folder()
- if file.exists():
- logger.info("File already exists at %s.", file)
- self.warning_renderer.render(
- f"The integration is already present and enabled at {file}! "
- "Restart your terminal or source ~/.bashrc in case it's not working."
- )
- return
+def _remove_bash_functions(utils, file: Path) -> None:
+ """Remove a bash integration."""
+ if not file.exists():
+ logger.debug("Couldn't find integration file at '%s'", str(file))
+ utils.render_warning(
+ "It seems that the integration is not enabled. Skipping operation."
+ )
+ return
- write_file(contents, file)
- self.text_renderer.render(
- f"Integration successfully added at {file}. "
- "In order to use it, please restart your terminal or source ~/.bashrc"
+ try:
+ file.unlink()
+ utils.render_success("Integration disabled successfully.")
+ except (FileExistsError, FileNotFoundError) as e:
+ logger.warning(
+ "Got an exception '%s'. Either file is missing or something removed just before this operation",
+ str(e),
)
- def _remove_bash_functions(self, file: Path) -> None:
- """Internal function to remove a bash integration
- Arguments:
- file (Path): The path object with the correct location
- """
- if not file.exists():
- logger.debug("Couldn't find integration file at '%s'", str(file))
- self.warning_renderer.render(
- "It seems that the integration is not enabled. Skipping operation."
- )
- return
-
- try:
- file.unlink()
- self.text_renderer.render("Integration disabled successfully.")
- except (FileExistsError, FileNotFoundError) as e:
- logger.warning(
- "Got an exception '%s'. Either file is missing or something removed just before this operation",
- str(e),
- )
+@command("shell", help="Manage shell integrations")
+@argument("--enable-capture", action="store_true", help="Enable terminal capture for the current terminal session.")
+@argument("--enable-interactive", action="store_true", help="Enable the shell integration for interactive mode on the system. Currently, only BASH is supported. After the interactive was sourced, hit Ctrl + G in your terminal to enable interactive mode.")
+@argument("--disable-interactive", action="store_true", help="Disable the shell integration for interactive mode on the system.")
+def shell_command(args: Namespace, context: CommandContext) -> int:
+ """Shell command implementation."""
+ utils = create_utils(context)
+
+ try:
+ if args.enable_interactive:
+ return _enable_interactive(utils)
+ elif args.disable_interactive:
+ return _disable_interactive(utils)
+ elif args.enable_capture:
+ return _enable_capture(utils)
+ else:
+ utils.render_error("No operation specified. Use --help for available options.")
+ return 1
+
+ except ShellCommandException as e:
+ logger.info("Failed to execute shell command: %s", str(e))
+ utils.render_error(str(e))
+ return e.code
-# Register operations using the decorator
-@ShellOperationFactory.register(ShellOperationType.ENABLE_INTERACTIVE)
-class EnableInteractiveMode(BaseShellOperation):
- """Class to hold the enable interactive mode operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- self._write_bash_functions(INTERACTIVE_MODE_INTEGRATION_FILE, BASH_INTERACTIVE)
+def _enable_interactive(utils) -> int:
+ """Enable interactive mode operation."""
+ _write_bash_functions(utils, INTERACTIVE_MODE_INTEGRATION_FILE, BASH_INTERACTIVE)
+ return 0
-@ShellOperationFactory.register(ShellOperationType.DISABLE_INTERACTIVE)
-class DisableInteractiveMode(BaseShellOperation):
- """Class to hold the disable interactive mode operation"""
-
- def execute(self) -> None:
- """Default method to execute the operation"""
- self._remove_bash_functions(INTERACTIVE_MODE_INTEGRATION_FILE)
+def _disable_interactive(utils) -> int:
+ """Disable interactive mode operation."""
+ _remove_bash_functions(utils, INTERACTIVE_MODE_INTEGRATION_FILE)
+ return 0
-@ShellOperationFactory.register(ShellOperationType.ENABLE_CAPTURE)
-class EnableTerminalCapture(BaseShellOperation):
- """Class to hold the enable terminal capture operation"""
+def _enable_capture(utils) -> int:
+ """Enable terminal capture operation."""
+ file_lock = NamedFileLock(name="terminal")
- def execute(self) -> None:
- """Default method to execute the operation"""
- file_lock = NamedFileLock(name="terminal")
+ if file_lock.is_locked:
+ raise ShellCommandException(
+ f"Detected a terminal capture session running with pid '{file_lock.pid}'."
+ " In order to start a new terminal capture session, you must stop the previous one."
+ )
- if file_lock.is_locked:
- raise ShellCommandException(
- f"Detected a terminal capture session running with pid '{file_lock.pid}'."
- " In order to start a new terminal capture session, you must stop the previous one."
- )
-
- with file_lock:
- self.text_renderer.render(
- "Starting terminal reader. Press Ctrl + D to stop the capturing."
- )
- self.text_renderer.render(
- f"Terminal capture log is being written to {TERMINAL_CAPTURE_FILE}"
- )
- self._initialize_bash_folder()
- start_capturing()
-
-
-class ShellCommand(BaseCLICommand):
- """Class that represents the history command."""
-
- def run(self) -> int:
- """Main entrypoint for the command to run.
-
- Returns:
- int: Return the status code for the operation
- """
- error_renderer: TextRenderer = create_error_renderer()
- operation_factory = ShellOperationFactory()
- try:
- # Get and execute the appropriate operation
- operation = operation_factory.create_operation(
- self._args, self._context, error_renderer=error_renderer
- )
- if operation:
- operation.execute()
-
- return 0
- except ShellCommandException as e:
- logger.info("Failed to execute shell command: %s", str(e))
- error_renderer.render(str(e))
- return e.code
-
-
-def register_subcommand(parser: SubParsersAction):
- """
- Register this command to argparse so it's available for the root parser.
-
- Arguments:
- parser (SubParsersAction): Root parser to register command-specific arguments
- """
- shell_parser = create_subparser(parser, "shell", "Manage shell integrations")
-
- terminal_capture_group = shell_parser.add_argument_group("Terminal Capture Options")
- terminal_capture_group.add_argument(
- "--enable-capture",
- action="store_true",
- help="Enable terminal capture for the current terminal session.",
- )
-
- interactive_mode = shell_parser.add_argument_group("Interactive Mode Options")
- interactive_mode.add_argument(
- "--enable-interactive",
- action="store_true",
- help=(
- "Enable the shell integration for interactive mode on the system. "
- "Currently, only BASH is supported. After the interactive was "
- "sourced, hit Ctrl + G in your terminal to enable interactive mode."
- ),
- )
- interactive_mode.add_argument(
- "--disable-interactive",
- action="store_true",
- help="Disable the shell integration for interactive mode on the system.",
- )
-
- shell_parser.set_defaults(func=_command_factory)
-
-
-def _command_factory(args: Namespace) -> ShellCommand:
- """Internal command factory to create the command class
-
- Arguments:
- args (Namespace): The arguments processed with argparse.
-
- Returns:
- ShellCommand: Return an instance of class
- """
- return ShellCommand(args)
+ with file_lock:
+ utils.render_success(
+ "Starting terminal reader. Press Ctrl + D to stop the capturing."
+ )
+ utils.render_success(
+ f"Terminal capture log is being written to {TERMINAL_CAPTURE_FILE}"
+ )
+ _initialize_bash_folder()
+ start_capturing()
+
+ return 0
\ No newline at end of file
diff --git a/command_line_assistant/commands/utils.py b/command_line_assistant/commands/utils.py
new file mode 100644
index 0000000..129c7f3
--- /dev/null
+++ b/command_line_assistant/commands/utils.py
@@ -0,0 +1,126 @@
+"""Utilities for simplified command implementation."""
+
+import logging
+from typing import Optional, cast
+
+from command_line_assistant.dbus.constants import (
+ CHAT_IDENTIFIER,
+ HISTORY_IDENTIFIER,
+ USER_IDENTIFIER,
+)
+from command_line_assistant.dbus.interfaces.chat import ChatInterface
+from command_line_assistant.dbus.interfaces.history import HistoryInterface
+from command_line_assistant.dbus.interfaces.user import UserInterface
+from command_line_assistant.rendering.renders.text import TextRenderer
+from command_line_assistant.utils.cli import CommandContext
+from command_line_assistant.utils.renderers import (
+ create_error_renderer,
+ create_text_renderer,
+ create_warning_renderer,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class CommandUtils:
+ """Utility class providing common functionality for commands."""
+
+ def __init__(self, context: CommandContext):
+ """Initialize command utilities.
+
+ Args:
+ context: Command context
+ """
+ self.context = context
+ self._text_renderer: Optional[TextRenderer] = None
+ self._warning_renderer: Optional[TextRenderer] = None
+ self._error_renderer: Optional[TextRenderer] = None
+ self._chat_proxy: Optional[ChatInterface] = None
+ self._history_proxy: Optional[HistoryInterface] = None
+ self._user_proxy: Optional[UserInterface] = None
+
+ @property
+ def text_renderer(self) -> TextRenderer:
+ """Get text renderer instance."""
+ if self._text_renderer is None:
+ self._text_renderer = create_text_renderer()
+ return self._text_renderer
+
+ @property
+ def warning_renderer(self) -> TextRenderer:
+ """Get warning renderer instance."""
+ if self._warning_renderer is None:
+ self._warning_renderer = create_warning_renderer()
+ return self._warning_renderer
+
+ @property
+ def error_renderer(self) -> TextRenderer:
+ """Get error renderer instance."""
+ if self._error_renderer is None:
+ self._error_renderer = create_error_renderer()
+ return self._error_renderer
+
+ @property
+ def chat_proxy(self) -> ChatInterface:
+ """Get chat proxy instance."""
+ if self._chat_proxy is None:
+ self._chat_proxy = cast(ChatInterface, CHAT_IDENTIFIER.get_proxy())
+ return self._chat_proxy
+
+ @property
+ def history_proxy(self) -> HistoryInterface:
+ """Get history proxy instance."""
+ if self._history_proxy is None:
+ self._history_proxy = cast(HistoryInterface, HISTORY_IDENTIFIER.get_proxy())
+ return self._history_proxy
+
+ @property
+ def user_proxy(self) -> UserInterface:
+ """Get user proxy instance."""
+ if self._user_proxy is None:
+ self._user_proxy = cast(UserInterface, USER_IDENTIFIER.get_proxy())
+ return self._user_proxy
+
+ def get_user_id(self) -> str:
+ """Get user ID for current context.
+
+ Returns:
+ User ID string
+ """
+ return self.user_proxy.GetUserId(self.context.effective_user_id)
+
+ def render_success(self, message: str) -> None:
+ """Render a success message.
+
+ Args:
+ message: Success message to render
+ """
+ self.text_renderer.render(message)
+
+ def render_warning(self, message: str) -> None:
+ """Render a warning message.
+
+ Args:
+ message: Warning message to render
+ """
+ self.warning_renderer.render(message)
+
+ def render_error(self, message: str) -> None:
+ """Render an error message.
+
+ Args:
+ message: Error message to render
+ """
+ self.error_renderer.render(message)
+
+
+def create_utils(context: CommandContext) -> CommandUtils:
+ """Create command utilities instance.
+
+ Args:
+ context: Command context
+
+ Returns:
+ CommandUtils instance
+ """
+ return CommandUtils(context)
\ No newline at end of file
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment