mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2025-07-18 06:24:25 +02:00
Added new logic in cli proxy script to allow better restructuring of cli files
This commit is contained in:
parent
74ebb375d0
commit
8457325b5c
140
main.py
140
main.py
@ -1,6 +1,5 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
|
|
||||||
import argparse
|
|
||||||
import os
|
import os
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
@ -10,7 +9,8 @@ import signal
|
|||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
import pty
|
import pty
|
||||||
|
|
||||||
from cli.sounds import Sound
|
from cli.sounds import Sound # ensure Sound imported
|
||||||
|
|
||||||
|
|
||||||
def format_command_help(name, description, indent=2, col_width=36, width=80):
|
def format_command_help(name, description, indent=2, col_width=36, width=80):
|
||||||
prefix = " " * indent + f"{name:<{col_width - indent}}"
|
prefix = " " * indent + f"{name:<{col_width - indent}}"
|
||||||
@ -21,16 +21,42 @@ def format_command_help(name, description, indent=2, col_width=36, width=80):
|
|||||||
)
|
)
|
||||||
return wrapper.fill(description)
|
return wrapper.fill(description)
|
||||||
|
|
||||||
|
|
||||||
def list_cli_commands(cli_dir):
|
def list_cli_commands(cli_dir):
|
||||||
return sorted(
|
"""Recursively list all .py files under cli_dir that use argparse (without .py)."""
|
||||||
os.path.splitext(f.name)[0] for f in os.scandir(cli_dir)
|
cmds = []
|
||||||
if f.is_file() and f.name.endswith(".py") and not f.name.startswith("__")
|
for root, _, files in os.walk(cli_dir):
|
||||||
)
|
for f in files:
|
||||||
|
if not f.endswith(".py") or f.startswith("__"):
|
||||||
|
continue
|
||||||
|
path = os.path.join(root, f)
|
||||||
|
try:
|
||||||
|
with open(path, 'r', encoding='utf-8') as fh:
|
||||||
|
content = fh.read()
|
||||||
|
if 'argparse' not in content:
|
||||||
|
continue
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
rel_dir = os.path.relpath(root, cli_dir)
|
||||||
|
name = os.path.splitext(f)[0]
|
||||||
|
if rel_dir == ".":
|
||||||
|
cmd = (None, name)
|
||||||
|
else:
|
||||||
|
cmd = (rel_dir.replace(os.sep, "/"), name)
|
||||||
|
cmds.append(cmd)
|
||||||
|
return sorted(cmds, key=lambda x: (x[0] or "", x[1]))
|
||||||
|
|
||||||
|
|
||||||
def extract_description_via_help(cli_script_path):
|
def extract_description_via_help(cli_script_path):
|
||||||
try:
|
try:
|
||||||
|
# derive module name from script path
|
||||||
|
script_dir = os.path.dirname(os.path.realpath(__file__))
|
||||||
|
cli_dir = os.path.join(script_dir, "cli")
|
||||||
|
rel = os.path.relpath(cli_script_path, cli_dir)
|
||||||
|
module = "cli." + rel[:-3].replace(os.sep, ".")
|
||||||
|
|
||||||
result = subprocess.run(
|
result = subprocess.run(
|
||||||
[sys.executable, cli_script_path, "--help"],
|
[sys.executable, "-m", module, "--help"],
|
||||||
capture_output=True,
|
capture_output=True,
|
||||||
text=True,
|
text=True,
|
||||||
check=True
|
check=True
|
||||||
@ -39,7 +65,7 @@ def extract_description_via_help(cli_script_path):
|
|||||||
for i, line in enumerate(lines):
|
for i, line in enumerate(lines):
|
||||||
if line.strip().startswith("usage:"):
|
if line.strip().startswith("usage:"):
|
||||||
continue
|
continue
|
||||||
if line.strip() == "":
|
if not line.strip():
|
||||||
for j in range(i+1, len(lines)):
|
for j in range(i+1, len(lines)):
|
||||||
desc = lines[j].strip()
|
desc = lines[j].strip()
|
||||||
if desc:
|
if desc:
|
||||||
@ -48,14 +74,16 @@ def extract_description_via_help(cli_script_path):
|
|||||||
except Exception:
|
except Exception:
|
||||||
return "-"
|
return "-"
|
||||||
|
|
||||||
|
|
||||||
def git_clean_repo():
|
def git_clean_repo():
|
||||||
"""Remove all Git-ignored files and directories in the current repository."""
|
|
||||||
subprocess.run(['git', 'clean', '-Xfd'], check=True)
|
subprocess.run(['git', 'clean', '-Xfd'], check=True)
|
||||||
|
|
||||||
|
|
||||||
def play_start_intro():
|
def play_start_intro():
|
||||||
Sound.play_start_sound()
|
Sound.play_start_sound()
|
||||||
Sound.play_cymais_intro_sound()
|
Sound.play_cymais_intro_sound()
|
||||||
|
|
||||||
|
|
||||||
def failure_with_warning_loop():
|
def failure_with_warning_loop():
|
||||||
Sound.play_finished_failed_sound()
|
Sound.play_finished_failed_sound()
|
||||||
print("Warning: command failed. Press Ctrl+C to stop sound warnings.")
|
print("Warning: command failed. Press Ctrl+C to stop sound warnings.")
|
||||||
@ -65,28 +93,15 @@ def failure_with_warning_loop():
|
|||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
print("Warnings stopped by user.")
|
print("Warnings stopped by user.")
|
||||||
|
|
||||||
from cli.sounds import Sound # ensure Sound imported
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
# Parse special flags early and remove from args
|
# Early parse special flags
|
||||||
no_sound = False
|
no_sound = '--no-sound' in sys.argv and (sys.argv.remove('--no-sound') or True)
|
||||||
log_enabled = False
|
log_enabled = '--log' in sys.argv and (sys.argv.remove('--log') or True)
|
||||||
git_clean = False
|
git_clean = '--git-clean' in sys.argv and (sys.argv.remove('--git-clean') or True)
|
||||||
infinite = False
|
infinite = '--infinite' in sys.argv and (sys.argv.remove('--infinite') or True)
|
||||||
if '--no-sound' in sys.argv:
|
|
||||||
no_sound = True
|
|
||||||
sys.argv.remove('--no-sound')
|
|
||||||
if '--log' in sys.argv:
|
|
||||||
log_enabled = True
|
|
||||||
sys.argv.remove('--log')
|
|
||||||
if '--git-clean' in sys.argv:
|
|
||||||
git_clean = True
|
|
||||||
sys.argv.remove('--git-clean')
|
|
||||||
if '--infinite' in sys.argv:
|
|
||||||
infinite = True
|
|
||||||
sys.argv.remove('--infinite')
|
|
||||||
|
|
||||||
# Setup segfault handler to catch crashes
|
# Segfault handler
|
||||||
def segv_handler(signum, frame):
|
def segv_handler(signum, frame):
|
||||||
if not no_sound:
|
if not no_sound:
|
||||||
Sound.play_finished_failed_sound()
|
Sound.play_finished_failed_sound()
|
||||||
@ -99,23 +114,23 @@ if __name__ == "__main__":
|
|||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
signal.signal(signal.SIGSEGV, segv_handler)
|
signal.signal(signal.SIGSEGV, segv_handler)
|
||||||
|
|
||||||
# Play intro sounds
|
# Play intro sounds asynchronously
|
||||||
if not no_sound:
|
if not no_sound:
|
||||||
threading.Thread(target=play_start_intro, daemon=True).start()
|
threading.Thread(target=play_start_intro, daemon=True).start()
|
||||||
|
|
||||||
# Change to script directory
|
|
||||||
script_dir = os.path.dirname(os.path.realpath(__file__))
|
script_dir = os.path.dirname(os.path.realpath(__file__))
|
||||||
cli_dir = os.path.join(script_dir, "cli")
|
cli_dir = os.path.join(script_dir, "cli")
|
||||||
os.chdir(script_dir)
|
os.chdir(script_dir)
|
||||||
|
|
||||||
# If requested, clean git-ignored files
|
|
||||||
if git_clean:
|
if git_clean:
|
||||||
git_clean_repo()
|
git_clean_repo()
|
||||||
|
|
||||||
available_cli_commands = list_cli_commands(cli_dir)
|
# Collect available commands
|
||||||
|
available = list_cli_commands(cli_dir)
|
||||||
|
args = sys.argv[1:]
|
||||||
|
|
||||||
# Handle help invocation
|
# Global help
|
||||||
if len(sys.argv) == 1 or sys.argv[1] in ('-h', '--help'):
|
if not args or args[0] in ('-h', '--help'):
|
||||||
print("CyMaIS CLI – proxy to tools in ./cli/")
|
print("CyMaIS CLI – proxy to tools in ./cli/")
|
||||||
print("Usage: cymais [--no-sound] [--log] [--git-clean] [--infinite] <command> [options]")
|
print("Usage: cymais [--no-sound] [--log] [--git-clean] [--infinite] <command> [options]")
|
||||||
print("Options:")
|
print("Options:")
|
||||||
@ -125,25 +140,46 @@ if __name__ == "__main__":
|
|||||||
print(" --infinite Run the proxied command in an infinite loop")
|
print(" --infinite Run the proxied command in an infinite loop")
|
||||||
print(" -h, --help Show this help message and exit")
|
print(" -h, --help Show this help message and exit")
|
||||||
print("Available commands:")
|
print("Available commands:")
|
||||||
for cmd in available_cli_commands:
|
|
||||||
path = os.path.join(cli_dir, f"{cmd}.py")
|
current_folder = None
|
||||||
desc = extract_description_via_help(path)
|
for folder, cmd in available:
|
||||||
print(format_command_help(cmd, desc))
|
if folder != current_folder:
|
||||||
|
if folder:
|
||||||
|
print(f"{folder}/")
|
||||||
|
current_folder = folder
|
||||||
|
desc = extract_description_via_help(
|
||||||
|
os.path.join(cli_dir, *(folder.split('/') if folder else []), f"{cmd}.py")
|
||||||
|
)
|
||||||
|
# Slight indent for subcommands
|
||||||
|
print(format_command_help(cmd, desc, indent=2))
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
|
|
||||||
# Special-case per-command help
|
# Per-command help
|
||||||
if len(sys.argv) >= 3 and sys.argv[1] in available_cli_commands and sys.argv[2] in ('-h', '--help'):
|
for n in range(len(args), 0, -1):
|
||||||
subprocess.run([sys.executable, os.path.join(cli_dir, f"{sys.argv[1]}.py"), "--help"])
|
candidate = os.path.join(cli_dir, *args[:n]) + ".py"
|
||||||
sys.exit(0)
|
if os.path.isfile(candidate) and len(args) > n and args[n] in ('-h', '--help'):
|
||||||
|
# derive module
|
||||||
|
rel = os.path.relpath(candidate, cli_dir)
|
||||||
|
module = "cli." + rel[:-3].replace(os.sep, ".")
|
||||||
|
subprocess.run([sys.executable, "-m", module, args[n]])
|
||||||
|
sys.exit(0)
|
||||||
|
|
||||||
# Execute chosen command
|
# Resolve script path by longest matching prefix
|
||||||
parser = argparse.ArgumentParser(add_help=False)
|
script_path = None
|
||||||
parser.add_argument('cli_command', choices=available_cli_commands)
|
cli_args = []
|
||||||
parser.add_argument('cli_args', nargs=argparse.REMAINDER)
|
module = None
|
||||||
args = parser.parse_args()
|
for n in range(len(args), 0, -1):
|
||||||
|
candidate = os.path.join(cli_dir, *args[:n]) + ".py"
|
||||||
|
if os.path.isfile(candidate):
|
||||||
|
script_path = candidate
|
||||||
|
cli_args = args[n:]
|
||||||
|
rel = os.path.relpath(candidate, cli_dir)
|
||||||
|
module = "cli." + rel[:-3].replace(os.sep, ".")
|
||||||
|
break
|
||||||
|
|
||||||
cmd_path = os.path.join(cli_dir, f"{args.cli_command}.py")
|
if not module:
|
||||||
full_cmd = [sys.executable, cmd_path] + args.cli_args
|
print(f"Error: command '{' '.join(args)}' not found.")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
log_file = None
|
log_file = None
|
||||||
if log_enabled:
|
if log_enabled:
|
||||||
@ -152,9 +188,10 @@ if __name__ == "__main__":
|
|||||||
timestamp = datetime.now().strftime('%Y%m%dT%H%M%S')
|
timestamp = datetime.now().strftime('%Y%m%dT%H%M%S')
|
||||||
log_file_path = os.path.join(log_dir, f'{timestamp}.log')
|
log_file_path = os.path.join(log_dir, f'{timestamp}.log')
|
||||||
log_file = open(log_file_path, 'a', encoding='utf-8')
|
log_file = open(log_file_path, 'a', encoding='utf-8')
|
||||||
# 📖 Tip: Check your logs at the path below
|
|
||||||
print(f"📖 Tip: Log file created at {log_file_path}")
|
print(f"📖 Tip: Log file created at {log_file_path}")
|
||||||
|
|
||||||
|
full_cmd = [sys.executable, "-m", module] + cli_args
|
||||||
|
|
||||||
def run_once():
|
def run_once():
|
||||||
try:
|
try:
|
||||||
if log_enabled:
|
if log_enabled:
|
||||||
@ -189,7 +226,7 @@ if __name__ == "__main__":
|
|||||||
log_file.close()
|
log_file.close()
|
||||||
|
|
||||||
if rc != 0:
|
if rc != 0:
|
||||||
print(f"Command '{args.cli_command}' failed with exit code {rc}.")
|
print(f"Command '{os.path.basename(script_path)}' failed with exit code {rc}.")
|
||||||
failure_with_warning_loop()
|
failure_with_warning_loop()
|
||||||
sys.exit(rc)
|
sys.exit(rc)
|
||||||
else:
|
else:
|
||||||
@ -202,7 +239,6 @@ if __name__ == "__main__":
|
|||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
if infinite:
|
if infinite:
|
||||||
# ♾️ Infinite mode activated
|
|
||||||
print("♾️ Starting infinite execution mode...")
|
print("♾️ Starting infinite execution mode...")
|
||||||
count = 1
|
count = 1
|
||||||
while True:
|
while True:
|
||||||
|
51
tests/integration/test_cli_help.py
Normal file
51
tests/integration/test_cli_help.py
Normal file
@ -0,0 +1,51 @@
|
|||||||
|
import unittest
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import subprocess
|
||||||
|
|
||||||
|
class CLIHelpIntegrationTest(unittest.TestCase):
|
||||||
|
@classmethod
|
||||||
|
def setUpClass(cls):
|
||||||
|
# Projekt-Root ermitteln
|
||||||
|
cls.project_root = os.path.abspath(
|
||||||
|
os.path.join(os.path.dirname(__file__), '..', '..')
|
||||||
|
)
|
||||||
|
cls.main_py = os.path.join(cls.project_root, 'main.py')
|
||||||
|
cls.cli_dir = os.path.join(cls.project_root, 'cli')
|
||||||
|
cls.python = sys.executable
|
||||||
|
|
||||||
|
def test_all_cli_commands_help(self):
|
||||||
|
"""
|
||||||
|
Iteriere über alle .py Dateien in cli/, baue daraus die
|
||||||
|
Subcommand-Pfade und prüfe, dass `python main.py <cmd> --help`
|
||||||
|
mit Exit-Code 0 endet.
|
||||||
|
"""
|
||||||
|
for root, _, files in os.walk(self.cli_dir):
|
||||||
|
for fname in files:
|
||||||
|
if not fname.endswith('.py') or fname.startswith('__'):
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Bestimme Subcommand-Segmente
|
||||||
|
rel_dir = os.path.relpath(root, self.cli_dir)
|
||||||
|
cmd_name = os.path.splitext(fname)[0]
|
||||||
|
if rel_dir == '.':
|
||||||
|
segments = [cmd_name]
|
||||||
|
else:
|
||||||
|
segments = rel_dir.split(os.sep) + [cmd_name]
|
||||||
|
|
||||||
|
with self.subTest(command=' '.join(segments)):
|
||||||
|
cmd = [self.python, self.main_py] + segments + ['--help']
|
||||||
|
result = subprocess.run(
|
||||||
|
cmd, capture_output=True, text=True
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
result.returncode, 0,
|
||||||
|
msg=(
|
||||||
|
f"Command `{ ' '.join(cmd) }` failed\n"
|
||||||
|
f"stdout:\n{result.stdout}\n"
|
||||||
|
f"stderr:\n{result.stderr}"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
Loading…
x
Reference in New Issue
Block a user