From 5c4e2116c8625595863e69c2b3cd3dd632e715ad Mon Sep 17 00:00:00 2001 From: Fede Luzzi Date: Tue, 12 May 2026 13:19:06 -0300 Subject: [PATCH] refactor to support future command center --- connpy/cli/terminal_ui.py | 137 ++++++++++++---------------------- connpy/core.py | 2 +- connpy/grpc_layer/stubs.py | 10 ++- connpy/services/ai_service.py | 41 ++++++++++ 4 files changed, 95 insertions(+), 95 deletions(-) diff --git a/connpy/cli/terminal_ui.py b/connpy/cli/terminal_ui.py index d02ec02..3b66061 100644 --- a/connpy/cli/terminal_ui.py +++ b/connpy/cli/terminal_ui.py @@ -18,99 +18,24 @@ from prompt_toolkit.formatted_text import HTML from prompt_toolkit.history import InMemoryHistory from ..printer import connpy_theme - -def log_cleaner(data: str) -> str: - """ - Stateless version of _logclean to remove ANSI sequences and process cursor movements. - """ - if not data: - return "" - - lines = data.split('\n') - cleaned_lines = [] - - # Regex to capture: ANSI sequences, control characters (\r, \b, etc), and plain text chunks - token_re = re.compile(r'(\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/ ]*[@-~])|\r|\b|\x7f|[\x00-\x1F]|[^\x1B\r\b\x7f\x00-\x1F]+)') - - for line in lines: - buffer = [] - cursor = 0 - - for token in token_re.findall(line): - if token == '\r': - cursor = 0 - elif token in ('\b', '\x7f'): - if cursor > 0: - cursor -= 1 - elif token == '\x1B[D': # Left Arrow - if cursor > 0: - cursor -= 1 - elif token == '\x1B[C': # Right Arrow - if cursor < len(buffer): - cursor += 1 - elif token == '\x1B[K': # Clear to end of line - buffer = buffer[:cursor] - elif token.startswith('\x1B'): - continue - elif len(token) == 1 and ord(token) < 32: - continue - else: - for char in token: - if cursor == len(buffer): - buffer.append(char) - else: - buffer[cursor] = char - cursor += 1 - cleaned_lines.append("".join(buffer)) - - return "\n".join(cleaned_lines).replace('\n\n', '\n').strip() +from connpy.utils import log_cleaner +from ..services.ai_service import AIService class CopilotInterface: - def __init__(self, config, history=None): + def __init__(self, config, history=None, pt_input=None, pt_output=None, rich_file=None): self.config = config - self.console = Console(theme=connpy_theme) self.history = history or InMemoryHistory() + self.pt_input = pt_input + self.pt_output = pt_output + self.ai_service = AIService(config) + + if rich_file: + self.console = Console(theme=connpy_theme, force_terminal=True, file=rich_file) + else: + self.console = Console(theme=connpy_theme) + self.mode_range, self.mode_single, self.mode_lines = 0, 1, 2 - def extract_blocks(self, raw_bytes: bytes, cmd_byte_positions: List[tuple], node_info: dict) -> List[tuple]: - """Identifies command blocks in the terminal history.""" - blocks = [] - if not (cmd_byte_positions and len(cmd_byte_positions) >= 2 and raw_bytes): - return blocks - - default_prompt = r'>$|#$|\$$|>.$|#.$|\$.$' - device_prompt = node_info.get("prompt", default_prompt) if isinstance(node_info, dict) else default_prompt - prompt_re_str = re.sub(r'(?Send? (y/n/e/number) [n]: "), key_bindings=c_bindings) + action = await confirm_session.prompt_async(HTML(f"Send? (y/n/e/range) [n]: "), key_bindings=c_bindings) except (KeyboardInterrupt, EOFError): action = "n" + def parse_indices(text, max_len): + """Helper to parse '1-3, 5, 7' into [0, 1, 2, 4, 6].""" + indices = [] + # Replace commas with spaces and split + parts = text.replace(',', ' ').split() + for part in parts: + if '-' in part: + try: + start, end = map(int, part.split('-')) + # Ensure inclusive and 0-indexed + indices.extend(range(start-1, end)) + except: continue + elif part.isdigit(): + indices.append(int(part)-1) + # Filter valid indices and remove duplicates + return [i for i in sorted(set(indices)) if 0 <= i < max_len] + action_l = (action or "n").lower().strip() if action_l in ('y', 'yes', 'all'): return "send_all", commands, None + + # Check for numeric selection (e.g., "1, 2-4") + if re.match(r'^[0-9,\-\s]+$', action_l): + selected_idxs = parse_indices(action_l, len(commands)) + if selected_idxs: + return "send_all", [commands[i] for i in selected_idxs], None + elif action_l.startswith('e'): - target = "\n".join(commands) + # Check if it's a selective edit like 'e1-2' + selection_str = action_l[1:].strip() + if selection_str: + idxs = parse_indices(selection_str, len(commands)) + cmds_to_edit = [commands[i] for i in idxs] if idxs else commands + else: + cmds_to_edit = commands + + target = "\n".join(cmds_to_edit) e_bindings = KeyBindings() @e_bindings.add('c-j') def _(ev): ev.app.exit(result=ev.app.current_buffer.text) diff --git a/connpy/core.py b/connpy/core.py index d18e43e..ae9152a 100755 --- a/connpy/core.py +++ b/connpy/core.py @@ -258,7 +258,7 @@ class node: @MethodHook def _logclean(self, logfile, var = False): """Remove special ascii characters and process terminal cursor movements to clean logs.""" - from .cli.terminal_ui import log_cleaner + from .utils import log_cleaner if var == False: try: diff --git a/connpy/grpc_layer/stubs.py b/connpy/grpc_layer/stubs.py index 93e0e0d..2db457f 100644 --- a/connpy/grpc_layer/stubs.py +++ b/connpy/grpc_layer/stubs.py @@ -9,7 +9,8 @@ from .utils import to_value, from_value, to_struct, from_struct from ..services.exceptions import ConnpyError from ..hooks import MethodHook from .. import printer -from ..cli.terminal_ui import log_cleaner, CopilotInterface +from ..cli.terminal_ui import CopilotInterface +from ..utils import log_cleaner def handle_errors(func): @wraps(func) @@ -87,11 +88,12 @@ class NodeStub: # Prepare final action for server action_sent = "cancel" - if action == "send_all": - action_sent = "send_all" + if action == "send_all" and commands: + # In remote mode, send the selected commands as a custom block + # so the server executes exactly what the user picked (e.g., selection '1') + action_sent = f"custom:{chr(10).join(commands)}" elif action == "custom" and custom_cmd: action_sent = f"custom:{chr(10).join(custom_cmd)}" - request_queue.put(connpy_pb2.InteractRequest(copilot_action=action_sent)) resume_generator() tty.setraw(sys.stdin.fileno()) diff --git a/connpy/services/ai_service.py b/connpy/services/ai_service.py index acd9715..6d2b86e 100644 --- a/connpy/services/ai_service.py +++ b/connpy/services/ai_service.py @@ -1,9 +1,50 @@ +import re from .base import BaseService from .exceptions import InvalidConfigurationError +from connpy.utils import log_cleaner class AIService(BaseService): """Business logic for interacting with AI agents and LLM configurations.""" + def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict) -> list: + """Identifies command blocks in the terminal history.""" + blocks = [] + if not (cmd_byte_positions and len(cmd_byte_positions) >= 2 and raw_bytes): + return blocks + + default_prompt = r'>$|#$|\$$|>.$|#.$|\$.$' + device_prompt = node_info.get("prompt", default_prompt) if isinstance(node_info, dict) else default_prompt + prompt_re_str = re.sub(r'(?