vs-cli

#!/usr/bin/env python3
"""
vs-cli: Send commands to a Docker container's console via docker attach.

This script connects to a running container and sends commands to the stdin
of PID 1, capturing output based on specified parameters.
"""

import argparse
import subprocess
import sys
import time
import select
from typing import Optional


class ContainerCommandError(Exception):
    """Base exception for container command errors."""
    pass


def send_command_to_container(
    container: str,
    command: str,
    response_lines: Optional[int] = None,
    response_timeout: Optional[float] = None,
    no_response: bool = False,
    verbose: bool = False
) -> int:
    """
    Send a command to a Docker container and optionally capture output.

    Args:
        container: Container name or ID
        command: Command to send (newline will be added automatically)
        response_lines: Number of lines to capture before exiting
        response_timeout: Seconds to capture output before exiting
        no_response: If True, don't capture any output
        verbose: If True, output debug information

    Returns:
        0 on success, 1 on error
    """
    process = None

    try:
        # Build docker attach command
        docker_cmd = ['docker', 'attach', '--sig-proxy=false', container]

        if verbose:
            print(f"[DEBUG] Executing: {' '.join(docker_cmd)}", file=sys.stderr)
            print(f"[DEBUG] Sending command: {command}", file=sys.stderr)

        # Start the docker attach process
        # Using binary mode for full control over input/output
        process = subprocess.Popen(
            docker_cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=0  # Unbuffered for real-time output
        )

        # Small delay to ensure connection is established
        time.sleep(0.1)

        # Send the command with newline
        command_bytes = (command + '\n').encode('utf-8')
        process.stdin.write(command_bytes)
        process.stdin.flush()

        if verbose:
            print("[DEBUG] Command sent successfully", file=sys.stderr)

        # Handle output based on mode
        if no_response:
            # No output needed, just wait briefly for command to be processed
            if verbose:
                print("[DEBUG] No response mode - waiting briefly before detaching", file=sys.stderr)
            time.sleep(0.2)

        elif response_lines is not None:
            # Capture specific number of lines
            if verbose:
                print(f"[DEBUG] Capturing {response_lines} lines", file=sys.stderr)

            lines_captured = 0
            start_time = time.time()
            timeout = 30  # Upper limit timeout for line capture

            while lines_captured < response_lines:
                # Check for timeout
                if time.time() - start_time > timeout:
                    if verbose:
                        print(f"[DEBUG] Timeout reached while waiting for lines", file=sys.stderr)
                    break

                # Use select to check for data (non-blocking)
                if sys.platform != 'win32':
                    ready, _, _ = select.select([process.stdout], [], [], 0.1)
                    if ready:
                        line = process.stdout.readline()
                        if line:
                            print(line.decode('utf-8', errors='replace'), end='')
                            lines_captured += 1
                        else:
                            # End of stream
                            break
                else:
                    # Windows doesn't support select on pipes, use blocking read with timeout
                    try:
                        line = process.stdout.readline()
                        if line:
                            print(line.decode('utf-8', errors='replace'), end='')
                            lines_captured += 1
                        else:
                            break
                    except Exception:
                        time.sleep(0.1)

        elif response_timeout is not None:
            # Capture output for specified duration
            if verbose:
                print(f"[DEBUG] Capturing output for {response_timeout} seconds", file=sys.stderr)

            start_time = time.time()

            while time.time() - start_time < response_timeout:
                remaining = response_timeout - (time.time() - start_time)
                if remaining <= 0:
                    break

                # Use select to check if data is available
                if sys.platform != 'win32':
                    timeout_select = min(remaining, 0.1)
                    ready, _, _ = select.select([process.stdout], [], [], timeout_select)
                    if ready:
                        line = process.stdout.readline()
                        if line:
                            print(line.decode('utf-8', errors='replace'), end='')
                else:
                    # Windows fallback
                    try:
                        line = process.stdout.readline()
                        if line:
                            print(line.decode('utf-8', errors='replace'), end='')
                    except Exception:
                        time.sleep(0.1)

        # Send detach sequence (Ctrl+P Ctrl+Q: \x10\x11)
        if verbose:
            print("[DEBUG] Sending detach sequence (Ctrl+P Ctrl+Q)", file=sys.stderr)

        detach_sequence = b'\x10\x11'
        process.stdin.write(detach_sequence)
        process.stdin.flush()

        # Give it a moment to process the detach
        time.sleep(0.2)

        # Close stdin
        process.stdin.close()

        # Wait for process to exit (with timeout)
        try:
            process.wait(timeout=2)
        except subprocess.TimeoutExpired:
            if verbose:
                print("[DEBUG] Process didn't exit cleanly, terminating", file=sys.stderr)
            process.terminate()
            try:
                process.wait(timeout=1)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait()

        if verbose:
            print("[DEBUG] Detached successfully", file=sys.stderr)

        return 0

    except FileNotFoundError:
        error_msg = "Error: 'docker' command not found. Is Docker installed and in PATH?"
        if verbose:
            import traceback
            print(error_msg, file=sys.stderr)
            traceback.print_exc()
        else:
            print(error_msg, file=sys.stderr)
        return 1

    except subprocess.CalledProcessError as e:
        error_msg = f"Error: Docker command failed with exit code {e.returncode}"
        if verbose:
            import traceback
            print(error_msg, file=sys.stderr)
            traceback.print_exc()
        else:
            print(error_msg, file=sys.stderr)
        return 1

    except PermissionError:
        error_msg = "Error: Permission denied. Do you have permission to access Docker?"
        if verbose:
            import traceback
            print(error_msg, file=sys.stderr)
            traceback.print_exc()
        else:
            print(error_msg, file=sys.stderr)
        return 1

    except KeyboardInterrupt:
        error_msg = "Error: Interrupted by user"
        print(error_msg, file=sys.stderr)
        return 1

    except Exception as e:
        error_msg = f"Error: {type(e).__name__}: {str(e)}"
        if verbose:
            import traceback
            print(error_msg, file=sys.stderr)
            traceback.print_exc()
        else:
            print(error_msg, file=sys.stderr)
        return 1

    finally:
        # Ensure process is cleaned up
        if process is not None and process.poll() is None:
            try:
                process.terminate()
                process.wait(timeout=1)
            except Exception:
                try:
                    process.kill()
                except Exception:
                    pass


def main():
    """Main entry point for the script."""
    parser = argparse.ArgumentParser(
        description='Send commands to a Docker container console via docker attach',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  vs-cli --name vs-server-1 /gm 2
  vs-cli --name vs-server-1 --response-lines 5 /status
  vs-cli --name vs-server-1 --response-timeout 2.5 /list players
  vs-cli --name vs-server-1 --no-response-output /save world

Notes:
  - Commands are automatically terminated with a newline
  - The script uses Ctrl+P Ctrl+Q to detach from the container
  - The container continues running after the script exits
  - Use --verbose to see debug output and full tracebacks
        """
    )

    parser.add_argument(
        '--name',
        dest='container',
        required=True,
        metavar='CONTAINER',
        help='Container name or ID'
    )

    # Output mode arguments (mutually exclusive)
    output_group = parser.add_mutually_exclusive_group()
    output_group.add_argument(
        '--response-lines',
        type=int,
        metavar='X',
        help='Capture and return X lines of output then exit'
    )
    output_group.add_argument(
        '--response-timeout',
        type=float,
        metavar='X',
        help='Capture output for X seconds then exit'
    )
    output_group.add_argument(
        '--no-response-output',
        action='store_true',
        help='Send command without capturing any output'
    )

    parser.add_argument(
        '--verbose',
        action='store_true',
        help='Enable verbose output with debug information and tracebacks'
    )

    parser.add_argument(
        'command',
        nargs='+',
        help='Command to send to the container (arguments will be joined with spaces)'
    )

    args = parser.parse_args()

    # Validate response-lines value
    if args.response_lines is not None and args.response_lines <= 0:
        print("Error: --response-lines must be a positive integer", file=sys.stderr)
        sys.exit(1)

    # Validate response-timeout value
    if args.response_timeout is not None and args.response_timeout <= 0:
        print("Error: --response-timeout must be a positive number", file=sys.stderr)
        sys.exit(1)

    # Join command arguments with spaces
    command = ' '.join(args.command)

    # Execute the command
    exit_code = send_command_to_container(
        container=args.container,
        command=command,
        response_lines=args.response_lines,
        response_timeout=args.response_timeout,
        no_response=args.no_response_output,
        verbose=args.verbose
    )

    sys.exit(exit_code)


if __name__ == '__main__':
    main()