|
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 |