Das deutsche Python-Forum

Code: Alles auswählen

def start_server(server_name, install_dir="servers"):
    global subprocess_server
    server_path = os.path.join(install_dir, server_name)

    # Ensure the start script exists
    run_script = "run.bat" if os.name == 'nt' else "run.sh"
    run_script_path = os.path.join(server_path, run_script)

    if not os.path.exists(run_script_path):
        print(f"Error: {run_script} not found. Ensure the server is correctly installed.")
        return False

    # Start the server
    try:
        if os.name == 'nt':  # Windows
            subprocess_server = subprocess.Popen(
                ["cmd", "/c", "start", "cmd", "/k", run_script],
                cwd=server_path
            )
        else:  # Unix-based systems
            subprocess_server = subprocess.Popen(
                ["bash", run_script],
                cwd=server_path,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
        print(f"Server '{server_name}' started.")
        return True
    except Exception as e:
        print(f"Error starting the server: {e}")
        return False

Code: Alles auswählen

import os
import subprocess
from pathlib import Path


def start_server(server_name, install_dir=Path("servers")):
    run_script_file_path = (install_dir / server_name / "run").with_suffix(
        ".bat" if os.name == "nt" else ".sh"
    )
    if not run_script_file_path.exists():
        raise ValueError(
            f"{run_script_file_path.name!r} not found."
            f" Ensure the server is correctly installed."
        )

    server_process = subprocess.Popen(
        [run_script_file_path.name], cwd=run_script_file_path.parent
    )
    #
    # TODO Don't use print for logging.
    #
    print(f"Server {server_name!r} started.")
    return server_process

Code: Alles auswählen

def stop_mc_server():
    global subprocess_server
    if subprocess_server:
        try:
            print("Stopping Minecraft server...")
            subprocess_server.stdin.write("stop\n")
            subprocess_server.stdin.flush()
            subprocess_server.wait()
            print("Server stopped.")
        except Exception as e:
            print(f"Error stopping the server: {e}")
        finally:
            subprocess_server = None
    else:
        print("No server is currently running.")

Code: Alles auswählen

def start_forge_server(server_name, install_dir="servers", min_memory="4G", max_memory="7G"):
    global global_forge_server_process
    try:
        server_path = os.path.join(install_dir, server_name)
        win_args_path = os.path.join(server_path, "libraries/net/minecraftforge/forge/1.20.2-48.1.0/win_args.txt")
        if not os.path.exists(win_args_path):
            raise FileNotFoundError("Required win_args.txt file not found in the server directory.")

        with open(win_args_path, "r") as file:
            win_args = file.read().strip().split()

        command = ["java", f"-Xms{min_memory}", f"-Xmx{max_memory}"] + win_args + ["nogui"]
        process = subprocess.Popen(command, cwd=server_path, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        global_forge_server_process = process

        def enqueue_output(out, log_entries):
            for line in iter(out.readline, ''):
                log_entries.append(line)
                if 'Done (' in line:
                    global server_status, status_color
                    server_status = "Eingeschaltet"
                    status_color = "on"
            out.close()

        threading.Thread(target=enqueue_output, args=(process.stdout, log_entries)).start()
        threading.Thread(target=enqueue_output, args=(process.stderr, log_entries)).start()
    except Exception as e:
        print(f"Error starting the server: {e}")
        
def send_command_to_server(command):
    global global_forge_server_process
    if global_forge_server_process and global_forge_server_process.stdin:
        try:
            global_forge_server_process.stdin.write(command + "\n")
            global_forge_server_process.stdin.flush()
        except Exception as e:
            print(f"Fehler beim Senden des Befehls: {e}")

Code: Alles auswählen

In [15]: " a b    c d \n".split()
Out[15]: ['a', 'b', 'c', 'd']

Code: Alles auswählen

import subprocess
from pathlib import Path
from threading import Thread


class Server:
    def __init__(self, name, install_path=Path("servers")):
        self.path = install_path / name
        self.process = None
        self.log_entries = []
        self.is_on = False

    def _process_output(self, file):
        with file:
            for line in file:
                self.log_entries.append(line)
                if "Done (" in line:
                    self.is_on = True

    def start(self, min_memory="4G", max_memory="7G"):
        arguments = (
            (
                self.path
                / "libraries"
                / "net"
                / "minecraftforge"
                / "forge"
                / "1.20.2-48.1.0"
                / "win_args.txt"
            )
            .read_text(encoding="utf-8")
            .split()
        )
        self.process = subprocess.Popen(
            [
                "java",
                f"-Xms{min_memory}",
                f"-Xmx{max_memory}",
                *arguments,
                "nogui",
            ],
            cwd=self.path,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        for file in [self.process.stdout, self.process.stderr]:
            Thread(
                target=self._process_output, args=[file], daemon=True
            ).start()

    def send_command(self, command):
        self.process.stdin.write(command + "\n")
        self.process.stdin.flush()

    def stop(self):
        self.send_command("stop")  # ???
        return_code = self.process.wait()
        self.process = None
        self.is_on = False
        return return_code