mirror of
https://github.com/johndoe6345789/docker-swarm-termina.git
synced 2026-04-24 13:45:01 +00:00
985c98339a
Split monolithic 631-line app.py into focused modules: **Structure:** - config.py - Configuration and shared state - utils/ - Utility functions (1-2 functions per file) - auth.py - Authentication helpers - docker_client.py - Docker client getter - exec_helpers.py - Command execution helpers - formatters.py - Data formatting utilities - terminal_helpers.py - Terminal operation helpers - diagnostics/docker_env.py - Docker diagnostics - routes/ - HTTP endpoints (1 endpoint per file) - login.py, logout.py, health.py - containers/list.py, exec.py, start.py, stop.py, restart.py, remove.py - handlers/ - WebSocket handlers (1 handler per file) - terminal/connect.py, disconnect.py, start.py, input.py, resize.py, register.py **Improvements:** - Reduced function complexity (from 21 locals to 18 max) - Fixed all pylint import order issues - Removed unused imports (select, timedelta, stat) - Applied lazy logging formatting throughout - Added comprehensive docstrings - Each file has focused responsibility - Easier to test, maintain, and extend **Pylint score improvement:** - Before: 25 problems (15 errors, 10 warnings) - After: Only duplicate code warnings (expected for similar routes) https://claude.ai/code/session_011PzvkCnVrsatoxbY3HbGXz
109 lines
3.0 KiB
Python
109 lines
3.0 KiB
Python
"""Helper functions for container exec operations."""
|
|
|
|
|
|
def build_bash_command(current_workdir, user_command, is_cd_command):
|
|
"""Build bash command for execution.
|
|
|
|
Args:
|
|
current_workdir: Current working directory
|
|
user_command: User's command
|
|
is_cd_command: Whether this is a cd command
|
|
|
|
Returns:
|
|
list: Command array for Docker exec
|
|
"""
|
|
path_export = 'export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'
|
|
|
|
if is_cd_command:
|
|
target_dir = user_command.strip()[3:].strip() or '~'
|
|
resolve_command = f'cd "{current_workdir}" && cd {target_dir} && pwd'
|
|
return ['/bin/bash', '-c', f'{path_export}; {resolve_command}']
|
|
|
|
return [
|
|
'/bin/bash', '-c',
|
|
f'{path_export}; cd "{current_workdir}" && {user_command}; echo "::WORKDIR::$(pwd)"'
|
|
]
|
|
|
|
|
|
def build_sh_command(current_workdir, user_command, is_cd_command):
|
|
"""Build sh command for execution (fallback).
|
|
|
|
Args:
|
|
current_workdir: Current working directory
|
|
user_command: User's command
|
|
is_cd_command: Whether this is a cd command
|
|
|
|
Returns:
|
|
list: Command array for Docker exec
|
|
"""
|
|
path_export = 'export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'
|
|
|
|
if is_cd_command:
|
|
target_dir = user_command.strip()[3:].strip() or '~'
|
|
resolve_command = f'cd "{current_workdir}" && cd {target_dir} && pwd'
|
|
return ['/bin/sh', '-c', f'{path_export}; {resolve_command}']
|
|
|
|
return [
|
|
'/bin/sh', '-c',
|
|
f'{path_export}; cd "{current_workdir}" && {user_command}; echo "::WORKDIR::$(pwd)"'
|
|
]
|
|
|
|
|
|
def execute_in_container(container, command):
|
|
"""Execute command in container.
|
|
|
|
Args:
|
|
container: Docker container object
|
|
command: Command to execute
|
|
|
|
Returns:
|
|
Docker exec instance
|
|
"""
|
|
return container.exec_run(
|
|
command,
|
|
stdout=True,
|
|
stderr=True,
|
|
stdin=False,
|
|
tty=True,
|
|
environment={'TERM': 'xterm-256color', 'LANG': 'C.UTF-8'}
|
|
)
|
|
|
|
|
|
def decode_output(exec_instance):
|
|
"""Decode exec output with fallback encoding.
|
|
|
|
Args:
|
|
exec_instance: Docker exec instance
|
|
|
|
Returns:
|
|
str: Decoded output
|
|
"""
|
|
if not exec_instance.output:
|
|
return ''
|
|
|
|
try:
|
|
return exec_instance.output.decode('utf-8')
|
|
except UnicodeDecodeError:
|
|
return exec_instance.output.decode('latin-1', errors='replace')
|
|
|
|
|
|
def extract_workdir(output, current_workdir, is_cd_command):
|
|
"""Extract working directory from command output.
|
|
|
|
Args:
|
|
output: Command output
|
|
current_workdir: Current working directory
|
|
is_cd_command: Whether this was a cd command
|
|
|
|
Returns:
|
|
tuple: (cleaned_output, new_workdir)
|
|
"""
|
|
if is_cd_command:
|
|
return '', output.strip()
|
|
|
|
if '::WORKDIR::' in output:
|
|
parts = output.rsplit('::WORKDIR::', 1)
|
|
return parts[0], parts[1].strip()
|
|
|
|
return output, current_workdir
|