env/minecraft_launcher.py

import os
import re

import minecraft_launcher_lib
import sys
import utils as U

from .process_monitor import SubprocessMonitor


class MinecraftInstance:
    def __init__(
        self,
        client_id,
        redirect_url,
        secret_value,
        version,
        mineflayer,
        log_path="logs",
    ):
        self.client_id = client_id
        self.redirect_url = redirect_url
        self.secret_value = secret_value
        self.version = version
        self.log_path = log_path
        self.mc_dir = minecraft_launcher_lib.utils.get_minecraft_directory()
        self.port = None

        def stop_mineflayer():
            print("Stopping mineflayer")
            try:
                mineflayer.stop()
            except Exception as e:
                print(e)

        self.mc_command = self.get_mc_command()
        self.mc_process = SubprocessMonitor(
            commands=self.mc_command,
            name="minecraft",
            ready_match=r"Started serving on (\d+)",
            log_path=self.log_path,
            callback=stop_mineflayer,
            callback_match=r"\[Server thread/INFO\]: bot left the game",
            finished_callback=stop_mineflayer,
        )

    def get_mineflayer_process(self, server_port):
        U.f_mkdir(self.log_path, "../mineflayer")
        file_path = os.path.abspath(os.path.dirname(__file__))
        return SubprocessMonitor(
            commands=[
                "node",
                U.f_join(file_path, "mineflayer/index.js"),
                str(server_port),
            ],
            name="mineflayer",
            ready_match=r"Server started on port (\d+)",
            log_path=U.f_join(self.log_path, "mineflayer"),
        )

    def get_mc_command(self):
        file_path = os.path.abspath(os.path.dirname(__file__))
        if not U.f_exists(file_path, "config.json"):
            (
                login_url,
                state,
                code_verifier,
            ) = minecraft_launcher_lib.microsoft_account.get_secure_login_data(
                self.client_id, self.redirect_url
            )
            print(
                f"Please open {login_url} in your browser and copy the url you are redirected into the prompt below."
            )
            code_url = input()

            try:
                auth_code = (
                    minecraft_launcher_lib.microsoft_account.parse_auth_code_url(
                        code_url, state
                    )
                )
            except AssertionError:
                print("States do not match!")
                sys.exit(1)
            except KeyError:
                print("Url not valid")
                sys.exit(1)

            login_data = minecraft_launcher_lib.microsoft_account.complete_login(
                self.client_id,
                self.secret_value,
                self.redirect_url,
                auth_code,
                code_verifier,
            )

            options = {
                "username": login_data["name"],
                "uuid": login_data["id"],
                "token": login_data["access_token"],
            }
            U.json_dump(options, file_path, "config.json")
            print(f"Login success, save to {U.f_join(file_path, 'config.json')}")

        options = U.json_load(file_path, "config.json")
        mc_command = minecraft_launcher_lib.command.get_minecraft_command(
            self.version, self.mc_dir, options
        )

        return mc_command

    def run(self):
        self.mc_process.run()
        pattern = r"Started serving on (\d+)"
        match = re.search(pattern, self.mc_process.ready_line)
        if match:
            self.port = int(match.group(1))
            print("The mc server is listening on port", self.port)
        else:
            raise RuntimeError("Port not found")

    def stop(self):
        self.mc_process.stop()

    @property
    def is_running(self):
        return self.mc_process.is_running