mirror of
https://github.com/Ayymoss/Plutainer
synced 2026-08-24 14:32:06 -04:00
103 lines
2.7 KiB
Python
103 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
rcon-cli — Interactive and one-shot RCON client for Plutainer.
|
|
|
|
Usage:
|
|
docker exec -i <container> rcon-cli # interactive mode
|
|
docker exec <container> rcon-cli <command> # one-shot mode
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
# Resolve symlinks so SCRIPT_DIR points to the real location,
|
|
# not the symlink's directory (e.g. /usr/local/bin/)
|
|
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
|
|
sys.path.insert(0, SCRIPT_DIR)
|
|
|
|
import pyquake3
|
|
|
|
|
|
def resolve_config():
|
|
"""Call game-config.sh to resolve port, config path, and RCON password."""
|
|
script = f"""
|
|
source "{SCRIPT_DIR}/game-config.sh"
|
|
detect_game_type || exit 1
|
|
ACTIVE_PORT="${{CUSTOM_PORT}}"
|
|
if [[ -z "$ACTIVE_PORT" ]]; then
|
|
resolve_default_port || exit 1
|
|
ACTIVE_PORT="${{DEFAULT_PORT}}"
|
|
fi
|
|
resolve_config_path || exit 1
|
|
extract_rcon_password || exit 1
|
|
echo "$ACTIVE_PORT"
|
|
echo "$CONFIG_PATH"
|
|
echo "$RCON_PASSWORD"
|
|
echo "$GAME_NAME"
|
|
echo "$GAME_TYPE"
|
|
"""
|
|
result = subprocess.run(
|
|
["bash", "-c", script],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
print(result.stderr.strip(), file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
lines = result.stdout.strip().split("\n")
|
|
if len(lines) < 5:
|
|
print("Error: Failed to resolve game configuration.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
return {
|
|
"port": int(lines[0]),
|
|
"config_path": lines[1],
|
|
"rcon_password": lines[2],
|
|
"game_name": lines[3],
|
|
"game_type": lines[4],
|
|
}
|
|
|
|
|
|
def send_command(server, cmd):
|
|
try:
|
|
_, response = server.rcon(cmd)
|
|
print(response.strip())
|
|
except Exception as e:
|
|
print(f"RCON error: {e}", file=sys.stderr)
|
|
|
|
|
|
def main():
|
|
config = resolve_config()
|
|
port = config["port"]
|
|
server = pyquake3.PyQuake3(f"127.0.0.1:{port}", rcon_password=config["rcon_password"])
|
|
|
|
# One-shot mode: arguments are the command
|
|
if len(sys.argv) > 1:
|
|
cmd = " ".join(sys.argv[1:])
|
|
send_command(server, cmd)
|
|
return
|
|
|
|
# Interactive mode
|
|
print(f"Connected to {config['game_name']} ({config['game_type']}) at 127.0.0.1:{port}")
|
|
print("Type 'quit' or 'exit' to close. Ctrl+C also works.")
|
|
print()
|
|
try:
|
|
while True:
|
|
try:
|
|
cmd = input("> ")
|
|
except EOFError:
|
|
break
|
|
cmd = cmd.strip()
|
|
if not cmd:
|
|
continue
|
|
if cmd.lower() in ("quit", "exit"):
|
|
break
|
|
send_command(server, cmd)
|
|
except KeyboardInterrupt:
|
|
print()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|